AI Agents Explained — The 3am Incident That Broke Our Multi-Agent Orchestrator

import json
from openai import OpenAI
from pydantic import BaseModel, Field
from typing import Literal

# Define the structured output schema for the agent's response
class AgentAction(BaseModel):
    action: Literal["search_knowledge_base", "calculate", "final_answer"]
    action_input: str = Field(description="Input for the action")
    reasoning: str = Field(description="Why this action was chosen")

client = OpenAI()

# Tool implementations
def search_knowledge_base(query: str) -> str:
    # In production, this would call a real API
    return f"Results for '{query}': No relevant documents found."

def calculate(expression: str) -> str:
    try:
        return str(eval(expression))
    except Exception as e:
        return f"Error: {e}"

def execute_tool(action: AgentAction) -> str:
    if action.action == "search_knowledge_base":
        return search_knowledge_base(action.action_input)
    elif action.action == "calculate":
        return calculate(action.action_input)
    else:
        raise ValueError(f"Unknown action: {action.action}")

# Agent loop with bounded iterations
def run_agent(task: str, max_iterations: int = 10) -> str:
    messages = [{"role": "user", "content": task}]
    for i in range(max_iterations):
        response = client.beta.chat.completions.parse(
            model="gpt-4o",
            messages=messages,
            response_format=AgentAction,
        )
        action = response.choices[0].message.parsed
        if action.action == "final_answer":
            return action.action_input
        tool_result = execute_tool(action)
        messages.append({"role": "assistant", "content": f"Action: {action.action}\nInput: {action.action_input}\nReasoning: {action.reasoning}"})
        messages.append({"role": "tool", "content": tool_result, "tool_call_id": str(i)})
    return "Max iterations reached without final answer."

print(run_agent("What is 2 + 2?"))

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator

# Define the shared state
class AgentState(TypedDict):
    messages: Annotated[List[dict], operator.add]  # Reducer appends to list
    next_agent: str
    final_output: str

# Node functions
def triage_agent(state: AgentState) -> AgentState:
    # In production, this would call an LLM to classify the task
    if "math" in state["messages"][-1]["content"]:
        return {"next_agent": "math_agent"}
    else:
        return {"next_agent": "general_agent"}

def math_agent(state: AgentState) -> AgentState:
    # Simple math logic
    result = eval(state["messages"][-1]["content"])
    return {"messages": [{"role": "assistant", "content": f"Result: {result}"}], "final_output": str(result)}

def general_agent(state: AgentState) -> AgentState:
    return {"messages": [{"role": "assistant", "content": "I can't help with that yet."}], "final_output": "Unsupported"}

def router(state: AgentState) -> str:
    # Decide which node to go to next
    if state["next_agent"] == "math_agent":
        return "math_agent"
    elif state["next_agent"] == "general_agent":
        return "general_agent"
    else:
        return END

# Build the graph
builder = StateGraph(AgentState)
builder.add_node("triage_agent", triage_agent)
builder.add_node("math_agent", math_agent)
builder.add_node("general_agent", general_agent)
builder.set_entry_point("triage_agent")
builder.add_conditional_edges("triage_agent", router)
builder.add_edge("math_agent", END)
builder.add_edge("general_agent", END)
graph = builder.compile()

# Run the graph
result = graph.invoke({"messages": [{"role": "user", "content": "What is 3 * 7?"}], "next_agent": "", "final_output": ""})
print(result["final_output"])

# Bad: Using an agent for a deterministic decision
# This is over-engineered and fragile
import openai
response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Should we send a password reset email to user@example.com?"}]
)
# The LLM might say 'yes', 'no', or 'call them'

# Good: Use a simple rule
if user_exists and user_requested_reset:
    send_reset_email(user_email)
else:
    log_error("Invalid reset request")

import hashlib
import time
from functools import lru_cache

# Simple in-memory cache for tool results
@lru_cache(maxsize=1000)
def cached_search(query: str) -> str:
    # Simulate a slow API call
    time.sleep(0.5)
    return f"Result for {query}"

# Rate limiter using a token bucket
class RateLimiter:
    def __init__(self, max_calls: int, period: float):
        self.max_calls = max_calls
        self.period = period
        self.tokens = max_calls
        self.last_refill = time.monotonic()

    def acquire(self) -> bool:
        now = time.monotonic()
        elapsed = now - self.last_refill
        self.tokens = min(self.max_calls, self.tokens + elapsed * (self.max_calls / self.period))
        self.last_refill = now
        if self.tokens >= 1:
            self.tokens -= 1
            return True
        return False

limiter = RateLimiter(max_calls=10, period=1.0)  # 10 calls per second

def rate_limited_search(query: str) -> str:
    if not limiter.acquire():
        raise Exception("Rate limit exceeded. Retry later.")
    return cached_search(query)

# Production agent loop with observability
def run_agent_with_telemetry(task: str) -> str:
    start_time = time.monotonic()
    token_count = 0
    try:
        result = run_agent(task)  # From earlier example
        token_count = len(task) + len(result)  # Simplified
        return result
    finally:
        elapsed = time.monotonic() - start_time
        # In production, send to OpenTelemetry
        print(f"TRACE: task={task[:50]}, duration={elapsed:.2f}s, tokens={token_count}")

# Mistake 1: Not validating tool inputs
# Bad: Agent can call search with any input
def search(query: str):
    return requests.get(f"https://api.example.com/search?q={query}").json()  # SQL injection risk

# Good: Validate and sanitize
import re
def safe_search(query: str):
    sanitized = re.sub(r'[^a-zA-Z0-9 ]', '', query)
    return requests.get(f"https://api.example.com/search?q={sanitized}").json()

# Mistake 2: Not handling context window
# Bad: Appending all messages indefinitely
messages.append(tool_result)

# Good: Summarize or truncate old messages
def summarize_messages(messages: list, max_tokens: int = 3000):
    # In production, use an LLM to summarize
    if len(str(messages)) > max_tokens:
        return [{"role": "system", "content": f"Previous context: {messages[0]['content'][:100]}..."}]
    return messages

# RAG example: Simple and cheap
from openai import OpenAI
client = OpenAI()

def rag_query(query: str, context: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o-mini",  # Cheaper model
        messages=[
            {"role": "system", "content": f"You are a helpful assistant. Use this context to answer: {context}"},
            {"role": "user", "content": query}
        ]
    )
    return response.choices[0].message.content

# Agent example: More capable but expensive
def agent_query(query: str) -> str:
    # Agent loop with tool calls
    return run_agent(query)  # From earlier example

# When to use what:
# - RAG: "What is the capital of France?" (single-turn, factual)
# - Agent: "Book a flight to Paris for next Tuesday, then email me the confirmation" (multi-step, tool use)

import json
import time

# Trace decorator to log every step
def trace_step(func):
    def wrapper(*args, **kwargs):
        start = time.monotonic()
        result = func(*args, **kwargs)
        elapsed = time.monotonic() - start
        trace = {
            "function": func.__name__,
            "args": args,
            "kwargs": kwargs,
            "result": str(result)[:200],
            "duration": elapsed
        }
        print(f"TRACE: {json.dumps(trace)}")
        return result
    return wrapper

@trace_step
def llm_call(prompt: str) -> str:
    # Simulate LLM call
    time.sleep(0.5)
    return f"Response to: {prompt[:50]}"

@trace_step
def tool_call(name: str, input: str) -> str:
    # Simulate tool call
    time.sleep(0.2)
    return f"Result from {name}"

# Replay a session from a trace
with open("trace_123.json", "r") as f:
    trace_data = json.load(f)

for step in trace_data["steps"]:
    print(f"Step {step['step']}: {step['function']} called with {step['args']}")
    print(f"  Result: {step['result']}")
    print(f"  Duration: {step['duration']:.2f}s")

# io.thecodeforge
from typing import List, Dict
import json

class AgentMemory:
    def __init__(self, window_size: int = 10):
        self.short_term: List[Dict] = []  # sliding window
        self.long_term: Dict[str, List[float]] = {}  # user_id -> embedding
        self.window_size = window_size

    def add_interaction(self, role: str, content: str):
        self.short_term.append({"role": role, "content": content})
        if len(self.short_term) > self.window_size:
            self.short_term.pop(0)

    def get_context(self) -> str:
        conversation = ""
        for msg in self.short_term:
            conversation += f"{msg['role']}: {msg['content']}\n"
        return conversation

# io.thecodeforge
from pydantic import BaseModel
from typing import Literal

class CalculatorTool(BaseModel):
    name: Literal["calculator"] = "calculator"
    parameters: dict = {
        "type": "object",
        "properties": {
            "expression": {
                "type": "string",
                "description": "Mathematical expression to evaluate (e.g., '2 + 2')"
            }
        },
        "required": ["expression"]
    }
    
    class Config:
        extra = "forbid"  # prevents made-up parameters