LangGraph：如何用“图思维”轻松管理多Agent协作？

架构精进之路

修改于 2025-04-08 06:58:08

68013

代码可运行

文章被收录于专栏：架构精进之路架构精进之路

运行总次数：3

代码可运行

引言

当AI任务变得复杂时，我们需要更好的“调度员”。

随着智能应用场景的深化，单一 Agent 在处理复杂任务（如电商智能客服、金融数据分析流水线）时显现出明显局限性。传统链式调用框架（如 LangChain）依赖开发者手动编排流程，在面对任务分支、动态决策和资源复用等场景时，往往陷入维护成本高、扩展性差的困境。

本文将深入解析：

1. LangGraph的核心原理：图结构如何模拟复杂任务流？

2. 多 Agent 应用实战：如何实现异构Agent的协作与通信？

3. 对比LangChain：何时该用LangGraph？

1、LangGraph核心原理

LangGraph 的核心原理，简单说就是：用“图”来重构任务流。

1.1. 图结构（Graph）的三大要素

节点（Node）：代表一个独立单元，可以是：
- Agent 节点：封装独立 Agent 能力（如调用GPT-4处理问题、知识检索）
- Tool 节点：调用具体工具（如搜索API、数据库查询），用于执行特定计算或处理任务
- END 节点：流程结束的标识
边（Edge）：标注 “Where to go”，表示状态流转的决策路径，决定下一步跳转到哪个节点
- 顺序执行（线性流程）
- 条件跳转（如“若用户投诉，转人工审核”）
状态（State）：贯穿整个流程，记录数据或交互状态（如用户输入的消息），驱动节点间的流转

针对如上图例，流程说明如下：

1）用户输入触发初始状态，进入 Agent 节点处理交互；

2）处理后通过 Edge 决策流向，可能跳转至 Tool 节点调用工具，或直接到达 END 节点结束流程；

3）若进入 Tool 节点，处理完成后状态可能再次流转，形成循环或推进至结束。该模型通过节点、边、状态的协作，实现交互任务的流程化处理。

1.2 动态编排：像流程图一样设计AI流程

LangGraph支持循环、分支、并行等控制流，例如：

# 示例：根据用户问题动态路由  
def route(state):  
    if "投诉" in state["input"]:  
        return "human_review"  # 转人工  
    else:  
        return "ai_agent"      # 由AI处理

2、多 Agent 协作实战

为了更好让大家理解，我们应用一个「客服工单处理案例」来进行介绍。

假设我们需要处理用户工单，主要流程包括：意图识别 → 专家Agent处理 → 结果审核

2.1 定义 Agent 节点

from langgraph.graph import StateGraph, MessagesState  
# Create the workflow
workflow = StateGraph(MessagesState)
# 意图识别  
def identify_intent(state):
    last_message = state["messages"][-1].content
    if "账号" in last_message or "登录" in last_message:
        intent = "technical"
    else:
        intent = "general"
    return {
        "messages": [AIMessage(content="意图识别完成")],
        "intent": intent
    }
# 专家处理
def call_expert(state):
    return {"messages": [AIMessage(content="专家处理结果：您的问题需要专业技术支持")]}
# 人工审核
def human_review(state):
    return {"messages": [AIMessage(content="人工审核完成：您的问题已处理")]}
    
    
# 添加三个Agent节点  
workflow.add_node("intent_agent", identify_intent)  # 意图识别  
workflow.add_node("expert_agent", call_expert)      # 专家处理  
workflow.add_node("review_agent", human_review)     # 人工审核

2.2 设计 workflow 逻辑

# 设置入口点
workflow.set_entry_point("intent_agent")
# 条件边：根据意图选择路径  
def router(state):  
    if state["intent"] == "technical":  
        return "expert_agent"  
    else:  
        return "review_agent"  
workflow.add_conditional_edges("intent_agent", router)  
# 专家处理后必须审核
workflow.add_edge("expert_agent", "review_agent")  
# 设置结束点
workflow.set_finish_point("review_agent")

2.3 运行 workflow

# Compile the workflow
app = workflow.compile()
output = app.invoke({"messages": [HumanMessage(content="我的账号无法登录")]})
print("执行结果：\n\n", final_state["messages"][-1].content)