"Building AuditFlow: A Multi-Agent Document Intelligence Platform"

The Problem

Traditional auditing involves manually reviewing hundreds of documents — PDFs, spreadsheets, scanned images — looking for inconsistencies, anomalies, and compliance issues. This process is slow, error-prone, and does not scale.

When I first encountered this problem while working with financial audit teams, I realized that the core challenge was not just about processing documents, but about understanding them at scale. Each audit engagement involves:

• 500-2000+ documents across multiple formats
• Tight deadlines (often 2-4 weeks)
• High accuracy requirements (99.5%+)
• Complex regulatory knowledge requirements

System Architecture

High-Level Design

AuditFlow uses a multi-agent architecture powered by LangGraph. Each agent specializes in a specific stage of the document intelligence pipeline:

User Upload → Document Parser → Chunking Agent → Embedding Pipeline → Vector Store
                                                                              ↓
User Query  →  Query Router  →  Knowledge Agent  →  RAG Pipeline  →  Response
                                    ↓
                            Workflow Agent → Multi-step Audit Procedures

Agent Breakdown

1. Document Parser Agent

The first challenge is handling diverse document formats. The parser supports:

SUPPORTED_FORMATS = {
    "pdf": ["PyMuPDF", "pdfplumber", "OCR (Tesseract)"],
    "docx": ["python-docx"],
    "xlsx": ["openpyxl"],
    "jpg/png": ["Tesseract OCR", "LayoutLM"],
    "html": ["BeautifulSoup"],
}

For complex PDFs with mixed layouts (tables, headers, footers), a hybrid approach is used:

class HybridParser:
    def parse(self, path: str) -> Document:
        # First pass: extract structured text
        text = PyMuPDFParser().extract(path)

        # Second pass: detect layout elements
        layout = LayoutLMParser().analyze(path)

        # Merge results
        return DocumentMerger().merge(text, layout)

2. Chunking Agent

Chunking quality directly determines retrieval accuracy. I implemented a multi-strategy system:

| Strategy | Use Case | Chunk Size | Overlap | |----------|----------|------------|---------| | Fixed-size | Simple text | 512 tokens | 64 tokens | | Semantic | Financial reports | Dynamic | None | | Recursive | Mixed content | 1024 tokens | 128 tokens | | Agentic | Complex docs | LLM-determined | N/A |

3. Embedding Pipeline

The embedding pipeline processes chunks in parallel batches:

async def embed_batch(chunks: list[Chunk]) -> list[Vector]:
    embeddings = await asyncio.gather(*[
        embed_client.embed(chunk.text)
        for chunk in chunks
    ])
    return [Vector(chunk.id, emb) for chunk, emb in zip(chunks, embeddings)]

Orchestration with LangGraph

All agents are orchestrated using LangGraph's state machine:

from langgraph.graph import StateGraph, END

workflow = StateGraph(AuditState)

# Register nodes
workflow.add_node("parse", DocumentParser())
workflow.add_node("chunk", ChunkingAgent())
workflow.add_node("embed", EmbeddingAgent())
workflow.add_node("index", VectorIndexer())
workflow.add_node("analyze", KnowledgeAgent())

# Define edges with conditional routing
workflow.add_conditional_edges(
    "parse",
    lambda state: "reprocess" if state.needs_ocr else "chunk"
)

workflow.add_edge("chunk", "embed")
workflow.add_edge("embed", "index")
workflow.add_edge("index", "analyze")
workflow.add_edge("analyze", END)

# Compile with checkpointing for fault tolerance
app = workflow.compile(checkpointer=PostgresSaver())

RAG Implementation

Hybrid Search

Pure semantic search is not enough for audit documents where exact terms matter. The system uses hybrid search combining:

class HybridRetriever:
    def __init__(self):
        self.dense_retriever = DenseRetriever(model="text-embedding-3-small")
        self.sparse_retriever = SparseRetriever(analyzer="english")

    async def retrieve(self, query: str, top_k: int = 10):
        # Dense retrieval (semantic)
        dense_results = await self.dense_retriever.search(query, top_k * 2)

        # Sparse retrieval (keyword)
        sparse_results = await self.sparse_retriever.search(query, top_k * 2)

        # Fusion with RRF (Reciprocal Rank Fusion)
        return rrf_fusion(dense_results, sparse_results, top_k)

Query Transformation

A key lesson was that raw user queries rarely work well for retrieval. The system transforms queries:

TRANSFORM_PROMPT = """
Given the original question, generate 3 search queries optimized for vector search.
Rewrite complex questions into simpler sub-questions.
Original: {question}
Queries:
"""

Engineering Challenges

Challenge 1: PDF Parsing Accuracy

Problem: Complex PDF layouts (multi-column, nested tables) caused text extraction errors.

Solution: Implemented a hybrid parser combining PyMuPDF for text extraction with LayoutLM for layout understanding. The system detects layout anomalies and falls back to OCR when confidence is low.

Result: Extraction accuracy improved from 85% to 99.2%.

Challenge 2: Processing Latency

Problem: 200-page documents took 45+ seconds to process end-to-end.

Solution: Designed a streaming pipeline with Redis-based caching and parallel chunk processing. Each document stage is independently cacheable.

Result: Reduced average processing time to 8 seconds for standard documents.

Challenge 3: Hallucination in Analysis

Problem: The knowledge agent occasionally fabricated evidence citations.

Solution: Implemented a two-stage verification: the agent first retrieves evidence, then a separate verifier agent confirms each claim maps to actual retrieved content. Claims without supporting evidence are flagged.

Key Results

| Metric | Before | After | |--------|--------|-------| | Manual review time | 40 hours | 4 hours | | Document accuracy | 92% | 99.5% | | Supported formats | 3 | 10+ | | Processing time/page | 15s | 2.4s | | Citation accuracy | N/A | 98.7% |

Lessons Learned

1. Hybrid approaches beat pure solutions — Every stage benefits from combining multiple strategies (parsing, retrieval, verification)
2. Agent orchestration requires fault tolerance — LangGraph's checkpointing was essential for production reliability
3. Evaluation drives improvement — Building a comprehensive test suite with 500+ test documents was the single most impactful investment
4. Start simple, iterate fast — The first prototype used a single LLM call. Agentic decomposition came later as complexity grew

Future Plans

• Multi-modal support (images, charts analysis)
• Real-time collaboration features
• Custom agent development SDK
• Integration with major audit platforms