Multi-Agent Python Script Generator

A reference implementation of a production-grade multi-agent pipeline built on LangGraph. Takes a plain-English data transformation request (or a screenshot) and delivers a tested, documented, sandboxed Python script.

The primary goal is engineering: demonstrating LangGraph's Send API for parallel fan-out, Annotated reducers for concurrent state, Docker SDK sandboxing, model tiering, and multimodal input — all wired together in one coherent system.

---

Pipeline

flowchart TD

A([User Input\ntext or image]) --> B{route_after_input}

B -->|image| C[Vision Agent\nSonnet]

B -->|text| D

C --> D[Orchestrator Agent\nSonnet]

D -->|Send API — parallel fan-out| E

D -->|Send API — parallel fan-out| F

D -->|Send API — parallel fan-out| G

D -->|Send API — parallel fan-out| H

subgraph Parallel ["

![⚡](https://fonts.gstatic.com/s/e/notoemoji/17.0/26a1/72.png)

Parallel Execution"]

E[Schema Agent\nHaiku]

F[Code Agent\nSonnet]

G[Test Agent\nHaiku]

H[Docs Agent\nHaiku]

end

E --> I[Merge\nAnnotated reducer]

F --> I

G --> I

H --> I

I --> J[Assemble\nfile bundle]

J --> K[Sandbox Executor\nDocker SDK]

K -->|success| L[Review\ninterrupt — HITL]

K -->|failure| M[Code Fix\n1 auto-retry]

M --> J

L -->|approved| N([Delivered])

L -->|rejected| O([Aborted])

Loading

Interactive architecture diagram (pipeline, AgentState, model tiering, tech stack) available at [http://localhost:8000/static/architecture.html](http://localhost:8000/static/architecture.html) when running locally.

---

Design decisions worth noting

• Send API for parallel fan-out.* The orchestrator dispatches four specialist agents simultaneously via Send. Total latency is max(agents) not sum(agents). This is the right pattern for any pipeline with independent specialist work.
• Annotated reducer for concurrent writes.* All four agents write to the same AgentState. Without a reducer, last-writer-wins drops data. Annotated[list[dict], operator.add] accumulates outputs safely; the merge node unpacks by discriminator field.

agent_outputs: Annotated[list[dict], operator.add]

• Model tiering.* Sonnet handles reasoning-heavy nodes (orchestrator, code, vision). Haiku handles mechanical work (schema inference, test generation, docs). Three of five agents run on Haiku — significant cost reduction with no quality loss on deterministic tasks.
• Docker SDK over subprocess.* Each execution gets an ephemeral container via client.containers.run() with mem_limit, cpu_quota, network_mode="none", and pids_limit. No persistent state between runs, no network egress, hard timeout. This is the same pattern as production code execution services.
• No ge/le on LLM output schemas.* Anthropic's structured output format rejects JSON Schema minimum/maximum keywords on number fields. Numeric bounds are encoded in field descriptions instead.

---

Stack

Layer	Technology
Orchestration	LangGraph — StateGraph, Send API, interrupt(), MemorySaver
LLM	Claude Sonnet 4.5 + Haiku 4.5 via OpenRouter
Serving	FastAPI + SSE (astream_events v2)
Sandboxing	Docker Python SDK — containers.run() with resource limits
Data generation	Faker + Polars — schema-aware synthetic CSV
Schemas	Pydantic v2 throughout — all agent I/O typed
Config	pydantic-settings with env-prefix namespacing
Observability	Langfuse (self-hosted) — parallel spans, session tracking
Prompts	Jinja2 templates
Tooling	uv, ruff, pytest — 50 tests, 0 lint errors

---

Local setup

• Prerequisites:* Python 3.12+, uv, Docker Desktop, OpenRouter API key.

git clone https://github.com/<your-username>/multi-agent-script-generator.git

cd multi-agent-script-generator

uv sync

cp .env.example .env          *# add ANTHROPIC_API_KEY (OpenRouter key)*

docker build -t script-sandbox:latest ./sandbox/

uv run serve                  *# [http://localhost:8000](http://localhost:8000/)*

Optional — Langfuse tracing:

docker compose up -d          *# UI at [http://localhost:3000](http://localhost:3000/)*

Set LANGFUSE_ENABLED=false in .env to skip.

---

API

• Generate* (SSE stream):

curl -N -X POST http://localhost:8000/generate \

-H 'Content-Type: application/json' \

-d '{"description": "Read a file with columns product and price. Add a discounted_price column at 10% off."}'

• Resume after HITL pause* (thread_id from done event):

curl -X POST http://localhost:8000/resume \

-H 'Content-Type: application/json' \

-d '{"thread_id": "<id>", "approved": true}'

• Image input:* add "image_b64": "<base64-encoded PNG>" to the generate request body.

SSE event types: node_start · node_end · stream · interrupt · error · done

---

Contributing

uv run ruff check . && uv run ruff format . && uv run pytest

Open an issue before submitting a PR for non-trivial changes.

---

License

MIT © 2026 Shreya Baid