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, Listing, Non-compulsory
from dataclasses import dataclass, area
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
consumer = 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
   writer: str = "unknown"


class BlueprintMemory(BaseModel):
   sort: 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: Listing[str] = []


class CognitiveBlueprint(BaseModel):
   identification: BlueprintIdentity
   objectives: Listing[str]
   constraints: Listing[str] = []
   instruments: Listing[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
 writer: Auton Framework Demo
objectives:
 - Reply consumer questions precisely utilizing accessible instruments
 - Present step-by-step reasoning for all solutions
 - Cite the strategy 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 exterior your software capabilities
instruments:
 - calculator
 - unit_converter
 - date_calculator
 - search_wikipedia_stub
reminiscence:
 sort: 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 don't know"
   - "I cannot determine"
"""


DATA_ANALYST_YAML = """
identification:
 title: DataAnalystBot
 model: 2.0.0
 description: Performs statistical evaluation and information summarization
 writer: Auton Framework Demo
objectives:
 - Compute descriptive statistics for given information
 - Establish developments and anomalies
 - Current findings clearly with numbers
constraints:
 - Solely work with numerical information
 - At all times report uncertainty when pattern dimension is small (< 5 objects)
instruments:
 - calculator
 - statistics_engine
 - list_sorter
reminiscence:
 sort: 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]
   perform: 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-compulsory[ToolSpec]:
       return self._tools.get(title)


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


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


   def list_tools(self) -> Listing[str]:
       return listing(self._tools.keys())


registry = ToolRegistry()


@registry.register(
   title="calculator",
   description="Evaluates a safe mathematical expression",
   parameters={"expression": "A math expression string, e.g. '2 ** 10 + 5 * 3'"},
   returns="Numeric result as float"
)
def calculator(expression: str) -> str:
   strive:
       allowed = {ok: v for ok, v in math.__dict__.objects() if not ok.startswith("_")}
       allowed.replace({"abs": abs, "round": 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 common units of measurement",
   parameters={
       "value": "Numeric value to convert",
       "from_unit": "Source unit (km, miles, kg, lbs, celsius, fahrenheit, liters, gallons, meters, feet)",
       "to_unit": "Target unit"
   },
   returns="Converted value as string with units"
)
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", "feet"): lambda x: x * 3.28084,
       ("feet", "meters"): lambda x: x / 3.28084,
   }
   key = (from_unit.decrease(), to_unit.decrease())
   if key in conversions:
       return f"{conversions[key](float(value)):.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 adds/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 number of days for add_days"
   },
   returns="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":
           end result = d1 + datetime.timedelta(days=int(date2))
           return f"{result.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 summary for well-known topics (demo — no live internet)",
   parameters={"topic": "Topic to look up"},
   returns="Short text summary"
)
def search_wikipedia_stub(subject: str) -> str:
   stubs = {
       "openai": "OpenAI is an AI research company founded in 2015. It created GPT-4 and the ChatGPT product.",
   }
   for key, val in stubs.objects():
       if key in subject.decrease():
           return val
   return f"No stub found for '{topic}'. In production, this would query Wikipedia's API."

@registry.register(
   title="statistics_engine",
   description="Computes descriptive statistics on a list of numbers",
   parameters={"numbers": "Comma-separated list of numbers, e.g. '4,8,15,16,23,42'"},
   returns="JSON with mean, median, std_dev, min, max, count"
)
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({
           "count": n, "mean": spherical(imply, 4), "median": spherical(median, 4),
           "std_dev": spherical(std_dev, 4), "min": min(nums),
           "max": max(nums), "range": max(nums) - min(nums)
       }, indent=2)
   besides Exception as e:
       return f"Error: {e}"


@registry.register(
   title="list_sorter",
   description="Sorts a comma-separated list of numbers",
   parameters={"numbers": "Comma-separated numbers", "order": "'asc' or 'desc'"},
   returns="Sorted comma-separated list"
)
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 a part of(str(n) for n in nums)


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


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


   def add(self, function: str, content material: str, metadata: Dict = None):
       self._history.append(MemoryEntry(function=function, content material=content material, metadata=metadata or {}))
       if (self.config.sort == 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 a part of(f"{e.role}: {e.content[:200]}" for e in to_compress)
       strive:
           resp = self.consumer.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.selections[0].message.content material.strip()
       besides Exception:
           self._summary += f" [compressed {len(to_compress)} messages]"


   def get_messages(self, system_prompt: str) -> Listing[Dict]:
       messages = [{"role": "system", "content": system_prompt}]
       if self._summary:
           messages.append({"role": "system",
               "content": f"[Memory Summary]: {self._summary.strip()}"})
       for entry in self._history[-self.config.window_size:]:
           messages.append({
               "role": entry.function if entry.function != "tool" else "assistant",
               "content": 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-compulsory[str]
   tool_args: Dict[str, Any]
   reasoning: str


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


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


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


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


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


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


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


           Given a consumer process, produce a JSON execution plan with this actual 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
           - Maintain 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({"role": "user", "content":
           f"Create a plan to complete this task: {task}"})
       resp = self.consumer.chat.completions.create(
           mannequin="gpt-4o-mini", messages=messages,
           max_tokens=1200, temperature=0.2
       )
       uncooked = resp.selections[0].message.content material.strip()
       uncooked = uncooked.substitute("```json", "").substitute("```", "").strip()
       information = json.hundreds(uncooked)
       steps = [
           PlanStep(
               step_id=s["step_id"], description=s["description"],
               software=s.get("tool"), tool_args=s.get("tool_args", {}),
               reasoning=s.get("reasoning", "")
           )
           for s in information["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-compulsory[str]
   error: Non-compulsory[str] = None


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


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

  def execute_plan(self, plan: Plan, reminiscence: MemoryManager,
                    verbose: bool = True) -> ExecutionTrace:
       outcomes: Listing[StepResult] = []
       if verbose:
           console.print(f"n[bold yellow]⚡ Executing:[/] {plan.task}")
           console.print(f"   Strategy: {plan.strategy} | 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"   🔧 Tool: [green]{step.tool}[/] | Args: {step.tool_args}")
                   output = self.registry.name(step.software, **step.tool_args)
                   end result = StepResult(step.step_id, True, str(output), step.software)
                   if verbose:
                       console.print(f"   ✅ Result: {output}")
               else:
                   context_text = "n".be a part of(
                       f"Step {r.step_id} result: {r.output}" for r in outcomes)
                   immediate = (
                       f"Previous results:n{context_text}nn"
                       f"Now complete this step: {step.description}n"
                       f"Reasoning hint: {step.reasoning}"
                   ) if context_text else (
                       f"Complete this step: {step.description}n"
                       f"Reasoning hint: {step.reasoning}"
                   )
                   sys_prompt = (
                       f"You are {self.blueprint.identity.name}. "
                       f"{self.blueprint.identity.description}. "
                       f"Constraints: {'; '.join(self.blueprint.constraints)}"
                   )
                   resp = self.consumer.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.selections[0].message.content material.strip()
                   end result = 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:
               end result = StepResult(step.step_id, False, "", step.software, str(e))
               if verbose:
                   console.print(f"   ❌ Error: {e}")
           outcomes.append(end result)


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


   def _synthesize(self, plan: Plan, outcomes: Listing[StepResult],
                   reminiscence: MemoryManager) -> str:
       steps_summary = "n".be a part of(
           f"Step {r.step_id} ({'✅' if r.success else '❌'}): {r.output[:300]}"
           for r in outcomes
       )
       synthesis_prompt = (
           f"Original task: {plan.task}nn"
           f"Step results:n{steps_summary}nn"
           f"Provide a clear, complete final answer. Integrate all step results."
       )
       sys_prompt = (
           f"You are {self.blueprint.identity.name}. "
           + ("Always show your reasoning. " if self.blueprint.validation.require_reasoning else "")
           + f"Goals: {'; '.join(self.blueprint.goals)}"
       )
       messages = reminiscence.get_messages(sys_prompt)
       messages.append({"role": "user", "content": synthesis_prompt})
       resp = self.consumer.chat.completions.create(
           mannequin="gpt-4o-mini", messages=messages,
           max_tokens=600, temperature=0.3
       )
       return resp.selections[0].message.content material.strip()


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


class Validator:
   def __init__(self, blueprint: CognitiveBlueprint, llm_client: OpenAI):
       self.blueprint = blueprint
       self.consumer    = 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 short: {len(answer)} 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 visible reasoning or explanation")


       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) -> Listing[str]:
       immediate = (
           f"Task: {task}nnAnswer: {answer[:500]}nn"
           f'Does this reply deal with the duty? Reply JSON: {{"on_topic": true/false, "issue": "..."}}'
       )
       strive:
           resp = self.consumer.chat.completions.create(
               mannequin="gpt-4o-mini",
               messages=[{"role": "user", "content": prompt}],
               max_tokens=100
           )
           uncooked = resp.selections[0].message.content material.strip().substitute("```json","").substitute("```","")
           information = json.hundreds(uncooked)
           if not information.get("on_topic", True):
               return [f"LLM quality check: {data.get('issue', 'off-topic')}"]
       besides Exception:
           go
       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.identity.name} v{bp.identity.version}n"
               f"[bold]Task:[/] {task}n"
               f"[bold]Strategy:[/] {bp.planning.strategy} | "
               f"Max Steps: {bp.planning.max_steps} | "
               f"Max Retries: {bp.planning.max_retries}",
               title="🚀 Runtime Engine Starting", border_style="blue"
           ))


       self.reminiscence.add("user", 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 {attempt}/{bp.planning.max_retries}[/]")
               console.print(f"  Issues: {', '.join(validation.issues)}")


           if verbose:
               console.print("n[bold magenta]📋 Phase 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]Tool:[/] {s.tool}")
                           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]⚡ Phase 2: Executing...[/]")
           hint = self.executor.execute_plan(plan, self.reminiscence, verbose=verbose)


           if verbose:
               console.print("n[bold magenta]✅ Phase 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: {status} | Score: {validation.score:.2f}")
               for challenge in validation.points:
                   console.print(f"  ⚠️  {issue}")


           if validation.handed:
               break


           retries += 1
           self.reminiscence.add("assistant", hint.final_answer)
           self.reminiscence.add("user",
               f"Your previous answer had issues: {'; '.join(validation.issues)}. "
               f"Please improve."
           )


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


       if verbose:
           console.print(Panel(
               hint.final_answer if hint else "No answer generated",
               title=f"🎯 Final Answer — {bp.identity.name}",
               border_style="green"
           ))


       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, consumer)
   research_engine.run(
       process=(
           "how many steps of 20cm height would that be? Also, if I burn 0.15 "
           "calories per step, what's the total calorie burn? Show all calculations."
       )
   )


   print("n" + "="*60)
   print("DEMO 2: DataAnalystBot")
   print("="*60)
   analyst_engine = build_engine(DATA_ANALYST_YAML, registry, consumer)
   analyst_engine.run(
       process=(
           "Analyze this dataset of monthly sales figures (in thousands): "
           "142, 198, 173, 155, 221, 189, 203, 167, 244, 198, 212, 231. "
           "Compute key statistics, identify the best and worst months, "
           "and calculate growth from first to last month."
       )
   )


   print("n" + "="*60)
   print("PORTABILITY DEMO: Same task → 2 different blueprints")
   print("="*60)
   SHARED_TASK = "Calculate 15% of 2,500 and tell me the result."


   responses = {}
   for title, yaml_str in [
       ("ResearchBot",    RESEARCH_AGENT_YAML),
       ("DataAnalystBot", DATA_ANALYST_YAML),
   ]:
       eng = build_engine(yaml_str, registry, consumer)
       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("Valid?",  width=7)
   desk.add_column("Score",   width=6)
   desk.add_column("Answer Preview", width=55)


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