---

name: langgraph description: LangGraph 专家——用于构建有状态、多角色 AI 应用的生产级框架。涵盖图构建、状态管理、循环与分支、检查点持久化、人机协作模式和 ReAct agent 模式。 risk: unknown source: vibeship-spawner-skills (Apache 2.0) date_added: 2026-02-27

LangGraph

LangGraph 专家——用于构建有状态、多角色 AI 应用的生产级框架。涵盖图构建、状态管理、循环与分支、检查点持久化、人机协作模式和 ReAct agent 模式。已在 LinkedIn、Uber 和 400+ 家公司的生产环境中使用。这是 LangChain 推荐的构建 agent 的方法。

角色：LangGraph Agent 架构师

你是使用 LangGraph 构建生产级 AI agent 的专家。你理解 agent 需要显式结构——图使流程可见且可调试。你精心设计状态，恰当使用 reducer，并始终考虑生产环境的持久化。你知道何时需要循环以及如何防止无限循环。

专业领域

• 图拓扑设计
• 状态 schema 模式
• 条件分支
• 持久化策略
• 人机协作
• 工具集成
• 错误处理与恢复

能力

• 图构建（StateGraph）
• 状态管理和 reducer
• 节点和边定义
• 条件路由
• 检查点和持久化
• 人机协作模式
• 工具集成
• 流式和异步执行

前置条件

• 0: Python 熟练
• 1: LLM API 基础
• 2: 异步编程概念
• 3: 图论基础
• 必需技能：Python 3.9+、langgraph 包、LLM API 访问（OpenAI、Anthropic 等）、图概念理解

范围

• 0: 仅 Python（TypeScript 处于早期阶段）
• 1: 图概念有学习曲线
• 2: 状态管理复杂度
• 3: 调试可能具有挑战性

生态系统

主要

• LangGraph
• LangChain
• LangSmith（可观测性）

常见集成

• OpenAI / Anthropic / Google
• Tavily（搜索）
• SQLite / PostgreSQL（持久化）
• Redis（状态存储）

平台

• Python 应用
• FastAPI / Flask 后端
• 云部署

模式

基础 Agent 图

带工具的简单 ReAct 风格 agent

使用场景：带工具调用的单 agent

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

# 1. 定义状态
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    # add_messages reducer 追加而非覆盖

# 2. 定义工具
@tool
def search(query: str) -> str:
    """Search the web for information."""
    # 实现代码
    return f"Results for: {query}"

@tool
def calculator(expression: str) -> str:
    """Evaluate a math expression."""
    return str(eval(expression))

tools = [search, calculator]

# 3. 创建带工具的 LLM
llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)

# 4. 定义节点
def agent(state: AgentState) -> dict:
    """Agent 节点 - 调用 LLM。"""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# 工具节点处理工具执行
tool_node = ToolNode(tools)

# 5. 定义路由
def should_continue(state: AgentState) -> str:
    """根据是否调用了工具进行路由。"""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

# 6. 构建图
graph = StateGraph(AgentState)

# 添加节点
graph.add_node("agent", agent)
graph.add_node("tools", tool_node)

# 添加边
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, ["tools", END])
graph.add_edge("tools", "agent")  # 循环回去

# 编译
app = graph.compile()

# 7. 运行
result = app.invoke({
    "messages": [("user", "What is 25 * 4?")]
})

带 Reducer 的状态

使用自定义 reducer 的复杂状态管理

使用场景：多个 agent 更新共享状态

from typing import Annotated, TypedDict
from operator import add
from langgraph.graph import StateGraph

# 自定义 reducer 用于合并字典
def merge_dicts(left: dict, right: dict) -> dict:
    return {**left, **right}

# 带多个 reducer 的状态
class ResearchState(TypedDict):
    # 消息追加（不覆盖）
    messages: Annotated[list, add_messages]

    # 研究发现合并
    findings: Annotated[dict, merge_dicts]

    # 来源累积
    sources: Annotated[list[str], add]

    # 当前步骤（覆盖 - 无 reducer）
    current_step: str

    # 错误计数（自定义 reducer）
    errors: Annotated[int, lambda a, b: a + b]

# 节点返回部分状态更新
def researcher(state: ResearchState) -> dict:
    # 仅返回正在更新的字段
    pass