From caa11bcd653ac7db375006b27087c08fa776a89b Mon Sep 17 00:00:00 2001 From: Adrian Lyjak Date: Thu, 12 Feb 2026 17:15:04 -0500 Subject: [PATCH] High level architecture docs (#357) --- AGENTS.md | 7 ++ architecture-docs/control-loop.md | 79 +++++++++++++++ architecture-docs/core-overview.md | 116 +++++++++++++++++++++++ architecture-docs/server-architecture.md | 97 +++++++++++++++++++ 4 files changed, 299 insertions(+) create mode 100644 architecture-docs/control-loop.md create mode 100644 architecture-docs/core-overview.md create mode 100644 architecture-docs/server-architecture.md diff --git a/AGENTS.md b/AGENTS.md index 96c4256f..2074950d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -42,6 +42,13 @@ uv run pre-commit run -a - `packages/llama-index-workflows/tests/` - Test suite - `examples/` - Usage examples +## Architecture + +See `architecture-docs/` for high-level architectural overviews: +- [`core-overview.md`](architecture-docs/core-overview.md) — Workflow, Context, Runtime, and event flow +- [`control-loop.md`](architecture-docs/control-loop.md) — The reducer-based execution engine +- [`server-architecture.md`](architecture-docs/server-architecture.md) — HTTP server, persistence, and runtime decorators + ## Key Components - **Workflow** - Main orchestration class - **Context** - State management across workflow steps diff --git a/architecture-docs/control-loop.md b/architecture-docs/control-loop.md new file mode 100644 index 00000000..71992294 --- /dev/null +++ b/architecture-docs/control-loop.md @@ -0,0 +1,79 @@ +# Control Loop Architecture + +The control loop is the core execution engine for workflows. It follows a **reducer pattern** — pure state transitions with side effects expressed as commands: + +``` +State + Tick --> (NewState, Commands) +``` + +[`control_loop.py`](../packages/llama-index-workflows/src/workflows/runtime/control_loop.py) + +## Main Loop + +```mermaid +flowchart TD + A[Initialize: queue StartEvent, schedule timeout] --> B[Drain tick buffer] + B --> C{Buffer empty?} + C -- No --> D[Reduce tick --> state + commands] + D --> E[Execute commands] + E --> B + C -- Yes --> F[Wait for next completion] + F --> G{What completed?} + G -- Timeout --> H[Pop due scheduled ticks into buffer] + G -- External tick --> I[Add to buffer] + G -- Worker result --> J[Add TickStepResult to buffer] + H --> B + I --> B + J --> B +``` + +1. **Initialize** — Queue `StartEvent`, schedule workflow timeout, rewind any in-progress work from a prior run. +2. **Drain tick buffer** — Process all queued ticks synchronously. Each tick runs through the reducer and its commands execute before the next tick. +3. **Wait for next completion** — Build a task set (worker tasks + one pull task), then wait for the first to complete. Workers have priority over pull tasks. +4. **Process completed task** — Route the result back into the tick buffer and loop. + +## Ticks and Commands + +**Ticks** are inputs to the reducer. They represent things that happen: events arriving, steps completing, cancellation requests, timeouts, and publish requests from steps. Each tick type dispatches to a dedicated reducer function. + +[`types/ticks.py`](../packages/llama-index-workflows/src/workflows/runtime/types/ticks.py) — all tick types + +**Commands** are outputs from the reducer — the side effects the loop executes. They represent actions to take: spawning step workers, queuing events (with optional delays), completing or failing the run, and publishing events to the external stream. + +[`types/commands.py`](../packages/llama-index-workflows/src/workflows/runtime/types/commands.py) — all command types + +## Runtime Integration + +The control loop is runtime-agnostic. It talks to the outside world exclusively through `InternalRunAdapter` (see [core-overview.md — Runtime and Adapters](./core-overview.md#runtime-and-adapters)). This is the extension point — runtime decorators wrap the adapter to add behavior like tick persistence, idle detection, or event recording. + +```mermaid +sequenceDiagram + participant CL as Control Loop + participant A as InternalRunAdapter + participant Ext as External (handler/client) + + Note over CL: Main loop iteration + CL->>A: wait_receive() [pull task] + Ext-->>A: send_event() delivers tick + A-->>CL: WaitResultTick + + CL->>CL: reduce tick --> (state, commands) + CL->>A: on_tick(tick) [journaling hook] + + Note over CL: Execute commands + CL->>A: write_to_event_stream(event) + CL->>CL: spawn worker task + + CL->>A: wait_for_next_task(task_set, timeout) + A-->>CL: completed task (worker or pull) +``` + +[`plugin.py`](../packages/llama-index-workflows/src/workflows/runtime/types/plugin.py) — full adapter interface + +## Key Design Decisions + +- **Deterministic replay** — The reducer is pure. Adapters can record ticks and replay them to reconstruct state, and override time functions for deterministic timestamps. +- **Priority ordering** — Worker tasks complete before pull tasks, ensuring in-flight work finishes before accepting new external events. +- **Optimistic execution with retry** — Workers receive a snapshot of collected events. If new events arrive during execution, the worker re-runs with the updated snapshot. +- **State rehydration** — On resume, in-progress events move back to the queue and worker IDs reset, allowing clean restart from stored ticks. +- **Idle detection** — When all steps are waiting on external input, the loop publishes `WorkflowIdleEvent`. Runtime decorators can use this signal to release idle workflows from memory. diff --git a/architecture-docs/core-overview.md b/architecture-docs/core-overview.md new file mode 100644 index 00000000..a1436473 --- /dev/null +++ b/architecture-docs/core-overview.md @@ -0,0 +1,116 @@ +# Core Architecture: Workflow, Context, and Runtime + +## Overview + +```mermaid +graph LR + subgraph Client ["Client (public API)"] + WR["Workflow.run()"] + WH[WorkflowHandler] + EC["Context\n(ExternalContext)"] + end + + subgraph Runtime + EA[ExternalRunAdapter] + CL[Control Loop] + IA[InternalRunAdapter] + end + + subgraph Worker ["Worker (per step)"] + IC["Context\n(InternalContext)"] + Steps[Step Functions] + end + + WR -->|launches| CL + WH --- EA + EC --- EA + EA --- CL + CL --- IA + IA --- IC + IA --- Steps +``` + +The system has three zones. **Client** code calls `Workflow.run()` and interacts through `WorkflowHandler` and `Context`. The **Runtime** sits in the middle — the control loop drives execution, with `ExternalRunAdapter` and `InternalRunAdapter` as its boundaries. **Workers** are step functions that see `Context` with a different face (InternalContext). + +The adapters are the key abstraction boundary. Everything on the client side goes through `ExternalRunAdapter`; everything on the worker side goes through `InternalRunAdapter`. The runtime is swappable by replacing or decorating these adapters. + +## Workflow + +Container for step definitions. `run()` selects a runtime, validates steps, and delegates to the runtime for execution. + +[`workflow.py`](../packages/llama-index-workflows/src/workflows/workflow.py) + +## Context Faces + +Context presents different interfaces depending on execution phase. Internally it holds a `_face` field that transitions through three types: + +```mermaid +graph LR + Pre[PreContext] -->|"_workflow_run()"| External[ExternalContext] + External -.->|"per step worker"| Internal[InternalContext] +``` + +| Face | When | Used By | +|------|------|---------| +| PreContext | Before `run()` | Setup code — configuration, serialization, state store init | +| ExternalContext | After `run()` | Handler / caller — sending events, streaming | +| InternalContext | During step execution | Step functions — collecting events, publishing to stream | + +Each face wraps an adapter from the runtime. Public methods on Context check the current face and raise `ContextStateError` if called in the wrong phase. + +[`context/`](../packages/llama-index-workflows/src/workflows/context/) — Context implementation and all face types + +## Runtime and Adapters + +The `Runtime` ABC uses a dual-adapter pattern. Each workflow run produces two adapters sharing a `run_id`: + +```mermaid +graph TB + Runtime -->|"run_workflow()"| ExternalRunAdapter + Runtime -->|"get_internal_adapter()"| InternalRunAdapter + ExternalRunAdapter --- run_id + InternalRunAdapter --- run_id +``` + +The **InternalRunAdapter** is used by the control loop — it handles receiving ticks, publishing events, timing, and task coordination. The **ExternalRunAdapter** is used by the WorkflowHandler — it handles sending events in, streaming published events out, getting results, and cancellation. + +Multiple base runtimes exist. `BasicRuntime` is the default in-memory asyncio runtime in the core package. `DBOSRuntime` provides durable distributed execution backed by a database. More runtimes can be added by implementing the `Runtime` ABC. Runtimes are also composable via decorators — see [server-architecture.md](./server-architecture.md#runtime-decorator-chain) for that pattern. + +[`plugin.py`](../packages/llama-index-workflows/src/workflows/runtime/types/plugin.py) — Runtime ABC, InternalRunAdapter, ExternalRunAdapter + +## Control Loop + +The control loop is the core execution engine. It follows a reducer pattern where pure state transitions produce side effects as commands. The control loop is runtime-agnostic — it interacts with the outside world exclusively through `InternalRunAdapter`. + +See [control-loop.md](./control-loop.md) for the full architecture. + +## Event Flow + +```mermaid +graph LR + subgraph "Events In (external to internal)" + H[WorkflowHandler] -->|send_event| EA[ExternalRunAdapter] + EA -->|receive queue| IA[InternalRunAdapter] + IA -->|tick| CL[Control Loop] + end +``` + +```mermaid +graph LR + subgraph "Events Out (internal to external)" + CL2[Control Loop] -->|command| IA2[InternalRunAdapter] + IA2 -->|publish queue| EA2[ExternalRunAdapter] + EA2 -->|stream_published_events| H2[WorkflowHandler] + end +``` + +## WorkflowHandler + +Returned by `Workflow.run()`. The user-facing handle for a running workflow. + +- `await handler` — blocks until StopEvent, returns the result +- `handler.stream_events()` — async iterator of published events (single consumption) +- `handler.send_event()` — send events into the running workflow +- `handler.cancel_run()` — graceful cancellation + +[`handler.py`](../packages/llama-index-workflows/src/workflows/handler.py) diff --git a/architecture-docs/server-architecture.md b/architecture-docs/server-architecture.md new file mode 100644 index 00000000..f8d86742 --- /dev/null +++ b/architecture-docs/server-architecture.md @@ -0,0 +1,97 @@ +# Server Architecture + +## Overview + +The server wraps the core workflow engine (see [core-overview.md](./core-overview.md)) with HTTP access, persistence, and durability. Components are layered with clear boundaries: + +```mermaid +graph TD + Client -->|HTTP| API["_WorkflowAPI"] + API --> Service["_WorkflowService"] + Service --> Runtime["Runtime decorator chain"] + Runtime --> ControlLoop["Control Loop"] + Runtime -.->|reads/writes| Store["WorkflowStore"] + Service -.->|queries| Store + API -.->|subscribes events| Store +``` + +The **store** is the shared persistence layer — the runtime writes to it, the service queries it, and the API streams from it. + +## Components + +**`WorkflowServer`** — Entry point. Assembles the runtime chain, service, and API into a Starlette app. Registers workflows. +[`server.py`](../packages/llama-agents-server/src/llama_agents/server/server.py) + +**`_WorkflowAPI`** — Starlette routes. Translates HTTP requests into service calls and streams events from the store to clients via SSE. +[`_api.py`](../packages/llama-agents-server/src/llama_agents/server/_api.py) + +**`_WorkflowService`** — Application logic. Starts workflows, manages handler lifecycle, coordinates event sending and cancellation. Bridges the API to the runtime. +[`_service.py`](../packages/llama-agents-server/src/llama_agents/server/_service.py) + +**Runtime decorators** — A chain of decorators that add server concerns (event recording, tick persistence, idle release, etc.) on top of a base runtime. See the section below. +[`_runtime/`](../packages/llama-agents-server/src/llama_agents/server/_runtime/) — standard server decorators + +**`AbstractWorkflowStore`** — Persistence contract shared by all layers above. +[`abstract_workflow_store.py`](../packages/llama-agents-server/src/llama_agents/server/_store/abstract_workflow_store.py) + +## Runtime Decorator Chain + +Runtimes compose via decoration. Each decorator wraps a `Runtime` and its adapters (see [core-overview.md — Runtime and Adapters](./core-overview.md#runtime-and-adapters)), overriding only the methods it needs to add a specific concern — event recording to a store, tick persistence for replay, idle detection and memory release, etc. + +```mermaid +graph LR + Outer["Server decorators\n(one per concern)"] -->|wraps| Inner["Base Runtime\n(BasicRuntime, DBOSRuntime, etc.)"] +``` + +`WorkflowServer` assembles a default decorator chain on top of whatever base runtime is provided. The base runtime is swappable — pass `runtime=` to `WorkflowServer` to use a different one (BasicRuntime, DBOSRuntime, or a custom implementation). See `server.py` for how the default chain is assembled. + +**Writing a decorator:** Extend `BaseRuntimeDecorator` and optionally `BaseInternalRunAdapterDecorator` / `BaseExternalRunAdapterDecorator`. These forward all methods to the inner runtime/adapter — override only what you need. +[`runtime_decorators.py`](../packages/llama-agents-server/src/llama_agents/server/_runtime/runtime_decorators.py) + +## Persistence (WorkflowStore) + +The store is the system's source of truth for anything that survives a restart or an idle-release cycle. All layers depend on it: + +| What's stored | Purpose | +|---|---| +| Handler records | Lifecycle tracking (status, timestamps, result) | +| Event log | Resumable event streaming to clients | +| Ticks | Rebuild workflow state after idle release or restart | +| State stores | Persistent key-value state per run | + +Two implementations: +- **`MemoryWorkflowStore`** — In-process dicts. No persistence across restarts. + [`memory_workflow_store.py`](../packages/llama-agents-server/src/llama_agents/server/_store/memory_workflow_store.py) +- **`SqliteWorkflowStore`** — SQLite-backed. Survives restarts. + [`sqlite_workflow_store.py`](../packages/llama-agents-server/src/llama_agents/server/_store/sqlite/sqlite_workflow_store.py) + +## Resumable Event Streams + +Events flow from step functions to clients through the store, which acts as both a write-ahead log and a subscription source: + +```mermaid +graph LR + Step["Step function"] -->|"ctx.write_event_to_stream()"| IA["InternalRunAdapter"] + IA -->|decorator intercepts| Store["WorkflowStore"] + Store -->|"subscribe_events(cursor)"| API["_WorkflowAPI"] + API -->|SSE| Client +``` + +The store assigns each event a monotonic **sequence number**. Clients track their position via this cursor: + +1. Client connects, receives events as SSE with `id: {sequence}` +2. Client disconnects (network drop, restart, etc.) +3. Client reconnects with `Last-Event-ID: {last_seen_sequence}` +4. Store replays all events after that sequence, then continues live + +This means the API layer never needs to hold event history in memory — it just opens a `subscribe_events(after_sequence=cursor)` iterator from the store each time a client connects. + +A special `"now"` cursor skips all historical events and streams only new ones. + +## Server Lifecycle + +**Start:** `WorkflowServer.start()` queries the store for handlers with `status=running` that aren't idle, and resumes each by rebuilding context from stored ticks. + +**Stop:** `WorkflowServer.stop()` aborts all active control loops. Handler records remain in the store — they'll resume on next start. + +**Idle release:** When the [control loop detects](./control-loop.md#key-design-decisions) all steps are waiting on external input, it publishes `WorkflowIdleEvent`. Runtime decorators can use this signal to release the workflow from memory. When a new event arrives for that workflow, it reloads from ticks transparently.