Tutorial de inicio de LangGraph

LangGraph es un framework desarrollado por el equipo de LangChainMarco de orquestación de agentes de bajo nivel, diseñado específicamente para construir flujos de trabajo de IA con estado (Stateful) y de larga duración.

A diferencia de las cadenas lineales tradicionales de llamadas a LLM, LangGraph modela el flujo de trabajo comoun grafo dirigido (Directed Graph)：

• Nodo (Node): función que ejecuta una operación específica (como llamar a un LLM, ejecutar herramientas, procesar datos)
• Arista (Edge): define las rutas de transición entre nodos, admite ramificaciones condicionales
• Estado (State): datos compartidos y transmitidos a lo largo de todo el flujo de trabajo

Dirección del código abierto:https://github.com/langchain-ai/langgraph。

Imagina que estás dirigiendo una actuación sinfónica: la cadena tradicional de LLM es como tocar una pieza de principio a fin, solo se puede reproducir en orden; LangGraph, en cambio, es como un director de orquesta que puede ajustar el orden de la interpretación en cualquier momento según la reacción del público, hacer que un movimiento se repita o saltar a un pasaje específico. Esto da a los flujos de trabajo de IA la sabiduría de un "director": ser capaz de hacer bucles, ramificarse y retroceder, logrando realmente una toma de decisiones autónoma compleja.

¿Por qué elegir LangGraph?

La siguiente tabla compara las principales diferencias entre la cadena de LLM tradicional y LangGraph:

Casos de uso

• Chatbot: necesita recordar el contexto de conversaciones de múltiples turnos
• Agente autónomo: capaz de planificar, usar herramientas, pensar de forma iterativa
• Sistema multi-agente: múltiples IA colaboran para completar tareas complejas
• Flujo de trabajo de aprobación: procesos automatizados que requieren revisión humana
• Asistente de investigación: requiere razonamiento de múltiples pasos y recuperación de información

Conceptos clave

Antes de empezar a escribir código, comprendamos los tres conceptos clave de LangGraph.

Graph (Grafo)

Graph es el plano de todo el flujo de trabajo, define la estructura lógica completa del agente. Está compuesto por nodos (Nodes) y aristas (Edges):

StateGraph
   |-- Nodes（节点）
   |     |-- node_a
   |     |-- node_b
   |     +-- node_c
   +-- Edges（边）
         |-- START -> node_a
         |-- node_a -> node_b（条件边）
         |-- node_a -> node_c（条件边）
         +-- node_b -> END

State (Estado)

State es, a lo largo de todo el grafo, laestructura de datos compartida. Cada nodo puede leer y actualizar State; el State actualizado se pasará al siguiente nodo.

from typing import TypedDict, Annotated
from langgraph.graph import add_messages

class MyState(TypedDict):
    messages: Annotated[list, add_messages]  # 消息列表（自动追加）
    user_name: str                            # 用户名称
    step_count: int                           # 步骤计数

Nodes (Nodos)

Los nodos son funciones comunes de Python que reciben el State actual y devuelven el State actualizado (campos parciales).

def my_node(state: MyState) -> dict:
    # 读取状态
    messages = state["messages"]
    
    # 执行操作...
    result = "处理结果"
    
    # 返回更新的字段（不需要返回所有字段）
    return {"messages": [{"role": "ai", "content": result}]}

Edges (Aristas)

Las aristas definen la forma de transición entre nodos:

• Arista normal: ruta fija,node_a -> node_b
• Arista condicional: enrutamiento dinámico según State,node_a -> node_b o node_c
• Arista de inicio：START -> 第一个节点
• Arista de fin：某节点 -> END

Configuración del entorno

Instalación de dependencias

Instale LangGraph y las dependencias relacionadas usando un espejo nacional (mirror doméstico):

# 创建虚拟环境（推荐）
python -m venv venv
source venv/bin/activate      # macOS/Linux
# venv\Scripts\activate       # Windows

# 安装 LangGraph 和 LangChain
pip install langgraph langchain langchain-openai python-dotenv -i https://mirrors.aliyun.com/pypi/simple/

# 可选：安装开发工具
pip install langgraph-cli jupyter -i https://mirrors.aliyun.com/pypi/simple/

Configuración de la API Key

Cree en la raíz del proyecto.envel archivo, y configure tanto OpenAI como DeepSeek:

# .env 文件内容

# OpenAI 配置（国外用户）
OPENAI_API_KEY=sk-xxx
OPENAI_BASE_URL=https://api.openai.com/v1

# DeepSeek 配置（国内用户推荐）
DEEPSEEK_API_KEY=sk-xxx
DEEPSEEK_BASE_URL=https://api.deepseek.com
DEEPSEEK_MODEL=deepseek-v4-pro

Verificar la instalación

import langgraph
print(f"LangGraph 版本: {langgraph.__version__}")

Primer programa LangGraph

Empecemos con el ejemplo más simple: un flujo de trabajo lineal con solo dos nodos.

Resultado de ejecución:

[greet_node] 收到消息: 世界
[process_node] 处理消息: 你好！世界

最终结果: {'message': '你好！世界', 'processed': True}

Visualizar la estructura del grafo

En Jupyter Notebook se puede visualizar directamente la estructura del grafo:

# 在 Jupyter Notebook 中可视化
from IPython.display import Image
Image(graph.get_graph().draw_mermaid_png())

# 或者打印 Mermaid 格式
print(graph.get_graph().draw_mermaid())

Gestión de State (estado)

Definir estado con TypedDict

from typing import TypedDict, Annotated, Optional
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    # 消息历史（add_messages reducer 自动追加而非覆盖）
    messages: Annotated[list, add_messages]
    
    # 普通字段（直接覆盖）
    user_id: str
    session_id: str
    
    # 可选字段
    error: Optional[str]
    
    # 计数器（使用 operator.add 作为 reducer）
    retry_count: Annotated[int, lambda x, y: x + y]

Definir estado con Pydantic (recomendado para producción)

from pydantic import BaseModel, Field
from typing import Annotated
from langgraph.graph.message import add_messages

class ProductionState(BaseModel):
    messages: Annotated[list, add_messages] = Field(default_factory=list)
    user_id: str = ""
    confidence_score: float = 0.0
    
    class Config:
        arbitrary_types_allowed = True

MessagesState (estado abreviado integrado)

LangGraph proporciona un estado integradoMessagesState, diseñado específicamente para escenarios de diálogo:

from langgraph.graph import MessagesState

# MessagesState 等价于:
# class MessagesState(TypedDict):
#     messages: Annotated[list[AnyMessage], add_messages]

# 直接使用，无需自定义
builder = StateGraph(MessagesState)

Nodes (Nodos)

Nodos de función normal

def simple_node(state: AgentState) -> dict:
    # 读取状态
    last_message = state["messages"][-1]
    
    # 执行操作
    response = f"收到: {last_message.content}"
    
    # 返回部分状态更新
    return {
        "messages": [{"role": "assistant", "content": response}]
    }

Nodo de llamada a LLM

Nodos asíncronos

import asyncio

async def async_node(state: AgentState) -> dict:
    """异步节点，适合 I/O 密集型操作"""
    # 模拟异步操作（如 API 调用、数据库查询）
    await asyncio.sleep(0.1)
    
    result = await some_async_api_call(state["messages"][-1].content)
    return {"messages": [{"role": "assistant", "content": result}]}

# 使用异步图
result = await graph.ainvoke({"messages": [...]})

Usar una clase como nodo

class RouterNode:
    def __init__(self, llm, system_prompt: str):
        self.llm = llm
        self.system_prompt = system_prompt
    
    def __call__(self, state: AgentState) -> dict:
        """类实例可以作为节点使用"""
        messages = [
            SystemMessage(content=self.system_prompt),
            *state["messages"]
        ]
        response = self.llm.invoke(messages)
        return {"messages": [response]}

# 添加类节点
router = RouterNode(llm, "你是一个专业的路由助手。")
builder.add_node("router", router)

Edges (Aristas) y enrutamiento condicional

Arista normal

# 固定路径：node_a 完成后始终执行 node_b
builder.add_edge("node_a", "node_b")

# 结束：node_a 完成后图结束
builder.add_edge("node_a", END)

Arista condicional

La arista condicional es una función central de LangGraph; decide dinámicamente el siguiente paso según el State actual.

def route_after_llm(state: AgentState) -> str:
    """
    路由函数：根据 LLM 的最新输出决定走哪条路径
    返回值必须是已注册节点名称或 END
    """
    last_message = state["messages"][-1]
    
    # 如果 LLM 请求使用工具
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    
    # 否则结束
    return END

# 添加条件边
builder.add_conditional_edges(
    "llm",              # 源节点
    route_after_llm,    # 路由函数
    {
        "tools": "tool_executor",   # 返回 "tools" 时 -> tool_executor 节点
        END: END                    # 返回 END 时 -> 结束
    }
)

Ejecución en paralelo (Fan-out)

# 从一个节点并行分叉到多个节点
builder.add_edge("start_node", "branch_a")
builder.add_edge("start_node", "branch_b")
builder.add_edge("start_node", "branch_c")

# 多个节点汇聚到一个节点（Fan-in）
builder.add_edge("branch_a", "merge_node")
builder.add_edge("branch_b", "merge_node")
builder.add_edge("branch_c", "merge_node")

Ejemplo completo de enrutamiento condicional

Ejemplo de resultado de ejecución:

Usuario: 北京今天天气怎么样？
Asistente: 很抱歉，我没有实时天气数据。不过北京现在是春季，建议您出门时关注天气预报...
--------------------------------------------------

Usuario: 帮我写一个 Python 快速排序
Asistente: 好的，下面是一个 Python 实现的快速排序算法：

```python
def quicksort(arr):
    if len(arr) <= 1:
        return arr
    ...
--------------------------------------------------

Usuario: 你好，介绍一下你自己
Asistente: 你好！我是一个 AI 助手，很高兴为你服务。我可以帮你回答各种问题...
--------------------------------------------------

Usuario: 再见啦！
Asistente: 再见！期待下次与你交流。
--------------------------------------------------

Construir un chatbot

Ahora, con lo aprendido, construyamos un robot que admita conversaciones de múltiples turnos. Este robot puede recordar el contexto de la conversación y ofrecer una experiencia de interacción continua.

Ejemplo de resultado de ejecución:

Tú: 你好，我叫小明
Asistente: 你好，小明！很高兴认识你。有什么我可以帮助你的吗？

Tú: 我想学习 Python 编程
Asistente: 太好了！Python 是一门很适合初学者的编程语言。我可以帮你从基础开始：
1. 首先了解变量和数据类型
2. 学习条件语句和循环
3. 掌握函数的定义和使用
你想从哪个部分开始呢？

Tú: 你还记得我叫什么吗？
Asistente: 当然记得，你叫小明！你刚才说想学习 Python 编程，我们可以继续这个话题。

Tú: 退出
再见！

Llamada a herramientas - ReAct Agent

Dotar a los agentes de la capacidad de usar herramientas externas es una de las características más potentes de LangGraph.ReAct(Reason + Act) es el patrón de agente más común: el LLM piensa -> elige una herramienta -> ejecuta la herramienta -> observa el resultado -> continúa pensando.

Definir herramientas

Primero, definamos las herramientas que el agente puede usar:

from langchain_core.tools import tool

@tool
def search_web(query: str) -> str:
    """搜索网络获取最新信息。
    
    Args:
        query: 搜索关键词
    
    Returns:
        搜索结果摘要
    """
    # 实际项目中替换为真实搜索 API
    return f"关于 '{query}' 的搜索结果：这是模拟的搜索结果..."

@tool
def calculate(expression: str) -> str:
    """计算数学表达式。
    
    Args:
        expression: 数学表达式，如 '2 + 2' o  '100 * 0.8'
    
    Returns:
        计算结果
    """
    import ast
    import operator
    
    # 安全的运算符映射
    ops = {
        ast.Add: operator.add,
        ast.Sub: operator.sub,
        ast.Mult: operator.mul,
        ast.Div: operator.truediv,
        ast.Pow: operator.pow,
        ast.USub: operator.neg,
    }
    
    def safe_eval(node):
        if isinstance(node, ast.Expression):
            return safe_eval(node.body)
        elif isinstance(node, ast.Constant):
            return node.value
        elif isinstance(node, ast.BinOp):
            left = safe_eval(node.left)
            right = safe_eval(node.right)
            return ops[type(node.op)](left, right)
        elif isinstance(node, ast.UnaryOp):
            operand = safe_eval(node.operand)
            return ops[type(node.op)](operand)
        else:
            raise ValueError(f"不支持的表达式类型: {type(node)}")
    
    try:
        tree = ast.parse(expression, mode='eval')
        result = safe_eval(tree)
        return f"计算结果: {expression} = {result}"
    except Exception as e:
        return f"计算错误: {str(e)}"

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息。
    
    Args:
        city: 城市名称
    
    Returns:
        天气信息
    """
    # 实际项目中替换为真实天气 API
    return f"{city} 今日天气：晴，温度 22C，湿度 60%"

tools = [search_web, calculate, get_weather]

Construir un ReAct Agent

Transmisión en flujo del proceso de llamada a herramientas

# 流式观察 Agent 的每一步
for chunk in graph.stream(
    {"messages": [HumanMessage(content="搜索 LangGraph 的最新特性")]},
    stream_mode="updates"
):
    for node_name, updates in chunk.items():
        print(f"\n=== 节点: {node_name} ===")
        for msg in updates.get("messages", []):
            if hasattr(msg, "tool_calls") and msg.tool_calls:
                for tc in msg.tool_calls:
                    print(f"  -> 调用工具: {tc['name']}({tc['args']})")
            else:
                print(f"  -> 输出: {msg.content[:200] if msg.content else '(无内容)'}")

Human-in-the-Loop (colaboración humano-máquina)

LangGraph admite de forma nativa la pausa durante la ejecución del flujo de trabajo, esperando una revisión o entrada humana. Esto es muy útil para operaciones sensibles que requieren confirmación humana.

Usar interrupt para pausar la ejecución

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

def sensitive_action_node(state: MessagesState) -> dict:
    """执行敏感操作前请求人工审批"""
    last_msg = state["messages"][-1].content
    
    # 暂停图的执行，等待人工决策
    human_decision = interrupt({
        "question": "是否批准执行以下操作？",
        "action": last_msg,
        "risk_level": "中等"
    })
    
    if human_decision == "approve":
        return {"messages": [{"role": "assistant", "content": "操作已批准并执行完毕。"}]}
    else:
        return {"messages": [{"role": "assistant", "content": "操作已取消。"}]}

# 必须使用 checkpointer 才能支持 interrupt
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

Flujo de trabajo de aprobación completo

Establecer puntos de interrupción en las aristas

# 另一种方式：在编译时指定断点
graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["sensitive_node"],   # 执行该节点前暂停
    # interrupt_after=["review_node"],     # 执行该节点后暂停
)

Memoria persistente

LangGraph proporciona un mecanismo integrado de persistencia de estado que permite al agente recordar el historial de conversación entre sesiones.

Almacenamiento en memoria (adecuado para desarrollo y pruebas)

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# 使用 thread_id 区分不同会话
config_user_a = {"configurable": {"thread_id": "user-alice"}}
config_user_b = {"configurable": {"thread_id": "user-bob"}}

# Alice 的对话
graph.invoke({"messages": [HumanMessage(content="我叫 Alice")]}, config=config_user_a)
graph.invoke({"messages": [HumanMessage(content="我叫什么名字？")]}, config=config_user_a)
# Agent 能记住：你叫 Alice

# Bob 的对话完全独立
graph.invoke({"messages": [HumanMessage(content="我叫什么名字？")]}, config=config_user_b)
# Agent 不知道 Bob 的名字（不同 thread_id）

Almacenamiento persistente con SQLite (adecuado para proyectos locales)

# 安装依赖
# pip install langgraph-checkpoint-sqlite

from langgraph.checkpoint.sqlite import SqliteSaver

# 数据持久化到文件，程序重启后对话历史仍存在
with SqliteSaver.from_conn_string("./chat_memory.db") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)
    
    config = {"configurable": {"thread_id": "persistent-chat"}}
    
    # 第一次运行
    graph.invoke({"messages": [HumanMessage(content="我叫张三")]}, config=config)
    
    # 程序重启后再次运行，记忆仍然存在
    result = graph.invoke(
        {"messages": [HumanMessage(content="你还记得我叫什么吗？")]},
        config=config
    )

Ver historial de conversación

# 获取某个 thread 的完整状态历史
history = list(graph.get_state_history(config))

for snapshot in history:
    print(f"时间: {snapshot.created_at}")
    print(f"消息数: {len(snapshot.values['messages'])}")
    print("---")

# 获取当前状态
current_state = graph.get_state(config)
print(f"当前消息数: {len(current_state.values['messages'])}")

Sistemas multi-agente

LangGraph destaca en coordinar múltiples agentes especializados para trabajar en conjunto. Al descomponer tareas complejas entre diferentes agentes expertos, se puede lograr una capacidad de resolución de problemas más potente.

Arquitectura maestro-esclavo (Patrón Supervisor)

Subgrafo (Subgraph)

Encapsular subprocesos complejos como subgrafos para reutilizarlos en el grafo principal:

# 将复杂子流程封装为子图，在主图中复用
sub_builder = StateGraph(MessagesState)
sub_builder.add_node("step1", step1_node)
sub_builder.add_node("step2", step2_node)
sub_builder.add_edge(START, "step1")
sub_builder.add_edge("step1", "step2")
sub_builder.add_edge("step2", END)
sub_graph = sub_builder.compile()

# 在主图中使用子图
main_builder = StateGraph(MessagesState)
main_builder.add_node("preprocessing", preprocess_node)
main_builder.add_node("sub_workflow", sub_graph)  # 直接使用编译好的子图
main_builder.add_node("postprocessing", postprocess_node)

main_builder.add_edge(START, "preprocessing")
main_builder.add_edge("preprocessing", "sub_workflow")
main_builder.add_edge("sub_workflow", "postprocessing")
main_builder.add_edge("postprocessing", END)

main_graph = main_builder.compile()

Depuración visual con LangGraph Studio

LangGraph Studio es el entorno de desarrollo visual proporcionado oficialmente, que te permite ver el proceso de ejecución del agente en tiempo real, mejorando significativamente la eficiencia de desarrollo y depuración.

Instalar LangGraph CLI

pip install langgraph-cli -i https://mirrors.aliyun.com/pypi/simple/

Crear archivo de configuración del proyecto

Crear en el directorio raíz del proyectolanggraph.jsonArchivo de configuración:

{
  "dependencies": ["."],
  "graphs": {
    "my_agent": "./my_agent.py:graph"
  },
  "env": ".env"
}

Iniciar servidor de desarrollo

langgraph dev

Después de iniciar, accede ahttp://localhost:8123para usar LangGraph Studio en el navegador.

Funciones principales de Studio

• Visualización en tiempo real: muestra gráficamente el proceso de ejecución de los nodos, permitiendo comprender de forma intuitiva el estado del flujo de trabajo
• Inspección de estado: pausar en cualquier nodo para ver el State actual, facilitando la resolución de problemas
• Viaje en el tiempo: reproducir los pasos de ejecución históricos y rastrear los cambios de estado en cada paso
• Recarga en caliente: actualiza automáticamente la estructura del grafo después de modificar el código, sin necesidad de reiniciar el servicio

Mejores prácticas y preguntas frecuentes

Mejores prácticas

Puntos clave del diseño de estado

• Mantener el State conciso, incluyendo solo los campos necesarios
• Definir reducers claros para campos complejos (por ejemplo,add_messages）
• Usar modelos Pydantic para validar los tipos de estado en entornos de producción
• Evitar almacenar objetos demasiado grandes en el State; considerar el uso de almacenamiento externo

Puntos clave del diseño de nodos

• Cada nodo debe tener una única responsabilidad, facilitando las pruebas y la reutilización
• Las funciones de nodo deben ser idempotentes (la misma entrada produce la misma salida)
• Evitar modificar directamente el state recibido en el nodo; en su lugar, devolver nuevos valores
• Usar adecuadamente nodos asíncronos para operaciones intensivas en E/S

Manejo de errores

def robust_node(state: AgentState) -> dict:
    try:
        result = risky_operation(state)
        return {"messages": [result], "error": None}
    except Exception as e:
        return {
            "error": str(e),
            "messages": [{"role": "assistant", "content": f"操作失败: {e}"}]
        }

Evitar bucles infinitos

def route_with_limit(state: AgentState) -> str:
    # 设置最大重试次数，防止无限循环
    if state.get("retry_count", 0) >= 3:
        return END
    
    if needs_retry(state):
        return "retry_node"
    return END

Preguntas frecuentes (FAQ)

P1: ¿Qué hacer si el formato del valor de retorno del nodo no es correcto?

# 错误：直接修改 state 对象
def bad_node(state):
    state["messages"].append(...)  # 不要直接修改
    return state

# 正确：返回需要更新的字段
def good_node(state):
    return {"messages": [new_message]}  # 只返回变更字段

P2: ¿Cómo pasar datos temporales entre nodos?

Agregue datos temporales a la definición de State, o use la convención de prefijo de subrayado para campos internos:

from typing import TypedDict

class PublicState(TypedDict):
    messages: list  # 对外暴露

class PrivateState(TypedDict):
    messages: list
    _internal_cache: dict  # 以下划线开头约定为内部使用

Q3: ¿Cómo depurar el proceso de ejecución de nodos?

# 使用 stream 模式观察每个节点的输出
for event in graph.stream(initial_state, stream_mode="updates"):
    for node_name, state_update in event.items():
        print(f"\n[节点: {node_name}]")
        print(f"更新: {state_update}")

Q4: ¿Cuál es la diferencia entre StateGraph y MessageGraph?

MessageGraphEs una API de versiones anteriores con funciones limitadas. Ahora se recomienda usar de manera uniformeStateGraph, que es más flexible y completa. Si necesita procesar mensajes, useStateGraph(MessagesState)o personalice un State que contengamessagesel campo.

Resumen

Este artículo presenta de manera sistemática los conceptos centrales y las aplicaciones prácticas del framework LangGraph. LangGraph, a través de la orquestación de flujos de trabajo basados en estructura de grafo, permite a los desarrolladores construir agentes de IA complejos con capacidades de bucle, ramificación y persistencia de estado. En comparación con las cadenas LLM lineales tradicionales, LangGraph tiene ventajas significativas en escenarios que requieren razonamiento de múltiples pasos, colaboración humano-máquina y coordinación de múltiples agentes.

Ruta de aprendizaje recomendada