最近在调研和落地 AI Agent 项目时发现很多开发者包括我自己都面临一个共同的困惑面对 Workflow、Router、Sub-agent、Skill 这些眼花缭乱的概念到底该如何选择它们之间是什么关系是互斥的选项还是可以组合使用的组件尤其是在 LangGraph、LangChain 等框架的语境下这些术语频繁出现但官方文档往往侧重于单个概念的讲解缺乏一个从系统设计视角出发的、指导性的选型指南。本文旨在填补这一空白我将结合 LangGraph 框架为你全景式解析这些核心组件的本质、适用场景以及如何根据你的业务需求进行合理选型帮你构建一个清晰、高效且可维护的 AI Agent 系统。1. 核心概念拆解它们到底是什么在深入选型之前我们必须先统一对这几个核心概念的理解。它们并非同一维度的比较而是 Agent 系统中不同层次的抽象。1.1 AI Agent智能体的宏观定义AI Agent 是一个能够感知环境、进行决策并执行行动以实现特定目标的自治实体。它通常由几个核心模块构成大脑LLM负责推理、规划和决策。记忆Memory存储对话历史、知识或状态。工具Tools扩展 Agent 的能力边界使其能执行具体操作如搜索、计算、调用 API。规划与执行循环Orchestration协调上述组件的工作流程。我们讨论的 Workflow、Router 等都属于“规划与执行循环”这个模块下的具体实现模式或组件。1.2 Workflow工作流任务的编排蓝图Workflow描述的是一个有向的任务执行图。它定义了完成一个复杂目标所需的一系列步骤节点以及这些步骤之间的流转条件边。Workflow 强调的是过程的顺序性、依赖性和状态流转。本质一个预定义的、结构化的执行计划。类比就像一份烹饪食谱明确列出了“准备食材 - 切菜 - 热锅 - 翻炒 - 调味 - 出锅”的固定流程。在 LangGraph 中StateGraph就是用来定义和运行 Workflow 的核心抽象。你通过add_node添加步骤通过add_conditional_edges或add_edge定义流转逻辑。# LangGraph 中一个简单线性 Workflow 示例 from langgraph.graph import StateGraph, END from typing import TypedDict class State(TypedDict): input: str step1_result: str step2_result: str final_output: str def step1(state: State): return {step1_result: fProcessed: {state[input]}} def step2(state: State): return {step2_result: fRefined: {state[step1_result]}} # 构建 Workflow graph StateGraph(State) graph.add_node(step1, step1) graph.add_node(step2, step2) graph.add_edge(step1, step2) # 固定顺序step1 后总是执行 step2 graph.add_edge(step2, END) app graph.compile()1.3 Router路由动态决策的调度中心Router是一个决策节点它根据当前的状态State或输入动态地决定下一步应该执行哪个节点或分支。Router 强调的是运行时的条件判断和分支选择。本质一个条件分发器。类比像一个智能客服系统的分流器根据用户的问题关键词如“账单”、“故障”、“人工”将对话路由到不同的处理模块。在 LangGraph 中通常通过add_conditional_edges方法实现其核心是一个返回下一个节点名称的函数。from langgraph.graph import StateGraph, END from typing import Literal class State(TypedDict): query: str category: str def classifier_node(state: State): # 一个简单的分类器决定路由方向 query state[query].lower() if weather in query: return {category: weather_agent} elif calculate in query: return {category: math_agent} else: return {category: general_agent} def route_decision(state: State) - Literal[weather_agent, math_agent, general_agent, __end__]: # Router 的核心逻辑根据 state 中的 category 决定下一跳 category state.get(category) if category weather_agent: return weather_agent elif category math_agent: return math_agent elif category general_agent: return general_agent else: return __end__ graph StateGraph(State) graph.add_node(classifier, classifier_node) graph.add_node(weather_agent, lambda s: {response: Calling weather API...}) graph.add_node(math_agent, lambda s: {response: Calculating...}) graph.add_node(general_agent, lambda s: {response: Im a general assistant.}) graph.set_entry_point(classifier) # 关键设置条件边实现路由 graph.add_conditional_edges( classifier, route_decision, # 路由决策函数 { weather_agent: weather_agent, math_agent: math_agent, general_agent: general_agent, } ) # 各个 Agent 执行完后结束 graph.add_edge(weather_agent, END) graph.add_edge(math_agent, END) graph.add_edge(general_agent, END)1.4 Sub-agent子智能体模块化与职责分离Sub-agent是指在一个主 Agent 系统内部承担特定子任务或拥有特定领域专长的独立功能单元。它本身可以是一个具备完整感知-决策-执行循环的 Agent也可以是更简单的函数或工具调用。本质功能模块或专门化的 Agent。类比一个大公司里的不同部门技术部、市场部、财务部每个部门负责一块专业事务协同完成公司目标。与 Workflow/Router 的关系Sub-agent 是 Workflow 中一个“节点”Node的具体实现。Router 则负责在不同的 Sub-agent 之间进行调度。例如一个“客服总机”Router根据问题类型将电话转接给“技术客服”Sub-agent A或“账单客服”Sub-agent B。1.5 Skill技能可复用的能力包Skill是一个比 Sub-agent 更细粒度、更强调可复用性和即插即用的概念。它通常指一个封装好的、解决特定问题的能力可能是一个复杂的工具调用链、一个精心设计的提示词模板、或一个微调的模型。本质预制的能力组件。类比乐高积木块。每个 Skill 是一个具有特定形状接口和功能实现的积木你可以用它快速搭建组装出复杂的结构Agent。与 Sub-agent 的关系界限有时模糊。一个 Skill 可以看作一个轻量级的 Sub-agent。通常Sub-agent 更强调其自主性和状态性而 Skill 更强调其工具性和无状态性。多个 Skill 可以组合成一个 Sub-agent。总结关系Skill是基础能力块Sub-agent是由一个或多个 Skill 组成的专职模块Workflow是编排多个 Sub-agent或节点完成任务的流程图而Router是 Workflow 中负责动态选择下一个 Sub-agent 的决策器。2. 选型决策矩阵何时用哪个理解了概念我们来看实战选型。选择的核心依据是任务的确定性程度和系统的复杂度。组件核心特点适用场景不适用场景Workflow流程固定顺序明确。像编写好的剧本。1. 客服工单处理创建-分配-处理-关闭2. 数据ETL管道抽取-清洗-转换-加载3. 内容审核流水线初筛-敏感词检测-人工复核1. 开放域对话用户意图千变万化。2. 探索性任务路径无法预先确定。Router动态决策条件分支。像十字路口的交通信号灯。1. 意图识别后的任务分发用户问天气-路由到天气Agent2. 多专家系统根据问题领域选择专家3. 故障处理流程根据错误码选择修复方案1. 线性、无分支的简单任务。2. 所有情况都走同一流程。Sub-agent模块化关注点分离。像公司的专业部门。1. 系统需要不同领域的专业知识代码、写作、分析。2. 单个Agent过于庞大需要拆解以降低复杂度。3. 需要独立管理不同模块的权限、资源或版本。1. 任务极其简单一个函数就能搞定。2. 模块间需要高度共享状态和记忆拆分反而增加通信成本。Skill即插即用高复用性。像瑞士军刀上的工具。1. 需要在多个不同Agent或Workflow中复用同一功能如“计算器”、“翻译器”。2. 快速原型验证通过组合现有Skill构建新Agent。3. 社区共享利用他人开发好的成熟能力。1. 功能与业务上下文深度耦合难以抽象。2. 对性能有极致要求封装带来不可接受的损耗。决策流程建议分析任务你的任务是步骤固定的用Workflow还是需要动态判断的引入Router分解功能这个任务可以拆分成几个相对独立的功能模块吗是-考虑Sub-agent或Skill评估复用这些功能模块会在别处用到吗是-优先设计为Skill权衡复杂度引入组件带来的管理收益是否大于其增加的架构复杂度3. LangGraph 全景解析如何落地实现LangGraph 是 LangChain 框架中专门用于构建复杂、有状态多智能体应用的库。它完美地支持了上述所有模式。3.1 LangGraph 的核心抽象State 与 GraphState一个贯穿整个工作流的、可变的共享数据容器。通常用TypedDict定义包含了所有节点需要读写的数据。Graph由Node节点和Edge边组成。Node是执行单元函数Edge定义了节点间的流转。3.2 在 LangGraph 中实现四种模式1. 实现一个线性 Workflow上文已给出示例。关键在于使用add_edge建立固定的前后继关系。2. 实现一个带 Router 的动态 Workflow上文也已给出示例。核心是add_conditional_edges它连接一个源节点到一个路由函数该函数返回下一个目标节点的名称。3. 实现 Sub-agent在 LangGraph 中每个Node都可以视作一个 Sub-agent 的入口。你可以将复杂的逻辑封装在一个节点函数内或者让该节点函数去调用另一个独立的 Agent 实例可能是另一个 LangChain Agent 或 LangGraph 子图。# 假设我们有一个专门处理数学的 Sub-agent class MathAgent: def run(self, problem: str) - str: # 这里可能包含复杂的逻辑比如调用计算工具、验证格式等 return fSolution to {problem} is 42. def math_agent_node(state: State): agent MathAgent() problem state[query] solution agent.run(problem) return {response: solution} # 将这个 Sub-agent 添加到图中作为一个节点 graph.add_node(math_sub_agent, math_agent_node)4. 集成 SkillSkill 通常被实现为Tool或一组Tool。在 LangGraph 的节点中你可以直接调用这些封装好的工具。from langchain.tools import tool # 定义一个 Skill单位换算 tool def unit_converter(amount: float, from_unit: str, to_unit: str) - str: Convert units, e.g., miles to kilometers. # ... 实现换算逻辑 ... return f{amount} {from_unit} {result} {to_unit} # 在某个 Agent 节点中使用这个 Skill def general_assistant_node(state: State): # 解析用户请求判断是否需要单位换算 if convert in state[query]: # 调用 Skill result unit_converter.invoke({amount: 5, from_unit: miles, to_unit: km}) return {response: result} # ... 其他逻辑 ...3.3 高级模式组合使用强大的系统往往是这些模式的组合。一个典型的架构可能是顶层一个主 Router根据用户意图选择不同的Workflow。中层每个 Workflow 由多个Sub-agent节点按顺序或条件边组成。底层每个 Sub-agent 在执行时调用一个或多个Skill工具来完成具体操作。# 概念性代码展示组合架构 def master_router(state): intent classify_intent(state[query]) if intent customer_service: return cs_workflow elif intent data_analysis: return da_workflow def cs_workflow(state): # 可能包含信息收集 - 问题分类 - 解决方案推荐 - 满意度调查 等多个 Sub-agent 节点 pass def da_workflow(state): # 可能包含数据查询 - 清洗 - 分析 - 可视化 等多个 Sub-agent 节点 pass # 构建主图 master_graph StateGraph(State) master_graph.add_node(router, master_router) master_graph.add_node(cs_workflow, cs_workflow) master_graph.add_node(da_workflow, da_workflow) master_graph.add_conditional_edges(router, ...)4. 实战案例构建一个智能任务处理中心让我们设计一个综合性的例子一个能处理“天气查询”、“简单计算”和“通用问答”的智能助手并记录交互历史。需求分析需要路由功能根据用户输入选择处理分支。每个分支天气、计算是一个独立的 Sub-agent。通用问答分支可以直接调用大模型。所有交互需要记录到记忆State中。实现步骤步骤1定义 State 和工具Skillfrom typing import TypedDict, List, Annotated import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage from langchain_community.tools import DuckDuckGoSearchRun # 定义共享状态 class AgentState(TypedDict): messages: Annotated[List, operator.add] # 对话历史 current_query: str response: str # 定义一些 Skill (Tools) search_tool DuckDuckGoSearchRun() llm ChatOpenAI(modelgpt-4o-mini) def calculate(expression: str) - str: 一个简单的计算 Skill注意生产环境需使用更安全的评估方式。 try: # 警告使用 eval 有安全风险此处仅用于演示 result eval(expression, {__builtins__: {}}) return str(result) except Exception as e: return f计算错误: {e}步骤2实现 Sub-agent 节点# Sub-agent 1: 天气查询专家 (模拟) def weather_agent(state: AgentState): query state[current_query] # 在实际项目中这里会调用天气API # 此处模拟并假设使用搜索工具获取信息 search_result search_tool.run(fweather {query}) response f关于【{query}】的天气信息{search_result[:200]}... return {response: response, messages: [AIMessage(contentresponse)]} # Sub-agent 2: 计算专家 def math_agent(state: AgentState): query state[current_query] # 简单提取计算表达式例如“计算 35*2” # 这里做简单演示直接尝试计算整个查询不安全仅演示 result calculate(query) response f计算结果{result} return {response: response, messages: [AIMessage(contentresponse)]} # Sub-agent 3: 通用助手 def general_agent(state: AgentState): query state[current_query] # 直接调用 LLM 进行回复并传入历史消息作为上下文 chat_history state[messages][-5:] # 取最近5条历史 messages [*chat_history, HumanMessage(contentquery)] ai_msg llm.invoke(messages) response ai_msg.content return {response: response, messages: [ai_msg]}步骤3实现路由决策逻辑def classify_and_route(state: AgentState) - str: 路由决策函数根据查询内容决定下一个节点 query state[current_query].lower() if any(word in query for word in [天气, weather, 气温, 预报]): return weather_node elif any(word in query for word in [计算, calculate, , -, *, /, 等于]): return math_node else: return general_node步骤4构建并编译 LangGraph# 初始化图 workflow StateGraph(AgentState) # 添加节点即我们的 Sub-agent workflow.add_node(weather_node, weather_agent) workflow.add_node(math_node, math_agent) workflow.add_node(general_node, general_agent) # 设置入口点一个分类器节点它只负责更新状态然后路由 def classifier_node(state: AgentState): # 这个节点可以做一些预处理这里直接传递 return state workflow.add_node(classifier, classifier_node) workflow.set_entry_point(classifier) # 关键从分类器节点添加条件边实现路由 workflow.add_conditional_edges( classifier, classify_and_route, # 路由决策函数 { weather_node: weather_node, math_node: math_node, general_node: general_node, } ) # 所有业务节点执行完后都流向 END workflow.add_edge(weather_node, END) workflow.add_edge(math_node, END) workflow.add_edge(general_node, END) # 编译成可执行应用 app workflow.compile()步骤5运行与测试# 初始化状态 initial_state {messages: [], current_query: 北京今天天气怎么样, response: } # 执行图 final_state app.invoke(initial_state) print(final_state[response]) # 输出示例关于【北京今天天气怎么样】的天气信息北京今日晴转多云最高气温25℃... # 第二次交互状态中包含了历史消息 next_state {messages: final_state[messages], current_query: 那计算一下 15*32 等于多少, response: } final_state2 app.invoke(next_state) print(final_state2[response]) # 输出示例计算结果47这个案例展示了一个结合了Routerclassify_and_route、Sub-agent三个*_agent函数和Skillcalculate函数、search_tool、llm的完整Workflow。State中的messages字段实现了简单的记忆功能。5. 常见问题与排查思路在构建和运行此类系统时你可能会遇到以下典型问题问题现象可能原因排查与解决思路图编译失败1. State 类型定义错误。2. 节点函数返回值格式与 State 不匹配。3. 边引用了不存在的节点。1. 检查TypedDict的字段和类型注解。2. 确保节点函数返回字典且键是 State 的字段名。3. 使用graph.get_graph().draw_mermaid()可视化图结构检查节点和边。路由死循环或无法结束1. 条件边函数逻辑错误导致在几个节点间循环。2. 忘记将某个分支连接到END。1. 在路由函数中添加详细日志打印其输入和输出。2. 确保每个业务节点后都有边指向END或其他收敛节点。3. 设置interrupt_before或interrupt_after进行调试。状态更新不符合预期1. 对Annotated归约操作符理解有误。2. 多个节点并发修改同一字段需用add_messages等线程安全方式。1. 复习operator.add用于列表拼接的逻辑。2. 对于非列表的复杂状态更新考虑在节点内手动合并字典。3. 使用State的checkpointer功能进行快照调试。Sub-agent 执行效率低1. 每个 Sub-agent 都初始化重量级资源如 LLM 客户端。2. 网络调用过多未做批处理或缓存。1. 将共享资源如 LLM 实例、数据库连接在节点函数外部初始化通过闭包或类属性传入。2. 对可缓存的请求如天气查询添加缓存层。3. 考虑将非严格顺序的节点并行化StateGraph支持并发。Skill/Tool 调用失败1. Tool 的输入参数格式不对。2. API 密钥未配置或网络错误。3. Tool 的输出解析失败。1. 使用tool.args_schema严格定义输入格式。2. 在调用 Tool 的地方添加try-except并返回友好的错误信息到 State。3. 使用 LangChain 的Tool封装它提供了更好的错误处理和解析。6. 最佳实践与工程建议始于简单渐进复杂不要一开始就设计庞大的图。从一个线性 Workflow 开始验证核心链路再逐步引入 Router 和 Sub-agent。State 设计要精简State 是所有节点共享的全局变量。只存放必要的数据避免过度耦合。考虑将大型数据如原始文档通过引用如 ID 或路径存储在 State 中而非数据本身。节点职责单一每个节点Sub-agent应只做一件事并做好。这有助于测试、调试和复用。如果一个节点过于复杂就拆分成多个节点。Skill 化与工具化将通用的功能如数据查询、格式转换、API调用封装成 ToolSkill。这不仅能提高复用性还能利用 LangChain 生态丰富的现有工具库。重视错误处理与回退在路由决策和节点执行中都要考虑失败情况。可以设计一个fallback_node或error_handler节点用于处理异常和提供降级响应。可视化与调试充分利用graph.get_graph().draw_mermaid()将你的工作流可视化。在开发阶段可以使用stream模式运行图观察每一步的状态变化。测试策略单元测试单独测试每个节点函数、路由函数和 Skill。集成测试测试从一个入口到结束的完整路径。压力测试模拟复杂、绕路的对话流检查状态管理是否正确。版本控制与部署将你的图定义Python代码和重要的提示词模板进行版本控制。考虑将编译好的app对象序列化或通过 API 提供服务便于部署和更新。7. 总结与学习路线通过本文的梳理希望你已经对 AI Agent 系统中的 Workflow、Router、Sub-agent 和 Skill 有了清晰的认识。它们不是四选一的关系而是构建智能、健壮 Agent 系统的不同维度的组件。Workflow是你的战略蓝图定义了任务的骨架。Router是你的战术决策者让系统具备动态响应能力。Sub-agent是你的专业团队负责具体领域的执行。Skill是你团队的工具箱是高效执行的基础。学习路线建议基础掌握从 LangGraph 官方文档的StateGraph基础教程开始亲手实现一个线性工作流。进阶理解尝试实现一个带条件分支Router的图例如一个简单的分类问答系统。实战深化设计一个包含 2-3 个不同 Sub-agent如检索专家、总结专家、润色专家的复杂工作流并尝试将一些功能封装成可复用的 ToolSkill。生态探索了解 LangChain 丰富的 Tool 生态如搜索引擎、代码执行器、各类 API 连接器思考如何将它们作为 Skill 集成到你的 Agent 中。工程化思考研究如何测试、监控、部署和维护一个基于图的 Agent 系统。记住没有“最好”的架构只有“最适合”当前需求的架构。从简单需求出发逐步迭代让系统的复杂度随着业务需求自然生长才是可持续的工程之道。