Skip to content

architecture.md

Latest commit

 

History

History
186 lines (151 loc) · 7.74 KB

architecture.md

File metadata and controls

186 lines (151 loc) · 7.74 KB
Raw

AgentFlow Architecture

Overview

AgentFlow is a monorepo consisting of two workspaces:

  • @agentflow/api — Express + TypeScript backend (port 3001)
  • @agentflow/web — React + TypeScript frontend (port 5173)

System Diagram

┌───────────────────────────────────────────────────┐
│                    Browser                         │
│  ┌─────────────────────────────────────────────┐  │
│  │            React Frontend (Vite)             │  │
│  │                                              │  │
│  │  ┌──────────┐  ┌──────────┐  ┌──────────┐  │  │
│  │  │  Canvas  │  │ Library  │  │Dashboard │  │  │
│  │  │ (R.Flow) │  │Templates │  │ Metrics  │  │  │
│  │  └──────────┘  └──────────┘  └──────────┘  │  │
│  │                                              │  │
│  │           Zustand State Store                │  │
│  └─────────────────────────────────────────────┘  │
└──────────────┬────────────────────┬───────────────┘
               │  REST API          │  SSE Stream
               ▼                    ▼
┌───────────────────────────────────────────────────┐
│              Express API (Node.js)                 │
│                                                    │
│  ┌────────────────┐   ┌─────────────────────────┐ │
│  │   REST Routes  │   │    Orchestration Engine  │ │
│  │  /pipelines    │   │                          │ │
│  │  /executions   │   │  DAG Traversal (Kahn's)  │ │
│  └────────────────┘   │  Parallel Wave Execution │ │
│                       │  SSE Token Streaming     │ │
│  ┌────────────────┐   └──────────┬──────────────┘ │
│  │  SQLite (WAL)  │              │                 │
│  │  pipelines     │              ▼                 │
│  │  executions    │   ┌─────────────────────────┐ │
│  └────────────────┘   │    Claude API Wrapper   │ │
│                       │  claude-sonnet-4-...    │ │
└───────────────────────┴─────────────────────────┘
                                    │
                                    ▼
                         ┌─────────────────┐
                         │  Anthropic API  │
                         └─────────────────┘

Key Design Decisions

1. DAG Execution with Kahn's BFS

Pipelines are directed acyclic graphs. The orchestration engine uses Kahn's topological sort algorithm, executing nodes in "waves" — all nodes whose predecessors are complete execute concurrently via Promise.all.

Wave 1: [A, B]   (no predecessors — execute in parallel)
Wave 2: [C]      (depends on A, B — executes after both complete)
Wave 3: [D]      (depends on C)

This naturally handles both sequential pipelines and fan-in/fan-out patterns.

2. SSE for Real-Time Streaming

Server-Sent Events are used instead of WebSockets because:

  • Unidirectional (server → client) matches the streaming use case perfectly
  • No handshake overhead
  • Automatic reconnection built into EventSource
  • Works through most proxies and firewalls

Each execution gets its own SSE channel at GET /api/executions/:id/stream.

3. Agent Context Propagation

When an agent runs, it receives a prompt containing:

  • The original user input
  • Outputs from all predecessor agents (separated by ---)

This allows downstream agents to build on upstream work without needing explicit routing logic.

4. SQLite with WAL Mode

SQLite with Write-Ahead Logging provides:

  • Zero external dependencies for development
  • ACID compliance for execution records
  • Concurrent reads during writes (important for SSE streaming while writing)
  • Easy deployment — single file, no server process

5. In-Memory DB Injection for Tests

The setDb() function allows tests to inject a fresh :memory: SQLite instance, giving each test suite complete isolation without file I/O or cleanup complexity.

Data Models

Pipeline

{
  id: string          // UUID
  name: string
  description: string
  nodes: PipelineNode[]   // Agent configs + React Flow positions
  edges: PipelineEdge[]   // Directed connections between nodes
  createdAt: string   // ISO 8601
  updatedAt: string
}

Execution

{
  id: string
  pipelineId: string
  status: 'pending' | 'running' | 'completed' | 'failed'
  input: string
  output: string           // Final output from terminal node(s)
  agents: AgentExecution[] // Per-agent results
  totalInputTokens: number
  totalOutputTokens: number
  durationMs: number
  startedAt: string
  completedAt: string | null
  error: string | null
}

SSE Event Protocol

Event Type When Payload
execution_start Pipeline begins { executionId }
agent_start Agent begins { agentId, agentName }
agent_token Each token streamed { agentId, token }
agent_done Agent completes { agentId, agentResult }
agent_error Agent fails { agentId, error }
execution_done All agents done { result: Execution }
execution_error Pipeline failed { error }

File Structure

agentflow/
├── apps/
│   ├── api/src/
│   │   ├── config.ts          Environment validation
│   │   ├── types.ts           All TypeScript types
│   │   ├── db/index.ts        SQLite setup + migrations
│   │   ├── models/
│   │   │   ├── pipeline.ts    Pipeline CRUD
│   │   │   └── execution.ts   Execution CRUD + metrics
│   │   ├── services/
│   │   │   ├── claude.ts      Anthropic SDK wrapper
│   │   │   ├── sse.ts         SSE connection manager
│   │   │   └── orchestration.ts  DAG execution engine
│   │   ├── routes/
│   │   │   ├── pipelines.ts   CRUD routes
│   │   │   └── executions.ts  Run + stream routes
│   │   └── middleware/
│   │       └── errorHandler.ts  Validation + errors
│   └── web/src/
│       ├── components/
│       │   ├── Canvas/        React Flow pipeline builder
│       │   ├── Execution/     Live streaming output panel
│       │   ├── Library/       Template browser
│       │   └── Dashboard/     Metrics + run history
│       ├── store/index.ts     Zustand global state
│       ├── utils/api.ts       Typed fetch + SSE client
│       └── utils/templates.ts Built-in pipeline templates
└── tests/api/
    ├── pipelines.test.ts      Route integration tests
    └── orchestration.test.ts  Engine unit tests