On this tutorial, we construct a whole cognitive blueprint and runtime agent framework. We outline structured blueprints for id, objectives, planning, reminiscence, validation, and power entry, and use them to create brokers that not solely reply but additionally plan, execute, validate, and systematically enhance their outputs. Alongside the tutorial, we present how the identical runtime engine can help a number of agent personalities and behaviors via blueprint portability, making the general design modular, extensible, and sensible for superior agentic AI experimentation.
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
attempt:
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):
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: Listing[str] = []
class CognitiveBlueprint(BaseModel):
id: 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 = """
id:
title: ResearchBot
model: 1.2.0
description: Solutions analysis questions utilizing calculation and reasoning
writer: Auton Framework Demo
objectives:
- Reply person questions precisely utilizing out there instruments
- Present step-by-step reasoning for all solutions
- Cite the strategy 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 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 not decide"
"""
DATA_ANALYST_YAML = """
id:
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
- All the time report uncertainty when pattern dimension is small (< 5 objects)
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: []
"""
We arrange the core setting and outline the cognitive blueprint, which constructions how an agent thinks and behaves. We create strongly typed fashions for id, reminiscence configuration, planning technique, and validation guidelines utilizing Pydantic and enums. We additionally outline two YAML-based blueprints, permitting us to configure completely different agent personalities and capabilities with out altering the underlying runtime system.
@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:
elevate ValueError(f"Device '{title}' not present 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 part of(f"{ok}: {v}" for ok, v in spec.parameters.objects())
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) -> Listing[str]:
return listing(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 end result as float"
)
def calculator(expression: str) -> str:
attempt:
allowed = {ok: v for ok, v in math.__dict__.objects() 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 frequent items of measurement",
parameters={
"worth": "Numeric worth to transform",
"from_unit": "Supply unit (km, miles, kg, lbs, celsius, fahrenheit, liters, gallons, meters, ft)",
"to_unit": "Goal unit"
},
returns="Transformed worth as string with items"
)
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", "ft"): lambda x: x * 3.28084,
("ft", "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="Consequence as string"
)
def date_calculator(operation: str, date1: str, date2: str) -> str:
attempt:
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"{end 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 abstract for well-known subjects (demo — no dwell web)",
parameters={"subject": "Matter to search for"},
returns="Quick textual content abstract"
)
def search_wikipedia_stub(subject: 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.objects():
if key in subject.decrease():
return val
return f"No stub discovered for '{subject}'. In manufacturing, this may question Wikipedia's API."We implement the software registry that enables brokers to find and use exterior capabilities dynamically. We design a structured system during which instruments are registered with metadata, together with parameters, descriptions, and return values. We additionally implement a number of sensible instruments, equivalent to a calculator, unit converter, date calculator, and a Wikipedia search stub that the brokers can invoke throughout execution.
@registry.register(
title="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:
attempt:
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(
title="list_sorter",
description="Kinds 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.kind(reverse=(order == "desc"))
return ", ".be 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.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.function}: {e.content material[:200]}" for e in to_compress)
attempt:
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.decisions[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({"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 != "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)We lengthen the software ecosystem and introduce the reminiscence administration layer that shops dialog historical past and compresses it when vital. We implement statistical instruments and sorting utilities that allow the information evaluation agent to carry out structured numerical operations. On the identical time, we design a reminiscence system that tracks interactions, summarizes lengthy histories, and supplies contextual messages to the language mannequin.
@dataclass
class PlanStep:
step_id: int
description: str
software: Non-compulsory[str]
tool_args: Dict[str, Any]
reasoning: str
@dataclass
class Plan:
activity: 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're {bp.id.title}, model {bp.id.model}.
{bp.id.description}
## Your Objectives:
{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 activity, 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, activity: str, reminiscence: MemoryManager) -> Plan:
system_prompt = self._build_planner_prompt()
messages = reminiscence.get_messages(system_prompt)
messages.append({"function": "person", "content material":
f"Create a plan to finish this activity: {activity}"})
resp = self.consumer.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()
information = json.masses(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 information["steps"]
]
return Plan(activity=activity, 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_clientWe implement the planning system that transforms a person activity right into a structured execution plan composed of a number of steps. We design a planner that instructs the language mannequin to supply a JSON plan containing reasoning, software choice, and arguments for every step. This planning layer permits the agent to interrupt advanced issues into smaller executable actions earlier than performing them.
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.activity}")
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}")
attempt:
if step.software and step.software != "null":
if verbose:
console.print(f" 🔧 Device: [green]{step.software}[/] | 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" ✅ Consequence: {output}")
else:
context_text = "n".be part of(
f"Step {r.step_id} end result: {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.id.title}. "
f"{self.blueprint.id.description}. "
f"Constraints: {'; '.be part of(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.decisions[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 part of(
f"Step {r.step_id} ({'✅' if r.success else '❌'}): {r.output[:300]}"
for r in outcomes
)
synthesis_prompt = (
f"Unique activity: {plan.activity}nn"
f"Step outcomes:n{steps_summary}nn"
f"Present a transparent, full last reply. Combine all step outcomes."
)
sys_prompt = (
f"You're {self.blueprint.id.title}. "
+ ("All the time present your reasoning. " if self.blueprint.validation.require_reasoning else "")
+ f"Objectives: {'; '.be part of(self.blueprint.objectives)}"
)
messages = reminiscence.get_messages(sys_prompt)
messages.append({"function": "person", "content material": synthesis_prompt})
resp = self.consumer.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: 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, activity: str,
use_llm_check: bool = False) -> ValidationResult:
points = []
v = self.blueprint.validation
if len(reply) < v.min_response_length:
points.append(f"Response too brief: {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.lengthen(self._llm_quality_check(reply, activity))
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, activity: str) -> Listing[str]:
immediate = (
f"Activity: {activity}nnAnswer: {reply[:500]}nn"
f'Does this reply handle the duty? Reply JSON: {{"on_topic": true/false, "challenge": "..."}}'
)
attempt:
resp = self.consumer.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("```","")
information = json.masses(uncooked)
if not information.get("on_topic", True):
return [f"LLM quality check: {data.get('issue', 'off-topic')}"]
besides Exception:
cross
return []We construct the executor and validation logic that really performs the steps generated by the planner. We implement a system that may both name registered instruments or carry out reasoning via the language mannequin, relying on the step definition. We additionally add a validator that checks the ultimate response in opposition to blueprint constraints equivalent to minimal size, reasoning necessities, and forbidden phrases.
@dataclass
class AgentResponse:
agent_name: str
activity: 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, activity: str, verbose: bool = True) -> AgentResponse:
bp = self.blueprint
if verbose:
console.print(Panel(
f"[bold]Agent:[/] {bp.id.title} v{bp.id.model}n"
f"[bold]Activity:[/] {activity}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", activity)
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...[/]")
attempt:
plan = self.planner.plan(activity, 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]Device:[/] {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, activity)
if verbose:
standing = "[green]PASSED[/]" if validation.handed else "[red]FAILED[/]"
console.print(f" Validation: {standing} | Rating: {validation.rating:.2f}")
for challenge in validation.points:
console.print(f" ⚠️ {challenge}")
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"🎯 Closing Reply — {bp.id.title}",
border_style="inexperienced"
))
return AgentResponse(
agent_name=bp.id.title, activity=activity,
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(
activity=(
"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, consumer)
analyst_engine.run(
activity=(
"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 one of the best and worst months, "
"and calculate development from first to final month."
)
)
print("n" + "="*60)
print("PORTABILITY DEMO: Similar activity → 2 completely different blueprints")
print("="*60)
SHARED_TASK = "Calculate 15% of two,500 and inform me the end 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("Legitimate?", width=7)
desk.add_column("Rating", width=6)
desk.add_column("Reply 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.rating:.2f}",
r.final_answer[:140] + "..."
)
console.print(desk)We assemble the runtime engine that orchestrates planning, execution, reminiscence updates, and validation into a whole autonomous workflow. We run a number of demonstrations displaying how completely different blueprints produce completely different behaviors whereas utilizing the identical core structure. Lastly, we illustrate blueprint portability by working the identical activity throughout two brokers and evaluating their outcomes.
In conclusion, we created a totally practical Auton-style runtime system that integrates cognitive blueprints, software registries, reminiscence administration, planning, execution, and validation right into a cohesive framework. We demonstrated how completely different brokers can share the identical underlying structure whereas behaving otherwise via custom-made blueprints, highlighting the design’s flexibility and energy. Via this implementation, we not solely explored how fashionable runtime brokers function but additionally constructed a powerful basis that we will lengthen additional with richer instruments, stronger reminiscence methods, and extra superior autonomous behaviors.
Try the Full Codes right here and Associated Paper. Additionally, be happy to comply with us on Twitter and don’t overlook to hitch our 120k+ ML SubReddit and Subscribe to our Publication. Wait! are you on telegram? now you may be part of us on telegram as effectively.
