architecture-synthesis

Architecture Synthesis

Generates a reference architecture specification for a new framework.

Process

1. Define primitives — Message, State, Result, Tool types
2. Specify interfaces — Protocols for LLM, Tool, Memory
3. Design the loop — Core execution algorithm
4. Create diagrams — Visual architecture representation
5. Produce roadmap — Implementation phases

Prerequisites

Before synthesis, ensure you have:

• Comparative matrix with decisions per dimension
• Anti-pattern catalog with "Do Not Repeat" list
• Design requirements document

Core Primitives Definition

Message Type

from typing import Literal
from pydantic import BaseModel

class Message(BaseModel):
    """Immutable message in the conversation."""
    role: Literal["system", "user", "assistant", "tool"]
    content: str
    name: str | None = None  # For tool messages
    tool_call_id: str | None = None
    
    class Config:
        frozen = True  # Immutable

State Type

from dataclasses import dataclass, field
from typing import Any

@dataclass(frozen=True)
class AgentState:
    """Immutable agent state - copy-on-write pattern."""
    messages: tuple[Message, ...]
    tool_results: tuple[ToolResult, ...] = ()
    metadata: dict[str, Any] = field(default_factory=dict)
    step_count: int = 0
    
    def with_message(self, msg: Message) -> "AgentState":
        """Return new state with message added."""
        return AgentState(
            messages=(*self.messages, msg),
            tool_results=self.tool_results,
            metadata=self.metadata,
            step_count=self.step_count
        )

Result Types

from typing import Union

@dataclass(frozen=True)
class ToolResult:
    """Result from tool execution."""
    tool_name: str
    success: bool
    output: str | None = None
    error: str | None = None
    
@dataclass(frozen=True)
class AgentFinish:
    """Agent completed its task."""
    output: str
    
@dataclass(frozen=True)
class AgentContinue:
    """Agent needs another step."""
    tool_calls: tuple[ToolCall, ...]

StepResult = Union[AgentFinish, AgentContinue]

Interface Protocols

LLM Protocol

from typing import Protocol, Iterator

class LLM(Protocol):
    """Minimal LLM interface."""
    
    def generate(self, messages: list[Message]) -> LLMResponse:
        """Generate a response."""
        ...
    
    def stream(self, messages: list[Message]) -> Iterator[str]:
        """Stream response tokens."""
        ...

@dataclass
class LLMResponse:
    """Full LLM response with metadata."""
    content: str
    tool_calls: list[ToolCall] | None
    usage: TokenUsage
    model: str
    raw: Any  # Original API response

Tool Protocol

class Tool(Protocol):
    """Minimal tool interface."""
    
    @property
    def name(self) -> str:
        """Tool identifier."""
        ...
    
    @property
    def description(self) -> str:
        """Human-readable description."""
        ...
    
    @property
    def schema(self) -> dict:
        """JSON Schema for parameters."""
        ...
    
    def execute(self, **kwargs) -> str:
        """Execute the tool."""
        ...

Memory Protocol

class Memory(Protocol):
    """Memory/context management interface."""
    
    def add(self, message: Message) -> None:
        """Add a message to memory."""
        ...
    
    def get_context(self, query: str, max_tokens: int) -> list[Message]:
        """Retrieve relevant context."""
        ...
    
    def clear(self) -> None:
        """Clear memory."""
        ...

Execution Loop Design

Algorithm Pseudocode

FUNCTION run_agent(input: str, max_steps: int) -> str:
    state = initial_state(input)
    
    FOR step IN range(max_steps):
        # 1. Build context
        messages = build_messages(state)
        
        # 2. Call LLM
        response = llm.generate(messages)
        
        # 3. Parse and decide
        result = parse_response(response)
        
        # 4. Handle result
        IF result IS AgentFinish:
            RETURN result.output
        
        IF result IS AgentContinue:
            # Execute tools
            FOR tool_call IN result.tool_calls:
                tool_result = execute_tool(tool_call)
                state = state.with_tool_result(tool_result)
            
            # Feed back to LLM
            state = state.with_message(format_observations(state))
        
        # 5. Emit events
        emit("step_complete", state)
    
    # Max steps reached
    RAISE MaxStepsExceeded(state)

Implementation Template

class Agent:
    def __init__(
        self,
        llm: LLM,
        tools: list[Tool],
        system_prompt: str,
        max_steps: int = 10
    ):
        self.llm = llm
        self.tools = {t.name: t for t in tools}
        self.system_prompt = system_prompt
        self.max_steps = max_steps
        self.callbacks: list[Callback] = []
    
    def run(self, input: str) -> str:
        state = AgentState(messages=(
            Message(role="system", content=self.system_prompt),
            Message(role="user", content=input)
        ))
        
        for step in range(self.max_steps):
            self._emit("step_start", step, state)
            
            # LLM call
            response = self.llm.generate(list(state.messages))
            self._emit("llm_response", response)
            
            # Parse
            result = self._parse_response(response)
            
            # Finish or continue
            if isinstance(result, AgentFinish):
                self._emit("agent_finish", result)
                return result.output
            
            # Execute tools
            for call in result.tool_calls:
                tool_result = self._execute_tool(call)
                state = state.with_tool_result(tool_result)
            
            # Update state
            state = state.with_message(
                Message(role="assistant", content=response.content)
            )
            for tr in state.tool_results[-len(result.tool_calls):]:
                state = state.with_message(
                    Message(role="tool", content=tr.output or tr.error, name=tr.tool_name)
                )
            
            self._emit("step_end", step, state)
        
        raise MaxStepsExceeded(f"Exceeded {self.max_steps} steps")
    
    def _execute_tool(self, call: ToolCall) -> ToolResult:
        tool = self.tools.get(call.name)
        if not tool:
            return ToolResult(call.name, success=False, error=f"Unknown tool: {call.name}")
        
        try:
            output = tool.execute(**call.arguments)
            return ToolResult(call.name, success=True, output=output)
        except Exception as e:
            return ToolResult(call.name, success=False, error=f"{type(e).__name__}: {e}")

Architecture Diagram

graph TB
    subgraph "Core Layer"
        MSG[Message]
        STATE[AgentState]
        RESULT[StepResult]
    end
    
    subgraph "Protocol Layer"
        LLM_P[LLM Protocol]
        TOOL_P[Tool Protocol]
        MEM_P[Memory Protocol]
    end
    
    subgraph "Execution Layer"
        LOOP[Agent Loop]
        PARSER[Response Parser]
        EXECUTOR[Tool Executor]
    end
    
    subgraph "Integration Layer"
        OPENAI[OpenAI LLM]
        ANTHROPIC[Anthropic LLM]
        TOOLS[Built-in Tools]
        VECTOR[Vector Memory]
    end
    
    MSG --> STATE
    STATE --> LOOP
    LOOP --> LLM_P
    LOOP --> PARSER
    PARSER --> RESULT
    RESULT --> EXECUTOR
    EXECUTOR --> TOOL_P
    
    LLM_P -.-> OPENAI
    LLM_P -.-> ANTHROPIC
    TOOL_P -.-> TOOLS
    MEM_P -.-> VECTOR

Implementation Roadmap

Phase 1: Core (Week 1-2)

• Define Message, State, Result types
• Implement LLM Protocol with OpenAI
• Implement basic Tool Protocol
• Create minimal Agent loop
• Add step limit termination

Phase 2: Robustness (Week 3-4)

• Add error handling and feedback
• Implement retry mechanisms
• Add comprehensive logging
• Create callback/event system
• Add token counting

Phase 3: Extensibility (Week 5-6)

• Add Memory Protocol
• Implement vector store integration
• Create tool discovery/registry
• Add configuration system
• Write documentation

Phase 4: Production (Week 7-8)

• Add tracing/observability
• Implement streaming
• Add rate limiting
• Create async version
• Performance optimization

Output Artifacts

reference-architecture/
├── docs/
│   ├── ARCHITECTURE.md      # This document
│   ├── PRIMITIVES.md        # Type definitions
│   ├── PROTOCOLS.md         # Interface specs
│   └── LOOP.md              # Algorithm details
├── diagrams/
│   ├── architecture.mermaid
│   ├── flow.mermaid
│   └── types.mermaid
├── examples/
│   ├── simple_agent.py
│   ├── multi_tool_agent.py
│   └── custom_llm.py
└── ROADMAP.md               # Implementation plan

Integration

• Inputs from: comparative-matrix, antipattern-catalog
• Produces: Reference architecture for implementation
• Validates against: Original protocol requirements