InkdownInkdown
Start writing

Study

59 filesยท8 subfolders

Shared Workspace

Study
core

20_ARCHITECTURE_PATTERNS

Shared from "Study" on Inkdown

Architecture Patterns - Comprehensive Deep Dive

Overview

This document describes common architectural patterns for building applications with the OpenAI Agents Python SDK. These patterns are proven approaches for structuring agent-based systems, drawn from real-world implementations and best practices. Understanding these patterns will help you design scalable, maintainable, and effective agent architectures.

Pattern Categories

  1. Agent Organization Patterns - How to organize agents
  2. Workflow Patterns - How agents work together
  3. Data Flow Patterns - How data flows through the system
  4. Integration Patterns - How to integrate with external systems
  5. Deployment Patterns - How to deploy agent applications

Agent Organization Patterns

1. Single Agent Pattern

Description: A single agent handles all tasks.

programming-language-concepts.md
zero-language-explanation.md
DB
01-introduction.md
02-relational-databases.md
03-database-design.md
04-indexing.md
05-transactions-acid.md
06-nosql-databases.md
07-query-optimization.md
08-replication-ha.md
09-sharding-partitioning.md
10-caching-strategies.md
11-cap-theorem.md
12-connection-pooling.md
13-backup-recovery.md
14-monitoring.md
15-database-selection.md
README.md
JS
Event loop
Merlin Backend
01-Orchestration.md
02-DeepResearch.md
03-Search.md
04-Scraping.md
05-Streaming.md
06-MultiProviderLLM.md
07-MemoryAndContext.md
08-ErrorHandling.md
09-RateLimiting.md
10-TaskQueue.md
11-SecurityAndAuth.md
Orchestration-2nd-draft
OpenAI Agents Python
00_OVERVIEW.md
01_AGENT_SYSTEM.md
02_RUNNER_SYSTEM.md
03_TOOL_SYSTEM.md
04_ITEMS_SYSTEM.md
05_GUARDRAILS.md
06_HANDOFFS.md
07_MEMORY_SESSIONS.md
08_MODEL_PROVIDERS.md
09_SANDBOX_SYSTEM.md
10_TRACING.md
11_RUN_STATE.md
12_CONTEXT.md
13_LIFECYCLE_HOOKS.md
14_CONFIGURATION.md
15_ERROR_HANDLING.md
16_STREAMING.md
17_EXTENSIONS.md
18_MCP_INTEGRATION.md
19_BEST_PRACTICES.md
20_ARCHITECTURE_PATTERNS.md
opencode-study
context-handling
core
Python
Alembic
Basics
sqlalchemy - fastapi
SQLAlchemy overview
tweets
system_design_for_agentic_apps.md
When to use:
  • Simple applications
  • Narrow domain
  • Single type of task

Example:

Python
agent = Agent(
    name="assistant",
    instructions="You are a helpful assistant for customer support",
    tools=[search_knowledge_base, create_ticket],
)

result = await Runner.run(agent, user_input)

Pros:

  • Simple to implement
  • Easy to understand
  • Low overhead

Cons:

  • Limited scalability
  • Hard to maintain for complex domains
  • Single point of failure
2. Specialist Pattern

Description: Multiple specialized agents, each handling a specific domain.

When to use:

  • Multiple domains
  • Different expertise required
  • Clear separation of concerns

Example:

Python
billing_agent = Agent(
    name="billing",
    instructions="Handle billing and payment issues",
    tools=[billing_api, refund_tool],
)

technical_agent = Agent(
    name="technical",
    instructions="Handle technical support issues",
    tools=[debug_tool, log_analyzer],
)

general_agent = Agent(
    name="general",
    instructions="Route to appropriate specialist",
    handoffs=[handoff(billing_agent), handoff(technical_agent)],
)

Pros:

  • Clear separation of concerns
  • Easy to add new specialists
  • Each agent can be optimized for its domain

Cons:

  • More complex to manage
  • Requires good handoff logic
  • Potential for handoff loops
3. Hierarchical Pattern

Description: Agents organized in a hierarchy, with supervisors overseeing specialists.

When to use:

  • Complex workflows
  • Multiple levels of abstraction
  • Need for oversight

Example:

Python
# Level 1: General triage
triage_agent = Agent(
    name="triage",
    instructions="Triage requests to appropriate category",
    handoffs=[handoff(billing_supervisor), handoff(technical_supervisor)],
)

# Level 2: Supervisors
billing_supervisor = Agent(
    name="billing_supervisor",
    instructions="Oversee billing specialists",
    handoffs=[handoff(payment_specialist), handoff(refund_specialist)],
)

technical_supervisor = Agent(
    name="technical_supervisor",
    instructions="Oversee technical specialists",
    handoffs=[handoff(debugging_specialist), handoff(maintenance_specialist)],
)

# Level 3: Specialists
payment_specialist = Agent(name="payment", instructions="Handle payment issues")
refund_specialist = Agent(name="refund", instructions="Handle refund requests")
debugging_specialist = Agent(name="debugging", instructions="Debug issues")
maintenance_specialist = Agent(name="maintenance", instructions="Handle maintenance")

Pros:

  • Clear chain of command
  • Scalable to many specialists
  • Good for large organizations

Cons:

  • More complex hierarchy
  • Longer handoff chains
  • Potential for bottlenecks
4. Collaborative Pattern

Description: Agents collaborate on tasks, with each contributing to the final result.

When to use:

  • Multi-step processes
  • Different agents have different skills
  • Need for peer review

Example:

Python
researcher = Agent(
    name="researcher",
    instructions="Research the topic and gather information",
    handoffs=[handoff(writer)],
)

writer = Agent(
    name="writer",
    instructions="Write content based on research",
    handoffs=[handoff(reviewer)],
)

reviewer = Agent(
    name="reviewer",
    instructions="Review and approve the content",
)

# Sequential collaboration
research_result = await Runner.run(researcher, "Research AI")
write_result = await Runner.run(writer, research_result.final_output)
review_result = await Runner.run(reviewer, write_result.final_output)

Pros:

  • Leverages different skills
  • Built-in review process
  • Quality assurance

Cons:

  • Sequential execution (slower)
  • More complex coordination
  • Potential for bottlenecks
5. Parallel Pattern

Description: Multiple agents work in parallel on the same task.

When to use:

  • Independent subtasks
  • Need for speed
  • Redundancy for reliability

Example:

Python
import asyncio

agent1 = Agent(name="agent1", instructions="Approach A")
agent2 = Agent(name="agent2", instructions="Approach B")
agent3 = Agent(name="agent3", instructions="Approach C")

# Parallel execution
results = await asyncio.gather(
    Runner.run(agent1, input),
    Runner.run(agent2, input),
    Runner.run(agent3, input),
)

# Combine results
final_result = combine_results(results)

Pros:

  • Faster execution
  • Redundancy (reliability)
  • Multiple perspectives

Cons:

  • Higher cost
  • Need to combine results
  • Potential for conflicts

Workflow Patterns

1. Linear Workflow

Description: Sequential execution of agents, each passing output to the next.

When to use:

  • Clear sequential steps
  • Each step depends on previous
  • Simple pipelines

Example:

Python
step1 = Agent(name="step1", instructions="First step")
step2 = Agent(name="step2", instructions="Second step")
step3 = Agent(name="step3", instructions="Third step")

# Execute sequentially
result1 = await Runner.run(step1, input)
result2 = await Runner.run(step2, result1.final_output)
result3 = await Runner.run(step3, result2.final_output)

Pros:

  • Simple to implement
  • Easy to debug
  • Clear data flow

Cons:

  • Sequential (slow)
  • No parallelism
  • Single point of failure per step
2. Branching Workflow

Description: Workflow branches based on conditions.

When to use:

  • Conditional logic
  • Different paths for different inputs
  • Decision trees

Example:

Python
triage = Agent(name="triage", instructions="Determine path")
path_a = Agent(name="path_a", instructions="Handle path A")
path_b = Agent(name="path_b", instructions="Handle path B")

result = await Runner.run(triage, input)

if "path_a" in result.final_output.lower():
    final_result = await Runner.run(path_a, result.final_output)
else:
    final_result = await Runner.run(path_b, result.final_output)

Pros:

  • Flexible logic
  • Handles different scenarios
  • Can optimize for common paths

Cons:

  • More complex logic
  • Harder to test all paths
  • Potential for path explosion
3. Loop Workflow

Description: Workflow loops until a condition is met.

When to use:

  • Iterative processes
  • Refinement tasks
  • Unknown number of iterations

Example:

Python
refiner = Agent(name="refiner", instructions="Refine the output")

result = await Runner.run(agent, input)
for i in range(5):  # Max 5 iterations
    if is_satisfactory(result.final_output):
        break
    result = await Runner.run(refiner, result.final_output)

Pros:

  • Handles iterative processes
  • Can refine results
  • Flexible iteration count

Cons:

  • Can be slow
  • Risk of infinite loops
  • Harder to predict runtime
4. Fan-Out/Fan-In Workflow

Description: Fan-out to multiple agents, then fan-in to combine results.

When to use:

  • Parallel subtasks
  • Multiple perspectives
  • Aggregation needed

Example:

Python
# Fan-out
agents = [
    Agent(name="agent1", instructions="Perspective 1"),
    Agent(name="agent2", instructions="Perspective 2"),
    Agent(name="agent3", instructions="Perspective 3"),
]

results = await asyncio.gather(*[
    Runner.run(agent, input) for agent in agents
])

# Fan-in
aggregator = Agent(name="aggregator", instructions="Combine perspectives")
final_result = await Runner.run(
    aggregator,
    "\n".join([r.final_output for r in results])
)

Pros:

  • Parallel execution
  • Multiple perspectives
  • Flexible aggregation

Cons:

  • Higher cost
  • Need good aggregation logic
  • Potential for inconsistent results
5. Human-in-the-Loop Workflow

Description: Workflow pauses for human intervention at key points.

When to use:

  • Critical decisions
  • Approval required
  • Quality gates

Example:

Python
@function_tool(needs_approval=True)
def sensitive_operation() -> str:
    return "Operation performed"

agent = Agent(
    name="agent",
    instructions="Use sensitive tool when needed",
    tools=[sensitive_operation],
)

result = await Runner.run(agent, input)

if result.interruptions:
    # Human reviews
    for interruption in result.interruptions:
        if should_approve(interruption):
            state = result.to_state()
            state.context.approve_tool(interruption)
            result = await Runner.run(agent, state)

Pros:

  • Human oversight
  • Quality control
  • Risk mitigation

Cons:

  • Slower execution
  • Requires human availability
  • Potential for bottlenecks

Data Flow Patterns

1. Request-Response Pattern

Description: Single request, single response.

When to use:

  • Simple queries
  • One-off tasks
  • Stateless operations

Example:

Python
result = await Runner.run(agent, "What is the capital of France?")
print(result.final_output)

Pros:

  • Simple
  • Stateless
  • Easy to cache

Cons:

  • No conversation history
  • Limited context
  • Not suitable for complex tasks
2. Conversational Pattern

Description: Multi-turn conversation with history.

When to use:

  • Chat interfaces
  • Ongoing dialogue
  • Context-dependent tasks

Example:

Python
session = SQLiteSession(db_path="conversations.db")

# Turn 1
result1 = await Runner.run(agent, "Hello", session=session, conversation_id="conv-1")

# Turn 2
result2 = await Runner.run(agent, "What's my name?", session=session, conversation_id="conv-1")

Pros:

  • Maintains context
  • Natural dialogue
  • Builds on previous exchanges

Cons:

  • Stateful
  • Token usage grows
  • More complex to manage
3. Streaming Pattern

Description: Real-time streaming of results.

When to use:

  • User interfaces
  • Long-running operations
  • Real-time feedback

Example:

Python
async for event in Runner.run_streamed(agent, input):
    if isinstance(event, RunItemStreamEvent):
        if isinstance(event.item, MessageOutputItem):
            for content in event.item.raw_item.content:
                if content.type == "text":
                    print(content.text, end="", flush=True)

Pros:

  • Real-time feedback
  • Better UX
  • Lower perceived latency

Cons:

  • More complex implementation
  • Need to handle streams
  • Potential for partial results
4. Batch Pattern

Description: Process multiple requests in batch.

When to use:

  • Bulk operations
  • Offline processing
  • High throughput needed

Example:

Python
inputs = ["input1", "input2", "input3", "input4", "input5"]

results = await asyncio.gather(*[
    Runner.run(agent, input) for input in inputs
])

for input, result in zip(inputs, results):
    print(f"{input}: {result.final_output}")

Pros:

  • High throughput
  • Efficient resource use
  • Good for bulk operations

Cons:

  • Higher cost
  • Need to handle failures
  • Potential for rate limiting
5. Event-Driven Pattern

Description: Agents triggered by events.

When to use:

  • Event-based systems
  • Reactive workflows
  • Asynchronous processing

Example:

Python
async def handle_event(event):
    """Handle an event."""
    agent = get_agent_for_event(event)
    result = await Runner.run(agent, event.data)
    return result

# Event loop
async for event in event_stream:
    await handle_event(event)

Pros:

  • Reactive
  • Decoupled
  • Scalable

Cons:

  • Complex event handling
  • Hard to debug
  • Potential for event storms

Integration Patterns

1. API Gateway Pattern

Description: Agents behind an API gateway.

When to use:

  • Web applications
  • Mobile apps
  • Multi-client support

Example:

Python
from fastapi import FastAPI

app = FastAPI()

@app.post("/agent")
async def run_agent_endpoint(request: AgentRequest):
    """Run agent via API."""
    result = await Runner.run(agent, request.input)
    return {"output": result.final_output}

Pros:

  • Centralized access
  • Authentication/authorization
  • Rate limiting

Cons:

  • Additional layer
  • Potential bottleneck
  • More infrastructure
2. Microservices Pattern

Description: Each agent as a separate microservice.

When to use:

  • Large systems
  • Independent scaling
  • Team ownership

Example:

Python
# Service 1: Billing Agent Service
@app.post("/billing")
async def billing_service(request):
    result = await Runner.run(billing_agent, request.input)
    return result

# Service 2: Technical Agent Service
@app.post("/technical")
async def technical_service(request):
    result = await Runner.run(technical_agent, request.input)
    return result

# Orchestration Service
@app.post("/orchestrate")
async def orchestrate(request):
    if request.type == "billing":
        return await call_billing_service(request)
    elif request.type == "technical":
        return await call_technical_service(request)

Pros:

  • Independent scaling
  • Team ownership
  • Failure isolation

Cons:

  • More infrastructure
  • Network overhead
  • Distributed complexity
3. Sidecar Pattern

Description: Agent as a sidecar to main application.

When to use:

  • Existing applications
  • Add AI capabilities
  • Minimal changes to main app

Example:

Python
# Main application
def main_app_logic():
    """Main application logic."""
    result = do_work()
    return result

# Sidecar agent
def enhance_with_agent(result):
    """Enhance result with agent."""
    agent_result = await Runner.run(agent, str(result))
    return agent_result.final_output

# Combined
result = main_app_logic()
enhanced = enhance_with_agent(result)

Pros:

  • Minimal changes to main app
  • Decoupled
  • Easy to remove

Cons:

  • Additional dependency
  • Network overhead
  • Potential for inconsistency
4. Plugin Pattern

Description: Agents as plugins to extensible system.

When to use:

  • Extensible applications
  • Third-party integrations
  • Modular architecture

Example:

Python
class AgentPlugin:
    """Base class for agent plugins."""
    
    def __init__(self, agent: Agent):
        self.agent = agent
    
    async def process(self, input: str) -> str:
        result = await Runner.run(self.agent, input)
        return result.final_output

# Register plugins
registry.register("billing", AgentPlugin(billing_agent))
registry.register("technical", AgentPlugin(technical_agent))

# Use plugins
plugin = registry.get("billing")
result = await plugin.process(input)

Pros:

  • Extensible
  • Modular
  • Easy to add/remove

Cons:

  • Plugin interface complexity
  • Compatibility issues
  • Discovery overhead
5. Proxy Pattern

Description: Proxy agents that forward to other services.

When to use:

  • Legacy integration
  • Protocol translation
  • Security boundary

Example:

Python
@function_tool
def legacy_api_call(endpoint: str) -> str:
    """Call legacy API."""
    # Translate to legacy format
    legacy_request = translate_to_legacy(endpoint)
    
    # Call legacy system
    response = legacy_client.call(legacy_request)
    
    # Translate back
    return translate_from_legacy(response)

agent = Agent(
    name="proxy",
    instructions="Proxy to legacy system",
    tools=[legacy_api_call],
)

Pros:

  • Encapsulates legacy
  • Protocol translation
  • Security boundary

Cons:

  • Translation overhead
  • Potential for bugs
  • Additional layer

Deployment Patterns

1. Serverless Pattern

Description: Deploy agents as serverless functions.

When to use:

  • Variable load
  • Cost optimization
  • Simple deployments

Example:

Python
# AWS Lambda
def lambda_handler(event, context):
    """AWS Lambda handler."""
    agent = load_agent()
    result = await Runner.run(agent, event["input"])
    return {"output": result.final_output}

# Google Cloud Functions
@gcp_cloud_functions.http
def gcp_handler(request):
    """GCP Cloud Function handler."""
    agent = load_agent()
    result = await Runner.run(agent, request.json["input"])
    return result.final_output

Pros:

  • Auto-scaling
  • Pay-per-use
  • No server management

Cons:

  • Cold starts
  • Limited execution time
  • Vendor lock-in
2. Container Pattern

Description: Deploy agents in containers.

When to use:

  • Consistent environments
  • Portable deployments
  • Microservices

Example:

Dockerfile
# Dockerfile
FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "80"]
YAML
# docker-compose.yml
version: '3'
services:
  agent-service:
    build: .
    ports:
      - "80:80"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}

Pros:

  • Consistent environment
  • Portable
  • Easy to scale

Cons:

  • Container management
  • Resource overhead
  • Orchestration complexity
3. Kubernetes Pattern

Description: Deploy agents on Kubernetes.

When to use:

  • Large scale
  • Complex orchestration
  • Production workloads

Example:

YAML
# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: agent-service
spec:
  replicas: 3
  selector:
    matchLabels:
      app: agent-service
  template:
    metadata:
      labels:
        app: agent-service
    spec:
      containers:
      - name: agent
        image: agent-service:latest
        ports:
        - containerPort: 80
        env:
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: openai-secret
              key: api-key

Pros:

  • Scalable
  • Self-healing
  • Production-grade

Cons:

  • Complex setup
  • Learning curve
  • Overhead for small apps
4. Queue-Based Pattern

Description: Use message queue for agent requests.

When to use:

  • High throughput
  • Decoupling
  • Asynchronous processing

Example:

Python
import asyncio
from fastapi import FastAPI
from aiokafka import AIOKafkaConsumer

app = FastAPI()

async def process_messages():
    """Process messages from Kafka."""
    consumer = AIOKafkaConsumer(
        "agent-requests",
        bootstrap_servers="localhost:9092",
    )
    await consumer.start()
    
    async for message in consumer:
        input = message.value.decode()
        result = await Runner.run(agent, input)
        await send_result(result)

@app.on_event("startup")
async def startup():
    """Start consumer on startup."""
    asyncio.create_task(process_messages())

Pros:

  • Decoupled
  • Scalable
  • Buffering

Cons:

  • Queue management
  • Complexity
  • Potential for backlog
5. Edge Deployment Pattern

Description: Deploy agents at the edge.

When to use:

  • Low latency required
  • Offline capability
  • Privacy requirements

Example:

Python
# Edge device deployment
class EdgeAgentService:
    """Agent service for edge devices."""
    
    def __init__(self, model_path: str):
        self.model = load_local_model(model_path)
        self.agent = Agent(model=self.model)
    
    async def process(self, input: str) -> str:
        """Process input locally."""
        result = await Runner.run(self.agent, input)
        return result.final_output

# Deploy to edge device
service = EdgeAgentService(model_path="/models/local")

Pros:

  • Low latency
  • Offline capability
  • Privacy

Cons:

  • Limited resources
  • Model management
  • Update complexity

Anti-Patterns

1. God Agent Anti-Pattern

Description: Single agent that does everything.

Why avoid:

  • Hard to maintain
  • Hard to test
  • Hard to scale

Solution: Use specialist pattern instead.

2. Tight Coupling Anti-Pattern

Description: Agents tightly coupled to each other.

Why avoid:

  • Hard to change
  • Hard to test
  • Brittle

Solution: Use handoffs with clear interfaces.

3. No Error Handling Anti-Pattern

Description: No error handling in agent workflows.

Why avoid:

  • Crashes on errors
  • Poor user experience
  • Hard to debug

Solution: Implement comprehensive error handling.

4. Hardcoded Configuration Anti-Pattern

Description: Configuration hardcoded in code.

Why avoid:

  • Hard to change
  • Security risk
  • Environment-specific

Solution: Use environment variables and config files.

5. No Monitoring Anti-Pattern

Description: No monitoring or observability.

Why avoid:

  • Can't detect issues
  • Hard to debug
  • No insight into usage

Solution: Implement comprehensive monitoring and tracing.

Pattern Selection Guide

Choose Pattern Based On:
CriteriaPattern
Simple taskSingle Agent
Multiple domainsSpecialist
Complex workflowHierarchical
Multi-step processLinear Workflow
Conditional logicBranching Workflow
Iterative refinementLoop Workflow
Parallel subtasksFan-Out/Fan-In
Human approval neededHuman-in-the-Loop
Web applicationAPI Gateway
Large systemMicroservices
Add AI to existing appSidecar
Extensible systemPlugin
Legacy integrationProxy
Variable loadServerless
Consistent environmentContainer
Production scaleKubernetes
High throughputQueue-Based
Low latencyEdge Deployment

Pattern Combinations

Patterns can be combined for complex systems:

Example: E-Commerce System
Python
# Agent Organization: Specialist Pattern
billing = Agent(name="billing", ...)
shipping = Agent(name="shipping", ...)
support = Agent(name="support", ...)
triage = Agent(name="triage", handoffs=[...])

# Workflow: Branching + Human-in-the-Loop
# Integration: API Gateway
# Deployment: Kubernetes
Example: Customer Support System
Python
# Agent Organization: Hierarchical
triage -> supervisors -> specialists

# Workflow: Human-in-the-Loop for approvals
# Data Flow: Conversational
# Integration: API Gateway
# Deployment: Container

Summary

Architectural patterns provide proven approaches for building agent systems. Key takeaways:

  1. Single Agent - Simple, single-purpose
  2. Specialist - Domain-specific agents
  3. Hierarchical - Supervised specialists
  4. Collaborative - Agents working together
  5. Parallel - Multiple agents simultaneously
  6. Linear Workflow - Sequential steps
  7. Branching - Conditional paths
  8. Loop - Iterative refinement
  9. Fan-Out/Fan-In - Parallel then aggregate
  10. Human-in-the-Loop - Human approval points
  11. Request-Response - Single request
  12. Conversational - Multi-turn dialogue
  13. Streaming - Real-time updates
  14. Batch - Bulk processing
  15. Event-Driven - Reactive processing
  16. API Gateway - Web API layer
  17. Microservices - Independent services
  18. Sidecar - Add-on to main app
  19. Plugin - Extensible modules
  20. Proxy - Legacy integration
  21. Serverless - Cloud functions
  22. Container - Docker deployment
  23. Kubernetes - Orchestration
  24. Queue-Based - Async processing
  25. Edge - Local deployment

Choose patterns based on your requirements, scale, and constraints. Combine patterns as needed for complex systems. Avoid anti-patterns that lead to unmaintainable code.