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, Checklist, Elective
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):
   identify: str
   model: str = "1.0.0"
   description: str
   creator: 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: Checklist[str] = []


class CognitiveBlueprint(BaseModel):
   identification: BlueprintIdentity
   objectives: Checklist[str]
   constraints: Checklist[str] = []
   instruments: Checklist[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:
 identify: ResearchBot
 model: 1.2.0
 description: Solutions analysis questions utilizing calculation and reasoning
 creator: Auton Framework Demo
objectives:
 - Reply consumer questions precisely utilizing accessible instruments
 - Present step-by-step reasoning for all solutions
 - Cite the tactic used for every calculation
constraints:
 - By no means fabricate numbers or statistics
 - All the time validate mathematical outcomes earlier than reporting
 - Don't reply questions outdoors your instrument 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 do not know"
   - "I can not decide"
"""


DATA_ANALYST_YAML = """
identification:
 identify: DataAnalystBot
 model: 2.0.0
 description: Performs statistical evaluation and knowledge summarization
 creator: Auton Framework Demo
objectives:
 - Compute descriptive statistics for given knowledge
 - Establish traits and anomalies
 - Current findings clearly with numbers
constraints:
 - Solely work with numerical knowledge
 - All the time report uncertainty when pattern dimension is small (< 5 gadgets)
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:
   identify: str
   description: str
   parameters: Dict[str, str]
   operate: Callable
   returns: str


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


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


   def get(self, identify: str) -> Elective[ToolSpec]:
       return self._tools.get(identify)


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


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


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


registry = ToolRegistry()


@registry.register(
   identify="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(
   identify="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(
   identify="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="Consequence 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(
   identify="search_wikipedia_stub",
   description="Returns a stub abstract for well-known matters (demo — no stay web)",
   parameters={"matter": "Matter to lookup"},
   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 might question Wikipedia's API."

@registry.register(
   identify="statistics_engine",
   description="Computes descriptive statistics on an inventory of numbers",
   parameters={"numbers": "Comma-separated listing of numbers, e.g. '4,8,15,16,23,42'"},
   returns="JSON with imply, median, std_dev, min, max, rely"
)
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({
           "rely": 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(
   identify="list_sorter",
   description="Types a comma-separated listing of numbers",
   parameters={"numbers": "Comma-separated numbers", "order": "'asc' or 'desc'"},
   returns="Sorted comma-separated listing"
)
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 = 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: Checklist[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.function}: {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) -> Checklist[Dict]:
       messages = [{"role": "system", "content": system_prompt}]
       if self._summary:
           messages.append({"function": "system",
               "content material": f"[Memory Summary]: {self._summary.strip()}"})
       for entry in self._history[-self.config.window_size:]:
           messages.append({
               "function": entry.function if entry.function != "instrument" 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
   instrument: Elective[str]
   tool_args: Dict[str, Any]
   reasoning: str


@dataclass
class Plan:
   job: str
   steps: Checklist[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.identify}, model {bp.identification.model}.
           {bp.identification.description}


           ## Your Targets:
           {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)}


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


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


           Given a consumer job, 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 instrument 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, job: str, reminiscence: MemoryManager) -> Plan:
       system_prompt = self._build_planner_prompt()
       messages = reminiscence.get_messages(system_prompt)
       messages.append({"function": "consumer", "content material":
           f"Create a plan to finish this job: {job}"})
       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"],
               instrument=s.get("instrument"), tool_args=s.get("tool_args", {}),
               reasoning=s.get("reasoning", "")
           )
           for s in knowledge["steps"]
       ]
       return Plan(job=job, steps=steps, technique=self.blueprint.planning.technique)


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


@dataclass
class ExecutionTrace:
   plan: Plan
   outcomes: Checklist[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: Checklist[StepResult] = []
       if verbose:
           console.print(f"n[bold yellow]⚡ Executing:[/] {plan.job}")
           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.instrument and step.instrument != "null":
                   if verbose:
                       console.print(f"   🔧 Instrument: [green]{step.instrument}[/] | Args: {step.tool_args}")
                   output = self.registry.name(step.instrument, **step.tool_args)
                   consequence = StepResult(step.step_id, True, str(output), step.instrument)
                   if verbose:
                       console.print(f"   ✅ Consequence: {output}")
               else:
                   context_text = "n".be a 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.identify}. "
                       f"{self.blueprint.identification.description}. "
                       f"Constraints: {'; '.be a 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.instrument, 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: Checklist[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"Authentic job: {plan.job}nn"
           f"Step outcomes:n{steps_summary}nn"
           f"Present a transparent, full closing reply. Combine all step outcomes."
       )
       sys_prompt = (
           f"You're {self.blueprint.identification.identify}. "
           + ("All the time present your reasoning. " if self.blueprint.validation.require_reasoning else "")
           + f"Targets: {'; '.be a part of(self.blueprint.objectives)}"
       )
       messages = reminiscence.get_messages(sys_prompt)
       messages.append({"function": "consumer", "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: Checklist[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, job: 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 rationalization")


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


       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, job: str) -> Checklist[str]:
       immediate = (
           f"Process: {job}nnAnswer: {reply[:500]}nn"
           f'Does this reply deal with the duty? Reply JSON: {{"on_topic": true/false, "concern": "..."}}'
       )
       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:
           cross
       return []

@dataclass
class AgentResponse:
   agent_name: str
   job: 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, job: str, verbose: bool = True) -> AgentResponse:
       bp = self.blueprint
       if verbose:
           console.print(Panel(
               f"[bold]Agent:[/] {bp.identification.identify} v{bp.identification.model}n"
               f"[bold]Process:[/] {job}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("consumer", job)
       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 a part of(validation.points)}")


           if verbose:
               console.print("n[bold magenta]📋 Section 1: Planning...[/]")
           strive:
               plan = self.planner.plan(job, self.reminiscence)
               if verbose:
                   tree = Tree(f"[bold]Plan ({len(plan.steps)} steps)[/]")
                   for s in plan.steps:
                       icon = "🔧" if s.instrument else "🤔"
                       department = tree.add(f"{icon} Step {s.step_id}: {s.description}")
                       if s.instrument:
                           department.add(f"[green]Instrument:[/] {s.instrument}")
                           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]⚡ Section 2: Executing...[/]")
           hint = self.executor.execute_plan(plan, self.reminiscence, verbose=verbose)


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


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


           if validation.handed:
               break


           retries += 1
           self.reminiscence.add("assistant", hint.final_answer)
           self.reminiscence.add("consumer",
               f"Your earlier reply had points: {'; '.be a 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"🎯 Remaining Reply — {bp.identification.identify}",
               border_style="inexperienced"
           ))


       return AgentResponse(
           agent_name=bp.identification.identify, job=job,
           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(
       job=(
           "what number of steps of 20cm top would that be? Additionally, if I burn 0.15 "
           "energy per step, what is the complete 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(
       job=(
           "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, determine the perfect and worst months, "
           "and calculate development from first to final month."
       )
   )


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


   responses = {}
   for identify, 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 identify, r in responses.gadgets():
       desk.add_row(
           identify, str(r.total_steps),
           "✅" if r.validation.handed else "❌",
           f"{r.validation.rating:.2f}",
           r.final_answer[:140] + "..."
       )
   console.print(desk)