Multi-Agent Document Intelligence API

Build a multi-agent RAG pipeline with LangGraph, ChromaDB, and FastAPI.

Introduction

30 Second Summary

When you read a long report, you naturally do several things at once. You summarize the key points, identify the people and organizations mentioned, and answer questions about what you read. Now imagine a system where specialized AI agents handle each of those tasks independently, then a separate agent checks their work.

In this project, you will build a multi-agent document intelligence pipeline that processes text documents through specialized AI agents, evaluates their outputs against local model baselines, and serves the results through an API with confidence scores.

What You'll Build

A locally-running FastAPI service that orchestrates multiple AI agents to analyze documents, then scores each output's reliability using comparative evaluation.

By the end of this project, you'll have:

  • A LangGraph state graph implementing a supervisor-worker architecture that routes documents to specialized Summarizer, Entity Extractor, and Q&A agents.
  • A RAG pipeline using ChromaDB for grounded document question-answering, where answers come only from retrieved context.
  • A comparative evaluation layer that scores Gemini outputs against local HuggingFace model baselines and produces per-output confidence metrics.
  • Secret Mission: Add an adaptive quality gate that automatically flags low-confidence outputs and triggers re-processing with an adjusted prompt strategy.

Are there any prerequisites?

This project assumes proficiency in Python, a basic understanding of APIs and HTTP requests, and a Google AI Studio account with an API key. You should be comfortable installing pip packages and running Python scripts from the terminal.

Set Up Your Environment

Building a multi-agent system requires multiple libraries working together. You need an orchestration framework, an LLM connection, a vector database, local ML models, and a web server.

In this step, you will install everything the pipeline depends on and verify that each integration works before writing any application logic.

In this step, get ready to:
  • Create a Python virtual environment and install all project dependencies.
  • Configure and verify your Gemini API connection.
  • Download and test HuggingFace models for local evaluation baselines.
Create your project and install dependencies

This project requires Python 3.10 or higher because both LangGraph and the google-genai SDK enforce this minimum.

  • Check your Python version by running this command:
python3 --version

✔️ I see version 3.10 or higher

Your Python version is compatible. Continue to the next instruction below.

ⓧ I see an older version

macOS sometimes ships with an older Python that shadows newer installs. You need 3.10+ for this project.

  • Install the latest Python from python.org/downloads.
  • After installing, run python3 --version again to confirm it reports 3.10 or higher.
  • If the version still shows the old one, try invoking the specific binary directly (e.g. python3.12 --version) and use that binary for the venv command below.

ⓧ Command not found

Python is not installed on your machine yet.

  • Download and install Python from python.org/downloads.
  • Follow the installer prompts, making sure to check any option that adds Python to your PATH.
  • Close and reopen your terminal, then run python3 --version to confirm the install succeeded.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir docintel
cd docintel
  • Create a Python virtual environment by running:
python3 -m venv venv
  • Activate the virtual environment:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

You should see (venv) appear at the start of your terminal prompt. This confirms the virtual environment is active.

  • Create the requirements.txt file that lists all project dependencies by running:
cat << 'EOF' > requirements.txt
langgraph
google-genai
chromadb
transformers
torch
fastapi
uvicorn
pydantic
EOF

What are these packages?

  • langgraph orchestrates multi-agent workflows as state graphs.
  • google-genai connects to the Gemini API for LLM calls.
  • chromadb provides a local vector database for RAG retrieval.
  • transformers and torch run HuggingFace models locally for evaluation baselines.
  • fastapi and uvicorn serve your pipeline as an HTTP API.
  • pydantic defines structured response schemas.
  • Install all dependencies by running:
pip install -r requirements.txt

This download is large

The torch package alone is over 2GB. The full install may take 5-10 minutes depending on your internet speed. Let it finish completely before continuing.

Once the install finishes without errors, you should see a Successfully installed message listing all packages.

Seeing install errors?

  • Make sure your virtual environment is active (you see (venv) in your prompt). Run the activate command again if needed.
  • If torch fails to install, check that you have at least 4GB of free disk space.
  • If you see permission errors, do not use sudo. Deactivate and recreate the venv instead.

help me troubleshoot my pip install errors for the docintel project

✔️ Awesome, I've got everything!

Great. Make sure you have saved requirements.txt and that your install completed without errors.

ⓧ I'd like to double check the full code

Here is what requirements.txt should contain:

langgraph
google-genai
chromadb
transformers
torch
fastapi
uvicorn
pydantic
Configure your Gemini API key

The pipeline uses Google's Gemini API for all LLM calls. You need an API key from Google AI Studio to authenticate.

  • Go to Google AI Studio and sign in with your Google account.
  • Click Create API key.
  • Copy the generated key. You will only see it once.

Is the Gemini API free?

The Gemini Developer API free tier provides free input and output tokens for gemini-2.0-flash. Content sent on the free tier is used to improve Google products.

  • Create a .env file in your docintel folder to store your key by running:
cat << 'EOF' > .env
GOOGLE_API_KEY=your_api_key_here
EOF
  • Open .env in your text editor and replace your_api_key_here with the actual API key you just copied.
  • Save the file.
  • Export the key into your current terminal session so your scripts can read it:

macOS

export GOOGLE_API_KEY=$(cat .env | grep GOOGLE_API_KEY | cut -d '=' -f2)

Windows

Open your .env file, copy your key value, and run:

$env:GOOGLE_API_KEY = "your_api_key_here"

Replace your_api_key_here with the actual key you pasted into .env.

Why a .env file and an export?

The .env file stores your key so you don't lose it between sessions. The export command makes it available to Python's os.environ in your current terminal. You will re-export each time you open a new terminal window.

Verify your setup end-to-end

Before writing any pipeline code, you will confirm that all three key integrations work: the Gemini API, the BERT NER model, and the BART summarization model.

  • Create the test script by running:
cat << 'EOF' > test_setup.py
import os
from google import genai
from transformers import pipeline

# Test Gemini
client = genai.Client(api_key=os.environ['GOOGLE_API_KEY'])
response = client.models.generate_content(model='gemini-2.0-flash', contents='Say hello')
print('Gemini:', response.text)

# Test BERT NER
ner_pipeline = pipeline('ner', model='dslim/bert-base-NER', aggregation_strategy='simple')
print('BERT NER:', ner_pipeline('My name is John and I work at Google in New York.'))

# Test BART summarization
summarizer = pipeline('summarization', model='facebook/bart-large-cnn')
print('BART:', summarizer('The quick brown fox jumps over the lazy dog. ' * 10, max_length=50, min_length=10))
EOF

What does this script do?

  • The Gemini section creates an API client and sends a simple prompt. If your key is valid, it prints a greeting.
  • The BERT NER section downloads dslim/bert-base-NER (~440MB) and runs entity extraction on a test sentence. It should find person, organization, and location entities.
  • The BART section downloads facebook/bart-large-cnn (~1.6GB) and summarizes repeated text. This confirms the model loads and generates output.
  • Run the test script:
python test_setup.py

First run downloads models

The first time you run this script, HuggingFace downloads both models to ~/.cache/huggingface (approximately 2GB total). This may take several minutes. Subsequent runs load from cache instantly.

You should see three lines of output. The Gemini line prints a greeting. The BERT NER line prints a list of detected entities with labels like PER, ORG, and LOC. The BART line prints a short summary.

Seeing an API key error?

  • Check that GOOGLE_API_KEY is set in your terminal by running echo $GOOGLE_API_KEY (macOS) or echo $env:GOOGLE_API_KEY (Windows PowerShell). If it prints nothing, re-run the export command from the previous substep.
  • If you see a 400 or 403 error, your key may be invalid. Return to Google AI Studio and generate a new one.
  • If a HuggingFace model download times out, check your internet connection and try running the script again.

help me debug my test_setup.py errors in the docintel project

✔️ Awesome, I've got everything!

Great. Make sure test_setup.py and .env are saved in your docintel folder.

ⓧ I'd like to double check the full code

Here are all the files you should have in your docintel folder:

GOOGLE_API_KEY=your_api_key_here
langgraph
google-genai
chromadb
transformers
torch
fastapi
uvicorn
pydantic
import os
from google import genai
from transformers import pipeline

# Test Gemini
client = genai.Client(api_key=os.environ['GOOGLE_API_KEY'])
response = client.models.generate_content(model='gemini-2.0-flash', contents='Say hello')
print('Gemini:', response.text)

# Test BERT NER
ner_pipeline = pipeline('ner', model='dslim/bert-base-NER', aggregation_strategy='simple')
print('BERT NER:', ner_pipeline('My name is John and I work at Google in New York.'))

# Test BART summarization
summarizer = pipeline('summarization', model='facebook/bart-large-cnn')
print('BART:', summarizer('The quick brown fox jumps over the lazy dog. ' * 10, max_length=50, min_length=10))

Your environment is fully configured. Next up, you will build the LangGraph supervisor-worker architecture that orchestrates your AI agents.

Build the LangGraph Supervisor-Worker Architecture

Your environment is verified and every dependency is ready. Now it's time to build the core of your document intelligence system: the multi-agent pipeline itself.

In a multi-agent system, you need something to decide which agent runs and when. LangGraph solves this with a state graph where nodes are agents and edges define how control flows between them. You'll implement a supervisor-worker pattern where a central Supervisor delegates tasks to specialized Worker agents.

In this step, get ready to:
  • Define the shared state schema and Supervisor routing logic.
  • Implement three Worker agent nodes that call Gemini for analysis.
  • Wire the full graph together, compile it, and test with a sample document.
Define the shared state schema and create the Supervisor node

Every agent in the pipeline reads from and writes to a shared state object. This state carries the document, results from each agent, and metadata about which tasks have been completed.

  • Create a new file called pipeline.py in your project folder.
  • Add the imports, client setup, and state schema by pasting this code:
import os
import json
from typing import TypedDict, Annotated, List, Dict, Any, Optional
from langgraph.graph import StateGraph, START, END
from google import genai
import chromadb

# Initialize Gemini and ChromaDB clients
client = genai.Client(api_key=os.environ['GOOGLE_API_KEY'])
chroma_client = chromadb.PersistentClient(path='./chroma')
collection = chroma_client.get_or_create_collection('documents')

# Shared state schema - every agent reads from and writes to this
class DocumentState(TypedDict):
    document_text: str
    question: str
    summary: str
    entities: List[Dict]
    qa_answer: str
    evaluation_scores: Dict[str, float]
    agent_trace: List[Dict]
    current_task: str
    tasks_completed: List[str]

What does this code do?

The imports pull in StateGraph, START, and END from LangGraph. These are the building blocks for defining your graph's structure and control flow.

The DocumentState TypedDict is the shared memory that every agent reads and updates. Key fields include current_task (what the Supervisor wants done next) and tasks_completed (which tasks are already finished). The Supervisor uses these to decide routing.

  • Now add the Supervisor logic below your DocumentState class by pasting this code:
def supervisor_node(state: DocumentState) -> DocumentState:
    # Check which tasks are done and assign the next one
    completed = state.get('tasks_completed', [])
    if 'summarize' not in completed:
        return {**state, 'current_task': 'summarize'}
    elif 'extract_entities' not in completed:
        return {**state, 'current_task': 'extract_entities'}
    elif 'qa' not in completed and state.get('question'):
        return {**state, 'current_task': 'qa'}
    else:
        return {**state, 'current_task': 'evaluate'}

def route_from_supervisor(state: DocumentState) -> str:
    # Map current_task to the correct worker node name
    task = state.get('current_task', '')
    routes = {
        'summarize': 'SummarizerAgent',
        'extract_entities': 'EntityExtractorAgent',
        'qa': 'QAAgent',
        'evaluate': 'EvaluatorAgent'
    }
    return routes.get(task, 'EvaluatorAgent')

What does this code do?

The supervisor_node function inspects the state to decide the next task. It checks tasks_completed in sequence: summarize first, then entity extraction, then Q&A (only if a question was provided), and finally evaluation.

The route_from_supervisor function translates the current_task value into the actual graph node name. LangGraph calls this function to determine which node runs next after the Supervisor.

  • Save pipeline.py.
  • Confirm there are no syntax errors by running:
python -c "from pipeline import DocumentState, supervisor_node, route_from_supervisor; print('Supervisor ready')"
  • You should see Supervisor ready printed in your terminal.

Seeing an ImportError or ModuleNotFoundError?

  • Make sure your virtual environment is activated. The prompt should show the env name in parentheses.
  • Confirm you are running the command from the same directory where pipeline.py lives.
  • Check for typos in the from langgraph.graph import StateGraph, START, END line.

Why can't Python find my pipeline module?

Implement the three Worker agent nodes

Each Worker agent handles one specific analysis task. They all follow the same pattern: call Gemini, parse the result, log a trace entry, and mark their task as complete.

  • Add the summarizer_agent function below route_from_supervisor in pipeline.py:
def summarizer_agent(state: DocumentState) -> DocumentState:
    import time
    start = time.time()
    # Call Gemini to summarize the document
    response = client.models.generate_content(
        model='gemini-2.0-flash',
        contents=f'Summarize the following document in 2-3 sentences:\n\n{state["document_text"]}'
    )
    summary = response.text
    # Log this agent's execution in the trace
    trace = state.get('agent_trace', [])
    trace.append({'agent': 'SummarizerAgent', 'model': 'gemini-2.0-flash', 'duration': time.time() - start})
    completed = state.get('tasks_completed', []) + ['summarize']
    return {**state, 'summary': summary, 'agent_trace': trace, 'tasks_completed': completed, 'current_task': 'supervisor'}

What does this code do?

The summarizer_agent sends the full document text to Gemini with a summarization prompt. It timestamps the request to measure latency, appends a trace dictionary recording agent name and duration, and marks 'summarize' as completed so the Supervisor knows not to re-run this task.

  • Add the entity_extractor_agent function below summarizer_agent:
def entity_extractor_agent(state: DocumentState) -> DocumentState:
    import time
    start = time.time()
    # Ask Gemini to extract entities as structured JSON
    response = client.models.generate_content(
        model='gemini-2.0-flash',
        contents=f'Extract all named entities from the text below. Return a JSON array of objects with keys: text, type (PERSON/ORG/LOCATION). Only return the JSON array, no other text.\n\n{state["document_text"]}'
    )
    # Parse the JSON response, handling potential markdown code fences
    try:
        raw = response.text.strip().strip('```json').strip('```').strip()
        entities = json.loads(raw)
    except Exception:
        entities = []
    trace = state.get('agent_trace', [])
    trace.append({'agent': 'EntityExtractorAgent', 'model': 'gemini-2.0-flash', 'duration': time.time() - start})
    completed = state.get('tasks_completed', []) + ['extract_entities']
    return {**state, 'entities': entities, 'agent_trace': trace, 'tasks_completed': completed, 'current_task': 'supervisor'}

What does this code do?

This agent prompts Gemini to return a JSON array of named entities. The parsing logic strips markdown code fences that Gemini sometimes wraps around JSON output. If parsing fails (malformed JSON), it falls back to an empty list rather than crashing the pipeline.

  • Add the qa_agent function below entity_extractor_agent:
def qa_agent(state: DocumentState) -> DocumentState:
    import time
    start = time.time()
    # Retrieve relevant chunks from ChromaDB
    results = collection.query(query_texts=[state['question']], n_results=3)
    chunks = results['documents'][0] if results['documents'] else []
    context = '\n\n'.join(chunks)
    # Generate answer grounded in retrieved context
    response = client.models.generate_content(
        model='gemini-2.0-flash',
        contents=f'Answer the question using ONLY the context provided. If the answer is not in the context, say so.\n\nContext:\n{context}\n\nQuestion: {state["question"]}'
    )
    trace = state.get('agent_trace', [])
    trace.append({'agent': 'QAAgent', 'model': 'gemini-2.0-flash', 'duration': time.time() - start, 'source_chunks': chunks})
    completed = state.get('tasks_completed', []) + ['qa']
    return {**state, 'qa_answer': response.text, 'agent_trace': trace, 'tasks_completed': completed, 'current_task': 'supervisor'}

What does this code do?

The qa_agent performs RAG (Retrieval-Augmented Generation). It queries ChromaDB for the 3 most relevant document chunks, formats them as context, and passes both context and question to Gemini. The prompt instructs Gemini to only answer from the provided context.

The trace entry also stores source_chunks so you can later verify which parts of the document the answer was based on.

  • Save pipeline.py and confirm the workers import correctly by running:
python -c "from pipeline import summarizer_agent, entity_extractor_agent, qa_agent; print('Workers ready')"
  • You should see Workers ready printed in your terminal.

Seeing a SyntaxError?

  • Check that the f-string in entity_extractor_agent uses \n\n (escaped newlines) inside the string, not raw line breaks.
  • Make sure the try/except block in entity_extractor_agent is properly indented within the function body.
  • Verify you have a closing parenthesis on every client.models.generate_content( call.

Help me fix a SyntaxError in my worker agents.

Wire the graph together and test it

With all the nodes defined, you now connect them into a directed graph. The Supervisor sits at the center, routing to workers via conditional edges. Workers always return to the Supervisor when done.

  • Add the Evaluator placeholder and graph wiring below qa_agent in pipeline.py:
def evaluator_node(state: DocumentState) -> DocumentState:
    # Placeholder - full evaluation logic comes in Step 4
    return {**state, 'evaluation_scores': {}, 'current_task': 'done'}

# Build the LangGraph state graph
builder = StateGraph(DocumentState)
builder.add_node('Supervisor', supervisor_node)
builder.add_node('SummarizerAgent', summarizer_agent)
builder.add_node('EntityExtractorAgent', entity_extractor_agent)
builder.add_node('QAAgent', qa_agent)
builder.add_node('EvaluatorAgent', evaluator_node)

# Wire edges: START -> Supervisor, Supervisor routes conditionally
builder.add_edge(START, 'Supervisor')
builder.add_conditional_edges('Supervisor', route_from_supervisor, {
    'SummarizerAgent': 'SummarizerAgent',
    'EntityExtractorAgent': 'EntityExtractorAgent',
    'QAAgent': 'QAAgent',
    'EvaluatorAgent': 'EvaluatorAgent'
})
# Workers always return to Supervisor for the next task
builder.add_edge('SummarizerAgent', 'Supervisor')
builder.add_edge('EntityExtractorAgent', 'Supervisor')
builder.add_edge('QAAgent', 'Supervisor')
# Evaluator is the terminal node
builder.add_edge('EvaluatorAgent', END)

# Compile the graph into a runnable
graph = builder.compile()

What does this code do?

The StateGraph(DocumentState) creates a graph whose nodes share the DocumentState schema. Each add_node call registers a function as a named node.

The add_conditional_edges call is what makes this a supervisor pattern. Instead of a fixed next node, the Supervisor dynamically picks the next worker based on route_from_supervisor's return value. The path map tells LangGraph which node names are valid destinations.

Workers have fixed edges back to the Supervisor (add_edge) because they always return for the next routing decision. The Evaluator connects to END to terminate the graph.

  • Save pipeline.py.
  • Test the full pipeline end-to-end by running:
python -c "
from pipeline import graph, collection
collection.add(documents=['DocIntel is a startup based in San Francisco. It was founded by Jane Smith in 2024. The company builds AI-powered document analysis tools.'], ids=['test1'])
result = graph.invoke({'document_text': 'DocIntel is a startup based in San Francisco. It was founded by Jane Smith in 2024. The company builds AI-powered document analysis tools.', 'question': '', 'summary': '', 'entities': [], 'qa_answer': '', 'evaluation_scores': {}, 'agent_trace': [], 'current_task': '', 'tasks_completed': []})
print('Summary:', result['summary'][:100])
print('Entities:', result['entities'])
print('Tasks completed:', result['tasks_completed'])
"
  • You should see a summary sentence, a list of entities (Jane Smith, DocIntel, San Francisco), and ['summarize', 'extract_entities'] in the tasks completed list.

Why only two tasks completed?

The test passes an empty question string. The Supervisor skips the QA task when there's no question, routing directly to the Evaluator after summarization and entity extraction. This is the conditional logic working correctly.

Seeing a KeyError or API error?

  • If you see a KeyError: 'GOOGLE_API_KEY', make sure you exported the environment variable in this terminal session.
  • If you see a rate limit or quota error, wait a few seconds and retry. The free tier has per-minute limits.
  • If entities is an empty list, Gemini may have returned malformed JSON. Run the test again. LLM outputs can vary between calls.

Help me debug my LangGraph pipeline test.

✔️ Awesome, I've got everything!

Great. Double check you've saved pipeline.py and that your test printed a summary and entities.

ⓧ I'd like to double check the full code

Here's the complete pipeline.py at this point:

import os
import json
from typing import TypedDict, Annotated, List, Dict, Any, Optional
from langgraph.graph import StateGraph, START, END
from google import genai
import chromadb

client = genai.Client(api_key=os.environ['GOOGLE_API_KEY'])
chroma_client = chromadb.PersistentClient(path='./chroma')
collection = chroma_client.get_or_create_collection('documents')

class DocumentState(TypedDict):
    document_text: str
    question: str
    summary: str
    entities: List[Dict]
    qa_answer: str
    evaluation_scores: Dict[str, float]
    agent_trace: List[Dict]
    current_task: str
    tasks_completed: List[str]

def supervisor_node(state: DocumentState) -> DocumentState:
    completed = state.get('tasks_completed', [])
    if 'summarize' not in completed:
        return {**state, 'current_task': 'summarize'}
    elif 'extract_entities' not in completed:
        return {**state, 'current_task': 'extract_entities'}
    elif 'qa' not in completed and state.get('question'):
        return {**state, 'current_task': 'qa'}
    else:
        return {**state, 'current_task': 'evaluate'}

def route_from_supervisor(state: DocumentState) -> str:
    task = state.get('current_task', '')
    routes = {
        'summarize': 'SummarizerAgent',
        'extract_entities': 'EntityExtractorAgent',
        'qa': 'QAAgent',
        'evaluate': 'EvaluatorAgent'
    }
    return routes.get(task, 'EvaluatorAgent')

def summarizer_agent(state: DocumentState) -> DocumentState:
    import time
    start = time.time()
    response = client.models.generate_content(
        model='gemini-2.0-flash',
        contents=f'Summarize the following document in 2-3 sentences:\n\n{state["document_text"]}'
    )
    summary = response.text
    trace = state.get('agent_trace', [])
    trace.append({'agent': 'SummarizerAgent', 'model': 'gemini-2.0-flash', 'duration': time.time() - start})
    completed = state.get('tasks_completed', []) + ['summarize']
    return {**state, 'summary': summary, 'agent_trace': trace, 'tasks_completed': completed, 'current_task': 'supervisor'}

def entity_extractor_agent(state: DocumentState) -> DocumentState:
    import time
    start = time.time()
    response = client.models.generate_content(
        model='gemini-2.0-flash',
        contents=f'Extract all named entities from the text below. Return a JSON array of objects with keys: text, type (PERSON/ORG/LOCATION). Only return the JSON array, no other text.\n\n{state["document_text"]}'
    )
    try:
        raw = response.text.strip().strip('```json').strip('```').strip()
        entities = json.loads(raw)
    except Exception:
        entities = []
    trace = state.get('agent_trace', [])
    trace.append({'agent': 'EntityExtractorAgent', 'model': 'gemini-2.0-flash', 'duration': time.time() - start})
    completed = state.get('tasks_completed', []) + ['extract_entities']
    return {**state, 'entities': entities, 'agent_trace': trace, 'tasks_completed': completed, 'current_task': 'supervisor'}

def qa_agent(state: DocumentState) -> DocumentState:
    import time
    start = time.time()
    results = collection.query(query_texts=[state['question']], n_results=3)
    chunks = results['documents'][0] if results['documents'] else []
    context = '\n\n'.join(chunks)
    response = client.models.generate_content(
        model='gemini-2.0-flash',
        contents=f'Answer the question using ONLY the context provided. If the answer is not in the context, say so.\n\nContext:\n{context}\n\nQuestion: {state["question"]}'
    )
    trace = state.get('agent_trace', [])
    trace.append({'agent': 'QAAgent', 'model': 'gemini-2.0-flash', 'duration': time.time() - start, 'source_chunks': chunks})
    completed = state.get('tasks_completed', []) + ['qa']
    return {**state, 'qa_answer': response.text, 'agent_trace': trace, 'tasks_completed': completed, 'current_task': 'supervisor'}

def evaluator_node(state: DocumentState) -> DocumentState:
    return {**state, 'evaluation_scores': {}, 'current_task': 'done'}

builder = StateGraph(DocumentState)
builder.add_node('Supervisor', supervisor_node)
builder.add_node('SummarizerAgent', summarizer_agent)
builder.add_node('EntityExtractorAgent', entity_extractor_agent)
builder.add_node('QAAgent', qa_agent)
builder.add_node('EvaluatorAgent', evaluator_node)

builder.add_edge(START, 'Supervisor')
builder.add_conditional_edges('Supervisor', route_from_supervisor, {
    'SummarizerAgent': 'SummarizerAgent',
    'EntityExtractorAgent': 'EntityExtractorAgent',
    'QAAgent': 'QAAgent',
    'EvaluatorAgent': 'EvaluatorAgent'
})
builder.add_edge('SummarizerAgent', 'Supervisor')
builder.add_edge('EntityExtractorAgent', 'Supervisor')
builder.add_edge('QAAgent', 'Supervisor')
builder.add_edge('EvaluatorAgent', END)

graph = builder.compile()

Your multi-agent pipeline is alive. The Supervisor routes tasks, Workers process them with Gemini, and the graph orchestrates the entire flow. Next up, you'll build the full RAG pipeline so the QA agent can answer questions grounded in document context.

Implement the RAG Pipeline

Your multi-agent pipeline can summarize documents and extract entities, but it cannot answer specific questions yet. The QAAgent needs a way to find the relevant parts of a document before passing them to Gemini.

This is where RAG (Retrieval-Augmented Generation) comes in. Instead of feeding the entire document to the LLM, you break it into smaller chunks, store them in a vector database, and retrieve only the most relevant pieces for each question.

In this step, get ready to:
  • Build a document ingestion function that chunks text and stores it in ChromaDB.
  • Update the QAAgent to track retrieved source chunks in state.
  • Test the full RAG flow end-to-end with a sample document and question.
Build the document ingestion function

Before the QAAgent can retrieve relevant chunks, documents need to be broken into smaller segments and stored in ChromaDB. This process is called ingestion.

You will create a function that splits text into 500-character chunks with 100-character overlap, then stores them in the collection. The overlap ensures context is not lost at chunk boundaries.

  • In pipeline.py, find the import lines at the top of the file:
import os
import json
from typing import TypedDict, Annotated, List, Dict, Any, Optional
from langgraph.graph import StateGraph, START, END
from google import genai
import chromadb
  • Replace those lines with the following updated imports (moves time to the top level and removes the unused Annotated import):
import os
import json
import time
from typing import TypedDict, List, Dict, Any, Optional
from langgraph.graph import StateGraph, START, END
from google import genai
import chromadb

Why move time to the top level?

Previously, each agent function imported time inside its own body. Moving it to the top level is cleaner and avoids redundant re-imports on every function call.

  • Next, find the DocumentState class and add a new field source_chunks between qa_answer and evaluation_scores. The full class should now look like this:
class DocumentState(TypedDict):
    document_text: str
    question: str
    summary: str
    entities: List[Dict]
    qa_answer: str
    source_chunks: List[str]
    evaluation_scores: Dict[str, float]
    agent_trace: List[Dict]
    current_task: str
    tasks_completed: List[str]

What does source_chunks store?

This field holds the actual text chunks that ChromaDB retrieved for the QAAgent's answer. Storing them in state lets the Evaluator (and later the API response) show which parts of the document the answer came from.

  • Now add the ingest_document() function directly below the DocumentState class and above supervisor_node. Copy the following code:
def ingest_document(text: str, doc_id: str = 'doc') -> None:
    # Split text into overlapping chunks for retrieval
    chunk_size = 500
    overlap = 100
    chunks = []
    ids = []
    i = 0
    idx = 0
    while i < len(text):
        chunk = text[i:i + chunk_size]
        chunks.append(chunk)
        ids.append(f'{doc_id}_chunk_{idx}')
        i += chunk_size - overlap
        idx += 1
    # Store chunks in ChromaDB (upsert avoids duplicates on re-run)
    collection.upsert(documents=chunks, ids=ids)

What does this function do?

The function walks through the document text with a sliding window. Each chunk is 500 characters long, and the window advances by 400 characters (500 minus the 100-character overlap).

Using upsert instead of add means re-running ingestion on the same document updates existing chunks rather than creating duplicates. ChromaDB handles embedding generation automatically using its built-in all-MiniLM-L6-v2 model.

  • Since time is now imported at the top level, remove the import time line from inside summarizer_agent, entity_extractor_agent, and qa_agent.
  • In each of those three functions, find this line:
    import time
  • Delete that line from all three functions. The start = time.time() call on the next line stays. It now uses the top-level import.
  • Save pipeline.py.

Seeing an IndentationError?

Make sure you deleted only the import time line inside each function. The start = time.time() line directly below it must remain and stay indented at the function body level.

Help me fix this indentation issue

Update the QAAgent to track source chunks

The QAAgent already retrieves chunks from ChromaDB, but it does not store them in the pipeline state. Adding the retrieved chunks to state makes them available to the Evaluator and the final API response.

  • In pipeline.py, scroll to the qa_agent function and find the return statement:
    return {**state, 'qa_answer': response.text, 'agent_trace': trace, 'tasks_completed': completed, 'current_task': 'supervisor'}
  • Replace it with this updated return that includes source_chunks:
    return {**state, 'qa_answer': response.text, 'source_chunks': chunks, 'agent_trace': trace, 'tasks_completed': completed, 'current_task': 'supervisor'}
  • Save pipeline.py.

What changed?

The chunks variable already existed in the function from the ChromaDB query. You are now writing it into the source_chunks field of the state so downstream nodes and the API can access the exact text that grounded the answer.

Test the full RAG flow end-to-end

Time to verify the pipeline works. You will ingest a sample document, ask a question about it, and confirm the answer is grounded in the retrieved context.

  • Create a test file called test_rag.py in the same folder as pipeline.py by copying the following code:
from pipeline import ingest_document, graph

# Sample document about a fictional company
sample_doc = """NextWork Inc. was founded in 2023 by Sarah Chen in San Francisco.
The company builds an educational platform for cloud computing and AI.
NextWork has 50 employees and raised $10 million in Series A funding
led by Horizon Ventures. The platform offers hands-on projects covering
AWS, Azure, and Google Cloud. Sarah Chen previously worked at Google
as a senior engineer for 8 years before starting NextWork."""

# Step 1: Ingest the document into ChromaDB
print("Ingesting document...")
ingest_document(sample_doc, doc_id='test')
print("Document ingested successfully.")

# Step 2: Run the pipeline with a question
print("\nRunning pipeline...")
result = graph.invoke({
    'document_text': sample_doc,
    'question': 'Who founded NextWork and where?',
    'summary': '',
    'entities': [],
    'qa_answer': '',
    'source_chunks': [],
    'evaluation_scores': {},
    'agent_trace': [],
    'current_task': '',
    'tasks_completed': []
})

# Step 3: Display results
print("\n--- Summary ---")
print(result['summary'])
print("\n--- Entities ---")
for entity in result['entities']:
    print(f"  {entity.get('text', '?')} ({entity.get('type', '?')})")
print("\n--- Q&A ---")
print(f"Question: Who founded NextWork and where?")
print(f"Answer: {result['qa_answer']}")
print(f"\n--- Source Chunks Used ---")
for i, chunk in enumerate(result.get('source_chunks', [])):
    print(f"  Chunk {i+1}: {chunk[:80]}...")

What does this test do?

The test script exercises the full RAG pipeline in three stages. First it ingests a sample document into ChromaDB chunks. Then it invokes the LangGraph pipeline with a question. Finally it prints the summary, entities, Q&A answer, and the source chunks used to ground the answer.

  • Run the test by executing this command in your terminal:
python test_rag.py

You should see output showing the ingestion confirmation, a 2-3 sentence summary, a list of extracted entities (people, organizations, locations), and a Q&A answer that mentions Sarah Chen and San Francisco.

The source chunks section confirms the answer came from the retrieved context, not from the model's general knowledge.

Seeing a KeyError or empty Q&A answer?

If you see a KeyError for source_chunks, make sure you added the source_chunks: List[str] field to DocumentState and included it in the graph.invoke() call in your test file.

If the Q&A answer says it cannot find the information, check that the ./chroma directory was created in your project folder after ingestion. This confirms ChromaDB persisted the chunks.

Help me debug my RAG pipeline

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline.py with all the changes from this step.

ⓧ I'd like to double check the full code

Here is the complete pipeline.py after all changes in this step:

import os
import json
import time
from typing import TypedDict, List, Dict, Any, Optional
from langgraph.graph import StateGraph, START, END
from google import genai
import chromadb

client = genai.Client(api_key=os.environ['GOOGLE_API_KEY'])
chroma_client = chromadb.PersistentClient(path='./chroma')
collection = chroma_client.get_or_create_collection('documents')

class DocumentState(TypedDict):
    document_text: str
    question: str
    summary: str
    entities: List[Dict]
    qa_answer: str
    source_chunks: List[str]
    evaluation_scores: Dict[str, float]
    agent_trace: List[Dict]
    current_task: str
    tasks_completed: List[str]

def ingest_document(text: str, doc_id: str = 'doc') -> None:
    chunk_size = 500
    overlap = 100
    chunks = []
    ids = []
    i = 0
    idx = 0
    while i < len(text):
        chunk = text[i:i + chunk_size]
        chunks.append(chunk)
        ids.append(f'{doc_id}_chunk_{idx}')
        i += chunk_size - overlap
        idx += 1
    collection.upsert(documents=chunks, ids=ids)

def supervisor_node(state: DocumentState) -> DocumentState:
    completed = state.get('tasks_completed', [])
    if 'summarize' not in completed:
        return {**state, 'current_task': 'summarize'}
    elif 'extract_entities' not in completed:
        return {**state, 'current_task': 'extract_entities'}
    elif 'qa' not in completed and state.get('question'):
        return {**state, 'current_task': 'qa'}
    else:
        return {**state, 'current_task': 'evaluate'}

def route_from_supervisor(state: DocumentState) -> str:
    task = state.get('current_task', '')
    routes = {
        'summarize': 'SummarizerAgent',
        'extract_entities': 'EntityExtractorAgent',
        'qa': 'QAAgent',
        'evaluate': 'EvaluatorAgent'
    }
    return routes.get(task, 'EvaluatorAgent')

def summarizer_agent(state: DocumentState) -> DocumentState:
    start = time.time()
    response = client.models.generate_content(
        model='gemini-2.0-flash',
        contents=f'Summarize the following document in 2-3 sentences:\n\n{state["document_text"]}'
    )
    summary = response.text
    trace = state.get('agent_trace', [])
    trace.append({'agent': 'SummarizerAgent', 'model': 'gemini-2.0-flash', 'duration': time.time() - start})
    completed = state.get('tasks_completed', []) + ['summarize']
    return {**state, 'summary': summary, 'agent_trace': trace, 'tasks_completed': completed, 'current_task': 'supervisor'}

def entity_extractor_agent(state: DocumentState) -> DocumentState:
    start = time.time()
    response = client.models.generate_content(
        model='gemini-2.0-flash',
        contents=f'Extract all named entities from the text below. Return a JSON array of objects with keys: text, type (PERSON/ORG/LOCATION). Only return the JSON array, no other text.\n\n{state["document_text"]}'
    )
    try:
        raw = response.text.strip().strip('```json').strip('```').strip()
        entities = json.loads(raw)
    except Exception:
        entities = []
    trace = state.get('agent_trace', [])
    trace.append({'agent': 'EntityExtractorAgent', 'model': 'gemini-2.0-flash', 'duration': time.time() - start})
    completed = state.get('tasks_completed', []) + ['extract_entities']
    return {**state, 'entities': entities, 'agent_trace': trace, 'tasks_completed': completed, 'current_task': 'supervisor'}

def qa_agent(state: DocumentState) -> DocumentState:
    start = time.time()
    results = collection.query(query_texts=[state['question']], n_results=3)
    chunks = results['documents'][0] if results['documents'] else []
    context = '\n\n'.join(chunks)
    response = client.models.generate_content(
        model='gemini-2.0-flash',
        contents=f'Answer the question using ONLY the context provided. If the answer is not in the context, say so.\n\nContext:\n{context}\n\nQuestion: {state["question"]}'
    )
    trace = state.get('agent_trace', [])
    trace.append({'agent': 'QAAgent', 'model': 'gemini-2.0-flash', 'duration': time.time() - start, 'source_chunks': chunks})
    completed = state.get('tasks_completed', []) + ['qa']
    return {**state, 'qa_answer': response.text, 'source_chunks': chunks, 'agent_trace': trace, 'tasks_completed': completed, 'current_task': 'supervisor'}

def evaluator_node(state: DocumentState) -> DocumentState:
    return {**state, 'evaluation_scores': {}, 'current_task': 'done'}

builder = StateGraph(DocumentState)
builder.add_node('Supervisor', supervisor_node)
builder.add_node('SummarizerAgent', summarizer_agent)
builder.add_node('EntityExtractorAgent', entity_extractor_agent)
builder.add_node('QAAgent', qa_agent)
builder.add_node('EvaluatorAgent', evaluator_node)

builder.add_edge(START, 'Supervisor')
builder.add_conditional_edges('Supervisor', route_from_supervisor, {
    'SummarizerAgent': 'SummarizerAgent',
    'EntityExtractorAgent': 'EntityExtractorAgent',
    'QAAgent': 'QAAgent',
    'EvaluatorAgent': 'EvaluatorAgent'
})
builder.add_edge('SummarizerAgent', 'Supervisor')
builder.add_edge('EntityExtractorAgent', 'Supervisor')
builder.add_edge('QAAgent', 'Supervisor')
builder.add_edge('EvaluatorAgent', END)

graph = builder.compile()

Your RAG pipeline is working. Documents get chunked, stored in ChromaDB, and retrieved at query time so the LLM only answers from real context. Next up, you will build the comparative evaluation layer that scores these outputs against local HuggingFace model baselines.

Build the Comparative Evaluation Layer

Your multi-agent pipeline can now summarize documents, extract entities, and answer questions using RAG. But how do you know if the outputs are any good?

In production AI systems, you need a way to measure output quality automatically. In this step, you will build a comparative evaluation layer that runs the same document through local HuggingFace models and compares their results against Gemini's outputs to produce confidence scores.

In this step, get ready to:
  • Create HuggingFace baseline functions using BERT-NER and BART-CNN.
  • Implement the Evaluator agent with F1 scoring and LLM-as-Judge.
  • Add trace logging to all agent nodes for observability.
Create the HuggingFace baseline functions

The evaluation strategy compares Gemini's outputs against specialized local models. These local models act as baselines. If Gemini's output closely matches what a purpose-built model produces, you can be more confident in the result.

  • In pipeline.py, find the existing import line that says import chromadb.
  • Add the HuggingFace pipeline import and model initializations directly below it by copying this code:
from transformers import pipeline as hf_pipeline

client = genai.Client(api_key=os.environ['GOOGLE_API_KEY'])
chroma_client = chromadb.PersistentClient(path='./chroma')
collection = chroma_client.get_or_create_collection('documents')

# Load HuggingFace models for baseline comparison
bert_ner = hf_pipeline('ner', model='dslim/bert-base-NER', aggregation_strategy='simple')
bart_summarizer = hf_pipeline('summarization', model='facebook/bart-large-cnn')

What does this code do?

You import the HuggingFace pipeline function with the alias hf_pipeline to avoid a name collision with other uses of "pipeline" in your code.

The two model initializations load dslim/bert-base-NER (a 110M parameter model fine-tuned for named entity recognition) and facebook/bart-large-cnn (a 406M parameter model fine-tuned for summarization). The aggregation_strategy='simple' parameter tells the NER pipeline to group subword tokens into complete entity words.

Now add the two baseline functions that use these models. These go after the ingest_document function.

  • In pipeline.py, find the end of the ingest_document function (the line collection.upsert(documents=chunks, ids=ids)).
  • Add the following two functions below it:
def get_bert_ner_baseline(text: str) -> List[str]:
    # Run BERT-NER on first 512 chars (model's max input)
    results = bert_ner(text[:512])
    return [r['word'].lower() for r in results]

def get_bart_summary_baseline(text: str) -> str:
    # Run BART summarizer on first 1024 chars (model's max input)
    truncated = text[:1024]
    result = bart_summarizer(truncated, max_length=130, min_length=30, do_sample=False)
    return result[0]['summary_text']

What do these baseline functions do?

get_bert_ner_baseline() runs the document text through BERT-NER and returns a list of lowercase entity strings. It truncates input to 512 characters because that is BERT's maximum token window.

get_bart_summary_baseline() runs the text through BART-CNN and returns a summary string. The max_length=130 and min_length=30 parameters control the summary length. do_sample=False ensures deterministic output.

  • Save pipeline.py.

You won't see a visible change yet. The next substep uses these baselines inside the Evaluator.

Seeing an import error?

Make sure you installed transformers and torch in your virtual environment in Step 1. If you see a module-not-found error, activate your virtual environment and run pip install transformers torch again.

help me fix a transformers import error in my pipeline

Implement the Evaluator agent node

The Evaluator needs to answer one question for each output: how confident are we in this result? It does this using two different techniques depending on the task.

For entity extraction, it computes an F1 score measuring the overlap between Gemini's entities and BERT-NER's entities. For summarization, it uses Gemini itself as an LLM-as-Judge to rate how similar its summary is to BART's summary on a 1-5 scale.

  • In pipeline.py, find the end of the get_bart_summary_baseline function you just added.
  • Add the F1 scoring function below it:
def entity_f1_score(gemini_entities: List[Dict], bert_entities: List[str]) -> float:
    # Extract entity text from Gemini's structured output
    gemini_set = set(e.get('text', '').lower() for e in gemini_entities if isinstance(e, dict))
    bert_set = set(bert_entities)
    # Handle edge cases where one or both sets are empty
    if not gemini_set and not bert_set:
        return 1.0
    if not gemini_set or not bert_set:
        return 0.0
    # Calculate F1 as harmonic mean of precision and recall
    intersection = gemini_set & bert_set
    precision = len(intersection) / len(gemini_set)
    recall = len(intersection) / len(bert_set)
    if precision + recall == 0:
        return 0.0
    return 2 * precision * recall / (precision + recall)

What is F1 score?

F1 score is the harmonic mean of precision and recall. Precision measures how many of Gemini's entities also appear in BERT's output ("how many of Gemini's guesses were correct?"). Recall measures how many of BERT's entities Gemini also found ("how many did Gemini miss?").

A score of 1.0 means perfect overlap. A score of 0.0 means no overlap at all. This gives you a normalized confidence metric between 0 and 1.

Now replace the stub evaluator_node function with the full implementation.

  • In pipeline.py, find the existing evaluator_node function (the one that returns empty evaluation_scores).
  • Replace the entire function with:
def evaluator_node(state: DocumentState) -> DocumentState:
    start = time.time()
    scores = {}

    # Entity confidence: F1 between Gemini and BERT-NER
    bert_entities = get_bert_ner_baseline(state['document_text'])
    scores['entities_confidence'] = entity_f1_score(state.get('entities', []), bert_entities)

    # Summary confidence: LLM-as-Judge comparing Gemini vs BART
    bart_summary = get_bart_summary_baseline(state['document_text'])
    judge_response = client.models.generate_content(
        model='gemini-2.0-flash',
        contents=f'Rate the similarity and quality of Summary A compared to Summary B on a scale of 1-5. Return only the integer score.\n\nSummary A (to evaluate): {state.get("summary", "")}\n\nSummary B (reference): {bart_summary}'
    )
    try:
        raw_score = int(judge_response.text.strip()[0])
        scores['summary_confidence'] = (raw_score - 1) / 4.0
    except Exception:
        scores['summary_confidence'] = 0.5

    trace = state.get('agent_trace', [])
    trace.append({'agent': 'EvaluatorAgent', 'model': 'gemini-2.0-flash+hf-baselines', 'timestamp': time.time(), 'duration': time.time() - start, 'scores': scores})
    return {**state, 'evaluation_scores': scores, 'agent_trace': trace, 'current_task': 'done'}

How does the LLM-as-Judge pattern work?

The LLM-as-Judge pattern uses an LLM to evaluate the quality of another LLM's output. Here, Gemini rates its own summary against BART's summary on a 1-5 scale.

The formula (raw_score - 1) / 4.0 normalizes the 1-5 rating to a 0.0-1.0 range. A score of 5 becomes 1.0 (maximum confidence). A score of 1 becomes 0.0 (minimum confidence). The except block defaults to 0.5 if parsing fails, giving a neutral confidence when the judge response is unexpected.

  • Save pipeline.py.

Getting a name error on entity_f1_score?

Make sure entity_f1_score is defined above evaluator_node in the file. Python reads top to bottom, so the function must be defined before it is called.

help me fix a NameError in my evaluator node

Add trace logging to all agent nodes

Production AI systems need observability. You need to know when each agent ran, how long it took, and what model it used. Your Evaluator already records this, but the other agents need timestamps too.

  • In pipeline.py, find the trace.append line inside summarizer_agent.
  • Find this line:
    trace.append({'agent': 'SummarizerAgent', 'model': 'gemini-2.0-flash', 'duration': time.time() - start})
  • Replace it with this line that adds a timestamp field:
    trace.append({'agent': 'SummarizerAgent', 'model': 'gemini-2.0-flash', 'timestamp': time.time(), 'duration': time.time() - start})
  • Now find the trace.append line inside entity_extractor_agent.
  • Find this line:
    trace.append({'agent': 'EntityExtractorAgent', 'model': 'gemini-2.0-flash', 'duration': time.time() - start})
  • Replace it with:
    trace.append({'agent': 'EntityExtractorAgent', 'model': 'gemini-2.0-flash', 'timestamp': time.time(), 'duration': time.time() - start})
  • Finally, find the trace.append line inside qa_agent.
  • Find this line:
    trace.append({'agent': 'QAAgent', 'model': 'gemini-2.0-flash', 'duration': time.time() - start, 'source_chunks': chunks})
  • Replace it with:
    trace.append({'agent': 'QAAgent', 'model': 'gemini-2.0-flash', 'timestamp': time.time(), 'duration': time.time() - start, 'source_chunks': chunks})

Why add timestamps?

Each timestamp records the exact moment (as a Unix epoch) when that agent finished processing. Combined with duration, this gives you a complete timeline of your pipeline execution. In production, this data feeds into dashboards that track latency regressions and identify bottleneck agents.

  • Save pipeline.py.
  • Test the full pipeline with evaluation by running this command in your terminal:
python -c "
import os
os.environ.setdefault('GOOGLE_API_KEY', os.environ.get('GOOGLE_API_KEY', ''))
from pipeline import graph, ingest_document

text = 'Elon Musk founded SpaceX in Hawthorne, California. The company launched its first Falcon 9 rocket in 2010. NASA awarded SpaceX a contract for crew transportation to the International Space Station.'
ingest_document(text)
result = graph.invoke({'document_text': text, 'question': 'When was the first Falcon 9 launched?', 'tasks_completed': [], 'agent_trace': []})
print('Summary:', result.get('summary', '')[:100])
print('Entities:', result.get('entities', []))
print('QA Answer:', result.get('qa_answer', '')[:100])
print('Confidence Scores:', result.get('evaluation_scores', {}))
print('Agent Trace Entries:', len(result.get('agent_trace', [])))
"

You should see confidence scores printed as a dictionary with keys summary_confidence and entities_confidence, each with a value between 0.0 and 1.0. The agent trace should show 4-5 entries (one per agent that ran).

Confidence scores showing 0.0?

A low entity confidence score is normal if the test text is short. BERT-NER and Gemini may tokenize entities differently (for example, "Elon Musk" vs "Elon" and "Musk" as separate entries). Longer documents with more entities will produce higher overlap.

If summary_confidence shows 0.5, the LLM-as-Judge response could not be parsed. Check that your GOOGLE_API_KEY environment variable is set correctly.

help me debug my evaluation confidence scores

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline.py and that your confidence scores print correctly.

ⓧ I'd like to double check the full code

Here is the complete pipeline.py at this point:

import os
import json
import time
from typing import TypedDict, List, Dict, Any, Optional
from langgraph.graph import StateGraph, START, END
from google import genai
import chromadb
from transformers import pipeline as hf_pipeline

client = genai.Client(api_key=os.environ['GOOGLE_API_KEY'])
chroma_client = chromadb.PersistentClient(path='./chroma')
collection = chroma_client.get_or_create_collection('documents')

bert_ner = hf_pipeline('ner', model='dslim/bert-base-NER', aggregation_strategy='simple')
bart_summarizer = hf_pipeline('summarization', model='facebook/bart-large-cnn')

class DocumentState(TypedDict):
    document_text: str
    question: str
    summary: str
    entities: List[Dict]
    qa_answer: str
    source_chunks: List[str]
    evaluation_scores: Dict[str, float]
    agent_trace: List[Dict]
    current_task: str
    tasks_completed: List[str]

def ingest_document(text: str, doc_id: str = 'doc') -> None:
    chunk_size = 500
    overlap = 100
    chunks = []
    ids = []
    i = 0
    idx = 0
    while i < len(text):
        chunk = text[i:i + chunk_size]
        chunks.append(chunk)
        ids.append(f'{doc_id}_chunk_{idx}')
        i += chunk_size - overlap
        idx += 1
    collection.upsert(documents=chunks, ids=ids)

def get_bert_ner_baseline(text: str) -> List[str]:
    results = bert_ner(text[:512])
    return [r['word'].lower() for r in results]

def get_bart_summary_baseline(text: str) -> str:
    truncated = text[:1024]
    result = bart_summarizer(truncated, max_length=130, min_length=30, do_sample=False)
    return result[0]['summary_text']

def entity_f1_score(gemini_entities: List[Dict], bert_entities: List[str]) -> float:
    gemini_set = set(e.get('text', '').lower() for e in gemini_entities if isinstance(e, dict))
    bert_set = set(bert_entities)
    if not gemini_set and not bert_set:
        return 1.0
    if not gemini_set or not bert_set:
        return 0.0
    intersection = gemini_set & bert_set
    precision = len(intersection) / len(gemini_set)
    recall = len(intersection) / len(bert_set)
    if precision + recall == 0:
        return 0.0
    return 2 * precision * recall / (precision + recall)

def supervisor_node(state: DocumentState) -> DocumentState:
    completed = state.get('tasks_completed', [])
    if 'summarize' not in completed:
        return {**state, 'current_task': 'summarize'}
    elif 'extract_entities' not in completed:
        return {**state, 'current_task': 'extract_entities'}
    elif 'qa' not in completed and state.get('question'):
        return {**state, 'current_task': 'qa'}
    else:
        return {**state, 'current_task': 'evaluate'}

def route_from_supervisor(state: DocumentState) -> str:
    task = state.get('current_task', '')
    routes = {
        'summarize': 'SummarizerAgent',
        'extract_entities': 'EntityExtractorAgent',
        'qa': 'QAAgent',
        'evaluate': 'EvaluatorAgent'
    }
    return routes.get(task, 'EvaluatorAgent')

def summarizer_agent(state: DocumentState) -> DocumentState:
    start = time.time()
    response = client.models.generate_content(
        model='gemini-2.0-flash',
        contents=f'Summarize the following document in 2-3 sentences:\n\n{state["document_text"]}'
    )
    summary = response.text
    trace = state.get('agent_trace', [])
    trace.append({'agent': 'SummarizerAgent', 'model': 'gemini-2.0-flash', 'timestamp': time.time(), 'duration': time.time() - start})
    completed = state.get('tasks_completed', []) + ['summarize']
    return {**state, 'summary': summary, 'agent_trace': trace, 'tasks_completed': completed, 'current_task': 'supervisor'}

def entity_extractor_agent(state: DocumentState) -> DocumentState:
    start = time.time()
    response = client.models.generate_content(
        model='gemini-2.0-flash',
        contents=f'Extract all named entities from the text below. Return a JSON array of objects with keys: text, type (PERSON/ORG/LOCATION). Only return the JSON array, no other text.\n\n{state["document_text"]}'
    )
    try:
        raw = response.text.strip().strip('```json').strip('```').strip()
        entities = json.loads(raw)
    except Exception:
        entities = []
    trace = state.get('agent_trace', [])
    trace.append({'agent': 'EntityExtractorAgent', 'model': 'gemini-2.0-flash', 'timestamp': time.time(), 'duration': time.time() - start})
    completed = state.get('tasks_completed', []) + ['extract_entities']
    return {**state, 'entities': entities, 'agent_trace': trace, 'tasks_completed': completed, 'current_task': 'supervisor'}

def qa_agent(state: DocumentState) -> DocumentState:
    start = time.time()
    results = collection.query(query_texts=[state['question']], n_results=3)
    chunks = results['documents'][0] if results['documents'] else []
    context = '\n\n'.join(chunks)
    response = client.models.generate_content(
        model='gemini-2.0-flash',
        contents=f'Answer the question using ONLY the context provided. If the answer is not in the context, say so.\n\nContext:\n{context}\n\nQuestion: {state["question"]}'
    )
    trace = state.get('agent_trace', [])
    trace.append({'agent': 'QAAgent', 'model': 'gemini-2.0-flash', 'timestamp': time.time(), 'duration': time.time() - start, 'source_chunks': chunks})
    completed = state.get('tasks_completed', []) + ['qa']
    return {**state, 'qa_answer': response.text, 'source_chunks': chunks, 'agent_trace': trace, 'tasks_completed': completed, 'current_task': 'supervisor'}

def evaluator_node(state: DocumentState) -> DocumentState:
    start = time.time()
    scores = {}

    # Entity confidence: F1 between Gemini and BERT-NER
    bert_entities = get_bert_ner_baseline(state['document_text'])
    scores['entities_confidence'] = entity_f1_score(state.get('entities', []), bert_entities)

    # Summary confidence: LLM-as-Judge comparing Gemini vs BART
    bart_summary = get_bart_summary_baseline(state['document_text'])
    judge_response = client.models.generate_content(
        model='gemini-2.0-flash',
        contents=f'Rate the similarity and quality of Summary A compared to Summary B on a scale of 1-5. Return only the integer score.\n\nSummary A (to evaluate): {state.get("summary", "")}\n\nSummary B (reference): {bart_summary}'
    )
    try:
        raw_score = int(judge_response.text.strip()[0])
        scores['summary_confidence'] = (raw_score - 1) / 4.0
    except Exception:
        scores['summary_confidence'] = 0.5

    trace = state.get('agent_trace', [])
    trace.append({'agent': 'EvaluatorAgent', 'model': 'gemini-2.0-flash+hf-baselines', 'timestamp': time.time(), 'duration': time.time() - start, 'scores': scores})
    return {**state, 'evaluation_scores': scores, 'agent_trace': trace, 'current_task': 'done'}

builder = StateGraph(DocumentState)
builder.add_node('Supervisor', supervisor_node)
builder.add_node('SummarizerAgent', summarizer_agent)
builder.add_node('EntityExtractorAgent', entity_extractor_agent)
builder.add_node('QAAgent', qa_agent)
builder.add_node('EvaluatorAgent', evaluator_node)

builder.add_edge(START, 'Supervisor')
builder.add_conditional_edges('Supervisor', route_from_supervisor, {
    'SummarizerAgent': 'SummarizerAgent',
    'EntityExtractorAgent': 'EntityExtractorAgent',
    'QAAgent': 'QAAgent',
    'EvaluatorAgent': 'EvaluatorAgent'
})
builder.add_edge('SummarizerAgent', 'Supervisor')
builder.add_edge('EntityExtractorAgent', 'Supervisor')
builder.add_edge('QAAgent', 'Supervisor')
builder.add_edge('EvaluatorAgent', END)

graph = builder.compile()

Your pipeline now produces confidence-scored outputs. Next up, you will wrap the entire system in a FastAPI service so it can accept requests over HTTP.

Serve via FastAPI

Your multi-agent pipeline is fully functional. It orchestrates specialist agents, retrieves grounded context from ChromaDB, and evaluates outputs against local HuggingFace baselines.

Now you need a way for other applications (or users) to interact with this pipeline over HTTP. In this step, you will wrap the entire system in a FastAPI service that accepts document uploads and returns structured, confidence-scored results.

In this step, get ready to:
  • Create a FastAPI app with a POST endpoint wired to the LangGraph pipeline.
  • Define Pydantic response models for structured, confidence-scored output.
  • Test the full pipeline end-to-end via curl and the interactive docs interface.
Create the FastAPI app and request model

The API layer lives in a separate file from the pipeline logic. This keeps concerns separated and lets you import the compiled graph as a dependency.

  • Create a new file called main.py in your project folder.
  • Add the imports, app initialization, and request model by pasting this code:
import time
from fastapi import FastAPI
from pydantic import BaseModel
from typing import List, Dict, Any, Optional
from pipeline import graph, ingest_document

# Initialize the FastAPI application
app = FastAPI(title='DocIntel API')

# Request body schema
class AnalyzeRequest(BaseModel):
    document_text: str
    question: Optional[str] = None

What does this code do?

  • Imports graph and ingest_document directly from pipeline.py. The compiled LangGraph is ready to invoke.
  • Creates a FastAPI instance titled DocIntel API. This title appears in the auto-generated documentation.
  • Defines AnalyzeRequest with a required document_text string and an optional question field. When no question is provided, the Q&A agent is skipped.
  • Save main.py.
  • You won't see a change yet. The next chunk adds the response models that shape the API output.
Define the Pydantic response models

Each piece of the pipeline's output needs a structured shape so consumers know exactly what to expect. You will define small models that compose into a single top-level response.

  • In main.py, add the following response model classes below the AnalyzeRequest class:
# Response models for structured output
class ConfidenceText(BaseModel):
    text: str
    confidence: float

class ConfidenceEntities(BaseModel):
    entities: List[Dict[str, Any]]
    confidence: float

class QAResult(BaseModel):
    answer: str
    confidence: float
    source_chunks: List[str]

class ProcessingMetadata(BaseModel):
    total_time: float
    agent_trace: List[Dict[str, Any]]

class AnalysisResult(BaseModel):
    summary: ConfidenceText
    entities: ConfidenceEntities
    qa_answer: Optional[QAResult] = None
    processing_metadata: ProcessingMetadata

What do these models represent?

  • ConfidenceText pairs a text output (the summary) with a confidence score from the Evaluator.
  • ConfidenceEntities wraps the extracted entities list with its F1-based confidence score.
  • QAResult includes the answer, confidence, and the source chunks used for grounding. It is Optional because Q&A only runs when a question is provided.
  • ProcessingMetadata captures the total request time and the full agent execution trace.
  • AnalysisResult composes all of the above into the top-level response that FastAPI serializes to JSON.
  • Save main.py.
  • You won't see a change yet. The next chunk wires the endpoint that ties everything together.
Wire the endpoint and run the server

The endpoint is where requests arrive, the graph executes, and structured results are returned. You will also add a runner block so you can start the server with a single command.

  • In main.py, add the endpoint function and the uvicorn runner below the AnalysisResult class:
@app.post('/analyze', response_model=AnalysisResult)
def analyze_document(request: AnalyzeRequest):
    start = time.time()
    # Ingest the document into ChromaDB for RAG retrieval
    ingest_document(request.document_text)
    # Build the initial state for the LangGraph pipeline
    initial_state = {
        'document_text': request.document_text,
        'question': request.question or '',
        'summary': '',
        'entities': [],
        'qa_answer': '',
        'source_chunks': [],
        'evaluation_scores': {},
        'agent_trace': [],
        'current_task': '',
        'tasks_completed': []
    }
    # Run the full multi-agent pipeline
    final_state = graph.invoke(initial_state)
    scores = final_state.get('evaluation_scores', {})
    total_time = time.time() - start
    # Build the optional QA result if a question was asked
    qa_result = None
    if request.question and final_state.get('qa_answer'):
        qa_result = QAResult(
            answer=final_state['qa_answer'],
            confidence=scores.get('qa_confidence', 0.5),
            source_chunks=final_state.get('source_chunks', [])
        )
    # Return the structured response
    return AnalysisResult(
        summary=ConfidenceText(text=final_state.get('summary', ''), confidence=scores.get('summary_confidence', 0.0)),
        entities=ConfidenceEntities(entities=final_state.get('entities', []), confidence=scores.get('entities_confidence', 0.0)),
        qa_answer=qa_result,
        processing_metadata=ProcessingMetadata(total_time=total_time, agent_trace=final_state.get('agent_trace', []))
    )

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=8000)

What does this endpoint do?

  • Calls ingest_document() to chunk the document and store it in ChromaDB for the QAAgent's retrieval step.
  • Constructs the full DocumentState dictionary matching the TypedDict schema from pipeline.py.
  • Invokes the compiled LangGraph with graph.invoke(). The Supervisor routes to each Worker, then the Evaluator scores everything.
  • Assembles the AnalysisResult from the final state, pulling confidence scores and agent traces.
  • The if __name__ == '__main__' block starts uvicorn on port 8000 so you can run the server directly with python main.py.
  • Save main.py.
  • Start the server by running this command in your terminal:
python main.py
  • You should see output indicating the server is running on http://0.0.0.0:8000.

What is the /docs page?

FastAPI automatically generates interactive API documentation at http://127.0.0.1:8000/docs. This is powered by Swagger UI and lets you test your endpoint directly in the browser without writing any client code.

  • Open a second terminal window (keep the server running in the first).
  • Test the endpoint by running this curl command:
curl -X POST http://localhost:8000/analyze \
  -H "Content-Type: application/json" \
  -d '{"document_text": "OpenAI was founded in San Francisco by Sam Altman and others. The company developed GPT-4 and ChatGPT, which are used by millions of people worldwide.", "question": "Who founded OpenAI?"}'
  • You should see a JSON response containing summary, entities, qa_answer, and processing_metadata fields, each with confidence scores between 0.0 and 1.0.

Use the interactive docs instead

Open http://127.0.0.1:8000/docs in your browser. Click the POST /analyze endpoint, then click Try it out to fill in the request body and execute the call. The response appears right on the page.

Server not responding?

  • Make sure your GOOGLE_API_KEY environment variable is set in the same terminal where you ran python main.py.
  • Check that pipeline.py is in the same directory as main.py. The import from pipeline import graph, ingest_document requires both files to be in the same folder.
  • If you see a port conflict, another process is using port 8000. Stop it or change the port in main.py.

Why isn't my FastAPI server responding to requests?

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py and that the server is running.

ⓧ I'd like to double check the full code

import time
from fastapi import FastAPI
from pydantic import BaseModel
from typing import List, Dict, Any, Optional
from pipeline import graph, ingest_document

app = FastAPI(title='DocIntel API')

class AnalyzeRequest(BaseModel):
    document_text: str
    question: Optional[str] = None

class ConfidenceText(BaseModel):
    text: str
    confidence: float

class ConfidenceEntities(BaseModel):
    entities: List[Dict[str, Any]]
    confidence: float

class QAResult(BaseModel):
    answer: str
    confidence: float
    source_chunks: List[str]

class ProcessingMetadata(BaseModel):
    total_time: float
    agent_trace: List[Dict[str, Any]]

class AnalysisResult(BaseModel):
    summary: ConfidenceText
    entities: ConfidenceEntities
    qa_answer: Optional[QAResult] = None
    processing_metadata: ProcessingMetadata

@app.post('/analyze', response_model=AnalysisResult)
def analyze_document(request: AnalyzeRequest):
    start = time.time()
    ingest_document(request.document_text)
    initial_state = {
        'document_text': request.document_text,
        'question': request.question or '',
        'summary': '',
        'entities': [],
        'qa_answer': '',
        'source_chunks': [],
        'evaluation_scores': {},
        'agent_trace': [],
        'current_task': '',
        'tasks_completed': []
    }
    final_state = graph.invoke(initial_state)
    scores = final_state.get('evaluation_scores', {})
    total_time = time.time() - start
    qa_result = None
    if request.question and final_state.get('qa_answer'):
        qa_result = QAResult(
            answer=final_state['qa_answer'],
            confidence=scores.get('qa_confidence', 0.5),
            source_chunks=final_state.get('source_chunks', [])
        )
    return AnalysisResult(
        summary=ConfidenceText(text=final_state.get('summary', ''), confidence=scores.get('summary_confidence', 0.0)),
        entities=ConfidenceEntities(entities=final_state.get('entities', []), confidence=scores.get('entities_confidence', 0.0)),
        qa_answer=qa_result,
        processing_metadata=ProcessingMetadata(total_time=total_time, agent_trace=final_state.get('agent_trace', []))
    )

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=8000)

Your document intelligence pipeline is now a fully functional API. Any client can send a document, ask a question, and receive structured, confidence-scored analysis. Next up, push this system further with an adaptive quality gate in the Secret Mission.

Secret mission

Adaptive Quality Gate

Your pipeline evaluates outputs, but what happens when confidence is low? In production systems, you don't just report bad results. You retry with a better strategy. Add a quality gate that automatically re-processes low-confidence outputs with enhanced prompts, and flags results that remain unreliable after retries.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally and uses the Gemini free tier. There are no ongoing costs or paid cloud resources to worry about.

Decide whether to keep your resources for continued development, or delete them to reclaim disk space.

What about Gemini API usage?

Your Gemini API calls consume your free-tier quota. The free tier provides free input and output tokens for gemini-3.5-flash, but your usage contributes to improving Google products.

For a portfolio project this is perfectly acceptable. No charges will appear on your account.

Resources you used:

  • HuggingFace model weights (~2GB cached in ~/.cache/huggingface/hub).
  • ChromaDB vector data (stored in the ./chroma directory inside your project folder).
  • Project files (pipeline.py, main.py, virtual environment, and any other files in your project folder).

Keep everything running

No action needed. Choose this if you plan to keep developing the DocIntel pipeline or want to reference it later.

  • Your HuggingFace model cache stays in ~/.cache/huggingface/hub (~2GB). Future projects that use the same models will load instantly without re-downloading.
  • Your ChromaDB data remains in ./chroma inside your project folder.
  • Your virtual environment and project files stay on your Desktop.

Pause - I'll come back to this later

Since everything is local, there are no running processes or services to pause. Your files stay on disk until you return.

  • If you left the FastAPI server running, stop it by pressing Ctrl+C in the terminal where uvicorn is running.
  • Everything else is just files on disk. No ongoing processes or costs.

Delete - I don't want to use this again

Remove all project resources and cached model weights to reclaim disk space.

  • If the FastAPI server is still running, stop it by pressing Ctrl+C in the terminal.
  • Delete the cached HuggingFace model weights by running:
rm -rf ~/.cache/huggingface/hub/models--dslim--bert-base-NER
rm -rf ~/.cache/huggingface/hub/models--facebook--bart-large-cnn
  • Delete the ChromaDB data directory by running this from inside your project folder:
rm -rf ./chroma
  • Delete the entire project folder (including the virtual environment and all project files) by running:
cd ~/Desktop
rm -rf docintel

What about the GOOGLE_API_KEY environment variable?

If you exported GOOGLE_API_KEY only in your terminal session, it disappears when you close that terminal window. If you added it to your shell profile (~/.zshrc or ~/.bashrc), open that file and remove the export line.

That's a Wrap!

That's a Wrap!

Nice work! You just built a production-grade document intelligence system that processes documents through specialized AI agents, evaluates their outputs against local model baselines, and serves everything through a structured API.

You've learned how to:

  • Build a multi-agent pipeline using LangGraph's supervisor-worker pattern to orchestrate specialized AI agents for summarization, entity extraction, and document Q&A.
  • Implement a RAG pipeline with ChromaDB for grounded document retrieval and a comparative evaluation system that scores LLM outputs against local HuggingFace model baselines using entity overlap F1 and LLM-as-Judge patterns.
  • Serve the entire system via FastAPI with structured Pydantic response models delivering per-output confidence scores and full processing metadata.
  • Secret Mission: Implemented an adaptive quality gate with automatic retry and escalation for low-confidence outputs.

Ready to quiz yourself?