Constructing Subsequent-Gen Agentic AI: A Full Framework for Cognitive Blueprint Pushed Runtime Brokers with Reminiscence Instruments and Validation

import json, yaml, time, math, textwrap, datetime, getpass, os
from typing import Any, Callable, Dict, Record, Non-obligatory
from dataclasses import dataclass, subject
from enum import Enum


from openai import OpenAI
from pydantic import BaseModel
from wealthy.console import Console
from wealthy.panel import Panel
from wealthy.desk import Desk
from wealthy.tree import Tree


strive:
   from google.colab import userdata
   OPENAI_API_KEY = userdata.get('OPENAI_API_KEY')
besides Exception:
   OPENAI_API_KEY = getpass.getpass("🔑 Enter your OpenAI API key: ")


os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY
shopper = OpenAI(api_key=OPENAI_API_KEY)
console = Console()


class PlanningStrategy(str, Enum):
   SEQUENTIAL   = "sequential"
   HIERARCHICAL = "hierarchical"
   REACTIVE     = "reactive"


class MemoryType(str, Enum):
   SHORT_TERM = "short_term"
   EPISODIC   = "episodic"
   PERSISTENT = "persistent"


class BlueprintIdentity(BaseModel):
   title: str
   model: str = "1.0.0"
   description: str
   creator: str = "unknown"


class BlueprintMemory(BaseModel):
   kind: MemoryType = MemoryType.SHORT_TERM
   window_size: int = 10
   summarize_after: int = 20


class BlueprintPlanning(BaseModel):
   technique: PlanningStrategy = PlanningStrategy.SEQUENTIAL
   max_steps: int = 8
   max_retries: int = 2
   think_before_acting: bool = True


class BlueprintValidation(BaseModel):
   require_reasoning: bool = True
   min_response_length: int = 10
   forbidden_phrases: Record[str] = []


class CognitiveBlueprint(BaseModel):
   identification: BlueprintIdentity
   objectives: Record[str]
   constraints: Record[str] = []
   instruments: Record[str] = []
   reminiscence: BlueprintMemory = BlueprintMemory()
   planning: BlueprintPlanning = BlueprintPlanning()
   validation: BlueprintValidation = BlueprintValidation()
   system_prompt_extra: str = ""


def load_blueprint_from_yaml(yaml_str: str) -> CognitiveBlueprint:
   return CognitiveBlueprint(**yaml.safe_load(yaml_str))


RESEARCH_AGENT_YAML = """
identification:
 title: ResearchBot
 model: 1.2.0
 description: Solutions analysis questions utilizing calculation and reasoning
 creator: Auton Framework Demo
objectives:
 - Reply person questions precisely utilizing out there instruments
 - Present step-by-step reasoning for all solutions
 - Cite the tactic used for every calculation
constraints:
 - By no means fabricate numbers or statistics
 - At all times validate mathematical outcomes earlier than reporting
 - Don't reply questions outdoors your software capabilities
instruments:
 - calculator
 - unit_converter
 - date_calculator
 - search_wikipedia_stub
reminiscence:
 kind: episodic
 window_size: 12
 summarize_after: 30
planning:
 technique: sequential
 max_steps: 6
 max_retries: 2
 think_before_acting: true
validation:
 require_reasoning: true
 min_response_length: 20
 forbidden_phrases:
   - "I do not know"
   - "I can't decide"
"""


DATA_ANALYST_YAML = """
identification:
 title: DataAnalystBot
 model: 2.0.0
 description: Performs statistical evaluation and knowledge summarization
 creator: Auton Framework Demo
objectives:
 - Compute descriptive statistics for given knowledge
 - Establish developments and anomalies
 - Current findings clearly with numbers
constraints:
 - Solely work with numerical knowledge
 - At all times report uncertainty when pattern dimension is small (< 5 gadgets)
instruments:
 - calculator
 - statistics_engine
 - list_sorter
reminiscence:
 kind: short_term
 window_size: 6
planning:
 technique: hierarchical
 max_steps: 10
 max_retries: 3
 think_before_acting: true
validation:
 require_reasoning: true
 min_response_length: 30
 forbidden_phrases: []
"""

@dataclass
class ToolSpec:
   title: str
   description: str
   parameters: Dict[str, str]
   operate: Callable
   returns: str


class ToolRegistry:
   def __init__(self):
       self._tools: Dict[str, ToolSpec] = {}


   def register(self, title: str, description: str,
                parameters: Dict[str, str], returns: str):
       def decorator(fn: Callable) -> Callable:
           self._tools[name] = ToolSpec(title, description, parameters, fn, returns)
           return fn
       return decorator


   def get(self, title: str) -> Non-obligatory[ToolSpec]:
       return self._tools.get(title)


   def name(self, title: str, **kwargs) -> Any:
       spec = self._tools.get(title)
       if not spec:
           increase ValueError(f"Instrument '{title}' not present in registry")
       return spec.operate(**kwargs)


   def get_tool_descriptions(self, allowed: Record[str]) -> str:
       strains = []
       for title in allowed:
           spec = self._tools.get(title)
           if spec:
               params = ", ".be part of(f"{ok}: {v}" for ok, v in spec.parameters.gadgets())
               strains.append(
                   f"• {spec.title}({params})n"
                   f"  → {spec.description}n"
                   f"  Returns: {spec.returns}"
               )
       return "n".be part of(strains)


   def list_tools(self) -> Record[str]:
       return record(self._tools.keys())


registry = ToolRegistry()


@registry.register(
   title="calculator",
   description="Evaluates a secure mathematical expression",
   parameters={"expression": "A math expression string, e.g. '2 ** 10 + 5 * 3'"},
   returns="Numeric consequence as float"
)
def calculator(expression: str) -> str:
   strive:
       allowed = {ok: v for ok, v in math.__dict__.gadgets() if not ok.startswith("_")}
       allowed.replace({"abs": abs, "spherical": spherical, "pow": pow})
       return str(eval(expression, {"__builtins__": {}}, allowed))
   besides Exception as e:
       return f"Error: {e}"


@registry.register(
   title="unit_converter",
   description="Converts between widespread models of measurement",
   parameters={
       "worth": "Numeric worth to transform",
       "from_unit": "Supply unit (km, miles, kg, lbs, celsius, fahrenheit, liters, gallons, meters, toes)",
       "to_unit": "Goal unit"
   },
   returns="Transformed worth as string with models"
)
def unit_converter(worth: float, from_unit: str, to_unit: str) -> str:
   conversions = {
       ("km", "miles"): lambda x: x * 0.621371,
       ("miles", "km"): lambda x: x * 1.60934,
       ("kg", "lbs"):   lambda x: x * 2.20462,
       ("lbs", "kg"):   lambda x: x / 2.20462,
       ("celsius", "fahrenheit"): lambda x: x * 9/5 + 32,
       ("fahrenheit", "celsius"): lambda x: (x - 32) * 5/9,
       ("liters", "gallons"): lambda x: x * 0.264172,
       ("gallons", "liters"): lambda x: x * 3.78541,
       ("meters", "toes"): lambda x: x * 3.28084,
       ("toes", "meters"): lambda x: x / 3.28084,
   }
   key = (from_unit.decrease(), to_unit.decrease())
   if key in conversions:
       return f"{conversions[key](float(worth)):.4f} {to_unit}"
   return f"Conversion from {from_unit} to {to_unit} not supported"


@registry.register(
   title="date_calculator",
   description="Calculates days between two dates, or provides/subtracts days from a date",
   parameters={
       "operation": "'days_between' or 'add_days'",
       "date1": "Date string in YYYY-MM-DD format",
       "date2": "Second date for days_between (YYYY-MM-DD), or variety of days for add_days"
   },
   returns="End result as string"
)
def date_calculator(operation: str, date1: str, date2: str) -> str:
   strive:
       d1 = datetime.datetime.strptime(date1, "%Y-%m-%d")
       if operation == "days_between":
           d2 = datetime.datetime.strptime(date2, "%Y-%m-%d")
           return f"{abs((d2 - d1).days)} days between {date1} and {date2}"
       elif operation == "add_days":
           consequence = d1 + datetime.timedelta(days=int(date2))
           return f"{consequence.strftime('%Y-%m-%d')} (added {date2} days to {date1})"
       return f"Unknown operation: {operation}"
   besides Exception as e:
       return f"Error: {e}"


@registry.register(
   title="search_wikipedia_stub",
   description="Returns a stub abstract for well-known matters (demo — no dwell web)",
   parameters={"matter": "Subject to search for"},
   returns="Quick textual content abstract"
)
def search_wikipedia_stub(matter: str) -> str:
   stubs = {
       "openai": "OpenAI is an AI analysis firm based in 2015. It created GPT-4 and the ChatGPT product.",
   }
   for key, val in stubs.gadgets():
       if key in matter.decrease():
           return val
   return f"No stub discovered for '{matter}'. In manufacturing, this may question Wikipedia's API."

@registry.register(
   title="statistics_engine",
   description="Computes descriptive statistics on a listing of numbers",
   parameters={"numbers": "Comma-separated record of numbers, e.g. '4,8,15,16,23,42'"},
   returns="JSON with imply, median, std_dev, min, max, depend"
)
def statistics_engine(numbers: str) -> str:
   strive:
       nums = [float(x.strip()) for x in numbers.split(",")]
       n = len(nums)
       imply = sum(nums) / n
       sorted_nums = sorted(nums)
       mid = n // 2
       median = sorted_nums[mid] if n % 2 else (sorted_nums[mid-1] + sorted_nums[mid]) / 2
       std_dev = math.sqrt(sum((x - imply) ** 2 for x in nums) / n)
       return json.dumps({
           "depend": n, "imply": spherical(imply, 4), "median": spherical(median, 4),
           "std_dev": spherical(std_dev, 4), "min": min(nums),
           "max": max(nums), "vary": max(nums) - min(nums)
       }, indent=2)
   besides Exception as e:
       return f"Error: {e}"


@registry.register(
   title="list_sorter",
   description="Types a comma-separated record of numbers",
   parameters={"numbers": "Comma-separated numbers", "order": "'asc' or 'desc'"},
   returns="Sorted comma-separated record"
)
def list_sorter(numbers: str, order: str = "asc") -> str:
   nums = [float(x.strip()) for x in numbers.split(",")]
   nums.type(reverse=(order == "desc"))
   return ", ".be part of(str(n) for n in nums)


@dataclass
class MemoryEntry:
   position: str
   content material: str
   timestamp: float = subject(default_factory=time.time)
   metadata: Dict = subject(default_factory=dict)


class MemoryManager:
   def __init__(self, config: BlueprintMemory, llm_client: OpenAI):
       self.config = config
       self.shopper = llm_client
       self._history: Record[MemoryEntry] = []
       self._summary: str = ""


   def add(self, position: str, content material: str, metadata: Dict = None):
       self._history.append(MemoryEntry(position=position, content material=content material, metadata=metadata or {}))
       if (self.config.kind == MemoryType.EPISODIC and
               len(self._history) > self.config.summarize_after):
           self._compress_memory()


   def _compress_memory(self):
       to_compress = self._history[:-self.config.window_size]
       self._history = self._history[-self.config.window_size:]
       textual content = "n".be part of(f"{e.position}: {e.content material[:200]}" for e in to_compress)
       strive:
           resp = self.shopper.chat.completions.create(
               mannequin="gpt-4o-mini",
               messages=[{"role": "user", "content":
                   f"Summarize this conversation history in 3 sentences:n{text}"}],
               max_tokens=150
           )
           self._summary += " " + resp.decisions[0].message.content material.strip()
       besides Exception:
           self._summary += f" [compressed {len(to_compress)} messages]"


   def get_messages(self, system_prompt: str) -> Record[Dict]:
       messages = [{"role": "system", "content": system_prompt}]
       if self._summary:
           messages.append({"position": "system",
               "content material": f"[Memory Summary]: {self._summary.strip()}"})
       for entry in self._history[-self.config.window_size:]:
           messages.append({
               "position": entry.position if entry.position != "software" else "assistant",
               "content material": entry.content material
           })
       return messages


   def clear(self):
       self._history = []
       self._summary = ""


   @property
   def message_count(self) -> int:
       return len(self._history)

@dataclass
class PlanStep:
   step_id: int
   description: str
   software: Non-obligatory[str]
   tool_args: Dict[str, Any]
   reasoning: str


@dataclass
class Plan:
   process: str
   steps: Record[PlanStep]
   technique: PlanningStrategy


class Planner:
   def __init__(self, blueprint: CognitiveBlueprint,
                registry: ToolRegistry, llm_client: OpenAI):
       self.blueprint = blueprint
       self.registry  = registry
       self.shopper    = llm_client


   def _build_planner_prompt(self) -> str:
       bp = self.blueprint
       return textwrap.dedent(f"""
           You're {bp.identification.title}, model {bp.identification.model}.
           {bp.identification.description}


           ## Your Targets:
           {chr(10).be part of(f'  - {g}' for g in bp.objectives)}


           ## Your Constraints:
           {chr(10).be part of(f'  - {c}' for c in bp.constraints)}


           ## Obtainable Instruments:
           {self.registry.get_tool_descriptions(bp.instruments)}


           ## Planning Technique: {bp.planning.technique}
           ## Max Steps: {bp.planning.max_steps}


           Given a person process, produce a JSON execution plan with this precise construction:
           {{
             "steps": [
               {{
                 "step_id": 1,
                 "description": "What this step does",
                 "tool": "tool_name or null if no tool needed",
                 "tool_args": {{"arg1": "value1"}},
                 "reasoning": "Why this step is needed"
               }}
             ]
           }}


           Guidelines:
           - Solely use instruments listed above
           - Set software to null for pure reasoning steps
           - Preserve steps <= {bp.planning.max_steps}
           - Return ONLY legitimate JSON, no markdown fences
           {bp.system_prompt_extra}
       """).strip()


   def plan(self, process: str, reminiscence: MemoryManager) -> Plan:
       system_prompt = self._build_planner_prompt()
       messages = reminiscence.get_messages(system_prompt)
       messages.append({"position": "person", "content material":
           f"Create a plan to finish this process: {process}"})
       resp = self.shopper.chat.completions.create(
           mannequin="gpt-4o-mini", messages=messages,
           max_tokens=1200, temperature=0.2
       )
       uncooked = resp.decisions[0].message.content material.strip()
       uncooked = uncooked.change("```json", "").change("```", "").strip()
       knowledge = json.hundreds(uncooked)
       steps = [
           PlanStep(
               step_id=s["step_id"], description=s["description"],
               software=s.get("software"), tool_args=s.get("tool_args", {}),
               reasoning=s.get("reasoning", "")
           )
           for s in knowledge["steps"]
       ]
       return Plan(process=process, steps=steps, technique=self.blueprint.planning.technique)


@dataclass
class StepResult:
   step_id: int
   success: bool
   output: str
   tool_used: Non-obligatory[str]
   error: Non-obligatory[str] = None


@dataclass
class ExecutionTrace:
   plan: Plan
   outcomes: Record[StepResult]
   final_answer: str


class Executor:
   def __init__(self, blueprint: CognitiveBlueprint,
                registry: ToolRegistry, llm_client: OpenAI):
       self.blueprint = blueprint
       self.registry  = registry
       self.shopper    = llm_client

  def execute_plan(self, plan: Plan, reminiscence: MemoryManager,
                    verbose: bool = True) -> ExecutionTrace:
       outcomes: Record[StepResult] = []
       if verbose:
           console.print(f"n[bold yellow]⚡ Executing:[/] {plan.process}")
           console.print(f"   Technique: {plan.technique} | Steps: {len(plan.steps)}")


       for step in plan.steps:
           if verbose:
               console.print(f"n  [cyan]Step {step.step_id}:[/] {step.description}")
           strive:
               if step.software and step.software != "null":
                   if verbose:
                       console.print(f"   🔧 Instrument: [green]{step.software}[/] | Args: {step.tool_args}")
                   output = self.registry.name(step.software, **step.tool_args)
                   consequence = StepResult(step.step_id, True, str(output), step.software)
                   if verbose:
                       console.print(f"   ✅ End result: {output}")
               else:
                   context_text = "n".be part of(
                       f"Step {r.step_id} consequence: {r.output}" for r in outcomes)
                   immediate = (
                       f"Earlier outcomes:n{context_text}nn"
                       f"Now full this step: {step.description}n"
                       f"Reasoning trace: {step.reasoning}"
                   ) if context_text else (
                       f"Full this step: {step.description}n"
                       f"Reasoning trace: {step.reasoning}"
                   )
                   sys_prompt = (
                       f"You're {self.blueprint.identification.title}. "
                       f"{self.blueprint.identification.description}. "
                       f"Constraints: {'; '.be part of(self.blueprint.constraints)}"
                   )
                   resp = self.shopper.chat.completions.create(
                       mannequin="gpt-4o-mini",
                       messages=[
                           {"role": "system", "content": sys_prompt},
                           {"role": "user",   "content": prompt}
                       ],
                       max_tokens=500, temperature=0.3
                   )
                   output = resp.decisions[0].message.content material.strip()
                   consequence = StepResult(step.step_id, True, output, None)
                   if verbose:
                       preview = output[:120] + "..." if len(output) > 120 else output
                       console.print(f"   🤔 Reasoning: {preview}")
           besides Exception as e:
               consequence = StepResult(step.step_id, False, "", step.software, str(e))
               if verbose:
                   console.print(f"   ❌ Error: {e}")
           outcomes.append(consequence)


       final_answer = self._synthesize(plan, outcomes, reminiscence)
       return ExecutionTrace(plan=plan, outcomes=outcomes, final_answer=final_answer)


   def _synthesize(self, plan: Plan, outcomes: Record[StepResult],
                   reminiscence: MemoryManager) -> str:
       steps_summary = "n".be part of(
           f"Step {r.step_id} ({'✅' if r.success else '❌'}): {r.output[:300]}"
           for r in outcomes
       )
       synthesis_prompt = (
           f"Authentic process: {plan.process}nn"
           f"Step outcomes:n{steps_summary}nn"
           f"Present a transparent, full remaining reply. Combine all step outcomes."
       )
       sys_prompt = (
           f"You're {self.blueprint.identification.title}. "
           + ("At all times present your reasoning. " if self.blueprint.validation.require_reasoning else "")
           + f"Targets: {'; '.be part of(self.blueprint.objectives)}"
       )
       messages = reminiscence.get_messages(sys_prompt)
       messages.append({"position": "person", "content material": synthesis_prompt})
       resp = self.shopper.chat.completions.create(
           mannequin="gpt-4o-mini", messages=messages,
           max_tokens=600, temperature=0.3
       )
       return resp.decisions[0].message.content material.strip()


@dataclass
class ValidationResult:
   handed: bool
   points: Record[str]
   rating: float


class Validator:
   def __init__(self, blueprint: CognitiveBlueprint, llm_client: OpenAI):
       self.blueprint = blueprint
       self.shopper    = llm_client


   def validate(self, reply: str, process: str,
                use_llm_check: bool = False) -> ValidationResult:
       points = []
       v = self.blueprint.validation


       if len(reply) < v.min_response_length:
           points.append(f"Response too quick: {len(reply)} chars (min: {v.min_response_length})")


       answer_lower = reply.decrease()
       for phrase in v.forbidden_phrases:
           if phrase.decrease() in answer_lower:
               points.append(f"Forbidden phrase detected: '{phrase}'")


       if v.require_reasoning:
           indicators = ["because", "therefore", "since", "step", "first",
                         "result", "calculated", "computed", "found that"]
           if not any(ind in answer_lower for ind in indicators):
               points.append("Response lacks seen reasoning or clarification")


       if use_llm_check:
           points.prolong(self._llm_quality_check(reply, process))


       return ValidationResult(handed=len(points) == 0,
                               points=points,
                               rating=max(0.0, 1.0 - len(points) * 0.25))


   def _llm_quality_check(self, reply: str, process: str) -> Record[str]:
       immediate = (
           f"Process: {process}nnAnswer: {reply[:500]}nn"
           f'Does this reply tackle the duty? Reply JSON: {{"on_topic": true/false, "subject": "..."}}'
       )
       strive:
           resp = self.shopper.chat.completions.create(
               mannequin="gpt-4o-mini",
               messages=[{"role": "user", "content": prompt}],
               max_tokens=100
           )
           uncooked = resp.decisions[0].message.content material.strip().change("```json","").change("```","")
           knowledge = json.hundreds(uncooked)
           if not knowledge.get("on_topic", True):
               return [f"LLM quality check: {data.get('issue', 'off-topic')}"]
       besides Exception:
           move
       return []

@dataclass
class AgentResponse:
   agent_name: str
   process: str
   final_answer: str
   hint: ExecutionTrace
   validation: ValidationResult
   retries: int
   total_steps: int


class RuntimeEngine:
   def __init__(self, blueprint: CognitiveBlueprint,
                registry: ToolRegistry, llm_client: OpenAI):
       self.blueprint = blueprint
       self.reminiscence    = MemoryManager(blueprint.reminiscence, llm_client)
       self.planner   = Planner(blueprint, registry, llm_client)
       self.executor  = Executor(blueprint, registry, llm_client)
       self.validator = Validator(blueprint, llm_client)


   def run(self, process: str, verbose: bool = True) -> AgentResponse:
       bp = self.blueprint
       if verbose:
           console.print(Panel(
               f"[bold]Agent:[/] {bp.identification.title} v{bp.identification.model}n"
               f"[bold]Process:[/] {process}n"
               f"[bold]Technique:[/] {bp.planning.technique} | "
               f"Max Steps: {bp.planning.max_steps} | "
               f"Max Retries: {bp.planning.max_retries}",
               title="🚀 Runtime Engine Beginning", border_style="blue"
           ))


       self.reminiscence.add("person", process)
       retries, hint, validation = 0, None, None


       for try in vary(bp.planning.max_retries + 1):
           if try > 0 and verbose:
               console.print(f"n[yellow]⟳ Retry {try}/{bp.planning.max_retries}[/]")
               console.print(f"  Points: {', '.be part of(validation.points)}")


           if verbose:
               console.print("n[bold magenta]📋 Part 1: Planning...[/]")
           strive:
               plan = self.planner.plan(process, self.reminiscence)
               if verbose:
                   tree = Tree(f"[bold]Plan ({len(plan.steps)} steps)[/]")
                   for s in plan.steps:
                       icon = "🔧" if s.software else "🤔"
                       department = tree.add(f"{icon} Step {s.step_id}: {s.description}")
                       if s.software:
                           department.add(f"[green]Instrument:[/] {s.software}")
                           department.add(f"[yellow]Args:[/] {s.tool_args}")
                   console.print(tree)
           besides Exception as e:
               if verbose: console.print(f"[red]Planning failed:[/] {e}")
               break


           if verbose:
               console.print("n[bold magenta]⚡ Part 2: Executing...[/]")
           hint = self.executor.execute_plan(plan, self.reminiscence, verbose=verbose)


           if verbose:
               console.print("n[bold magenta]✅ Part 3: Validating...[/]")
           validation = self.validator.validate(hint.final_answer, process)


           if verbose:
               standing = "[green]PASSED[/]" if validation.handed else "[red]FAILED[/]"
               console.print(f"  Validation: {standing} | Rating: {validation.rating:.2f}")
               for subject in validation.points:
                   console.print(f"  ⚠️  {subject}")


           if validation.handed:
               break


           retries += 1
           self.reminiscence.add("assistant", hint.final_answer)
           self.reminiscence.add("person",
               f"Your earlier reply had points: {'; '.be part of(validation.points)}. "
               f"Please enhance."
           )


       if hint:
           self.reminiscence.add("assistant", hint.final_answer)


       if verbose:
           console.print(Panel(
               hint.final_answer if hint else "No reply generated",
               title=f"🎯 Last Reply — {bp.identification.title}",
               border_style="inexperienced"
           ))


       return AgentResponse(
           agent_name=bp.identification.title, process=process,
           final_answer=hint.final_answer if hint else "",
           hint=hint, validation=validation,
           retries=retries,
           total_steps=len(hint.outcomes) if hint else 0
       )


   def reset_memory(self):
       self.reminiscence.clear()


def build_engine(blueprint_yaml: str, registry: ToolRegistry,
                llm_client: OpenAI) -> RuntimeEngine:
   return RuntimeEngine(load_blueprint_from_yaml(blueprint_yaml), registry, llm_client)


if __name__ == "__main__":


   print("n" + "="*60)
   print("DEMO 1: ResearchBot")
   print("="*60)
   research_engine = build_engine(RESEARCH_AGENT_YAML, registry, shopper)
   research_engine.run(
       process=(
           "what number of steps of 20cm top would that be? Additionally, if I burn 0.15 "
           "energy per step, what is the whole calorie burn? Present all calculations."
       )
   )


   print("n" + "="*60)
   print("DEMO 2: DataAnalystBot")
   print("="*60)
   analyst_engine = build_engine(DATA_ANALYST_YAML, registry, shopper)
   analyst_engine.run(
       process=(
           "Analyze this dataset of month-to-month gross sales figures (in 1000's): "
           "142, 198, 173, 155, 221, 189, 203, 167, 244, 198, 212, 231. "
           "Compute key statistics, establish the very best and worst months, "
           "and calculate development from first to final month."
       )
   )


   print("n" + "="*60)
   print("PORTABILITY DEMO: Identical process → 2 totally different blueprints")
   print("="*60)
   SHARED_TASK = "Calculate 15% of two,500 and inform me the consequence."


   responses = {}
   for title, yaml_str in [
       ("ResearchBot",    RESEARCH_AGENT_YAML),
       ("DataAnalystBot", DATA_ANALYST_YAML),
   ]:
       eng = build_engine(yaml_str, registry, shopper)
       responses[name] = eng.run(SHARED_TASK, verbose=False)


   desk = Desk(title="🔄 Blueprint Portability", show_header=True, show_lines=True)
   desk.add_column("Agent",   fashion="cyan",   width=18)
   desk.add_column("Steps",   fashion="yellow", width=6)
   desk.add_column("Legitimate?",  width=7)
   desk.add_column("Rating",   width=6)
   desk.add_column("Reply Preview", width=55)


   for title, r in responses.gadgets():
       desk.add_row(
           title, str(r.total_steps),
           "✅" if r.validation.handed else "❌",
           f"{r.validation.rating:.2f}",
           r.final_answer[:140] + "..."
       )
   console.print(desk)