Agentic AI Patterns
Overview
An agent is an LLM in a loop: observe → think → act → observe. The hard part isn't the LLM — it's the loop control, error recovery, and knowing when to stop.
When to Use
- Building AI agents that take actions (not just chat)
- Designing tool-use patterns
- Orchestrating multiple agents
- Debugging agent loops that get stuck or go off-track
Agent Architecture
User Goal → Planner → [Tool Call → Observe → Decide]* → Final Answer
Core Components
| Component | Purpose |
|---|---|
| Planner | Break goal into steps |
| Executor | Run tools, API calls |
| Observer | Parse tool results |
| Decider | Continue, retry, or stop |
| Memory | Track state across steps |
Patterns
1. ReAct (Reasoning + Acting)
Thought: I need to find the user's order status
Action: query_database(user_id=123, table="orders")
Observation: [{"id": 456, "status": "shipped"}]
Thought: Found it. The order is shipped.
Answer: Your order #456 has been shipped.
2. Plan-and-Execute
Plan: 1) Search docs 2) Extract relevant info 3) Summarize
Execute step 1: search("refund policy") → [doc1, doc2]
Execute step 2: extract(doc1) → "30-day refund window..."
Execute step 3: summarize → "You have 30 days to request a refund."
3. Multi-Agent (specialist delegation)
Router Agent → classify intent
├── Research Agent → search + summarize
├── Code Agent → write + test code
└── Data Agent → query + analyze data
Graph-Based Agents (LangGraph)
Beyond simple loops, production agents use directed graphs for complex branching logic:
from langgraph.graph import StateGraph, END
def should_continue(state):
if state["tool_calls"]: return "tools"
if state["needs_human"]: return "human_review"
return END
graph = StateGraph(AgentState)
graph.add_node("llm", call_llm)
graph.add_node("tools", execute_tools)
graph.add_node("human_review", request_approval)
graph.add_conditional_edges("llm", should_continue)
graph.add_edge("tools", "llm") # loop back after tool execution
Key patterns:
- Parallel tool execution — fan-out to multiple tools simultaneously, merge results
- Conditional branching — route based on output content, confidence, or tool results
- Subgraph composition — nest agent graphs for modularity
- Persistence — LangGraph's checkpointing enables pause/resume across sessions
Tool Design Rules
- Clear names —
search_documentsnottool_1 - Typed parameters — JSON schema for every tool
- Bounded output — truncate large results, paginate
- Error messages — return actionable errors, not stack traces
- Idempotent — safe to retry on failure
Error Recovery
| Failure | Recovery |
|---|---|
| Tool returns error | Retry once, then report to user |
| Agent loops >10 steps | Force stop, summarize progress |
| Off-topic drift | Check goal alignment every 3 steps |
| Hallucinated tool call | Validate tool name exists before calling |
| Timeout | Set max execution time, graceful exit |
State Management
state = {
"goal": "Find cheapest flight to Tokyo",
"steps_completed": ["searched flights", "compared prices"],
"current_step": "booking confirmation",
"attempts": 2,
"max_attempts": 5,
"context": {...}
}
- Persist state for long-running agents
- Log every step for debugging
- Set hard limits: max steps, max time, max cost
Memory Architecture
Agents need different memory types for different purposes:
| Type | Storage | Lifetime | Use case |
|---|---|---|---|
| In-context | Token window | Current session | Active task state, recent tool results |
| Session | DB (Redis/Postgres) | One conversation | User preferences, conversation history |
| Long-term | Vector DB | Persistent | User facts, past decisions, learned patterns |
| Episodic | DB + embeddings | Persistent | Past task completions, examples |
Context window management:
# Summarize old messages to prevent overflow
def compress_history(messages, max_tokens=4000):
if count_tokens(messages) < max_tokens:
return messages
# Keep system + last 5 messages, summarize the rest
summary = llm.summarize(messages[1:-5])
return [messages[0], HumanMessage(f"[Summary: {summary}]")] + messages[-5:]
When to use external memory:
- Conversation > 20 turns → summarize and store
- User mentions facts that apply beyond this session → upsert to long-term store
- Agent needs to "remember" past tasks → episodic store with semantic search
Human-in-the-Loop
Design confidence-based escalation rather than binary human/autonomous:
def route_by_confidence(result, confidence_threshold=0.85):
if result.confidence >= confidence_threshold:
return "auto_proceed"
elif result.confidence >= 0.6:
return "notify_and_proceed" # log but continue
else:
return "require_approval" # block and wait
# Approval checkpoint in LangGraph
def human_approval_node(state):
# Pause execution, notify human, wait for response
send_notification(state["pending_action"])
approval = wait_for_human_input(timeout=3600) # 1 hour timeout
return {"approved": approval, "human_feedback": approval.comment}
When to require human approval:
- Irreversible actions (delete, send, purchase, deploy)
- Low confidence + high stakes
- Novel situation not seen in training
- User explicitly requested oversight
- Regulatory requirement (financial, medical)
Graceful timeout: If no response within timeout → escalate or abort safely, never proceed on assumption.
Anti-Patterns
| Pattern | Fix |
|---|---|
| No max iterations | Always set a loop limit |
| Agent calls itself recursively | Detect cycles, break loop |
| Too many tools (>15) | Group into categories, use router |
| No logging | Log every thought/action/observation |
| Trusting agent output blindly | Validate before executing side effects |
Cost-Aware Agent Design
Agent costs compound: each step adds tokens. Design for efficiency:
Token budgets:
class CostAwareAgent:
def __init__(self, max_tokens_per_task=50000):
self.token_budget = max_tokens_per_task
self.tokens_used = 0
def should_continue(self, step_estimate):
if self.tokens_used + step_estimate > self.token_budget * 0.9:
return "summarize_and_stop" # graceful degradation
return "continue"
Tool selection strategy:
- Cheap tools first (search before synthesize)
- Cache tool results for duplicate calls in same session
- Use smaller model for tool selection, larger for final synthesis
- Parallel tools when independent (fan-out saves latency and doesn't add cost)
Per-task cost tracking:
# Log cost per agent task for accountability
log_task_cost(task_id, input_tokens, output_tokens, tool_calls, total_usd)
Integration
- magic-powers:prompt-engineering — design agent system prompts
- magic-powers:llm-evaluation — evaluate agent task completion
- magic-powers:ai-safety-guardrails — guard agent actions
- magic-powers:llm-observability — monitor agent runs