LangGraph in Action: Practical Guide

LangGraph is an agent framework built on the LangChain ecosystem that uses a Directed Cyclic Graph (DCG) architecture. Unlike linear chains, it allows for stateful, iterative, and looping workflows.

Core Components

Nodes

Nodes represent units of work in the graph:

• LLM Call Nodes: Interact with language models for reasoning.
• Tool Call Nodes: Execute external tools or APIs.
• Custom Function Nodes: Encapsulate business or data processing logic.
• Subgraph Nodes: Modularize complex workflows.

Edges

Edges define the flow between nodes:

• Normal Edges: Define direct, sequential execution.
• Conditional Edges: Dynamically determine the next node based on the current state.
• Entry Points: Define where the workflow starts.
• Conditional Entry Points: Select the starting node based on initial state.

State

State is the shared data container that acts as the graph's memory. It is used for:

• Storing context (conversation history, task progress).
• Passing data between nodes.
• Persistence (memory and fault tolerance).
• Multi-agent collaboration.

In the Functional API, short-term memory (session state) can be implemented by using the previous parameter in an @entrypoint and the entrypoint.final() function.

• previous parameter: Automatically injects the state saved from the last execution of the same thread_id.
• entrypoint.final(value=..., save=...): Allows you to separate the value returned to the caller (value) from the state that is persisted for the next call (save).

This pattern is ideal for conversational agents where you need to maintain message history or session variables across multiple turns.

@entrypoint(checkpointer=MemorySaver())
def counter_workflow(increment: int, *, previous: Any = None) -> 'entrypoint.final[int, int]':
    """使用先前状态维护计数器的工作流。"""
    current_count = previous if previous is not None else 0
    new_count = current_count + increment
    return entrypoint.final(value=current_count, save=new_count)

# To invoke with persistence:
config = {"configurable": {"thread_id": "counter_workflow_1"}}
result = counter_workflow.invoke(increment, config)

Subgraphs allow you to nest one LangGraph within another, enabling modular design and complex system management. There are two primary ways to use them:

1. Shared State (Simple): Add a pre-compiled subgraph directly as a node in the parent graph. This works best if the parent and child share the same state schema (e.g., both use a messages key).
2. Heterogeneous State (Advanced): Use a node function in the parent graph to wrap the subgraph call. This allows you to perform State Transformation:
   • Input Transformation: Convert ParentState keys to the format expected by ChildState.
   • Subgraph Execution: Call subgraph.invoke(transformed_input).
   • Output Transformation: Map the ChildState results back into the ParentState format.

Benefits:

• Isolation: Subgraphs have independent state spaces, preventing key collisions.
• Reusability: A subgraph can be used in multiple different parent graphs.
• Complexity Management: Breaks large graphs into smaller, testable units.

# Example of State Transformation pattern for Heterogeneous Subgraphs

def call_child_graph(state: ParentState) -> ParentState:
    # 1. Input Transformation: Parent -> Child
    child_input = {"child_key": state["parent_key"]}
    
    # 2. Subgraph Execution
    child_output = child_graph.invoke(child_input)
    
    # 3. Output Transformation: Child -> Parent
    return {"parent_key": child_output["child_key"]}

# Add the wrapper function as a node in the parent graph
builder.add_node("child_node", call_child_graph)

LangGraph supports parallel execution through Fan-out (one node triggering multiple downstream nodes) and Fan-in (multiple nodes converging into a single downstream node).

To implement this:

1. Fan-out: Add multiple outgoing edges from a single node to different target nodes. LangGraph will trigger these nodes concurrently.
2. Fan-in: Connect multiple nodes to the same downstream node. This downstream node acts as a synchronization point, waiting for all incoming parallel branches to complete before executing.
3. State Management: Use a Reducer (e.g., operator.add) in your TypedDict state definition to ensure that updates from concurrent branches are correctly merged rather than overwritten.

import operator
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

# Use operator.add to merge results from parallel branches
class ParallelState(TypedDict):
    aggregate: Annotated[list, operator.add]

def node_a(state: ParallelState):
    return {"aggregate": ["A"]}

def node_b(state: ParallelState):
    return {"aggregate": ["B"]}

def node_c(state: ParallelState):
    return {"aggregate": ["C"]}

def node_d(state: ParallelState):
    return {"aggregate": ["D"]}

builder = StateGraph(ParallelState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_node("node_c", node_c)
builder.add_node("node_d", node_d)

# Fan-out: node_a triggers node_b and node_c concurrently
builder.add_edge(START, "node_a")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_a", "node_c")

# Fan-in: node_d waits for both node_b and node_c
builder.add_edge("node_b", "node_d")
builder.add_edge("node_c", "node_d")
builder.add_edge("node_d", END)

parallel_graph = builder.compile()

A Reducer allows you to define custom logic for how a state key is updated when a node returns a value. Instead of overwriting the existing value, a reducer can merge, append, or filter data.

Common use cases:

• Appending to lists: Using operator.add or add_messages to maintain history.
• Custom merging: Creating a function to merge lists and remove duplicates.

To use a reducer, wrap the state field in Annotated[Type, reducer_function] within your TypedDict definition.

from typing_extensions import TypedDict, Annotated
from operator import add
from langgraph.graph.message import add_messages

# Example 1: Using built-in add_messages for chat history
class ChatState(TypedDict):
    message_history: Annotated[list, add_messages]

# Example 2: Using a custom reducer for unique items
def reducer_extend_unique(left: list[str] | None, right: list[str] | None) -> list[str]:
    existing_items = left if left else []
    new_items = right if right else []
    return list(set(existing_items + new_items))

class CustomState(TypedDict):
    item_list: Annotated[list[str], reducer_extend_unique]

The Functional API allows you to build workflows using standard Python functions and decorators, providing a more intuitive programming model than the explicit Graph API.

• @entrypoint: Defines the entry point and boundary of a workflow. It can accept several injectable parameters like store, writer, and config.
• @task: Encapsulates an independent unit of work. Tasks can be called within an @entrypoint and their results retrieved using .result().

Key Features:

• Task Orchestration: Use standard Python if/else and loops to control flow.
• Parallel Execution: Call multiple @task functions without awaiting them immediately to run them in parallel. Use .result() to collect their outputs.
• Retries: Configure RetryPolicy on @task to handle transient errors automatically.

from langgraph.func import entrypoint, task
from langgraph.types import RetryPolicy

retry_policy = RetryPolicy(max_attempts=3, retry_on=TimeoutError)

@task(retry_policy=retry_policy)
def fetch_data(url: str):
    # Logic that might fail
    return {"data": "some data"}

@entrypoint()
def my_workflow(url: str):
    # Parallel execution pattern
    task1 = fetch_data(url)
    task2 = fetch_data(url + "/extra")
    
    # Wait for results
    res1 = task1.result()
    res2 = task2.result()
    return {"combined": [res1, res2]}