Chapter 10 · Deep Dive: LangChain & LangGraph

Part of Part III · Deep Dives

LangChain is the React of AI -- ubiquitous, powerful, and easy to misuse. This chapter teaches you to wield it without getting burned.

It is the most popular agent framework in the world. It is also the most criticized. Developers love it for getting started and hate it for getting serious. The paradox has a simple explanation: LangChain makes easy things easier and hard things harder.

This chapter is a map. You will learn which parts of the ecosystem earn their keep, which to avoid, and how to use LangGraph -- the real gem in the stack. By the end, you will have a complete research agent and know exactly when to reach for a framework and when to stay raw.

This is not a tutorial. The docs run over a thousand pages. You need the 20% that does 80% of the work. That is what this chapter delivers: strategic guidance, working code, and honest assessments of where LangChain shines and where it burns you.

Section 1: LangChain -- What to Use, What to Skip

The LangChain ecosystem is not one library. It is a family of packages, and understanding which is which is the first step to using them effectively.

langchain-core: The foundation. Runnable interface, prompt templates, output parsers, tool definitions, LCEL primitives. The only package you absolutely need. Well-maintained and stable.

langchain-community: A graveyard of integrations. Hundreds of connectors -- most volunteer-maintained, many out of date, some broken. Use it for document loaders and text splitters. Be skeptical of everything else.

langchain-experimental: Exactly what it sounds like. Do not put this in production.

LangGraph: A state machine framework for agent workflows. This is the good part. Not "LangChain but for graphs" -- a fundamentally different abstraction that models agents as explicit state machines. Most of this chapter is about LangGraph.

LangSmith: SaaS platform for tracing, monitoring, and evaluation. Not required, but makes debugging dramatically easier. Think Datadog for your agents.

Here is the rule of thumb this chapter operates under:

Use: LCEL, tool definitions (@tool decorator), document loaders, text splitters, LangGraph, LangSmith for tracing.

Skip: The old Chain API (LLMChain, ConversationChain, SequentialChain -- all deprecated patterns), most of langchain-community (unmaintained integrations), agents in langchain.agents (use LangGraph instead), and anything in langchain-experimental for production.

The key insight: LangChain's value is in its composability, not its high-level abstractions. The old Chain API tried to hide complexity behind magic class names. It failed. LCEL and LangGraph succeed because they make the composition explicit. You can see how data flows. You can debug it. You can reason about it.

The Old Way vs. The New Way

Here is the old Chain API -- the pattern you should never write:

# DO NOT WRITE THIS -- deprecated pattern
from langchain.chains import LLMChain
from langchain.llms import OpenAI

chain = LLMChain(
    llm=OpenAI(),
    prompt="Answer this question: {question}"
)
result = chain.run(question="What is agentic AI?")

This looks simple. It is not. LLMChain is a black box. You cannot see how the prompt is constructed. You cannot easily add a parser. You cannot stream the output. You cannot compose it with other components without learning a separate SequentialChain API. The abstraction hides too much and composes too poorly.

Here is the same thing with LCEL:

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

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{question}")
])
model = ChatOpenAI(model="gpt-4o")
chain = prompt | model | StrOutputParser()

result = chain.invoke({"question": "What is agentic AI?"})

More lines, less magic. Every component is explicit. The pipe operator shows you exactly how data flows: prompt to model to parser. You can swap any component without changing the rest. You can add a retriever, a tool, or a conditional branch by inserting another pipe. This is composability that scales.

Section 2: LCEL -- The Good Part of LangChain

LCEL is LangChain Expression Language. It is a declarative way to compose LLM calls, tools, retrievers, and parsers using the | (pipe) operator. If you use one thing from LangChain, use LCEL.

The Pipe Operator

Every LCEL component is a Runnable. A Runnable is anything that can be invoked with input and produces output. Prompts are Runnables. Models are Runnables. Output parsers are Runnables. The pipe operator chains them together:

chain = prompt | model | output_parser

Data flows left to right. The prompt receives your input dictionary and produces a list of formatted messages. The model receives those messages and produces an AIMessage. The output parser receives the AIMessage and produces a string. Each component's output becomes the next component's input.

This is not syntactic sugar. It is a protocol. Every Runnable supports the same interface:

# Synchronous invocation
result = chain.invoke({"input": "Hello"})

# Batch invocation -- runs in parallel
results = chain.batch([
    {"input": "Hello"},
    {"input": "What is AI?"},
    {"input": "Write a haiku"}
])

# Streaming -- yields tokens as they arrive
for chunk in chain.stream({"input": "Tell me a story"}):
    print(chunk, end="", flush=True)

# Async variants
result = await chain.ainvoke({"input": "Hello"})
async for chunk in chain.astream({"input": "Tell me a story"}):
    print(chunk, end="", flush=True)

Every Runnable supports invoke, batch, stream, and their async counterparts. This uniformity is LCEL's superpower. You can take any chain, wrap it in another chain, and the interface stays the same. A complex RAG pipeline with retrieval, reranking, and structured output has the same .invoke() signature as a simple prompt-to-model chain.

The Building Blocks

LCEL gives you three primitives for shaping data flow:

RunnablePassthrough passes data through unchanged. Use it when you need to forward input to multiple downstream components or when a step in your chain should not modify the data:

from langchain_core.runnables import RunnablePassthrough

chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

Here, RunnablePassthrough() forwards the user's question unchanged while retriever fetches relevant documents. Both values are assembled into a dictionary and passed to the prompt.

RunnableLambda wraps any Python function as a Runnable. Use it for custom transformations that do not fit into a pre-built component:

from langchain_core.runnables import RunnableLambda

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

chain = (
    {"context": retriever | RunnableLambda(format_docs), "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

RunnableParallel executes multiple Runnables concurrently and merges their outputs. Use it when you need to fetch data from multiple sources at once:

from langchain_core.runnables import RunnableParallel

# Fetch from two retrievers in parallel
parallel_retrieval = RunnableParallel(
    web_results=web_retriever,
    doc_results=doc_retriever
)

chain = parallel_retrieval | prompt | model | StrOutputParser()

A Complete RAG Chain with LCEL

Here is a retrieval-augmented generation chain built entirely with LCEL. It loads documents, splits them, creates embeddings, stores them in a vector database, retrieves relevant chunks, and generates an answer -- all composed with pipes:

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import Chroma

# 1. Load and split documents
loader = TextLoader("knowledge_base.txt")
documents = loader.load()
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = text_splitter.split_documents(documents)

# 2. Create vector store and retriever
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# 3. Build the RAG prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", """Answer the question based on the provided context.
If the context does not contain the answer, say so.

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

# 4. Compose the chain
model = ChatOpenAI(model="gpt-4o")

def format_docs(docs):
    return "\n\n---\n\n".join(doc.page_content for doc in docs)

rag_chain = (
    {"context": retriever | RunnableLambda(format_docs), "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

# 5. Use it
answer = rag_chain.invoke("What is the company's refund policy?")
print(answer)

This is about 40 lines of code. It loads documents, creates embeddings, builds a retriever, and answers questions with grounded responses. Every component is swappable. Want to use Anthropic instead of OpenAI? Change the model. Want to use Pinecone instead of Chroma? Change the vector store. The chain structure stays the same.

Adding Tool Calling to an LCEL Chain

LCEL chains can call tools. Here is a chain that can search the web when it needs information it does not have:

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

@tool
def web_search(query: str) -> str:
    """Search the web for current information. Use for facts beyond your knowledge cutoff."""
    # In production, this would call a real search API
    return f"Search results for '{query}': [simulated results]"

model = ChatOpenAI(model="gpt-4o").bind_tools([web_search])

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a research assistant. Use web_search when you need current information."),
    ("user", "{question}")
])

chain = prompt | model

result = chain.invoke({"question": "What is the current price of Bitcoin?"})
print(result.tool_calls)
# [{'name': 'web_search', 'args': {'query': 'current Bitcoin price 2026'}, 'id': '...'}]

The .bind_tools() method attaches tool definitions to the model. When the model decides a tool is needed, it returns a response with tool_calls instead of content. Your code executes the tool and feeds the result back. This is the foundation of tool-using agents, and LCEL makes it a one-liner.

Section 3: LangChain Tools and Agents

LangChain provides a @tool decorator that converts any Python function into a tool definition the model can understand. It is one of the few LangChain abstractions that genuinely reduces boilerplate.

Defining Tools

from langchain_core.tools import tool
from pydantic import BaseModel, Field

# Simple tool -- LangChain infers the schema from type hints
@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression. Supports +, -, *, /, **, and parentheses."""
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return str(result)
    except Exception as e:
        return f"Error: {e}"

# Complex tool -- explicit Pydantic schema for the arguments
class SendEmailInput(BaseModel):
    to: str = Field(description="Recipient email address")
    subject: str = Field(description="Email subject line")
    body: str = Field(description="Email body text")

@tool(args_schema=SendEmailInput)
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email. Use only when explicitly asked to send an email."""
    # In production, call your email service here
    return f"Email sent to {to} with subject '{subject}'"

# Async tool
@tool
async def query_database(sql: str) -> str:
    """Run a read-only SQL query against the company database."""
    # In production, execute the query
    return "[query results]"

The @tool decorator does three things. It reads the function's docstring and uses it as the tool description -- this is what the model sees when deciding whether to call the tool. It reads the function's type hints and generates a JSON schema for the arguments. It wraps the function so it can be called by LangChain's tool-execution infrastructure.

The AgentExecutor Pattern (Legacy, But You Will See It)

Before LangGraph, LangChain's standard way to build agents was AgentExecutor. It is still common in tutorials and older codebases. You should recognize it, but you should not build new things with it:

# Legacy pattern -- recognize it, do not build with it
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_openai import ChatOpenAI

agent = create_openai_functions_agent(
    llm=ChatOpenAI(model="gpt-4o"),
    tools=[calculate, web_search],
    prompt=prompt
)
agent_executor = AgentExecutor(
    agent=agent,
    tools=[calculate, web_search],
    verbose=True,
    max_iterations=10
)
result = agent_executor.invoke({"input": "What is 15% of 87?"})

AgentExecutor works. It has a loop, tool execution, and error handling. But it is a black box. You cannot easily add human-in-the-loop approval. You cannot checkpoint state. You cannot branch conditionally based on custom logic. You cannot compose agents within agents. These are not edge cases -- they are requirements for production agents. LangGraph solves all of them.

When to Use LangChain Tools vs. Defining Your Own

LangChain provides hundreds of pre-built tool integrations: SQL databases, GitHub, Gmail, Slack, Jira, and more. They are convenient. They are also heavy. Each integration pulls in its own dependencies, and many are thinly maintained.

The rule: use LangChain tools for complex integrations where the boilerplate is genuinely painful (SQL with connection pooling, GitHub with OAuth). Define your own tools for everything else. A tool is just a function with a description. You do not need a framework for that.

# Your own tool -- zero LangChain dependency
def search_knowledge_base(query: str) -> str:
    """Search the internal knowledge base for relevant documents."""
    results = vectorstore.similarity_search(query, k=3)
    return "\n\n".join(r.page_content for r in results)

# Wrap it with the @tool decorator if you want LangChain compatibility
search_tool = tool(search_knowledge_base)

The @tool decorator is worth using. The pre-built tool integrations are worth evaluating case by case. The AgentExecutor is worth migrating away from.

Section 4: LangGraph -- The Real Power

LangGraph is a state machine framework for building agent workflows. It is the answer to the question: "How do I build an agent that does more than call tools in a loop?"

Why LangGraph Matters

Agents are not linear. A real agent branches: "Should I search again, or do I have enough information?" It loops: "That tool call failed. Retry with different parameters." It waits: "I need human approval before I send this email." It recovers: "The API returned an error. Let me try the fallback endpoint."

A linear chain cannot model this. A black-box executor cannot model this. You need a graph -- nodes for actions, edges for transitions, state that flows through the whole thing. LangGraph gives you exactly that.

Core Concepts

LangGraph has four core concepts. Understand these and you understand the framework:

State is a typed dictionary that flows through the graph. It holds everything the agent knows and has done: messages, intermediate results, tool outputs, flags. State is the agent's working memory. You define it as a TypedDict or a Pydantic model.

Nodes are functions that read state and return state updates. A node can be anything: an LLM call, a tool execution, a human input step, a data transformation. Nodes are the verbs in your graph.

Edges are transitions between nodes. An edge can be unconditional (always go from A to B) or conditional (go to B, C, or D based on the current state). Edges are the control flow.

The Graph is the compiled state machine. You add nodes, connect them with edges, set an entry point, and compile. The result is a Runnable that you can invoke, stream, and debug.

Your First LangGraph Agent

Here is a complete agent built with LangGraph. It calls an LLM, executes tools, and routes between thinking and acting until the task is done:

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
import json

# --- State ---
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    # add_messages appends new messages instead of replacing the list

# --- Tools ---
@tool
def web_search(query: str) -> str:
    """Search the web for current information."""
    return f"Search results for '{query}': [simulated results]"

@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression."""
    try:
        return str(eval(expression, {"__builtins__": {}}, {}))
    except Exception as e:
        return f"Error: {e}"

tools = [web_search, calculate]
tool_map = {t.name: t for t in tools}

# --- Model ---
model = ChatOpenAI(model="gpt-4o").bind_tools(tools)

# --- Nodes ---
def call_model(state: AgentState) -> dict:
    """Call the LLM with the current message history."""
    response = model.invoke(state["messages"])
    return {"messages": [response]}

def execute_tools(state: AgentState) -> dict:
    """Execute any tool calls in the last AI message."""
    last_message = state["messages"][-1]
    tool_messages = []

    for tool_call in last_message.tool_calls:
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]
        print(f"  Calling tool: {tool_name}({tool_args})")

        tool_func = tool_map[tool_name]
        result = tool_func.invoke(tool_args)

        tool_messages.append(
            ToolMessage(content=str(result), tool_call_id=tool_call["id"])
        )

    return {"messages": tool_messages}

# --- Routing ---
def should_continue(state: AgentState) -> Literal["tools", "end"]:
    """Decide: call tools, or end the conversation?"""
    last_message = state["messages"][-1]

    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"

    return "end"

# --- Build the Graph ---
graph = StateGraph(AgentState)

graph.add_node("agent", call_model)
graph.add_node("tools", execute_tools)

graph.set_entry_point("agent")

graph.add_conditional_edges(
    "agent",
    should_continue,
    {
        "tools": "tools",
        "end": END
    }
)
graph.add_edge("tools", "agent")  # After tools, always go back to the agent

app = graph.compile()

# --- Run It ---
result = app.invoke({
    "messages": [HumanMessage(content="What is 15% of 87? Also, what is the current Bitcoin price?")]
})

for msg in result["messages"]:
    if hasattr(msg, "content") and msg.content:
        print(f"{msg.type.upper()}: {msg.content[:200]}...")

This is about 80 lines. It is explicit. You can see every node, every edge, every decision point. You can add a new node -- say, a human approval step -- by inserting it between agent and tools. You can add a new conditional edge -- say, a retry counter that gives up after three failed tool calls -- by modifying should_continue. The graph is your control flow, and you control it completely.

What the Graph Looks Like

    +---------+       +---------+
    |  agent  |<------|  tools  |
    +---------+       +---------+
         |                 ^
         |                 |
    [should_continue?]     |
         |                 |
    "tools" --------------+
         |
    "end"
         |
        END

The agent node calls the LLM. If the LLM requests a tool, the graph routes to the tools node, which executes the tool and routes back to the agent. The agent sees the tool result and decides whether to call another tool or respond to the user. This loop continues until the agent produces a response without tool calls, at which point the graph ends.

This is the same agent loop you built in Chapter 4. The difference is that LangGraph makes the loop explicit, typed, and debuggable. You are not parsing strings to detect termination. You are not manually managing message history. You are building a state machine, and the framework handles the plumbing.

Section 5: LangGraph Advanced Patterns

The basic agent graph is useful. The advanced patterns are where LangGraph earns its place in your stack.

Human-in-the-Loop

Some actions should not happen without human approval. Sending an email. Making a purchase. Deleting data. LangGraph supports this with interrupts -- points in the graph where execution pauses and waits for human input:

from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt

def sensitive_action(state: AgentState) -> dict:
    """Perform an action that requires human approval."""
    last_message = state["messages"][-1]
    action_details = last_message.tool_calls[0]["args"]

    # Pause and ask for approval
    approval = interrupt(f"Approve this action?\n\n{json.dumps(action_details, indent=2)}")

    if approval.get("approved"):
        # Execute the action
        result = execute_sensitive_action(action_details)
        return {"messages": [ToolMessage(content=result, tool_call_id=last_message.tool_calls[0]["id"])]}
    else:
        return {"messages": [ToolMessage(content="Action rejected by user.", tool_call_id=last_message.tool_calls[0]["id"])]}

# Compile with a checkpointer -- required for interrupts
graph = StateGraph(AgentState)
# ... add nodes and edges ...
app = graph.compile(checkpointer=MemorySaver())

# Run until the interrupt
config = {"configurable": {"thread_id": "session-123"}}
events = app.stream(
    {"messages": [HumanMessage(content="Send $500 to bob@example.com")]},
    config
)

# The graph pauses at the interrupt. You inspect the state, then resume:
for event in events:
    if "interrupt" in event:
        print(f"Graph interrupted: {event['interrupt']}")
        # Human reviews and approves
        app.invoke(None, config)  # Resume with no new input (approval was given)

The interrupt() function is the key. It pauses graph execution and surfaces a message to the human operator. The operator inspects the state, makes a decision, and resumes the graph. The graph continues from exactly where it stopped, with all state intact.

Checkpointing and Replay

Every LangGraph graph can be compiled with a checkpointer. The checkpointer saves the graph state after every step -- every node execution, every edge transition. This enables three powerful capabilities:

Resume from failure. If your graph crashes on step 7, you do not restart from step 1. You fix the bug and resume from step 6.

Debug by replaying. You can load a past execution and step through it node by node, inspecting state at every point. This is dramatically faster than adding print statements.

Time travel. You can rewind to any previous state and explore alternative paths. What if the agent had chosen a different tool? What if it had searched with different keywords?

from langgraph.checkpoint.memory import MemorySaver

# Compile with checkpointer
app = graph.compile(checkpointer=MemorySaver())

config = {"configurable": {"thread_id": "session-456"}}

# Run the graph
result = app.invoke(
    {"messages": [HumanMessage(content="Research quantum computing advances in 2026")]},
    config
)

# Inspect the checkpointed state at each step
checkpointer = MemorySaver()
state_history = list(checkpointer.list(config))
for state in state_history:
    print(f"Step {state.metadata['step']}: {state.metadata['source']}")
    # Inspect messages, tool calls, intermediate results

# Resume from a specific checkpoint
app.invoke(None, config)  # Continues from the last saved state

The checkpointer is pluggable. MemorySaver is for development. For production, use SqliteSaver (local) or PostgresSaver (server). The interface is the same.

Subgraphs

Complex agents contain sub-agents. A research agent might contain a search sub-agent that handles query formulation, result extraction, and relevance filtering. LangGraph models this with subgraphs -- graphs compiled as nodes within a parent graph:

# Define a sub-agent for web research
def build_search_agent():
    search_graph = StateGraph(SearchState)
    search_graph.add_node("formulate_query", formulate_query)
    search_graph.add_node("execute_search", execute_search)
    search_graph.add_node("extract_results", extract_results)
    search_graph.add_edge("formulate_query", "execute_search")
    search_graph.add_edge("execute_search", "extract_results")
    search_graph.add_edge("extract_results", END)
    search_graph.set_entry_point("formulate_query")
    return search_graph.compile()

# Use it as a node in the parent graph
parent_graph = StateGraph(ParentState)
parent_graph.add_node("research_plan", plan_research)
parent_graph.add_node("search", build_search_agent())  # Subgraph as a node
parent_graph.add_node("synthesize", synthesize_results)
parent_graph.add_edge("research_plan", "search")
parent_graph.add_edge("search", "synthesize")
parent_graph.add_edge("synthesize", END)
parent_graph.set_entry_point("research_plan")

The parent graph treats the subgraph as a single node. It invokes the subgraph with the relevant subset of state, waits for it to complete, and merges the results back. Subgraphs can be nested arbitrarily deep.

Parallel Execution

Some tasks benefit from parallelism. You want to search three different sources simultaneously, or call two different models and compare their outputs. LangGraph supports this with Send -- a way to dispatch to multiple nodes in parallel:

from langgraph.types import Send
from typing import List

class ResearchState(TypedDict):
    queries: List[str]
    search_results: Annotated[list, add_messages]  # Collect results from parallel searches

def continue_to_searches(state: ResearchState):
    """Fan out: send each query to the search node in parallel."""
    return [Send("execute_search", {"query": q}) for q in state["queries"]]

def execute_search(state: dict) -> dict:
    """Execute a single search query."""
    query = state["query"]
    results = search_api(query)
    return {"search_results": [f"Results for '{query}': {results}"]}

graph = StateGraph(ResearchState)
graph.add_node("generate_queries", generate_queries)
graph.add_node("execute_search", execute_search)
graph.add_node("synthesize", synthesize)

graph.set_entry_point("generate_queries")
graph.add_conditional_edges("generate_queries", continue_to_searches)
graph.add_edge("execute_search", "synthesize")
graph.add_edge("synthesize", END)

Each Send creates an independent execution of the target node. They run concurrently. Results are collected via the annotated state reducer (add_messages appends results from all parallel branches). The graph continues to synthesize only after all parallel branches complete.

Streaming

LangGraph graphs are Runnables, which means they support streaming. You can watch your agent think in real time:

# Stream node outputs as they happen
for event in app.stream(
    {"messages": [HumanMessage(content="Write a report on AI safety")]},
    config,
    stream_mode="values"
):
    # event contains the full state after each node
    last_message = event["messages"][-1]
    if hasattr(last_message, "content") and last_message.content:
        print(f"[{event.get('current_node', '')}] {last_message.content}")

# Stream individual tokens from the LLM
for event in app.stream(
    {"messages": [HumanMessage(content="Explain quantum computing")]},
    config,
    stream_mode="messages"
):
    if event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="", flush=True)

stream_mode="values" emits the full state after each node completes. stream_mode="messages" emits individual LLM tokens as they are generated. You can use both simultaneously for different levels of observability.

Section 6: LangSmith -- Observability

LangSmith is a SaaS platform for tracing, monitoring, and evaluating LLM applications. It is not required to use LangChain or LangGraph. It is, however, the fastest way to debug an agent that is not behaving.

What LangSmith Traces

Every LCEL chain and LangGraph graph is traced automatically when you set the appropriate environment variables:

export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY="ls_..."
export LANGCHAIN_PROJECT="my-agent-project"

With these set, every invocation produces a trace. A trace contains:

• Every LLM call: the prompt sent, the response received, token counts, latency, and cost.
• Every tool execution: the tool name, arguments, result, and execution time.
• Every state transition (LangGraph): which node ran, what the state was before and after, which edge was taken.
• Every retriever call: the query, the documents returned, and relevance scores.
• Errors and exceptions: full stack traces with the exact state at the point of failure.

Reading a LangSmith Trace

Here is what you see when you open a trace for a LangGraph agent run:

[Run: research_agent]
  |
  +-- [Node: agent] (2.3s, 1,245 tokens)
  |     Input: messages=[HumanMessage("Research quantum computing...")]
  |     LLM Call: gpt-4o
  |       Prompt tokens: 512
  |       Completion tokens: 733
  |     Output: AIMessage with tool_calls=[web_search("quantum computing 2026")]
  |
  +-- [Edge: should_continue -> "tools"]
  |
  +-- [Node: tools] (0.8s)
  |     Tool: web_search("quantum computing advances 2026")
  |     Result: "Search results: [10 links]..."
  |
  +-- [Edge: tools -> agent]
  |
  +-- [Node: agent] (3.1s, 1,890 tokens)
  |     Input: messages=[..., ToolMessage("Search results...")]
  |     LLM Call: gpt-4o
  |       Prompt tokens: 1,102
  |       Completion tokens: 788
  |     Output: AIMessage with tool_calls=[web_search("quantum error correction 2026")]
  |
  +-- [Edge: should_continue -> "tools"]
  |
  +-- [Node: tools] (0.6s)
  |     Tool: web_search("quantum error correction 2026")
  |     Result: "Search results: [8 links]..."
  |
  +-- [Edge: tools -> agent]
  |
  +-- [Node: agent] (4.2s, 2,450 tokens)
  |     Input: messages=[..., ToolMessage("Search results...")]
  |     LLM Call: gpt-4o
  |       Prompt tokens: 1,890
  |       Completion tokens: 560
  |     Output: AIMessage("Based on my research, here are the key advances...")
  |
  +-- [Edge: should_continue -> "end"]
  |
  +-- [END]

This trace tells you everything. The agent made three LLM calls and two tool calls. Total latency: ~11 seconds. Total tokens: ~5,585. Cost: approximately $0.03. You can see exactly what the model was thinking at each step, what tools it called, and what results it got. When the agent produces a wrong answer, you can pinpoint exactly where the reasoning went off the rails.

Evaluation with LangSmith

LangSmith also supports evaluation. You create a dataset of input-output pairs, run your agent against it, and compare results:

from langsmith import Client

client = Client()

# Create a dataset
dataset = client.create_dataset(
    "research_qa",
    description="Research question-answer pairs for agent evaluation"
)

# Add examples
client.create_examples(
    inputs=[
        {"question": "What is the capital of France?"},
        {"question": "Explain quantum entanglement"},
        {"question": "What were the major AI breakthroughs in 2025?"},
    ],
    outputs=[
        {"answer": "Paris"},
        {"answer": "Quantum entanglement is a phenomenon where..."},
        {"answer": "In 2025, major AI breakthroughs included..."},
    ],
    dataset_id=dataset.id,
)

# Run evaluation
from langsmith.evaluation import evaluate

results = evaluate(
    lambda inputs: app.invoke({"messages": [HumanMessage(content=inputs["question"])]}),
    data=dataset.name,
    evaluators=[
        "correctness",    # LLM-as-judge evaluation
        "helpfulness",    # Is the answer useful?
        "concise"         # Is it appropriately brief?
    ],
    experiment_prefix="research-agent-v2",
)

Evaluation turns agent development from alchemy into engineering. You make a change, run the eval, and see whether your agent got better or worse. Without this, you are guessing.

Section 7: The Complete LangGraph Research Agent

You now have all the pieces. Let us assemble them into a complete research agent -- one that takes a research topic, plans a strategy, searches the web, analyzes results, synthesizes findings, and produces a report. With human-in-the-loop review before the final output.

from typing import TypedDict, Annotated, Literal, List
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Send
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage, SystemMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
import json

# ============================================================
# STATE
# ============================================================
class ResearchState(TypedDict):
    topic: str
    messages: Annotated[list, add_messages]
    research_plan: List[str]
    search_queries: List[str]
    search_results: Annotated[list, add_messages]
    analysis_notes: str
    draft_report: str
    final_report: str
    iteration_count: int

# ============================================================
# TOOLS
# ============================================================
@tool
def web_search(query: str) -> str:
    """Search the web. Returns a list of relevant results with snippets."""
    # In production, use a real search API (Tavily, SerpAPI, Brave)
    return json.dumps([
        {"title": f"Result for '{query}' - 1", "snippet": f"Key information about {query}..."},
        {"title": f"Result for '{query}' - 2", "snippet": f"Additional context on {query}..."},
        {"title": f"Result for '{query}' - 3", "snippet": f"Recent developments in {query}..."},
    ])

@tool
def extract_key_facts(text: str) -> str:
    """Extract key facts, dates, names, and statistics from a block of text."""
    # In production, this could be another LLM call with structured output
    return f"Extracted facts from text: [key points from '{text[:100]}...']"

tools = [web_search, extract_key_facts]
tool_map = {t.name: t for t in tools}

# ============================================================
# MODEL
# ============================================================
model = ChatOpenAI(model="gpt-4o").bind_tools(tools)

# ============================================================
# NODES
# ============================================================
def plan_research(state: ResearchState) -> dict:
    """Create a research plan: what questions need answering?"""
    planner = ChatOpenAI(model="gpt-4o")

    system = """You are a research planner. Given a research topic, produce a plan
with 3-5 specific questions that need to be answered. Output as a JSON list of strings."""

    response = planner.invoke([
        SystemMessage(content=system),
        HumanMessage(content=f"Research topic: {state['topic']}")
    ])

    try:
        plan = json.loads(response.content)
    except json.JSONDecodeError:
        # Fallback: extract list from text
        import re
        plan = re.findall(r'"([^"]+)"', response.content)

    return {
        "research_plan": plan,
        "search_queries": plan,  # Initial queries = plan questions
        "iteration_count": 0,
        "messages": [AIMessage(content=f"Research plan: {json.dumps(plan)}")]
    }

def call_model(state: ResearchState) -> dict:
    """Call the LLM with current context."""
    system = SystemMessage(content="""You are a research agent. Your goal is to thoroughly
research the given topic. Use web_search to find information. Use extract_key_facts
to pull out important details. When you have enough information, synthesize your findings
into a comprehensive report. Do not end until you have covered all questions in the research plan.""")

    messages = [system] + state["messages"]
    response = model.invoke(messages)
    return {"messages": [response]}

def execute_tools(state: ResearchState) -> dict:
    """Execute any pending tool calls."""
    last_message = state["messages"][-1]
    tool_messages = []

    for tool_call in last_message.tool_calls:
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]
        tool_func = tool_map[tool_name]
        result = tool_func.invoke(tool_args)
        tool_messages.append(
            ToolMessage(content=str(result), tool_call_id=tool_call["id"])
        )

    return {
        "messages": tool_messages,
        "search_results": [str(tm.content) for tm in tool_messages]
    }

def analyze_results(state: ResearchState) -> dict:
    """Analyze collected search results and extract key findings."""
    if not state.get("search_results"):
        return {"analysis_notes": "No results to analyze yet."}

    analyzer = ChatOpenAI(model="gpt-4o")
    results_text = "\n".join(state["search_results"][-10:])  # Last 10 results

    response = analyzer.invoke([
        SystemMessage(content="Analyze these search results. Extract key findings, contradictions, and gaps."),
        HumanMessage(content=f"Research topic: {state['topic']}\n\nSearch results:\n{results_text}")
    ])

    return {"analysis_notes": response.content}

def write_report(state: ResearchState) -> dict:
    """Synthesize findings into a draft report."""
    writer = ChatOpenAI(model="gpt-4o")

    response = writer.invoke([
        SystemMessage(content="""Write a comprehensive research report based on the analysis notes.
Include: Executive Summary, Key Findings, Detailed Analysis, and Sources.
Be thorough but concise. Use markdown formatting."""),
        HumanMessage(content=f"""Topic: {state['topic']}
Research Plan: {json.dumps(state.get('research_plan', []))}
Analysis Notes: {state.get('analysis_notes', 'No analysis available.')}""")
    ])

    return {"draft_report": response.content}

def human_review(state: ResearchState) -> dict:
    """Pause for human review of the draft report."""
    approval = interrupt({
        "message": "Review the draft research report",
        "draft": state["draft_report"],
        "options": ["approve", "request_revisions", "reject"]
    })

    if approval == "approve":
        return {"final_report": state["draft_report"]}
    elif approval == "request_revisions":
        return {
            "messages": [HumanMessage(content=f"Please revise the report with these notes: {approval.get('notes', '')}")],
            "final_report": ""
        }
    else:
        return {"final_report": "Report rejected by reviewer."}

# ============================================================
# ROUTING
# ============================================================
def should_continue(state: ResearchState) -> Literal["tools", "analyze", "end"]:
    """Decide next step after model call."""
    last_message = state["messages"][-1]

    # If the model wants to call tools, route to tool execution
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"

    # If we have search results but haven't analyzed them, analyze
    if state.get("search_results") and not state.get("analysis_notes"):
        return "analyze"

    # If we have analysis but no draft, write the report
    if state.get("analysis_notes") and not state.get("draft_report"):
        return "write_report"

    # Otherwise, we're done with this phase
    return "end"

def after_tools(state: ResearchState) -> Literal["agent", "analyze"]:
    """After executing tools, decide: more research or analyze?"""
    iteration = state.get("iteration_count", 0) + 1

    if iteration >= 3:
        # Max research iterations reached -- move to analysis
        return "analyze"

    return "agent"

def after_analysis(state: ResearchState) -> Literal["agent", "write_report"]:
    """After analysis, decide: more research needed or write report?"""
    # If analysis mentions gaps, do more research
    analysis = state.get("analysis_notes", "").lower()
    if "gap" in analysis or "missing" in analysis or "further research" in analysis:
        if state.get("iteration_count", 0) < 5:
            return "agent"

    return "write_report"

# ============================================================
# BUILD THE GRAPH
# ============================================================
graph = StateGraph(ResearchState)

# Add nodes
graph.add_node("plan", plan_research)
graph.add_node("agent", call_model)
graph.add_node("tools", execute_tools)
graph.add_node("analyze", analyze_results)
graph.add_node("write_report", write_report)
graph.add_node("review", human_review)

# Set entry
graph.set_entry_point("plan")

# Plan always goes to agent
graph.add_edge("plan", "agent")

# Agent routes conditionally
graph.add_conditional_edges(
    "agent",
    should_continue,
    {
        "tools": "tools",
        "analyze": "analyze",
        "write_report": "write_report",
        "end": END
    }
)

# After tools, decide: more agent or analyze
graph.add_conditional_edges(
    "tools",
    after_tools,
    {
        "agent": "agent",
        "analyze": "analyze"
    }
)

# After analysis, decide: more research or write
graph.add_conditional_edges(
    "analyze",
    after_analysis,
    {
        "agent": "agent",
        "write_report": "write_report"
    }
)

# Write report goes to human review
graph.add_edge("write_report", "review")

# Review ends the graph
graph.add_edge("review", END)

# Compile with checkpointer for human-in-the-loop
app = graph.compile(checkpointer=MemorySaver())

# ============================================================
# RUN IT
# ============================================================
config = {"configurable": {"thread_id": "research-session-1"}}

# First run: will pause at human review
print("Starting research agent...")
events = app.stream(
    {"topic": "Advances in quantum computing error correction in 2026"},
    config,
    stream_mode="values"
)

for event in events:
    node_name = event.get("current_node", "")
    if node_name:
        print(f"\n--- Completed node: {node_name} ---")
    if event.get("draft_report"):
        print("\n=== DRAFT REPORT ===")
        print(event["draft_report"][:500] + "...")

# The graph pauses at the review node.
# In a real application, a human reviews the draft and resumes:
# app.invoke(None, config)  # Resume with approval

This is about 200 lines. It is a complete research agent with planning, tool use, analysis, report generation, and human review. Every decision point is explicit. Every state transition is visible. You can add new nodes, new tools, new routing logic without touching the existing code. That is the power of LangGraph.

The Graph Structure

[plan] --> [agent] <--> [tools]
              |
              v
          [analyze] <--> [agent]
              |
              v
        [write_report]
              |
              v
          [review] --> END

The agent researches (agent <--> tools loop), analyzes findings, optionally does more research if gaps are found, writes a report, and pauses for human review. Every phase is a node. Every decision is a conditional edge. The state carries everything forward.

Section 8: LangChain/LangGraph vs. Raw -- When to Use What

You now have three ways to build agents: raw API calls (Chapters 4-9), LangChain/LCEL, and LangGraph. Here is when to use each.

Use Raw API/SDK When:

• You are learning. Frameworks hide details. When you are building your first agents, you need to see every API call, every message, every loop iteration. Raw code teaches you what is actually happening.
• Your agent is simple. A single model call with one or two tools does not need a framework. The OpenAI or Anthropic SDK is sufficient.
• You need maximum control. Frameworks make assumptions about how agents work. When your agent has an unusual pattern -- a custom memory structure, a non-standard tool execution model, a novel routing strategy -- the framework fights you.
• You are optimizing for latency or cost. Frameworks add overhead. Sometimes a few milliseconds. Sometimes more. When every millisecond counts, raw API calls give you the tightest loop.

Use LangChain When:

• You need document loaders and text splitters. LangChain's document processing utilities are genuinely useful and well-maintained. Loading PDFs, scraping web pages, splitting text into chunks -- this is boilerplate you do not want to write.
• You want the @tool decorator. It is a small thing, but it eliminates a surprising amount of boilerplate around JSON Schema generation and tool execution.
• You are building RAG pipelines. LCEL's composability shines for retrieval-augmented generation. The pipe operator makes data flow explicit and swappable.

Use LangGraph When:

• Your agent has complex control flow. Branching, looping, parallel execution, conditional routing -- LangGraph makes these explicit and debuggable.
• You need human-in-the-loop. The interrupt() mechanism is the cleanest implementation of human approval in any agent framework.
• You need state persistence. Checkpointing gives you resume-from-failure, replay-for-debugging, and time-travel exploration for free.
• You are building production agents. LangGraph's explicit state machine model is easier to test, monitor, and maintain than a raw loop with implicit state management.

The Hybrid Approach

In practice, you will often use all three. Raw API calls for the core agent logic. LangChain for data loading and tool definitions. LangGraph for orchestration and state management. They compose:

# Raw API for a custom reasoning step
def deep_reasoning(state):
    response = anthropic_client.messages.create(
        model="claude-sonnet-4-20250514",
        system="You are an expert reasoner...",
        messages=state["messages"]
    )
    return {"messages": [response]}

# LangChain for document loading
from langchain_community.document_loaders import WebBaseLoader
loader = WebBaseLoader("https://example.com/research-paper")
documents = loader.load()

# LangGraph for orchestration
graph = StateGraph(State)
graph.add_node("reason", deep_reasoning)  # Raw API node
graph.add_node("load_docs", load_documents)  # LangChain node
graph.add_node("synthesize", call_model)  # LCEL node
# ... edges and routing ...

The frameworks are tools, not religions. Use the right one for each part of the problem.

The Turn

You came into this chapter with a question: "Should I use LangChain?" The answer is not yes or no. It is: use the parts that earn their keep.

LCEL earns its keep. The pipe operator, the Runnable interface, the uniform support for invoke/batch/stream -- these are genuine improvements over raw API calls for composing LLM pipelines. The @tool decorator earns its keep. Document loaders and text splitters earn their keep.

LangGraph earns its keep emphatically. It is not "LangChain for graphs." It is a state machine framework that happens to be maintained by the same organization. It models agents the way they actually work -- as stateful, branching, looping systems -- rather than the way frameworks wish they worked -- as linear chains of magical abstractions.

The old Chain API does not earn its keep. Most of langchain-community does not earn its keep. AgentExecutor does not earn its keep. You now know to skip them.

More importantly, you now understand LangChain strategically, not just syntactically. You can look at a LangChain tutorial and know whether it is showing you the good parts or the deprecated parts. You can evaluate a new LangChain feature and decide whether it solves a real problem or adds another layer of magic you will regret. You can build complex agent workflows with LangGraph and debug them with LangSmith.

That is the difference between using a framework and understanding it.

Close

You have now mastered single agents -- from raw loops in Chapter 4 to structured outputs in Chapter 5, from tool use in Chapter 6 to the state-machine workflows of LangGraph in this chapter. You can build an agent that researches, reasons, uses tools, and produces structured output. You can debug it, checkpoint it, and put a human in the loop.

But the real power of agentic AI emerges when agents work together.

A single agent, no matter how well-designed, has a single perspective. It can search, but it cannot debate. It can analyze, but it cannot get a second opinion. It can write, but it cannot get an editor. The most interesting agent systems -- the ones that produce genuinely surprising results -- are multi-agent systems. Agents that collaborate. Agents that challenge each other. Agents that specialize and coordinate.

In the next chapter, you will build your first multi-agent system. You will create agents with different personalities and tools, give them a shared task, and watch them figure out how to work together. You will learn about agent communication protocols, task delegation, and the patterns that make multi-agent systems more than the sum of their parts.

The single agent was the foundation. The multi-agent system is where things get interesting.

---

What you built in this chapter:

Component	What It Does
LCEL chain	Composable LLM pipelines with the pipe operator
Runnable interface	Uniform invoke/batch/stream across all components
@tool decorator	Converts Python functions into model-callable tools
LangGraph StateGraph	Explicit state machine for agent workflows
Agent + tools + routing	Complete tool-using agent with conditional edges
Human-in-the-loop	Pause execution for human approval with interrupt()
Checkpointing	Save and resume graph state at any point
Subgraphs	Compose graphs within graphs for modular agents
Parallel execution	Fan out to multiple nodes with Send
Research agent	Complete 200-line agent with planning, search, analysis, and review
LangSmith tracing	Automatic observability for every LLM call and state transition

Key takeaways:

• LangChain's value is in composability (LCEL), not high-level abstractions (old Chain API).
• Use LCEL, @tool, document loaders, text splitters, and LangGraph. Skip the rest.
• LangGraph models agents as explicit state machines: State, Nodes, Edges, Graph.
• Every LangGraph graph is a Runnable -- invoke, batch, stream, and async all work.
• Human-in-the-loop, checkpointing, subgraphs, and parallel execution are first-class features.
• LangSmith gives you automatic tracing for debugging and evaluation for measurement.
• The hybrid approach -- raw API for logic, LangChain for data, LangGraph for orchestration -- is often the right answer.
• A framework is a tool, not a requirement. Use the parts that earn their keep.