Multi-Agent Pipelines for Document Analysis: Architecture and Tradeoffs

Multi-Agent Pipelines for Document Analysis: Architecture and Tradeoffs

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.

Why Multiple Agents?

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:

  • Splitter: segments the raw document into individual change entries
  • Classifier: labels each entry by revision motive (design change, site condition, owner request, etc.)
  • Impact extractor: pulls cost delta, time delta, and drawing references from each entry
  • Cross-reference resolver: matches drawing references across entries to detect cascading changes

Each agent is a Python function that calls the OpenAI API with a structured prompt and parses the response into a typed Pydantic model.

Async Orchestration

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.

Storage: PostgreSQL with SQLAlchemy

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.

Deployment: Azure Container Registry and GitHub Actions

The pipeline runs as a containerized job. The deployment flow is:

  1. A GitHub Actions workflow builds the Docker image on every push to main
  2. The image is pushed to Azure Container Registry (ACR)
  3. A downstream job pulls the image and runs the pipeline on a scheduled trigger

The 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 }}

Tradeoffs and Lessons

Structured output vs. free-form parsing

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.

Agent granularity

Four agents felt right for this task. Fewer agents made individual prompts too long and outputs too heterogeneous. More agents added latency and made the dependency graph hard to reason about. The right granularity is roughly one agent per distinct output schema.

Error handling

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.

Cost

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.

What I Would Do Differently

  • Add an evaluation harness earlier. We built manual spot-checks at first. A small labeled test set with automated accuracy checks would have caught regressions faster.
  • Use a task queue for large corpora. For corpora over a few thousand entries, asyncio.gather is not enough. A proper task queue (Celery, ARQ) with worker scaling is the right tool.
  • Version the prompts. Prompt changes that improve one agent can silently break another. Tracking prompts as versioned artifacts alongside the code prevents this.

Conclusion

Multi-agent pipelines are not inherently complex. The complexity comes from managing dependencies between agents, handling failures gracefully, and keeping the system testable. In this project, async orchestration with asyncio, typed outputs via Pydantic, and PostgreSQL for persistence covered 90% of the real engineering work. The LLM calls themselves were the easy part.

If you are building something similar and want to discuss the architecture, feel free to reach out.