Building agentic systems with LangGraph: ReAct agents and RAG.

In the agentic AI post we built an agent loop by hand: a while loop, a model, some tools. That works, and it's the right way to learn the mechanics — but real systems quickly outgrow it. You want persistent conversation state, streaming, branching logic, retries, human approval steps, and the ability to actually see what your agent did. LangGraph is a framework built for exactly this. In this post we'll build two of the most useful patterns with it: a ReAct agent with tools and memory, and a RAG system — first as a fixed pipeline, then as an agent that decides for itself when to retrieve.

The core idea: agents as graphs

LangGraph models an LLM application as a state machine. Three concepts cover almost everything:

• State — a typed dictionary that flows through the system. For a chat agent it's usually just the message history; for a pipeline it can be anything (question, retrieved documents, draft answer, …).
• Nodes — plain Python functions. Each takes the current state and returns a partial update to it. A node might call the model, run a tool, or execute ordinary code.
• Edges — the wiring between nodes. Edges can be fixed ("after retrieval, always generate") or conditional ("if the model requested a tool, go to the tool node; otherwise finish"). Conditional edges are where agency lives: a cycle between a model node and a tool node is the agent loop.

Compared to the hand-rolled loop, the payoff is structure you don't have to build yourself: every step is checkpointable (so conversations persist and can be resumed), streamable, interruptible (pause for human approval, then continue), and inspectable. The graph is the architecture diagram, except it runs.

Setup

pip install langgraph langchain-anthropic langchain-huggingface langchain-text-splitters

We'll use Claude as the model via the langchain-anthropic integration (set ANTHROPIC_API_KEY in your environment). Everything below is model-agnostic, though — swap in any chat model with tool-calling support, including a local one served by Ollama via langchain-ollama.

Part 1: a ReAct agent from scratch

ReAct (Reason + Act) is the canonical agent pattern: the model alternates between reasoning about what to do and acting through tools, folding each tool result into its next reasoning step. In graph terms it's beautifully small — two nodes and one conditional edge:

START ──▶ agent ──(tool calls?)──▶ tools ──▶ agent ──▶ ... ──▶ END

First, the tools. LangGraph uses LangChain's @tool decorator — the function's signature and docstring become the schema the model sees, so write the docstring as if you were explaining the tool to a new colleague:

from langchain_core.tools import tool

@tool
def multiply(a: float, b: float) -> float:
    """Multiply two numbers and return the result."""
    return a * b

@tool
def get_exchange_rate(currency: str) -> str:
    """Get the current exchange rate from EUR to the given currency code (e.g. 'USD')."""
    rates = {"USD": 1.09, "GBP": 0.85, "JPY": 168.2}   # in real life: call an API
    return f"1 EUR = {rates.get(currency.upper(), 'unknown')} {currency.upper()}"

tools = [multiply, get_exchange_rate]

Now the graph itself:

from langchain_anthropic import ChatAnthropic
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode, tools_condition

llm = ChatAnthropic(model="claude-opus-4-8", max_tokens=1024)
llm_with_tools = llm.bind_tools(tools)

# The "agent" node: one model call on the running message history
def call_model(state: MessagesState):
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}   # partial update: appended, not replaced

builder = StateGraph(MessagesState)
builder.add_node("agent", call_model)
builder.add_node("tools", ToolNode(tools))     # executes whatever tools the model requested

builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition)  # tool calls? → "tools", else → END
builder.add_edge("tools", "agent")             # results flow back to the model

graph = builder.compile()

A few things worth pausing on. MessagesState is a prebuilt state schema with a single messages field whose update rule is "append" — that's why call_model returns a one-message list rather than the whole history. ToolNode reads the tool calls from the last model message, executes the matching Python functions, and appends the results as tool messages. And tools_condition is the entire ReAct decision: it routes to the tool node if the model asked for tools, and to END when it produced a final answer.

Run it:

result = graph.invoke(
    {"messages": [{"role": "user", "content":
        "What is 2500 EUR in USD, and what is that squared?"}]}
)
print(result["messages"][-1].content)

Behind that one call, the model looked up the exchange rate, multiplied, called multiply again to square the result, and composed an answer — four model turns and three tool executions, none of which you orchestrated.

Adding memory (checkpointing)

As written, the graph forgets everything between calls. LangGraph's checkpointer saves the state after every node, keyed by a thread_id — which gives you persistent multi-turn conversations for free:

from langgraph.checkpoint.memory import InMemorySaver

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "user-42"}}

graph.invoke({"messages": [{"role": "user", "content": "Convert 100 EUR to GBP."}]}, config)
result = graph.invoke({"messages": [{"role": "user", "content": "And to JPY?"}]}, config)
# "And to JPY?" is understood — the thread's history was restored from the checkpoint

In production you'd swap InMemorySaver for a database-backed checkpointer (SQLite and Postgres implementations ship as packages) — same code, durable state. Checkpointing is also what enables human-in-the-loop interrupts: you can compile the graph to pause before the tool node, inspect the pending tool call, and resume.

If you don't need to see the wiring, the prebuilt one-liner builds this exact graph for you: create_agent(model, tools) (from langchain.agents; older LangGraph versions call it create_react_agent). Knowing the explicit version pays off the moment you need to customize — which, in my experience, is always.

Part 2: RAG as a graph

Retrieval-augmented generation (RAG) grounds the model's answers in your own documents: retrieve the passages relevant to a question, put them in the prompt, and instruct the model to answer from them. It's the standard cure for two LLM diseases at once — hallucination, and ignorance of anything private or recent.

First we need an indexed document collection. This part is plain LangChain, no graph yet:

from langchain_core.documents import Document
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

docs = [
    Document(page_content=open("handbook.md", encoding="utf-8").read(),
             metadata={"source": "handbook.md"}),
    # ... in practice: load PDFs, wikis, tickets with document loaders
]

# Split into overlapping chunks — retrieval works on chunks, not whole files
chunks = RecursiveCharacterTextSplitter(
    chunk_size=1000, chunk_overlap=200
).split_documents(docs)

# Embed the chunks and index them (local, free embedding model)
embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")
vector_store = InMemoryVectorStore(embeddings)
vector_store.add_documents(chunks)

(For anything beyond a demo, use a persistent vector store — Chroma, Qdrant, pgvector — via the same interface. The chunking parameters matter more than people expect: chunks need to be small enough to be specific, large enough to be self-contained.)

Now the pipeline as a two-node graph, with a custom state instead of MessagesState — this is where LangGraph's "state is whatever you say it is" design shines:

from typing_extensions import TypedDict

class RAGState(TypedDict):
    question: str
    context: list      # retrieved Documents
    answer: str

def retrieve(state: RAGState):
    docs = vector_store.similarity_search(state["question"], k=4)
    return {"context": docs}

def generate(state: RAGState):
    context_text = "\n\n".join(
        f"[{d.metadata.get('source', '?')}]\n{d.page_content}" for d in state["context"]
    )
    prompt = (
        "Answer the question using ONLY the context below. "
        "If the context is not sufficient, say so.\n\n"
        f"Context:\n{context_text}\n\nQuestion: {state['question']}"
    )
    response = llm.invoke(prompt)
    return {"answer": response.content}

builder = StateGraph(RAGState)
builder.add_node("retrieve", retrieve)
builder.add_node("generate", generate)
builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "generate")
builder.add_edge("generate", END)
rag_graph = builder.compile()

result = rag_graph.invoke({"question": "What is our parental leave policy?"})
print(result["answer"])
for doc in result["context"]:
    print("source:", doc.metadata["source"])   # citations come free with the state

Notice what keeping context in the state buys you: the retrieved documents come back alongside the answer, so showing sources to the user — the single best trust feature a RAG system can have — is a loop over state, not extra plumbing.

A fixed graph like this is a workflow, not an agent — and per the rule from the agents post, that's a feature: it always retrieves exactly once, costs the same every time, and is trivial to evaluate. Make it more sophisticated only when the failure cases demand it. Which brings us to…

Part 3: agentic RAG

The fixed pipeline has two blind spots. It retrieves even when the question doesn't need it ("summarize your last answer"), and it retrieves once even when the question needs several searches ("compare the 2024 and 2025 policies"). The fix is to hand the retriever to the ReAct agent from Part 1 as just another tool — the model then decides whether, when, and how often to search, and can rewrite its query when the first attempt comes back empty:

@tool
def search_handbook(query: str) -> str:
    """Search the company handbook for passages relevant to the query.
    Use a short, specific search phrase, not a full sentence."""
    docs = vector_store.similarity_search(query, k=4)
    if not docs:
        return "No results. Try a different phrasing."
    return "\n\n".join(f"[{d.metadata.get('source', '?')}]\n{d.page_content}" for d in docs)

llm_with_tools = llm.bind_tools([search_handbook])

def call_model(state: MessagesState):
    system = ("You answer questions about the company handbook. "
              "Search before answering anything factual; cite the sources you used. "
              "If the handbook doesn't cover it, say so.")
    response = llm_with_tools.invoke(
        [{"role": "system", "content": system}] + state["messages"]
    )
    return {"messages": [response]}

builder = StateGraph(MessagesState)
builder.add_node("agent", call_model)
builder.add_node("tools", ToolNode([search_handbook]))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition)
builder.add_edge("tools", "agent")
agentic_rag = builder.compile(checkpointer=InMemorySaver())

Same graph shape as Part 1 — that's the point. "Agentic RAG" isn't a new architecture, it's the ReAct pattern with retrieval as a tool. On a multi-part question you'll now see the agent search two or three times with different queries before answering; on small talk it won't search at all.

The trade-offs cut both ways, and it's worth being explicit about them:

A pattern that serves production systems well: pipeline RAG as the default path, agentic RAG for queries the pipeline flags as complex. My NeuroLex system — multi-agent RAG over a century of neuroscience literature — is built on exactly these ingredients, with separate agents for query understanding, retrieval, and answer composition.

Where to go from here

• Streaming — graph.stream(..., stream_mode="values") yields state after each step, so users watch the agent work instead of staring at a spinner.
• Human-in-the-loop — compile with interrupt_before=["tools"] to pause for approval before any tool runs; resume from the checkpoint after sign-off.
• Subgraphs and multi-agent — a compiled graph can be a node in a bigger graph, which is how orchestrator/specialist architectures are assembled.
• Observability — trace your graphs (LangSmith or any OpenTelemetry-compatible backend) from day one; debugging an agent without traces is archaeology.
• Evaluation — for RAG especially: build a small set of question/expected-answer pairs early, and measure retrieval quality separately from answer quality. Most RAG failures are retrieval failures.

All posts