Building Next-Gen Agentic AI: A Complete Framework for Cognitive Blueprint Driven Runtime Agents with Memory Tools and Validation

import json, yaml, time, math, textwrap, datetime, getpass, os
from typing import Any, Callable, Dict, List, Optional
from dataclasses import dataclass, field
from enum import Enum

from openai import OpenAI
from pydantic import BaseModel
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from rich.tree import Tree

try:
from google.colab import userdata
OPENAI_API_KEY = userdata.get(‘OPENAI_API_KEY’)
except Exception:
OPENAI_API_KEY = getpass.getpass(“🔑 Enter your OpenAI API key: “)

os.environ[“OPENAI_API_KEY”] = OPENAI_API_KEY
client = 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):
name: str
version: str = “1.0.0”
description: str
author: str = “unknown”

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

class BlueprintPlanning(BaseModel):
strategy: 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: List[str] = []

class CognitiveBlueprint(BaseModel):
identity: BlueprintIdentity
goals: List[str]
constraints: List[str] = []
tools: List[str] = []
memory: 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 = “””
identity:
name: ResearchBot
version: 1.2.0
description: Answers research questions using calculation and reasoning
author: Auton Framework Demo
goals:
– Answer user questions accurately using available tools
– Show step-by-step reasoning for all answers
– Cite the method used for each calculation
constraints:
– Never fabricate numbers or statistics
– Always validate mathematical results before reporting
– Do not answer questions outside your tool capabilities
tools:
– calculator
– unit_converter
– date_calculator
– search_wikipedia_stub
memory:
type: episodic
window_size: 12
summarize_after: 30
planning:
strategy: 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 = “””
identity:
name: DataAnalystBot
version: 2.0.0
description: Performs statistical analysis and data summarization
author: Auton Framework Demo
goals:
– Compute descriptive statistics for given data
– Identify trends and anomalies
– Present findings clearly with numbers
constraints:
– Only work with numerical data
– Always report uncertainty when sample size is small (< 5 items)
tools:
– calculator
– statistics_engine
– list_sorter
memory:
type: short_term
window_size: 6
planning:
strategy: hierarchical
max_steps: 10
max_retries: 3
think_before_acting: true
validation:
require_reasoning: true
min_response_length: 30
forbidden_phrases: []
“””

@dataclass
class ToolSpec:
name: str
description: str
parameters: Dict[str, str]
function: Callable
returns: str

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

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

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

def call(self, name: str, **kwargs) -> Any:
spec = self._tools.get(name)
if not spec:
raise ValueError(f”Tool ‘{name}’ not found in registry”)
return spec.function(**kwargs)

def get_tool_descriptions(self, allowed: List[str]) -> str:
lines = []
for name in allowed:
spec = self._tools.get(name)
if spec:
params = “, “.join(f”{k}: {v}” for k, v in spec.parameters.items())
lines.append(
f”• {spec.name}({params})\n”
f” → {spec.description}\n”
f” Returns: {spec.returns}”
)
return “\n”.join(lines)

def list_tools(self) -> List[str]:
return list(self._tools.keys())

registry = ToolRegistry()

@registry.register(
name=”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:
try:
allowed = {k: v for k, v in math.__dict__.items() if not k.startswith(“_”)}
allowed.update({“abs”: abs, “round”: round, “pow”: pow})
return str(eval(expression, {“__builtins__”: {}}, allowed))
except Exception as e:
return f”Error: {e}”

@registry.register(
name=”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(value: 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.lower(), to_unit.lower())
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(
name=”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:
try:
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”:
result = d1 + datetime.timedelta(days=int(date2))
return f”{result.strftime(‘%Y-%m-%d’)} (added {date2} days to {date1})”
return f”Unknown operation: {operation}”
except Exception as e:
return f”Error: {e}”

@registry.register(
name=”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(topic: 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.items():
if key in topic.lower():
return val
return f”No stub found for ‘{topic}’. In production, this would query Wikipedia’s API.”

@registry.register(
name=”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:
try:
nums = [float(x.strip()) for x in numbers.split(“,”)]
n = len(nums)
mean = 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 – mean) ** 2 for x in nums) / n)
return json.dumps({
“count”: n, “mean”: round(mean, 4), “median”: round(median, 4),
“std_dev”: round(std_dev, 4), “min”: min(nums),
“max”: max(nums), “range”: max(nums) – min(nums)
}, indent=2)
except Exception as e:
return f”Error: {e}”

@registry.register(
name=”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.sort(reverse=(order == “desc”))
return “, “.join(str(n) for n in nums)

@dataclass
class MemoryEntry:
role: str
content: str
timestamp: float = field(default_factory=time.time)
metadata: Dict = field(default_factory=dict)

class MemoryManager:
def __init__(self, config: BlueprintMemory, llm_client: OpenAI):
self.config = config
self.client = llm_client
self._history: List[MemoryEntry] = []
self._summary: str = “”

def add(self, role: str, content: str, metadata: Dict = None):
self._history.append(MemoryEntry(role=role, content=content, metadata=metadata or {}))
if (self.config.type == 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:]
text = “\n”.join(f”{e.role}: {e.content[:200]}” for e in to_compress)
try:
resp = self.client.chat.completions.create(
model=”gpt-4o-mini”,
messages=[{“role”: “user”, “content”:
f”Summarize this conversation history in 3 sentences:\n{text}”}],
max_tokens=150
)
self._summary += ” ” + resp.choices[0].message.content.strip()
except Exception:
self._summary += f” [compressed {len(to_compress)} messages]”

def get_messages(self, system_prompt: str) -> List[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.role if entry.role != “tool” else “assistant”,
“content”: entry.content
})
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
tool: Optional[str]
tool_args: Dict[str, Any]
reasoning: str

@dataclass
class Plan:
task: str
steps: List[PlanStep]
strategy: PlanningStrategy

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

def _build_planner_prompt(self) -> str:
bp = self.blueprint
return textwrap.dedent(f”””
You are {bp.identity.name}, version {bp.identity.version}.
{bp.identity.description}

## Your Goals:
{chr(10).join(f’ – {g}’ for g in bp.goals)}

## Your Constraints:
{chr(10).join(f’ – {c}’ for c in bp.constraints)}

## Available Tools:
{self.registry.get_tool_descriptions(bp.tools)}

## Planning Strategy: {bp.planning.strategy}
## Max Steps: {bp.planning.max_steps}

Given a user task, produce a JSON execution plan with this exact structure:
{{
“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”
}}
]
}}

Rules:
– Only use tools listed above
– Set tool to null for pure reasoning steps
– Keep steps <= {bp.planning.max_steps}
– Return ONLY valid JSON, no markdown fences
{bp.system_prompt_extra}
“””).strip()

def plan(self, task: str, memory: MemoryManager) -> Plan:
system_prompt = self._build_planner_prompt()
messages = memory.get_messages(system_prompt)
messages.append({“role”: “user”, “content”:
f”Create a plan to complete this task: {task}”})
resp = self.client.chat.completions.create(
model=”gpt-4o-mini”, messages=messages,
max_tokens=1200, temperature=0.2
)
raw = resp.choices[0].message.content.strip()
raw = raw.replace(““`json”, “”).replace(““`”, “”).strip()
data = json.loads(raw)
steps = [
PlanStep(
step_id=s[“step_id”], description=s[“description”],
tool=s.get(“tool”), tool_args=s.get(“tool_args”, {}),
reasoning=s.get(“reasoning”, “”)
)
for s in data[“steps”]
]
return Plan(task=task, steps=steps, strategy=self.blueprint.planning.strategy)

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

@dataclass
class ExecutionTrace:
plan: Plan
results: List[StepResult]
final_answer: str

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

def execute_plan(self, plan: Plan, memory: MemoryManager,
verbose: bool = True) -> ExecutionTrace:
results: List[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}”)
try:
if step.tool and step.tool != “null”:
if verbose:
console.print(f” 🔧 Tool: [green]{step.tool}[/] | Args: {step.tool_args}”)
output = self.registry.call(step.tool, **step.tool_args)
result = StepResult(step.step_id, True, str(output), step.tool)
if verbose:
console.print(f” ✅ Result: {output}”)
else:
context_text = “\n”.join(
f”Step {r.step_id} result: {r.output}” for r in results)
prompt = (
f”Previous results:\n{context_text}\n\n”
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.client.chat.completions.create(
model=”gpt-4o-mini”,
messages=[
{“role”: “system”, “content”: sys_prompt},
{“role”: “user”, “content”: prompt}
],
max_tokens=500, temperature=0.3
)
output = resp.choices[0].message.content.strip()
result = StepResult(step.step_id, True, output, None)
if verbose:
preview = output[:120] + “…” if len(output) > 120 else output
console.print(f” 🤔 Reasoning: {preview}”)
except Exception as e:
result = StepResult(step.step_id, False, “”, step.tool, str(e))
if verbose:
console.print(f” ❌ Error: {e}”)
results.append(result)

final_answer = self._synthesize(plan, results, memory)
return ExecutionTrace(plan=plan, results=results, final_answer=final_answer)

def _synthesize(self, plan: Plan, results: List[StepResult],
memory: MemoryManager) -> str:
steps_summary = “\n”.join(
f”Step {r.step_id} ({‘✅’ if r.success else ‘❌’}): {r.output[:300]}”
for r in results
)
synthesis_prompt = (
f”Original task: {plan.task}\n\n”
f”Step results:\n{steps_summary}\n\n”
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 = memory.get_messages(sys_prompt)
messages.append({“role”: “user”, “content”: synthesis_prompt})
resp = self.client.chat.completions.create(
model=”gpt-4o-mini”, messages=messages,
max_tokens=600, temperature=0.3
)
return resp.choices[0].message.content.strip()

@dataclass
class ValidationResult:
passed: bool
issues: List[str]
score: float

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

def validate(self, answer: str, task: str,
use_llm_check: bool = False) -> ValidationResult:
issues = []
v = self.blueprint.validation

if len(answer) < v.min_response_length:
issues.append(f”Response too short: {len(answer)} chars (min: {v.min_response_length})”)

answer_lower = answer.lower()
for phrase in v.forbidden_phrases:
if phrase.lower() in answer_lower:
issues.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):
issues.append(“Response lacks visible reasoning or explanation”)

if use_llm_check:
issues.extend(self._llm_quality_check(answer, task))

return ValidationResult(passed=len(issues) == 0,
issues=issues,
score=max(0.0, 1.0 – len(issues) * 0.25))

def _llm_quality_check(self, answer: str, task: str) -> List[str]:
prompt = (
f”Task: {task}\n\nAnswer: {answer[:500]}\n\n”
f’Does this answer address the task? Reply JSON: {{“on_topic”: true/false, “issue”: “…”}}’
)
try:
resp = self.client.chat.completions.create(
model=”gpt-4o-mini”,
messages=[{“role”: “user”, “content”: prompt}],
max_tokens=100
)
raw = resp.choices[0].message.content.strip().replace(““`json”,””).replace(““`”,””)
data = json.loads(raw)
if not data.get(“on_topic”, True):
return [f”LLM quality check: {data.get(‘issue’, ‘off-topic’)}”]
except Exception:
pass
return []

@dataclass
class AgentResponse:
agent_name: str
task: str
final_answer: str
trace: ExecutionTrace
validation: ValidationResult
retries: int
total_steps: int

class RuntimeEngine:
def __init__(self, blueprint: CognitiveBlueprint,
registry: ToolRegistry, llm_client: OpenAI):
self.blueprint = blueprint
self.memory = MemoryManager(blueprint.memory, 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, task: 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.memory.add(“user”, task)
retries, trace, validation = 0, None, None

for attempt in range(bp.planning.max_retries + 1):
if attempt > 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…[/]”)
try:
plan = self.planner.plan(task, self.memory)
if verbose:
tree = Tree(f”[bold]Plan ({len(plan.steps)} steps)[/]”)
for s in plan.steps:
icon = “🔧” if s.tool else “🤔”
branch = tree.add(f”{icon} Step {s.step_id}: {s.description}”)
if s.tool:
branch.add(f”[green]Tool:[/] {s.tool}”)
branch.add(f”[yellow]Args:[/] {s.tool_args}”)
console.print(tree)
except Exception as e:
if verbose: console.print(f”[red]Planning failed:[/] {e}”)
break

if verbose:
console.print(“\n[bold magenta]⚡ Phase 2: Executing…[/]”)
trace = self.executor.execute_plan(plan, self.memory, verbose=verbose)

if verbose:
console.print(“\n[bold magenta]✅ Phase 3: Validating…[/]”)
validation = self.validator.validate(trace.final_answer, task)

if verbose:
status = “[green]PASSED[/]” if validation.passed else “[red]FAILED[/]”
console.print(f” Validation: {status} | Score: {validation.score:.2f}”)
for issue in validation.issues:
console.print(f” ⚠️ {issue}”)

if validation.passed:
break

retries += 1
self.memory.add(“assistant”, trace.final_answer)
self.memory.add(“user”,
f”Your previous answer had issues: {‘; ‘.join(validation.issues)}. ”
f”Please improve.”
)

if trace:
self.memory.add(“assistant”, trace.final_answer)

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

return AgentResponse(
agent_name=bp.identity.name, task=task,
final_answer=trace.final_answer if trace else “”,
trace=trace, validation=validation,
retries=retries,
total_steps=len(trace.results) if trace else 0
)

def reset_memory(self):
self.memory.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, client)
research_engine.run(
task=(
“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, client)
analyst_engine.run(
task=(
“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 name, yaml_str in [
(“ResearchBot”, RESEARCH_AGENT_YAML),
(“DataAnalystBot”, DATA_ANALYST_YAML),
]:
eng = build_engine(yaml_str, registry, client)
responses[name] = eng.run(SHARED_TASK, verbose=False)

table = Table(title=”🔄 Blueprint Portability”, show_header=True, show_lines=True)
table.add_column(“Agent”, style=”cyan”, width=18)
table.add_column(“Steps”, style=”yellow”, width=6)
table.add_column(“Valid?”, width=7)
table.add_column(“Score”, width=6)
table.add_column(“Answer Preview”, width=55)

for name, r in responses.items():
table.add_row(
name, str(r.total_steps),
“✅” if r.validation.passed else “❌”,
f”{r.validation.score:.2f}”,
r.final_answer[:140] + “…”
)
console.print(table)