DEV Community: David Archanjo

Chat with Your Docs: Working with LangChain, RAG, MCP, ChromaDB Made Simple

David Archanjo — Mon, 27 Jul 2026 18:40:10 +0000

Have you ever wished you could ask questions from a pile of documents, spreadsheets, or PDFs sitting on your machine and magically get answered? It's not magic and in this guide, we will build exactly that: a local Q&A system that lets we chat with our documents using AI techniques and tools.

We will combine Retrieval Augmented Generation (RAG) with the Model Context Protocol (MCP) to create a modular, production-ready pipeline. By the end of this article, we will have a working system composed of three services: a document ingestion server, an MCP tool server, and a conversational AI LangChain-powered agent.

What We Will Build

A powerful Q&A system from documents.
Understand the core concepts of RAG architecture.
Set up and interact with ChromaDB vector store.
Implement data loading using LangChain document loaders.
Stream responses for a great user experience.
Integrate Retrieval Augmented Generation (RAG).

About Me

I am David Archanjo, a Full-Stack Engineer with 10+ years of experience in the technology industry. I am passionate about software development and AI Engineering, and I enjoy sharing practical knowledge through technical articles that help developers build real-world applications.

Background
Why RAG and MCP?
Step 1: Setting Up the Environment
Step 2: Loading and Indexing Documents into ChromaDB
Step 3: Exposing MCP tools
Step 4: Building the RAG chain
Step 5: Running and Testing the Q&A Agent
Wrapping Up

Background

Large Language Models (LLMs) excel at reasoning and text generation, but their knowledge is limited to what they were trained on. They cannot access our private documents, internal company data, or any information not included in their training, which makes them unable to answer questions about our knowledge base without additional context.

Retrieval-Augmented Generation (RAG) addresses this limitation by retrieving more relevant information, for instance, from our documents at query time and providing it to the LLM as context, enabling grounded responses. In this guide, we expose that retrieval layer as a Model Context Protocol (MCP) tool, creating a reusable service that any MCP-compatible agent or application can leverage.

Why RAG and MCP?

The Case for RAG

Without RAG, an LLM can only answer questions based on what it learned during training.

It cannot:

Answer questions about our private or proprietary documents.
Reflect updates that happened after its training cutoff.
Cite specific sources or passages to back up its answers.

RAG solves these problems by combining semantic search with LLM reasoning. Our documents are first split into small, semantically meaningful chunks, converted into embeddings (vector representations of their meaning), and stored in a vector database. When a user submits a question, the query is embedded into the same vector space, allowing the system to retrieve the most relevant chunks based on semantic similarity rather than keyword matching. Those retrieved passages are then injected into the LLM's prompt as context, enabling the model to generate responses that are grounded in our data, more accurate, and less prone to hallucinations.

Key benefits of RAG:

Accuracy: answers come from our actual documents.
Transparency: we can trace every answer back to a specific source.
Cost efficiency: we only process what is relevant to the question.
Freshness: new documents can be added at any time without retraining the model.

The Case for MCP

The Model Context Protocol is an open standard that defines how AI agents discover and invoke external tools.

By exposing our document search capability as an MCP tool, we gain several architectural advantages:

Modularity: the retrieval service is completely decoupled from the agent.
Reusability: the same MCP server can serve multiple agents or applications simultaneously.
Discoverability: agents can automatically discover available tools and their descriptions at runtime.
Standardization: an open protocol decouples our integration from any single framework, reducing vendor lock-in and ensuring long-term interoperability.

Combining RAG with MCP gives the best of both worlds: the accuracy and contextual grounding of retrieval-augmented generation with the flexibility, interoperability, and maintainability of a standardized protocol for integrating tools and data sources.

Step 1: Setting Up the Environment

Before writing any code, let us get the project structure in place and install the required dependencies.

Project Structure

.
├── .env                        # Environment variables
├── .db/                        # ChromaDB persistent storage
├── agent.py                    # Conversational RAG agent
├── assets/                     # Sample documents for testing
│   ├── car_prices.csv
│   ├── company_credentials.txt
│   ├── meeting_notes.md
│   └── resume.pdf
├── config.py                   # Centralized configuration
├── doc_ingestion_server.py     # Document ingestion pipeline & API
├── doc_mcp_server.py           # MCP tool server
├── docs/                       # Watched document directory
└── requirements.txt            # Python dependencies

NOTE: You can find these sample documents in the accompanying GitHub repository.

Our project is organized around three separate services, each with a clear responsibility:

Service	File	Responsibility
Ingestion Server	`doc_ingestion_server.py`	Load, chunk, embed, and store documents
MCP Tool Server	`doc_mcp_server.py`	Expose documents' tools via MCP protocol
AI Agent	`agent.py`	Accept user queries and orchestrate retrieval

Installing Dependencies

Create and activate a virtual environment:

python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

Create a requirements.txt file and add the following:

fastapi>=0.139.2
fastmcp>=3.4.4
langchain>=1.3.14
langchain-chroma>=1.1.0
langchain-community>=0.4.2
langchain-mcp-adapters>=0.3.0
langchain-openai>=1.4.0
pypdf>=6.14.2
pydantic-settings
python-dotenv

Install the dependencies:

pip install -r requirements.txt

Environment Configuration

Create a .env at the root of the project with the following contents:

DOCS_DIR="./docs"
CHROMA_DIR="./.db"
CHROMA_COLLECTION_NAME="documents"
EMBEDDING_MODEL_NAME="mxbai-embed-large-v1"
EMBEDDING_MODEL_BASE_URL="http://localhost:8001/v1"
CHUNK_SIZE=500
CHUNK_OVERLAP=50
LLM_API_KEY="sk-xxxxx"
LLM_BASE_URL="http://localhost:8000/v1"
LLM_MODEL_NAME="gpt-4o"
LLM_MODEL_PROVIDER="openai"
LLM_TEMPERATURE=0.2
MCP_SERVER_NAME="document-search"
MCP_SERVER_URL="http://localhost:3001/mcp"

Breaking down each variable:

Variable	Purpose
`DOCS_DIR`	Directory where uploaded documents are stored
`CHROMA_DIR`	Persistent storage directory for ChromaDB
`CHROMA_COLLECTION_NAME`	Name of the ChromaDB collection
`EMBEDDING_MODEL_NAME`	Name of the embedding model
`EMBEDDING_MODEL_BASE_URL`	Base URL for the embedding model API
`CHUNK_SIZE`	Maximum number of characters per document chunk
`CHUNK_OVERLAP`	Overlap in characters between adjacent chunks
`LLM_API_KEY`	API key for the language model
`LLM_BASE_URL`	Base URL for the LLM API
`LLM_MODEL_NAME`	Name of the language model to use
`LLM_MODEL_PROVIDER`	Name of the language model's provider (e.g. `openai`, `anthropic`, `bedrock`)
`LLM_TEMPERATURE`	Controls the creativity of model responses
`MCP_SERVER_NAME`	Identifier for the MCP server
`MCP_SERVER_URL`	URL where the MCP server is reachable

Local OpenAI-Compatible API Service

In this article, we will interact with language and embedding models through an OpenAI-compatible API. LangChain's integrations can communicate with any server implementing this interface. Luckily this API does not require the official OpenAI service, so any server implementing the same HTTP interface can be used transparently.

We will run both the chat model and the embedding model locally. I recommend using llamafile, a standalone distribution of llama.cpp, as this approach keeps the entire stack running on our own machine while allowing the application code to remain identical to what we would write for the official OpenAI API.

The following llamafile models are recommended:

Purpose	Llamafile Model
Chat model	Qwen3.5-2B-Q8_0.llamafile
Embedding model	mxbai-embed-large-v1-f16.llamafile

Start the embedding model:

./mxbai-embed-large-v1-f16.llamafile \
  --server \
  --host 0.0.0.0 \
  --port 8001 \
  --nobrowser \
  --embedding \
  --alias mxbai-embed-large-v1

In another terminal, start the chat model :

./Qwen3.5-2B-Q8_0.llamafile \
  --server \
  --host 0.0.0.0 \
  --port 8000 \
  --alias gpt-4o

The --alias gpt-4o option exposes our local language model under the name gpt-4o. From LangChain's perspective, it behaves exactly like if it was connected to the OpenAI GPT-4o model, allowing us to work with the same LLM initialization setup.

NOTE: If you prefer using the official OpenAI service, simply replace LLM_API_KEY with your own API key, and update EMBEDDING_MODEL_BASE_URL if your embedding service is running elsewhere, or remove EMBEDDING_MODEL_BASE_URL if you're using OpenAI embeddings. No other change is required.

Why llamafile? I personally recommend llamafile because it packages llama.cpp, the model weights, and all required runtime components into a single executable. There is no installation, daemon service, package manager, or additional Python environment to maintain. We simply download the file, make it executable, and run it. Since it is built on top of llama.cpp, it inherits excellent performance while remaining portable across Linux, macOS and Windows.

The Configuration Class

# config.py
from pydantic_settings import BaseSettings

class EnvironmentConfiguration(BaseSettings):
    DOCS_DIR: str
    CHROMA_DIR: str
    CHROMA_COLLECTION_NAME: str
    EMBEDDING_MODEL_NAME: str
    EMBEDDING_MODEL_BASE_URL: str
    CHUNK_SIZE: int
    CHUNK_OVERLAP: int
    LLM_API_KEY: str
    LLM_BASE_URL: str
    LLM_MODEL_NAME: str
    LLM_MODEL_PROVIDER: str
    LLM_TEMPERATURE: float
    MCP_SERVER_NAME: str
    MCP_SERVER_URL: str

    class Config:
        env_file = ".env"

The config.py file extends BaseSettings class from pydantic-settings to load all environment variables automatically from the .env file. By centralizing configuration this way, every service (i.e. doc_ingestion_server.py, doc_mcp_server.py and agent.py) imports the same EnvironmentConfiguration class and receives consistent, validated settings.

Step 2: Loading and Indexing Documents into ChromaDB

The ingestion pipeline is the foundation of our RAG system. It is responsible for reading raw documents, splitting them into chunks, converting those chunks into vector embeddings, and persisting everything in ChromaDB so the MCP server can query it later.

The full implementation will live in doc_ingestion_server.py, which is built as a FastAPI web server so we can upload new documents via API request at runtime without having to restart the server.

"""
Document ingestion pipeline & server.

Start the server:
    uvicorn doc_ingestion_server:app --host 0.0.0.0 --port 3002

The ingestion pipeline is automatically executed once during server startup.

Upload new documents through:
    POST /documents
"""

import sys
import logging
import shutil
from config import EnvironmentConfiguration
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, UploadFile
from langchain_chroma import Chroma
from langchain_community.document_loaders import (
    CSVLoader,
    DirectoryLoader,
    TextLoader,
    PyPDFLoader
)
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from pathlib import Path


ALLOWED_FILE_EXTENSIONS = {".csv", ".txt", ".md", ".pdf"}

config = EnvironmentConfiguration()

logger = logging.getLogger("uvicorn.error")

embeddings = OpenAIEmbeddings(
    model=config.EMBEDDING_MODEL_NAME,
    base_url=config.EMBEDDING_MODEL_BASE_URL,
    check_embedding_ctx_length=False
)


def split_documents(loader):
    documents = loader.load()
    if not documents:
        return []

    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=config.CHUNK_SIZE,
        chunk_overlap=config.CHUNK_OVERLAP,
        separators=["\n\n", "\n", ". ", " ", ""],
    )
    chunks = text_splitter.split_documents(documents)
    logger.info("Loaded %d document(s) → %d chunk(s).", len(documents), len(chunks))
    return chunks


def embed_and_write(chunks):
    if len(chunks) == 0:
        Chroma(
            collection_name=config.CHROMA_COLLECTION_NAME,
            persist_directory=config.CHROMA_DIR,
            embedding_function=embeddings
        )
        return

    Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        collection_name=config.CHROMA_COLLECTION_NAME,
        persist_directory=config.CHROMA_DIR,
    )
    logger.info("Ingestion complete. %d chunks indexed into ChromaDB.", len(chunks))


def ingest():
    logger.info("Starting ingestion pipeline.")

    loader = DirectoryLoader(
        config.DOCS_DIR,
        glob=["**/*.txt", "**/*.md"],
        loader_cls=TextLoader,
        loader_kwargs={"encoding": "utf-8"}
    )
    chunks = split_documents(loader)

    csv_loader = DirectoryLoader(
        config.DOCS_DIR,
        glob="**/*.csv",
        loader_cls=CSVLoader,
        loader_kwargs={"encoding": "utf-8"}
    )
    chunks.extend(split_documents(csv_loader))

    pdf_loader = DirectoryLoader(
        config.DOCS_DIR,
        glob="**/*.pdf",
        loader_cls=PyPDFLoader
    )
    chunks.extend(split_documents(pdf_loader))

    embed_and_write(chunks)


@asynccontextmanager
async def lifespan(app: FastAPI):
    try:
        Path(config.DOCS_DIR).mkdir(parents=True, exist_ok=True)
        ingest()
        yield
    except Exception as e:
        logger.critical("Failed to initialize: %s", e)
        sys.exit(1)


app = FastAPI(title="Document Ingestion Server", lifespan=lifespan)


@app.post("/documents")
async def upload_document(file: UploadFile):
    if not file.filename:
        raise HTTPException(status_code=400, detail="Filename is required.")
    extension = Path(file.filename).suffix.lower()
    if extension not in ALLOWED_FILE_EXTENSIONS:
        raise HTTPException(status_code=400, detail=f"Only '{ALLOWED_FILE_EXTENSIONS}' files are supported.")

    destination = Path(config.DOCS_DIR) / file.filename
    with destination.open("wb") as buffer:
        shutil.copyfileobj(file.file, buffer)

    try:
        if extension == ".csv":
            loader = CSVLoader(str(destination), encoding="utf-8")
        elif extension == ".pdf":
            loader = PyPDFLoader(str(destination))
        else:
            loader = TextLoader(str(destination), encoding="utf-8")
        chunks = split_documents(loader)
        embed_and_write(chunks)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"File saved but ingestion failed: {e}")

    return {"message": "Document uploaded and ingested successfully.", "filename": file.filename}

Breaking Down the Code

Allowed File Types

ALLOWED_FILE_EXTENSIONS = {".csv", ".txt", ".md", ".pdf"}

We restrict ingestion to four file formats. Each format has a dedicated LangChain's community loader which understands its structure:

CSVLoader parses rows and columns
PyPDFLoader extracts text page by page
TextLoader handles plain text and Markdown files

Embeddings Initialization

embeddings = OpenAIEmbeddings(
    model=config.EMBEDDING_MODEL_NAME,
    base_url=config.EMBEDDING_MODEL_BASE_URL,
    check_embedding_ctx_length=False
)

We initialize an OpenAIEmbeddings instance pointing to our locally hosted embedding model (configured via EMBEDDING_MODEL_BASE_URL). Notice we have check_embedding_ctx_length=False settings. This specific setting disables the token-length guard which is required when working with locally served models that may not report context limits the same way the OpenAI API does.

The split_documents Function

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=config.CHUNK_SIZE,
    chunk_overlap=config.CHUNK_OVERLAP,
    separators=["\n\n", "\n", ". ", " ", ""],
)

This is the heart of the chunking strategy. RecursiveCharacterTextSplitter tries to split text at natural boundaries, i.e. first at double newlines (paragraph breaks), then single newlines, then sentence boundaries, and so on, working down the list of separators until the resulting chunk is within the configured CHUNK_SIZE (500 by default).

The CHUNK_OVERLAP (50 characters by default) ensures that context is not lost at chunk boundaries. If a sentence spans two chunks, the tail of the first chunk appears at the head of the second, so retrieval can always return complete thoughts.

The embed_and_write Function

Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    collection_name=config.CHROMA_COLLECTION_NAME,
    persist_directory=config.CHROMA_DIR,
)

Chroma.from_documents takes the list of chunks, calls the embedding model to convert each chunk into a dense vector, and writes both the vectors and the original text to disk under CHROMA_DIR. ChromaDB automatically persists this data, so the MCP server can load it later without re-embedding anything.

When the chunk list is empty (no documents found during startup), we initialize an empty ChromaDB collection as fallback. This prevents errors downstream when the MCP server tries to connect before any documents have been uploaded.

The ingest Function

def ingest():
    # Text and Markdown
    loader = DirectoryLoader(config.DOCS_DIR, glob=["**/*.txt", "**/*.md"], ...)
    chunks = split_documents(loader)

    # CSV
    csv_loader = DirectoryLoader(config.DOCS_DIR, glob="**/*.csv", ...)
    chunks.extend(split_documents(csv_loader))

    # PDF
    pdf_loader = DirectoryLoader(config.DOCS_DIR, glob="**/*.pdf", ...)
    chunks.extend(split_documents(pdf_loader))

    embed_and_write(chunks)

The ingest function runs three separate DirectoryLoader passes: one per document family because each file type requires a different loader class. All resulting chunks are collected into a single list before being written to ChromaDB in one batch, which is more efficient than writing each file type separately, at least for the purpose of this guide.

The FastAPI Lifespan and Upload Endpoint

@asynccontextmanager
async def lifespan(app: FastAPI):
    Path(config.DOCS_DIR).mkdir(parents=True, exist_ok=True)
    ingest()
    yield

FastAPI's lifespan context manager runs ingest() exactly once at startup. This means that if documents already exist in DOCS_DIR from a previous run, they are re-indexed automatically. No manual intervention is required.

The POST /documents endpoint handles runtime uploads. A user can submit a new file and it will be chunked and embedded immediately, making it searchable without restarting either service. upload_document processes the file upload by selecting the appropriate loader class based on the file extension.

Ingestion Server Startup

uvicorn doc_ingestion_server:app --host 0.0.0.0 --port 3002

Once running, we can upload new documents like this:

curl -X POST http://localhost:3002/documents -F "file=@assets/meeting_notes.md"

Step 3: Exposing MCP Tools

Now with our documents properly indexed in ChromaDB, we need to equip our AI agent with tool calling capability. This is where the Model Context Protocol (MCP) comes in. Instead of hard-coding retrieval logic inside the agent (though this is valid in some context), we publish it as a discoverable, decoupled, protocol-compliant tool which any MCP-compatible client can use.

The full implementation will live in doc_mcp_server.py, which is also built as a FastAPI web server.

"""
FastMCP ASGI server exposing a local document search service.

Start the server:
    uvicorn doc_mcp_server:app --host 0.0.0.0 --port 3001
"""

import logging
from config import EnvironmentConfiguration
from pathlib import Path
from fastmcp import FastMCP
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from typing import Any

config = EnvironmentConfiguration()
logger = logging.getLogger("uvicorn.error")


def create_app():
    if not Path(config.CHROMA_DIR).exists():
        raise RuntimeError(
            f"Chroma database '{config.CHROMA_DIR}' does not exist.\n"
            "Run doc_ingestion_server.py first."
        )

    embeddings = OpenAIEmbeddings(
        model=config.EMBEDDING_MODEL_NAME,
        base_url=config.EMBEDDING_MODEL_BASE_URL,
        check_embedding_ctx_length=False,
    )

    vector_store = Chroma(
        collection_name=config.CHROMA_COLLECTION_NAME,
        persist_directory=config.CHROMA_DIR,
        embedding_function=embeddings,
    )

    mcp = FastMCP(config.MCP_SERVER_NAME)

    @mcp.tool
    def search_documents(query: str) -> list[dict[str, Any]]:
        """
        Search the indexed knowledge base using semantic similarity.
        ...
        """
        logger.info("Executing tool: search_documents ['%s']", query)
        try:
            retriever = vector_store.as_retriever(search_kwargs={"k": 3})
            documents = retriever.invoke(query)
            return [
                {"metadata": doc.metadata, "content": doc.page_content}
                for doc in documents
            ] if documents else []
        except Exception as ex:
            logger.error("search_documents failed: %s", ex)
            return []

    @mcp.tool
    def list_documents() -> list[str]:
        """
        Return the names of every indexed document.
        ...
        """
        logger.info("Executing tool: list_documents")
        data = vector_store.get(include=["metadatas"])
        sources = set()
        for metadata in (data.get("metadatas") or []):
            if metadata and "source" in metadata:
                sources.add(Path(metadata["source"]).name)
        return sorted(sources)

    return mcp.http_app()


app = create_app()

Breaking Down the Code

The Application Factory Pattern

def create_app():
    ...
    return mcp.http_app()

app = create_app()

We wrap the entire server setup in a create_app factory function rather than initializing everything at module level. This pattern makes it easy to test the server in isolation, swap out dependencies, or run multiple configurations side by side. The factory runs all initialization eagerly at import time, i.e. if ChromaDB is missing, the error is raised immediately during startup rather than at the first tool invocation.

Safety Guard at Startup

if not Path(config.CHROMA_DIR).exists():
    raise RuntimeError(
        f"Chroma database '{config.CHROMA_DIR}' does not exist.\n"
        "Run doc_ingestion.py first."
    )

This guard ensures the MCP server refuses to start if the vector database has not been created yet.

The search_documents Tool

@mcp.tool
def search_documents(query: str) -> list[dict[str, Any]]:
    retriever = vector_store.as_retriever(search_kwargs={"k": 3})
    documents = retriever.invoke(query)
    return [
        {"metadata": doc.metadata, "content": doc.page_content}
        for doc in documents
    ] if documents else []

The @mcp.tool decorator registers the function as a discoverable MCP tool. FastMCP automatically reads the function's docstring, type annotations, and parameter names to generate the tool's schema. Our AI agent uses this schema to understand when and how to call the tool.

Internally, as_retriever(search_kwargs={"k": 3}) converts the vector store into a LangChain retriever that returns the top 3 most semantically similar chunks for any query. The similarity is computed using cosine similarity between the query's embedding vector and all stored chunk vectors.

Each result is serialized as a plain dictionary containing two fields:

metadata: information like the source file path and page number, which the agent can use to cite its sources.
content: the actual text of the chunk, which becomes the context for the LLM's answer.

The list_documents Tool

@mcp.tool
def list_documents() -> list[str]:
    data = vector_store.get(include=["metadatas"])
    sources = set()
    for metadata in (data.get("metadatas") or []):
        if metadata and "source" in metadata:
            sources.add(Path(metadata["source"]).name)
    return sorted(sources)

This tool addresses a common user need: "What documents do you know about?" or "How many documents do I have in my knowledge base?". Rather than searching by content, it scans all stored metadata records, extracts the source field from each, and returns a deduplicated, sorted list of file names.

Using here a set for deduplication is important because every chunk from the same document carries the same source metadata. For example, A 50-page PDF might produce hundreds of chunks, but list_documents should report it as a single entry.

Running the MCP Server

uvicorn doc_mcp_server:app --host 0.0.0.0 --port 3001

The MCP server is now reachable at http://localhost:3001/mcp. Any MCP-compatible client — including our agent — can connect, discover the two tools, and start invoking them.

Testing the MCP Server

Before integrating our document MCP server with our AI agent, it's useful to verify that it is working correctly. The easiest way to do this is with the official MCP Inspector, which provides an web application for connecting to our server, discovering available tools, and invoking them manually.

We can permanently install it globally:

npm install -g @modelcontextprotocol/inspector

Or, we can run it directly without installing it:

npx @modelcontextprotocol/inspector

Once the Inspector opens in the default browser, connect it to the document MCP server. Now we can initialize a session, list the available tools as well as execute one of them (i.e. search_documents or list_documents).

Step 4: Building the RAG Chain

With both the ingestion server and the MCP tool server running, we can now build the agent that ties everything together. The agent accepts natural language questions, decides which MCP tools to call, interprets the retrieved documents, and produces grounded answers.

The full implementation is contained in agent.py, leveraging LangChain to orchestrate the workflow.

import asyncio
from config import EnvironmentConfiguration
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain_mcp_adapters.client import MultiServerMCPClient

config = EnvironmentConfiguration()


async def run_agent():
    print(f"📡 Connecting to the MCP Server [{config.MCP_SERVER_NAME}] at [{config.MCP_SERVER_URL}]...")
    client = MultiServerMCPClient(connections={
        config.MCP_SERVER_NAME: {"transport": "http", "url": config.MCP_SERVER_URL}
    })

    try:
        print("📥 Fetching and registering MCP tools...")
        discovered_tools = await client.get_tools()
        print(f"✅ Successfully registered {len(discovered_tools)} tools:")
        tool_map = {tool.name: tool for tool in discovered_tools}
        for name in tool_map.keys():
            print(f"   ➡️ [Discovered] Name: '{name}'")

        llm = init_chat_model(
            api_key=config.LLM_API_KEY
            base_url=config.LLM_BASE_URL,
            model=config.LLM_MODEL_NAME,
            model_provider=config.LLM_MODEL_PROVIDER,
            temperature=config.LLM_TEMPERATURE
        )

        system_prompt = """
        You are a document retrieval assistant.

        Whenever the user asks a question whose answer may exist in the document store:

        1. Call the document search tool.
        2. Examine the retrieved documents.
        3. Answer exclusively from the retrieved content.
        4. If no relevant documents are found, reply:
          "I couldn't find that information in your documents."

        Never skip the search step.
        Never answer using your own knowledge when the answer should come from the document store.
        Do not fabricate information.
        Keep responses concise and factual.
        """

        agent_executor = create_agent(model=llm, tools=discovered_tools, system_prompt=system_prompt)

        session_state = {"messages": []}
        print("-" * 50)

        while True:
            user_prompt = await asyncio.to_thread(input, "\n👨 User Query ('quit' to exit): ")

            if user_prompt.strip().lower() == "quit":
                print("\n👋 Terminating session. Goodbye!")
                break

            if not user_prompt.strip():
                continue

            session_state["messages"].append(("user", user_prompt))
            result = await agent_executor.ainvoke(session_state)
            session_state["messages"] = result["messages"]

            print(f"\n🤖 Agent Answer: {session_state['messages'][-1].content}")
            print("-" * 50)

    except Exception as e:
        print(f"❌ Error encountered during client execution: {e}")


if __name__ == "__main__":
    asyncio.run(run_agent())

Breaking Down the Code

Connecting to the MCP Server

client = MultiServerMCPClient(connections={
    config.MCP_SERVER_NAME: {"transport": "http", "url": config.MCP_SERVER_URL}
})
discovered_tools = await client.get_tools()

MultiServerMCPClient acts as a universal MCP client. It connects to one or more MCP servers, fetches their tool schemas, and wraps each tool as a native LangChain BaseTool object that the agent can invoke directly.

The transport: "http" setting tells the client to communicate over standard HTTP, which is the same transport our doc_mcp_server.py is getting served. This makes the integration completely network-transparent: the agent, the MCP server, and the ingestion server can all run on different machines/environments — that's the beauty of decoupled architecture and HTTP-based communication.

Configuring the Language Model

llm = init_chat_model(
    api_key=config.LLM_API_KEY
    base_url=config.LLM_BASE_URL,
    model=config.LLM_MODEL_NAME,
    model_provider=config.LLM_MODEL_PROVIDER,
    temperature=config.LLM_TEMPERATURE
)

We use LangChain's init_chat_model factory function to create the chat model. The LLM_API_KEY authenticates requests, while LLM_MODEL_PROVIDER, LLM_BASE_URL, and LLM_MODEL_NAME configure the provider, endpoint, and model to use. With LLM_TEMPERATURE=0.2, the model produces focused, deterministic responses, making it well suited for our RAG application.

The System Prompt

system_prompt = """
You are a document retrieval assistant.

Whenever the user asks a question whose answer may exist in the document store:
1. Call the document search tool.
2. Examine the retrieved documents.
3. Answer exclusively from the retrieved content.
4. If no relevant documents are found, reply:
  "I couldn't find that information in your documents."

Never skip the search step.
Never answer using your own knowledge when the answer should come from the document store.
Do not fabricate information.
"""

The system prompt is the most important part of an agent's configuration. Without explicit instructions, an LLM naturally relies on its pre-trained knowledge instead of the retrieved context. This completely defeats the purpose of RAG.

The four numbered rules enforce a strict retrieve-then-answer workflow. They require searching the document store before answering and prohibit relying on the model's internal knowledge. The fallback response prevents hallucinations when the requested information is not found.

The Agent Executor

agent_executor = create_agent(model=llm, tools=discovered_tools, system_prompt=system_prompt)

create_agent assembles a ReAct-style agent that can iteratively call tools and reason about the results before producing a final answer. The agent receives the full list of discovered MCP tools and uses the LLM to decide, at each step, whether to call a tool or respond directly.

Maintaining Conversation History

session_state = {"messages": []}

session_state["messages"].append(("user", user_prompt))
result = await agent_executor.ainvoke(session_state)
session_state["messages"] = result["messages"]

The agent maintains a running list of messages in session_state. After each turn, the full updated message history (including tool calls and their results) is stored back into session_state. This gives the agent memory across turns as it can answer follow-up questions that reference earlier exchanges without losing context.

Asynchronous Input Handling

user_prompt = await asyncio.to_thread(input, "\n👨 User Query ('quit' to exit): ")

Python's built-in input() function is blocking, i.e. it would freeze the entire async event loop while waiting for the user to type. asyncio.to_thread offloads the blocking call to a thread pool, keeping the event loop free to process ongoing async operations (like the agent's tool calls) concurrently.

Step 5: Running and Testing the Q&A Agent

Now that we walked through all three services implementation, it's time to start them up and put the system through its paces.

Starting All Services

Let's open three separate terminals and run the following commands, in this order:

Terminal 1 (Ingestion Server):

uvicorn doc_ingestion_server:app --host 0.0.0.0 --port 3002

Terminal 2 (MCP Tool Server):

uvicorn doc_mcp_server:app --host 0.0.0.0 --port 3001

Terminal 3 (Agent):

python agent.py

The order matters because the ingestion server must run first to create the ChromaDB database. The MCP tool server must start second because it reads from that database on startup. The agent can then connect to the live MCP tool server.

Uploading Sample Documents

Before querying, upload the sample files from the assets/ directory using the ingestion server's REST endpoint POST /documents:

curl -X POST http://localhost:3002/documents -F "file=@assets/meeting_notes.md"
curl -X POST http://localhost:3002/documents -F "file=@assets/car_prices.csv"
curl -X POST http://localhost:3002/documents -F "file=@assets/resume.pdf"
curl -X POST http://localhost:3002/documents -F "file=@assets/company_credentials.txt"

Each upload triggers the full pipeline: file saved → chunked → embedded → written to ChromaDB. You will see confirmation logs in Terminal 1 like this:

INFO:     New document upload: 'meeting_notes.md'.
INFO:     Loaded 1 document(s) → 5 chunk(s).
INFO:     Ingestion complete. 5 chunks indexed into ChromaDB.
INFO:     Success document upload: 'meeting_notes.md'.

Sample Questions for Testing the Knowledge Base

General questions

How many documents do I have in my knowledge base?

`assets/company_credentials.txt`:

What is my Jira login credential?
How often should I change my passwords based on my company rules?
What is the URL for my company's GitHub Enterprise instance?
What is my personal AWS Account ID?

`assets/meeting_notes.md`

What specific tasks was the QA team assigned across all the strategy meetings?
How did the engineering team's database priorities change between July and September?
What decisions were made at the Q1 strategy meeting?

`assets/resume.pdf`

Do we have any candidates whose first name is John?
What degree does John Doe hold?
Which university did John Doe attend, and were there any academic honors or scholarships?
Breakdown John Doe's experiences and see if he speaks Portuguese.

`assets/car_prices.csv`

List the cheapest car by brands.
What is the price of a red Honda with high horsepower?
What is the most expensive Chevrolet option?
How many white cars are there for each brand?

Agent Example Interaction

Once the agent is running, we can interact with the agent. A sample session looks like this:

📡 Connecting to the MCP Server [document-search] at [http://localhost:3001/mcp]...
📥 Fetching and registering MCP tools...
✅ Successfully registered 2 tools:
   ➡️ [Discovered] Name: 'search_documents'
   ➡️ [Discovered] Name: 'list_documents'

--------------------------------------------------

👨 User Query ('quit' to exit): What documents do I have indexed?

🤖 Agent Answer: You have the following documents indexed:

*   car_prices.csv
*   company_credentials.txt
*   meeting_notes.md
*   resume.pdf

--------------------------------------------------

👨 User Query ('quit' to exit): What were the key decisions made in Q2?

🤖 Agent Answer: Based on the meeting notes, here are the key decisions made in Q2:

1.  **Data Quality**: Added automated validation for ingested documents.
2.  **UX Improvements**: Improved search results by adding feedback when low-confidence results are returned.
3.  **Operations**: Deployed a new monitoring dashboard to track vector store health.

--------------------------------------------------

👨 User Query ('quit' to exit): quit

👋 Terminating session. Goodbye!

Verifying Retrieval Quality

If the agent's answers seem off or incomplete, here are the main levers to adjust:

Parameter	Location	Effect
`CHUNK_SIZE`	`.env`	Larger chunks preserve more context per result; smaller chunks improve precision
`CHUNK_OVERLAP`	`.env`	More overlap reduces information loss at boundaries
`LLM_TEMPERATURE`	`.env`	Lower values produce more conservative, factual answers
`k` in `search_kwargs`	`doc_mcp_server.py`	Retrieve more chunks for complex questions; fewer for focused ones

After changing chunk size or overlap, re-run the ingestion server to re-index all documents with the new settings.

The Full Execution Flow

Wrapping Up

In this guide, we built a complete document Q&A system powered by LangChain, RAG, ChromaDB, and the Model Context Protocol (MCP). By separating document ingestion, semantic retrieval, and conversational orchestration into independent services, we created a modular architecture that is easy to extend, maintain, and reuse.

Although the implementation runs entirely on local, open-source models through llamafile, the same application can target any OpenAI-compatible service by changing only the configuration. From here, you can continue expanding the system by adding more document loaders, improving retrieval strategies, or exposing additional capabilities as MCP tools.

BTW, this guide has an accompanying GitHub repository, making it easy to run, modify, and extend on your own.

Happy AI coding! 🚀

Building a Speech-to-Speech AI Chatbot with Python, LangChain, Whisper, and Kokoro

David Archanjo — Wed, 22 Jul 2026 03:13:41 +0000

Like many of you, I use LLMs every day for practically everything I do. Whether it's to generate content on a subject, resolve a bug in code, summarize a document, or just think through a problem out loud. However, sometimes the laziness of having to type out my question into a prompt is something that bothers me — especially when I already know exactly what I want to say.

That got me thinking: what if I could just speak to it? Not through a polished commercial product with its own constraints and opinions, but through something I built myself, using the tools I already work with. So I put together a voice chatbot in Python that listens to what I say, thinks about it using an LLM, and speaks the answer back to me — all running locally, with persistent memory across conversations.

In this article, we walk through how to build exactly that using three libraries: OpenAI Whisper for speech recognition, LangChain for conversational AI, and Kokoro for text-to-speech synthesis. The result is a fully functional, speech-to-speech AI assistant that you can run on your own machine and extend however you like.

About Me

Introduction
Prerequisites
Architecture Overview
Project Setup
Configuration
Speech Recognition with Whisper
Conversational AI with LangChain
Persistent Conversation History
Text-to-Speech with Kokoro
Put It All Together
Running the Application
Next Steps
Wrap Up

Introduction

This guide walks through building a fully functional, voice-driven AI chatbot in Python. The chatbot listens to spoken input, understands it using a Large Language Model (LLM), and responds in a natural synthesized voice. Always in a continuous, persistent conversation loop.

We combine three distinct technologies to achieve this:

Whisper: OpenAI's speech recognition model, used here via pywhispercpp, to transcribe microphone input into text.
LangChain: a framework for building LLM-powered applications, used to manage the conversational chain and persistent memory.
Kokoro: a lightweight, ONNX-based text-to-speech engine that converts the LLM's text response into spoken audio.

Motivation

Most LLM chatbot tutorials stop at text input and text output. Real conversational interfaces require two additional layers: understanding spoken language and producing spoken responses. Building these layers separately is straightforward, but connecting them into a single, cohesive, stateful pipeline is where the real work is.

This guide covers exactly that connection, i.e. from raw audio input to spoken AI response, with persistent conversation memory stored in a SQL database between sessions.

What We Will Build

By the end of this guide, we will have a working Python application which:

Continuously listens to microphone input handling silence detection.
Transcribes speech into text using Whisper.
Sends transcribed text through a LangChain conversational chain powered by an LLM.
Stores conversation history persistently in a SQLite database.
Converts the LLM's response into audio using Kokoro and plays it back through the speakers.

Prerequisites

Before we get started, make sure you have the following in place:

Python 3.10 or later, and pip installed
Basic familiarity with Python virtual environments
A working microphone and audio output device

Common Terms and Abbreviations

You will encounter the following terms and abbreviations throughout this guide:

ONNX: Open Neural Network Exchange, a standard for representing machine learning models which enables models trained in one framework to be executed in another runtime.
VAD: Voice Activity Detection, a technique used to detect speech in audio.
STT: Speech-to-Text, a process that converts audio into text.
TTS: Text-to-Speech, a process that converts text into audio.

API Key Setup

For this article, you can use a locally hosted language model exposed through an OpenAI-compatible API using, for instance llama.cpp. This setup allows every example to run entirely on our machine while keeping the code identical to what we would write when using the official OpenAI service.

# OpenAI
export OPENAI_API_KEY="sk-your-openai-api-key"

# Anthropic (optional)
export ANTHROPIC_API_KEY="sk-ant-your-anthropic-api-key"

I personally use and recommend llamafile, a standalone distribution of llama.cpp, to serve open models locally, for instance the Gemma 4:

./llamafile \
  -m gemma-4-E2B-it-Q4_K_M.gguf \
  --server \
  --host 0.0.0.0 \
  --port 8000 \
  --alias gpt-4o

--alias gpt-4o tells the server to expose the local model under the name gpt-4o. From LangChain's perspective, it behaves exactly like if it was connected to the OpenAI GPT-4o model, allowing every example in this article to work with the same init_chat_model setup, without requiring any code changes.:

from langchain.chat_models import init_chat_model

llm = init_chat_model(model="gpt-4o", model_provider="openai")

To make this local setup work transparently, you must also configure the following environment variable:

export OPENAI_BASE_URL="http://localhost:8000/v1"

NOTE: If you already have an OpenAI API key, simply remove the OPENAI_BASE_URL environment variable and replace the placeholder value of OPENAI_API_KEY with your own key. The examples throughout this article will continue to work without modification.

Architecture Overview

Before jumping into the code, we should understand how the components interact. The chatbot operates as a single continuous loop driven by audio events.

The flow is strictly linear and blocking by design. The assistant does not listen while it is speaking nor speaks while it is thinking. This design keeps the implementation simple and avoids issues like audio feedback, though it comes at the cost of a less fluid conversation than humans naturally have with each other.

Component Responsibilities

Component	Library	Responsibility
Voice Activity Detection	`pywhispercpp`	Detects speech boundaries and silence in microphone stream
Speech Recognition	`pywhispercpp` (Whisper)	Transcribes audio segments into text
Conversational Chain	`langchain`	Assembles prompt, invokes LLM, parses output
Conversation Memory	`langchain-community`	Persists chat history in SQLite database via SQLAlchemy
Language Model	`langchain-openai`	Generates natural language responses
Text-to-Speech	`kokoro-onnx`	Synthesizes text into audio samples
Audio Playback	`sounddevice`	Plays audio through system output device
Terminal UI	`rich`	Displays colored, formatted status messages

Project Setup

This project will be organized into a minimal, flat structure:

/
├── ai_chatbot.py       # Main application entry point
├── config.py           # All configuration constants and environment loading
├── .env                # Environment variables (not committed to version control)
├── requirements.txt    # Python dependencies
└── models/             # Directory for local model files (Kokoro)
    ├── kokoro-v1.0.onnx
    └── voices.bin

Application Dependencies (`requirements.txt`)

This project depends on the following Python packages:

langchain==0.3.27
langchain-openai==0.3.27
langchain-community==0.3.27

numpy>=2.0.2
SQLAlchemy>=2.0.51

pywhispercpp==1.5.0
sounddevice==0.5.5
webrtcvad==2.0.10

kokoro-onnx==0.5.0
phonemizer-fork==3.3.2
espeakng-loader==0.2.4

python-dotenv
rich

These packages provide the components required for LLM orchestration, speech recognition (Whisper.cpp), voice activity detection, speech synthesis (Kokoro ONNX), configuration management, and terminal output.

Installing Dependencies

We create a virtual environment and install the required packages from the provided requirements.txt.

python -m venv .venv
source .venv/bin/activate   # On Windows: .venv\Scripts\activate
pip install -r requirements.txt

System Dependencies

At the system level, we need to install the following tools before running the application:

On Ubuntu/Debian, use apt:

sudo apt-get install espeak-ng portaudio19-dev

On macOS, use Homebrew:

brew install espeak-ng portaudio

On Windows, use Chocolatey:

choco install espeak-ng portaudio

Checking eSpeak-NG Installation

Open a new terminal window and run:

espeak-ng "Testing eSpeak-NG is working"

We should hear the indicated phrase "Testing eSpeak-NG is working" through the default audio output (probably in a robotic voice).

Downloading Model Files

pywhispercpp downloads Whisper model weights automatically on first run. We configure which model size to use in config.py. However, Kokoro model files must be downloaded manually from the kokoro-onnx releases page and placed in the models/ directory.

Configuration

Rather than scattering constants and environment variables throughout the codebase, we centralize all configuration in a single config.py module. Every other module imports from it. This makes the application easier to adjust and keeps secrets and static configurations out of application logic.

The `.env` File

We use python-dotenv to load environment variables from the .env file.

MODEL=gpt-4o
MODEL_PROVIDER=openai
SESSION_ID=user_session
DATABASE_URL=sqlite:///chat_history.db
KOKORO_MODEL_PATH=./models/kokoro-v1.0.onnx
KOKORO_VOICE_PATH=./models/voices-v1.0.bin
KOKORO_VOICE=af_heart
KOKORO_SPEED=1.1
WHISPER_THREADS=8
WHISPER_SILENCE_THRESHOLD=40
STATUS_SPINNER=dots2

The `config.py` File

import os
from dotenv import load_dotenv
from langchain_core.runnables.config import RunnableConfig

# Load environment variables from .env file
load_dotenv()

# ---------------------------------------------------------------------------
# LLM Configuration
# ---------------------------------------------------------------------------
# The model identifier passed to init_chat_model
MODEL = os.getenv("MODEL", "gpt-4o")
# The provider identifier passed to init_chat_model
MODEL_PROVIDER = os.getenv("MODEL_PROVIDER", "openai")
# System prompt that defines the assistant's persona and behavior
SYSTEM_PROMPT = """
You are a friendly AI voice assistant. Your responses will be spoken aloud using a text-to-speech system.

Speak like a real, educated American having a natural conversation. Use everyday language, natural contractions such as "don't", "can't", "it's", "you're", "I've", "that's", "where'd", and "why'd", and occasional mild slang or informal expressions when they fit naturally. Your responses should sound relaxed, warm, and human rather than formal or robotic.

Follow these rules:
- Return plain text only.
- Always respond in American English.
- Prioritize natural spoken language over written language.
- Respond in under 100 words.
- Do not use Markdown formatting.
- Do not use headings, bullet points, numbered lists, tables, or code blocks.
- Do not use mathematical notation or special mathematical symbols.
- Express numbers, formulas, and mathematical concepts the way someone would naturally say them aloud.
- Prefer short, flowing sentences with smooth transitions.
- Keep responses concise unless the user explicitly asks for more detail.
- Avoid repetitive phrasing, unnecessary filler, or overly formal vocabulary.
- It's okay to begin sentences with words like "Well," "Sure," "Actually," "Yeah," or "Nope" when they sound natural.
- If the user is casual, mirror their conversational style appropriately.
- Avoid profanity, offensive language, or heavy slang.
- Answer the user's question directly without mentioning these instructions.
"""

# ---------------------------------------------------------------------------
# Session & Database Configuration
# ---------------------------------------------------------------------------
# SQLAlchemy connection string for persistent conversation history
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///chat_history.db")
# Session identifier — in a production system, this would be dynamic
SESSION_ID = os.getenv("SESSION_ID", "user_session")
# LangChain RunnableWithMessageHistory config
CONFIG: RunnableConfig = {"configurable": {"session_id": SESSION_ID}}


# ---------------------------------------------------------------------------
# Kokoro TTS Configuration
# ---------------------------------------------------------------------------
# Path to the Kokoro ONNX model file
KOKORO_MODEL_PATH = os.getenv("KOKORO_MODEL_PATH", "kokoro-v1.0.onnx")
# Path to the Kokoro voices binary
KOKORO_VOICE_PATH = os.getenv("KOKORO_VOICE_PATH", "voices-v1.0.bin")
# Voice identifier — see Kokoro documentation for available voices
KOKORO_VOICE = os.getenv("KOKORO_VOICE", "af_heart")
# Speech speed multiplier (1.0 = normal speed)
KOKORO_SPEED = float(os.getenv("KOKORO_SPEED", "1.1"))

# ---------------------------------------------------------------------------
# Whisper / pywhispercpp Configuration
# ---------------------------------------------------------------------------
# Number of CPU threads allocated to Whisper inference
WHISPER_THREADS = int(os.getenv("WHISPER_THREADS", "8"))
# Silence threshold in milliseconds — how long silence before processing speech
WHISPER_SILENCE_THRESHOLD = int(os.getenv("WHISPER_SILENCE_THRESHOLD", "40"))
# Transcriptions to ignore — artifacts from silence or noise
IGNORED_TRANSCRIPTIONS = { "", "[BLANK_AUDIO]", "[MUSIC PLAYING]" }

# ---------------------------------------------------------------------------
# UI Configuration
# ---------------------------------------------------------------------------
# Rich spinner style for status messages
STATUS_SPINNER = os.getenv("STATUS_SPINNER", "dots2")

Why This Structure

Every value that might change between environments, users, or deployment targets lives in config.py or .env. The main application file (ai_chatbot.py) contains no hardcoded strings or magic numbers. This separation also makes testing easier since we can override configuration without touching application logic.

The IGNORED_TRANSCRIPTIONS set handles a practical issue with Whisper: when it receives near-silence or background noise, it sometimes produces placeholder strings like [BLANK_AUDIO] rather than returning an empty string. We filter these out before passing anything to the the LLM.

Speech Recognition with Whisper

Whisper is a general-purpose speech recognition model developed by OpenAI. It is trained on a large and diverse dataset of audio and supports transcription in multiple languages. We use it here through pywhispercpp, which provides Python bindings for whisper.cpp — a C++ implementation of Whisper optimized for CPU inference.

The `Assistant` Class

pywhispercpp includes a high-level Assistant utility class that encapsulates the entire speech recognition pipeline. Rather than handling audio capture, VAD, and Whisper inference ourself, we only need to provide a callback function. Assistant automatically invokes the given callback whenever it finishes transcribing a complete utterance.

This is the key integration point between audio input and the rest of our application.

How Voice Activity Detection Works

Assistant uses webrtcvad under the hood to detect speech boundaries. It continuously reads from the microphone and buffers audio. When it detects that a person has stopped speaking (a silence longer than silence_threshold milliseconds), it passes the buffered audio to Whisper for transcription, then delivers the result to the assigned callback.

Initializing the A`sistant`

The Assistant initialization goes like this:

from pywhispercpp.examples.assistant import Assistant

# Initialize the voice assistant
assistant = Assistant(
    silence_threshold=WHISPER_SILENCE_THRESHOLD,
    commands_callback=process_input,
    n_threads=WHISPER_THREADS
)

# Start listening
console.print(
    "[bold blue]Listening...[/bold blue] "
    "[dim]Speak when you're ready.[/dim]"
)
assistant.start()

About the three constructor parameters:

Parameter	Type	Purpose
`silence_threshold`	`int` (ms)	Duration of silence required before processing the buffered audio
`commands_callback`	`callable`	Function called with the transcription string after each utterance
`n_threads`	`int`	Number of CPU threads for Whisper inference

assistant.start() blocks the main thread. The callback-driven design means our application logic lives entirely inside process_input, which Whisper calls whenever it has something to say.

Filtering Transcription Output

Whisper does not always produce clean empty strings when given silence or noise. The process_input function handles this at the top:

def process_input(text: str):
    # Strip whitespace from the input
    text = text.strip()

    # Skip empty input or predefined ignored transcriptions
    if not text or text in IGNORED_TRANSCRIPTIONS:
        return

We strip whitespace first, then check against our IGNORED_TRANSCRIPTIONS set. If the transcription matches any of those values, we return early without invoking the LLM. This prevents unnecessary API calls and avoids the assistant responding to silence with spoken output.

Choosing a Whisper Model Size

pywhispercpp downloads Whisper model weights automatically on first use. The model size is a balance between transcription quality and inference speed. On a modern CPU, tiny and base are suitable for real-time use. Larger models produce better transcriptions but introduce noticeable latency and requires more CPU resources.

Model	Parameters	Relative Speed	Relative Accuracy
`tiny`	39M	Fastest	Lower
`base`	74M	Fast	Moderate
`small`	244M	Moderate	Good
`medium`	769M	Slow	High
`large`	1550M	Slowest	Highest

For a conversational assistant running on CPU, base is a reasonable starting point. The default model used by pywhispercpp's Assistant class can be configured via its constructor.

Conversational AI with LangChain

TL;DR If you are new to LangChain and really want to learn it from a pratical guide, check out my LangChain Explained: A Practical Guide For Beginners.

LangChain organizes LLM interactions as composable pipelines called chains. A chain is a sequence of components where the output of one becomes the input of the next. In our case, the chain consists of three components:

A prompt template which formats the conversation history and user input.
A language model which generates a response based on the transcribed audio.
An output parser that extracts the plain text from the model's response. LangChain uses the pipe (ǀ) operator to compose these components, following the same convention as Unix pipes.

Initializing the Language Model

from langchain.chat_models import init_chat_model

# Initialize the language model
llm = init_chat_model(model=MODEL, model_provider=MODEL_PROVIDER)

init_chat_model is a provider-agnostic factory function. It accepts a model name and a provider string, then returns the appropriate BaseChatModel subclass. With MODEL = "gpt-4o" and MODEL_PROVIDER = "openai", it instantiates a ChatOpenAI internally. Switching providers requires only changing those two constants in config.py.

Defining the Prompt

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# Define the prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", SYSTEM_PROMPT),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{user_input}"),
])

ChatPromptTemplate defines the structure of the message list sent to the LLM. We use three message slots:

("system", SYSTEM_PROMPT): a fixed system message that establishes the assistant's persona and constraints. This is where we instruct the model to respond in plain spoken English without markdown formatting, since responses will be read aloud by Kokoro.
MessagesPlaceholder(variable_name="chat_history"): a dynamic slot where the conversation history will be injected. The variable_name must match the history_messages_key we configure later.
("human", "{user_input}"): the current user message. The {user_input} placeholder must match the input_messages_key we configure later.

Assembling the Chain

from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import Runnable

# Initialize the output parser
parser = StrOutputParser()

# Create the LangChain chain
chain: Runnable = prompt | llm | parser

The StrOutputParser extracts the content field from the AIMessage returned by the LLM, giving us a plain Python string. Without it, chain.invoke() would return an AIMessage object rather than a string, and we would need to access .content manually everywhere we use the response.

The type annotation chain: Runnable is not strictly required but documents the expected interface. Every component in a LangChain expression, e.g. prompts, models, parsers, implements the Runnable interface, which defines .invoke, .stream, .batch, and their async equivalents.

Invoking the Chain

response = chain.invoke(
    {"user_input": text},
    config=CONFIG
)

Inside process_input, we invoke the chain with the current user input. The config dictionary carries session metadata (we will cover its role in the next section). At this point, chain has been wrapped with history management, so .invoke will automatically load conversation history before calling the LLM and save the new exchange afterward.

Persistent Conversation History

LLMs are stateless, i.e. each call to the model is independent, as the model has no memory of previous turns unless we explicitly include them in the prompt. For a chatbot, this means we must load prior messages and append them to every request.

LangChain addresses this through RunnableWithMessageHistory, a wrapper that automatically retrieves conversation history before each invocation and saves new messages afterward. The underlying storage mechanism is configurable.

SQLChatMessageHistory

We use SQLChatMessageHistory from langchain-community package, which persists conversation history in a SQL database via SQLAlchemy.

On every invocation, it:

Reads all previous messages for the given session_id from the database.
Injects them into the chat_history placeholder in the prompt.
After the chain completes, writes the new human message and AI response back to the database.

from langchain_community.chat_message_histories import SQLChatMessageHistory

# Define the function to retrieve the persistent session history
def get_session_history(session_id: str):
    return SQLChatMessageHistory(session_id=session_id, connection=DATABASE_URL)

get_session_history is a factory function responsible for returning the chat history associated with a given session. On every invocation, RunnableWithMessageHistory calls this function, passing the session ID extracted from the config dictionary. For example, if the history is stored in a SQLite database configured as DATABASE_URL = "sqlite:///chat_history.db".

Wrapping the Chain

from langchain_core.runnables.history import RunnableWithMessageHistory

# Add persistent conversation history
chain = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="user_input",
    history_messages_key="chat_history"
)

The four arguments:

Argument	Value	Purpose
`runnable`	`chain`	The base chain to wrap
`get_session_history`	Our factory function	Called to retrieve or create a history store
`input_messages_key`	`"user_input"`	Must match the input key in `chain.invoke()`
`history_messages_key`	`"chat_history"`	Must match the `MessagesPlaceholder` variable name in the prompt

The key names must be consistent across three places: the prompt template, the RunnableWithMessageHistory constructor, and the chain.invoke() call. A mismatch in any of these will cause a runtime error.

Session Management

The config dictionary passed to chain.invoke() carries the session identifier:

# config.py
SESSION_ID = "default-session"
CONFIG = {"configurable": {"session_id": SESSION_ID}}

In this application, we use a single fixed session for learning purpose, but in a production context with multiple users, each user or conversation would have its own session_id, keeping conversation histories isolated. The session_id can be any string, for instance a UUID, a username, or any identifier that uniquely scopes a conversation.

Execution Flow with History

Text-to-Speech with Kokoro

Kokoro is a text-to-speech model distributed in ONNX format. It runs entirely on CPU without requiring a GPU or an external API call. The kokoro-onnx Python package loads the ONNX model and provides a simple interface to synthesize speech from text.

Kokoro's ONNX format enables efficient CPU inference using ONNX Runtime, without requiring PyTorch or TensorFlow. This eliminates the need to install heavyweight deep learning frameworks while still providing high-performance inference.

Initializing Kokoro

from kokoro_onnx import Kokoro

# Initialize Kokoro
kokoro = Kokoro(model_path=KOKORO_MODEL_PATH, voices_path=KOKORO_VOICE_PATH)

We initialize Kokoro once at startup and reuse the instance for every speak() call. Loading the ONNX model has a non-trivial startup cost, so we avoid re-initializing it per request.

The voices_path points to a binary file that contains pre-computed speaker embeddings. Kokoro uses these embeddings to control the vocal characteristics of the synthesized speech. Each named voice (like "af_heart") is an identifier into this file.

Generating and Playing Speech

import sounddevice as sd

# Generate and play speech
def speak(text: str):
    # 1. Generate speech samples from text
    samples, sample_rate = kokoro.create(text, voice=KOKORO_VOICE, speed=KOKORO_SPEED)

    # 2. Ensure array shape and dtype are correct for sounddevice
    wav = samples.squeeze().astype("float32")

    # 3. Play audio synchronously
    sd.play(wav, samplerate=sample_rate, blocking=True)

kokoro.create() returns a NumPy array of audio samples and the corresponding sample rate (typically 24000 Hz). The samples represent normalized floating-point audio in the range [-1.0, 1.0].

We call .squeeze() to remove any extra dimensions that Kokoro may add to the array shape, and .astype("float32") to ensure the dtype matches what sounddevice expects.

sd.play() with blocking=True means the function does not return until the audio has finished playing. This is intentional as we do not want the assistant to start listening again before it has finished speaking, right?.

Voice Options

Kokoro ships with several pre-trained voices. The voice is controlled by the voice parameter in kokoro.create(). The KOKORO_VOICE constant in config.py can be changed to any voice included in the voices.bin file. Check out the Kokoro's HuggingFace page for the full list of available voice identifiers.

Suppressing Phonemizer Logs

Kokoro internally uses phonemizer to convert text into phoneme sequences before synthesis. phonemizer is verbose by default. We suppress its logs in the application:

import logging
logging.getLogger("phonemizer").setLevel(logging.ERROR)

This line only affects the phonemizer logger, leaving all other loggers at their default level.

Put It All Together

Now that we've covered each component individually, it's time to walk through the complete chatbot implementation. Each section of the file initializes a component or defines a callback, while the assistant.start() call at the end brings everything together.

# ai_chatbot.py

from config import *
import logging
import sounddevice as sd
from kokoro_onnx import Kokoro
from langchain.chat_models import init_chat_model
from langchain_community.chat_message_histories import SQLChatMessageHistory
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import Runnable
from langchain_core.runnables.history import RunnableWithMessageHistory
from pywhispercpp.examples.assistant import Assistant
from rich.console import Console

# Configure phonemizer's logging to suppress verbose output
logging.getLogger("phonemizer").setLevel(logging.ERROR)

# Initialize Rich console for formatted terminal output
console = Console()

# Defines a function to get a persistent conversation history
def get_session_history(session_id: str):
    return SQLChatMessageHistory(session_id=session_id, connection=DATABASE_URL)

# Initialize the LLM
llm = init_chat_model(model=MODEL, model_provider=MODEL_PROVIDER)

# Defines a parser to extract the LLM response
parser = StrOutputParser()

# Defines a prompt template for the LLM
prompt = ChatPromptTemplate.from_messages([
    ("system", SYSTEM_PROMPT),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{user_input}"),
])

# Defines a runnable chain for the LLM
chain: Runnable = prompt | llm | parser

# Defines a runnable chain with a persistent conversation history
chain = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="user_input",
    history_messages_key="chat_history"
)

# Initialize Kokoro once at startup
kokoro = Kokoro(model_path=KOKORO_MODEL_PATH, voices_path=KOKORO_VOICE_PATH)

# Defines a function to speak text
def speak(text: str) -> None:
    """Convert text to speech and play it through the audio output device."""
    # Synthesize speech samples from text
    samples, sample_rate = kokoro.create(text, voice=KOKORO_VOICE, speed=KOKORO_SPEED)

    # Normalize array shape and dtype for sounddevice
    wav = samples.squeeze().astype("float32")

    # Play audio synchronously — blocks until playback is complete
    sd.play(wav, samplerate=sample_rate, blocking=True)

# Defines a callback for the Whisper Assistant
def process_input(text: str) -> None:
    """
    Callback invoked by the Whisper Assistant after each transcribed utterance.
    Filters noise, invokes the LangChain chain, and speaks the response.
    """
    # Remove leading/trailing whitespace
    text = text.strip()

    # Discard empty strings and known Whisper noise artifacts
    if not text or text in IGNORED_TRANSCRIPTIONS:
        return

    # Display the recognized user input
    console.print(f"\n[bold cyan]You:[/bold cyan] {text}")

    # Invoke the LLM chain with a loading indicator
    with console.status("[bold yellow]Thinking...[/bold yellow]", spinner=STATUS_SPINNER):
        try:
            response = chain.invoke(
                {"user_input": text},
                config=CONFIG
            )
        except Exception as e:
            console.print(f"[red]Error:[/red] {e}")
            speak("Sorry, something went wrong.")
            return

    # Display the assistant's text response
    console.print(f"[bold green]Assistant:[/bold green] {response}")

    # Synthesize and play the response with a status indicator
    with console.status("[bold magenta]Speaking...[/bold magenta]", spinner=STATUS_SPINNER):
        try:
            speak(response)
        except Exception as e:
            console.print(f"[red]Speech error:[/red] {e}")

    # Signal that the assistant is ready for the next input
    console.print(
        "\n[bold blue]Listening...[/bold blue] "
        "[dim]Speak when you're ready.[/dim]"
    )

# Initializes the Whisper-backed voice assistant
assistant = Assistant(
    silence_threshold=WHISPER_SILENCE_THRESHOLD,
    commands_callback=process_input,
    n_threads=WHISPER_THREADS
)

# Prints the initial listening prompt and begins the audio capture loop
console.print(
    "[bold blue]Listening...[/bold blue] "
    "[dim]Speak when you're ready.[/dim]"
)
assistant.start()

Breaking Down the Code

We imported the required libraries, including LangChain, Whisper, Kokoro, Rich, and our application configuration.
We configured logging to suppress unnecessary messages from the phonemizer library.
We created a Rich console to display colored messages and loading animations.
We defined get_session_history(), which tells LangChain how to load and store conversation history in a SQLite database.
We initialized the language model using init_chat_model().
We created a StrOutputParser to extract the model's response as plain text.
We built a prompt template containing the system prompt, the conversation history, and the user's latest message.
We assembled a LangChain pipeline that sends the prompt to the language model and parses the generated response.
We wrapped the pipeline with RunnableWithMessageHistory so previous conversations are automatically loaded and persisted.
We initialized the Kokoro text-to-speech engine once at startup.
We implemented speak(), which converts text into speech and plays it through the speakers.
We implemented process_input(), the callback responsible for processing each Whisper transcription, generating a response, displaying it, and speaking it aloud.
We initialized the Whisper Assistant, configuring it to invoke process_input() whenever speech is recognized.
Finally, we displayed a listening message and started the assistant, allowing it to continuously listen for the user's voice.

The Execution Flow

Running the Application

With the virtual environment activated and the .env file populated:

python ai_chatbot.py

On the first run, pywhispercpp downloads the configured Whisper model. This is a one-time operation. Kokoro loads the ONNX model from the local models/ directory.

Expected Terminal Output

Listening... Speak when you're ready.

You: What is the capital of France?
Assistant: The capital of France is Paris.

Listening... Speak when you're ready.

You: And what is it known for?
Assistant: Paris is known for the Eiffel Tower, world-class cuisine, fashion, art museums like the Louvre, and its rich history as a cultural capital of Europe.

Listening... Speak when you're ready.

All responses are printed to the terminal, and their corresponding audio is played through the system's default audio output device.

Verifying Conversation Persistence

After interacting with the chatbot through several conversations, you can inspect the SQLite database to verify that the conversation history has been persisted correctly. The database should look like to the following:

Stopping the application with Ctrl+C and restarting it will resume the conversation from the stored history. The LLM will have access to all prior messages from the same session_id.

Next Steps

This implementation intentionally focuses on the core concepts. As a result, it has a few limitations, such as supporting voice input only (no text input), processing requests sequentially (listen → think → speak), and using a single conversation session.

Possible improvements include:

Stream the LLM response as it's generated
Stream speech synthesis for lower latency
Add tool calling
Support multiple conversation sessions
Add a wake word
Replace SQLite with PostgreSQL

Wrap Up

We have built a complete, voice-driven AI chatbot from raw audio input to spoken response. The system is composed of four layers: voice capture and transcription, conversational reasoning, persistent memory, and speech synthesis. Each component is implemented by a specific library and connected through a clean, callback-driven architecture.

This guide has an accompanying GitHub repository, making it easy to run, modify, and extend the chatbot on your own.

Happy AI coding 🚀

LangGraph Explained: A Practical Guide For Beginners

David Archanjo — Sun, 19 Jul 2026 08:56:29 +0000

In this article, I will explain what LangGraph is, why it exists, and how it enables the development of stateful, multi-step AI applications powered by Large Language Models (LLMs). Instead of presenting isolated concepts, we will build our understanding step by step — from a simple single-node graph to a complete, production-ready AI workflow with state management, conditional routing, memory, human oversight, and multi-agent collaboration.

By the end of this article, you will understand how all these building blocks fit together and how LangGraph orchestrates them into a cohesive, controllable, and scalable AI system.

What We Will Build

By the end of this guide, we will have built a complete AI research and reporting workflow that:

Researches a topic using multiple specialized agents
Loops back for revisions until quality is approved
Runs independent tasks in parallel
Persists state across sessions
Pauses for human review before finalizing

About Me

Prerequisites
What Is LangGraph?
Why Learn and Use LangGraph?
LangChain vs LangGraph
The Core Concept
Hello, World! — Our First Graph
State Management
Nodes
Edges
Conditional Routing
Cycles and Loops
Parallel Execution
Persistence and Checkpointing
Human-in-the-Loop
Subgraphs
Building a Complete AI Workflow
Multi-Agent Systems
Production Features
Next Steps
Wrap Up

Prerequisites

To get the most out of this guide, you should have:

Python 3.10 or later and pip installed.
Basic knowledge of Python.
A text editor or IDE such as Visual Studio Code or PyCharm.
LangChain Basics: LangGraph is built on top of LangChain. You should understand what is an LLM, how to invoke one, what a Prompt Template is, and how chains work. If you are new to LangChain, you can check my LangChain Explained: A Practical Guide For Beginners.
Basic knowledge on flowcharts and how to visualize them using tools like Mermaid.

Python Core Dependencies

pip install langgraph langchain-openai

API Key Setup

# OpenAI
export OPENAI_API_KEY="sk-your-openai-api-key"

# Anthropic (optional)
export ANTHROPIC_API_KEY="sk-ant-your-anthropic-api-key"

For this article, you can use a locally hosted language model exposed through an OpenAI-compatible API using, for instance using llama.cpp. This setup allows every example to run entirely on our machine while keeping the code identical to what we would write when using the official OpenAI service.

I personally use llamafile, a standalone distribution of llama.cpp, to serve open models locally, like the Gemma 4:

./llamafile \
  -m gemma-4-E2B-it-Q4_K_M.gguf \
  --server \
  --host 0.0.0.0 \
  --port 8000 \
  --alias gpt-4o

To make this local setup work transparently, you must also configure the following environment variable:

export OPENAI_BASE_URL="http://localhost:8000/v1"

If you already have an OpenAI API key, simply remove the OPENAI_BASE_URL environment variable and replace the placeholder value of OPENAI_API_KEY with your own key. The examples throughout this article will continue to work without modification.

What Is LangGraph?

LangGraph is a framework for building stateful, multi-step AI applications where the flow of logic is modeled as a graph.

Think of it as a flowchart for our AI application.

Each box in the flowchart is a Node (a step that does something), and each arrow is an Edge (a connection that determines what happens next).

On its own, a standard LangChain chain:

Always executes steps in a fixed, linear order
Cannot loop back and retry a step
Cannot make dynamic decisions about which path to take next
Does not natively support pausing for human input in the middle of execution

LangGraph gives our AI application the ability to branch, loop, pause, and coordinate multiple agents through a controllable graph structure.

Why Learn and Use LangGraph?

To understand why we would consider LangGraph, we first need to understand what LangChain already does well, and where its linear model reaches its limits.

LangChain is excellent at building linear pipelines. For instance, a prompt flows into an LLM, the output flows into a parser, and the result is returned. With LangChain Expression Language (LCEL), a chain also supports parallel branches via RunnableParallel, memory via RunnableWithMessageHistory, and tool-calling via AgentExecutor. Frankly speaking, for a large number of real-world applications, this is everything we need.

However, as applications grow in complexity, certain patterns become difficult or impossible to express with a linear chain:

1. Cycles and Self-Correction

LangChain chains are directed and non-cyclic by design, i.e. data flows forward and never loops back. But many advanced AI workflows require iterationm, for instance:

Generate a draft → evaluate it → if it fails quality checks → revise and try again
Attempt a task → detect an error → retry with a corrected approach

In other words:

With LangChain: A chain does not have a native way to loop back to a previous step. We would have to wrap the entire chain in a manual while loop in our Python code, managing state and exit conditions by ourself.
With LangGraph: Cycles are a native feature of the graph. A node can route back to a previous node, creating a clean, controlled loop with a well-defined exit condition, having all the details expressed within the graph itself.

2. Complex Stateful Workflows

LangChain's memory tools work well for conversational history, i.e. tracking what was said in a chat. But for complex workflows where many different pieces of data need to be created, transformed, and passed across many steps, managing state through LangChain's memory primitives becomes awkward, to say the least.

With LangChain: For a chain, state beyond chat history must be managed externally, i.e. passed manually between steps or stored in application-level variables.
With LangGraph: State is a first-class citizen. We define a typed State object that is automatically shared across every node in the graph, always reflecting the latest values without any manual wiring.

3. Dynamic Conditional Routing

While LangChain's RunnableParallel handles static parallel branches, expressing dynamic routing, where the next step depends on the current result, requires conditional logic that does not fit naturally into a chain.

With LangChain: We must implement branching logic outside the chain, in our application code, using if/else statements that inspect the chain's output and decide what to do/call next.
With LangGraph: Conditional routing is a native graph concept. A routing function inspects the current state and returns the name of the next node to execute, keeping all the flow logic inside the graph where it is visible and testable.

4. Human-in-the-Loop Control

Real-world AI applications often need a human to review, approve, or correct AI decisions before the workflow continues.

With LangChain: There is no built-in mechanism to pause mid-execution and wait for human input. The implementation of such feature would require building custom infrastructure around the chain, e.g. queues, state serialization, resume logic, etc.
With LangGraph: Human-in-the-Loop is a built-in feature. The graph can be configured to pause at any node, wait for human input, and resume exactly where it left off, with the full state preserved automatically.

5. Multi-Agent Orchestration at Scale

LangChain's AgentExecutor is powerful for single-agent scenarios where one LLM decides which tools to be invoked. But coordinating multiple specialized agents, having each with their own role, tools, and sub-workflows that communicate with each other is beyond what a single AgentExecutor is designed for.

With LangChain: Each agent is a standalone executor. Coordinating multiple agents requires a hardcoded orchestration that manages inter-agent communication, shared context, and execution order.
With LangGraph: Each agent is a node in the graph. LangGraph handles routing between agents, maintains the shared state that all agents read from and write to, and provides a clear, visual structure for the entire collaboration ecosystem.

LangChain vs LangGraph

One of the most common questions when comes to LangGraph is: "I already use LangChain. Why would I need LangGraph?".

The answer depends entirely on the complexity of our workflow.

LangGraph and LangChain are not competitors; they complement each other. LangGraph builds on top of LangChain by adding an orchestration layer for stateful and complex workflows. We can still use LangChain components such as Prompt Templates, Output Parsers, Tools, and Retrievers inside our LangGraph nodes, while LangGraph controls how and when those components are executed.

Think in this way:

LangChain gives us the individual instruments.
LangGraph is the maestro that tells each instrument when to play.

Side-by-Side Comparison

Feature	LangChain	LangGraph
Execution Flow	Linear, sequential	Graph-based, flexible
Branching Logic	Via application code `if/else`	Native conditional edges
Loops / Cycles	Not supported natively	Native feature
State Management	Chat history via `RunnableWithMessageHistory`	Typed `State` object shared across all nodes
Parallel Execution	Via `RunnableParallel` and `.batch()`	Via `Send` API for dynamic fan-out
Memory / Persistence	`InMemoryChatMessageHistory`, `RedisChatMessageHistory`, etc.	Built-in checkpointers (Memory, SQLite, PostgreSQL)
Human-in-the-Loop	Not supported natively	Built-in interrupt and resume support
Tool Calling / Agents	`AgentExecutor` for single-agent scenarios	Multi-agent coordination as graph nodes
Best For	Linear pipelines, RAG, single agents	Complex stateful workflows, cycles, multi-agent systems

When to Use Which

1. Question: Is our workflow linear with no branching or loops?

Answer: LangChain is sufficient

2. Question: Does our app need to loop back and self-correct?

Answer: LangGraph (native cycles)

3. Question: Does our workflow have complex shared state beyond chat history?

Answer: LangGraph (typed State object)

4. Question: Do we need dynamic routing based on runtime results?

Answer: LangGraph (conditional edges)

5. Question: Do we need a human to approve AI decisions mid-workflow?

Answer: LangGraph (Human-in-the-Loop)

6. Question: Do we need multiple specialized agents collaborating?

Answer: LangGraph (Multi-Agent graph)

Key Takeaway: LangChain and LangGraph are designed to work together. We start with LangChain for simpler use cases and introduce LangGraph when our workflow requires cycles, complex state management, dynamic routing, human oversight, or multi-agent coordination.

The Core Concept

At its core, LangGraph is built around one simple idea: model AI application as a graph.

A graph is a structure made of two things:

Nodes: They are the steps — each node is a Python function that performs one unit of work, for instance: calling an LLM, running a tool, making a decision, or transforming data.
Edges: They are the connections between nodes — each edge defines what happens after a node finishes, i.e. which node to execute next. Together, nodes and edges define the flow of logic through our application.

The Three Core Building Blocks

Building Block	What It Does
State	A shared data object that carries information through the entire graph. Every node can read from and write to it.
Nodes	Python functions that perform work. They receive the current state as input and return updates to the state as output.
Edges	As connections between nodes, they can be fixed (always go to node B after node A) or conditional (go to B or C depending on the state).

Types of Edges

Edge	Type Description
Normal Edge	Always routes to the same next node.
Conditional Edge	Calls a routing function that inspects the state and returns the name of the next node to execute.
Entry Point	Defines which node the graph starts from.
`END`	A special destination that tells the graph the workflow is complete.

Types of Nodes

Nodes can perform virtually any operation required by our workflow. Some common types include:

LLM Nodes: Call a language model to generate, analyze, classify, or transform content.
Tool Nodes: Execute external tools, APIs, database queries, or other services.
Decision Nodes: Evaluate the current state and produce information used to determine what should happen next.
Pre-built Nodes: Ready-to-use nodes provided by LangGraph, such as ToolNode, which automatically executes tools requested by an LLM.

Regardless of what a node does internally, the basic pattern remains the same: it receives the current state, performs a unit of work, and returns the fields that should be updated.

How It All Flows

A basic LangGraph execution workflow looks like this:

Unlike a LangChain chain which is linear, a LangGraph graph can loop back:

This ability to loop, branch, and converge is what makes LangGraph fundamentally different and complementary from a standard LangChain chain.

Hello, World! — Our First Graph

It's time for a Hello, World!. Let's build the simplest possible LangGraph application: a graph with a single node which calls an LLM and returns a response.

from typing import TypedDict
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, END

# Step 1: Define the State
class State(TypedDict):
    question: str
    answer: str

# Step 2: Initialize the LLM
llm = init_chat_model(model="gpt-4o", model_provider="openai")

# Step 3: Define a Node
def answer_question(state: State) -> dict:
    response = llm.invoke(state["question"])
    return {"answer": response.content}

# Step 4: Build the Graph
graph_builder = StateGraph(State)

graph_builder.add_node("answer_question", answer_question)
graph_builder.add_edge(START, "answer_question")
graph_builder.add_edge("answer_question", END)

graph = graph_builder.compile()

# Step 5: Run the Graph
result = graph.invoke({"question": "What is LangGraph in one sentence?"})
print(result["answer"])

Output:

LangGraph is a library that allows developers to build complex, stateful, and cyclical LLM agent workflows by modeling them as directed graphs.

Breaking Down The Code

We defined a State using TypedDict with two fields: question and answer.
We initialized an LLM using init_chat_model.
We created a Node as a plain Python function that reads the question from the state, calls the LLM, and writes the generated answer back to the state.
We built a Graph using StateGraph, connected START to our node, and connected the node to END.
We compiled the graph into a runnable workflow and invoked it with an initial state.

This is the foundation, and everything else in LangGraph builds on top of this pattern.

The Execution Flow

State Management

State is the most important concept in LangGraph. It is the shared memory of our graph, i.e. the single source of truth that every node reads from and writes to as the workflow progresses.

What is a State?

State is a Python TypedDict that we define. Think of it as a baton in a relay race, in which each runner (node) picks it up, does their part, and passes it on.

from typing import TypedDict, List, Optional

class ResearchState(TypedDict):
    topic: str                 # Set at the start
    search_results: List[str]  # Populated by the researcher node
    draft: str                 # Written by the writer node
    feedback: Optional[str]    # Provided by the reviewer node
    final_report: str          # Finalized output

How Nodes Update State

A node receives the full current state and returns only the field(s) it changed. LangGraph merges those updates automatically:

def fetch_search_results(state: ResearchState) -> dict:
    topic = state["topic"]
    results = [
        f"Article 1 about {topic}",
        f"Article 2 about {topic}",
        f"Article 3 about {topic}"
    ]
    # Only return the fields being updated
    return {"search_results": results}

Using Reducers for Accumulating Data

By default, LangGraph replaces a state field with the new value. When we need to append instead, for instance to accumulate chat messages, we should use a reducer:

from typing import Annotated
from langgraph.graph.message import add_messages

class ChatState(TypedDict):
    # New messages are appended, not replaced
    messages: Annotated[list, add_messages]

State Evolution Through the Graph

Nodes

As stated previously, a Node is a trivial Python function which receives the current state and returns a dictionary of updates. It can be considered as the unit of work in a LangGraph graph.

Anatomy of a Node

def my_node(state: State) -> dict:
    # 1. Read from state
    input_data = state["some_field"]

    # 2. Do some work
    result = do_something_with_input_data(input_data)

    # 3. Return only the fields being updated
    return {"output_field": result}

Categories of Nodes

Although LangGraph nodes are generally Python functions or runnables, we can categorize them based on the role they perform within a graph. Some common categories include:

LLM Node: calls a language model

An LLM Node calls a language model to perform tasks that require natural language understanding or generation. It can be used to answer questions, generate content, summarize information, classify text, or make decisions based on the current graph state.

from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate

llm = init_chat_model(model="gpt-4o", model_provider="openai")

def generate_summary(state: State) -> dict:
    prompt = ChatPromptTemplate.from_template("Summarize the following text in 3 bullet points:\n\n{text}")
    chain = prompt | llm
    response = chain.invoke({"text": state["raw_text"]})
    return {"summary": response.content}

Tool Node: calls an external API or service

A Tool Node executes an external operation that extends the capabilities of the workflow beyond the language model itself. It can call an API, query a database, search the web, access a file system, or execute any other function exposed as a tool.

import requests

def fetch_weather(state: State) -> dict:
    city = state["city"]
    response = requests.get(f"https://api.weather.com/current?city={city}")
    return {"weather_data": response.json()}

Decision Node: performs logic without calling an LLM

A Decision Node performs deterministic logic without calling a language model. It typically inspects the current state and returns an update that can be used by a conditional edge to determine which execution path the graph should follow.

def check_quality(state: State) -> dict:
    word_count = len(state["draft"].split())
    is_sufficient = word_count >= 200
    return {"quality_passed": is_sufficient}

Pre-built ToolNode: LangGraph's built-in node for executing LangChain tools automatically

LangGraph provides the pre-built ToolNode for executing LangChain tools requested by a language model. Instead of manually creating a node that identifies and executes each tool call, ToolNode handles the tool execution automatically and returns the results to the graph.

from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool

@tool
def search_web(query: str) -> str:
    """Search the web for information about a query."""
    return f"Search results for: {query}"

@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression."""
    return str(eval(expression))

# ToolNode automatically executes whichever tool the LLM requests
tools = [search_web, calculate]
tool_node = ToolNode(tools)

graph_builder.add_node("tools", tool_node)

Edges

Edges are the connections between Nodes. They define the flow of execution through the graph.

Normal Edges

A fixed, unconditional connection. After node A finishes, always go to node B:

from typing import TypedDict
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, END, START

class State(TypedDict):
    topic: str
    title: str
    body: str
    conclusion: str

llm = init_chat_model(model="gpt-4o", model_provider="openai")

def generate_title(state: State) -> dict:
    response = llm.invoke(f"Generate a blog title about: {state['topic']}")
    return {"title": response.content}

def write_body(state: State) -> dict:
    response = llm.invoke(f"Write a blog post titled: {state['title under 30 words.']}")
    return {"body": response.content}

def write_conclusion(state: State) -> dict:
    response = llm.invoke(f"Write a conclusion for:\n{state['body']} under 20 words.")
    return {"conclusion": response.content}

graph_builder = StateGraph(State)

graph_builder.add_node("generate_title", generate_title)
graph_builder.add_node("write_body", write_body)
graph_builder.add_node("write_conclusion", write_conclusion)

# Normal edges — always go from A to B
graph_builder.add_edge(START, "generate_title")
graph_builder.add_edge("generate_title", "write_body")
graph_builder.add_edge("write_body", "write_conclusion")
graph_builder.add_edge("write_conclusion", END)

graph = graph_builder.compile()
result = graph.invoke({"topic": "LangGraph"})
print(result["title"])
print('\n---\n')
print(result["body"])
print('\n---\n')
print(result["conclusion"])

Breaking Down The Code

State: The State schema defines the data shared across the workflow: the blog topic, generated title, body, and conclusion.
Nodes: Each node performs a specific task and updates a different field in the shared state.
Normal Edges: The nodes are connected using add_edge(), creating a fixed execution sequence where each node always runs after the previous one.
Execution Flow: The graph starts at START, executes generate_title → write_body → write_conclusion, and finally reaches END.

The Execution Flow

Conditional Routing

Normal edges are great for linear workflows, but real-world AI applications need to make decisions: "Is this output good enough? Which path should we take?"

That's where Conditional Edges come in.

What is a Conditional Edge?

A conditional edge connects one node to multiple possible next nodes via a routing function. Its a plain Python function that inspects the current state and returns the name of the next node:

def routing_function(state: State) -> str:
    return "node_b" if state["some_condition"] else "node_c"

graph_builder.add_conditional_edges(
    "node_a",          # From this node
    routing_function,  # Call this function
    {
        "node_b": "node_b",   # If it returns "node_b", go here
        "node_c": "node_c"    # If it returns "node_c", go here
    }
)

Example: Intent-Based Routing

In the following example, we build a graph that classifies a user's message and dynamically routes the execution to the appropriate response node based on the detected intent:

from typing import TypedDict
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START, END

llm = init_chat_model(model="gpt-4o", model_provider="openai")

class State(TypedDict):
    user_input: str
    intent: str
    response: str

def classify_intent(state: State) -> dict:
    prompt = (
        "Classify the user's primary intent as exactly one word: "
        "question, complaint, or compliment. "
        "A message expressing dissatisfaction is a complaint, even if phrased as a question.\n\n"
        f"Message: {state['user_input']}"
    )
    intent = llm.invoke(prompt).content.strip().lower()
    return {"intent": intent}

def handle_question(state: State) -> dict:
    response = llm.invoke(f"Answer helpfully: {state['user_input']}")
    return {"response": response.content}

def handle_complaint(state: State) -> dict:
    response = llm.invoke(f"Respond empathetically: {state['user_input']}")
    return {"response": response.content}

def handle_compliment(state: State) -> dict:
    response = llm.invoke(f"Respond warmly: {state['user_input']}")
    return {"response": response.content}

def route_by_intent(state: State) -> str:
    return state["intent"]

builder = StateGraph(State)

builder.add_node("classify", classify_intent)
builder.add_node("question", handle_question)
builder.add_node("complaint", handle_complaint)
builder.add_node("compliment", handle_compliment)

builder.add_edge(START, "classify")

builder.add_conditional_edges(
    "classify",
    route_by_intent,
    {
        "question": "question",
        "complaint": "complaint",
        "compliment": "compliment",
    },
)

builder.add_edge("question", END)
builder.add_edge("complaint", END)
builder.add_edge("compliment", END)

graph = builder.compile()

result = graph.invoke({
    "user_input": "Why is my order still not delivered?",
    "intent": "",
    "response": "",
})

print(f"Intent: {result['intent']}")
print(f"Response: {result['response']}")

Breaking Down The Code

State: The State schema stores the user's input, the classified intent, and the generated response.
Classification: The classify_intent node uses the LLM to classify the user's input as a question, complaint, or compliment.
Routing Function: The route_by_intent function reads the classified intent from the state and returns the route that should be followed.
Conditional Edges: add_conditional_edges() maps each possible route to its corresponding node: question, complaint, or compliment.
Execution Flow: After the selected node generates the appropriate response, execution follows a normal edge to END.

The Execution Flow

Cycles and Loops

One of LangGraph's most powerful capabilities, and one that is not natively possible in a linear LangChain chain, is the ability to create cycles: graphs where execution can loop back to the previous node.

Why Cycles Matter

Cycles enable self-correcting AI workflows:

Generate an output → evaluate it → if not good enough → generate again
Try a tool → check the result → if wrong or failed → try a different approach
Draft a plan → execute it → check for errors → revise and retry

Example: Self-Correcting Code Generator

In the following example, we build a graph that generates Python code, evaluates its quality, and loops back to generate an improved version until the code passes validation or reaches the maximum number of attempts:

from typing import TypedDict, Optional
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START, END

llm = init_chat_model(model="gpt-4o", model_provider="openai")

class State(TypedDict):
    task: str
    code: str
    error: Optional[str]
    attempts: int
    final_code: str

def generate_code(state: State) -> dict:
    if state.get("error"):
        prompt = (
            f"Fix this Python code:\n{state['code']}\n\n"
            f"Error: {state['error']}\n"
            "Return only valid Python code without Markdown."
        )
    else:
        prompt = (
            f"Write a Python function that {state['task']}. "
            "Return only valid Python code without Markdown."
        )

    code = llm.invoke(prompt).content.strip()

    return {
        "code": code,
        "attempts": state.get("attempts", 0) + 1,
        "error": None,
    }

def validate_code(state: State) -> dict:
    try:
        compile(state["code"], "<string>", "exec")
        return {"error": None, "final_code": state["code"]}
    except SyntaxError as e:
        return {"error": str(e), "final_code": ""}

def route(state: State) -> str:
    if not state.get("error") or state["attempts"] >= 3:
        return "done"
    return "retry"

builder = StateGraph(State)
builder.add_node("generate", generate_code)
builder.add_node("validate", validate_code)
builder.add_edge(START, "generate")
builder.add_edge("generate", "validate")
builder.add_conditional_edges(
    "validate",
    route,
    {"retry": "generate", "done": END},
)

graph = builder.compile()

result = graph.invoke({
    "task": "calculate the factorial of a number recursively",
    "code": "",
    "error": None,
    "attempts": 0,
    "final_code": "",
})

print("Final Code:")
print(result["final_code"])
print(f"Generated in {result['attempts']} attempt(s)")

The Execution Flow

We must always add an exit condition to our cycles. A cycle without a maximum attempt counter or a clear success condition creates an infinite loop.

Parallel Execution

LangGraph supports parallel execution, i.e. run multiple nodes simultaneously and merging their results. This is ideal when we have independent tasks that do not depend on each other. Parallel execution in LangGraph is powered by the Send API. It allows a node to dispatch multiple parallel tasks dynamically, each with its own state slice.

Example: Parallel Content Generation

In the following example, we build a graph that dynamically dispatches multiple content-generation tasks to run in parallel and combines their results into the shared graph state:

from typing import TypedDict, List, Annotated
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
import operator

# Step 1: Initialize the LLM
llm = init_chat_model(model="gpt-4o", model_provider="openai")

# Step 2: Define the States
class State(TypedDict):
    topic: str
    results: Annotated[List[str], operator.add]

class BranchState(TypedDict):
    topic: str
    task: str

# Step 3: Define the Nodes
def write_section(state: BranchState) -> dict:
    response = llm.invoke(
        f"For a blog post about '{state['topic']}', write {state['task']}. "
        "Keep it concise."
    )
    return {"results": [f"{state['task']}:\n{response.content}"]}

def dispatch_tasks(state: State):
    tasks = ["a catchy title", "a 5-point outline", "an engaging introduction"]
    return [
        Send("write_section", {"topic": state["topic"], "task": task})
        for task in tasks
    ]

# Step 4: Define the Edges
builder = StateGraph(State)
builder.add_node("write_section", write_section)
builder.add_conditional_edges(
    START,
    dispatch_tasks,
    ["write_section"],
)
builder.add_edge("write_section", END)
graph = builder.compile()

# Step 5: Run the graph
result = graph.invoke({
    "topic": "LangGraph for AI Developers",
    "results": [],
})

# Print the result
for section in result["results"]:
    print(section)
    print("---")

The Execution Flow

When to Use Parallel Execution

Scenario	Benefit
Generating multiple independent sections	Significantly faster than sequential
Calling multiple APIs simultaneously	Reduces total wait time
Running the same task on multiple inputs	Processes all inputs at once
Evaluating output from multiple perspectives	Gets parallel viewpoints simultaneously

Persistence and Checkpointing

By default, a LangGraph graph is stateless between separate .invoke() calls. Each run starts fresh and real-world applications (e.g. chatbots, long-running workflows, multi-session assistants) need to remember previous interactions and resume from where they left off.

That's where Checkpointers come in.

What is a Checkpointer?

A Checkpointer automatically saves the full graph state after every node execution. This means:

If the graph is interrupted, it can be resumed from the last saved checkpoint.
Multiple .invoke() calls on the same thread_id share and continue the same state.
The complete execution history is available for inspection and debugging.

Example: In-Memory Checkpointing

In the following example, we build a conversational graph that uses an in-memory checkpointer to persist message history across multiple graph invocations sharing the same thread_id:

from typing import TypedDict, Annotated
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage

# Step 1: Initialize the LLM
llm = init_chat_model(model="gpt-4o", model_provider="openai")

# Step 2: Define the State
class ChatState(TypedDict):
    messages: Annotated[list, add_messages]

# Step 3: Define the Node
def chat(state: ChatState) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# Step 4: Define the Edge and Build the Graph
graph_builder = StateGraph(ChatState)
graph_builder.add_node("chat", chat)
graph_builder.add_edge(START, "chat") 
graph_builder.add_edge("chat", END)

# Step 5: Compile the Graph assigning a Checkpointer with Memory
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "session_001"}}

# Testing the graph
## Turn 1
result1 = graph.invoke(
    {"messages": [HumanMessage(content="My name is David.")]},
    config=config
)
print(result1["messages"][-1].content)
# Output: Hello David! It's nice to meet you. How can I help you today?

## Turn 2 — the graph remembers the full conversation
result2 = graph.invoke(
    {"messages": [HumanMessage(content="What is my name?")]},
    config=config
)
print(result2["messages"][-1].content)
# Output: Your name is David.

The Execution Flow

For applications where state must survive application restarts, we can use a database-backed checkpointer. LangGraph provides different checkpointer implementations depending on the application's requirements:

Checkpointer	Package	Use Case
`InMemorySaver`	`langgraph-checkpoint`	Development, debugging, and testing
`SqliteSaver`	`langgraph-checkpoint-sqlite`	Experimentation and lightweight local workflows
`AsyncSqliteSaver`	`langgraph-checkpoint-sqlite`	Async experimentation and lightweight local workflows
`PostgresSaver`	`langgraph-checkpoint-postgres`	Production workloads
`AsyncPostgresSaver`	`langgraph-checkpoint-postgres`	Async production workloads

For instance, we can use SqliteSaver to persist graph checkpoints in a physical SQLite database file:

import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver

connection = sqlite3.connect(
    "checkpoints.db",
    check_same_thread=False
)

checkpointer = SqliteSaver(connection)

graph = graph_builder.compile(
    checkpointer=checkpointer
)

Unlike InMemorySaver, checkpoints stored in SQLite are persisted to disk and can be retrieved after the application restarts. When invoking the graph, we must provide a thread_id, which the checkpointer uses to identify and retrieve the persisted state associated with a particular execution thread:

config = {
    "configurable": {
        "thread_id": "session_001"
    }
}

result = graph.invoke(
    {"messages": [...]},
    config=config
)

By reusing the same thread_id across graph invocations, LangGraph can retrieve the previously persisted state and continue the same execution thread.

Inspecting Saved State

Once a graph uses a checkpointer, we can inspect both the current state of a thread and its previous checkpoints. This is useful for debugging, monitoring long-running workflows, and understanding how the state evolved throughout the graph execution:

# View the current state of a thread
current_state = graph.get_state(config)
print(current_state.values)

# View the full history of a thread
for checkpoint in graph.get_state_history(config):
    print(f"Step: {checkpoint.metadata.get('step')}")
    print(f"State: {checkpoint.values}")
    print("---")

How Persistence Works

The thread_id acts as the identifier for a persistent execution thread. When we invoke the graph with a particular thread_id, the checkpointer retrieves the latest state associated with that thread, applies the new input, executes the graph, and saves the updated state. Reusing the same thread_id allows subsequent invocations to continue from the previously persisted state:

Human-in-the-Loop

One of LangGraph's most distinctive features is native support for Human-in-the-Loop (HITL), which is the ability to pause a running graph, present information to a human, collect their input or approval, and then resume exactly where the graph left off.

This is critical for production AI systems where autonomous decisions carry real consequences.

Why Human-in-the-Loop Matters

Safety: A human can catch and correct AI mistakes before they cause harm.
Trust: Users trust AI systems more when they know a human is in the control loop.
Compliance: Many industries require human approval before acting on AI recommendations.

How It Works: Interrupt

LangGraph implements HITL through interrupts — checkpoints in the graph where execution pauses and waits:

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
graph = graph_builder.compile(
    checkpointer=memory,
    interrupt_before=["execute_action"]  # Pause before this node
)

Example: AI Action Approval Workflow

In the following example, we build a graph that pauses execution before performing a proposed action, allowing a human to review and approve or reject it before the workflow continues:

from typing import TypedDict, Optional
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END

llm = init_chat_model(model="gpt-4o", model_provider="openai")

class State(TypedDict):
    user_request: str
    proposed_action: str
    human_approved: Optional[bool]
    result: str

def propose_action(state: State) -> dict:
    response = llm.invoke(
        f"A user wants to: '{state['user_request']}'\n\n"
        f"Propose a specific action to fulfill this request. "
        f"Be concise and specific."
    )
    return {"proposed_action": response.content}

def execute_action(state: State) -> dict:
    if not state.get("human_approved"):
        return {"result": "Action was rejected by the human reviewer."}

    response = llm.invoke(
        f"Execute the following action and describe what was done:\n"
        f"{state['proposed_action']}"
    )
    return {"result": response.content}

graph_builder = StateGraph(State)
graph_builder.add_node("propose_action", propose_action)
graph_builder.add_node("execute_action", execute_action)

graph_builder.add_edge(START, "propose_action")
graph_builder.add_edge("propose_action", "execute_action")
graph_builder.add_edge("execute_action", END)

memory = MemorySaver()
graph = graph_builder.compile(
    checkpointer=memory,
    interrupt_before=["execute_action"]
)

config = {"configurable": {"thread_id": "approval_001"}}

# Step 1: Run until the interrupt
print("=== Running until human review point ===")
graph.invoke(
    {
        "user_request": "Delete all files older than 30 days from the archive folder",
        "proposed_action": "",
        "human_approved": None,
        "result": ""
    },
    config=config
)

# Show the proposed action to the human
current_state = graph.get_state(config)
proposed = current_state.values["proposed_action"]
print(f"\nProposed Action:\n{proposed}")

# Step 2: Human reviews and provides input
print("\n=== Waiting for human approval ===")
human_decision = input("Do you approve this action? (yes/no): ").strip().lower()
approved = human_decision == "yes"

# Step 3: Update state with human decision and resume
graph.update_state(config, {"human_approved": approved})

print("\n=== Resuming graph after human input ===")
final_result = graph.invoke(None, config=config)
print(f"\nResult: {final_result['result']}")

The Execution Flow

Subgraphs

As our LangGraph applications grow in complexity, a single flat graph becomes difficult to manage. Subgraphs allow us to break a large, complex graph into smaller, reusable, self-contained graphs, being each responsible for one part of the workflow.

What is a Subgraph?

A subgraph is a fully compiled LangGraph graph that is used as a node inside a parent graph. From the parent's perspective, it looks and behaves like any other node, i.e. it receives state, does its work, and returns updates.

Why Use Subgraphs?

Benefit	Description
Modularity	Break complex workflows into focused, manageable units
Reusability	Use the same subgraph in multiple parent graphs
Testability	Test each subgraph independently before composing them
Readability	Parent graph stays clean and high-level

Example: Research and Writing Subgraphs

In the following example, we build a graph that consists of two subgraphs: one for research and another for writing. Each subgraph is responsible for a different part of the workflow, and they are composed into a single parent graph:

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model

llm = init_chat_model(model="gpt-4o", model_provider="openai")

# ─────────────────────────────────────────
# Subgraph 1: Research Subgraph
# ─────────────────────────────────────────
class ResearchState(TypedDict):
    topic: str
    research_notes: str
    key_facts: str

def gather_notes(state: ResearchState) -> dict:
    response = llm.invoke(
        f"Research '{state['topic']}' and provide detailed notes."
    )
    return {"research_notes": response.content}

def extract_key_facts(state: ResearchState) -> dict:
    response = llm.invoke(
        f"Extract the 5 most important facts from these notes:\n{state['research_notes']}"
    )
    return {"key_facts": response.content}

research_builder = StateGraph(ResearchState)
research_builder.add_node("gather_notes", gather_notes)
research_builder.add_node("extract_key_facts", extract_key_facts)
research_builder.add_edge(START, "gather_notes")
research_builder.add_edge("gather_notes", "extract_key_facts")
research_builder.add_edge("extract_key_facts", END)

research_subgraph = research_builder.compile()

# ─────────────────────────────────────────
# Subgraph 2: Writing Subgraph
# ─────────────────────────────────────────
class WritingState(TypedDict):
    topic: str
    key_facts: str
    draft: str
    polished_report: str

def write_draft(state: WritingState) -> dict:
    response = llm.invoke(
        f"Write a report about '{state['topic']}' using these facts:\n{state['key_facts']}"
    )
    return {"draft": response.content}

def polish_report(state: WritingState) -> dict:
    response = llm.invoke(
        f"Polish and improve this report for clarity and flow:\n{state['draft']}"
    )
    return {"polished_report": response.content}

writing_builder = StateGraph(WritingState)
writing_builder.add_node("write_draft", write_draft)
writing_builder.add_node("polish_report", polish_report)
writing_builder.add_edge(START, "write_draft")
writing_builder.add_edge("write_draft", "polish_report")
writing_builder.add_edge("polish_report", END)

writing_subgraph = writing_builder.compile()

# ─────────────────────────────────────────
# Parent Graph: Orchestrates both subgraphs
# ─────────────────────────────────────────
class ParentState(TypedDict):
    topic: str
    research_notes: str
    key_facts: str
    draft: str
    polished_report: str

parent_builder = StateGraph(ParentState)

parent_builder.add_node("research", research_subgraph)
parent_builder.add_node("writing", writing_subgraph)

parent_builder.add_edge(START, "research")
parent_builder.add_edge("research", "writing")
parent_builder.add_edge("writing", END)

parent_graph = parent_builder.compile()

result = parent_graph.invoke({
    "topic": "The future of AI agents",
    "research_notes": "",
    "key_facts": "",
    "draft": "",
    "polished_report": ""
})

print("=== FINAL REPORT ===")
print(result["polished_report"])

The Execution Flow

Building a Complete AI Workflow

Now let's bring everything together. We will build a complete, end-to-end AI research assistant that combines state management, conditional routing, cycles, parallel execution, persistence, and human-in-the-loop into a single cohesive workflow.

The Workflow

Our complete AI workflow will:

Research the topic (gather notes in parallel from multiple angles)
Write a draft report from the research
Review the draft and loop back for revisions if needed
Pause for human approval before finalizing
Finalize the approved report

State Definition

class WorkflowState(TypedDict):
    topic: str
    research_angles: List[str]
    research_results: Annotated[List[str], operator.add]
    combined_research: str
    draft: str
    review_feedback: Optional[str]
    revision_count: int
    human_approved: Optional[bool]
    final_report: str

Node Definitions

# Node 1: Plan research angles
def plan_research(state: WorkflowState) -> dict:
    response = llm.invoke(
        f"List exactly 3 distinct research angles for the topic: '{state['topic']}'. "
        f"Return them as a simple numbered list."
    )
    angles = [
        line.strip()
        for line in response.content.split("\n")
        if line.strip() and line.strip()[0].isdigit()
    ]
    return {"research_angles": angles[:3]}

# Node 2: Research one angle (used in parallel)
def research_angle(state: dict) -> dict:
    response = llm.invoke(
        f"Research this specific angle about '{state['topic']}':\n"
        f"{state['angle']}\n\nProvide 3-4 key insights."
    )
    return {"research_results": [f"[{state['angle']}]\n{response.content}"]}

# Dispatcher: fan out to parallel research nodes
def dispatch_research(state: WorkflowState):
    return [
        Send("research_angle", {"topic": state["topic"], "angle": angle})
        for angle in state["research_angles"]
    ]

# Node 3: Combine parallel research results
def combine_research(state: WorkflowState) -> dict:
    combined = "\n\n".join(state["research_results"])
    response = llm.invoke(
        f"Synthesize these research findings into a cohesive summary:\n\n{combined}"
    )
    return {"combined_research": response.content}

# Node 4: Write the draft
def write_draft(state: WorkflowState) -> dict:
    feedback = state.get("review_feedback", "")
    revision = state.get("revision_count", 0)

    if feedback and revision > 0:
        prompt = (
            f"Revise this draft about '{state['topic']}' based on feedback.\n\n"
            f"Original Draft:\n{state['draft']}\n\n"
            f"Feedback:\n{feedback}\n\nWrite the improved version."
        )
    else:
        prompt = (
            f"Write a professional report about '{state['topic']}' "
            f"using this research:\n\n{state['combined_research']}"
        )

    response = llm.invoke(prompt)
    return {
        "draft": response.content,
        "revision_count": revision + 1,
        "review_feedback": None
    }

# Node 5: Review the draft
def review_draft(state: WorkflowState) -> dict:
    response = llm.invoke(
        f"Review this report and respond with either:\n"
        f"'APPROVED' — if complete and high quality\n"
        f"'REVISE: <feedback>' — if improvements are needed\n\n"
        f"Report:\n{state['draft']}"
    )
    feedback_text = response.content

    if "APPROVED" in feedback_text:
        return {"review_feedback": None}
    else:
        return {"review_feedback": feedback_text}

# Node 6: Finalize the report
def finalize_report(state: WorkflowState) -> dict:
    return {"final_report": state["draft"]}

# Routing: after AI review
def route_after_review(state: WorkflowState) -> str:
    if state.get("revision_count", 0) >= 3:
        return "finalize"
    if state.get("review_feedback"):
        return "revise"
    return "finalize"

Graph Definition

graph_builder = StateGraph(WorkflowState)

graph_builder.add_node("plan_research",    plan_research)
graph_builder.add_node("research_angle",   research_angle)
graph_builder.add_node("combine_research", combine_research)
graph_builder.add_node("write_draft",      write_draft)
graph_builder.add_node("review_draft",     review_draft)
graph_builder.add_node("finalize_report",  finalize_report)

graph_builder.add_edge(START, "plan_research")

graph_builder.add_conditional_edges(
    "plan_research",
    dispatch_research,
    ["research_angle"]
)

graph_builder.add_edge("research_angle", "combine_research")
graph_builder.add_edge("combine_research", "write_draft")
graph_builder.add_edge("write_draft", "review_draft")

graph_builder.add_conditional_edges(
    "review_draft",
    route_after_review,
    {
        "revise":   "write_draft",
        "finalize": "finalize_report"
    }
)

graph_builder.add_edge("finalize_report", END)

memory = MemorySaver()
graph = graph_builder.compile(
    checkpointer=memory,
    interrupt_before=["finalize_report"]
)

Running the Workflow

config = {"configurable": {"thread_id": "complete_workflow_001"}}

print("=== PHASE 1: AI Research and Writing ===")
graph.invoke(
    {
        "topic": "The impact of AI agents on software development",
        "research_angles": [],
        "research_results": [],
        "combined_research": "",
        "draft": "",
        "review_feedback": None,
        "revision_count": 0,
        "human_approved": None,
        "final_report": ""
    },
    config=config
)

current_state = graph.get_state(config)
print("\n=== PHASE 2: Human Review ===")
print("Draft Report:")
print(current_state.values["draft"])
print(f"\nRevisions made by AI: {current_state.values['revision_count']}")

decision = input("\nApprove this report? (yes/no): ").strip().lower()
graph.update_state(config, {"human_approved": decision == "yes"})

print("\n=== PHASE 3: Finalizing ===")
final = graph.invoke(None, config=config)
print("\n=== FINAL REPORT ===")
print(final["final_report"])

The Execution Flow

Multi-Agent Systems

LangGraph's most advanced capability is enabling multi-agent systems, i.e. architectures where multiple specialized AI agents collaborate, each focusing on what it does best, coordinated by LangGraph's graph structure.

What is a Multi-Agent System?

Instead of one LLM trying to do everything, a multi-agent system divides responsibilities:

Agent	Role
Orchestrator / Supervisor	Coordinates the other agents, decides who works next
Researcher	Gathers and summarizes information
Writer	Drafts and revises content
Reviewer	Critiques and provides quality feedback
Executor	Takes actions in the real world (APIs, databases)

Supervisor Pattern

The most common multi-agent pattern in LangGraph is the Supervisor Pattern, in which one supervisor agent routes work to specialized worker agents:

from typing import TypedDict
from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command

# Step 1: Initialize the LLM
llm = init_chat_model(model="gpt-4o", model_provider="openai")


# Step 2: Define the state
class AgentState(TypedDict):
    task: str
    research: str
    draft: str
    feedback: str
    final: str
    next_agent: str
    iteration: int


# Step 3: Define Supervisor Agent
def supervisor(state: AgentState) -> dict:
    iteration = state.get("iteration", 0)

    prompt = ChatPromptTemplate.from_messages([
        ("system",
         "You are a supervisor managing a team of AI agents.\n"
         "Available agents: researcher, writer, reviewer, human_review\n\n"
         "Rules:\n"
         "- Start with 'researcher' if there is no research yet\n"
         "- Use 'writer' after research is complete\n"
         "- Use 'reviewer' after a draft exists\n"
         "- Use 'human_review' after review or after 3 iterations\n\n"
         "Respond with only the agent name to call next."),
        ("human",
         "Task: {task}\n"
         "Research done: {has_research}\n"
         "Draft done: {has_draft}\n"
         "Last feedback: {feedback}\n"
         "Iterations: {iteration}\n\n"
         "Which agent should work next?")
    ])

    response = (prompt | llm).invoke({
        "task": state["task"],
        "has_research": "yes" if state.get("research") else "no",
        "has_draft": "yes" if state.get("draft") else "no",
        "feedback": state.get("feedback", "none"),
        "iteration": iteration
    })

    return {"next_agent": response.content.strip().lower()} # type: ignore


# Step 4: Define Worker Agents
def researcher(state: AgentState) -> dict:
    response = llm.invoke(
        f"Research this task thoroughly and provide key findings:\n{state['task']}"
    )
    return {
        "research": response.content,
        "iteration": state.get("iteration", 0) + 1
    }


def writer(state: AgentState) -> dict:
    response = llm.invoke(
        f"Write a professional report for this task:\n{state['task']}\n\n"
        f"Based on this research:\n{state['research']}"
    )
    return {
        "draft": response.content,
        "iteration": state.get("iteration", 0) + 1
    }


def reviewer(state: AgentState) -> dict:
    response = llm.invoke(
        f"Review this draft critically. Be specific about what is good "
        f"and what needs improvement:\n{state['draft']}"
    )
    return {
        "feedback": response.content,
        "iteration": state.get("iteration", 0) + 1
    }


# Human-in-the-Loop Agent: pause the graph and request human approval
def human_review(state: AgentState) -> dict:
    approved = interrupt({
        "draft": state["draft"],
        "feedback": state["feedback"],
        "question": "Do you approve this report?"
    })

    # Route to finalizer if approved, otherwise return to the writer
    return {
        "next_agent": "finalizer" if approved else "writer"
    }


def finalizer(state: AgentState) -> dict:
    response = llm.invoke(
        f"Polish this draft into a final, publication-ready report:\n{state['draft']}"
    )
    return {"final": response.content}


# Routing: supervisor decides next agent
def route_by_supervisor(state: AgentState) -> str:
    next_agent = state.get("next_agent", "researcher")
    valid_agents = ["researcher", "writer", "reviewer", "human_review"]
    return next_agent if next_agent in valid_agents else "human_review"


# Routing: human decision determines whether to finalize or revise
def route_after_human_review(state: AgentState) -> str:
    return state["next_agent"]


# Step 5: Build the Multi-Agent Graph
graph_builder = StateGraph(AgentState)

graph_builder.add_node("supervisor",   supervisor)
graph_builder.add_node("researcher",   researcher)
graph_builder.add_node("writer",       writer)
graph_builder.add_node("reviewer",     reviewer)
graph_builder.add_node("human_review", human_review)
graph_builder.add_node("finalizer",    finalizer)

graph_builder.add_edge(START, "supervisor") # or `graph_builder.set_entry_point("supervisor")`

graph_builder.add_conditional_edges(
    "supervisor",
    route_by_supervisor,
    {
        "researcher":   "researcher",
        "writer":       "writer",
        "reviewer":     "reviewer",
        "human_review": "human_review"
    }
)

# All workers report back to supervisor (except human_review and finalizer)
graph_builder.add_edge("researcher", "supervisor")
graph_builder.add_edge("writer",     "supervisor")
graph_builder.add_edge("reviewer",   "supervisor")

# Human approval routes to finalizer; rejection routes back to writer
graph_builder.add_conditional_edges(
    "human_review",
    route_after_human_review,
    {
        "writer":    "writer",
        "finalizer": "finalizer"
    }
)

graph_builder.add_edge("finalizer", END)


# Step 6: Compile the Graph with a checkpointer
# A checkpointer is required to pause and resume execution with interrupt()
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

# The same thread_id must be used when starting and resuming the graph
config = {"configurable": {"thread_id": "report_001"}}


# Step 7: Run the graph until the Human-in-the-Loop interrupt
result = graph.invoke({
    "task": "Write a comprehensive overview of LangGraph for a technical blog",
    "research": "",
    "draft": "",
    "feedback": "",
    "final": "",
    "next_agent": "",
    "iteration": 0
}, config=config) # type: ignore


# Step 8: Display the information sent by the Human-in-the-Loop interrupt
interrupt_data = result["__interrupt__"][0].value

print("=== HUMAN REVIEW ===")
print("\nDraft:")
print(interrupt_data["draft"])

print("\nAI Reviewer Feedback:")
print(interrupt_data["feedback"])


# Step 9: Collect the human decision
decision = input("\nApprove report? (yes/no): ").strip().lower()
approved = decision == "yes"


# Step 10: Resume the graph with the human decision
# Command(resume=...) becomes the return value of interrupt()
result = graph.invoke(
    Command(resume=approved),
    config=config # type: ignore
)

# Step 11: Print the final result
print("\n=== FINAL OUTPUT ===")
print(result["final"])

The Execution Flow

Production Features

Taking a LangGraph application from prototype to production requires attention to reliability, observability, scalability, and error handling.

1. LangSmith Tracing

LangSmith automatically traces every step of our graph, making it easy to debug failures, measure latency, and monitor quality:

export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY="my-langsmith-api-key"
export LANGCHAIN_PROJECT="my-langgraph-app"

Once set, every graph invocation is automatically traced — no code changes required.

2. Error Handling and Retries

LLM calls and external services can fail temporarily due to network errors, rate limits, timeouts, or unavailable dependencies. Production nodes should handle these failures gracefully and retry transient errors instead of immediately failing the entire workflow. A common strategy is to retry the operation a limited number of times using exponential backoff:

import time

def resilient_node(state: State) -> dict:
    max_retries = 3
    for attempt in range(max_retries):
        try:
            response = llm.invoke(state["input"])
            return {"output": response.content, "error": ""}
        except Exception as e:
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)  # Exponential backoff
                continue
            return {"output": "", "error": str(e)}

3. Streaming Graph Output

For long-running workflows, waiting until the entire graph finishes before displaying any result can create a poor user experience. LangGraph supports streaming, allowing us to observe state updates and intermediate results as the graph executes:

config = {"configurable": {"thread_id": "stream_session"}}

for event in graph.stream(
    {"question": "Explain LangGraph in detail"},
    config=config,
    stream_mode="values"
):
    if "messages" in event:
        last_message = event["messages"][-1]
        if hasattr(last_message, "content"):
            print(last_message.content, end="", flush=True)

4. Async Execution

LangGraph also supports asynchronous execution through methods such as .ainvoke() and .astream(). Async execution is particularly useful for applications that perform I/O-bound operations, such as calling LLMs, APIs, databases, or other external services, because it allows the application to perform other work while waiting for those operations to complete:

import asyncio

async def async_node(state: State) -> dict:
    response = await llm.ainvoke(state["question"])
    return {"answer": response.content}

async def main():
    result = await graph.ainvoke(
        {"question": "What is LangGraph?"},
        config=config
    )
    print(result["answer"])

asyncio.run(main())

5. Runtime Configuration

Not every execution of a graph needs to use exactly the same configuration. LangGraph allows runtime values to be passed through the config object, making it possible to dynamically configure parameters such as the model, temperature, user identifiers, or other application-specific settings without modifying the graph state itself:

from langchain_core.runnables import RunnableConfig

def configurable_node(state: State, config: RunnableConfig) -> dict:
    model_name = config.get("configurable", {}).get("model", "gpt-4o")
    model_provider = config.get("configurable", {}).get("provider", "openai")
    temperature = config.get("configurable", {}).get("temperature", 0.7)

    llm = init_chat_model(model=model_name, model_provider=model_provider, temperature=temperature)
    response = llm.invoke(state["question"])
    return {"answer": response.content}

result = graph.invoke(
    {"question": "Explain AI agents"},
    config={
        "configurable": {
            "thread_id": "session_001",
            "model": "gpt-4o",
            "provider": "openai",
            "temperature": 0.2
        }
    }
)

6. Production Checkpointing with Persistent Storage

MemorySaver is useful during development, but its checkpoints exist only in application memory and are lost when the process restarts. Production applications should use a persistent checkpointer backed by durable storage, such as PostgreSQL, allowing graph state to survive application restarts and enabling workflows to be resumed across different application instances:

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:password@localhost:5432/langgraph_db"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()
    graph = graph_builder.compile(checkpointer=checkpointer)

    result = graph.invoke(
        {"question": "What is LangGraph?"},
        config={"configurable": {"thread_id": "prod_session_001"}}
    )

Production Readiness Checklist

Feature	Tool	Purpose
Observability	LangSmith, Langfuse	Trace, debug, and monitor graph execution
Error Handling	`try/except` + retries	Resilience against transient failures
Streaming	`graph.stream()`	Responsive real-time user experience
Async	`graph.ainvoke()`	High-throughput concurrent processing
Persistence	Persistent storage checkpointer	Durable state across server restarts
Configuration	`RunnableConfig`	Runtime flexibility without code changes

Next Steps

Now that you have a solid understanding of LangGraph's core concepts and have built real-world applications with them, here are the natural next steps to expand your knowledge:

Explore the Ecosystem

LangGraph Studio: A visual IDE for building, running, and debugging LangGraph graphs interactively.
LangSmith: Set up evaluation pipelines, monitor production traces, and measure the quality of AI outputs.
Langfuse: An open-source alternative to LangSmith for LLM observability, tracing, evaluation, and debugging, with support for self-hosting.
LangGraph Cloud: Deploy your LangGraph applications to a fully managed, scalable cloud infrastructure.

Advanced Patterns to Learn

Reflection Agents: Agents that critique their own outputs and iteratively self-improve.
Plan-and-Execute: Agents that first create a full plan and then execute each step systematically.
Corrective RAG: A RAG pattern where the agent checks if retrieved documents are relevant and searches again if they are not.
Multi-Agent Debate: Multiple agents with opposing perspectives debate a topic to reach a higher-quality conclusion.

Wrap Up

Throughout this guide, we built our understanding of LangGraph from the ground up, from a single-node graph to a complete, production-ready multi-agent AI workflow.

Here is a summary of everything we covered:

Topic	Purpose	Key Takeaway
Prerequisites	Foundation knowledge	Know Python, LangChain basics, and LLM concepts
What Is LangGraph?	Core definition	A graph-based framework for stateful AI workflows
Why LangGraph Exists?	Motivation	Extends LangChain with cycles, complex state, and human oversight
LangChain vs LangGraph	Choosing the right tool	Complementary tools — LangGraph extends LangChain for complex workflows
The Core Concept	Mental model	Nodes do the work; edges control the flow
Our First Graph	Getting started	State + Node + Edge = a working LangGraph app
State Management	Shared memory	A TypedDict that evolves as it flows through the graph
Nodes	Units of work	Plain Python functions that read state and return updates
Edges	Flow control	Fixed or conditional connections between nodes
Conditional Routing	Dynamic decisions	Routing functions that inspect state and choose the next node
Cycles and Loops	Self-correction	Native looping — not possible in a linear LangChain chain
Parallel Execution	Speed and efficiency	Fan out independent tasks with the Send API
Persistence	Memory across sessions	Checkpointers save and restore full graph state automatically
Human-in-the-Loop	Safety and control	Pause, collect human input, and resume seamlessly
Subgraphs	Modularity	Compose large workflows from reusable sub-workflows
Complete Workflow	Everything together	Research → Write → Review → Human Approval → Finalize
Multi-Agent Systems	Collaboration	Specialized agents orchestrated by a supervisor
Production Features	Going to production	Tracing, streaming, async, error handling, and persistence

The Big Picture

LangGraph represents a natural evolution of LangChain, not a replacement, but an extension that unlocks capabilities that linear chains cannot express: cycles, complex shared state, dynamic conditional routing, human oversight, and multi-agent collaboration.

The best way to master LangGraph is to start simple and grow deliberately:

Begin with a single node and a clear state
Add edges to connect your steps
Introduce conditional routing when decisions are needed
Add cycles when self-correction is required
Layer in persistence, human oversight, and multiple agents as complexity demands

Remember, every production AI system we admire is built on exactly these principles.

All the Python examples presented in this guide are available in this accompanying GitHub repository, making it easy to run, modify, and experiment with each example on your own.

Happy AI coding 🚀

LangChain Explained: A Practical Guide For Beginners

David Archanjo — Wed, 15 Jul 2026 03:08:50 +0000

In this guide, I will explain what LangChain is, why it exists, and how it simplifies the development of LLM-powered applications. Instead of presenting isolated concepts, we will build our understanding step by step, starting with a simple LLM integration and gradually exploring prompt templates, output parsers, memory, Retrieval-Augmented Generation (RAG), and tool calling.

By the end of this article, you will understand how these building blocks fit together and how LangChain orchestrates them into a cohesive workflow.

About Me

Prerequisites
What Is LangChain?
Why Learn and Use LangChain?
The Core Concept of LangChain
Hello, World!: LLM Integration
Prompt Templates
Output Parsers
LangChain Expression Language (LCEL)
Conversational Memory Management
Retrieval-Augmented Generation (RAG)
Tool Calling and Agents
Next Steps
Wrap Up

Prerequisites

To get the most out of this guide, you should have:

Python 3.10 or later and pip installed.
Basic knowledge of Python.
A text editor or IDE such as Visual Studio Code or PyCharm.
A basic understanding of Large Language Models (LLMs).

What Is LangChain?

LangChain is a toolkit that helps developers build smart AI applications powered by language models.

Think of a language model as a very smart brain, but on its own, it:

Cannot access our private files
Cannot search the web
Does not remember past conversations
Cannot use tools like calculators or databases

LangChain connects that brain to tools, data, and memory.

Why Learn and Use LangChain?

To understand why we need LangChain, we first have to understand the limitation of LLMs like GPT-4 or Claude.

While LLMs are great, they are like "brains in a jar". They are powerful at processing information but have no internet access, no memory of past conversations (beyond a single session), and no way to interact with our specific private data, just to name a few of their limitations.

LangChain is the "Swiss Army Knife" that connects LLMs to the rest of the world.

Five Reasons Why LangChain is Necessary

1. "Chaining" Multiple Steps

A simple prompt is easy, but complex tasks (like writing a research paper) require a sequence.

Without LangChain: We have to manually take the output of one prompt, clean it up, and paste it into the next.
With LangChain: We create a "Chain." For example: Step 1: Search Google -> Step 2: Summarize results -> Step 3: Translate to Portuguese. LangChain takes all of these pieces and runs as a flow.

2. Context Management (Memory)

LLMs are technically "stateless". Once a session ends, they don't remember who we are, what we talked about.

Without LangChain: Developers have to manually store chat logs in a database and feed them back to the AI every time, which is problematic and time-consuming.
With LangChain: It provides built-in "Memory" modules. It automatically tracks the conversation history and feeds the relevant parts back to the LLM so it remembers what we said five minutes ago.

3. Data Awareness (RAG)

LLMs are trained on public data up to a certain date. They don't know about our private stuff, like emails, spreadsheets, or today’s news.

The Solution: How LangChain helps:
Without LangChain: We must feed LLMs manually with additional data in such a way they are eventually able to reason based on what was provide and then answer ours questions or perform the given action accordingly.
With LangChain: It provides "Document Loaders" and "Vector Store" integrations, allowing LLMs to "read" our specific documents and answer questions based only on that data. That's Retrieval Augmented Generation (RAG) to the rescue.

4. Model-agnostic Approach

The AI field moves fast. Today OpenAI's model is the leader; tomorrow it might Anthropic's Claude or an open-source model like Gemma 4.

Without LangChain: If we want or need to switch models, we have to rewrite our entire codebase or application because every AI company has a different API.
With LangChain: It acts as an abstraction layer. We write our code once using LangChain’s syntax, and we can swap the underlying LLM with just one line of code.

5. Using "Agents"

This is the most advanced part of LangChain. An Agent is an AI that can decide for itself which tool to use.

Without LangChain: We must manually code the logic that decides which tool to use and when. For example, if a user asks, "What is Apple’s stock price and how does it affect my investements?", we would need to:
- Detect that stock data is required
- Call a stock price API
- Parse the response
- Retrieve the user’s portfolio data
- Run calculations
- Format the final answer
- All decision-making and orchestration logic have to be hard-coded
With LangChain: We define tools (e.g. Stock API, Calculator, Database), and the Agent automatically decides which tool to use and in what order. The LLM reasons through the task, selects the appropriate tools, executes them, and combines the results into a final response. All this without we manually scripting every decision step.

The Core Concept of LangChain

At its core, LangChain is built around one simple idea: everything is a composable building block.

Just like LEGO bricks, each component in LangChain does one job and is designed to connect with other components. We stack them together to build something bigger and more powerful.

These building blocks are:

Building Block	What It Does
LLM / Chat Model	The brain. Receives input and generates a response.
Prompt Template	Structures and formats the input sent to the LLM.
Output Parser	Transforms the LLM's raw text response into structured data.
Memory	Stores and retrieves conversation history across interactions.
Retriever / Vector Store	Fetches relevant documents from our private data.
Tools	External capabilities the LLM can call (APIs, calculators, databases).
Agent	The decision-maker that chooses which tools to use and when.

How They Connect: The Chain

The term "LangChain" comes from the idea of chaining these blocks together into a pipeline. A typical flow looks like this:

  User Input
      ↓
Prompt Template (formats the input)
      ↓
     LLM        (generates a response)
      ↓
 Output Parser  (structures the response)
      ↓
 Final Output

As the application grows in complexity, more blocks are added to the chain:

  User Input
      ↓
   Memory       (adds conversation history)
      ↓
  Retriever     (fetches relevant documents)
      ↓
Prompt Template (formats everything together)
      ↓
     LLM        (reasons and generates a response)
      ↓
 Output Parser  (structures the response)
      ↓
 Final Output

In the section that follow, we will explore each of these building blocks individually, starting from the simplest possible example and progressively adding more components until we are able to get a complete AI application powered by LangChain.

Hello, World!: LLM Integration

As with any new technology, the best place to start is with a basic "Hello, World!". In LangChain, the most fundamental building block is LLM integration: sending a prompt to a language model and receiving its response.

So let's get started!

Installation

First, we install LangChain and the OpenAI integration packages:

pip install langchain langchain-openai

Each package serves a specific purpose:

Package	Purpose
`langchain`	Provides the core LangChain framework, including abstractions for prompts, document loaders, text splitters, embeddings, vector stores, retrievers, and chains.
`langchain-openai`	Adds integration with OpenAI-compatible models, allowing LangChain to interact with chat models through the `ChatOpenAI` class.

Then, set our OpenAI API key as an environment variable:

export OPENAI_API_KEY="sk-xxxxx"

Our First LLM Call

from langchain_openai import ChatOpenAI

# Initialize the model
llm = ChatOpenAI(model="gpt-4o")

# Invoke the model with a simple prompt
response = llm.invoke("What is LangChain in one sentence?")

print(response.content)

Output:

LangChain is a framework that helps developers build applications powered by large language models by connecting them to tools, data, and memory.

Breaking Down The Code

We instantiated a ChatOpenAI model, pointing to gpt-4o.
We called llm.invoke() with a plain string message as argument.
LangChain handled the API communication and returned an AIMessage object.
We printed out response.content to display the response in plain text.

Running LangChain with a Local OpenAI-Compatible API

Although the code examples in this article use the ChatOpenAI class, they do not require the official OpenAI API. One of the major advantages of LangChain is that its OpenAI integration can communicate with any inference server implementing the OpenAI API specification. This allows us to switch between cloud providers and locally hosted models with little or no code changes.

For this article, we can use a locally hosted language model exposed through an OpenAI-compatible API using, for instance, llama.cpp. This setup allows every example to run entirely on our machine while keeping the code identical to what we would write when using the official OpenAI service.

I personally use llamafile, a standalone distribution of llama.cpp, to serve open models locally, like the Gemma 4:

./llamafile \
  -m gemma-4-E2B-it-Q4_K_M.gguf \
  --server \
  --host 0.0.0.0 \
  --port 8000 \
  --alias gpt-4o

Here, --alias gpt-4o tells the server to expose the local model under the name gpt-4o. From LangChain's perspective, it behaves exactly like the OpenAI model with the same name, allowing every example in this article to work with the same ChatOpenAI initialization:

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

without requiring any code changes.

To make this work transparently, we MUST configure the following environment variables:

export OPENAI_API_KEY="sk-xxxxx"
export OPENAI_BASE_URL="http://localhost:8000/v1"

The OPENAI_BASE_URL environment variable overrides the default OpenAI endpoint, causing every request made through ChatOpenAI to be sent to the local inference server instead of OpenAI's servers.

Although the local server does not validate API keys, the OpenAI client library still expects the OPENAI_API_KEY environment variable to exist. Since the key is never verified by llama.cpp, any non-empty value is sufficient.

Using a local OpenAI-compatible API offers several benefits while learning LangChain:

No OpenAI subscription required.
No API usage costs.
Complete privacy, since prompts and responses remain on our machine.
Offline development, provided the model has already been downloaded.
The same LangChain code works with both local and cloud-hosted models.

If you already have an OpenAI API key, simply remove the OPENAI_BASE_URL environment variable and replace the placeholder value of OPENAI_API_KEY with your own key. The examples throughout this article will continue to work without modification.

In addition to llama.cpp, ChatOpenAI can communicate with any inference server exposing an OpenAI-compatible API, which includes other tools such as LM Studio, vLLM, Ollama and SGLang.

Switching to a Different Model

Because LangChain acts as an abstraction layer, swapping the underlying model requires changing just one line. The following shows how would be using Anthropic's Claude:

# !pip install langchain-anthropic
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-sonnet-4-6")

response = llm.invoke("What is LangChain in one sentence?")
print(response.content)

The rest of our application code stays exactly the same. This is the power of LangChain's model-agnostic approach.

Prompt Templates

Hardcoding a message directly into .invoke() works for simple cases, but real applications need dynamic, reusable prompts, i.e. prompts that change based on user input or context.

That's when Prompt Templates comes to rescue!

What is a Prompt Template?

A Prompt Template is a reusable prompt with placeholders that get filled in at runtime. Think of it like a form letter where we define the structure once and fill in the specific details later.

Without Prompt Template

# Manually formatting strings every time is messy and error-prone
user_language = "Portuguese"
user_topic = "LangChain"
prompt = f"Explain {user_topic} in simple terms. Answer in {user_language}."

response = llm.invoke(prompt)

print(response.content)

This version works, but it becomes unmanageable as prompts grow more complex over time.

With Prompt Template

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# Define the prompt structure once
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful technical writer who explains concepts simply."),
    ("human", "Explain {topic} in simple terms. Answer in {language}.")
])

# Initialize the model
llm = ChatOpenAI(model="gpt-4o")

# Create a chain that combines the prompt template and the model
chain = prompt | llm

# Invoke with dynamic values
response = chain.invoke({"topic": "LangChain", "language": "Portuguese"})

print(response.content)

Key Benefits of Prompt Templates

Reusability: Define the prompt once, use it many times with different inputs.
Separation of Concerns: The prompt structure is separate from the application logic.
Role-based Messaging: Easily define system, human, and ai roles in a conversation.
Composability: Templates connect directly to LLMs using the | pipe operator, forming a chain.

Output Parsers

By default, LangChain returns LLM's response as an AIMessage object. While this works well for conversational applications, many real-world systems require structured data that can be processed programmatically.

For example, an application may need to:

store the response in a database;
serialize it as JSON;
populate a web interface;
call another service;
execute business rules based on individual fields.

Working with plain text quickly becomes inconvenient.

That's when Output Parsers comes to rescue!

The Problem: Default Response Output

Suppose we are building an AI assistant that recommends learning resources.

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

response = llm.invoke("Recommend a resource for learning LangChain.")

print(type(response))   # <class 'langchain_core.messages.ai.AIMessage'>
print(response.content) # I recommend starting with the official LangChain documentation...

The response is human-readable, but extracting fields such as the resource name or URL would require additional parsing.

Parsing to Plain Text

When the application only needs the generated text, StrOutputParser removes the surrounding AIMessage object and returns a simple string.

from langchain_core.output_parsers import StrOutputParser

parser = StrOutputParser()

chain = llm | parser

response = chain.invoke("Explain what a Vector Database is.")

print(type(response)) # <class 'langchain_core.messages.base.TextAccessor'>
print(response)       # A vector database is a specialized database...

This is the simplest parser and the one we will probably use most often in chatbot applications.

NOTE: langchain_core.messages.base.TextAccessor is a specialized, string-like object in LangChain created to ensure backward compatibility during the framework's transition from method-based to property-based text access for messages.

Parsing to Structured Data (JSON)

For more complex use cases, we can instruct the LLM to return structured data and parse it automatically:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser

# Define the prompt structure once
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant. Always respond in valid JSON."),
    ("human", "{question}")
])

# Initialize the model
llm = ChatOpenAI(model="gpt-4o")

# Initialize the output parser
parser = JsonOutputParser()

# Create a chain that combines the prompt template, the model, and the output parser
chain = prompt | llm | parser

# Invoke with a dynamic question
response = chain.invoke({
    "question": "Give me three benefits of LangChain as a JSON array with 'title' and 'description' fields. Items in the JSON array must be sorted alphabetically by the 'title' field."
})

# Now the response is a Python list we can iterate over
for benefit in response:
    print(f"- {benefit['title']}: {benefit['description']}")

Output:

- Composability: LangChain allows you to chain components together into powerful pipelines.
- Ecosystem: A rich library of integrations for tools, memory, and data retrieval.
- Model Agnostic: Switch between LLM providers with a single line of code.

Producing Structured Objects

Now imagine we are developing an learning platform.

Instead of free-form text, we would like the LLM to recommend a course using a predefined schema.

from pydantic import BaseModel, Field

# Define a structured output model for course recommendations
class CourseRecommendation(BaseModel):
    course: str = Field(description="Course name")
    level: str = Field(description="Difficulty level")
    duration_hours: int = Field(description="Estimated duration")
    reason: str = Field(description="Why the course is recommended")

LangChain offers two ways to obtain this object.

Option 1: Native Structured Output

Modern language models can generate structured data directly.

LangChain exposes this capability through with_structured_output(), allowing the model to return a validated Pydantic object without requiring an explicit parser.

# Initialize the model
llm = ChatOpenAI(model="gpt-4o")

# Configure the LLM to use the structured output model
llm = llm.with_structured_output(CourseRecommendation, method="json_schema")

# Invoke the model with a prompt
recommendation = llm.invoke("Recommend a beginner course for learning LangChain.")

# The response is now a structured object we can access its fields directly
print(f"Course: {recommendation.course}")
print(f"Level: {recommendation.level}")
print(f"Duration: {recommendation.duration_hours} hours")
print(f"Reason: {recommendation.reason}")

Output:

Course: LangChain for Beginners
Level: Beginner
Duration: 80 hours
Reason: Hands-on implementation of core concepts, focusing on prompt engineering, chaining, and retrieval mechanisms.

Notice that recommendation is already a CourseRecommendation instance — no manual JSON parsing required. This is the preferred approach whenever the underlying model supports structured outputs.

Option 2: Using `PydanticOutputParser`

Not every model supports native structured output.

In these situations, LangChain can instruct the model to produce JSON matching a predefined schema and then automatically convert it into a Python object.

from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import ChatPromptTemplate

# Initialize the Pydantic parser for the structured output model
parser = PydanticOutputParser(pydantic_object=CourseRecommendation)

# Define the prompt structure with format instructions for the Pydantic parser
prompt = ChatPromptTemplate.from_messages([
    (
        "system",
        """
You are an AI education advisor.

Return ONLY valid JSON.

{format_instructions}
"""
    ),
    (
        "human",
        "Recommend a {level} course for learning {topic}."
    )
])

# Create a chain that combines the prompt template, the model, and the Pydantic output parser
chain = prompt | llm | parser

# Invoke with dynamic values and format instructions
recommendation = chain.invoke({"topic": "LangChain", "level": "beginner", "format_instructions": parser.get_format_instructions()})

# The response is a structured object we can access its fields directly
print(f"Course: {recommendation.course}")
print(f"Level: {recommendation.level}")
print(f"Duration: {recommendation.duration_hours} hours")
print(f"Reason: {recommendation.reason}")

PydanticOutputParser validates the generated JSON and converts it into a CourseRecommendation object before returning it the final result.

Which Approach Should Be Used?

Approach	Recommended Use
`StrOutputParser`	When the application only needs plain text.
`with_structured_output()`	When the model supports native structured outputs. Preferred approach.
`PydanticOutputParser`	When working with models that don't support native structured outputs or when explicit formatting instructions are required.

LangChain Expression Language (LCEL)

In the previous sections, we observed the ǀ pipe operator appearing in the code examples, right?

chain = prompt | llm | parser

This is LangChain Expression Language (LCEL), which is the standard, unified way to compose LangChain components into pipelines, and it is worth understanding deeply because it underpins everything in LangChain.

What is LCEL?

LCEL is a declarative syntax for building chains. The ǀ operator works like a Unix pipe: the output of the component on the left becomes the input of the component on the right.

This single line chain = prompt | llm | parser is read as: "Take the prompt, pipe it into the LLM, then pipe the result into the parser". This piping instruction replaces what used to require many lines of manual orchestration code:

# Step 1: Format the prompt
prompt = prompt.invoke({"topic": "LangChain"})

# Step 2: Invoke the LLM
response = llm.invoke(prompt)

# Step 3: Parse the output
result = parser.invoke(response)

Why LCEL is Better?

The pipe operator isn't just syntactic sugar. It creates a Runnable pipeline.

Instead of this:

prompt = prompt.format(topic="LangChain")
response = llm.invoke(prompt)
result = parser.invoke(response)

we simply write:

chain = prompt | llm | parser

result = chain.invoke({"topic": "LangChain"})

The Runnable Interface

LCEL works because every LangChain component (prompts, LLMs, parsers, retrievers, tools) implements the same interface called Runnable. A Runnable is any object that exposes these standard methods:

Method	Description
`.invoke(input)`	Runs the chain with a single input and returns a single output.
`.stream(input)`	Runs the chain and streams the output token by token.
`.batch(inputs)`	Runs the chain on a list of inputs in parallel.
`.ainvoke(input)`	Async version of `.invoke()` for non-blocking execution.
`.astream(input)`	Async version of `.stream()`.
`.abatch(inputs)`	Async version of `.batch()`.

You find more information about the Runnable interface here.

Because every component shares this interface, they can be freely composed with the ǀ operator without any adapter code.

Streaming with LCEL

One of LCEL's most practical benefits is first-class streaming support. Instead of waiting for the full response, we can stream tokens as they are generated:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Initialize the model
llm = ChatOpenAI(model="gpt-4o")

# Initialize the output parser
parser = StrOutputParser()

# Initialize the prompt template to be used with dynamic questions
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("human", "{question}")
])

# Create the chain combining the prompt template, the model, and the output parser
chain = prompt | llm | parser

# Stream the response token by token
for chunk in chain.stream({"question": "Explain LangChain in 3 sentences."}):
    print(chunk, end="", flush=True)

This gives us a responsive, real-time experience with no additional configuration required.

Batch Processing with LCEL

Need to process multiple inputs at once? Use .batch():

responses = chain.batch([
    {"question": "What is LangChain?"},
    {"question": "What is a Prompt Template?"},
    {"question": "What is RAG?"}
])

for response in responses:
    print(response)
    print("---")

LangChain automatically runs these in parallel, making batch processing significantly faster than sequential calls.

Branching with RunnableParallel

LCEL also supports parallel branches, which consist of running multiple chains simultaneously and merging their results:

from langchain_core.runnables import RunnableParallel
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(model="gpt-4o")
parser = StrOutputParser()

# Set two separate chains
summary_chain = (
    ChatPromptTemplate.from_template("Summarize this topic in one sentence: {topic}")
    | llm
    | parser
)

example_chain = (
    ChatPromptTemplate.from_template("Give one practical example of: {topic}")
    | llm
    | parser
)

# Run both chains in parallel and combine the results
chain = RunnableParallel(
    summary=summary_chain,
    example=example_chain
)

result = chain.invoke({"topic": "LangChain"})

print("Summary:", result["summary"])
print("Example:", result["example"])

Output:

Summary: LangChain is a framework for building LLM-powered applications by composing modular components.
Example: A customer support chatbot that retrieves answers from a company's private knowledge base using RAG.

Conversational Memory Management

So far, our chain treats every invocation as an isolated interaction. Send a message, get a response, and it's done. But real AI-powered conversational applications need to remember what was said before.

That's when Conversational Memory Management comes to rescue!

The Problem without Memory

chain.invoke({"question": "Hey, David here from Brazil."})
# Response: "Nice to meet you, David!"

chain.invoke({"question": "What's my home country?"})
# Response: "I don't know which country you are." ← The LLM has no clue :)

Adding Memory with Conversation History

LangChain provides memory through chat history management, which in our example is leveraged by InMemoryChatMessageHistory. The most practical approach is using RunnableWithMessageHistory, which automatically injects past messages into every new invocation:

from langchain_openai import ChatOpenAI
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory

# Store for session histories
session_store = {}

# Define the function to retrieve or create a session history
def get_session_history(session_id: str):
    if session_id not in session_store:
        session_store[session_id] = InMemoryChatMessageHistory()
    return session_store[session_id]

# Prompt now includes a placeholder for chat history
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant. Remember the conversation context."),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{question}")
])

llm = ChatOpenAI(model="gpt-4o")
parser = StrOutputParser()

chain = prompt | llm | parser

# Wrap the chain with memory management
chain = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="question",
    history_messages_key="chat_history"
)

# Session config to identify the conversation
config = {"configurable": {"session_id": "user_david_session"}}

# First message
response1 = chain.invoke(
    {"question": "Hey, David here from Brazil."},
    config=config # type: ignore
)
print(response1)  # "Nice to meet you, David!"

# Second message: now the LLM remembers
response2 = chain.invoke(
    {"question": "What's my home country?"},
    config=config # type: ignore
)
print(response2)  # "Based on what you told me, your home country is Brazil!"

Note: InMemoryChatMessageHistory stores conversations only in memory, so all chat history is lost when the application exits. For long-lived AI assistants, LangChain provides persistent implementations that store conversation history in external databases (PostgresChatMessageHistory, RedisChatMessageHistory) or storage services (DynamoDBChatMessageHistory, S3ChatMessageHistory), allowing users to resume previous conversations across application restarts.

Breaking Down The Code

The user sends a new message.
LangChain fetches the chat history for the given session_id.
The history is injected into the prompt via the MessagesPlaceholder.
The LLM receives the full conversation context and responds accordingly.
The new message and response are saved back to the history automatically.

Retrieval-Augmented Generation (RAG)

Even with memory, our LLM only knows what it was trained on. It has no access to our private documents, internal knowledge bases, or real-time data. RAG solves this by fetching relevant information and injecting it into the prompt before the LLM responds.

How RAG Works

 User Question
      ↓
  Retriever      (searches our documents for relevant chunks)
      ↓
Relevant Chunks  (injected into the prompt as context)
      ↓
Prompt Template  (combines question + context)
      ↓
     LLM         (answers based on the retrieved context)
      ↓
 Final Answer

RAG Implementation

Unlike chat models, RAG applications also require an embedding model capable of converting text into vector embeddings.

For this guide, I run locally a dedicated nomic-embed-text model on port 8001, exposing the standard OpenAI Embeddings API:

./llamafile \
  -m nomic-embed-text-v1.5.f16.gguf \
  --embedding \
  --server \
  --host 0.0.0.0 \
  --port 8001 \
  --alias nomic-embed-text \
  --rope-scaling yarn \
  --rope-freq-scale 0.75 \
  -c 8192 \
  -b 8192 \
  --ubatch-size 8192

This separation is a common production architecture, where specialized models are used for different tasks and runs in dedicated environment.

Step 1: Install dependencies

RAG requires two additional dependencies to be installed besides the packages introduced earlier in this guide (i.e. langchain and langchain-openai):

pip install langchain-community langchain-chroma

Chroma is an open-source vector database used to store embeddings and perform similarity searches over our documents

Step 2: Load and split a document

The first step is loading the document(s) that will serve as the application's knowledge base. Since LLMs have a limited context window, storing an entire document as a single block of text is inefficient and usually ineffective.

Instead, we split the document(s) into smaller overlapping chunks. During retrieval, the vector database can search these chunks individually and return only the most relevant pieces for the user's question.

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

# Load the target document
loader = TextLoader("company_credentials.txt")
documents = loader.load()

# Split into smaller chunks for better retrieval
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = text_splitter.split_documents(documents)

Step 3: Create a Vector Store

The next step is converting every text chunk into a numerical representation called embedding.

Embeddings capture the semantic meaning of the text, allowing similar pieces of information to be found even when different words are used.

Once the embeddings are generated, they are stored in a Chroma vector database so we can perform similarity searches over them later.

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

# Initialize the embedding model
embeddings = OpenAIEmbeddings(
    model="nomic-embed-text",
    base_url="http://localhost:8001/v1",
    check_embedding_ctx_length=False
)

# Generate embeddings and index them in Chroma (in-memory)
vector_store = Chroma.from_documents(documents=chunks, embedding=embeddings)

# Create a retriever
retriever = vector_store.as_retriever(search_kwargs={"k": 3})

The retriever will now be responsible for searching the vector database and returning the most relevant document chunks whenever a user asks a question.

Step 4: Build the RAG Chain

With the retriever in place, we can finally build our RAG pipeline.

When a user submits a question, LangChain first sends it to the retriever, which searches the vector store for the most relevant chunks. Those retrieved chunks are then injected into the prompt as contextual information before the prompt is sent to the language model.

Finally, the LLM generates an answer based on both the user's question and the retrieved context.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a helpful assistant. Answer the user's question based only on the context provided below. If the answer cannot be found in the context, say that you don't know.

     Context: {context}
     """),
    ("human", "{question}")
])

llm = ChatOpenAI(model="gpt-4o")
parser = StrOutputParser()

# Set up the RAG chain
chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
    | parser
)

# Ask a question with information only found in the ingested document
response = chain.invoke("What are my login credentials for Jira?")
print(response)

Notice that the retriever is automatically executed as part of the chain. The retrieved document chunks populate the {context} placeholder, while the original user question is passed unchanged to {question} through RunnablePassthrough().

The internal RAG workflow:

Company Handbook
       ↓
 Text Splitter
       ↓
     Chunks
       ↓
   Embeddings
       ↓
Chroma Vector Store
       ↑ (similarity search)
 User Question
       ↓
Retrieved Chunks
       ↓
Prompt Template
       ↓
      LLM
       ↓
  Final Output

Why RAG Matters?

Without RAG	With RAG
LLM answers from training data only	LLM answers from our private documents
Knowledge is frozen at training cutoff	Knowledge can be updated anytime
May hallucinate missing information	Grounds answers in real retrieved content
One-size-fits-all responses	Responses tailored to our specific data

Tool Calling and Agents

The final and most powerful building block is Tool Calling. This is what transforms a passive LLM into an active Agent, i.e. one that can take actions in the real world.

What is Tool Calling?

Tool Calling allows the LLM to:

Recognize that it needs external information or capability to answer a question.
Select the right tool from a set of available tools.
Call the tool with the correct arguments.
Use the tool's response to formulate the final answer.

Defining a Tool

In LangChain, any Python function can become a tool using the @tool decorator.

Imagine we are building a minimal personal AI assistant with two capabilities: telling the current time and identifying the operating system it's running on.

import platform
from datetime import datetime
from langchain_core.tools import tool

@tool
def get_current_time() -> str:
    """Return the current local date and time."""
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

@tool
def get_operating_system() -> str:
    """Returns the name and architecture of the host operating system.

    Use this tool whenever the user asks about their system environment, 
    OS name, computer platform, or machine architecture.
    """
    os_name = platform.system()
    architecture = platform.machine()

    return f"Operating System: {os_name}, Architecture: {architecture}"

Although these tools are simple, they illustrate an important concept: the LLM does not execute business logic itself. Instead, it decides when a tool should be called and what arguments should be passed.

Building an Agent with Tools

Now let's give the LLM access to these tools.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain.agents import create_tool_calling_agent, AgentExecutor

tools = [get_current_time, get_operating_system]

# Initialize the model
llm = ChatOpenAI(model="gpt-4o")

# Define the prompt structure
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant. Use tools whenever they are useful."),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

# Build the agent and executor
agent = create_tool_calling_agent(llm=llm, tools=tools, prompt=prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# Example usage
input = "What time is it right now?"
response = agent_executor.invoke({"input": input})
print(f"{input}:", response["output"])

input = "What is today's date?"
response = agent_executor.invoke({"input": input})
print(f"{input}:", response["output"])

input = "What platform is the user running on?"
response = agent_executor.invoke({"input": input})
print(f"{input}:", response["output"])

Output (with verbose=True, the execution flow becomes visible):

> Entering new AgentExecutor chain...

Invoking: `get_current_time` with `{}`


2026-07-14 21:51:01The current time is 21:51:01 on July 14, 2026.

> Finished chain.
What time is it right now?: The current time is 21:51:01 on July 14, 2026.


> Entering new AgentExecutor chain...

Invoking: `get_current_time` with `{}`


2026-07-14 21:51:03The current date is July 14, 2026.

> Finished chain.
What is today's date?: The current date is July 14, 2026.


> Entering new AgentExecutor chain...

Invoking: `get_operating_system` with `{}`


Operating System: Linux, Architecture: x86_64The user is running on **Linux** with an **x86_64** architecture.

> Finished chain.
What platform is the user running on?: The user is running on **Linux** with an **x86_64** architecture.

Notice that the LLM never executes the Python functions directly. It only determines which tool should be called and what arguments should be passed. LangChain is responsible for executing the functions and returning their results to the model.

The Agent Loop

What makes agents powerful is their reasoning loop:

         User Request
              ↓
      +----------------+
      |      LLM       |
      +----------------+
              │
      "Do I need a tool?"
      Yes ↓       ↓ No
Select a Tool   Final Output
          ↓
  Execute Tool
          ↓
  Tool Returns Result
          ↓
      +----------------+
      |      LLM       |
      +----------------+
              │
    "Do I need another tool?"

This loop continues until the LLM determines it has enough information to answer the user's question.

Next Steps

Now that we understand the foundational building blocks of LangChain, here are some natural next steps:

LangGraph: A framework built on top of LangChain for building stateful, multi-agent workflows as graphs.
LangSmith: LangChain's observability platform for debugging and monitoring our chains and agents in production.
LangChain Templates: Pre-built application templates to accelerate development.
LangChain Integrations Hub: Explore the hundreds of available integrations for document loaders, vector stores, tools, and LLM providers.

LangChain is a rapidly evolving ecosystem. The best way to master it is to build something real. Start small, add one building block at a time, and let complexity grow naturally as our application demands it.

Wrap Up

Throughout this guide, we explored LangChain from the ground up, building our understanding one component at a time.

Here is a summary of everything we covered:

Component	Purpose	Key Takeaway
LLM Integration	Connect to a language model	One unified API for all LLM providers
Prompt Templates	Structure and reuse prompts	Separate prompt logic from application logic
Output Parsers	Structure the LLM's response	Convert raw text into usable data formats
LCEL	Compose components into pipelines	The `ǀ` operator connects everything together
Memory	Persist conversation history	Give the LLM context across multiple turns
RAG	Connect the LLM to private data	Ground responses in our own documents
Tool Calling / Agents	Give the LLM the ability to act	Let the LLM decide which tools to use and when

All the Python examples presented in this guide are available in this accompanying GitHub repository, making it easy to run, modify, and experiment with each example on your own.

Happy AI coding 🚀

DEV Community: David Archanjo

Chat with Your Docs: Working with LangChain, RAG, MCP, ChromaDB Made Simple

What We Will Build

About Me

Table Of Contents

Background

Why RAG and MCP?

The Case for RAG

The Case for MCP

Step 1: Setting Up the Environment

Project Structure

Installing Dependencies

Environment Configuration

Breaking down each variable:

Local OpenAI-Compatible API Service

The Configuration Class

Step 2: Loading and Indexing Documents into ChromaDB

Breaking Down the Code

Ingestion Server Startup

Step 3: Exposing MCP Tools

Breaking Down the Code

Running the MCP Server

Testing the MCP Server

Step 4: Building the RAG Chain

Breaking Down the Code

Step 5: Running and Testing the Q&A Agent

Starting All Services

Uploading Sample Documents

Sample Questions for Testing the Knowledge Base

General questions

assets/company_credentials.txt:

assets/meeting_notes.md

assets/resume.pdf

assets/car_prices.csv

Agent Example Interaction

Verifying Retrieval Quality

The Full Execution Flow

Wrapping Up

Building a Speech-to-Speech AI Chatbot with Python, LangChain, Whisper, and Kokoro

About Me

Table of Contents

Introduction

Motivation

What We Will Build

Prerequisites

Common Terms and Abbreviations

API Key Setup

Architecture Overview

Component Responsibilities

Project Setup

Application Dependencies (requirements.txt)

Installing Dependencies

System Dependencies

Checking eSpeak-NG Installation

Downloading Model Files

Configuration

The .env File

The config.py File

Why This Structure

Speech Recognition with Whisper

The Assistant Class

How Voice Activity Detection Works

Initializing the Asistant

Filtering Transcription Output

Choosing a Whisper Model Size

Conversational AI with LangChain

Initializing the Language Model

Defining the Prompt

Assembling the Chain

Invoking the Chain

Persistent Conversation History

SQLChatMessageHistory

Wrapping the Chain

The four arguments:

Session Management

Execution Flow with History

Text-to-Speech with Kokoro

Initializing Kokoro

Generating and Playing Speech

Voice Options

`assets/company_credentials.txt`:

`assets/meeting_notes.md`

`assets/resume.pdf`

`assets/car_prices.csv`

Application Dependencies (`requirements.txt`)

The `.env` File

The `config.py` File

The `Assistant` Class

Initializing the A`sistant`