AI Agent Pipeline System - Implementation Spec

## Background & Motivation

Build a modular AI pipeline system that takes multimedia documents as input and automatically composes a set of relevant AI agents to analyze the content and produce insights. The system should intelligently select which agents to run based on the input content and execute them in the correct order based on their dependencies.

**Example Flow:**
```
Input: Paper on "Quantum Nanodots in OLED Displays"
↓
System evaluates all available agents
↓
Selected: PhysicistAgent, MarketAnalysisAgent, TechnicalFeasibilityAgent
Skipped: CreativeEndeavorAgent, LegalComplianceAgent
↓
Build dependency graph (e.g., PhysicistAgent → TechnicalFeasibilityAgent)
↓
Execute DAG and collect insights
```

## System Architecture

### Agent Structure
Each agent is a directory containing:

```
agents/
├── physicist_agent/
│ ├── TRIGGER.md # When should this agent run?
│ ├── SKILL.md # What does this agent do?
│ ├── tools/ # Agent-specific tools (optional)
│ │ └── equation_solver.py
│ └── config.json # Metadata (dependencies, timeout, etc.)
```

**config.json schema (example structure):**
```json
{
"name": "physicist_agent",
"version": "1.0.0",
"dependencies": ["prerequisite_agent"],
"timeout_seconds": 60,
"description": "Analyzes physical and scientific concepts"
}
```

### Core Components

1. **Agent Loader**: Discovers and loads agents from the agents directory
2. **Trigger Evaluator**: Runs TRIGGER.md prompts to determine which agents should execute
3. **DAG Builder**: Constructs execution graph from agent dependencies
4. **Pipeline Executor Interface**: Abstraction layer for execution platforms (Temporal, Airflow, etc.)
5. **Tool System**: Registry for shared and agent-specific tools
6. **Context Manager**: Maintains shared state between agents

### Project Structure
```
ai-agent-pipeline/
├── src/
│ └── agent_pipeline/
│ ├── agent.py # Agent data class
│ ├── loader.py # AgentLoader
│ ├── trigger.py # TriggerEvaluator
│ ├── dag.py # DAGBuilder
│ ├── executor.py # PipelineExecutor interface + InMemoryExecutor
│ ├── context.py # Context management
│ ├── tool.py # Tool definitions and registry
│ └── tools/ # Shared tool implementations
├── agents/ # Agent definitions
├── tests/ # Unit and integration tests
└── examples/ # Sample documents
```

## Phase 1: Foundation

### Milestone 1.1: Project Setup & Agent Definition
- [x] Create `Agent` class representing an agent
- [x] Implement `AgentLoader` that discovers agents from directory
- [x] Parse TRIGGER.md, SKILL.md, config.json files
- [x] Create 3 example agents for testing

### Milestone 1.2: Trigger Evaluation
- [x] Create `TriggerEvaluator` class
- [x] Integrate with Anthropic API for trigger evaluation
- [x] Support multimodal inputs (text + images)
- [x] Return structured decision (should_run, confidence, reasoning)

**Environment Setup:**
Use `uv`. Install dependencies with:
```bash
uv pip install anthropic
```

## Phase 2: DAG Construction & Execution

### Milestone 2.1: DAG Builder
- [ ] Create `DAGBuilder` class
- [ ] Parse dependencies from agent configs
- [ ] Detect circular dependencies
- [ ] Implement topological sort (Kahn's algorithm or DFS-based)
- [ ] Output execution plan

### Milestone 2.2: Pipeline Executor Interface
- [ ] Design `PipelineExecutor` abstract base class
- [ ] Define execution contract (input: plan + document, output: results)
- [ ] Implement `InMemoryExecutor` for testing
- [ ] Design context passing between agents
- [ ] Document how to implement for Temporal.io, Airflow, etc.

```python
class PipelineExecutor(ABC):
@abstractmethod
def execute(self, execution_plan: ExecutionPlan, input_doc: Document) -> PipelineResults:
pass

class InMemoryExecutor(PipelineExecutor):
def execute(self, execution_plan: ExecutionPlan, input_doc: Document) -> PipelineResults:
# Sequential execution for POC
pass
```

**Context structure:**
```python
{
"input_document": {...},
"agent_outputs": {
"physicist_agent": {
"output": "Analysis text...",
"metadata": {...},
"timestamp": "2024-01-15T10:30:00Z"
}
}
}
```

### Milestone 2.3: Tool System
- [ ] Define `Tool` interface (name, description, parameters, execute callable)
- [ ] Implement `ToolRegistry` for tool discovery and management
- [ ] Support both shared tools (in `src/agent_pipeline/tools/`) and agent-specific tools
- [ ] Integrate with LLM function calling
- [ ] Create 2-3 example tools (web_search, calculate, data_parser)
- [ ] Handle tool execution and errors

```python
class Tool:
name: str
description: str
parameters: dict
execute: callable

# Option 1: Registry loads tools (pull pattern)
tool_registry = ToolRegistry()
tool_registry.register_global_tools(load_tools_from("./tools"))
tool_registry.register_agent_tools("physicist_agent", load_tools_from("./agents/physicist_agent/tools"))

# Option 2: Agent registers its own tools (inversion of control)
physicist_agent = Agent(...) # Agent loaded by AgentLoader
physicist_agent.register_tools(tool_registry) # Agent knows and registers its own tools
available_tools = tool_registry.get_tools_for_agent(physicist_agent)
```

## Phase 3: Integration

### Milestone 3.1: Output & CLI
- [ ] Aggregate agent outputs
- [ ] Generate structured output (JSON, Markdown)
- [ ] Export DAG visualization
- [ ] Create CLI interface

```bash
python -m agent_pipeline analyze document.pdf --output report.json
```

## Concepts to Review

- **Directed Acyclic Graphs (DAGs)**: Understanding topological ordering
- **Dependency Resolution**: How package managers solve this problem
- **LLM Prompt Engineering**: Writing effective trigger and skill prompts
- **Multimodal AI**: Working with text + images in LLM APIs