Building Agentic Workflows

LangGraph is the framework our team uses to build stateful, multi-step agentic workflows. This guide walks you through the core concepts and has you running a working agent in minutes.

If you've already built LangChain chains and want to move beyond linear pipelines into workflows with loops, branching, and persistent state, this is the place to start.

---

What Is LangGraph

LangGraph is a low-level orchestration framework for building, managing, and deploying long-running, stateful agents. It provides the infrastructure for durable execution, streaming, human-in-the-loop, persistence, and comprehensive memory.

How It Differs From LangChain

LangChain chains are linear pipelines: input flows through a sequence of components and produces output. They work well for straightforward tasks, but fall short when you need:

• Loops — an agent that iteratively calls tools until it has an answer.
• Conditional branching — different paths depending on intermediate results.
• Persistent state — remembering context across steps, failures, or even sessions.
• Human-in-the-loop — pausing execution to get human input before continuing.

LangGraph models workflows as graphs: nodes do work, edges route control flow, and a shared state object carries data between them. Under the hood, LangChain's own agent abstractions are built on LangGraph.

Why We Use It

• Control — explicit graph structure means you know exactly how your agent executes.
• Persistence — checkpoints let agents survive failures and resume from where they left off.
• Streaming — observe intermediate state, messages, and tool calls in real time.
• Human-in-the-loop — interrupt execution at any node for review or modification.
• Composability — graphs can be nested as subgraphs for modular design.

High-Level Mental Model

User Input → [Node A] ──edge──→ [Node B] ──conditional──→ [Node C] or END
                ↑                    │
                └─────edge────────────┘
                    (loop back)

Shared State flows through every node, accumulating messages,
tool results, and intermediate data.

---

Core Concepts

Before writing code, understand the six building blocks you will combine.

Graphs — StateGraph

The StateGraph class is the foundation. You define a graph by specifying a state schema, then adding nodes and edges. Once compiled, the graph becomes a runnable object you can invoke:

from langgraph.graph import StateGraph, START, END

# Create a graph with a state schema
graph = StateGraph(MyState)

# Add nodes (functions that do work)
graph.add_node("my_node", my_node_function)

# Add edges (control flow)
graph.add_edge(START, "my_node")
graph.add_edge("my_node", END)

# Compile into a runnable
app = graph.compile()

The constants START and END mark entry and exit points. Every execution begins at START and terminates at END (unless interrupted).

Nodes

Nodes are Python callables that receive the current state and return a partial state update. They are the workhorses of your graph:

def process_data(state: MyState) -> dict:
    # Read from state
    input_text = state["text"]
    # Do work
    result = do_something(input_text)
    # Return partial state update
    return {"result": result}

A node function:

• Receives the full current state as its argument.
• Returns a partial update — only the keys you want to change.
• Does not modify state in place. LangGraph merges your return value into the existing state.

Edges

Edges define control flow between nodes. There are two types:

• Static edges — always route from one node to another.
• Conditional edges — route based on a function that inspects state.

# Static: always go from node A to node B
graph.add_edge("node_a", "node_b")

# Conditional: choose the next node based on state
graph.add_conditional_edges(
    "node_a",
    routing_function,  # returns a node name
    ["node_b", "node_c", END],
)

State

State is the shared data structure that flows through every node. It is the single most important concept in LangGraph. You define it as either a TypedDict or a Pydantic model:

from typing_extensions import TypedDict

class AgentState(TypedDict):
    messages: list
    steps: int

Every node reads from and writes to this shared state. LangGraph merges updates automatically.

Conditional Edges

Conditional edges enable dynamic routing. The routing function inspects state and returns the name of the next node (or END):

from typing import Literal

def should_continue(state: AgentState) -> Literal["analyze", "summarize", END]:
    if state.get("needs_analysis"):
        return "analyze"
    if state.get("ready_to_summarize"):
        return "summarize"
    return END

The Compiler

The .compile() method transforms your graph definition into a runnable. Compilation validates the graph structure, checks that all referenced nodes exist, and builds the execution engine:

app = graph.compile()

# Now you can invoke it
result = app.invoke({"messages": [...]})

# Or stream it
for event in app.stream({"messages": [...]}):
    print(event)

---

Defining State

State is the backbone of every LangGraph application. Getting state design right is more important than getting any other part right.

Using TypedDict for State

TypedDict is the simplest and most common approach. It gives you a lightweight, dictionary-like structure with type hints:

from typing_extensions import TypedDict

class AgentState(TypedDict):
    messages: list
    query: str
    context: str
    step_count: int

Each key in the state is a field. Nodes read fields they need and return updates for the fields they modify.

Using Pydantic Models for State

Pydantic models add validation, defaults, and richer type information. Use them when you need schema enforcement:

from pydantic import BaseModel, Field

class AgentState(BaseModel):
    messages: list = []
    query: str = ""
    context: str = ""
    step_count: int = 0
    status: str = Field(default="idle", description="Current agent status")

Reducers — How State Updates Work

When multiple nodes write to the same state field, LangGraph needs a rule for merging updates. By default, later writes overwrite earlier values. For fields that should accumulate (like message lists), you use a reducer:

from typing_extensions import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    # Default: last write wins
    query: str

    # Reducer: new values are appended, not replacing
    messages: Annotated[list, operator.add]

The Annotated type wraps a field with a reducer function. operator.add concatenates lists, so each node's returned messages are appended to the existing list rather than replacing it.

Common reducer patterns:

Pattern	Use Case
Annotated[list, operator.add]	Accumulating messages, tool results, or any list
Annotated[dict, operator.or_]	Merging dictionary updates
Default (no reducer)	Scalar values that get overwritten each step

Annotated Patterns

The Annotated type is the standard way to attach reducer behavior. Here is a complete state definition showing multiple patterns:

from typing_extensions import TypedDict, Annotated
from langchain.messages import AnyMessage
import operator

class ResearchState(TypedDict):
    # Messages accumulate via operator.add
    messages: Annotated[list[AnyMessage], operator.add]

    # Search results accumulate
    search_results: Annotated[list[str], operator.add]

    # Scalar fields — last write wins
    query: str
    topic: str
    draft: str
    step_count: int

Do This / Don't Do This

State Design Rules

Do:

• Keep state minimal. Include only fields that multiple nodes need to share.
• Use Annotated[list, operator.add] for any field that should accumulate across steps.
• Name state fields explicitly. A well-named state is self-documenting.
• Define state at the top of your file or in a dedicated state.py module.

Don't:

• Put large blobs of data in state unless necessary. State is checkpointed.
• Mutate state in place inside nodes. Always return a new partial update.
• Use generic names like data or result. Be specific: search_results, analysis, summary.
• Forget that non-annotated list fields will be replaced, not appended.

---

Building Your First Agent

Let's build a research agent that can search for information and summarize findings. This agent will loop: call the LLM, check if it wants to use a tool, execute the tool, and repeat until it has an answer.

Step 1 — Install Dependencies

We use uv for package management:

uv init research-agent
cd research-agent

# Install LangGraph and a model provider
uv add langgraph langchain langchain-openai
# Or for Anthropic:  uv add langgraph langchain langchain-anthropic

uv sync

Step 2 — Set Environment Variables

export OPENAI_API_KEY="sk-..."
# Or for Anthropic: export ANTHROPIC_API_KEY="sk-ant-..."

Step 3 — Define the State Schema

The state carries messages and metadata through the agent's execution:

from typing_extensions import TypedDict, Annotated
from langchain.messages import AnyMessage
import operator


class AgentState(TypedDict):
    """State shared across all nodes in the research agent."""
    messages: Annotated[list[AnyMessage], operator.add]
    """Chat messages — accumulates across steps."""
    query: str
    """The original user query."""
    step_count: int
    """Number of LLM invocations (for debugging and limiting loops)."""

Step 4 — Define Tools

Tools extend the agent's capabilities. Here, we define a mock search tool and a summarize utility:

from langchain.tools import tool


@tool
def web_search(query: str) -> str:
    """Search the web for information about a topic.

    Args:
        query: The search query string.
    """
    # In production, this would call a real search API
    # (Tavily, Brave, SerpAPI, etc.)
    return (
        f"Search results for '{query}':\n"
        "1. LangGraph is a framework for building stateful, multi-step agents.\n"
        "2. It uses a graph-based model with nodes, edges, and shared state.\n"
        "3. State is defined using TypedDict or Pydantic models with optional reducers."
    )


@tool
def summarize_text(text: str) -> str:
    """Summarize a block of text into key points.

    Args:
        text: The text to summarize.
    """
    # In production, this would call an LLM with a summarization prompt
    return "Key points: LangGraph enables stateful agent workflows using graphs with nodes, edges, and shared state."


# Build a lookup by name for the tool execution node
tools = [web_search, summarize_text]
tools_by_name = {tool.name: tool for tool in tools}

Step 5 — Set Up the Model

Initialize the LLM and bind tools to it:

from langchain.chat_models import init_chat_model

# Initialize the model — swap the string to change providers
model = init_chat_model("gpt-4o")

# Bind tools so the model can decide to call them
model_with_tools = model.bind_tools(tools)

Step 6 — Create Nodes

Each node is a function that reads state, does work, and returns a partial state update:

from langchain.messages import SystemMessage, ToolMessage


def call_llm(state: AgentState) -> dict:
    """Call the LLM with the current message history and system prompt."""
    system = SystemMessage(
        content=(
            "You are a research assistant. Search for information when needed, "
            "summarize findings, and provide clear answers to the user's question."
        )
    )
    response = model_with_tools.invoke([system] + state["messages"])
    return {
        "messages": [response],
        "step_count": state.get("step_count", 0) + 1,
    }


def execute_tools(state: AgentState) -> dict:
    """Execute tool calls requested by the LLM."""
    tool_results = []
    last_message = state["messages"][-1]

    for tool_call in last_message.tool_calls:
        tool_name = tool_call["name"]
        tool_fn = tools_by_name[tool_name]
        observation = tool_fn.invoke(tool_call["args"])
        tool_results.append(
            ToolMessage(
                content=observation,
                tool_call_id=tool_call["id"],
            )
        )

    return {"messages": tool_results}

Step 7 — Add Routing Logic

The conditional edge function decides whether to call tools or finish:

from typing import Literal
from langgraph.graph import END


def route_after_llm(state: AgentState) -> Literal["execute_tools", END]:
    """Route based on whether the LLM requested tool calls."""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "execute_tools"
    return END

Step 8 — Build and Compile the Graph

Wire everything together:

from langgraph.graph import StateGraph, START

# Create the graph
agent_builder = StateGraph(AgentState)

# Register nodes
agent_builder.add_node("call_llm", call_llm)
agent_builder.add_node("execute_tools", execute_tools)

# Define control flow
agent_builder.add_edge(START, "call_llm")
agent_builder.add_conditional_edges(
    "call_llm",
    route_after_llm,
    ["execute_tools", END],
)
agent_builder.add_edge("execute_tools", "call_llm")

# Compile
agent = agent_builder.compile()

The execution flow:

START → call_llm ──has tool calls?──→ execute_tools → call_llm (loop)
                │
                └──no tool calls──→ END

Step 9 — Run the Agent

from langchain.messages import HumanMessage

result = agent.invoke({
    "messages": [HumanMessage(content="What is LangGraph and how does it manage state?")],
    "query": "What is LangGraph and how does it manage state?",
    "step_count": 0,
})

# Print the conversation
for message in result["messages"]:
    message.pretty_print()

The agent will:

1. Receive the user question.
2. Call the LLM, which decides to use web_search.
3. Execute the search tool and feed results back to the LLM.
4. The LLM processes results and either calls more tools or produces a final answer.
5. Return the complete message history.

Full Code Example

The complete agent code is available as a single runnable script. Copy it into a file and run with uv run agent.py.

---

Working with Tools

Tool integration is central to agent design. Tools extend the agent's capabilities beyond what the LLM can generate from its training data.

Binding Tools to the LLM

The bind_tools() method attaches tool definitions to the model, enabling the model to decide when and how to call them:

from langchain.chat_models import init_chat_model
from langchain.tools import tool


@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"The weather in {city} is sunny, 72°F."


@tool
def get_time(city: str) -> str:
    """Get the current time in a city."""
    return f"The current time in {city} is 3:45 PM."


model = init_chat_model("gpt-4o")
model_with_tools = model.bind_tools([get_weather, get_time])

When the model receives a request, it can respond with either text, tool calls, or both. The tool_calls attribute on the response indicates whether tools were requested.

Tool Calling Nodes

The tool execution node iterates over the LLM's tool calls, executes each one, and returns ToolMessage objects:

from langchain.messages import ToolMessage


def tool_node(state: AgentState) -> dict:
    """Execute all tool calls from the last LLM response."""
    results = []
    last_message = state["messages"][-1]

    for tool_call in last_message.tool_calls:
        tool_name = tool_call["name"]
        tool_fn = tools_by_name[tool_name]

        try:
            observation = tool_fn.invoke(tool_call["args"])
            results.append(ToolMessage(
                content=str(observation),
                tool_call_id=tool_call["id"],
            ))
        except Exception as e:
            # Return an error message instead of crashing the agent
            results.append(ToolMessage(
                content=f"Error executing {tool_name}: {e}",
                tool_call_id=tool_call["id"],
            ))

    return {"messages": results}

Handling Tool Errors in the Graph

Tools can fail — APIs time out, inputs are invalid, external services are down. Handle errors gracefully so the agent can recover:

def resilient_tool_node(state: AgentState) -> dict:
    """Execute tools with error handling and retry logic."""
    results = []
    last_message = state["messages"][-1]

    for tool_call in last_message.tool_calls:
        tool_name = tool_call["name"]
        tool_fn = tools_by_name.get(tool_name)

        if tool_fn is None:
            results.append(ToolMessage(
                content=f"Unknown tool: {tool_name}",
                tool_call_id=tool_call["id"],
            ))
            continue

        try:
            observation = tool_fn.invoke(tool_call["args"])
            results.append(ToolMessage(
                content=str(observation),
                tool_call_id=tool_call["id"],
            ))
        except Exception as e:
            results.append(ToolMessage(
                content=f"Tool '{tool_name}' failed: {type(e).__name__}: {e}. "
                        "Try a different approach or rephrase your request.",
                tool_call_id=tool_call["id"],
            ))

    return {"messages": results}

Tool Error Strategy

Never let tool exceptions bubble up and crash the agent. Always wrap tool execution in try/except and return a ToolMessage with the error. The LLM can then decide how to proceed — retry with different arguments, try a different tool, or conclude without that information.

---

Message Passing

Messages are the primary communication channel in LangGraph agents. Understanding how messages flow through state is essential.

How Messages Flow Through State

The messages field in state is annotated with operator.add, so each node's returned messages are appended:

Step 1: [HumanMessage("What is LangGraph?")]
         ↓ call_llm node
Step 2: [HumanMessage(...), AIMessage(..., tool_calls=[...])]
         ↓ execute_tools node
Step 3: [HumanMessage(...), AIMessage(...), ToolMessage("Search results...")]
         ↓ call_llm node again
Step 4: [HumanMessage(...), AIMessage(...), ToolMessage(...), AIMessage("Final answer")]
         ↓ route → END

Each step adds messages to the growing list. The LLM sees the full conversation history, giving it context for decisions.

Message Types

LangGraph works with four message roles:

Type	Role	Direction
HumanMessage	User input or instructions	Human → Agent
AIMessage	LLM response, may include tool_calls	Agent → Human (or triggers tools)
ToolMessage	Result from a tool execution	Tool → Agent
SystemMessage	System-level instructions	Developer → Model

from langchain.messages import (
    HumanMessage,
    AIMessage,
    ToolMessage,
    SystemMessage,
)

# A typical conversation state after two steps:
messages = [
    SystemMessage(content="You are a helpful research assistant."),
    HumanMessage(content="What is LangGraph?"),
    AIMessage(content="", tool_calls=[{"name": "web_search", "args": {"query": "LangGraph"}}]),
    ToolMessage(content="Search results: ...", tool_call_id="call_abc123"),
    AIMessage(content="LangGraph is a framework for building stateful agents..."),
]

Message History Management

In long-running agents, message history can grow large. Here are strategies for managing it:

Limit message count:

MAX_MESSAGES = 20

def trim_messages(messages: list) -> list:
    """Keep only the most recent messages to fit within context window."""
    if len(messages) <= MAX_MESSAGES:
        return messages
    # Keep system message + most recent messages
    system_msgs = [m for m in messages if m.type == "system"]
    recent = messages[-MAX_MESSAGES:]
    return system_msgs + recent

Summarize older messages:

def summarize_history(messages: list, model) -> str:
    """Compress older messages into a summary."""
    summary_prompt = ChatPromptTemplate.from_messages([
        ("system", "Summarize this conversation in 2-3 sentences."),
        ("human", "{messages}"),
    ])
    chain = summary_prompt | model | StrOutputParser()
    return chain.invoke({"messages": "\n".join(str(m) for m in messages)})

Message Accumulation

Always use Annotated[list, operator.add] for the messages field. Without the reducer, each node's messages would replace the entire list, losing conversation history.

---

Next Steps

You now understand the core building blocks of LangGraph and have a working agent. Here is where to go next:

• LangGraph Best Practices — our team's standards for production-grade agent design, including error handling, testing, and performance patterns.
• PostgreSQL Checkpointing & Memory Setup — how to persist agent state to PostgreSQL for durability, time travel, and human-in-the-loop workflows.
• LangGraph Official Docs — comprehensive reference for the Graph API, Functional API, streaming, and deployment.
• LangSmith Tracing — set up observability to trace, debug, and evaluate your agents.

tip

Set up LangSmith tracing early. Enable it with LANGSMITH_TRACING=true and your API key. Every agent invocation will be traced, giving you visibility into execution paths, state transitions, and tool call details — invaluable for debugging and optimization.