Multi-agent pipeline code — ai-agents-production-systems

Repository status

Code complete · 18/18 tests passing · Published on GitHub. Everything below is the real repository, verbatim — not a summary, not pseudocode. No network, no API keys; stdlib + pytest only. Clone it, run the failure drills, steal the patterns.

File tree

ai-agents-production-systems/ ├── README.md ├── LICENSE ├── requirements.txt # pytest only — the pipeline is stdlib ├── src/workflow/ │ ├── __init__.py │ ├── types.py # typed handoffs: every agent gets/returns a dataclass │ ├── intake.py # normalize + validate the raw request │ ├── classifier.py # booking / question / complaint │ ├── drafter.py # composes the reply — no invented facts │ ├── approval.py # the human gate: approve or reject │ ├── sender.py # idempotent delivery, injected failure drills │ ├── pipeline.py # wires the agents; reject halts, nothing sends │ └── runlog.py # structured run log for every route └── tests/ ├── conftest.py ├── test_agents.py ├── test_edge_cases.py └── test_pipeline.py # incl. reject-halts, idempotency, fail_once/fail_always

The parts that matter

1 · The approval gate: nothing sends without a human

approval.py. The machine does the reading, routing, and drafting; a human decides what goes out under the business's name. The hard rule is enforced in code, not documentation: the pipeline reaches the sender only through an approved decision.

def approval_gate(draft: Draft, decide: DecideFn, log: RunLog) -> Decision: log.record("approval_gate", "awaiting_human", f"draft for {draft.recipient} ({draft.category})") decision = decide(draft) log.record( "approval_gate", "approved" if decision.approved else "rejected", decision.reason, ) return decision

Reject at the gate and the run halts — the outbox stays empty and the log records the rejection. test_reject_halts_and_sends_nothing asserts it.

2 · Idempotent sends: retries never double-deliver

sender.py. Every send carries an idempotency key. A retried or re-run pipeline with the same key returns the original receipt and delivers nothing twice.

def send(self, draft: Draft, idempotency_key: str, log: RunLog) -> SendReceipt: # Rule 1: same key -> same receipt, no second delivery. if idempotency_key in self._registry: log.record("sender", "duplicate_suppressed", idempotency_key) return self._registry[idempotency_key] ... receipt = SendReceipt( idempotency_key=idempotency_key, message_id=f"msg-{idempotency_key}", recipient=draft.recipient, ) self._registry[idempotency_key] = receipt self.outbox.append(receipt)

Failure drills are injected, not hoped for: fail_once (flaky provider — retry then succeed), fail_always (outage — escalate, send nothing).

3 · Typed handoffs: no free-form dicts cross agent boundaries

types.py. Each agent receives a dataclass and returns a dataclass. If a field matters, it is named here.

@dataclass(frozen=True) class Draft: """Outbound message composed but NOT sent. Sending requires a human.""" request_id: str category: Category recipient: str subject: str body: str

Get the full repo

Run it yourself: pip install -r requirements.txt && pytest — 18 tests, all passing. Try the failure drills first.

View on GitHub Full repo · README · tests · failure drills

Go deeper: the interactive demo — run the workflow yourself, including the 2am failure scenario — and the article on what makes a workflow hold up at 2am.

Want this running your business?

Same shape, your systems, your voice — scoped, priced, and monitored before you commit. Fixed scope, fixed price, 14-day defect window.

Start the free assessment Request a quote

The multi-agent pipeline, in code.