Chapter 3: Autonomous Agent Demo - System Architecture

Overview

This project demonstrates six key standardization patterns for building reliable, maintainable autonomous AI agents. It showcases best practices for production-ready agent systems through a working RAG (Retrieval-Augmented Generation) implementation.

🔒 Strong Typing

Pydantic models with runtime validation and mypy strict type checking

⚠️ Error Handling

Result<T, E> pattern for explicit error states without exceptions

📊 Structured Logging

JSON logs with correlation IDs for distributed tracing

⚡ Async Orchestration

asyncio-based concurrent operations for better performance

📝 Versioned Prompts

Declarative prompt management with A/B testing support

🔌 Plugin Registry

Protocol-based tool registry for extensible architecture

System Architecture

The system follows a clean, modular architecture with clear separation of concerns:

Data Flow

CLI Entry: User query enters through the CLI interface
Agent Orchestration: Agent core loads prompts and orchestrates async operations
Parallel Retrieval: Multiple data sources queried concurrently using asyncio
Decision Making: Results validated through Pydantic models and processed
Structured Output: JSON logs with correlation IDs track the entire flow

Standardization Patterns

1. Strong Typing & Validation

class Passage(BaseModel):
    id: str
    text: str
    score: float = Field(ge=0, le=1)  # Runtime validation

@runtime_checkable
class Tool(Protocol):
    def execute(self, **kwargs) -> dict: ...

Benefits: Catch errors at design time, self-documenting APIs, runtime validation prevents invalid data propagation.

2. Result-Based Error Handling

@dataclass
class Result(Generic[T, E]):
    ok: Optional[T] = None
    err: Optional[E] = None

    @property
    def is_ok(self) -> bool: return self.err is None

# Usage
result = decide(passages)
if result.is_err:
    log.info("plan_failed", extra={"error": result.err})
    return {"error": result.err}

Benefits: Explicit error states, no silent failures, forces error handling at call sites.

3. Structured JSON Logging

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "level": record.levelname,
            "msg": record.getMessage(),
            "logger": record.name,
            "ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S"),
            "correlation_id": getattr(record, "correlation_id", None),
        }
        return json.dumps(payload)

Benefits: Machine-readable logs, easy to parse and analyze, correlation IDs enable distributed tracing.

4. Async Orchestration

async def plan(q: str) -> Result[Decision, str]:
    log.info("plan_start", extra={"query": q})
    p_task = asyncio.create_task(get_passages(q))
    w_task = asyncio.create_task(get_web(q))
    passages_a, passages_b = await asyncio.gather(p_task, w_task)
    # Process results...

Benefits: Concurrent I/O operations, better resource utilization, cleaner than threading.

5. Versioned Prompt Management

{
  "name": "summarize",
  "version": "v1",
  "system": "You are a precise technical assistant...",
  "user_template": "Summarize the following text in at most ${max_bullets} bullet points.",
  "constraints": {
    "max_tokens": 400
  }
}

Benefits: Prompts versioned in git, A/B testing via environment variables, audit trail for prompt changes.

6. Plugin Registry Pattern

registry = Registry()
registry.register("search", SearchTool())

# Anywhere in code:
result = registry.use("search", query="RAG systems")

# Add new tools without modifying agent code
registry.register("summarizer", SummarizerTool())

Benefits: Decoupled architecture, easy to add/remove tools, testable in isolation.

Core Components

cli.py

Entry point for the agent-demo command. Sets up environment and runs the agent with a sample query.

agent-demo

agent.py

Core agent logic with async orchestration. Manages retrieval, decision-making, and prompt rendering.

run_agent() | plan()

prompt_store.py

Loads and renders versioned prompts from JSON files with template substitution.

PromptStore.load()

registry.py

Protocol-based plugin system for registering and using tools dynamically.

Registry.register()

result.py

Generic Result<T, E> type for explicit error handling without exceptions.

Result[Decision, str]

models.py

Pydantic models with runtime validation for Passages and Decisions.

Passage | Decision

logging.py

Custom JSON formatter with correlation IDs for distributed tracing.

JsonFormatter

demo.py

Bootstrap script that creates venv, installs package, and runs the demo.

python demo.py

Getting Started

Quick Start

# Clone the repository
git clone https://github.com/ranjanarajendran/engineering-autonomous-ai.git
cd engineering-autonomous-ai/chapter3-agentic-standardization-demo

# Run the one-command demo
python demo.py

# Output: JSON with prompt details, decision, and citations

Run Tests

python -m venv .venv && source .venv/bin/activate
pip install -e .
pip install pytest
pytest -q

Type Checking

pip install mypy
mypy --strict src/autonomous_agent_demo

Switch Prompt Versions

export PROMPT_VARIANT=v2
python demo.py

Why Standardization Matters

Without these patterns, autonomous agents suffer from:

Silent Failures: Errors hidden in try-catch blocks or ignored return codes
Type Confusion: Runtime type errors that could have been caught earlier
Debugging Nightmares: Unstructured logs making it impossible to trace execution
Thread Complexity: Race conditions and deadlocks from manual thread management
Prompt Chaos: String-embedded prompts scattered throughout code
Tight Coupling: Agent logic intertwined with tool implementations

This demo shows how adopting industry-standard patterns creates agents that are observable, testable, and maintainable at scale.