Library Architecture
This document explains the internal architecture of Agent Patterns, how patterns work, and how they integrate with LangGraph and LangChain.
Overview
Agent Patterns is built on three foundational components:
LangChain: Provides LLM abstractions and integrations
LangGraph: Manages state graphs and workflow execution
Agent Patterns: Implements reusable agent workflow patterns
┌─────────────────────────────────────┐
│ Your Application │
├─────────────────────────────────────┤
│ Agent Patterns Library │
│ ┌─────────────┬─────────────┐ │
│ │ Pattern │ Pattern │ │
│ │ Library │ Base │ │
│ │ │ Classes │ │
│ └─────────────┴─────────────┘ │
├─────────────────────────────────────┤
│ LangGraph │
│ (State Graph Management) │
├─────────────────────────────────────┤
│ LangChain │
│ (LLM Integrations) │
└─────────────────────────────────────┘
Core Architecture
Component Hierarchy
BaseAgent (Abstract)
├── Single-Agent Patterns
│ ├── ReActAgent
│ ├── ReflectionAgent
│ ├── PlanAndSolveAgent
│ ├── ReflexionAgent
│ ├── LLMCompilerAgent
│ ├── REWOOAgent
│ ├── LATSAgent
│ ├── SelfDiscoveryAgent
│ └── STORMAgent
└── MultiAgentBase (Abstract)
└── (Future multi-agent patterns)
BaseAgent Class
All patterns inherit from BaseAgent, which provides:
class BaseAgent(abc.ABC):
"""Abstract base for all patterns."""
def __init__(
self,
llm_configs: Dict[str, Dict[str, Any]],
prompt_dir: str = "prompts",
custom_instructions: Optional[str] = None,
prompt_overrides: Optional[Dict[str, Dict[str, str]]] = None
):
# LLM configuration and caching
self.llm_configs = llm_configs
self._llm_cache: Dict[str, BaseChatModel] = {}
# Prompt customization
self.prompt_dir = prompt_dir
self.custom_instructions = custom_instructions
self.prompt_overrides = prompt_overrides or {}
# State graph
self.graph: Optional[CompiledStateGraph] = None
# Build the pattern's graph
self.build_graph()
@abc.abstractmethod
def build_graph(self) -> None:
"""Build the LangGraph state graph."""
pass
@abc.abstractmethod
def run(self, input_data: Any) -> Any:
"""Execute the pattern."""
pass
Key Features:
LLM Management: Initializes and caches LLM instances
Prompt Loading: Loads prompts from files with customization support
Graph Building: Constructs the pattern’s state machine
Lifecycle Hooks: Provides extension points for logging/monitoring
State Graph Architecture
Every pattern is a state graph managed by LangGraph.
Graph Components
┌─────────────┐
│ Entry Node │ ← Starting point
└──────┬──────┘
│
▼
┌─────────────┐
│ Node 1 │ ← Processing step
└──────┬──────┘
│
▼
┌───┴───┐
│ Edge │ ← Transition logic
└───┬───┘
│
▼
┌─────────────┐
│ Node 2 │ ← Another step
└──────┬──────┘
│
▼
┌─────────────┐
│ END Node │ ← Terminal state
└─────────────┘
Example: ReAct Pattern Graph
def build_graph(self):
workflow = StateGraph(dict)
# Add nodes (processing steps)
workflow.add_node("thought_step", self._generate_thought_and_action)
workflow.add_node("action_step", self._execute_action)
workflow.add_node("observation_step", self._observation_handler)
workflow.add_node("final_answer", self._format_final_answer)
# Set entry point
workflow.set_entry_point("thought_step")
# Add edges (transitions)
workflow.add_edge("thought_step", "action_step")
workflow.add_edge("action_step", "observation_step")
# Conditional edge (routing logic)
workflow.add_conditional_edges(
"observation_step",
self._should_continue, # Decision function
{
"continue": "thought_step", # Loop back
"finish": "final_answer", # Exit loop
},
)
workflow.add_edge("final_answer", END)
# Compile to executable graph
self.graph = workflow.compile()
Graph Execution:
# Initialize state
initial_state = {
"input": "What is 2+2?",
"thought": "",
"action": {},
"observation": None,
"final_answer": None,
}
# Execute graph
final_state = self.graph.invoke(initial_state)
# Extract result
result = final_state["final_answer"]
State Management
State Dictionary
Each pattern defines its state schema:
# ReAct Pattern State
state = {
"input": str, # User query
"thought": str, # Current reasoning
"action": Dict, # Tool to call
"observation": Any, # Tool result
"intermediate_steps": List,# History
"final_answer": str, # Result
"iteration_count": int, # Loop counter
"max_iterations": int, # Loop limit
}
State Flow
State flows through the graph:
┌──────────────────────────────────────┐
│ Initial State │
│ {"input": "query", ...} │
└────────────┬─────────────────────────┘
│
▼
┌──────────────────────────────────────┐
│ Node updates state │
│ state["thought"] = "I should..." │
└────────────┬─────────────────────────┘
│
▼
┌──────────────────────────────────────┐
│ State passed to next node │
│ state = next_node(state) │
└────────────┬─────────────────────────┘
│
▼
┌──────────────────────────────────────┐
│ Final State │
│ {"input": "query", "final_answer":..}│
└──────────────────────────────────────┘
Important: Each node receives the state, modifies it, and returns the updated state.
LLM Integration
LLM Configuration
Patterns use role-based LLM configuration:
llm_configs = {
"thinking": {
"provider": "openai",
"model_name": "gpt-4-turbo",
"temperature": 0.7,
"max_tokens": 2000,
},
"reflection": {
"provider": "anthropic",
"model_name": "claude-3-5-sonnet-20241022",
"temperature": 0.5,
"max_tokens": 1000,
},
}
LLM Initialization and Caching
def _get_llm(self, role: str) -> BaseChatModel:
"""Get or create LLM for role."""
# Check cache
if role in self._llm_cache:
return self._llm_cache[role]
# Get config
config = self.llm_configs[role]
provider = config["provider"]
model_name = config["model_name"]
# Initialize LLM
if provider == "openai":
llm = ChatOpenAI(
model=model_name,
temperature=config.get("temperature", 0.7),
max_tokens=config.get("max_tokens", 2000)
)
elif provider == "anthropic":
llm = ChatAnthropic(
model=model_name,
temperature=config.get("temperature", 0.7),
max_tokens=config.get("max_tokens", 2000)
)
# Cache and return
self._llm_cache[role] = llm
return llm
Benefits:
LLMs initialized once and reused
Different models for different roles
Easy to swap providers/models
LLM Invocation
# Get LLM for role
llm = self._get_llm("thinking")
# Prepare messages
messages = [
SystemMessage(content=system_prompt),
HumanMessage(content=user_prompt),
]
# Invoke LLM
response = llm.invoke(messages)
result = response.content
Prompt Management
Three-Layer Prompt System
1. Base Prompts (Files)
↓
2. Custom Instructions (Appended)
↓
3. Prompt Overrides (Replacement)
File-Based Prompts
prompts/
└── ReActAgent/
└── ThoughtStep/
├── system.md # System prompt
└── user.md # User prompt template
Loading:
def _load_prompt(self, step_name: str) -> Dict[str, str]:
"""Load prompts for a step."""
# Get class name for directory
class_name = self.__class__.__name__
# Build path
prompt_path = Path(self.prompt_dir) / class_name / step_name
# Load files
system_prompt = (prompt_path / "system.md").read_text()
user_prompt = (prompt_path / "user.md").read_text()
return {
"system": system_prompt,
"user": user_prompt
}
Custom Instructions
Appended to all system prompts:
agent = ReActAgent(
llm_configs=configs,
custom_instructions="""
You are an expert in the medical domain.
Always cite sources and include disclaimers.
"""
)
# System prompt becomes:
# <original system prompt>
#
# ## Custom Instructions
#
# <custom instructions>
Prompt Overrides
Complete replacement for specific steps:
overrides = {
"ThoughtStep": {
"system": "You are a cautious reasoner...",
"user": "Task: {input}\n\nWhat should I do?"
}
}
agent = ReActAgent(
llm_configs=configs,
prompt_overrides=overrides
)
Priority: Overrides > File System > Custom Instructions
Pattern Execution Flow
Complete Execution Lifecycle
1. Initialization
├─ Load LLM configs
├─ Load/override prompts
└─ Build state graph
2. Invocation (run method)
├─ Create initial state
├─ Call on_start() hook
└─ Invoke graph
3. Graph Execution
├─ Enter first node
├─ Process state
├─ Invoke LLM if needed
├─ Update state
├─ Transition to next node
├─ Repeat until END
└─ Return final state
4. Completion
├─ Extract result from state
├─ Call on_finish() hook
└─ Return result
Example: ReAct Execution
# 1. User creates agent
agent = ReActAgent(
llm_configs={"thinking": {...}},
tools={"search": search_fn},
max_iterations=5
)
# → build_graph() called automatically
# 2. User invokes agent
result = agent.run("What is the weather in Paris?")
# 3. Internal execution:
initial_state = {
"input": "What is the weather in Paris?",
"thought": "",
"action": {},
"observation": None,
"intermediate_steps": [],
"final_answer": None,
"iteration_count": 0,
"max_iterations": 5,
}
# Graph executes:
# thought_step → action_step → observation_step → decision
# ↑ ↓
# └──────────────────┘
# (loop if continue)
# 4. Result extracted
result = final_state["final_answer"]
# → "The weather in Paris is..."
Tool Integration
Tool Definition
Tools are simple Python functions:
def search_tool(query: str) -> str:
"""Search the web for information."""
# Implementation
results = api.search(query)
return results
Tool Registration
agent = ReActAgent(
llm_configs=configs,
tools={
"search": search_tool,
"calculator": calc_tool,
"database": db_tool,
}
)
Tool Execution
Within a pattern node:
def _execute_action(self, state: Dict) -> Dict:
"""Execute tool specified in action."""
action = state["action"]
tool_name = action["tool_name"]
tool_input = action["tool_input"]
# Get tool function
if tool_name in self.tools:
try:
# Execute tool
observation = self.tools[tool_name](tool_input)
except Exception as e:
observation = f"Error: {str(e)}"
else:
observation = f"Tool {tool_name} not found"
# Update state
state["observation"] = observation
return state
Lifecycle Hooks
Patterns provide hooks for monitoring and logging:
class BaseAgent:
def on_start(self, input_data: Any) -> None:
"""Called before execution starts."""
pass
def on_finish(self, result: Any) -> None:
"""Called after execution completes."""
pass
def on_error(self, error: Exception) -> None:
"""Called when an error occurs."""
pass
Custom Hook Implementation
class MonitoredReActAgent(ReActAgent):
def on_start(self, input_data):
print(f"[START] Query: {input_data}")
self.start_time = time.time()
def on_finish(self, result):
duration = time.time() - self.start_time
print(f"[FINISH] Duration: {duration:.2f}s")
print(f"[RESULT] {result[:100]}...")
def on_error(self, error):
print(f"[ERROR] {type(error).__name__}: {error}")
# Log to monitoring system
logger.error(f"Agent error: {error}")
Extending Patterns
Creating Custom Patterns
Inherit from BaseAgent:
from agent_patterns.core import BaseAgent
from langgraph.graph import StateGraph, END
class MyCustomPattern(BaseAgent):
def __init__(self, llm_configs, **kwargs):
super().__init__(llm_configs, **kwargs)
def build_graph(self):
"""Define your workflow."""
workflow = StateGraph(dict)
# Add your nodes
workflow.add_node("step1", self._step1)
workflow.add_node("step2", self._step2)
# Define flow
workflow.set_entry_point("step1")
workflow.add_edge("step1", "step2")
workflow.add_edge("step2", END)
# Compile
self.graph = workflow.compile()
def run(self, input_data):
"""Execute your pattern."""
initial_state = {
"input": input_data,
"output": None,
}
final_state = self.graph.invoke(initial_state)
return final_state["output"]
def _step1(self, state):
"""First processing step."""
# Use LLM
llm = self._get_llm("thinking")
# Load prompts
prompts = self._load_prompt("Step1")
# Process
# ...
return state
def _step2(self, state):
"""Second processing step."""
# ...
return state
Extending Existing Patterns
Override specific methods:
class EnhancedReActAgent(ReActAgent):
def _generate_thought_and_action(self, state):
"""Override thought generation with custom logic."""
# Add logging
print(f"Iteration {state['iteration_count']}")
# Call parent implementation
state = super()._generate_thought_and_action(state)
# Add custom processing
# ...
return state
Design Principles
1. Synchronous Execution
All patterns are synchronous - no async/await:
# Synchronous
result = agent.run(input)
# NOT async
# result = await agent.run(input)
Why: Simplicity, debuggability, and sufficient for most use cases.
2. Immutable by Default
State updates create new state:
def _update_state(self, state):
# Return updated state
return {**state, "new_field": value}
# NOT in-place modification
# state["new_field"] = value
# return state
3. Explicit Over Implicit
Everything is explicit:
State is a visible dictionary
LLM calls are clear
Tool execution is traceable
No hidden magic
4. Composition Over Inheritance
Prefer composition:
# Good: Compose with tools
agent = ReActAgent(tools={"search": search_fn})
# Less good: Inherit for simple customization
class SearchReActAgent(ReActAgent):
def __init__(self):
super().__init__(tools={"search": search_fn})
Performance Considerations
LLM Caching
LLMs are initialized once and cached:
# First call: initializes LLM
llm = self._get_llm("thinking")
# Subsequent calls: returns cached instance
llm = self._get_llm("thinking") # Instant
State Graph Compilation
Graphs are compiled once during initialization:
def __init__(self, ...):
# ...
self.build_graph() # Compiled once
# graph.invoke() is fast
# Each run() reuses compiled graph
result1 = agent.run(input1) # Uses compiled graph
result2 = agent.run(input2) # Uses compiled graph
Prompt Loading
Prompts are loaded on-demand and could be cached:
# Loaded when needed
prompts = self._load_prompt("Step1")
# For optimization, cache in subclass:
def __init__(self, ...):
super().__init__(...)
self._prompt_cache = {}
def _load_prompt(self, step_name):
if step_name not in self._prompt_cache:
self._prompt_cache[step_name] = super()._load_prompt(step_name)
return self._prompt_cache[step_name]
Next Steps
API Reference: See BaseAgent API for complete method documentation
Pattern Details: Learn about each pattern’s implementation
Examples: Study real implementations
Type Reference: Review type definitions