⚡

LangGraph（语言图）Skill langgraph

LangGraph是一个专为构建状态化、多参与者AI应用设计的生产级框架，支持图构造、状态管理、条件路由和持久化。它适用于开发AI智能体，实现可调试的代理流程，广泛应用于AI代理、状态管理、图计算和ReAct模式。关键词：LangGraph, AI智能体, 状态管理, 图构造, 代理开发, 生产级框架, 可调试AI

AI智能体 0 次安装 0 次浏览 更新于 3/21/2026

name: langgraph description: “LangGraph专家 - 用于构建状态化、多参与者AI应用的生产级框架。涵盖图构造、状态管理、周期和分支、使用检查点的持久化、人机交互模式以及ReAct代理模式。在LinkedIn、Uber和400多家公司中生产使用。这是LangChain推荐的构建代理的方法。使用场景：langgraph、langchain代理、状态化代理、代理图、react代理。” source: vibeship-spawner-skills (Apache 2.0)

LangGraph

角色: LangGraph代理架构师

您是一位使用LangGraph构建生产级AI代理的专家。您理解代理需要明确的结构 - 图使流程可见且可调试。您精心设计状态，适当使用归约器，并始终为生产考虑持久化。您知道何时需要周期以及如何防止无限循环。

能力

  • 图构造（StateGraph）
  • 状态管理和归约器
  • 节点和边定义
  • 条件路由
  • 检查点和持久化
  • 人机交互模式
  • 工具集成
  • 流式和异步执行

要求

  • Python 3.9+
  • langgraph包
  • LLM API访问（OpenAI、Anthropic等）
  • 图概念理解

模式

基础代理图

简单的ReAct风格代理与工具

使用时机: 单代理工具调用

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归约器追加，不覆盖

# 2. 定义工具
@tool
def search(query: str) -> str:
    """搜索网络获取信息。"""
    # 在此实现
    return f"结果查询: {query}"

@tool
def calculator(expression: str) -> str:
    """评估数学表达式。"""
    return str(eval(expression))

tools = [search, calculator]

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

# 4. 定义节点
def agent(state: AgentState) -> dict:
    """代理节点 - 调用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", "25 * 4等于多少？")]
})

带归约器的状态

复杂状态管理与自定义归约器

使用时机: 多个代理更新共享状态

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

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

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

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

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

    # 当前步骤（覆盖 - 无归约器）
    current_step: str

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

# 节点返回部分状态更新
def researcher(state: ResearchState) -> dict:
    # 只返回正在更新的字段
    return {
        "findings": {"topic_a": "新发现"},
        "sources": ["source1.com"],
        "current_step": "研究中"
    }

def writer(state: ResearchState) -> dict:
    # 访问累积状态
    all_findings = state["findings"]
    all_sources = state["sources"]

    return {
        "messages": [("assistant", f"基于{len(all_sources)}个来源的报告")],
        "current_step": "写作中"
    }

# 构建图
graph = StateGraph(ResearchState)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)
# ... 添加边

条件分支

基于状态路由到不同路径

使用时机: 多个可能的工作流

from langgraph.graph import StateGraph, START, END

class RouterState(TypedDict):
    query: str
    query_type: str
    result: str

def classifier(state: RouterState) -> dict:
    """分类查询类型。"""
    query = state["query"].lower()
    if "code" in query or "program" in query:
        return {"query_type": "coding"}
    elif "search" in query or "find" in query:
        return {"query_type": "search"}
    else:
        return {"query_type": "chat"}

def coding_agent(state: RouterState) -> dict:
    return {"result": "这是您的代码..."}

def search_agent(state: RouterState) -> dict:
    return {"result": "搜索结果..."}

def chat_agent(state: RouterState) -> dict:
    return {"result": "让我帮助..."}

# 路由函数
def route_query(state: RouterState) -> str:
    """路由到适当的代理。"""
    query_type = state["query_type"]
    return query_type  # 返回节点名称

# 构建图
graph = StateGraph(RouterState)

graph.add_node("classifier", classifier)
graph.add_node("coding", coding_agent)
graph.add_node("search", search_agent)
graph.add_node("chat", chat_agent)

graph.add_edge(START, "classifier")

# 从分类器的条件边
graph.add_conditional_edges(
    "classifier",
    route_query,
    {
        "coding": "coding",
        "search": "search",
        "chat": "chat"
    }
)

# 所有代理导向END
graph.add_edge("coding", END)
graph.add_edge("search", END)
graph.add_edge("chat", END)

app = graph.compile()

反模式

❌ 无退出的无限循环

为什么不好: 代理永远循环。 消耗代币和成本。 最终出错。

替代: 始终有退出条件：

  • 状态中的最大迭代计数器
  • 路由中的明确END条件
  • 应用级别超时

def should_continue(state): if state[“iterations”] > 10: return END if state[“task_complete”]: return END return “agent”

❌ 无状态节点

为什么不好: 失去LangGraph的好处。 状态未持久化。 无法恢复对话。

替代: 始终使用状态进行数据流。 从节点返回状态更新。 使用归约器进行累积。 让LangGraph管理状态。

❌ 巨型单体状态

为什么不好: 难以推理。 上下文中的不必要数据。 序列化开销。

替代: 使用输入/输出模式实现干净接口。 内部数据的私有状态。 清晰的责任分离。

限制

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

相关技能

良好配合: crewai, autonomous-agents, langfuse, structured-output