In a recent consulting project, I built a multi-agent pipeline to automate analysis of construction change logs. The input was a corpus of change request documents. The output was structured data: revision motives, cost and time impact estimates, cross-references between drawings, and flags for anomalies.
This post describes the architecture, the tradeoffs I encountered, and what I would do differently.
A single LLM call over a large document is unreliable for structured extraction. The model loses coherence over long contexts, conflates tasks, and produces inconsistent output formats. Splitting the work across specialized agents gives each agent a focused task with a clear input and output schema.
In this project, the pipeline had four agents:
Each agent is a Python function that calls the OpenAI API with a structured prompt and parses the response into a typed Pydantic model.
The most important architectural decision was running agents asynchronously. A corpus of 300 change
entries processed sequentially would take minutes. With async orchestration via Python's
asyncio, independent entries run in parallel, bounded by a semaphore to stay within
API rate limits.
import asyncio
async def process_entries(entries: list[str], concurrency: int = 10) -> list[Result]:
sem = asyncio.Semaphore(concurrency)
async def run_one(entry):
async with sem:
classified = await classify(entry)
impact = await extract_impact(entry, classified)
return Result(entry=entry, classification=classified, impact=impact)
return await asyncio.gather(*[run_one(e) for e in entries])
The semaphore keeps the pipeline from hammering the API. The gather call collects results
in order, which matters when writing to the database.
Results are persisted in PostgreSQL via SQLAlchemy ORM. Each agent's output maps to a table. The cross-reference resolver reads from the classifier and impact tables to do its work, which is why the pipeline has a dependency stage: the first three agents run in parallel, the resolver runs after.
Using SQLAlchemy (rather than raw SQL or a document store) made it easy to write pytest fixtures that spin up an in-memory SQLite database for testing, keeping tests fast and hermetic.
The pipeline runs as a containerized job. The deployment flow is:
mainThe GitHub Actions workflow uses OIDC authentication to ACR, so there are no long-lived credentials in the repository. The image tag is the git commit SHA, which makes rollbacks straightforward.
- name: Build and push
uses: docker/build-push-action@v5
with:
context: .
push: true
tags: myregistry.azurecr.io/change-log-pipeline:${{ github.sha }}
I started with free-form LLM responses parsed with regex. This broke constantly. Switching to OpenAI's structured output mode (with a JSON schema) reduced parsing failures to near zero and made the Pydantic validation layer the single source of truth for the schema.
Four agents turned out to be the right number, though I didn't arrive at that on purpose. An earlier version collapsed the classifier and impact extractor into one agent, and the prompt got long enough that the model started skipping the cost-delta field on maybe one entry in twenty. Splitting it back out fixed that immediately. Going the other direction, toward more agents than four, mostly just added latency without buying anything, since the dependency graph between agents gets harder to reason about faster than the individual prompts get simpler. One agent per distinct output schema is the rule I'd start from next time, but I'd still expect to adjust it once I saw real failures.
API calls fail. The pipeline uses exponential backoff with jitter on each agent call. Entries that fail after retries are logged with their error and skipped, not dropped silently. A separate review queue in the database holds failed entries for manual inspection.
Four API calls per entry on a 300-entry corpus is 1,200 calls per run. With gpt-4o-mini this is a few dollars. With gpt-4o it would be closer to $50. Model selection per agent matters: the splitter and classifier use a cheaper model; the cross-reference resolver, which requires more reasoning, uses the more capable one.
The biggest gap was not having an evaluation harness from day one. We relied on manual spot-checks early on, which meant a prompt tweak that quietly dropped accuracy on one agent could sit unnoticed for days. A small labeled test set with an automated accuracy check per agent would have caught that in the next CI run instead.
The other two things I'd change are smaller but real. asyncio.gather stopped being
enough once I mentally scaled the corpus past a few thousand entries, since one slow or hung call
can hold up the whole batch, a proper task queue like Celery or ARQ with worker scaling is the
better fit at that size. And prompts need to be versioned alongside the code, not edited in place,
because a change that improves the classifier can silently regress the cross-reference resolver
three commits later with nothing in the diff to flag it.
What surprised me most, looking back, is how little of the actual work was the LLM part. Getting
the prompts to produce something usable took maybe a week. Getting the orchestration, retries, and
storage layer to a point where I trusted the pipeline to run unattended overnight took the rest of
the project. That ratio is probably the real lesson here, more than any specific choice of
asyncio over a task queue or Pydantic over raw JSON parsing.