Building Autonomous AI Agents: Architecture, Memory, and Tool Use
AI agents have evolved from simple chatbots into autonomous systems that can plan, reason, use tools, and coordinate with other agents. This guide covers the architecture patterns, memory systems, and tool orchestration strategies used in production agent systems.
Agent Architecture Patterns
Sponsored
1. ReAct (Reason + Act)
The most common pattern for single-agent systems:
python
class ReActAgent:
def __init__(self, llm, tools):
self.llm = llm
self.tools = {tool.name: tool for tool in tools}
self.memory = AgentMemory()
async def run(self, task: str) -> str:
self.memory.add_user_message(task)
for step in range(self.max_steps):
# Reason: LLM decides what to do
response = await self.llm.generate(
messages=self.memory.get_messages(),
tools=self._format_tools()
)
# Check if agent wants to use a tool
if response.tool_calls:
for tool_call in response.tool_calls:
# Act: Execute the tool
result = await self.tools[tool_call.name].execute(
**tool_call.arguments
)
# Store result in memory
self.memory.add_tool_result(
tool_call.name, result
)
else:
# Agent is done — return final answer
return response.content
return "Max steps reached without final answer"
2. Plan-and-Execute
For complex tasks that benefit from upfront planning:
python
class PlanAndExecuteAgent:
def __init__(self, planner_llm, executor_llm, tools):
self.planner = planner_llm
self.executor = executor_llm
self.tools = tools
async def run(self, task: str) -> str:
# Step 1: Create a plan
plan = await self._create_plan(task)
# Step 2: Execute each step
results = []
for step in plan.steps:
result = await self._execute_step(step, results)
results.append(result)
# Re-plan if something goes wrong
if result.failed:
plan = await self._replan(task, results)
# Step 3: Synthesize final answer
return await self._synthesize(task, results)
async def _create_plan(self, task: str) -> Plan:
response = await self.planner.generate(
f"""Create a step-by-step plan to accomplish this task:
Task: {task}
Available tools: {self._tool_descriptions()}
Return a JSON plan with steps, each specifying:
- description: What to do
- tool: Which tool to use (if any)
- depends_on: Previous step indices this depends on
"""
)
return Plan.parse(response)
3. Multi-Agent Systems
For tasks requiring specialized expertise:
python
class MultiAgentSystem:
def __init__(self):
self.agents = {
'researcher': Agent(
name="Researcher",
tools=[web_search, wikipedia, arxiv],
system_prompt="You research topics thoroughly."
),
'coder': Agent(
name="Coder",
tools=[code_interpreter, file_editor],
system_prompt="You write and test code."
),
'reviewer': Agent(
name="Reviewer",
tools=[],
system_prompt="You review work for quality and correctness."
)
}
self.coordinator = CoordinatorAgent(self.agents)
async def run(self, task: str) -> str:
# Coordinator decides which agents to involve
workflow = await self.coordinator.plan_workflow(task)
results = {}
for step in workflow:
agent = self.agents[step.agent_name]
result = await agent.run(
step.instruction,
context=results # Previous results as context
)
results[step.id] = result
return self._compile_final_output(results)
Memory Systems
Short-Term Memory (Conversation Buffer)
python
class ConversationBuffer:
def __init__(self, max_tokens=4000):
self.messages = []
self.max_tokens = max_tokens
def add(self, message: Message):
self.messages.append(message)
self._trim_to_fit()
def _trim_to_fit(self):
"""Keep only recent messages that fit in token limit"""
while self._count_tokens() > self.max_tokens:
# Remove oldest non-system messages
for i, msg in enumerate(self.messages):
if msg.role != "system":
self.messages.pop(i)
break
Long-Term Memory (Vector Store)
python
class LongTermMemory:
def __init__(self, vector_store, embedding_model):
self.vector_store = vector_store
self.embedding_model = embedding_model
async def store(self, content: str, metadata: dict):
"""Store information for later retrieval"""
embedding = self.embedding_model.encode(content)
await self.vector_store.upsert(
embedding=embedding,
content=content,
metadata=metadata
)
async def retrieve(self, query: str, top_k: int = 5):
"""Retrieve relevant memories"""
query_embedding = self.embedding_model.encode(query)
results = await self.vector_store.search(
query_embedding, top_k=top_k
)
return results
async def consolidate(self, conversation: list[Message]):
"""Extract and store key facts from conversation"""
facts = await self.llm.extract_facts(conversation)
for fact in facts:
await self.store(fact.content, {
'type': 'fact',
'source': conversation[0].timestamp,
'importance': fact.importance
})
Episodic Memory (Event Store)
python
class EpisodicMemory:
def __init__(self):
self.episodes = []
def record_episode(self, task: str, steps: list[Step], outcome: str):
"""Record a complete task execution"""
self.episodes.append({
'task': task,
'steps': steps,
'outcome': outcome,
'timestamp': datetime.now(),
'success': 'error' not in [s.status for s in steps]
})
async def find_similar_episodes(self, task: str, top_k: int = 3):
"""Find past episodes similar to current task"""
# Use embedding similarity or keyword matching
similar = []
task_embedding = self.embed(task)
for episode in self.episodes:
ep_embedding = self.embed(episode['task'])
similarity = cosine_similarity(task_embedding, ep_embedding)
similar.append((episode, similarity))
similar.sort(key=lambda x: x[1], reverse=True)
return [ep for ep, _ in similar[:top_k]]
Tool Orchestration
Tool Definition Pattern
python
from pydantic import BaseModel, Field
class ToolDefinition(BaseModel):
name: str
description: str
parameters: dict
returns: str
def to_openai_format(self) -> dict:
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters
}
}
# Define tools
SEARCH_TOOL = ToolDefinition(
name="web_search",
description="Search the web for current information",
parameters={
"type": "object",
"properties": {
"query": {"type": "string", "description": "Search query"}
},
"required": ["query"]
},
returns="List of search results with titles, URLs, and snippets"
)
CODE_TOOL = ToolDefinition(
name="execute_code",
description="Execute Python code in a sandboxed environment",
parameters={
"type": "object",
"properties": {
"code": {"type": "string", "description": "Python code to execute"},
"timeout": {"type": "integer", "description": "Timeout in seconds"}
},
"required": ["code"]
},
returns="Execution output (stdout, stderr, or error)"
)
Dynamic Tool Discovery
python
class DynamicToolRegistry:
def __init__(self):
self.tools = {}
self.tool_embeddings = {}
async def discover_tools(self, task_description: str):
"""Automatically find relevant tools for a task"""
task_embedding = await self.embed(task_description)
relevant_tools = []
for name, tool in self.tools.items():
tool_embedding = self.tool_embeddings[name]
similarity = cosine_similarity(task_embedding, tool_embedding)
if similarity > self.threshold:
relevant_tools.append(tool)
return relevant_tools
def register_tool(self, tool: ToolDefinition, implementation: callable):
"""Register a new tool dynamically"""
self.tools[tool.name] = {
'definition': tool,
'implementation': implementation
}
# Pre-compute embedding for discovery
self.tool_embeddings[tool.name] = self.embed(tool.description)
Error Handling and Recovery
Advertisement
Graceful Degradation
python
class ResilientAgent:
async def execute_with_recovery(self, step: AgentStep):
"""Execute a step with automatic error recovery"""
max_retries = 3
for attempt in range(max_retries):
try:
result = await self._execute_step(step)
return result
except ToolError as e:
# Tool failed — try alternative approach
if attempt < max_retries - 1:
alternative = await self._suggest_alternative(step, e)
step = alternative
else:
# All retries failed — report error
return StepResult(
status="failed",
error=str(e),
suggestion="Manual intervention required"
)
except LLMError as e:
# LLM failed — try with different model
if attempt < max_retries - 1:
self.llm = self._fallback_model()
else:
raise
Human-in-the-Loop
python
class HumanInTheLoopAgent:
def __init__(self, llm, tools, human_approval_threshold=0.7):
self.llm = llm
self.tools = tools
self.approval_threshold = human_approval_threshold
async def execute(self, step: AgentStep):
# Check if step requires human approval
risk_score = await self._assess_risk(step)
if risk_score > self.approval_threshold:
# Request human approval
approval = await self._request_approval(step)
if not approval.approved:
return StepResult(
status="rejected",
reason=approval.reason,
suggestion=approval.alternative
)
# Execute the step
return await self._execute_step(step)
Real-World Deployment Patterns
1. Research Agent
python
class ResearchAgent:
"""Automated research agent that gathers and synthesizes information"""
async def research(self, topic: str) -> ResearchReport:
# Phase 1: Initial search
sources = await self.web_search(topic, num_results=10)
# Phase 2: Read and extract key information
extracts = []
for source in sources:
content = await self.web_reader.read(source.url)
key_points = await self.extract_key_points(content, topic)
extracts.append(key_points)
# Phase 3: Cross-reference and verify
verified_facts = await self.cross_reference(extracts)
# Phase 4: Synthesize report
report = await self.synthesize_report(
topic=topic,
facts=verified_facts,
sources=sources
)
return report
2. Code Development Agent
python
class DevAgent:
"""End-to-end code development agent"""
async def develop(self, requirements: str) -> CodeResult:
# Phase 1: Understand requirements
spec = await self.analyze_requirements(requirements)
# Phase 2: Design architecture
design = await self.design_architecture(spec)
# Phase 3: Implement code
code = await self.implement(design)
# Phase 4: Write tests
tests = await self.write_tests(code, spec)
# Phase 5: Run tests and fix issues
test_results = await self.run_tests(tests)
if test_results.failed:
code = await self.fix_issues(code, test_results.errors)
# Phase 6: Code review
review = await self.review_code(code)
return CodeResult(
code=code,
tests=tests,
review=review
)
Production Checklist
- Error handling: Graceful degradation, retry logic, fallback models
- Memory management: Context window limits, long-term storage, consolidation
- Tool validation: Input validation, output verification, timeout handling
- Cost control: Token budgets, tool call limits, caching
- Monitoring: Step-by-step logging, performance metrics, error tracking
- Safety: Human approval for high-risk actions, sandboxed execution
- Evaluation: Task completion rates, accuracy, efficiency metrics
Conclusion
Building production AI agents requires careful attention to architecture, memory, tool orchestration, and error handling. The key insights are:
- Start with ReAct — it's the simplest pattern that works for most use cases
- Memory is critical — agents without memory are stateless and limited
- Tool orchestration matters — dynamic tool discovery beats fixed tool sets
- Error handling is essential — agents will fail, plan for it
- Monitor everything — you can't improve what you can't measure
The future of AI agents is multi-agent systems with specialized capabilities working together to solve complex problems.