这次我们来看一个 AI Agent 系统选型的问题。当你准备动手搭建一个 AI Agent 时面对 Workflow、Router、Sub-agent、Skill 这些概念是不是感觉有点懵它们到底有什么区别在 LangGraph、LangChain 这些框架里又该怎么用这篇文章不绕弯子直接帮你理清思路告诉你什么场景该选什么组件以及如何用 LangGraph 把它们高效地组织起来。本文的核心是帮你做出可落地的技术决策。我们会先快速对比这几个核心组件的职责和适用场景然后重点解析 LangGraph 如何作为“总指挥”来编排它们最后给出从零搭建一个具备路由和技能调用能力的 Agent 的实操步骤。无论你是想快速验证一个想法还是为复杂业务构建稳定的 Agent 系统这里都有清晰的路径。1. 核心能力速览四大组件与 LangGraph 定位在深入细节前我们先通过一个表格快速把握这几个关键概念的核心职责和关系这是后续选型的基础。组件/概念核心职责类比适用场景在 LangGraph 中的角色Workflow (工作流)定义 Agent 执行任务的固定流程和步骤。工厂的生产线步骤固定。流程明确、步骤顺序固定的任务如数据提取-清洗-分析-报告。通过StateGraph定义节点和边构建图的骨架。Router (路由)根据当前状态或输入动态决定下一步执行哪个节点或子流程。交通调度中心根据路况决定车辆路线。处理复杂、分支多的任务需要根据上下文做决策如客服场景根据用户意图转接不同技能。通常实现为一个决策节点其run函数返回下一个节点的名称。Sub-agent (子智能体)一个具备特定能力的独立 Agent可被主 Agent 调用。公司的专业部门如财务部、技术部。模块化复杂系统将大问题分解由专精的子 Agent 处理子任务。作为 Workflow 中的一个功能节点其内部可能又是一个复杂的图。Skill (技能)Agent 可调用的一个具体工具或函数通常完成一个原子操作。员工的具体技能如使用Excel、编写代码。需要与外部系统交互、执行具体操作如搜索网络、查询数据库、调用API。通常封装在节点的run函数中或被Tool节点直接调用。LangGraph编排框架将以上组件连接、协调成一个可运行的整体系统。公司的CEO项目管理软件负责战略和协调各部门。构建任何需要状态管理、循环、分支、多Agent协作的复杂AI应用。基础设施提供图定义、状态管理、持久化、并发等底层支持。简单来说Workflow 是“骨架”规定了大体的行进路线。Router 是“决策大脑”在岔路口决定走哪条路。Sub-agent 是“专业团队”负责处理某个领域的复杂子问题。Skill 是“工具包”完成具体的、单一的动作。LangGraph 是“舞台和导演”提供了让上述角色协同演出的剧场和调度规则。2. 适用场景与使用边界理解了组件是什么下一步就是判断什么时候该用谁。选型错误可能导致系统过度复杂或能力不足。2.1 何时选择 Workflow当你需要处理流程清晰、步骤线性的任务时Workflow 是第一选择。它的优势在于可预测性和可维护性。典型场景文档处理流水线上传 - OCR识别 - 关键信息抽取 - 归档。内容生成流水线确定主题 - 搜集资料 - 生成大纲 - 撰写正文 - 润色排版。数据分析报告连接数据源 - 执行查询 - 数据清洗 - 生成图表 - 编写结论。使用边界Workflow 不擅长处理需要大量动态分支和上下文依赖决策的场景。如果你发现需要在一个流程中嵌入大量的“如果...就...”判断纯 Workflow 会变得臃肿且难以维护这时就需要引入 Router。2.2 何时引入 Router当你的 Agent 需要像人类一样根据对话或任务上下文做出不同反应时Router 就变得至关重要。它负责动态规划执行路径。典型场景智能客服用户输入问题 - Router 判断意图如“退货”、“咨询产品”、“投诉”- 路由到相应的处理模块。任务分解助手用户提出一个复杂需求如“帮我策划一次旅行”- Router 将其分解为“查询目的地”、“预订机票”、“安排行程”等子任务并决定执行顺序。编码助手根据用户指令“修复bug”、“添加功能”、“重构代码”路由到不同的代码分析或生成子流程。使用边界Router 的决策依赖于准确的意图识别或状态判断。如果判断逻辑过于复杂或模糊Router 可能做出错误路由导致任务失败。需要精心设计状态State和路由逻辑。2.3 何时设计 Sub-agent当系统复杂度上升单一 Agent 变得笨重或者需要复用特定领域的专家能力时就应该考虑 Sub-agent 模式。典型场景多模态处理一个主 Agent 接收用户请求调用“图像理解 Sub-agent”分析图片再调用“文本生成 Sub-agent”撰写描述。企业级系统一个“接待 Agent”将技术问题路由给“技术支持 Sub-agent”将合同问题路由给“法务 Sub-agent”。游戏 NPC每个 NPC 都是一个独立的 Sub-agent拥有自己的记忆、目标和行为树由主控系统协调。使用边界Sub-agent 增加了系统的通信开销和协调复杂度。对于简单任务使用 Skill 或简单的函数调用即可无需上升至 Sub-agent 层级。要明确界定主 Agent 和 Sub-agent 的通信协议和职责边界。2.4 何时封装 SkillSkill 是最基础的构建块任何需要与“外部世界”交互或执行确定性计算的操作都应封装为 Skill。典型场景工具调用搜索SerpAPI、计算Calculator、文件读写File IO。API 集成调用天气预报 API、发送邮件、操作数据库。原子操作字符串格式化、时间转换、简单的数据验证。使用边界Skill 应该是无状态或副作用明确的单一功能。避免在一个 Skill 里做太多事情。Skill 的实现质量直接决定了 Agent 执行具体任务的能力上限。2.5 LangGraph 的统领作用LangGraph 并非与上述组件并列而是承载它们的框架。无论你采用 Workflow、Router 还是 Sub-agent 模式最终都需要一个像 LangGraph 这样的框架来定义状态统一管理任务执行过程中的所有上下文数据。编排节点将 Workflow 的步骤、Router 的决策点、Sub-agent 的调用、Skill 的执行定义为图中的节点。控制流通过条件边conditional_edge实现 Router 逻辑通过普通边实现线性 Workflow。持久化与并发支持长时间运行的任务、暂停、恢复以及多线程/异步处理。总结一下选型心法从 Skill 开始组合成线性 Workflow当流程需要分支时引入 Router当模块足够复杂且独立时升格为 Sub-agent而 LangGraph 是你从头到尾构建这个系统所依赖的蓝图和运行时。3. 环境准备与前置条件在开始用 LangGraph 搭建系统之前需要准备好开发和运行环境。以下是一个通用的 Python 环境配置清单适用于大多数 AI Agent 开发场景。操作系统Windows 10/11, macOS, 或 Linux (推荐 Ubuntu)。LangGraph 是跨平台的。Python 版本Python 3.10 或 3.11。这是目前主流 AI 框架最兼容的版本避免使用 3.12 以上可能存在的未适配问题。包管理工具使用pip或更推荐的uv、poetry进行虚拟环境管理以隔离项目依赖。核心依赖langgraph: 本文的核心框架用于编排 Agent 工作流。langchain: 提供构建 Agent 所需的大量组件LLM 集成、Tools、Memory 等。虽然 LangGraph 可独立使用但结合 LangChain 生态更高效。openai或其他 LLM SDK根据你选择的模型提供商安装如openai,anthropic,cohere或本地模型库ollama。pydantic: LangGraph 的状态管理严重依赖 Pydantic 模型进行类型验证。开发工具一个趁手的 IDE 或编辑器如 VSCode、PyCharm。硬件要求AI Agent 开发本身对 GPU 没有强制要求因为 LLM 推理通常由云端 API 或本地模型服务提供。重点在于内存建议 8GB 以上用于处理复杂的逻辑和状态。网络稳定访问所选 LLM API如 OpenAI、Claude的网络环境。磁盘空间预留足够空间安装 Python 包和可能的本地模型文件如果使用 Ollama 等。4. 安装部署与启动方式LangGraph 本身是一个库不存在“启动服务”的概念。它的“启动”就是你的 Python 脚本开始运行。这里我们演示如何安装并初始化一个最基本的 LangGraph 环境。首先创建并激活一个虚拟环境以venv为例# 创建项目目录并进入 mkdir ai-agent-demo cd ai-agent-demo # 创建虚拟环境 python -m venv .venv # 激活虚拟环境 # Windows: .venv\Scripts\activate # Linux/macOS: source .venv/bin/activate接下来安装核心依赖。我们使用pip进行安装pip install langgraph langchain-openai pydantic如果你计划使用 OpenAI 的模型还需要设置你的 API 密钥。通常通过环境变量管理# 在命令行中设置临时 export OPENAI_API_KEYyour-api-key-here # Linux/macOS set OPENAI_API_KEYyour-api-key-here # Windows # 更推荐的做法是创建 .env 文件 echo OPENAI_API_KEYyour-api-key-here .env然后在 Python 代码中通过dotenv加载from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量至此基础环境就搭建完成了。LangGraph 应用的“启动”就是执行你编写的 Python 脚本。5. 功能测试与效果验证构建一个带路由的客服 Agent理论说再多不如动手试。我们将构建一个简单的智能客服 Agent它能够根据用户意图动态路由到不同的处理节点。这个例子将串联 Workflow、Router 和 Skill。5.1 定义状态与技能首先定义整个系统共享的状态并封装几个简单的技能函数。from typing import TypedDict, Annotated from langgraph.graph.message import add_messages import operator # 1. 定义状态这是贯穿整个图的数据结构 class State(TypedDict): # 消息历史LangGraph 提供了便捷的注解来处理消息列表 messages: Annotated[list, add_messages] # 路由决策结果用于存储下一步该去哪个节点 next: str # 2. 封装技能Skill def handle_refund_request(state: State): 处理退款请求的技能 latest_message state[“messages”][-1].content # 这里可以集成复杂的退款逻辑如查询订单、调用退款API等 # 此处简化为生成回复 response f“已收到您的退款请求{latest_message}。我们的客服专员将在24小时内联系您处理。” return {“messages”: [HumanMessage(contentresponse)], “next”: “end”} def handle_product_inquiry(state: State): 处理产品咨询的技能 latest_message state[“messages”][-1].content # 模拟查询产品数据库 response f“关于您咨询的产品{latest_message}。这是我们的旗舰产品支持30天无理由退换货目前有现货。” return {“messages”: [HumanMessage(contentresponse)], “next”: “end”} def handle_complaint(state: State): 处理投诉的技能 latest_message state[“messages”][-1].content response f“非常抱歉给您带来不好的体验{latest_message}。我们已经记录您的投诉并会优先处理请保持电话畅通。” return {“messages”: [HumanMessage(contentresponse)], “next”: “end”} def general_fallback(state: State): 默认回复技能 response “抱歉我暂时无法处理您的问题已为您转接人工客服。” return {“messages”: [HumanMessage(contentresponse)], “next”: “end”}5.2 实现路由决策逻辑这是 Router 的核心它根据当前对话内容决定下一步该执行哪个节点。from langchain_openai import ChatOpenAI # 初始化 LLM用于意图识别 llm ChatOpenAI(model“gpt-3.5-turbo”) def router(state: State): 路由节点分析用户意图决定下一步 latest_message state[“messages”][-1].content # 使用 LLM 进行意图分类。在实际应用中可以使用更精细的提示词和少样本学习。 prompt f“”” 请判断以下用户输入的意图类别只返回类别名称不要解释。 可选类别[“退款”, “产品咨询”, “投诉”, “其他”] 用户输入“{latest_message}” 意图类别是 “”” response llm.invoke(prompt) intent response.content.strip() # 根据意图返回下一个节点的名称 if “退款” in intent: return {“next”: “refund_node”} elif “产品咨询” in intent: return {“next”: “inquiry_node”} elif “投诉” in intent: return {“next”: “complaint_node”} else: return {“next”: “fallback_node”}5.3 构建并运行 LangGraph 工作流现在我们将 Skill 和 Router 组装成一个完整的 Workflow。from langgraph.graph import StateGraph, END # 1. 创建图的工作流构建器 workflow StateGraph(State) # 2. 添加节点 # 路由决策节点 workflow.add_node(“router”, router) # 各个技能处理节点 workflow.add_node(“refund_node”, handle_refund_request) workflow.add_node(“inquiry_node”, handle_product_inquiry) workflow.add_node(“complaint_node”, handle_complaint) workflow.add_node(“fallback_node”, general_fallback) # 3. 设置入口点所有对话都先经过路由节点 workflow.set_entry_point(“router”) # 4. 添加条件边根据路由节点的结果动态跳转到不同的技能节点 workflow.add_conditional_edges( “router”, # 源节点 # 这是一个函数它接收路由节点更新后的状态并返回下一个节点的名称 lambda state: state[“next”], # 映射关系返回的值 - 对应的节点 { “refund_node”: “refund_node”, “inquiry_node”: “inquiry_node”, “complaint_node”: “complaint_node”, “fallback_node”: “fallback_node”, } ) # 5. 从各个技能节点连接到结束点 workflow.add_edge(“refund_node”, END) workflow.add_edge(“inquiry_node”, END) workflow.add_edge(“complaint_node”, END) workflow.add_edge(“fallback_node”, END) # 6. 编译图得到可执行的应用 app workflow.compile() # 7. 运行测试 # 初始化状态 initial_state {“messages”: [HumanMessage(content“我买的衣服尺码不对想申请退款。”)], “next”: “”} # 执行图 final_state app.invoke(initial_state) # 查看结果 print(final_state[“messages”][-1].content) # 预期输出”已收到您的退款请求我买的衣服尺码不对想申请退款。。我们的客服专员将在24小时内联系您处理。”通过这个例子你可以清晰地看到Workflow由StateGraph定义包含router-处理节点-END的流程。Routerrouter节点根据 LLM 分析的意图动态决策。Skillhandle_refund_request等具体处理函数。LangGraph提供了StateGraph,add_node,add_conditional_edges,compile,invoke这一整套编排和执行机制。6. 接口 API 与批量任务一个成熟的 Agent 系统往往需要以服务的形式提供 API并能处理批量任务。虽然 LangGraph 核心是库但可以轻松地与 Web 框架结合。6.1 将 LangGraph App 封装为 FastAPI 服务以下示例展示如何将上面编译好的app暴露为 HTTP API。from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List # 假设上面定义的 app 已经存在 # from your_agent_builder import app app_fastapi FastAPI(title“AI Agent 客服 API”) class ChatRequest(BaseModel): message: str # 可以扩展更多参数如 session_id, user_id 等 class ChatResponse(BaseModel): response: str next_action: str # 例如 “end”, “request_more_info” 等 app_fastapi.post(“/chat”, response_modelChatResponse) async def chat_endpoint(request: ChatRequest): try: # 1. 构造初始状态 initial_state { “messages”: [HumanMessage(contentrequest.message)], “next”: “” } # 2. 调用 LangGraph App result app.invoke(initial_state) # 3. 提取最后一条 AI 回复 last_message result[“messages”][-1] response_text last_message.content if hasattr(last_message, ‘content’) else str(last_message) return ChatResponse(responseresponse_text, next_actionresult.get(“next”, “end”)) except Exception as e: raise HTTPException(status_code500, detailf“Agent 处理失败: {str(e)}”) # 启动命令 (需安装 fastapi 和 uvicorn): uvicorn api_main:app_fastapi --reload --host 0.0.0.0 --port 8000启动服务后你就可以通过POST /chat接口与 Agent 交互。6.2 批量任务处理对于批量处理任务如处理一批用户反馈你需要考虑并发和状态隔离。关键是为每个任务创建独立的执行线程或进程并管理好各自的状态。import asyncio from concurrent.futures import ThreadPoolExecutor from langgraph.checkpoint import MemorySaver # 1. 使用检查点存储器便于管理状态可选但对于复杂任务推荐 checkpointer MemorySaver() # 在编译图时传入 app_with_checkpoint workflow.compile(checkpointercheckpointer) def process_single_query(user_query: str, thread_id: str): 处理单个查询任务 # 使用唯一的 thread_id 来隔离不同任务的状态 config {“configurable”: {“thread_id”: thread_id}} initial_state {“messages”: [HumanMessage(contentuser_query)], “next”: “”} # 调用应用传入 config 以关联检查点 result app_with_checkpoint.invoke(initial_state, configconfig) return result[“messages”][-1].content async def batch_process(queries: List[str]): 批量处理查询列表 results [] # 使用线程池控制并发度避免过度消耗 API 额度或本地资源 with ThreadPoolExecutor(max_workers5) as executor: loop asyncio.get_event_loop() tasks [] for i, query in enumerate(queries): # 为每个任务生成唯一ID thread_id f“batch_task_{i}” task loop.run_in_executor(executor, process_single_query, query, thread_id) tasks.append(task) # 等待所有任务完成 results await asyncio.gather(*tasks, return_exceptionsTrue) # 处理结果和可能的异常 processed_results [] for r in results: if isinstance(r, Exception): processed_results.append(f“处理失败: {r}”) else: processed_results.append(r) return processed_results # 使用示例 if __name__ “__main__”: query_list [“我要退款”, “这个产品怎么用”, “服务质量太差了”, “今天天气怎么样”] final_results asyncio.run(batch_process(query_list)) for q, r in zip(query_list, final_results): print(f“Q: {q} - A: {r}”)批量任务的核心要点状态隔离使用thread_id或session_id确保不同任务的状态不会互相污染。并发控制使用ThreadPoolExecutor或asyncio.Semaphore限制同时进行的请求数防止对 LLM API 造成冲击或耗尽本地资源。错误处理批量任务中个别失败不应导致整个任务中断要做好异常捕获和重试机制。检查点对于长时任务利用 LangGraph 的Checkpointer实现中断恢复。7. 资源占用与性能观察AI Agent 系统的性能瓶颈主要不在 LangGraph 框架本身而在于其调用的组件尤其是 LLM。以下是需要重点观察的方面LLM API 调用开销延迟网络往返时间 LLM 生成时间。这是最主要的耗时部分。使用asyncio进行异步调用可以显著提升吞吐量。成本监控 Token 消耗。复杂的 Router 逻辑和多次 Tool 调用会显著增加 Token 使用量。限流注意 API 的 RPM每分钟请求数和 TPM每分钟 Token 数限制批量任务时需要做限速。本地资源占用内存LangGraph 状态和消息历史会驻留在内存中。对于高并发或长对话场景需要监控内存增长考虑定期清理历史或使用外部存储如数据库作为检查点后端。CPU本地工具函数Skill的计算开销。如果 Skill 涉及大量数据处理可能成为瓶颈。图执行性能节点优化确保每个节点尤其是 Router的逻辑高效。避免在节点内进行不必要的 IO 或复杂计算。状态设计State应该只包含必要的数据。过大的状态对象会在节点间传递时增加序列化/反序列化开销。并发与持久化使用checkpointer时存储后端内存、数据库的性能会影响图的执行速度。监控建议在关键节点添加日志记录输入、输出和耗时。使用time模块或logging记录每个invoke的总时间。对于 Web API使用 FastAPI 的中间件或类似 Prometheus 的指标系统来监控接口响应时间和错误率。8. 常见问题与排查方法在开发和运行 LangGraph Agent 时你可能会遇到以下典型问题。问题现象可能原因排查方式解决方案invoke后状态没有更新1. 节点函数没有正确返回更新后的状态字典。2. 返回的字典键名与State定义不匹配。1. 检查节点函数的返回值确保是dict。2. 打印返回值确认键名如messages,next拼写正确。确保节点函数返回{“state_key”: new_value}且state_key在State中有定义。条件路由不生效总是走默认分支1. Router 节点的返回值{“next”: “node_name”}中node_name与add_conditional_edges中定义的映射不匹配。2. Router 的逻辑判断有误。1. 打印 Router 函数返回的next值。2. 检查add_conditional_edges的映射字典。确保 Router 返回的字符串与映射字典的键完全一致。注意大小写和空格。错误State字段类型注解错误在Annotated中使用了不正确的缩减函数Reducer。检查State定义中Annotated的第二个参数。对于list类型的messages应使用add_messages。参考 LangGraph 文档为list、dict等类型使用正确的 Reducer如add_messages、operator.add。LLM 调用超时或失败1. 网络问题。2. API 密钥无效或额度不足。3. 请求速率超限。1. 检查网络连接。2. 检查 API 密钥和环境变量。3. 查看 LLM 服务商的控制台错误信息。1. 增加超时设置。2. 实现重试机制如tenacity库。3. 在批量任务中加入延迟。图编译失败1. 节点未定义就被边引用。2.State定义与节点函数输入/输出不兼容。仔细阅读编译时的错误堆栈信息。1. 确保所有在边中引用的节点都已通过add_node添加。2. 确保所有节点函数都接受State作为参数并返回一个更新State部分字段的字典。多轮对话中状态混乱没有正确使用或重置thread_id导致不同会话的状态互相覆盖。检查调用invoke时传入的config中的thread_id是否唯一。为每个独立的对话会话使用唯一的thread_id。对于 Web 服务可以使用用户 ID 或会话 ID。9. 最佳实践与使用建议基于上述分析和实践总结出以下构建稳健 AI Agent 系统的最佳实践始于简单迭代复杂不要一开始就设计庞大的图。从一个线性 Workflow 开始验证核心链路。然后逐步引入 Router 处理分支再将复杂的节点重构为 Sub-agent。精心设计 StateState是你的应用的“全局变量”。保持其精简只包含节点间需要传递的必要数据。使用 Pydantic 进行严格的类型校验可以在开发早期发现很多错误。Skill 要单一职责每个 Skill工具函数只做一件事并做好错误处理。这有利于测试、复用和替换。Router 决策可降级Router 的决策逻辑尤其是依赖 LLM 的要有降级策略。例如当 LLM 无法识别意图时应有一个默认路径如fallback_node或请求用户澄清。利用检查点对于任何可能长时间运行或需要中断恢复的 Agent务必使用Checkpointer。内存存储MemorySaver适合开发生产环境应考虑 Redis、PostgreSQL 等持久化存储。测试驱动开发为每个节点函数编写单元测试为整个图编写集成测试。模拟不同的输入状态验证路由逻辑和输出是否符合预期。监控与可观测性在关键节点记录日志和指标耗时、Token 数、决策路径。这对于调试复杂问题和优化性能至关重要。安全与合规权限控制Skill 中调用外部 API 或访问数据库时需遵循最小权限原则。内容过滤在 Agent 的最终输出前加入对有害、偏见或敏感内容的过滤层。用户数据妥善处理State中可能包含的用户个人信息遵守数据隐私法规。AI Agent 系统的构建是一个在“规划”与“反应”之间寻找平衡的艺术。Workflow 提供了规划和确定性Router 赋予了反应和灵活性Sub-agent 实现了模块化和复用Skill 夯实了能力基础而 LangGraph 则是那个让一切协同起来的粘合剂。从明确你所要解决问题的核心流程开始选择最匹配的组件小步快跑持续迭代你就能搭建出既强大又可控的智能体系统。