By Gopi Krishna Tummala

1. From “Parrot” to “Doer”

Imagine you walk into a room and there’s a parrot. A very, very clever parrot. You say “two times three” and it squawks “six.” You say “What’s the capital of France?” and it goes “Paris!” Perfectly. You are impressed.

That’s where large language models were two years ago: the world’s most over-educated parrot.

Now imagine the same parrot suddenly grows arms, legs, and an attitude. You say, “I’m hungry, make me dinner,” and instead of just repeating recipes, it walks to the fridge, grabs eggs, finds a pan, turns on the stove, and twenty minutes later hands you an omelette while muttering, “Next time buy better cheese.”

That’s the jump we just made. We stopped building parrots and started building little artificial people — tiny, tireless, slightly clumsy assistants that actually do things in the real world.

We call them agentic AI.

And the craziest part? We didn’t give them new brains. We just gave the old parrot some tools, a bit of memory, and taught it to shut up and think before squawking.

These machines can now:

1. Interpret goals — understand what you actually want

2. Break them into substeps — figure out the steps to get there

3. Choose tools — pick the right hammer for the right nail

4. Execute actions — actually do the thing

5. Observe the world — see what happened

6. Self-correct — try again when they mess up

Think of the jump from calculator → spreadsheet: one computes, the other coordinates.

Agentic AI is that jump for LLMs. Let’s see how that actually works.

2. The Simplest Possible Picture

Picture yourself trying to find where you left your keys this morning.

You don’t just stand there yelling “KEYS!” like an idiot. You do three things over and over:

1. Think - “I had coffee, so maybe the kitchen counter?”

2. Act - walk to the kitchen

3. Look - oh, there they are next to the mug. Done.

   (Or: nope, not there. Back to step 1.)

That tiny loop — Think → Act → Look → repeat — is literally the first “design pattern” of every serious agent today. The paper that wrote it down in 2022 just called it ReAct (Reason + Act).

It looks stupidly simple. It is. And yet 90% of the magic comes from getting this loop right.

In one sentence: Agentic AI = LLM + memory + reasoning + tool use + feedback loops.

In math-speak (which we’ll translate immediately):

Translation: “What should I do next, given everything I’ve seen so far, what I’m trying to achieve, and what I remember?”

Where:

• $o_{\le t}$ = everything the agent has observed up to now
• $g$ = the goal (what you asked for)
• $M$ = memory (what it remembers from before)
• $a_t$ = the action it picks (think, use a tool, write code, etc.)

This transforms LLMs from static predictors (parrots that just say words) into sequential decision-makers (assistants that actually do stuff).

3. Why Does It Still Screw Up So Often?

Because it’s basically a super-smart toddler with a credit card and internet access.

Toddlers do three things wrong that agents copy perfectly:

1. They forget what they were doing five minutes ago (context window problem)

2. They get distracted by shiny objects (goal drift)

3. They keep touching the hot stove even after burning their hand (infinite loops)

Agent systems fail in repeatable ways:

• Hallucinations — making stuff up confidently
• Infinite loops — doing the same wrong thing forever
• Wrong tools — using a hammer when you need a screwdriver
• Missing steps — skipping crucial parts
• Inconsistent memory — forgetting important details
• Fragile prompts — breaking when you change one word

The same way software engineering evolved its own patterns (like “always validate input” or “use try-catch blocks”), agent systems now have emerging design patterns—architectural principles that reduce chaos.

So we had to invent toddler-proofing for AIs:

• Sticky notes that say “REMEMBER: WE ARE BOOKING A FLIGHT, NOT RESEARCHING CRYPTO”
• A babysitter (the Supervisor) who watches and says “No, put down the flamethrower.”
• A budget tracker that yanks the credit card when the bill hits $2.

All the fancy code blocks you’ll see? They’re just fancy versions of toddler gates.

Let’s explore the important ones.

Core Agentic AI Design Patterns (2024-2025)

Pattern #1 — The ReAct Loop (Reason + Act + Observe)

This is the “hello world” design of LLM agents. It’s the key-finding loop, but for everything.

The Loop:

Translation: Think → Do → See what happened → Think again.

ReAct Loop Flow:

flowchart TD
    Start([User Query]) --> Thought[Generate Thought]
    Thought --> Action[Execute Action]
    Action --> Observation[Observe Result]
    Observation --> Check{Task Complete?}
    Check -->|No| Thought
    Check -->|Yes| End([Final Answer])
    
    style Thought fill:#e1f5ff
    style Action fill:#fff4e1
    style Observation fill:#e8f5e9

Example:

Here’s what it looks like inside the machine:

Thought: User wants cheapest flight to Austin.

Thought: I need prices, so I should search.

Action: call the flight-search tool with "Austin"

Observation: here are 50 flights, cheapest is Delta $187

Thought: That seems good, but let me double-check Southwest…

Action: call tool again with Southwest filter

Observation: actually $174!

Thought: Cool, that's the winner.

Final Answer: Book the Southwest flight for $174.

It’s just the key-finding loop, but with airplane tickets instead of keys.

Strengths: You can see exactly what it’s thinking (transparency). You can stop it if it goes wrong (controllability).

Weakness: Sometimes it talks too much and overthinks simple things. Like a teenager narrating every thought out loud.

Implementation:

Modern frameworks implement ReAct with a simple interface:

from langchain.agents import AgentExecutor, create_react_agent
from langchain_openai import ChatOpenAI

# Initialize the agent with tools
llm = ChatOpenAI(model="gpt-4")
tools = [search_tool, calculator_tool, code_executor]

# Create ReAct agent
agent = create_react_agent(llm, tools)
agent_executor = AgentExecutor(agent=agent, max_iterations=10)

# Run the agent
result = agent_executor.invoke({
    "input": "Book me the cheapest flight to Austin"
})

The agent automatically alternates between reasoning (generating thoughts) and acting (calling tools) until it reaches a final answer or hits the iteration limit.

Citation:

Yao et al. (2023). “ReAct: Synergizing Reasoning and Acting in Language Models.” arXiv:2210.03629

Pattern #2 — Plan-Execute-Reflect (PER)

Sometimes the omelette is more complicated: you want a three-course dinner for six people.

You don’t want one frantic parrot running around the kitchen. You want:

• One calm chef who writes the full menu and timeline on a whiteboard (the Planner)

• Several cooks who actually chop onions and stir sauce (Executors)

• One annoying food critic who tastes everything and yells “THIS SOUP HAS NO SOUL!” (Reflector)

Only when the critic finally says “Okay, I guess this is edible” do you serve the guests.

This is the second big pattern: Plan → Execute → Reflect, repeated until the critic shuts up.

It sounds overkill, but it stops the agent from serving raw chicken because it got excited and skipped steps.

How It Works:

Formally:

Plan-Execute-Reflect Flow:

flowchart TD
    Start([Goal]) --> Planner[Planner LLM<br/>Create Sub-tasks]
    Planner --> Executor[Executor LLM<br/>Perform Tasks]
    Executor --> Reflector[Reflector LLM<br/>Critique Results]
    Reflector --> Check{Goal Satisfied?}
    Check -->|No| Executor
    Check -->|Yes| End([Final Output])
    
    style Planner fill:#e1f5ff
    style Executor fill:#fff4e1
    style Reflector fill:#f3e5f5

Implementation:

LangGraph’s StateGraph provides a clean abstraction for PER:

from langgraph.graph import StateGraph, END
from typing import TypedDict

class AgentState(TypedDict):
    goal: str
    plan: list[str]
    results: dict
    reflection: str

def planner(state: AgentState) -> AgentState:
    """Break goal into sub-tasks"""
    plan = llm.invoke(f"Create a plan for: {state['goal']}")
    return {"plan": parse_plan(plan)}

def executor(state: AgentState) -> AgentState:
    """Execute each task in the plan"""
    results = {}
    for task in state['plan']:
        results[task] = execute_task(task)
    return {"results": results}

def reflector(state: AgentState) -> AgentState:
    """Critique results against original goal"""
    reflection = llm.invoke(
        f"Goal: {state['goal']}\n"
        f"Results: {state['results']}\n"
        "Does this satisfy the goal? What's missing?"
    )
    return {"reflection": reflection}

# Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("planner", planner)
workflow.add_node("executor", executor)
workflow.add_node("reflector", reflector)
workflow.set_entry_point("planner")
workflow.add_edge("planner", "executor")
workflow.add_edge("executor", "reflector")
workflow.add_conditional_edges("reflector", 
    lambda x: "executor" if needs_rework(x) else END)

Modern frameworks like OpenAI’s Swarm, LangGraph, and Instructor patterns use this.

Citation:

Shinn et al. (2023). “Reflexion: Language Agents with Verbal Reinforcement Learning.” arXiv:2303.11366

Pattern #3 — Tools Are Just Extra Hands

Imagine you’re a carpenter with no arms. Someone glues a hammer, a saw, and a drill to long sticks and says “Here, use these.”

At first you wave the sticks around like a drunk octopus. After a while you learn exactly when to pick the hammer-stick versus the saw-stick.

That’s what “tool use” is for an AI. The tools are:

• Google search
• a calculator
• your email inbox
• a code-running sandbox
• the mouse and keyboard of your computer

The agent doesn’t have them built in — they’re just extra hands it can grab when needed.

The clever part: modern agents don’t wait for you to say “use the calculator.” They decide themselves, the same way you don’t ask permission to pick up a hammer when you see a nail.

In math-speak:

Translation: “Which tool is most useful right now?” The agent picks the best one automatically.

Tools include:

• Web search

• Code execution

• Databases

• Embeddings

• Email/calendar

• Vision models

• Simulators

Implementation:

Using Pydantic/Instructor pattern ensures type-safe tool calls:

from pydantic import BaseModel
from instructor import patch

class SearchTool(BaseModel):
    """Search the web for information"""
    query: str
    max_results: int = 5

class CalculatorTool(BaseModel):
    """Evaluate a mathematical expression"""
    expression: str

# Patch the LLM to use structured outputs
client = patch(ChatOpenAI())
tools = [SearchTool, CalculatorTool]

# The LLM automatically chooses and calls tools
response = client.chat.completions.create(
    model="gpt-4",
    response_format=SearchTool,  # Enforces schema
    messages=[{"role": "user", "content": "Find recent papers on agentic AI"}]
)

# Type-safe tool call
search_result = response.parsed  # Automatically a SearchTool instance

This is crucial for enterprise AI.

Citation:

Schick et al. (2023). “Toolformer: Language Models Can Teach Themselves to Use Tools.” arXiv:2302.04761

Tool Router Pattern (Simpler Alternative to RL-Based Tool Use)

Note: Full PPO training on tool policies is rarely used in production (2025). Instead, most teams use a simple Tool Router—a lightweight classifier that decides which tool/specialist to call.

Why Tool Router Instead of RL:

• Simplicity: Easy to implement and debug
• Fast: No training required, works immediately
• Production-Ready: Used in 90%+ of deployments
• Effective: 85-90% accuracy for tool selection

Implementation:

from sklearn.ensemble import RandomForestClassifier
import numpy as np

class ToolRouter:
    """Lightweight classifier for tool selection"""
    def __init__(self):
        self.classifier = RandomForestClassifier(n_estimators=100)
        self.tool_features = {}  # Pre-computed tool embeddings
    
    def route(self, query: str, available_tools: list) -> str:
        """Route query to best tool"""
        # Extract features from query
        features = self.extract_features(query)
        
        # Predict best tool
        tool_scores = {}
        for tool in available_tools:
            score = self.classifier.predict_proba([features])[0]
            tool_scores[tool] = score
        
        return max(tool_scores, key=tool_scores.get)
    
    def extract_features(self, query: str) -> np.array:
        """Extract features for routing"""
        # Simple keyword-based features (can use embeddings)
        features = [
            "search" in query.lower(),
            "calculate" in query.lower(),
            "code" in query.lower(),
            "database" in query.lower(),
            len(query),
            query.count("?"),
        ]
        return np.array(features)

For production: Use this simple router instead of full RL training. It’s faster, cheaper, and easier to maintain.

Pattern #4 — Self-Consistency Sampling

You know how when you’re not sure about something, you ask multiple people and go with what most of them say? That’s this pattern.

Instead of picking a single chain of thought, agents generate many and vote.

Translation: “Ask 10 experts instead of 1, then pick the most common answer.”

Why This Works:

Implementation:

This is an efficient form of Monte Carlo Tree Search (MCTS) applied to the thought space:

def self_consistency_sampling(prompt: str, n_samples: int = 10):
    """Generate multiple reasoning paths and vote"""
    samples = []
    
    for _ in range(n_samples):
        # Generate a chain of thought
        response = llm.generate(
            prompt,
            temperature=0.7,  # Higher temp = more diversity
            max_tokens=500
        )
        samples.append(extract_answer(response))
    
    # Return the most common answer
    from collections import Counter
    return Counter(samples).most_common(1)[0][0]

# Example: Math problem
question = "Solve: 2x + 5 = 15"
answer = self_consistency_sampling(question, n_samples=10)
# If 8/10 samples say x=5, that's the answer

Used in:

• Math reasoning

• Code generation

• Planning

• Safety filtering

Citation:

Wang et al. (2022). “Self-Consistency Improves Chain of Thought Reasoning in Language Models.” arXiv:2203.11171

Pattern #5 — Graph-of-Thoughts (GoT)

You know how when you’re brainstorming, you don’t think in a straight line? You have one idea, it branches into three, two of those merge back together, you cross out the bad one, and keep going.

That’s Graph-of-Thoughts. Instead of linear reasoning (A → B → C), thoughts form a graph:

• Parallel branches — explore multiple ideas at once
• Merging — combine good ideas from different paths
• Pruning — cut off dead ends
• Revising — go back and fix earlier thoughts

The Analogy:

Brainstorming on a whiteboard with arrows and clusters. Like a mind map, but the AI actually uses it to solve problems.

Graph-of-Thoughts Structure:

graph TD
    Root[Initial Thought] --> Branch1[Thought Branch 1]
    Root --> Branch2[Thought Branch 2]
    Root --> Branch3[Thought Branch 3]
    Branch1 --> Merge1[Merge Point]
    Branch2 --> Merge1
    Branch3 --> Prune1[Pruned Branch]
    Merge1 --> Final[Final Thought]
    
    style Root fill:#e1f5ff
    style Merge1 fill:#fff4e1
    style Final fill:#e8f5e9
    style Prune1 fill:#ffebee,stroke-dasharray: 5 5

Implementation:

GoT allows parallel exploration and merging of reasoning paths:

from typing import List, Set

class ThoughtNode:
    def __init__(self, content: str, score: float):
        self.content = content
        self.score = score
        self.children: List[ThoughtNode] = []
        self.parents: List[ThoughtNode] = []

def graph_of_thoughts(initial_prompt: str, max_depth: int = 3):
    """Build a graph of reasoning paths"""
    root = ThoughtNode(initial_prompt, 0.0)
    frontier = [root]
    
    for depth in range(max_depth):
        # Expand all nodes in parallel
        new_nodes = []
        for node in frontier:
            # Generate multiple child thoughts
            children = generate_thoughts(node.content, n=3)
            for child_content, score in children:
                child = ThoughtNode(child_content, score)
                child.parents.append(node)
                node.children.append(child)
                new_nodes.append(child)
        
        # Merge similar thoughts
        merged = merge_similar_thoughts(new_nodes)
        # Prune low-scoring branches
        frontier = [n for n in merged if n.score > threshold]
    
    # Extract best path
    return extract_best_path(root)

Modern agent frameworks now implement explicit DAG-based reasoning.

Citation:

Besta et al. (2023). “Graph of Thoughts: Solving Elaborate Problems with Large Language Models.” arXiv:2308.09687

Pattern #12 — Search-Augmented Agents (When Thoughts Become a Tree)

Instead of single ReAct trajectories or self-consistency voting, modern agents use systematic search over the space of possible reasoning paths.

Analogy:

Like a chess engine exploring possible moves, but for reasoning and planning.

MCTS Search Tree:

graph TD
    Root[Root State] --> N1[Node 1]
    Root --> N2[Node 2]
    Root --> N3[Node 3]
    N1 --> N4[Node 4]
    N1 --> N5[Node 5]
    N2 --> N6[Node 6]
    N3 --> N7[Node 7]
    N4 --> Best[Best Path]
    N5 --> Best
    
    style Root fill:#e1f5ff
    style Best fill:#e8f5e9

Formal Structure:

Each search node represents:

Where:

• $s$ = current state
• $a$ = proposed action
• $r$ = predicted reward
• $s'$ = predicted next state
• $\text{thought}$ = LLM-generated reasoning

Implementation:

LLM-MCTS (Monte-Carlo Tree Search with LLM guidance):

from collections import defaultdict
import numpy as np

class SearchNode:
    def __init__(self, state, thought, action=None):
        self.state = state
        self.thought = thought
        self.action = action
        self.visits = 0
        self.value = 0.0
        self.children = []
        self.parent = None

def llm_mcts(root_state, goal, max_iterations=100):
    """Monte-Carlo Tree Search guided by LLM"""
    root = SearchNode(root_state, llm.generate_thought(root_state, goal))
    
    for _ in range(max_iterations):
        # Selection: traverse to leaf
        node = root
        while node.children:
            node = select_best_child(node)  # UCB1 or similar
        
        # Expansion: LLM proposes actions
        if not is_terminal(node.state):
            actions = llm.propose_actions(node.state, goal)
            for action in actions:
                next_state = world_model.predict(node.state, action)
                child = SearchNode(next_state, 
                    llm.generate_thought(next_state, goal), action)
                child.parent = node
                node.children.append(child)
        
        # Simulation: roll out to terminal
        value = simulate_rollout(node, goal)
        
        # Backpropagation: update values
        backpropagate(node, value)
    
    # Return best action
    return max(root.children, key=lambda n: n.value / n.visits).action

def world_model_predict(state, action):
    """Learned or LLM-based world model"""
    # Can be a neural network, physics simulator, or LLM
    return model.predict(state, action)

def simulate_rollout(node, goal, max_depth=10):
    """Roll out trajectory and evaluate"""
    current = node
    for depth in range(max_depth):
        if is_terminal(current.state):
            return evaluate(current.state, goal)
        action = llm.sample_action(current.state, goal)
        current = world_model_predict(current.state, action)
    return evaluate(current.state, goal)

Key Advantages:

• Systematic exploration vs. single-shot reasoning
• World model integration for accurate predictions
• Optimal path selection via search
• Handles long-horizon planning better than ReAct alone

Used In:

• Robotics (manipulation, navigation)
• Theorem proving
• Autonomous coding (multi-file refactoring)
• Web navigation (complex multi-step tasks)
• Game playing (Chess, Go, strategy games)

Citation:

Recent work combining LLMs with MCTS and learned world models (2024-2025)

Pattern #6 — Memory-Attentive Agents

You know how you have different kinds of memory? Your phone number (long-term), what you had for breakfast (short-term), and that thing you’re actively thinking about right now (working memory)?

Agents need the same thing. Memory is not an afterthought anymore.

Types of Memory:

1. Short-Term Scratchpads — like sticky notes on your desk (local CoT)

2. Long-Term Semantic Memory — like a filing cabinet organized by topic (vector DBs)

3. Episodic Memory — like a diary of everything that happened (conversation history)

4. Working Memory — like the stuff you’re actively juggling in your head right now (structured key-value)

Formally:

Memory retrieval becomes:

Implementation:

Multi-memory system with retrieval:

from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from typing import Dict, List

class AgentMemory:
    def __init__(self):
        self.episodic: List[Dict] = []  # Conversation history
        self.semantic = Chroma(embedding_function=OpenAIEmbeddings())  # Vector DB
        self.working: Dict = {}  # Structured key-value
    
    def retrieve(self, query: str, k: int = 5) -> List[str]:
        """Retrieve from all memory types"""
        results = []
        
        # Semantic search
        semantic_results = self.semantic.similarity_search(query, k=k)
        results.extend([r.page_content for r in semantic_results])
        
        # Episodic search (recent context)
        recent = self.episodic[-10:]  # Last 10 turns
        results.extend([turn['content'] for turn in recent])
        
        # Working memory (exact matches)
        if query in self.working:
            results.append(self.working[query])
        
        return results[:k]
    
    def store(self, content: str, memory_type: str = "semantic"):
        """Store in appropriate memory"""
        if memory_type == "semantic":
            self.semantic.add_texts([content])
        elif memory_type == "episodic":
            self.episodic.append({"content": content, "timestamp": time.time()})
        elif memory_type == "working":
            # Extract key-value pairs
            self.working.update(extract_kv(content))

The Analogy:

A scientist with notebooks (long-term memory), lab records (episodic memory), and mental reminders (working memory). All working together.

Citation:

Park et al. (2023). “Generative Agents: Interactive Simulacra of Human Behavior.” arXiv:2304.03442

Pattern #16 — Memory Rewriting (Compression as Control)

You know how your phone storage fills up and you have to delete old photos? Agents have the same problem, but worse.

Long-lived agents must actively manage memory: compress, rewrite, and discard irrelevant information to avoid “memory rot” and context overflow.

The Problem:

Imagine you’re trying to remember everything that happened in the last month. Your brain would explode. Agents have the same issue:

As agents operate over days/weeks:

• Memory grows unbounded (like a hoarder’s house)
• Old information becomes stale (like expired milk in the fridge)
• Context windows fill up (like a backpack that won’t zip)
• Irrelevant facts obscure important ones (like trying to find your keys in a messy room)

Solution: Active Memory Management

Memory evolves via compression:

Where:

• $M_t$ = current memory state
• $\Delta_t$ = new observations
• $g$ = current goal (for relevance filtering)

Implementation:

class MemoryCompressor:
    def __init__(self, max_memory_size: int = 1000):
        self.max_size = max_memory_size
        self.memory: List[Dict] = []
    
    def compress(self, new_observations: List[str], goal: str):
        """Compress and rewrite memory"""
        # Add new observations
        for obs in new_observations:
            self.memory.append({
                "content": obs,
                "timestamp": time.time(),
                "relevance": self.compute_relevance(obs, goal)
            })
        
        # If over limit, compress
        if len(self.memory) > self.max_size:
            self.memory = self.rewrite_memory(goal)
    
    def compute_relevance(self, memory_item: str, goal: str) -> float:
        """Score memory relevance to current goal"""
        return llm.invoke(
            f"Memory: {memory_item}\n"
            f"Goal: {goal}\n"
            "Rate relevance 0-1:"
        )
    
    def rewrite_memory(self, goal: str) -> List[Dict]:
        """Compress memory by summarizing and pruning"""
        # Group by topic
        topics = self.cluster_memories(self.memory)
        
        compressed = []
        for topic, memories in topics.items():
            # Summarize related memories
            summary = llm.invoke(
                f"Summarize these related memories:\n{memories}\n"
                f"Keep only facts relevant to: {goal}"
            )
            compressed.append({
                "content": summary,
                "timestamp": max(m['timestamp'] for m in memories),
                "relevance": max(m['relevance'] for m in memories)
            })
        
        # Keep only top-K by relevance
        compressed.sort(key=lambda x: x['relevance'], reverse=True)
        return compressed[:self.max_size]
    
    def forget_irrelevant(self, threshold: float = 0.3):
        """Remove low-relevance memories"""
        self.memory = [
            m for m in self.memory 
            if m['relevance'] > threshold
        ]

Memory Rewriting Strategies:

1. Temporal Compression: Merge similar events over time
2. Goal-Based Pruning: Remove memories irrelevant to current goal
3. Semantic Summarization: Replace detailed memories with summaries
4. Forgetting Curves: Decay old memories unless frequently accessed

Citation:

Recent work on memory management in long-lived agents (2024-2025)

Pattern #7 — Environment Loop (Agents That Touch the Real World)

Agents interact with:

• browsers

• OS

• file system

• cloud

• emails

• Slack

• simulators

Using a step function:

This is classic reinforcement learning but driven by LLM policies.

Example:

• Autonomous coding agents

• Browsing agents

• Automated analysts

• DevOps agents

Citation:

Zhou et al. (2023). “WebArena: A Realistic Web Environment for Building Autonomous Agents.” arXiv:2307.13854

Pattern #14 — 3D Grounded Agents (Scene Graph → Plan → Act)

Modern agents operating in physical 3D spaces build scene graphs to reason about spatial relationships, affordances, and constraints.

The 3D Challenge:

Unlike 2D web interfaces, 3D environments require:

• Spatial reasoning (reachability, visibility, occlusion)
• Object affordances (what actions are possible)
• Scene understanding (relationships between objects)
• Long-horizon manipulation (multi-step physical tasks)

Scene Graph Representation:

Agents construct a graph:

Where:

• $V$ = nodes (objects, locations, agents)
• $E$ = edges (spatial relationships: on, near, inside, occludes)

3D Scene Graph Example:

graph TD
    Cup[Cup] -->|on| Table[Table]
    Robot[Robot Arm] -->|near| Table
    Table -->|supports| Cup
    Cup -->|contains| Water[Water]
    Robot -.->|can reach| Cup
    
    style Cup fill:#e1f5ff
    style Table fill:#fff4e1
    style Robot fill:#e8f5e9
    style Water fill:#f3e5f5

Implementation:

from dataclasses import dataclass
from typing import List, Dict

@dataclass
class SceneObject:
    id: str
    type: str  # "cup", "table", "robot_arm"
    position: tuple[float, float, float]
    affordances: List[str]  # ["graspable", "movable"]

@dataclass
class SpatialRelation:
    subject: str
    relation: str  # "on", "near", "inside", "occludes"
    object: str
    confidence: float

class SceneGraph:
    def __init__(self):
        self.objects: Dict[str, SceneObject] = {}
        self.relations: List[SpatialRelation] = []
    
    def build_from_observation(self, observation: Dict):
        """Construct scene graph from 3D observation"""
        # Extract objects (from vision model or 3D detector)
        objects = self.detect_objects(observation)
        
        # Extract spatial relations
        relations = self.detect_relations(objects)
        
        self.objects = {obj.id: obj for obj in objects}
        self.relations = relations
    
    def plan_action(self, goal: str) -> List[str]:
        """Plan actions using scene graph constraints"""
        # Check reachability
        reachable = self.check_reachability()
        
        # Check affordances
        valid_actions = self.filter_by_affordances(goal)
        
        # Generate plan with spatial constraints
        plan = llm.invoke(
            f"Goal: {goal}\n"
            f"Scene Graph: {self.to_text()}\n"
            f"Reachable objects: {reachable}\n"
            f"Valid actions: {valid_actions}\n"
            "Generate a step-by-step plan:"
        )
        return parse_plan(plan)
    
    def check_reachability(self) -> List[str]:
        """Check which objects are reachable by agent"""
        reachable = []
        agent_pos = self.objects["robot_arm"].position
        
        for obj_id, obj in self.objects.items():
            distance = euclidean_distance(agent_pos, obj.position)
            if distance < REACH_THRESHOLD:
                reachable.append(obj_id)
        
        return reachable
    
    def filter_by_affordances(self, goal: str) -> List[str]:
        """Filter actions based on object affordances"""
        required_affordance = extract_affordance(goal)  # e.g., "graspable"
        
        valid_objects = [
            obj_id for obj_id, obj in self.objects.items()
            if required_affordance in obj.affordances
        ]
        
        return valid_objects
    
    def to_text(self) -> str:
        """Convert graph to text for LLM prompt"""
        obj_desc = [f"{o.type} at {o.position}" for o in self.objects.values()]
        rel_desc = [f"{r.subject} is {r.relation} {r.object}" for r in self.relations]
        return f"Objects: {'; '.join(obj_desc)}. Relations: {'; '.join(rel_desc)}."

# Example: The LLM receives the text description and plans:
# "Objects: cup at (0.5, 0.1, 0.9); table at (0, 0, 0). Relations: cup is on table."
# Goal: "Pick up the cup."
# Plan: "1. Move arm to (0.5, 0.1, 0.9). 2. Grasp cup. 3. Lift."

3D Planning with Constraints:

Agents reason about:

• Reachability: Can the robot arm reach object X?
• Visibility: Is object Y occluded by object Z?
• Affordances: Can object W be grasped/moved?
• Collisions: Will action A cause a collision?

Applications:

• Home robotics (cleaning, organizing, cooking)
• Factory automation (assembly, quality control)
• Autonomous vehicles (navigation, manipulation)
• VR/AR agents (interactive virtual environments)

Citation:

Recent CVPR 2024-2025 work on 3D scene understanding and grounded agents

Pattern #6 — Supervisor / Orchestrator (The #1 Production Pattern in 2025)

Remember the chef-cooks-critic pattern? This is like that, but the chef is also a traffic controller.

This is the single most deployed pattern in 2025 production systems.

Instead of having all your agents talk to each other in a chaotic free-for-all (which costs a fortune and breaks constantly), you have one smart supervisor who says:

“Hey, this is a research task. Researcher, you handle it.” “Okay, this needs code. Coder, your turn.” “Wait, that looks wrong. Critic, check this.”

The supervisor routes tasks to the right specialist, reducing token costs by 40-70% and dramatically improving reliability.

It’s like having a restaurant manager instead of letting all the chefs just yell at each other.

Why It Matters:

• Cost Efficiency: Supervisor routes to the right specialist, avoiding unnecessary agent calls
• Reliability: Centralized error handling and retry logic
• Scalability: Easy to add new specialist agents without changing the core flow
• Production-Ready: Used in OpenAI Swarm, CrewAI, LangGraph, Azure Agent Factory

Architecture:

flowchart TD
    Start([User Request]) --> Supervisor[Supervisor Agent<br/>Routes to Specialist]
    Supervisor -->|Research Task| Researcher[Researcher Agent]
    Supervisor -->|Code Task| Coder[Coder Agent]
    Supervisor -->|Write Task| Writer[Writer Agent]
    Supervisor -->|Review Task| Critic[Critic Agent]
    Researcher --> Tools[Tool Node]
    Coder --> Tools
    Writer --> Tools
    Critic --> Supervisor
    Tools --> Supervisor
    Supervisor --> Check{Task Complete?}
    Check -->|No| Supervisor
    Check -->|Yes| End([Final Output])
    
    style Supervisor fill:#e1f5ff
    style Researcher fill:#fff4e1
    style Coder fill:#e8f5e9
    style Writer fill:#f3e5f5
    style Critic fill:#fce4ec

Implementation (LangGraph 0.2+ Style — 2025 Standard):

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from typing import Literal

class AgentState(TypedDict):
    messages: list
    next_agent: str

def supervisor_agent(state: AgentState) -> AgentState:
    """Supervisor decides which specialist to call"""
    last_message = state["messages"][-1]
    
    # Simple routing logic (can be LLM-based)
    if "research" in last_message.content.lower():
        return {"next_agent": "researcher"}
    elif "code" in last_message.content.lower() or "function" in last_message.content.lower():
        return {"next_agent": "coder"}
    elif "write" in last_message.content.lower() or "draft" in last_message.content.lower():
        return {"next_agent": "writer"}
    else:
        return {"next_agent": "researcher"}  # Default

def route_next_agent(state: AgentState) -> Literal["researcher", "coder", "writer", "tools", END]:
    """Route to the next agent based on supervisor decision"""
    if state.get("next_agent") == "researcher":
        return "researcher"
    elif state.get("next_agent") == "coder":
        return "coder"
    elif state.get("next_agent") == "writer":
        return "writer"
    elif state.get("next_agent") == "tools":
        return "tools"
    else:
        return END

# Build the workflow
workflow = StateGraph(AgentState)

# Add nodes
workflow.add_node("supervisor", supervisor_agent)
workflow.add_node("researcher", researcher_agent)
workflow.add_node("coder", coder_agent)
workflow.add_node("writer", writer_agent)
workflow.add_node("tools", ToolNode([search_tool, code_executor]))

# Add edges
workflow.add_edge(START, "supervisor")
workflow.add_conditional_edges("supervisor", route_next_agent)  # ← Key line
workflow.add_edge("researcher", "tools")
workflow.add_edge("coder", "tools")
workflow.add_edge("tools", "supervisor")  # Return to supervisor
workflow.add_edge("writer", END)

app = workflow.compile()

Cost Comparison:

The Supervisor pattern reduces costs by 40-70% by avoiding unnecessary agent-to-agent communication.

Production Best Practices:

1. Use SLMs for specialists (Llama-3.2-8B, Qwen2.5-14B) when possible
2. Supervisor can be lightweight (even a simple classifier works)
3. Add budget guardrails at the supervisor level
4. Implement retry logic in the supervisor, not individual agents

Citation:

Wu et al. (2023). “AutoGen: Enabling Next-Gen LLM Applications via Multi-Agent Conversation.” arXiv:2308.08155

Pattern #7 — Parallel Tool Use / Fan-Out (2025 Standard)

You know how when you’re cooking, you don’t wait for the water to boil before you start chopping vegetables? You do things in parallel.

Modern LLMs (Claude 3.5/4, GPT-4o-2025-08, Grok-3) support parallel tool calls natively. This single change delivers 60-80% latency reduction for multi-tool tasks.

The Old Way (Sequential):

Like doing laundry one sock at a time:

# Old way (sequential)
result1 = await llm.call_tool(search_tool, query1)  # 200ms
result2 = await llm.call_tool(db_tool, query2)      # 150ms
result3 = await llm.call_tool(calc_tool, expr)     # 100ms
# Total: 450ms

The Solution:

Parallel tool calls:

# New way (parallel) - 2025 standard
tools_to_call = [
    (search_tool, query1),
    (db_tool, query2),
    (calc_tool, expr)
]
results = await llm.parallel_tool_call(tools_to_call)  # One round-trip: 200ms
# Total: 200ms (60% faster)

Implementation:

from openai import AsyncOpenAI

client = AsyncOpenAI()

async def parallel_tool_execution(prompt: str, tools: list):
    """Execute multiple tools in parallel"""
    response = await client.chat.completions.create(
        model="gpt-4o-2025-08",
        messages=[{"role": "user", "content": prompt}],
        tools=tools,
        tool_choice="auto"  # Model decides which tools to call
    )
    
    # Model returns multiple tool calls in one response
    tool_calls = response.choices[0].message.tool_calls
    
    # Execute all tools in parallel
    import asyncio
    results = await asyncio.gather(*[
        execute_tool(call.function.name, call.function.arguments)
        for call in tool_calls
    ])
    
    return results

Latency Comparison:

When to Use:

• ✅ Multiple independent tool calls
• ✅ Tool calls that don’t depend on each other
• ✅ When latency is critical
• ❌ When tools depend on previous results (use sequential)

Citation:

Native support in Claude 3.5+, GPT-4o-2025-08+, Grok-3 (2025)

Pattern #8 — The Big Secret Nobody Says Out Loud (2025 Edition)

Here’s the part that will sound like science fiction, but it’s actually shipping today:

The very best agents in 2025 barely talk out loud anymore.

Old way (2023-2024):

Thought → Action → Thought → Action → blab blab blab

Like a teenager narrating every thought: “I’m opening the fridge… now I’m looking for milk…”

New way (late 2025):

[silent for 15 seconds, burning 30,000 invisible “thinking tokens”]

“Here’s your perfectly booked flight + hotel + restaurant reservation. You’re welcome.”

Like an adult who just quietly makes you a sandwich.

The biggest shift in late 2024-mid 2025: reasoning is increasingly hidden inside the model (o1, Claude “thinking”, Grok-3 “reasoning tokens”, Gemini 2.5 Flash Thinking). This has made many explicit ReAct loops obsolete for medium-difficulty tasks.

Before (2023-2024): Explicit reasoning in prompts

Thought: I need to search for flights
Action: search_flights
Observation: [results]
Thought: Sort by price
Action: filter_by_price

Now (2025): Hidden reasoning inside the model

<thinking>
[Model internally reasons for 10k-50k tokens]
</thinking>
<answer>
The cheapest flight to Austin is...
</answer>

We call it hidden reasoning or “test-time compute scaling.”

Normal people call it “finally shutting up and thinking.”

OpenAI’s o1, Claude’s thinking mode, Grok-3’s reasoning tokens — they all do this now.

And honestly? For hard problems it destroys every loud, ReAct-style agent we built before.

Why This Matters:

• Simpler prompts: No need for explicit ReAct structure
• Better reasoning: Model can spend 10-100× more tokens thinking
• Cost-aware: Cap reasoning tokens at 16k-32k for hard tasks
• Production-ready: o1-preview achieves 96% reliability vs 72% for pure ReAct

Implementation:

from openai import OpenAI

client = OpenAI()

def hidden_reasoning_agent(prompt: str, max_reasoning_tokens: int = 32000):
    """Use model's internal reasoning capability"""
    response = client.chat.completions.create(
        model="o1-preview",  # or "claude-3-5-sonnet-20241022"
        messages=[{"role": "user", "content": prompt}],
        max_tokens=max_reasoning_tokens,  # Allow extensive reasoning
        # Model handles reasoning internally
    )
    
    # For models that expose reasoning tokens
    if hasattr(response, 'reasoning_tokens'):
        print(f"Reasoning tokens used: {response.reasoning_tokens}")
    
    return response.choices[0].message.content

Cost & Performance Comparison:

When to Use Hidden Reasoning:

• ✅ Complex multi-step problems
• ✅ Tasks requiring deep reasoning
• ✅ When reliability > cost
• ❌ Simple tool-calling tasks (use ReAct)
• ❌ When latency is critical (use parallel tools)

Production Tip:

For cost-sensitive applications, use a hybrid approach:

1. Try hidden reasoning for hard tasks
2. Fall back to ReAct for simple tasks
3. Use SLMs with hidden reasoning for 80% of cases

Citation:

OpenAI o1, Anthropic Claude thinking mode, xAI Grok-3 (2024-2025)

Pattern #13 — Multi-Agent Societies (Specialists + Protocols)

You wouldn’t ask a single person to run a startup. You need a CEO, a coder, a marketer, and an accountant. Agentic AI is moving to the same model: societies of specialized LLMs.

The Analogy:

A modern microservices architecture, but instead of software components, the services are specialized LLMs. Slack is their bloodstream.

Common Specializations:

Communication Protocol:

Where $\pi_i$ is the policy of agent $i$ that generates a structured message $m$ intended for agent $j$ based on their shared observation $o$ and the global goal $g$.

Implementation:

Frameworks like AutoGen and CrewAI implement this by defining specific roles and termination criteria.

from autogen import UserProxyAgent, AssistantAgent, GroupChat, GroupChatManager

# 1. Define Agents (Roles)
researcher = AssistantAgent(
    name="Researcher",
    system_message="Synthesizes web search results."
)

coder = AssistantAgent(
    name="Coder",
    system_message="Writes and executes Python code in a sandbox."
)

user_proxy = UserProxyAgent(name="Client", human_input_mode="NEVER")

# 2. Define the Protocol (GroupChat)
groupchat = GroupChat(
    agents=[user_proxy, researcher, coder],
    messages=[],
    max_round=12,
    speaker_selection_method="auto"  # LLM decides who talks next
)

manager = GroupChatManager(groupchat=groupchat)

# 3. Execute the Multi-Agent Mission
user_proxy.initiate_chat(
    manager,
    message="Find the latest stock price for NVDA and calculate its P/E ratio."
)

# The manager orchestrates Researcher (to get price) and Coder (to calculate P/E).

Modern examples:

• AutoGen
• OpenAI Swarm
• CrewAI
• LangGraph agent clusters

Citation:

Wu et al. (2023). “AutoGen: Enabling Next-Gen LLM Applications via Multi-Agent Conversation.” arXiv:2308.08155

Pattern #17 — Compensatory Reflexes (Error Recovery)

When the agent touches the hot stove (an API call fails, a runtime error occurs, a key file is missing), it needs an instant, pre-programmed response.

The Reflex:

The reflex policy $\pi_{\text{reflex}}$ overrides the main reasoning loop temporarily.

Example Errors and Reflexes:

This dramatically improves reliability from the $\sim 50\%$ success rate of brittle scripts to the $\sim 95\%$ required for enterprise use.

Implementation:

Error handling with automatic retry and correction:

def execute_with_reflex(action: str, max_retries: int = 3):
    """Execute action with automatic error recovery"""
    for attempt in range(max_retries):
        try:
            result = execute_action(action)
            return result
        except ToolError as e:
            if attempt < max_retries - 1:
                # Reflex: ask LLM to fix the error
                corrected_action = llm.invoke(
                    f"Action failed: {action}\n"
                    f"Error: {e}\n"
                    "Generate a corrected action:"
                )
                action = parse_action(corrected_action)
            else:
                raise
        except ValidationError as e:
            # Pydantic validation failed - fix schema
            fixed_schema = llm.invoke(
                f"Fix this tool call to match the schema:\n{action}\n"
                f"Schema error: {e}"
            )
            action = fixed_schema

Analogy:

Touch a hot stove → pull hand instantly.

This dramatically improves reliability.

Citation:

Madaan et al. (2023). “Self-Refine: Iterative Refinement with Self-Feedback.” arXiv:2303.17651

Pattern #18 — Introspective Agents (Self-Debugging AI)

Agents must be their own critics, checking their output against constraints before delivery. This is the unit test for thoughts.

Self-Evaluation Loop:

If $\text{score} < \text{threshold}$, the agent is sent back to the ReAct loop with the failure message as the new observation.

Example:

1. Thought: “I will use this two-year-old API documentation.”

2. Action: $\text{call\_api}$

3. Observation: $\text{API\_ERROR}$

4. Self-Evaluation: $\pi_{\text{eval}}$ checks the output against the constraint: Was an error received? Yes. $\text{score} = 0.2$.

5. New Thought: “The API call failed. I will search for current documentation.”

This is the refinement that turns a brittle planner into a resilient problem-solver.

Implementation:

Self-evaluation with automatic correction:

def introspective_agent(prompt: str, constraints: List[str]):
    """Agent that evaluates and corrects its own output"""
    max_iterations = 5
    
    for iteration in range(max_iterations):
        # Generate output
        output = llm.generate(prompt)
        
        # Self-evaluate
        evaluation = llm.invoke(
            f"Output: {output}\n"
            f"Constraints: {constraints}\n"
            "Rate this output 0-10 and explain why:"
        )
        score = extract_score(evaluation)
        
        if score >= 8.0:
            return output
        
        # Self-correct
        prompt = llm.invoke(
            f"Original prompt: {prompt}\n"
            f"Previous output: {output}\n"
            f"Evaluation: {evaluation}\n"
            "Generate an improved version:"
        )
    
    return output  # Return best attempt

Example:

Chain-of-Thought → Self-Check → Correction.

This resembles unit tests for thoughts.

Citation:

Shinn et al. (2023). “Reflexion: Language Agents with Verbal Reinforcement Learning.” arXiv:2303.11366

Pattern #11 — Embodied Agent Loops (LLMs + Perception + Control)

Remember the parrot that grew arms? This is that, but for robots.

Modern embodied agents operate via closed-loop control in physical environments, combining high-level natural language planning with low-level motor control.

The Embodied Challenge:

Unlike digital agents (which live in computers), embodied agents live in the real world. They have to:

• Perceive the world through vision/sensors (like eyes and touch)
• Ground language to physical actions (understand “pick up the cup” means moving a robot arm)
• Control motors, grippers, joints (actually move things)
• Compose skills to achieve long-horizon goals (pick up cup → pour water → hand it to you)

It’s like teaching a robot to make coffee. The high-level plan (“make coffee”) gets broken down into low-level commands (“move arm 5cm left, close gripper, lift 10cm”).

Formal Structure:

Where:

• $s_t$ = physical state (joint angles, positions, velocities)
• $a_t$ = motor commands (torques, velocities)
• $o_{t+1}$ = sensor observations (RGB, depth, proprioception)
• $f_{\text{dynamics}}$ = physics simulator or real-world dynamics

Embodied Agent Architecture:

flowchart TD
    Goal([Natural Language Goal]) --> LLM[High-Level Planner<br/>LLM]
    LLM --> SkillLib[Skill Library<br/>Learned Skills]
    SkillLib --> Policy[Low-Level Policy<br/>Neural Network]
    Policy --> Robot[Robot<br/>Motor Commands]
    Robot --> Sensors[Sensors<br/>Vision, Proprioception]
    Sensors --> Perception[Perception<br/>Scene Understanding]
    Perception --> LLM
    
    style Goal fill:#e1f5ff
    style LLM fill:#fff4e1
    style SkillLib fill:#e8f5e9
    style Policy fill:#f3e5f5
    style Robot fill:#fce4ec
    style Sensors fill:#fff9c4
    style Perception fill:#e0f2f1

Implementation:

class EmbodiedAgent:
    def __init__(self):
        self.llm_planner = ChatOpenAI(model="gpt-4")
        self.vision_model = load_vision_model()  # e.g., OpenVLA
        self.skill_library = SkillLibrary()
        self.low_level_policy = LowLevelPolicy()
    
    def step(self, observation, goal: str):
        """Execute one step in the environment"""
        # 1. Perceive
        scene_description = self.vision_model.describe(observation)
        
        # 2. Plan (high-level)
        plan = self.llm_planner.invoke(
            f"Goal: {goal}\n"
            f"Current scene: {scene_description}\n"
            "What skill should I execute next?"
        )
        skill_name = extract_skill(plan)
        
        # 3. Execute skill (low-level)
        skill = self.skill_library.get_skill(skill_name)
        motor_commands = self.low_level_policy.execute(
            skill, observation
        )
        
        # 4. Act
        next_obs, reward, done = self.env.step(motor_commands)
        
        return next_obs, reward, done
    
    def compose_skills(self, goal: str) -> List[str]:
        """Chain atomic skills for long-horizon tasks"""
        # LLM breaks goal into skill sequence
        plan = self.llm_planner.invoke(
            f"Goal: {goal}\n"
            f"Available skills: {self.skill_library.list_skills()}\n"
            "Generate a sequence of skills:"
        )
        return parse_skill_sequence(plan)

class SkillLibrary:
    """Library of learned manipulation skills"""
    def __init__(self):
        self.skills = {
            "grasp": self.load_skill("grasp.pkl"),
            "place": self.load_skill("place.pkl"),
            "push": self.load_skill("push.pkl"),
            "pour": self.load_skill("pour.pkl"),
        }
    
    def get_skill(self, name: str):
        """Retrieve a learned skill policy"""
        return self.skills[name]

Key Components:

1. Vision-Language Models: Ground natural language to visual observations
2. Skill Learning: Learn atomic actions from demonstrations (imitation learning)
3. Skill Composition: Chain skills for complex tasks
4. Low-Level Control: Convert high-level plans to motor commands

Applications:

• Home robotics (cleaning, cooking, organizing)
• Factory automation (assembly, quality control)
• Healthcare (assistive robots)
• Simulated environments (training in simulation, deploying in reality)

Citation:

Recent CVPR/ICRA/CoRL 2024-2025 work on embodied AI and vision-language-action models

Pattern #15 — Imagination Loop (Predict → Simulate → Select)

This is the ultimate self-correction mechanism: the ability to mentally rehearse a future before living it.

The Analogy:

Before a surgeon makes an incision, they don’t just guess. They study the medical model, visualize the outcome, and mentally rehearse the steps. Agents do this by incorporating a World Model—often a generative AI (like a Video-LLM or 3D Diffusion model).

The Loop:

1. LLM Proposes Action ($a_t$): “I think I should open the door.”

2. World Model Simulates Outcome ($o_{t+1}, s_{t+1}$): The model generates a short video or 3D simulation of the agent opening the door and seeing a monster.

3. Agent Selects/Revises Trajectory ($\tau^*$): “Wait, that resulted in a negative outcome (monster). I will revise the action to ‘Check keyhole with camera.’”

Formal Structure:

The agent selects an action based on the predicted future return $R$ over a trajectory $\tau$:

Where $R(\tau)$ is calculated by rolling out the trajectory using the internal world model.

Why it Matters:

This dramatically improves safety and planning quality in high-risk, real-world environments like robotics or autonomous driving, moving decision-making from reactive to proactive.

Imagination Loop Flow:

flowchart TD
    State[Current State] --> Propose[Propose Actions]
    Propose --> Sim1[Simulate Action 1]
    Propose --> Sim2[Simulate Action 2]
    Propose --> Sim3[Simulate Action 3]
    Sim1 --> Eval1[Evaluate Outcome 1]
    Sim2 --> Eval2[Evaluate Outcome 2]
    Sim3 --> Eval3[Evaluate Outcome 3]
    Eval1 --> Select[Select Best]
    Eval2 --> Select
    Eval3 --> Select
    Select --> Execute[Execute Action]
    Execute --> NextState[Next State]
    
    style State fill:#e1f5ff
    style Propose fill:#fff4e1
    style Sim1 fill:#e8f5e9
    style Sim2 fill:#e8f5e9
    style Sim3 fill:#e8f5e9
    style Select fill:#f3e5f5
    style Execute fill:#e8f5e9

Formal Structure:

Implementation:

class ImaginationAgent:
    def __init__(self):
        self.llm = ChatOpenAI()
        self.world_model = WorldModel()  # Video/3D diffusion model
        self.value_estimator = ValueNetwork()
    
    def imagine_and_select(self, state, goal: str):
        """Imagine futures and select best action"""
        # 1. LLM proposes candidate actions
        candidates = self.llm.invoke(
            f"State: {state}\n"
            f"Goal: {goal}\n"
            "Propose 5 possible next actions:"
        )
        actions = parse_actions(candidates)
        
        # 2. Simulate outcomes for each
        trajectories = []
        for action in actions:
            # World model predicts future
            future_states = self.world_model.rollout(state, action, horizon=10)
            
            # Estimate value
            value = self.value_estimator(future_states[-1], goal)
            
            trajectories.append({
                "action": action,
                "future": future_states,
                "value": value
            })
        
        # 3. Select best trajectory
        best = max(trajectories, key=lambda t: t["value"])
        
        return best["action"]
    
    def world_model_rollout(self, state, action, horizon: int):
        """Simulate future states"""
        states = [state]
        for _ in range(horizon):
            # Predict next state using world model
            next_state = self.world_model.predict(states[-1], action)
            states.append(next_state)
        return states

class WorldModel:
    """Generative model that predicts future observations"""
    def __init__(self):
        # Can be: video prediction model, 3D diffusion, physics simulator
        self.model = load_pretrained_world_model()
    
    def predict(self, state, action):
        """Predict next state given current state and action"""
        # Encode state (image, 3D scene, etc.)
        state_encoding = self.encode_state(state)
        
        # Encode action
        action_encoding = self.encode_action(action)
        
        # Predict next state
        next_state_encoding = self.model(state_encoding, action_encoding)
        
        # Decode to observation space
        return self.decode_state(next_state_encoding)
    
    def rollout(self, initial_state, action_sequence, horizon: int):
        """Roll out a sequence of actions"""
        states = [initial_state]
        for action in action_sequence[:horizon]:
            next_state = self.predict(states[-1], action)
            states.append(next_state)
        return states

World Model Types:

1. Video Prediction Models: Predict future frames
2. 3D Diffusion Models: Generate 3D scene futures
3. Physics Simulators: Accurate physics-based prediction
4. Learned Dynamics: Neural networks trained on experience

Benefits:

• Error Prevention: Catch bad actions before execution
• Better Planning: Evaluate multiple futures
• Sample Efficiency: Learn from imagined experiences
• Robustness: Handle unseen scenarios via simulation

Citation:

Recent NeurIPS/ICLR 2024-2025 work on world models and imagination in agents

Putting It All Together: A Modern Agent Stack (2025)

Architecture Diagram:

flowchart TD
    Goal([Goal]) --> Supervisor[Supervisor<br/>Pattern #6]
    Supervisor --> Budget[Budget Guardrail<br/>Stop if > $2]
    Budget --> Memory[Memory<br/>Pattern #5]
    Memory --> Specialist{Route to Specialist}
    Specialist -->|Research| Researcher[Researcher Agent<br/>SLM]
    Specialist -->|Code| Coder[Coder Agent<br/>SLM]
    Specialist -->|Write| Writer[Writer Agent<br/>SLM]
    Researcher --> ParallelTools[Parallel Tools<br/>Pattern #7]
    Coder --> ParallelTools
    Writer --> ParallelTools
    ParallelTools --> SelfEval[Self-Eval / Correction<br/>Pattern #2]
    SelfEval --> Verify[Verification & Safety]
    Verify --> LoopCheck{Loop Detected?}
    LoopCheck -->|Yes| Supervisor
    LoopCheck -->|No| Output([Final Output])
    
    Supervisor -.->|Feedback Loop| Supervisor
    
    style Goal fill:#e1f5ff
    style Supervisor fill:#fff4e1
    style Budget fill:#ffebee
    style Memory fill:#e8f5e9
    style Researcher fill:#f3e5f5
    style Coder fill:#f3e5f5
    style Writer fill:#f3e5f5
    style ParallelTools fill:#fce4ec
    style SelfEval fill:#fff9c4
    style Verify fill:#ffebee
    style Output fill:#e8f5e9

Component Flow:

1. Goal → Input objective
2. Planner (Pattern #2) → Breaks goal into sub-tasks
3. Memory (Pattern #6, #16) → Retrieves relevant context
4. ReAct Loop (Pattern #1) → Alternates reasoning and action
5. Tool Use (Pattern #3, #13) → Executes tools/code
6. Self-Evaluation (Pattern #10) → Checks and corrects
7. Verification → Multi-layer safety checks
8. Imagination (Pattern #15) → Optional future simulation

This stack can solve tasks like:

• doing full research

• writing + executing code

• debugging itself

• solving math proofs

• handling enterprise workflows

• orchestrating other agents

Modern systems (OpenAI’s Swarm, Anthropic’s agent API, LangGraph’s StateGraph) are formalizing this.

Where the Math Actually Lives

1. Action Selection:

2. Planning:

3. Self-Consistency:

4. Retrieval:

5. Reflection:

These mathematical abstractions map to practical code patterns across agent frameworks.

So What Are We Actually Building?

We’re not building better chatbots.

We’re building the first generation of machines that can be told “make me money,” “fix my bugs,” or “plan my vacation” and actually go do it while we take a nap.

It’s like giving a 12-year-old unlimited caffeine, perfect memory, and the internet — then being shocked when they either take over the world or burn the house down.

Our job right now is mostly the “don’t burn the house down” part.

But the exciting thing — the really exciting thing — is that we’re finally past the “wow it can write poetry” phase.

We’re in the “wow it can book the flight, negotiate the price, and expense it correctly without ever waking me up” phase.

That’s not a better parrot.

That’s the first baby step toward a machine that can run errands, write code, do research, maybe one day run companies or laboratories, while the humans do… whatever humans turn out to be good for.

And the wild part? All of this is built out of the same dumb next-word predictor we started with five years ago.

We just taught it to stop predicting words and start predicting what to do next.

That’s it.

That’s the whole trick.

What we’re really building:

• Software that rewrites itself
• Systems that reason before acting
• Tools that orchestrate other tools
• AI workflows that improve with time

The next OS won’t be Windows, macOS, or Linux.

It will be an agent runtime.

Pretty neat, huh?

Part II: The Agent’s Paradox — Common Failure Modes and Mitigation

Remember the toddler with a credit card? This section is about all the ways they can mess up, and how we stop them.

Just as software must account for concurrency and exceptions (like “what if two people try to buy the last item?”), agentic AI must anticipate these common, repeatable failures. Understanding failure modes is crucial for building production-ready agent systems.

The good news? They fail in predictable ways. The bad news? You have to plan for all of them.

A. The “Tool Overuse” Trap

Failure:

The agent defaults to calling a tool (e.g., web search) even when the answer is in its context or memory. This wastes tokens, latency, and API costs.

Example: User asks “What is 2+2?” and the agent calls a calculator tool instead of using its internal knowledge.

Mitigation (Pattern #3):

Implement a “Triage” Prompt—a meta-step before $\text{Thought}_t$ that explicitly asks the LLM to decide: Internal Knowledge vs. Tool Use.

def triage_step(query: str, context: str) -> str:
    """Decide if tool use is necessary"""
    decision = llm.invoke(
        f"Query: {query}\n"
        f"Context: {context}\n\n"
        "Can this be answered from context alone? "
        "Respond: INTERNAL or TOOL_NEEDED"
    )
    return decision

# Use before tool selection
if triage_step(user_query, agent_memory.retrieve(user_query)) == "INTERNAL":
    return llm.invoke(user_query)  # No tool call
else:
    return agent.select_and_call_tool(user_query)

B. The Contextual Amnesia Loop

Failure:

The LLM’s finite context window forces it to “forget” crucial observations from $t-5$ steps ago, leading to re-planning or repeating failed actions.

Example: Agent searches for flights, finds results, but 10 steps later forgets the search results and searches again.

Mitigation (Pattern #6):

Implement Structured Working Memory ($M_{work}$). Force the agent to distill the core findings of every $N$ steps into a structured JSON/YAML object that always gets injected into the next prompt.

class WorkingMemory:
    def __init__(self):
        self.facts: Dict[str, Any] = {}
        self.decisions: List[str] = []
    
    def distill_step(self, observations: List[str], step_num: int):
        """Compress observations into structured facts"""
        if step_num % 5 == 0:  # Every 5 steps
            summary = llm.invoke(
                f"Observations: {observations}\n"
                "Extract key facts as JSON: {fact: value, ...}"
            )
            self.facts.update(json.loads(summary))
    
    def inject_into_prompt(self, base_prompt: str) -> str:
        """Always include working memory in prompt"""
        memory_context = f"""
Working Memory:
{json.dumps(self.facts, indent=2)}

Recent Decisions:
{self.decisions[-3:]}
"""
        return f"{base_prompt}\n\n{memory_context}"

C. The Goal Drift Problem (The Agent’s “Shiny Object Syndrome”)

Failure:

The agent gets distracted by an interesting sub-problem and loses sight of the original, top-level goal ($g$).

Example: Goal is “Book a flight to Austin,” but agent gets sidetracked researching hotel prices and never books the flight.

Mitigation (Pattern #2):

Enforce the Plan-Execute-Reflect (PER) structure. The Planner’s output is immutable for $k$ steps. The Reflector’s primary job is to check the current output against the original $g$, not just the sub-task.

class GoalAwareReflector:
    def __init__(self, original_goal: str):
        self.original_goal = original_goal
        self.plan: List[str] = []
    
    def reflect(self, current_state: Dict) -> Dict:
        """Check if we're still aligned with original goal"""
        reflection = llm.invoke(
            f"Original Goal: {self.original_goal}\n"
            f"Current Plan: {self.plan}\n"
            f"Current State: {current_state}\n\n"
            "Are we still working toward the original goal? "
            "If not, what corrective action is needed?"
        )
        
        if "DRIFT_DETECTED" in reflection:
            # Reset to original goal
            return {"action": "reset_to_goal", "goal": self.original_goal}
        return {"action": "continue"}

D. The Hallucinated API Call

Failure:

The LLM invents a non-existent tool name or generates correct code with entirely fabricated function arguments.

Example: Agent calls search_flights_api(destination="Austin", date="2025-01-20") but the actual API requires to_city and departure_date.

Mitigation (Pattern #9):

Utilize Pydantic/Instructor Pattern for all tool calls. Force the tool-call LLM to output a JSON object strictly conforming to a defined schema. If the JSON parsing fails (a non-LLM error), trigger the Compensatory Reflex to generate a corrected JSON structure.

from pydantic import BaseModel, ValidationError
from instructor import patch

class FlightSearchTool(BaseModel):
    """Search for flights - schema enforced"""
    to_city: str  # Not "destination"!
    departure_date: str  # Format: YYYY-MM-DD
    from_city: str = "SFO"  # Default

client = patch(ChatOpenAI())

def safe_tool_call(user_request: str) -> FlightSearchTool:
    """Tool call with automatic schema correction"""
    max_retries = 3
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model="gpt-4",
                response_format=FlightSearchTool,
                messages=[{"role": "user", "content": user_request}]
            )
            return response.parsed  # Type-safe!
        except ValidationError as e:
            if attempt < max_retries - 1:
                # Reflex: fix the schema error
                user_request = f"{user_request}\n\n"
                f"Previous error: {e}\n"
                "Generate a corrected request matching the schema."
            else:
                raise

E. The Infinity Loop (The Circular Argument)

Failure:

The agent falls into a closed loop, e.g., $\text{Action}_1$ fails $\rightarrow \text{Thought}_2$ says “Try $\text{Action}_1$ again” $\rightarrow \text{Action}_1$ fails…

Example: Agent tries to call an API, gets 404, thinks “maybe the URL is wrong,” tries again with same URL, gets 404 again, repeats.

Mitigation:

Implement Episodic Memory Pruning and a Backtrack Limit. Store a hash of the last three actions/thoughts. If the current sequence matches a recent pattern, trigger a hard reflex action like $\text{Action}_{\text{backtrack}}$ or $\text{Action}_{\text{reset\_plan}}$.

import hashlib
from collections import deque

class LoopDetector:
    def __init__(self, window_size: int = 3):
        self.action_history: deque = deque(maxlen=window_size)
        self.seen_patterns: set = set()
    
    def detect_loop(self, current_action: str) -> bool:
        """Check if we're repeating a pattern"""
        # Add current action
        self.action_history.append(current_action)
        
        # Create pattern hash
        pattern = " -> ".join(self.action_history)
        pattern_hash = hashlib.md5(pattern.encode()).hexdigest()
        
        if pattern_hash in self.seen_patterns:
            return True  # Loop detected!
        
        self.seen_patterns.add(pattern_hash)
        return False
    
    def trigger_backtrack(self, agent_state: Dict) -> Dict:
        """Hard reset when loop detected"""
        return {
            "action": "backtrack",
            "reset_to_step": len(agent_state['history']) - 5,
            "new_strategy": "try_different_approach"
        }

# Usage in agent loop
detector = LoopDetector()
for step in agent_execution:
    if detector.detect_loop(step.action):
        # Break the loop
        agent_state = detector.trigger_backtrack(agent_state)
        break

F. The Premature Termination Problem

Failure:

The agent stops before completing all sub-tasks, thinking it’s done when it’s only partially complete.

Example: Goal is “Research and write a blog post about agentic AI.” Agent researches, writes introduction, then stops without writing the body or conclusion.

Mitigation:

Implement explicit Task Completion Verification using Pattern #10 (Introspective Agents). Before terminating, the agent must verify all planned sub-tasks are complete.

def verify_completion(original_goal: str, plan: List[str], results: Dict) -> bool:
    """Check if all planned tasks are actually complete"""
    verification = llm.invoke(
        f"Original Goal: {original_goal}\n"
        f"Plan: {plan}\n"
        f"Results: {results}\n\n"
        "Are ALL planned tasks complete? Respond: COMPLETE or INCOMPLETE"
    )
    return "COMPLETE" in verification

# In agent execution
if agent_thinks_done():
    if not verify_completion(goal, plan, results):
        # Continue execution
        agent_state['action'] = 'continue_plan'

G. Verifiable Agent Pipelines (Safety & Grounding)

Failure:

Agents produce outputs that are:

• Unverifiable (no way to check correctness)
• Ungrounded (hallucinated facts)
• Uncertain (no confidence scores)

Mitigation:

Implement multi-layer verification:

class VerifiableAgent:
    def __init__(self):
        self.llm = ChatOpenAI()
        self.code_executor = CodeExecutor()
        self.uncertainty_estimator = UncertaintyModel()
    
    def generate_with_verification(self, prompt: str):
        """Generate output with automatic verification"""
        # 1. Generate initial output
        output = self.llm.generate(prompt)
        
        # 2. Estimate uncertainty
        confidence = self.uncertainty_estimator(output, prompt)
        
        # 3. Self-check
        self_check = self.llm.invoke(
            f"Output: {output}\n"
            "Check for: hallucinations, contradictions, errors. "
            "Respond: VERIFIED or ISSUES_FOUND"
        )
        
        # 4. Code execution verification (if applicable)
        if contains_code(output):
            execution_result = self.code_executor.run(output)
            if execution_result.error:
                output = self.fix_code(output, execution_result.error)
        
        # 5. Tool-grounded cross-check
        if contains_facts(output):
            facts = extract_facts(output)
            verified_facts = self.verify_facts(facts)  # Use web search/tools
            output = replace_unverified_facts(output, verified_facts)
        
        return {
            "output": output,
            "confidence": confidence,
            "verified": self_check == "VERIFIED"
        }
    
    def verify_facts(self, facts: List[str]) -> Dict[str, bool]:
        """Cross-check facts using tools"""
        verified = {}
        for fact in facts:
            # Use web search to verify
            search_results = web_search(fact)
            verified[fact] = self.check_consistency(fact, search_results)
        return verified