引言
当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处理 # 示例:根据用户问题动态路由 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(cnotallow="意图识别完成")], "intent": intent } # 专家处理 def call_expert(state): return {"messages": [AIMessage(cnotallow="专家处理结果:您的问题需要专业技术支持")]} # 人工审核 def human_review(state): return {"messages": [AIMessage(cnotallow="人工审核完成:您的问题已处理")]} # 添加三个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(cnotallow="我的账号无法登录")]}) print("执行结果:\n\n", final_state["messages"][-1].content)
执行结果:
图片
程序执行流程为:用户输入 → 意图识别 → 技术问题 → 专家Agent → 人工审核 → 返回结果
LangGraph 图例如下:
图片
3、LangChain vs LangGraph
当LangChain 遇上 LangGraph,我们应该如何选择呢?
图片
简单下个结论:
- 如果是固定流程(如“搜索→总结”),用LangChain更轻量。
- 如果是动态、多角色协作(如客服、数据分析流水线),LangGraph是更优解。
结语
LangGraph的图思维将复杂任务分解为可编排的节点,通过动态路由和共享状态实现高效协作。如果你正面临多Agent的“spaghetti code”难题,不妨用LangGraph重构你的流程!
图片