AI智能体框架实战：LangGraph状态图编排与多Agent协作机制设计

小编 • • 人工智能

AI智能体（Agent）开发正在从简单的Prompt链式调用演变为复杂的状态图编排。LangGraph作为LangChain团队推出的Agent编排框架，基于有向无环图（DAG）和循环图模型，解决了多Agent协作场景下的状态管理、条件分支、人工干预等工程问题。AI智能体的核心价值在于让大模型从被动回答转变为主动规划、调用工具、迭代执行的自主系统，而这背后的技术支撑就是状态图编排。

LangGraph核心概念与状态图模型

LangGraph的设计灵感来源于Pregel图计算框架和State Machine理论。每个Agent节点接收共享状态（State），执行特定逻辑后返回状态更新。与线性Chain不同，LangGraph允许定义循环边，使Agent能够在多轮交互中不断修正策略。

核心概念包括：

State（状态）：一个TypedDict或Pydantic Model，贯穿整个图的执行过程，所有节点共享读写。

Node（节点）：处理函数，接收当前State，返回部分状态更新。每个节点可以是一个LLM调用、一个工具执行，或一个完整的子图。

Edge（边）：定义节点间的流转关系。普通边是固定跳转，条件边根据State中的字段值动态决定下一个节点。

单Agent状态图搭建与工具调用绑定

以下是一个完整的单Agent实现，包含工具调用和条件路由：

from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from typing import TypedDict, Annotated, List
from langgraph.graph.message import add_messages

# 定义共享状态
class AgentState(TypedDict):
    messages: Annotated[List, add_messages]
    tool_call_count: int

# 定义工具
@tool
def search_db(query: str) -> str:
    """搜索内部数据库"""
    return f"查询结果：{query} 对应的记录为 128 条"

@tool
def calculate(expression: str) -> str:
    """执行数学计算"""
    try:
        result = eval(expression)  # 生产环境替换为安全计算库
        return str(result)
    except Exception:
        return "计算失败"

tools = [search_db, calculate]
llm = ChatOpenAI(model="gpt-4o", temperature=0)
llm_with_tools = llm.bind_tools(tools)

# 定义Agent节点
def agent_node(state: AgentState) -> dict:
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response], "tool_call_count": state.get("tool_call_count", 0)}

# 定义工具执行节点
def tool_node(state: AgentState) -> dict:
    last_msg = state["messages"][-1]
    results = []
    for tool_call in last_msg.tool_calls:
        tool_map = {t.name: t for t in tools}
        result = tool_map[tool_call["name"]].invoke(tool_call["args"])
        results.append(HumanMessage(content=result, name=tool_call["name"]))
    return {
        "messages": results,
        "tool_call_count": state["tool_call_count"] + 1
    }

# 条件路由：判断是否需要调用工具
def should_use_tool(state: AgentState) -> str:
    last_msg = state["messages"][-1]
    if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
        if state.get("tool_call_count", 0) >= 5:
            return END  # 防止无限循环
        return "tools"
    return END

# 构建状态图
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_use_tool)
workflow.add_edge("tools", "agent")  # 工具执行后回到Agent

graph = workflow.compile()

# 执行
result = graph.invoke({
    "messages": [HumanMessage(content="帮我搜索销量数据，然后计算月平均")],
    "tool_call_count": 0
})
print(result["messages"][-1].content)

这段代码实现了一个完整的ReAct模式Agent：LLM决定是否调用工具、执行工具、将结果反馈给LLM、循环直到给出最终答案。tool_call_count限制防止Agent陷入无限工具调用循环。

多Agent协作架构与消息路由设计

实际业务场景中，单个Agent难以处理复杂任务。LangGraph支持多Agent编排，核心思路是用一个Supervisor Agent作为路由器，将任务分发给专业子Agent。

from langgraph.graph import StateGraph, END, START
from typing import TypedDict, Annotated, List, Dict, Any
from langchain_core.messages import HumanMessage, AIMessage
import operator

class MultiAgentState(TypedDict):
    messages: Annotated[List, operator.add]
    next_agent: str
    task_completed: bool

def supervisor(state: MultiAgentState) -> dict:
    """supervisor决定下一个执行的Agent"""
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    last_msg = state["messages"][-1]
    
    routing_prompt = f"""
    根据用户请求，决定由哪个Agent处理：
    - research: 数据检索与信息搜集
    - coder: 代码编写与执行
    - reviewer: 代码审查与质量检查
    - FINISH: 任务完成
    
    用户请求: {last_msg.content}
    只返回一个词: research / coder / reviewer / FINISH
    """
    decision = llm.invoke(routing_prompt).content.strip()
    return {"next_agent": decision}

def research_agent(state: MultiAgentState) -> dict:
    return {"messages": [AIMessage(content="找到相关数据：用户增长23%")]}

def coder_agent(state: MultiAgentState) -> dict:
    return {"messages": [AIMessage(content="\n```python\nimport pandas as pd\ndf = pd.read_csv('data.csv')\nprint(df.head())\n```\n")]}

def reviewer_agent(state: MultiAgentState) -> dict:
    return {"messages": [AIMessage(content="代码审查通过，异常处理完善")]}

def route_from_supervisor(state: MultiAgentState) -> str:
    nxt = state["next_agent"]
    if nxt == "FINISH":
        return END
    return nxt

# 构建多Agent图
workflow = StateGraph(MultiAgentState)
workflow.add_node("supervisor", supervisor)
workflow.add_node("research", research_agent)
workflow.add_node("coder", coder_agent)
workflow.add_node("reviewer", reviewer_agent)

workflow.add_edge(START, "supervisor")
workflow.add_conditional_edges("supervisor", route_from_supervisor)
workflow.add_edge("research", "supervisor")
workflow.add_edge("coder", "supervisor")
workflow.add_edge("reviewer", "supervisor")

multi_agent_graph = workflow.compile()

Supervisor模式下，每个子Agent专注于自身领域，Supervisor根据当前状态决定下一步路由。这种架构的本质是将复杂任务分解为有向图遍历问题，每个Agent节点处理一个子任务，Supervisor节点负责编排和汇总。

人工干预节点与中断恢复机制

高风险场景（如金融决策、合约生成）需要在Agent执行链中插入人工审批环节。LangGraph通过interrupt机制实现：

from langgraph.checkpoint.memory import MemorySaver

def human_review(state: AgentState) -> dict:
    # 该节点暂停执行，等待人工输入
    pass

workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("review", human_review)
workflow.add_node("execute", lambda state: {"messages": [AIMessage(content="执行完毕")]})

workflow.set_entry_point("agent")
workflow.add_edge("agent", "review")
workflow.add_edge("review", "execute")
workflow.add_edge("execute", END)

checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)

# 执行到review节点会暂停
config = {"configurable": {"thread_id": "session-1"}}
result = app.invoke({"messages": [HumanMessage(content="生成合同草稿")]}, config=config)

# 人工审核后恢复执行
app.invoke(None, config=config)

MemorySaver会持久化每个节点的状态快照。当Agent在review节点暂停时，完整的上下文状态被保存。人工确认后，调用invoke(None)恢复执行，后续节点可以从检查点继续。生产环境中，checkpointer应替换为PostgresSaver或RedisSaver以支持跨进程恢复。

生产部署注意事项

状态序列化：自定义State中的Pydantic Model需要保证JSON可序列化，避免包含不可序列化的对象（如数据库连接池）。

并发控制：LangGraph原生支持Super-Step并行执行同一层级的多个节点。对于有副作用的工具（如写数据库），需在节点函数内加锁或使用幂等设计。

循环深度限制：条件边可能产生死循环。建议在State中维护迭代计数器，在条件函数中设置上限。recursion_limit参数也可以限制图的最大执行轮次：

result = graph.invoke(
    {"messages": [HumanMessage(content="复杂查询")]},
    config={"recursion_limit": 20}
)

可观测性：LangGraph与LangSmith深度集成，开启后每个节点的输入输出、token消耗、延迟数据都会自动上报。对于调试复杂Agent链，这是定位状态漂移和意外循环的关键手段。

原创文章，作者：小编，如若转载，请注明出处：https://www.yunthe.com/ai-zhi-neng-ti-kuang-jia-shi-zhan-langgraph-zhuang-tai-tu/

赞 (0)
小编小编
0
上一篇 7小时前
下一篇 7小时前

相关推荐