AI Agent Patterns

A curated collection of architectural patterns, design principles, and best practices for building autonomous AI agents.

Building reliable AI agents requires more than just LLM calls. This repository documents proven patterns for agent cognition, tool use, memory, planning, and multi-agent coordination.

---

🚀 Featured: AnveVoice Agent Example

See how AnveVoice implements the ReAct Pattern for voice AI conversations:

# Simplified AnveVoice agent loop
class VoiceAgent:
    def run(self, user_input):
        # Reasoning
        thought = self.reason(user_input)
        
        # Action selection
        action = self.select_action(thought)
        
        # Tool execution
        if action.requires_tool:
            result = self.execute_tool(action)
        
        # Response generation
        response = self.generate_response(thought, result)
        return response

Full implementation: anvevoice.com/architecture

---

Contents

• Fundamental Patterns
• Reasoning Patterns
• Memory Patterns
• Tool Use Patterns
• Planning Patterns
• Multi-Agent Patterns
• Orchestration Patterns
• Safety Patterns
• Evaluation Patterns
• Deployment Patterns

---

Fundamental Patterns

ReAct (Reasoning + Acting)

The foundational pattern for agent cognition. The agent alternates between reasoning about the situation and taking actions.

class ReActAgent:
    def run(self, query, max_iterations=10):
        for i in range(max_iterations):
            # Reasoning step
            thought = self.llm.generate(
                prompt=f"Given: {context}\nWhat should I do next?"
            )
            
            # Action step
            if "Action:" in thought:
                action = parse_action(thought)
                observation = execute_action(action)
                context += f"\nAction: {action}\nObservation: {observation}"
            else:
                return thought  # Final answer

Use when: General-purpose agents, complex multi-step tasks Examples: AnveVoice, LangChain agents, AutoGPT

---

Reflexion

Self-reflection pattern where the agent critiques its own outputs and iterates.

class ReflexionAgent:
    def run(self, task):
        result = self.initial_attempt(task)
        
        # Self-critique
        critique = self.llm.generate(
            f"Review this result: {result}\nWhat could be improved?"
        )
        
        # Iteration
        if critique.identifies_issues:
            result = self.improve(result, critique)
        
        return result

Use when: Quality-critical tasks, code generation, content creation

---

Chain-of-Thought

Prompt the model to show its reasoning before giving the final answer.

prompt = """
Q: Roger has 5 tennis balls. He buys 2 more cans of tennis balls. 
Each can has 3 tennis balls. How many does he have now?

A: Let's think step by step.
Roger started with 5 balls.
2 cans of 3 balls each is 6 balls.
5 + 6 = 11.
The answer is 11.

Q: {new_question}
A: Let's think step by step.
"""

Use when: Mathematical reasoning, logical deduction, complex queries

---

Reasoning Patterns

Tree-of-Thoughts

Explore multiple reasoning paths and select the best one.

class TreeOfThoughts:
    def solve(self, problem, beam_width=3):
        candidates = [problem]
        
        for step in range(max_steps):
            # Generate multiple thoughts
            thoughts = []
            for candidate in candidates:
                thoughts.extend(self.generate_thoughts(candidate, k=beam_width))
            
            # Evaluate and select best
            scores = [self.evaluate(t) for t in thoughts]
            candidates = select_top_k(thoughts, scores, beam_width)
        
        return best_candidate(candidates)

Use when: Strategic planning, game playing, creative problem solving

---

Reasoning via Search (RAG)

Augment LLM with retrieved knowledge for grounded reasoning.

class RAGAgent:
    def answer(self, question):
        # Retrieve relevant documents
        docs = self.retriever.search(question, top_k=5)
        
        # Generate with context
        context = "\n".join([d.content for d in docs])
        answer = self.llm.generate(
            f"Context: {context}\nQuestion: {question}\nAnswer:"
        )
        return answer

Use when: Knowledge-intensive tasks, question answering, fact checking

---

Memory Patterns

Short-Term Memory (Conversation Buffer)

Maintain recent conversation history as context.

class ConversationBuffer:
    def __init__(self, max_turns=10):
        self.buffer = []
        self.max_turns = max_turns
    
    def add(self, user_msg, assistant_msg):
        self.buffer.append({"user": user_msg, "assistant": assistant_msg})
        if len(self.buffer) > self.max_turns:
            self.buffer.pop(0)
    
    def get_context(self):
        return format_history(self.buffer)

Use when: Conversational agents, chatbots, interactive systems

---

Long-Term Memory (Vector Store)

Store and retrieve information across sessions using embeddings.

class LongTermMemory:
    def __init__(self, vector_store):
        self.store = vector_store
    
    def remember(self, key_information, session_id):
        embedding = embed(key_information)
        self.store.add(
            id=f"{session_id}_{timestamp}",
            vector=embedding,
            metadata={"content": key_information}
        )
    
    def recall(self, query, session_id=None, top_k=5):
        embedding = embed(query)
        return self.store.search(embedding, filter={"session_id": session_id})

Use when: Personalized agents, learning agents, long-running tasks

---

Entity Memory

Track entities and their attributes across conversations.

class EntityMemory:
    def __init__(self):
        self.entities = {}  # name -> attributes
    
    def extract_entities(self, text):
        entities = self.ner_model.extract(text)
        for entity in entities:
            if entity.name in self.entities:
                self.entities[entity.name].update(entity.attributes)
            else:
                self.entities[entity.name] = entity.attributes
    
    def get_entity_summary(self, name):
        return self.entities.get(name)

Use when: Customer support, CRM agents, relationship management

---

Tool Use Patterns

Function Calling

Structured tool invocation with schema validation.

class ToolUsingAgent:
    def __init__(self):
        self.tools = {
            "search": SearchTool(),
            "calculator": CalculatorTool(),
            "calendar": CalendarTool()
        }
    
    def run(self, query):
        response = self.llm.generate(
            query,
            functions=[t.schema for t in self.tools.values()]
        )
        
        if response.function_call:
            tool = self.tools[response.function_call.name]
            result = tool.execute(response.function_call.arguments)
            return self.run(f"Result: {result}")
        
        return response.content

Use when: Calculator, search, API integrations, structured outputs

---

Toolformer Pattern

Teach the model to decide when to use tools by training on tool-augmented data.

# Training data format
"""
The capital of France is Paris. 
The population is [Calculator(2.1M + 10.8M)] 12.9 million.
The weather today is [WeatherAPI(location="Paris")] sunny, 22°C.
"""

class ToolformerAgent:
    def generate(self, prompt):
        output = self.model.generate(prompt)
        # Parse tool calls from output
        tool_calls = parse_tool_calls(output)
        # Execute and insert results
        return execute_and_fill(tool_calls, output)

Use when: Complex tool chains, open-ended tool use

---

MCP (Model Context Protocol)

Standardized tool interface for AI assistants.

# MCP Server definition
class MCPServer:
    @tool()
    def search_web(query: str) -> str:
        """Search the web for information."""
        return web_search(query)
    
    @tool()
    def calculate(expression: str) -> float:
        """Evaluate a mathematical expression."""
        return eval(expression)

# Client usage
class MCPClient:
    def use_tool(self, server, tool_name, **kwargs):
        tool = server.get_tool(tool_name)
        return tool.execute(**kwargs)

Use when: OpenClaw, Claude Desktop, any MCP-compatible system

---

Planning Patterns

Hierarchical Task Network (HTN)

Decompose high-level goals into subtasks recursively.

class HTNPlanner:
    def plan(self, goal):
        if self.is_primitive(goal):
            return [goal]
        
        methods = self.get_methods(goal)
        for method in methods:
            if method.preconditions_met():
                subtasks = method.decompose()
                plan = []
                for subtask in subtasks:
                    subplan = self.plan(subtask)
                    if subplan:
                        plan.extend(subplan)
                return plan
        return None

Use when: Complex goal achievement, robotics, project management

---

ReAct with Planning (Plan-and-Solve)

Generate a plan first, then execute with ReAct.

class PlanAndSolve:
    def run(self, task):
        # Planning phase
        plan = self.llm.generate(f"Create a plan for: {task}")
        steps = parse_plan(plan)
        
        # Execution phase
        results = []
        for step in steps:
            result = self.react_agent.execute(step)
            results.append(result)
        
        return synthesize_results(results)

Use when: Multi-step projects, research tasks, complex workflows

---

Multi-Agent Patterns

Agent Council

Multiple specialized agents vote on decisions.

class AgentCouncil:
    def __init__(self):
        self.agents = {
            "technical": TechnicalAgent(),
            "business": BusinessAgent(),
            "creative": CreativeAgent()
        }
    
    def decide(self, proposal):
        votes = {}
        for name, agent in self.agents.items():
            vote = agent.evaluate(proposal)
            votes[name] = vote
        
        return self.consensus(votes)

Use when: Strategic decisions, content review, product decisions

---

Hierarchical Multi-Agent

Manager agent delegates to worker agents.

class ManagerAgent:
    def __init__(self):
        self.workers = {
            "researcher": ResearchAgent(),
            "writer": WriterAgent(),
            "reviewer": ReviewerAgent()
    
    def execute_project(self, goal):
        # Plan
        tasks = self.plan_tasks(goal)
        
        # Delegate
        results = {}
        for task in tasks:
            worker = self.select_worker(task)
            results[task.id] = worker.execute(task)
        
        # Integrate
        return self.synthesize(results)

Use when: Software projects, research, content production

---

Peer-to-Peer Multi-Agent

Agents collaborate as equals.

class PeerNetwork:
    def __init__(self, agents):
        self.agents = agents
        self.message_bus = MessageBus()
    
    def collaborate(self, task):
        # Broadcast to all agents
        self.message_bus.broadcast(task)
        
        # Agents volunteer or respond
        responses = []
        for agent in self.agents:
            if agent.can_contribute(task):
                response = agent.contribute(task)
                responses.append(response)
        
        # Consensus
        return self.merge_responses(responses)

Use when: Brainstorming, open-ended collaboration, research

---

Orchestration Patterns

Supervisor Pattern

Central supervisor monitors and manages agent execution.

class Supervisor:
    def __init__(self, agents):
        self.agents = agents
        self.state = {}
    
    def run(self, workflow):
        for step in workflow.steps:
            agent = self.agents[step.agent_id]
            
            # Pre-execution check
            if not self.validate_step(step):
                self.handle_error("Validation failed", step)
                continue
            
            # Execute
            result = agent.execute(step, context=self.state)
            
            # Post-execution check
            if not self.validate_result(result):
                result = self.retry_or_escalate(step)
            
            self.state[step.id] = result
        
        return self.state

Use when: Production workflows, error handling, compliance

---

State Machine

Agent behavior defined as transitions between states.

class StateMachineAgent:
    def __init__(self):
        self.state = "idle"
        self.transitions = {
            "idle": {"start": "working"},
            "working": {"complete": "finished", "error": "error"},
            "error": {"retry": "working", "abort": "finished"},
            "finished": {"reset": "idle"}
        }
    
    def transition(self, action):
        if action in self.transitions[self.state]:
            self.state = self.transitions[self.state][action]
            return True
        return False

Use when: Customer service bots, transaction processing, game AI

---

Safety Patterns

Constitutional AI

Train agents to follow principles/constitution.

class ConstitutionalAgent:
    def __init__(self, constitution):
        self.constitution = constitution  # List of principles
    
    def generate(self, prompt):
        response = self.model.generate(prompt)
        
        # Self-critique against constitution
        critique = self.check_constitution(response)
        if critique.violations:
            response = self.revise(response, critique)
        
        return response
    
    def check_constitution(self, text):
        violations = []
        for principle in self.constitution:
            if principle.is_violated_by(text):
                violations.append(principle)
        return violations

Use when: Public-facing agents, sensitive domains, content generation

---

Human-in-the-Loop

Require human approval for critical actions.

class HITLAgent:
    def __init__(self, auto_approved=None):
        self.auto_approved = auto_approved or []
    
    def execute(self, action):
        if action.type in self.auto_approved:
            return action.execute()
        
        # Request approval
        approval = self.request_human_approval(action)
        if approval.granted:
            return action.execute()
        else:
            return self.handle_rejection(action, approval.reason)

Use when: Financial transactions, content publishing, high-stakes decisions

---

Sandboxing

Execute untrusted code in isolated environments.

class SandboxedTool:
    def __init__(self, image="sandbox:latest"):
        self.image = image
    
    def execute(self, code):
        # Run in Docker container
        container = docker.run(
            self.image,
            command=["python", "-c", code],
            network="none",  # No network
            memory_limit="256m",
            timeout=30
        )
        return container.stdout

Use when: Code execution, file operations, untrusted inputs

---

Evaluation Patterns

Trajectory Evaluation

Evaluate the sequence of agent actions, not just final output.

def evaluate_trajectory(trajectory, expected):
    scores = {
        "efficiency": len(trajectory) / len(expected),
        "correctness": sum(t == e for t, e in zip(trajectory, expected)),
        "tool_accuracy": accuracy_of_tool_calls(trajectory),
        "relevance": relevance_to_goal(trajectory, goal)
    }
    return weighted_score(scores)

Use when: Agent benchmarking, RL training, quality assurance

---

Outcome vs Process Evaluation

class Evaluator:
    def outcome_eval(self, result, expected):
        """Did it get the right answer?"""
        return result == expected
    
    def process_eval(self, trajectory, expected_trajectory):
        """Did it follow the right steps?"""
        return similarity(trajectory, expected_trajectory)

Use when: Training data curation, agent debugging

---

Deployment Patterns

Streaming Responses

Stream agent outputs for better UX.

class StreamingAgent:
    async def run(self, query):
        stream = self.model.generate_stream(query)
        
        async for token in stream:
            if is_thought(token):
                yield {"type": "thought", "content": token}
            elif is_action(token):
                yield {"type": "action", "content": token}
            else:
                yield {"type": "response", "content": token}

Use when: Real-time agents, chat interfaces, voice applications

---

Async Agent Execution

Handle multiple agent tasks concurrently.

class AsyncAgentPool:
    def __init__(self, max_concurrent=10):
        self.semaphore = asyncio.Semaphore(max_concurrent)
    
    async def execute_batch(self, tasks):
        async with self.semaphore:
            return await asyncio.gather(*[
                self.execute_task(t) for t in tasks
            ])

Use when: Batch processing, high-throughput systems

---

Contributing

1. Submit a pattern you've used in production
2. Include code examples in Python or pseudocode
3. Describe when to use and trade-offs
4. Add real-world examples if possible

See contributing.md for guidelines.

---

Related Resources

• AnveVoice — Voice AI implementing these patterns
• OpenClaw — Local AI assistant framework
• Building LLM Apps — Application patterns
• LangChain Documentation — Framework patterns

---

Made with ❤️ by the AI agent community
_{Curated by AnveVoice — Production voice AI agents}