Build a RAG Agent with LangChain

Build a PDF research agent with RAG, tool calling, and a streaming API.

Introduction

30 Second Summary

Imagine you have a stack of research papers and need to find specific answers buried across hundreds of pages. Manually searching through each document is slow, and you might miss the most relevant passage entirely.

In this project, you will build a research agent that ingests PDF documents, retrieves relevant passages, and answers your questions with source citations. You'll expose it through a streaming API with full observability tracing every decision the agent makes.

What You'll Build

A modular Python application that combines a RAG ingestion pipeline, a LangChain agent with custom tools, a streaming FastAPI server, and LangSmith observability.

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

  • A RAG pipeline that loads PDFs, chunks text, creates Gemini embeddings, and stores them in a ChromaDB vector store.
  • A LangChain agent with custom retrieval and web search tools that cites its sources and handles errors gracefully through middleware.
  • A streaming FastAPI endpoint that delivers agent responses token-by-token using Server-Sent Events.
  • Full LangSmith observability tracing every agent decision, tool call, and retrieval step.
  • Secret Mission: Add conversation memory so the agent remembers previous questions within a session.

Are there any prerequisites?

This project assumes intermediate Python experience and familiarity with virtual environments. You'll need Python 3.10+ installed, a Google API key for Gemini (free tier, no credit card required), and a free LangSmith account.

Set Up Your Project and Dependencies

Before building your research agent, you need a clean project environment with the right packages in place. Getting the setup right from the start prevents dependency headaches later.

You'll create a virtual environment and install packages for LangChain, Gemini, ChromaDB, and FastAPI. Then you'll run a quick test to confirm your API key is working.

In this step, get ready to:
  • Create a project directory and Python virtual environment.
  • Install dependencies and configure API keys.
  • Set up project files and verify your Gemini API key.
Create your project directory and virtual environment

A virtual environment isolates your project's packages from the rest of your system. This keeps dependencies clean and avoids version conflicts between projects.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir research-agent
cd research-agent
  • Create a Python virtual environment by running:
uv venv

What does uv venv do?

The uv venv command creates a .venv folder inside your project directory. This folder contains an isolated Python environment where all your packages will be installed.

Using uv instead of the built-in venv module is significantly faster. It's written in Rust and handles dependency resolution 10-100x quicker than pip.

  • Confirm the .venv folder was created by running:
ls -a

You should see .venv listed in the output.

Now activate the virtual environment so that packages install into it instead of your system Python.

macOS

  • Activate the virtual environment by running:
source .venv/bin/activate

Windows

  • Activate the virtual environment by running:
.venv\Scripts\activate

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

Don't see (.venv) in your prompt?

Make sure you ran the activation command from inside the research-agent folder, not from your Desktop.

On Windows, if you see an execution policy error, run Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned first, then try activation again.

help me activate my Python virtual environment with uv

Install dependencies and configure your API keys

Your research agent relies on several packages for different parts of its architecture. You'll install them all at once, then set up a configuration file for your API keys.

  • Install all required packages by running:
uv pip install langchain langchain-google-genai langchain-text-splitters langchain-community chromadb fastapi uvicorn python-dotenv pypdf

What are all these packages?

  • langchain is the core framework for building agents and tool-calling workflows.
  • langchain-google-genai connects LangChain to Google's Gemini models for chat and embeddings.
  • langchain-text-splitters breaks documents into smaller chunks for embedding.
  • langchain-community provides integrations like the PDF loader and ChromaDB vector store.
  • chromadb is a local vector database for storing and searching document embeddings.
  • fastapi and uvicorn together power the streaming API server.
  • python-dotenv loads environment variables from a .env file.
  • pypdf is the PDF parsing engine used by the document loader.
  • Wait for the installation to finish. You should see a message listing all installed packages and their versions.

Seeing installation errors?

Make sure (.venv) appears in your terminal prompt. If it does not, the virtual environment is not active and packages may install to the wrong location.

If you see a No module named langchain.agents error later, you may need to install the pre-release version. Run uv pip install --pre langchain to get the latest LangChain v1 release.

help me troubleshoot my package installation with uv

Your API keys and tracing configuration need to live in a .env file so your code can load them securely at runtime.

  • Create a .env file in the research-agent folder by running:
touch .env
  • Open .env in your text editor and add the following content:
GOOGLE_API_KEY=your-google-api-key-here
LANGSMITH_API_KEY=your-langsmith-api-key-here
LANGSMITH_TRACING=true

What do these variables do?

  • GOOGLE_API_KEY authenticates your requests to the Gemini API. You created this key during your Google AI Studio setup.
  • LANGSMITH_API_KEY connects your agent to LangSmith for observability tracing. You can find this key in your LangSmith dashboard at smith.langchain.com.
  • LANGSMITH_TRACING=true enables automatic trace collection. Every agent decision and tool call will be logged.
  • Replace your-google-api-key-here with your actual Google API key.
  • Replace your-langsmith-api-key-here with your LangSmith API key.
  • Save the .env file.

Keep .env out of version control

If you plan to use Git with this project, add .env to your .gitignore file. API keys should never be committed to a repository.

  • Confirm the file contents by running:
cat .env

You should see your three environment variables listed with your actual key values.

✔️ Awesome, I've got everything!

Great. Make sure you saved your .env file before moving on.

ⓧ I'd like to double check the full code

Your .env file should look exactly like this (with your real keys substituted in):

GOOGLE_API_KEY=your-google-api-key-here
LANGSMITH_API_KEY=your-langsmith-api-key-here
LANGSMITH_TRACING=true
Create the project files and test your setup

Your agent will be split across four Python files, each handling a different responsibility. You'll also create a folder to hold the PDF documents your agent will ingest.

  • Create the empty project files and the docs folder by running these commands:
touch ingest.py tools.py agent.py server.py
mkdir docs

What will each file do?

  • ingest.py loads PDFs, splits them into chunks, and stores embeddings in the vector database.
  • tools.py defines the retrieval and web search tools the agent can call.
  • agent.py assembles the LangChain agent with tools, a system prompt, and error-handling middleware.
  • server.py exposes the agent through a streaming FastAPI endpoint.
  • docs/ is where you'll place PDF files for the RAG pipeline to ingest.
  • Confirm all files were created by running:
ls

You should see ingest.py, tools.py, agent.py, server.py, the docs folder, and .env in the output.

Time to verify your Gemini API key works before building anything on top of it.

  • Create a temporary test file called test_setup.py with the following code:
from langchain_google_genai import ChatGoogleGenerativeAI
from dotenv import load_dotenv

# Load API keys from the .env file
load_dotenv()

# Create a Gemini LLM instance
llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash")

# Send a test message and print the response
response = llm.invoke("Say hello in one sentence")
print(response.content)

What does this test do?

This script loads your GOOGLE_API_KEY from the .env file using python-dotenv. Then it creates a Gemini chat model through the LangChain integration and sends a simple prompt.

If the API key is valid and the packages are installed correctly, you'll see a short greeting printed in your terminal.

  • Run the test by executing:
python test_setup.py

You should see a short greeting from Gemini printed in your terminal, something like "Hello there, how can I help you today?"

Seeing an error instead of a greeting?

If you see GOOGLE_API_KEY not found or Invalid API key, open your .env file and confirm you pasted your actual key after the = sign with no extra spaces or quotes.

If you see ModuleNotFoundError, your virtual environment may not be active. Check for (.venv) in your terminal prompt and re-run the activation command if needed.

help me fix my Gemini API test

  • Clean up the test file by running:
rm test_setup.py

Your project environment is ready and your API key is confirmed working. Next up, you'll build the RAG ingestion pipeline to load and embed PDF documents into a vector store.

Build the RAG Ingestion Pipeline

Your project is set up, your dependencies are installed, and your Gemini API key is confirmed working. Now it's time to build the heart of your research agent: the ingestion pipeline.

An ingestion pipeline takes raw PDF documents, breaks them into smaller text chunks, converts those chunks into numerical representations called vector embeddings, and stores everything in a database for fast retrieval. This is the foundation that makes your agent capable of answering questions about specific documents.

In this step, get ready to:
  • Load PDF documents and split them into text chunks.
  • Embed the chunks and store them in a ChromaDB vector database.
  • Verify the pipeline by running a similarity search against the stored data.
Write the ingestion script

Your ingestion pipeline lives in ingest.py. It needs to handle four responsibilities: loading PDFs, splitting text into chunks, creating embeddings, and storing everything in ChromaDB.

  • Open ingest.py in your editor and add the imports and configuration by pasting this code:
import os
from pathlib import Path

from dotenv import load_dotenv
from langchain_community.document_loaders import PyPDFLoader
from langchain_google_genai import GoogleGenerativeAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma

# Load environment variables from .env
load_dotenv()

# Paths for document source and vector storage
DOCS_DIR = Path("docs")
CHROMA_DIR = Path("chroma_db")

What does this code do?

These imports bring in the four key tools your pipeline needs.

  • PyPDFLoader reads PDF files and converts each page into a LangChain Document object.
  • RecursiveCharacterTextSplitter breaks long documents into smaller chunks while trying to keep paragraphs and sentences intact.
  • GoogleGenerativeAIEmbeddings converts text into numerical vectors using Gemini's embedding model.
  • Chroma is the vector store that saves and searches those embeddings locally on disk.

DOCS_DIR points to the folder where your PDFs live. CHROMA_DIR is where ChromaDB will persist its data between runs.

  • Save ingest.py.

You won't see a change yet. The next chunk adds the function that uses all of these imports.

  • Below the constants you just added in ingest.py, paste the ingest_documents() function and the script's entry point:

def ingest_documents():
    """Load PDFs, split into chunks, embed, and store in ChromaDB."""
    # Loop through every PDF in the docs/ folder
    documents = []
    for pdf_path in DOCS_DIR.glob("*.pdf"):
        loader = PyPDFLoader(str(pdf_path))
        documents.extend(loader.load())

    # Exit early if no PDFs were found
    if not documents:
        print("No PDF files found in docs/ folder.")
        return

    print(f"Loaded {len(documents)} pages from {len(list(DOCS_DIR.glob('*.pdf')))} PDFs.")

    # Split documents into smaller chunks for better retrieval
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200,
    )
    chunks = text_splitter.split_documents(documents)
    print(f"Split into {len(chunks)} chunks.")

    # Create vector embeddings using Gemini
    embeddings = GoogleGenerativeAIEmbeddings(model="gemini-embedding-2-preview")

    # Store the embedded chunks in ChromaDB
    vector_store = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=str(CHROMA_DIR),
        collection_name="research_docs",
    )
    print(f"Stored {len(chunks)} chunks in ChromaDB at {CHROMA_DIR}.")

    # Quick verification: search for chunks related to 'summary'
    results = vector_store.similarity_search("summary", k=2)
    print(f"\nVerification - top 2 results for 'summary':")
    for i, doc in enumerate(results):
        source = doc.metadata.get("source", "unknown")
        page = doc.metadata.get("page", "?")
        print(f"  [{i+1}] {source} (page {page}): {doc.page_content[:100]}...")


if __name__ == "__main__":
    ingest_documents()

What does this function do?

This single function handles the entire RAG ingestion pipeline in four stages.

  • Load: It loops through every PDF in docs/ and extracts each page as a Document with metadata (file name, page number).
  • Split: RecursiveCharacterTextSplitter breaks pages into 1000-character chunks with 200-character overlap. The overlap ensures context is not lost at chunk boundaries.
  • Embed: Each chunk is converted into a vector using Gemini's gemini-embedding-2-preview model.
  • Store: Chroma.from_documents() saves every embedded chunk to disk in the chroma_db/ directory so it persists between runs.

The final section runs a quick similarity search for "summary" to confirm the vector store is working. The if __name__ == "__main__" guard ensures this function only runs when you execute the file directly.

Why 1000 characters with 200 overlap?

Chunk size controls how much text each vector represents. Chunks that are too large dilute the meaning. Chunks that are too small lose context.

1000 characters is a solid default for most document types. The 200-character overlap means neighboring chunks share some text, so important sentences that fall on a boundary are not cut in half.

  • Save ingest.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved ingest.py before moving on.

ⓧ I'd like to double check the full code

Here is the complete ingest.py file. Make sure yours matches this exactly.

import os
from pathlib import Path

from dotenv import load_dotenv
from langchain_community.document_loaders import PyPDFLoader
from langchain_google_genai import GoogleGenerativeAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma

load_dotenv()

DOCS_DIR = Path("docs")
CHROMA_DIR = Path("chroma_db")


def ingest_documents():
    """Load PDFs, split into chunks, embed, and store in ChromaDB."""
    documents = []
    for pdf_path in DOCS_DIR.glob("*.pdf"):
        loader = PyPDFLoader(str(pdf_path))
        documents.extend(loader.load())

    if not documents:
        print("No PDF files found in docs/ folder.")
        return

    print(f"Loaded {len(documents)} pages from {len(list(DOCS_DIR.glob('*.pdf')))} PDFs.")

    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200,
    )
    chunks = text_splitter.split_documents(documents)
    print(f"Split into {len(chunks)} chunks.")

    embeddings = GoogleGenerativeAIEmbeddings(model="gemini-embedding-2-preview")

    vector_store = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=str(CHROMA_DIR),
        collection_name="research_docs",
    )
    print(f"Stored {len(chunks)} chunks in ChromaDB at {CHROMA_DIR}.")

    results = vector_store.similarity_search("summary", k=2)
    print(f"\nVerification - top 2 results for 'summary':")
    for i, doc in enumerate(results):
        source = doc.metadata.get("source", "unknown")
        page = doc.metadata.get("page", "?")
        print(f"  [{i+1}] {source} (page {page}): {doc.page_content[:100]}...")


if __name__ == "__main__":
    ingest_documents()
Add a sample PDF and run the pipeline

Your script is ready, but it needs at least one PDF in the docs/ folder to process. Any PDF will work here.

  • Find any PDF file on your computer (a research paper, a report, or any multi-page document).
  • Copy or move that PDF into the docs/ folder inside your project directory.

Need a quick PDF?

You can download any free research paper from arxiv.org. Search for a topic you find interesting, open a paper, and click the PDF link. Save it directly into your docs/ folder.

  • Run the ingestion script by executing this command in your terminal:
python ingest.py

The script takes a few seconds to run. Gemini needs to generate an embedding for every chunk, so the time depends on the size of your PDF.

What should I see?

You should see output similar to this (exact numbers depend on your PDF):

Loaded 12 pages from 1 PDFs.
Split into 24 chunks.
Stored 24 chunks in ChromaDB at chroma_db.

Verification - top 2 results for 'summary':
  [1] docs/your-paper.pdf (page 0): The first 100 characters of a matching chunk...
  [2] docs/your-paper.pdf (page 3): The first 100 characters of another chunk...

The key things to look for: a non-zero number of pages loaded, chunks created, and two verification results printed at the bottom.

Seeing an error?

  • If you see No PDF files found in docs/ folder. make sure you placed a .pdf file directly inside the docs/ folder (not in a subfolder).
  • If you see an API error mentioning your key, double check that GOOGLE_API_KEY is correctly set in your .env file and that you activated your virtual environment.
  • If you see a ModuleNotFoundError, confirm you installed all dependencies in the previous step and that your virtual environment is active.

Help me debug my ingest.py script

Verify the vector store

The verification output at the bottom of the script confirms that your pipeline worked end-to-end. The similarity search queried the vector store for the word "summary" and returned the two most relevant chunks.

  • Confirm that the chroma_db/ directory was created by running:
ls chroma_db/

You should see files and folders inside chroma_db/. This is your persisted vector store on disk.

What just happened?

Each verification result shows the source file, page number, and a preview of the matching text. This metadata is what your agent will use later to cite its sources when answering questions.

The chroma_db/ directory stores all your embedded chunks on disk. Your agent can load this data in future sessions without re-running the ingestion.

Your vector store is loaded and ready to search. Next up, you'll build the custom tools that let your agent query this data and fall back to web search when needed.

Create Custom Tools for the Agent

Your ingestion pipeline is up and running. The ChromaDB vector store is populated with document chunks, ready to be searched.

Now you need to give your agent the ability to actually use that data. LangChain agents interact with external systems through tools. Each tool is a Python function that the agent can call when it needs specific information. In this step, you'll build two tools: a retrieval tool that searches your vector store, and a web search fallback for when the documents don't have the answer.

In this step, get ready to:
  • Build a retrieve_context tool that searches ChromaDB for relevant document chunks.
  • Add a web_search fallback tool using the Wikipedia API.
  • Test both tools independently to verify they return the expected format.
Set up the tools file and vector store helper

Before writing the tools themselves, you need a helper function that loads the existing ChromaDB vector store. Both tools will rely on this helper to access your embedded documents.

  • Create a new file called tools.py in your project folder and add the imports, constants, and helper function by pasting this code:
import os
import urllib.request
import urllib.error
import json
from pathlib import Path

from dotenv import load_dotenv
from langchain.tools import tool
from langchain_google_genai import GoogleGenerativeAIEmbeddings
from langchain_community.vectorstores import Chroma

load_dotenv()

CHROMA_DIR = Path("chroma_db")


def get_vector_store():
    """Load the existing ChromaDB vector store."""
    embeddings = GoogleGenerativeAIEmbeddings(model="gemini-embedding-2-preview")
    return Chroma(
        persist_directory=str(CHROMA_DIR),
        embedding_function=embeddings,
        collection_name="research_docs",
    )

What does this code do?

The imports pull in everything both tools need: urllib for web requests, json for parsing responses, and the LangChain packages for tool creation and vector store access.

The get_vector_store() helper reconnects to the same ChromaDB collection you populated in the previous step. It uses the same embedding model (gemini-embedding-2-preview) and collection name (research_docs) so the queries match the stored embeddings.

You won't see a change yet. The next chunk adds the first tool that uses this helper.

  • Below the get_vector_store function in tools.py, add the retrieval tool by pasting this code:
@tool(response_format="content_and_artifact")
def retrieve_context(query: str):
    """Search the document knowledge base for information relevant to the query.
    Use this tool first before trying web search.
    Returns relevant text passages with source citations."""
    vector_store = get_vector_store()
    results = vector_store.similarity_search(query, k=3)

    if not results:
        return "No relevant documents found in the knowledge base.", []

    serialized = "\n\n".join(
        f"Source: {doc.metadata.get('source', 'unknown')} | "
        f"Page: {doc.metadata.get('page', '?')}\n"
        f"Content: {doc.page_content}"
        for doc in results
    )
    return serialized, results

What does this tool do?

The @tool(response_format="content_and_artifact") decorator registers this function as a LangChain tool. The content_and_artifact format means the tool returns two things: serialized text for the LLM to read, and the raw document objects for the application to use.

The function runs a similarity_search against ChromaDB, pulling the 3 most relevant chunks. It formats each result with its source file name and page number so the agent can cite where the information came from.

  • Save tools.py.

You won't see a change yet. You'll test this tool after adding the web search fallback.

Add the web_search fallback tool

Sometimes the ingested documents won't contain the answer. Your agent needs a fallback for those situations. This tool queries the Wikipedia API, which is free and requires no additional API keys.

  • Below the retrieve_context function in tools.py, add the web search fallback tool by pasting this code:
@tool
def web_search(query: str) -> str:
    """Search the web for information when the document knowledge base
    does not contain relevant results. Use this as a fallback only."""
    try:
        encoded_query = urllib.parse.quote(query)
        url = f"https://en.wikipedia.org/w/api.php?action=query&list=search&srsearch={encoded_query}&format=json&srlimit=3"
        req = urllib.request.Request(
            url,
            headers={"User-Agent": "ResearchAgent/1.0"},
        )
        with urllib.request.urlopen(req, timeout=10) as resp:
            data = json.loads(resp.read().decode("utf-8"))

        results = data.get("query", {}).get("search", [])
        if not results:
            return "No web results found."

        formatted = "\n\n".join(
            f"Title: {r['title']}\nSnippet: {r['snippet']}"
            for r in results
        )
        return f"Web search results for '{query}':\n\n{formatted}"
    except (urllib.error.URLError, Exception) as e:
        return f"Web search failed: {e}"

What does this tool do?

This tool URL-encodes the query and sends it to Wikipedia's search API, requesting the top 3 results. Each result comes back with a title and a text snippet.

The entire function is wrapped in a try/except block. If the network request fails, the tool returns a helpful error message instead of crashing the agent.

Why Wikipedia instead of Google Search?

Google Search APIs require their own API key and billing setup. Wikipedia's search API is completely free, requires no authentication, and still demonstrates the tool-calling pattern perfectly. You can swap in a production search API later without changing the agent architecture.

  • Save tools.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved tools.py before moving on.

ⓧ I'd like to double check the full code

Here is the complete tools.py file. Compare your code against it:

import os
import urllib.request
import urllib.error
import json
from pathlib import Path

from dotenv import load_dotenv
from langchain.tools import tool
from langchain_google_genai import GoogleGenerativeAIEmbeddings
from langchain_community.vectorstores import Chroma

load_dotenv()

CHROMA_DIR = Path("chroma_db")


def get_vector_store():
    """Load the existing ChromaDB vector store."""
    embeddings = GoogleGenerativeAIEmbeddings(model="gemini-embedding-2-preview")
    return Chroma(
        persist_directory=str(CHROMA_DIR),
        embedding_function=embeddings,
        collection_name="research_docs",
    )


@tool(response_format="content_and_artifact")
def retrieve_context(query: str):
    """Search the document knowledge base for information relevant to the query.
    Use this tool first before trying web search.
    Returns relevant text passages with source citations."""
    vector_store = get_vector_store()
    results = vector_store.similarity_search(query, k=3)

    if not results:
        return "No relevant documents found in the knowledge base.", []

    serialized = "\n\n".join(
        f"Source: {doc.metadata.get('source', 'unknown')} | "
        f"Page: {doc.metadata.get('page', '?')}\n"
        f"Content: {doc.page_content}"
        for doc in results
    )
    return serialized, results


@tool
def web_search(query: str) -> str:
    """Search the web for information when the document knowledge base
    does not contain relevant results. Use this as a fallback only."""
    try:
        encoded_query = urllib.parse.quote(query)
        url = f"https://en.wikipedia.org/w/api.php?action=query&list=search&srsearch={encoded_query}&format=json&srlimit=3"
        req = urllib.request.Request(
            url,
            headers={"User-Agent": "ResearchAgent/1.0"},
        )
        with urllib.request.urlopen(req, timeout=10) as resp:
            data = json.loads(resp.read().decode("utf-8"))

        results = data.get("query", {}).get("search", [])
        if not results:
            return "No web results found."

        formatted = "\n\n".join(
            f"Title: {r['title']}\nSnippet: {r['snippet']}"
            for r in results
        )
        return f"Web search results for '{query}':\n\n{formatted}"
    except (urllib.error.URLError, Exception) as e:
        return f"Web search failed: {e}"
Test both tools independently

Before wiring these tools into an agent, verify each one works on its own. You'll call each tool directly and check that the output format is correct.

  • Test the retrieval tool by running this command:
python -c "from tools import retrieve_context; print(retrieve_context.invoke('summary'))"

You should see formatted text with source file names, page numbers, and content snippets from your PDF. This confirms the retrieval tool is connecting to your ChromaDB vector store and finding relevant chunks.

Seeing an error about the collection or embeddings?

Make sure you ran ingest.py successfully in the previous step. The chroma_db/ folder must exist and contain data.

Check that the collection_name in get_vector_store() matches exactly: research_docs. A mismatch means the tool can't find the stored embeddings.

Help me debug my retrieve_context tool

  • Test the web search fallback tool by running this command:
python -c "from tools import web_search; print(web_search.invoke('machine learning'))"

You should see Wikipedia search results with titles and text snippets related to machine learning. This confirms the web search fallback is working and can fetch results when the local documents don't have the answer.

Web search returning an error?

If you see Web search failed: followed by a URL error, check your internet connection. The tool needs network access to reach the Wikipedia API.

If you see a timeout error, try running the command again. The Wikipedia API occasionally takes longer than the 10-second timeout.

Help me debug my web_search tool

Your tools are built and tested. Next up, you'll assemble the agent itself and wire these tools into a LangChain agent that can reason about which tool to call and when.

Assemble the Agent with Middleware

Your retrieval and web search tools are tested and ready. The next challenge is giving your agent the intelligence to decide when to use each one.

In this step, you'll assemble a LangChain agent that reasons about which tool to call, follows rules you define in a system prompt, and recovers gracefully when things go wrong.

In this step, get ready to:
  • Define a system prompt with retrieval-first rules and citation instructions.
  • Build error-handling middleware to catch tool failures.
  • Test the agent end-to-end and verify the trace in LangSmith.
Define the system prompt and imports

The system prompt is the set of instructions your agent follows every time it handles a question. It controls the agent's behaviour: which tool to try first, when to fall back, and how to format citations.

  • Open agent.py in your editor and add the imports and system prompt by copying the following code:
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage

from tools import retrieve_context, web_search

load_dotenv()

# Instructions the agent follows for every question
SYSTEM_PROMPT = """You are a research assistant that answers questions using a document knowledge base.

Rules:
1. Always use the retrieve_context tool FIRST to search the knowledge base.
2. If the knowledge base does not have relevant information, use the web_search tool as a fallback.
3. Always cite your sources. Include the document name and page number when citing from the knowledge base.
4. If you cannot find relevant information from either source, say so honestly.
5. Keep answers clear, structured, and concise."""

What does this code do?

  • create_agent is LangChain v1's agent factory. It wires a model, tools, and a prompt into a runnable agent.
  • wrap_tool_call and ToolMessage are the building blocks for middleware. You'll use them in the next chunk.
  • retrieve_context and web_search are imported from your tools.py file so the agent can use them.
  • SYSTEM_PROMPT tells the agent to always search the knowledge base first, cite sources with document name and page number, and only fall back to web search when needed.
  • Save agent.py.

You won't see any output yet. The next chunk adds the middleware layer that protects the agent from tool failures.

Add error-handling middleware

When a tool crashes (network timeout, missing file, bad query), you don't want the entire agent to fail. Middleware wraps each tool call so the agent receives a helpful error message instead of a stack trace.

  • In agent.py, add the middleware function below the SYSTEM_PROMPT by copying this code:
@wrap_tool_call
def handle_tool_errors(request, handler):
    """Middleware that catches tool errors and returns a helpful message."""
    try:
        return handler(request)
    except Exception as e:
        return ToolMessage(
            content=f"Tool '{request.tool_call['name']}' failed: {str(e)}. "
            "Try rephrasing or using a different tool.",
            tool_call_id=request.tool_call["id"],
        )

What does this code do?

  • @wrap_tool_call registers this function as LangChain middleware that runs around every tool call.
  • handler(request) executes the actual tool. If the tool succeeds, its result passes through unchanged.
  • If the tool throws an exception, the except block catches it and returns a ToolMessage with a friendly error. The agent sees this message and can decide to try a different tool or rephrase the query.
  • Save agent.py.

You won't see a change yet. The next chunk creates the agent function that uses this middleware.

Why middleware instead of try/except inside each tool?

You could add error handling inside every tool function, but that duplicates the same try/except pattern everywhere. Middleware lets you write the error logic once and apply it to all tools automatically.

It also keeps your tools focused on their core job (retrieval, search) while the middleware handles the cross-cutting concern of error recovery.

Create and test the research agent

The create_research_agent function is where everything comes together. It passes the model, tools, system prompt, and middleware into LangChain's create_agent factory.

  • In agent.py, add the agent factory function and a test block below the middleware by copying this code:
def create_research_agent():
    """Create and return the research agent."""
    agent = create_agent(
        # Use Gemini as the reasoning model
        model="google_genai:gemini-2.5-flash",
        # Give the agent access to both tools
        tools=[retrieve_context, web_search],
        # Set the system-level instructions
        system_prompt=SYSTEM_PROMPT,
        # Wrap every tool call in error handling
        middleware=[handle_tool_errors],
    )
    return agent


if __name__ == "__main__":
    agent = create_research_agent()
    result = agent.invoke(
        {"messages": [{"role": "user", "content": "What are the main topics covered in the documents?"}]}
    )
    print(result["messages"][-1].content)

What does this code do?

  • create_agent builds a runnable agent from four pieces: the Gemini model, the two tools, the system prompt, and the middleware.
  • model="google_genai:gemini-2.5-flash" tells LangChain to use the Gemini 2.5 Flash model via the Google GenAI integration.
  • The __main__ block sends a test question and prints the agent's final response. This lets you run the file directly to verify everything works.
  • Save agent.py.
  • Run the agent by running this command in your terminal:
python agent.py

You should see the agent's answer printed in your terminal. The response should reference content from your ingested PDF and include source citations with document names and page numbers.

What should I see?

The agent takes a few seconds to think. Behind the scenes, it calls retrieve_context to search your ChromaDB vector store, reads the results, and generates a structured answer.

Your output will look different depending on the PDF you ingested, but you should see a coherent summary that names specific topics from the document.

Seeing an error?

  • If you see ModuleNotFoundError: No module named 'langchain.agents', make sure your virtual environment is activated and that you installed langchain in Step 1.
  • If you see an API key error, double-check that your .env file contains a valid GOOGLE_API_KEY and that load_dotenv() is at the top of your file.
  • If the agent returns a response with no document citations, make sure you ran ingest.py in Step 2 and that the chroma_db/ folder exists in your project directory.
  • Still stuck? Help me debug an error when running agent.py

✔️ Awesome, I've got everything!

Great. Double check you've saved agent.py before moving on.

ⓧ I'd like to double check the full code

Here is the complete agent.py file. Compare it with yours line by line.

import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage

from tools import retrieve_context, web_search

load_dotenv()

SYSTEM_PROMPT = """You are a research assistant that answers questions using a document knowledge base.

Rules:
1. Always use the retrieve_context tool FIRST to search the knowledge base.
2. If the knowledge base does not have relevant information, use the web_search tool as a fallback.
3. Always cite your sources. Include the document name and page number when citing from the knowledge base.
4. If you cannot find relevant information from either source, say so honestly.
5. Keep answers clear, structured, and concise."""


@wrap_tool_call
def handle_tool_errors(request, handler):
    """Middleware that catches tool errors and returns a helpful message."""
    try:
        return handler(request)
    except Exception as e:
        return ToolMessage(
            content=f"Tool '{request.tool_call['name']}' failed: {str(e)}. "
            "Try rephrasing or using a different tool.",
            tool_call_id=request.tool_call["id"],
        )


def create_research_agent():
    """Create and return the research agent."""
    agent = create_agent(
        model="google_genai:gemini-2.5-flash",
        tools=[retrieve_context, web_search],
        system_prompt=SYSTEM_PROMPT,
        middleware=[handle_tool_errors],
    )
    return agent


if __name__ == "__main__":
    agent = create_research_agent()
    result = agent.invoke(
        {"messages": [{"role": "user", "content": "What are the main topics covered in the documents?"}]}
    )
    print(result["messages"][-1].content)

Now let's check that LangSmith captured the full trace of what the agent just did.

  • Open your browser and go to smith.langchain.com.
  • Sign in if prompted.
  • Click on your most recent trace in the dashboard.

You should see a trace that shows the agent's full reasoning chain: the initial message, the retrieve_context tool call, the retrieved chunks, and the final generated answer.

Why is LangSmith tracing useful?

LangSmith records every decision your agent makes. You can see exactly which tool it called, what input it sent, what the tool returned, and how the model used that context to form its answer.

This is essential for debugging. If the agent gives a bad answer, the trace tells you whether the problem was in retrieval (wrong chunks), the prompt (unclear instructions), or the model itself.

Your research agent is alive and reasoning through questions with full observability. Next up, you'll expose it through a streaming FastAPI endpoint so clients can receive answers token by token in real time.

Build the Streaming FastAPI Server

Your research agent can answer questions, retrieve context from your documents, and fall back to web search. Right now, you can only interact with it by running agent.py directly in the terminal.

To make the agent useful to other applications (a web frontend, a mobile app, or even another script), you need to expose it through an API. In this step, you will build a FastAPI server that streams the agent's response token by token using Server-Sent Events.

In this step, get ready to:
  • Write a FastAPI server with a streaming /chat endpoint.
  • Implement Server-Sent Events streaming with LangChain's async streaming API.
  • Test the server with a Python client and verify the full trace in LangSmith.
Set up the FastAPI server file

FastAPI is a Python web framework that makes it easy to define API endpoints with automatic request validation. You will use it to create a /chat endpoint that accepts a question and streams the agent's answer back.

  • Open server.py in your editor and add the server foundation by pasting this code:
import os
import json

from dotenv import load_dotenv
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

# Import the agent factory from agent.py
from agent import create_research_agent

load_dotenv()

# Create the FastAPI app and a single agent instance
app = FastAPI(title="Research Agent API")
agent = create_research_agent()


# Define the expected shape of incoming requests
class ChatRequest(BaseModel):
    question: str

What does this code do?

FastAPI creates the web server. StreamingResponse is how FastAPI sends data back to the client in chunks instead of all at once.

The agent is created once at module level using create_research_agent() so it gets initialized when the server starts. Every incoming request shares this same agent instance.

ChatRequest is a Pydantic model that tells FastAPI to expect a JSON body with a question field. FastAPI validates this automatically and returns an error if the field is missing.

You will not see a change yet. The next chunk adds the streaming logic that powers the endpoint.

Implement the streaming response logic

When a user sends a question, you do not want them to wait for the entire answer to generate before seeing anything. Server-Sent Events (SSE) let the server push individual tokens to the client as they are produced.

What are Server-Sent Events?

SSE is a standard protocol where the server sends a stream of data: lines over a single HTTP connection. The client reads each line as it arrives. This is how ChatGPT and other AI tools show text appearing word by word.

  • In server.py, add the streaming generator function below the ChatRequest class:
async def stream_agent_response(question: str):
    """Stream the agent response as Server-Sent Events."""
    input_message = {"role": "user", "content": question}

    # Stream tokens as they are generated by the agent
    async for chunk in agent.astream(
        {"messages": [input_message]},
        stream_mode="messages",
        version="v2",
    ):
        # Each chunk with type 'messages' contains a token and metadata
        if chunk["type"] == "messages":
            token, metadata = chunk["data"]
            if hasattr(token, "content") and token.content:
                # Format each token as an SSE event
                event_data = json.dumps({"token": token.content})
                yield f"data: {event_data}\n\n"

    # Signal the end of the stream
    yield "data: [DONE]\n\n"

What does this code do?

This is an async generator function. Instead of returning one value, it yields multiple values over time.

agent.astream() is LangChain's async streaming method. The stream_mode="messages" setting tells it to stream individual LLM tokens. The version="v2" flag uses LangChain's recommended streaming format.

Each token is wrapped in a JSON object and sent as an SSE data: line. The final [DONE] event tells the client the stream is complete.

You will not see a change yet. The next chunk wires this generator to the API endpoints.

  • Add the route handlers and the server entry point at the bottom of server.py:
@app.post("/chat")
async def chat(request: ChatRequest):
    """Stream a response to the user's question."""
    return StreamingResponse(
        stream_agent_response(request.question),
        media_type="text/event-stream",
    )


@app.get("/health")
async def health():
    return {"status": "ok"}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run("server:app", host="0.0.0.0", port=8000, reload=True)

What does this code do?

  • The /chat endpoint accepts a POST request, passes the question to the streaming generator, and returns a StreamingResponse with the SSE media type.
  • The /health endpoint is a simple health check that returns {"status": "ok"}. This is useful for monitoring tools to verify the server is running.
  • The __main__ block starts the server with uvicorn on port 8000 with hot reload enabled, so changes to your code restart the server automatically.
  • Save server.py.

Seeing an import error for FastAPI?

  • Make sure you installed fastapi and uvicorn back in Step 1. Run uv pip install fastapi uvicorn if they are missing.
  • Confirm your virtual environment is activated. You should see (.venv) at the start of your terminal prompt.

Help me fix an import error in my FastAPI server.

Double check your full server.py file against the reference below.

✔️ Awesome, I've got everything!

Make sure you have saved server.py before moving on.

ⓧ I'd like to double check the full code

import os
import json

from dotenv import load_dotenv
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

from agent import create_research_agent

load_dotenv()

app = FastAPI(title="Research Agent API")
agent = create_research_agent()


class ChatRequest(BaseModel):
    question: str


async def stream_agent_response(question: str):
    """Stream the agent response as Server-Sent Events."""
    input_message = {"role": "user", "content": question}

    async for chunk in agent.astream(
        {"messages": [input_message]},
        stream_mode="messages",
        version="v2",
    ):
        if chunk["type"] == "messages":
            token, metadata = chunk["data"]
            if hasattr(token, "content") and token.content:
                event_data = json.dumps({"token": token.content})
                yield f"data: {event_data}\n\n"

    yield "data: [DONE]\n\n"


@app.post("/chat")
async def chat(request: ChatRequest):
    """Stream a response to the user's question."""
    return StreamingResponse(
        stream_agent_response(request.question),
        media_type="text/event-stream",
    )


@app.get("/health")
async def health():
    return {"status": "ok"}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run("server:app", host="0.0.0.0", port=8000, reload=True)
Start and test the streaming server

The server is written. Now you will start it, send a real question, and watch tokens stream back in real time.

  • Start the server by running this command:
python server.py
  • You should see output that includes Uvicorn running on http://0.0.0.0:8000. This means the server is ready to accept requests.

Seeing 'Address already in use'?

Another process is using port 8000. Close any other servers running on that port, or stop the process using it.

Help me find and stop the process using port 8000.

The server is running, but you need to keep it running while you test it. You will open a second terminal window for the test commands.

  • Open a new terminal window or tab. Make sure your virtual environment is activated in this new terminal as well.
  • Test the health endpoint first by running this command:
curl http://localhost:8000/health
  • You should see {"status":"ok"} printed in the terminal. This confirms the server is running and reachable.

Now you will create a small test client script to verify the streaming behavior. A simple curl command works too, but a Python script lets you parse the SSE stream and print tokens as they arrive.

  • Create a new file called test_client.py in your project folder and paste this code:
import urllib.request
import json


def test_chat():
    """Simple test client for the streaming endpoint."""
    url = "http://localhost:8000/chat"
    # Build the request with a test question
    data = json.dumps({"question": "What are the main topics in the documents?"}).encode("utf-8")
    req = urllib.request.Request(
        url,
        data=data,
        headers={"Content-Type": "application/json"},
        method="POST",
    )

    print("Streaming response:\n")
    # Read the SSE stream line by line
    with urllib.request.urlopen(req) as resp:
        for line in resp:
            decoded = line.decode("utf-8").strip()
            if decoded.startswith("data: "):
                payload = decoded[6:]
                if payload == "[DONE]":
                    print("\n\n[Stream complete]")
                    break
                # Parse each token and print it immediately
                token_data = json.loads(payload)
                print(token_data.get("token", ""), end="", flush=True)


if __name__ == "__main__":
    test_chat()

What does this code do?

The test client sends a POST request to your /chat endpoint with a test question. It then reads the response line by line, looking for SSE data: lines.

Each token is extracted from the JSON and printed with flush=True so it appears immediately in your terminal. You should see text appearing word by word, just like a ChatGPT-style interface.

  • Save test_client.py.
  • Run the test client in your second terminal by running this command:
python test_client.py
  • You should see the agent's answer appearing token by token in your terminal, followed by [Stream complete] at the end. The answer should reference content from the PDF you ingested earlier.

Seeing 'Connection refused'?

  • Check that the server is still running in your first terminal. You should see the Uvicorn running on http://0.0.0.0:8000 message.
  • Make sure your virtual environment is activated in both terminal windows.
  • Look at the server terminal for error messages. If you see an import error or traceback, the server crashed on startup.

Help me debug why my test client cannot connect to the server.

  • Open your LangSmith dashboard at smith.langchain.com.
  • Find the most recent trace. You should see the full agent execution for the API request, including the retrieval tool call, the documents it found, and the final answer generation.

Why check LangSmith for API requests?

When the agent runs behind an API, you cannot see its internal steps in the terminal anymore. LangSmith captures every agent decision, tool call, and retrieval step regardless of how the agent was invoked. This is what makes it essential for debugging production agents.

Your research agent is now live as a streaming API. Any application can send a question to /chat and receive the agent's answer token by token. Next up, a Secret Mission to give your agent a memory so it can handle follow-up questions.

Secret mission

Add Conversation Memory

Your agent answers questions well, but it forgets everything between requests. In this secret mission, you will add session-based conversation memory so the agent remembers previous questions and can handle follow-up queries like a real research assistant.

Before you go

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing cloud costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Python virtual environment (.venv folder inside your project directory).
  • Project files: ingest.py, tools.py, agent.py, server.py, test_client.py, and .env.
  • ChromaDB persistent storage (chroma_db/ folder).
  • Sample PDFs in the docs/ folder.
  • Google API key (Gemini free tier, no charges).
  • LangSmith account (free tier, 5,000 traces/month).

Keep everything running

No action needed. Choose this if you are still actively building or plan to extend the project soon.

  • Your virtual environment, ChromaDB data, and project files stay in place on your Desktop.
  • Your Google API key and LangSmith account remain active on their free tiers at no cost.
  • If the uvicorn server is still running, it continues to use a small amount of memory.

Pause - I'll come back to this later

Shut down any running processes to free up memory. Your files and data stay on disk so you can pick up where you left off.

  • Stop the uvicorn server by pressing Ctrl+C in the terminal where it is running.
  • Deactivate your virtual environment by running deactivate in your terminal.
  • Your project folder, ChromaDB data, and .env file remain on disk. When you are ready to resume, reactivate the virtual environment and start the server again.

Delete - I don't want to use this again

Remove all project resources and start fresh. This permanently deletes your project files, virtual environment, and ingested data.

  • Stop the uvicorn server by pressing Ctrl+C if it is still running.
  • Delete the entire project folder from your Desktop. This removes all Python files, the .venv virtual environment, the chroma_db/ data, and the docs/ folder.
  • Optionally, revoke your Google API key in the Google AI Studio console if you no longer need it.
  • Optionally, delete your LangSmith API key from the LangSmith settings page if you no longer need it.

Mission accomplished

That's a Wrap!

Nice work! You just built a production-grade RAG research agent powered by LangChain, Gemini, and LangSmith. From PDF ingestion to streaming API responses with full observability, this project covers six distinct AI engineering competencies.

Here's what you learned:

  • Built a complete RAG ingestion pipeline with PDF loading, text chunking, Gemini embeddings, and ChromaDB vector storage.
  • Created a LangChain agent with custom retrieval and web search tools, plus error-handling middleware for graceful tool failure recovery.
  • Exposed the agent through a streaming FastAPI endpoint using Server-Sent Events and wired up LangSmith observability to trace every agent decision and tool call.
  • Added conversation memory so the agent maintains context across questions within a session.

Ready to quiz yourself?