AI Agent架构模式解析:Workflow、Router、Sub-agent与Skill在LangGraph中的实践
2026/8/20 8:10:57 网站建设 项目流程

如果你正在构建一个 AI Agent 系统,面对 Workflow、Router、Sub-agent、Skill 这些眼花缭乱的概念,是不是感觉无从下手?你可能会想:它们到底有什么区别?我的项目应该用哪个?为什么别人的 Agent 能处理复杂任务,而我的却像个“人工智障”?

问题的核心往往不在于模型本身,而在于架构模式的选择。选错了模式,你的 Agent 要么能力单一,要么逻辑混乱,维护成本直线上升。最近,LangGraph 作为 LangChain 生态中专门用于构建有状态、多步骤 Agent 的框架,成为了解决这些问题的热门选择。但 LangGraph 本身也引入了 State、Node、Edge 等新概念,让选型变得更加复杂。

这篇文章不会给你一个“万能公式”,而是帮你建立一个清晰的决策框架。我们将深入解析 Workflow、Router、Sub-agent、Skill 这四种核心模式,并结合 LangGraph 的实践,告诉你:

  1. 每种模式解决什么本质问题?(是任务编排、动态路由、能力解耦还是功能复用?)
  2. 它们之间的边界在哪里?(为什么有时 Router 和 Sub-agent 看起来很像?)
  3. 在 LangGraph 中如何具体实现?(用代码和 State 设计说话)
  4. 你的项目到底该选哪个?(根据任务复杂度、团队规模、维护性需求做判断)

读完本文,你将能像搭积木一样,为你的 AI Agent 选择合适的“心智模式”,并利用 LangGraph 将其高效、清晰地构建出来。

1. 核心问题:为什么你的 AI Agent 需要“架构”?

在深入具体模式之前,我们必须先达成一个共识:AI Agent 的核心价值是“自主决策与执行”,而不仅仅是“调用一次大模型”

一个简单的 Q&A 机器人,用 Prompt + RAG 或许就够了。但当你需要 Agent 完成“分析市场报告、总结要点、生成邮件草稿并预约会议”这一系列任务时,问题就来了:

  • 状态如何管理?上一步的分析结果如何传递给下一步的邮件生成?
  • 流程如何控制?如果总结要点失败了,是重试还是跳过?是否需要人工审核?
  • 能力如何组织?是写一个超级庞大的 Prompt,还是把分析、写作、日历操作拆分成独立模块?

这就是 Workflow、Router、Sub-agent、Skill 这些模式要解决的问题。它们不是互斥的,而是构建复杂 Agent 系统的不同抽象层次和设计模式

我们可以用一个简单的类比来理解:

  • Skill(技能):就像乐高积木的最小单元块(一块2x4的砖)。它是一个具体的、可复用的能力,比如“调用搜索API”、“执行一段Python代码”、“查询数据库”。
  • Sub-agent(子智能体):像是一个预组装好的乐高功能模块(比如一辆小车、一座小房子)。它内部可能用了多个Skill,有自己相对独立和复杂的逻辑,负责一个特定的子目标,比如“数据分析专家”或“文案写手”。
  • Router(路由):像是乐高说明书里的决策分支。“如果拿到的是轮子零件,就交给小车模块;如果拿到的是窗户零件,就交给房子模块”。它根据当前状态或输入,决定下一步由哪个Skill或Sub-agent来执行。
  • Workflow(工作流):则是完整的乐高搭建说明书。它定义了从拆箱到成品,所有步骤(包括Router的决策点)的执行顺序和条件。它管理着整个构建过程的状态(哪些步骤已完成,当前在拼什么)。

LangGraph 正是描述和运行这份“说明书”(Workflow)的绝佳工具。它通过State对象来记录我们的搭建进度,通过Node来定义每个步骤(可以是Skill、Sub-agent或Router),通过Edge来定义步骤之间的流转逻辑。

接下来,我们逐一拆解这四种模式,并用 LangGraph 的视角重新审视它们。

2. 基础概念深度解析:四种核心模式

2.1 Skill(技能):原子化可复用单元

是什么?Skill 是 Agent 能够执行的最小、最具体的操作单元。它通常对应一次工具调用(Tool Call)或一个精心设计的 Prompt 模板。其核心目标是高内聚、低耦合和可复用

解决了什么问题?

  • 代码复用:避免在多个Agent中重复编写相同的工具调用逻辑。
  • 维护性:当某个API接口变更时,只需修改对应的Skill,所有使用它的Agent都会受益。
  • 能力标准化:团队可以沉淀一个统一的“技能库”,新成员可以快速组合出新的Agent。

LangGraph 中的体现:在 LangGraph 中,一个 Skill 通常被实现为一个Node(节点)。这个 Node 是一个函数,它读取全局的State,执行特定操作(如调用工具、LLM),并更新State

示例:一个搜索 Skill

# 这是一个 LangGraph Node,也是一个 Skill from langchain_community.tools import TavilySearchResults from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated import operator # 1. 定义状态 State class AgentState(TypedDict): question: str search_results: Annotated[list, operator.add] # 这是一个追加式状态 answer: str # 2. 实现 Skill Node def search_node(state: AgentState): """Skill: 执行网络搜索""" question = state[“question”] search_tool = TavilySearchResults(max_results=3) results = search_tool.invoke(question) # 更新状态 return {“search_results”: results} # 3. 在图中使用这个 Skill Node graph_builder = StateGraph(AgentState) graph_builder.add_node(“search_web”, search_node) # “search_web” 就是一个 Skill Node # ... 后续添加边 (Edges)

如何判断是否需要 Skill?当你发现一段操作(工具调用、数据转换、特定Prompt)会在不同任务或Agent中被重复使用时,就应该将其抽象为 Skill。

2.2 Sub-agent(子智能体):专精特定领域的“专家”

是什么?Sub-agent 是一个具备一定自主性的、负责完成特定子目标的 Agent。它内部可能封装了多个 Skill 和复杂的决策逻辑。你可以把它理解为一个功能更完备的“小 Agent”

解决了什么问题?

  • 复杂度管理:将庞大复杂的任务分解,由不同的“专家”处理。例如,一个主Agent协调一个“研究Sub-agent”和一个“写作Sub-agent”。
  • 职责分离:每个Sub-agent可以针对其领域进行深度优化(使用不同的模型、不同的Prompt)。
  • 并行与协作:在某些架构下,Sub-agent可以并行执行任务。

与 Skill 的关键区别:Skill 是“做什么”(动作),Sub-agent 是“谁来做”(责任主体)。Sub-agent 内部会决定“如何做”,可能会调用多个 Skill。

LangGraph 中的体现:在 LangGraph 中,一个 Sub-agent 通常也表现为一个Node,但这个 Node 的内部逻辑可能非常复杂,它本身可能又是一个嵌套的、小型的 LangGraph。

示例:一个数据分析 Sub-agent

def data_analysis_agent_node(state: AgentState): """Sub-agent: 专精于数据分析的智能体""" # 这个 Sub-agent 内部可能有自己的逻辑链 from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI data = state[“raw_data”] llm = ChatOpenAI(model=“gpt-4o”) # 第一步:清理数据 (可能调用一个数据清洗 Skill) cleaned_data = clean_data_function(data) # 第二步:分析趋势 (调用 LLM) analysis_prompt = ChatPromptTemplate.from_template(“”” 请分析以下数据的主要趋势和关键洞察: {data} ”””) analysis_chain = analysis_prompt | llm analysis_result = analysis_chain.invoke({“data”: cleaned_data}) # 第三步:生成报告摘要 summary_prompt = ChatPromptTemplate.from_template(“”” 根据分析,生成一段摘要: {analysis} ”””) summary_chain = summary_prompt | llm summary = summary_chain.invoke({“analysis”: analysis_result.content}) # 更新主状态 return {“data_analysis”: analysis_result.content, “summary”: summary} # 将这个 Sub-agent 添加到主图中 graph_builder.add_node(“data_analyst”, data_analysis_agent_node)

如何判断是否需要 Sub-agent?当任务的一个环节足够复杂,需要独立的规划、执行和错误处理逻辑,并且这个“专家模块”可能在多个工作流中被重用时,就适合设计为 Sub-agent。

2.3 Router(路由):工作流中的决策者

是什么?Router 是一个决策节点,它根据当前工作流的状态(State)或输入,动态地决定下一步应该执行哪个 Node(Skill 或 Sub-agent)。它是实现条件逻辑和动态工作流的关键。

解决了什么问题?

  • 非线性的任务流:任务不是固定顺序,需要根据中间结果做判断。例如,“如果用户查询天气,就调用天气Skill;如果查询新闻,就调用搜索Skill”。
  • 错误处理与重试:“如果API调用失败,是重试3次,还是转人工处理?”
  • 多路径选择:根据内容类型(文本、图片、代码)路由到不同的处理管道。

LangGraph 中的体现:在 LangGraph 中,Router 通过Conditional Edges(条件边)来实现。你定义一个路由函数,它根据State返回下一个要执行的Node的名称。

示例:一个简单的分类 Router

from langgraph.graph import StateGraph, START def route_question(state: AgentState): """Router: 根据问题类型决定下一步""" question = state[“question”].lower() if “weather” in question: return “weather_tool” # 跳转到天气 Skill Node elif “calculate” in question or “what is” in question: return “calculator_tool” # 跳转到计算器 Skill Node else: return “general_qa” # 跳转到通用问答 Sub-agent Node # 在图中添加条件边 graph_builder.add_conditional_edges( “classify”, # 起始 Node(可以是一个专门做分类的Node) route_question, # 路由函数 {“weather_tool”: “weather_tool”, “calculator_tool”: “calculator_tool”, “general_qa”: “general_qa”} # 可能的目的地 ) # 另一种常见模式:从 START 就开始路由 graph_builder.add_edge(START, “classify”)

如何判断是否需要 Router?当你的工作流存在“如果...那么...”这类分支判断时,就必须引入 Router。它是让 Agent 从“固定脚本”走向“智能决策”的标志。

2.4 Workflow(工作流):整体的编排与状态管理

是什么?Workflow 是最高层次的抽象,它定义了完成一个宏观目标所需要的完整过程,包括所有步骤(Nodes)的执行顺序、分支逻辑(Conditional Edges)以及全局状态(State)的流转。LangGraph 的核心就是帮助你定义和运行 Workflow。

解决了什么问题?

  • 端到端任务自动化:将复杂的多步骤任务串联起来。
  • 状态持久化与共享:确保每个步骤都能获取到之前步骤的结果,并将自己的产出传递给后续步骤。
  • 流程可视化与调试:LangGraph 的图结构使得整个工作流清晰可见,便于理解和调试。

LangGraph 中的体现:整个StateGraph对象及其配置,就是 Workflow 的定义。compile()方法将其编译为可执行的图。

示例:一个简单的客服工单处理 Workflow

# 定义状态 class SupportState(TypedDict): user_input: str intent: str retrieved_kb: str response: str needs_human: bool # 构建 Workflow graph_builder = StateGraph(SupportState) # 1. 添加 Nodes (Skills & Sub-agents) graph_builder.add_node(“classify_intent”, classify_intent_node) # Skill: 意图分类 graph_builder.add_node(“retrieve_kb”, retrieve_kb_node) # Skill: 知识库检索 graph_builder.add_node(“generate_response”, generate_response_node) # Sub-agent: 生成回复 graph_builder.add_node(“human_escalation”, human_escalation_node) # Skill: 转人工 # 2. 设置流程和路由 graph_builder.add_edge(START, “classify_intent”) graph_builder.add_edge(“classify_intent”, “retrieve_kb”) graph_builder.add_edge(“retrieve_kb”, “generate_response”) # 3. 条件边:判断是否需要人工介入 def need_human_router(state: SupportState): if state[“needs_human”]: return “human_escalation” else: return END graph_builder.add_conditional_edges( “generate_response”, need_human_router, {“human_escalation”: “human_escalation_node”, END: END} ) graph_builder.add_edge(“human_escalation”, END) # 4. 编译 Workflow workflow = graph_builder.compile()

Workflow 是必然存在的吗?是的,只要你构建的 Agent 涉及多个步骤或决策,你就在定义一个 Workflow。使用 LangGraph 是将其显式化、结构化和管理起来的最佳实践。

3. 选型决策指南:为你的项目选择正确模式

理解了概念,我们来看实战。如何为你的项目选择组合这些模式?下面是一个决策流程图和场景分析。

3.1 决策流程图

开始 │ ├── 你的任务是否只有一个步骤?(例如:调用一次搜索) │ │ │ ├── 是 → 只需要一个 **Skill**。 │ │ │ └── 否 → 进入多步骤任务分析。 │ ├── 多步骤任务中,步骤顺序是否固定且无分支? │ │ │ ├── 是 → 使用线性 **Workflow**,串联多个 **Skill**。 │ │ (例如:获取数据 → 清洗数据 → 分析数据) │ │ │ └── 否 → 任务中存在条件判断。 │ ├── 条件判断是基于输入或中间结果,选择不同的“动作”吗? │ │ │ ├── 是 → 引入 **Router**,连接不同的 **Skill**。 │ │ (例如:用户问天气 → 天气Skill;用户问计算 → 计算器Skill) │ │ │ └── 否 → 条件判断后,需要执行的是一个复杂的“子任务”。 │ ├── 这个“子任务”内部是否包含多步骤和私有状态,且可被抽象为独立模块? │ │ │ ├── 是 → 将子任务封装为 **Sub-agent**,在主 **Workflow** 中用 **Router** 调用它。 │ │ (例如:主Agent接到“写报告”任务,路由到“写作专家Sub-agent”,该Sub-agent内部有搜集资料、拟定大纲、撰写、润色等步骤) │ │ │ └── 否 → 重新审视任务分解,可能仍需用 **Router** + 复杂 **Skill** 解决。 │ └── 最终,用 **LangGraph** 将所有的 **Skill**、**Sub-agent**、**Router** 编排成一个完整的 **Workflow**。

3.2 典型场景与模式组合

场景一:智能客服助手

  • 需求:识别用户意图 → 查询知识库 → 生成回答。如果问题复杂,转人工。
  • 模式组合
    • classify_intent(Skill): 意图分类。
    • retrieve_kb(Skill): 检索知识库。
    • generate_answer(Skill/Sub-agent): 生成回答。如果逻辑简单就是Skill,如果需要调用LLM并考虑上下文就是轻量级Sub-agent。
    • human_escalation(Skill): 转人工工单。
    • Router: 在generate_answer后,根据置信度或用户反馈,决定流向END还是human_escalation
    • Workflow:START → classify_intent → retrieve_kb → generate_answer → Router → (END / human_escalation → END)

场景二:自动化数据分析报告生成

  • 需求:用户上传数据 → 自动选择分析模型(统计/机器学习)→ 执行分析 → 生成可视化图表 → 编写报告摘要。
  • 模式组合
    • data_loader(Skill): 加载和初步清洗数据。
    • model_router(Router): 根据数据特征,决定使用statistical_analyst还是ml_analyst
    • statistical_analyst(Sub-agent): 内部包含描述性统计、相关性分析等Skill。
    • ml_analyst(Sub-agent): 内部包含特征工程、模型训练、评估等Skill。
    • visualization(Skill): 生成图表。
    • report_writer(Sub-agent): 汇总分析和图表,撰写摘要。
    • Workflow:START → data_loader → model_router → {statistical_analyst | ml_analyst} → visualization → report_writer → END

场景三:多工具协作的“超级助理”

  • 需求:理解用户自然语言指令(如“帮我查下北京天气,然后算算出差补贴”),拆解任务,按顺序调用不同工具。
  • 模式组合
    • task_planner(Sub-agent): 理解指令,拆解为工具执行序列([“search_weather”, “calculate_allowance”])。这是一个典型的规划型Sub-agent。
    • search_weather(Skill): 调用天气API。
    • calculate_allowance(Skill): 调用计算器工具。
    • response_synthesizer(Skill): 将多个工具结果整合成自然语言回复。
    • Workflow:START → task_planner → [按序列执行各个Skill] → response_synthesizer → END。这里的序列执行可以通过一个循环节点或动态边来实现。

4. LangGraph 实战:构建一个混合模式 Agent

让我们用一个综合案例,将上述模式串联起来。我们构建一个“技术调研助手”,它能根据用户提出的技术概念,决定是直接回答、进行网络搜索,还是启动一个深度的“对比分析”子任务。

4.1 环境准备与依赖安装

# 创建虚拟环境(推荐) python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装核心依赖 pip install langgraph langchain-openai langchain-community # 可选:安装搜索工具依赖(如Tavily,需API Key) # pip install tavily-python

4.2 定义全局状态(State)

State 是 LangGraph 的“记忆中枢”,所有节点都读写它。

# state.py from typing import TypedDict, List, Optional, Annotated import operator class ResearchState(TypedDict): """技术调研助手的状态定义""" # 用户输入 user_query: str # 由分类节点产生 query_type: Optional[str] # ‘simple‘, ‘need_search‘, ‘need_comparison‘ # 由搜索节点追加 search_context: Annotated[List[str], operator.add] # 由对比分析子任务产生 comparison_result: Optional[str] # 最终答案 final_answer: str
  • Annotated[List[str], operator.add]是 LangGraph 的一个强大特性,表示这个字段是一个列表,后续节点可以向其中“追加”内容,而不是覆盖。

4.3 实现 Skill 节点

# nodes.py from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate # 假设我们有一个搜索工具 from langchain_community.tools import TavilySearchResults llm = ChatOpenAI(model=“gpt-3.5-turbo”) def classify_query_node(state: ResearchState) -> dict: """Skill: 查询分类""" query = state[“user_query”] prompt = ChatPromptTemplate.from_messages([ (“system”, “你是一个技术查询分类器。请将用户查询分类为:’simple‘ (可直接回答)、’need_search‘ (需要搜索)、’need_comparison‘ (需要对比分析)。只返回分类标签。”), (“human”, “{query}”) ]) chain = prompt | llm response = chain.invoke({“query”: query}) query_type = response.content.strip().lower() return {“query_type”: query_type} def simple_qa_node(state: ResearchState) -> dict: """Skill: 简单问答""" query = state[“user_query”] prompt = ChatPromptTemplate.from_messages([ (“system”, “你是一个技术专家,请用简洁的语言回答以下问题。”), (“human”, “{query}”) ]) chain = prompt | llm response = chain.invoke({“query”: query}) return {“final_answer”: response.content} def search_node(state: ResearchState) -> dict: """Skill: 执行网络搜索(需要TAVILY_API_KEY环境变量)""" query = state[“user_query”] search = TavilySearchResults(max_results=2) try: results = search.invoke(query) # 提取摘要信息 contexts = [f“来源 {i+1}: {r[‘content’]}” for i, r in enumerate(results)] return {“search_context”: contexts} except Exception as e: return {“search_context”: [f“搜索失败: {str(e)}”]} def synthesize_from_search_node(state: ResearchState) -> dict: """Skill: 基于搜索内容合成答案""" query = state[“user_query”] context = “\n”.join(state.get(“search_context”, [])) prompt = ChatPromptTemplate.from_messages([ (“system”, “你是一个技术研究员。请基于以下搜索内容,回答用户的问题。如果内容不相关或不足,请说明。\n搜索内容:{context}”), (“human”, “问题:{query}”) ]) chain = prompt | llm response = chain.invoke({“query”: query, “context”: context}) return {“final_answer”: response.content}

4.4 实现 Sub-agent 节点

# nodes.py (续) def comparison_agent_node(state: ResearchState) -> dict: """ Sub-agent: 深度对比分析专家。 这是一个复杂的子智能体,内部包含多步逻辑。 """ query = state[“user_query”] # 步骤1: 拆解对比项 decompose_prompt = ChatPromptTemplate.from_template(“”” 请从以下技术对比问题中,提取出需要对比的A和B两个实体。 问题:{query} 输出格式:A: [实体A], B: [实体B] “””) decompose_chain = decompose_prompt | llm entities = decompose_chain.invoke({“query”: query}).content # 简单解析,实际应用应更健壮 a, b = “EntityA”, “EntityB” if “A:“ in entities and ”B:“ in entities: parts = entities.split(”B:“) a = parts[0].replace(”A:“, ”“).strip() b = parts[1].strip() # 步骤2: 为每个实体搜索信息 search_tool = TavilySearchResults(max_results=1) info_a = search_tool.invoke(f“{a} technology overview”) info_b = search_tool.invoke(f“{b} technology overview”) # 步骤3: 执行对比分析 compare_prompt = ChatPromptTemplate.from_template(“”” 请从技术特点、适用场景、优缺点等方面对比以下两项技术: 技术A: {a} 信息A: {info_a} 技术B: {b} 信息B: {info_b} 请生成一份结构化的对比报告。 “””) compare_chain = compare_prompt | llm comparison_report = compare_chain.invoke({ “a”: a, “info_a”: info_a, “b”: b, “info_b”: info_b }).content # 更新状态,这个结果可以被后续节点使用 return { “comparison_result”: comparison_report, “final_answer”: f“# {a} vs {b} 对比分析\n\n{comparison_report}” # 这里也直接更新最终答案 }

4.5 实现 Router 逻辑

Router 通常体现为条件边(Conditional Edge)的判断函数。

# edges.py def route_after_classify(state: ResearchState) -> str: """Router: 根据分类结果,决定下一步走向哪个节点""" query_type = state.get(“query_type”) if query_type == “simple”: return “simple_qa” elif query_type == “need_search”: return “search” elif query_type == “need_comparison”: return “comparison_agent” else: # 默认情况,进入搜索流程 return “search” def route_after_search(state: ResearchState) -> str: """Router: 搜索完成后,决定是合成答案还是结束""" if state.get(“search_context”): return “synthesize” else: # 如果搜索没拿到内容,直接结束 return END

4.6 组装完整 Workflow

# graph.py from langgraph.graph import StateGraph, START, END from state import ResearchState from nodes import classify_query_node, simple_qa_node, search_node, synthesize_from_search_node, comparison_agent_node from edges import route_after_classify, route_after_search # 1. 创建图构建器 graph_builder = StateGraph(ResearchState) # 2. 添加所有节点 graph_builder.add_node(“classify”, classify_query_node) graph_builder.add_node(“simple_qa”, simple_qa_node) graph_builder.add_node(“search”, search_node) graph_builder.add_node(“synthesize”, synthesize_from_search_node) graph_builder.add_node(“comparison_agent”, comparison_agent_node) # 3. 设置入口 graph_builder.add_edge(START, “classify”) # 4. 添加条件边:分类后的路由 graph_builder.add_conditional_edges( “classify”, route_after_classify, { “simple_qa”: “simple_qa”, “search”: “search”, “comparison_agent”: “comparison_agent” } ) # 5. 添加普通边和条件边 # 简单问答直接结束 graph_builder.add_edge(“simple_qa”, END) # 对比分析直接结束(因为它已经生成了final_answer) graph_builder.add_edge(“comparison_agent”, END) # 搜索后需要合成 graph_builder.add_conditional_edges( “search”, route_after_search, { “synthesize”: “synthesize”, END: END } ) graph_builder.add_edge(“synthesize”, END) # 6. 编译图,得到可执行的Workflow research_agent_workflow = graph_builder.compile() # 可选:可视化图(需要安装 pygraphviz) # try: # from IPython.display import Image, display # display(Image(research_agent_workflow.get_graph().draw_mermaid_png())) # except: # print(“可视化需要安装 pygraphviz 和 IPython”)

4.7 运行与测试

# run.py from graph import research_agent_workflow # 测试用例1:简单问题 print(“=== 测试1: 简单问题 ===") initial_state = {“user_query”: “什么是Python的装饰器?”, “final_answer”: “”} result = research_agent_workflow.invoke(initial_state) print(f“最终答案:{result[‘final_answer’][:200]}...”) print(f“经过的节点:{list(result[‘__pregel_messages’].keys()) if ‘__pregel_messages’ in result else ‘N/A’}”) # 测试用例2:需要搜索的问题 print(“\n=== 测试2: 需要搜索的问题 ===") initial_state = {“user_query”: “LangGraph 最新版本有什么新特性?”, “final_answer”: “”} result = research_agent_workflow.invoke(initial_state) print(f“最终答案:{result[‘final_answer’][:300]}...”) print(f“搜索内容:{result.get(‘search_context’, [])}”) # 测试用例3:需要对比分析的问题 print(“\n=== 测试3: 对比分析问题 ===") initial_state = {“user_query”: “对比一下 FastAPI 和 Django REST framework 的优缺点”, “final_answer”: “”} result = research_agent_workflow.invoke(initial_state) print(f“最终答案:\n{result[‘final_answer’]}”)

5. 常见问题与排查思路

在构建基于 LangGraph 的 Agent 系统时,你可能会遇到以下典型问题:

问题现象可能原因排查方式解决方案
State字段更新不生效1. 字段未在TypedDict中正确定义。
2. 使用了错误的更新方式(如对Annotated列表用了赋值而非追加)。
1. 检查State类定义。
2. 打印每个节点执行前后的state
1. 确保字段类型正确。对于列表追加,使用operator.add注解并返回{“field”: [new_value]}
图编译失败1. 存在未连接的节点。
2.Conditional Edges返回了图中不存在的节点名。
3. 节点函数签名不符合要求。
查看graph_builder.compile()时的错误信息。1. 确保所有节点都被边连接或条件边覆盖。
2. 检查路由函数返回值是否在add_conditional_edges的映射中。
3. 确保节点函数接收并返回state字典。
工作流陷入无限循环1. 图中存在循环,但没有终止条件。
2.Router逻辑错误,总是在几个节点间来回跳转。
1. 可视化你的图结构。
2. 在State中添加step_count字段并设置最大步数限制。
1. 确保至少有一条路径能到达END
2. 在路由逻辑中加入终止条件(如最大重试次数、超时)。
3. 使用add_edge(node, END)明确结束点。
Sub-agent 内部错误导致整个流程失败Sub-agent 节点内部代码有 bug 或外部服务(如 API)不可用。1. 对 Sub-agent 节点函数进行单独测试和异常捕获。
2. 查看 LangGraph 执行时的详细日志。
1. 在 Sub-agent 内部实现健壮的错误处理和回退机制。
2. 考虑使用try...except包裹关键操作,并在State中设置错误标志,由后续节点处理。
性能瓶颈1. 所有节点线性执行,无法并行。
2. LLM 调用耗时过长。
1. 使用性能分析工具。
2. 检查是否有节点可以并行化。
1. 评估是否可将无依赖的节点并行化(LangGraph 支持分支与合并)。
2. 为 LLM 调用设置合理的超时和重试。
3. 考虑缓存频繁使用的 LLM 响应或工具调用结果。

6. 最佳实践与工程建议

  1. State 设计要精简而充分

    • 只存储必要的共享数据:避免将临时变量或节点内部状态放入全局 State。
    • 使用Annotated处理集合操作:对于列表追加、计数器累加等操作,利用operator.add等注解,让 LangGraph 自动处理并发安全。
    • 为复杂状态设计清晰的 Schema:使用TypedDict或 Pydantic 模型,并添加文档字符串,便于团队理解。
  2. 节点(Node)保持单一职责

    • 每个 Node 应只做一件事。如果一个 Node 过于复杂,考虑将其拆分为多个 Node 或升级为一个 Sub-agent。
    • Node 函数应尽量是纯函数,或副作用可预测(如只调用外部工具)。这有利于测试和调试。
  3. 善用 Router 实现灵活逻辑,但避免过度复杂

    • 路由逻辑应清晰、可预测。复杂的路由函数本身可以抽离成一个专门的“决策 Node”。
    • 避免创建“蜘蛛网”状的图。如果条件分支太多,考虑是否应该引入更上层的规划型 Sub-agent 来动态生成执行路径。
  4. Sub-agent 的粒度要适中

    • 太粗:失去模块化优势,变得难以维护和复用。
    • 太细:增加不必要的编排复杂度,通信开销大。
    • 判断标准:如果一个模块有独立的“目标”、内部包含多个步骤、并且可能被多个不同的 Workflow 调用,它就适合成为 Sub-agent。
  5. 测试策略

    • 单元测试:单独测试每个 Skill 和 Router 函数。
    • 集成测试:测试 Sub-agent 内部的逻辑流。
    • 工作流测试:使用不同的初始 State 调用完整的graph.invoke(),验证端到端结果和状态流转是否符合预期。
    • 模拟外部依赖:在测试中,模拟 LLM 调用和工具 API,保证测试的稳定性和速度。
  6. 版本控制与文档

    • 将 Workflow 的定义(图结构)视为重要的应用代码,纳入版本控制。
    • 使用代码注释和图可视化工具(如 Mermaid)来文档化你的 Agent 架构,这对于团队协作和后期维护至关重要。

选择 Workflow、Router、Sub-agent 还是 Skill,从来不是一道单选题,而是一道关于如何组织智能的架构设计题。LangGraph 提供的图计算模型,是表达这种复杂编排关系的优雅工具。

核心要诀是:从任务本身出发,自顶向下分解。先定义你的 Workflow 要解决的宏观目标,然后识别其中的关键决策点(引入 Router),再将复杂的、可复用的子目标模块化(设计 Sub-agent),最后用原子化的 Skill 来实现具体的操作。

不要试图在第一个版本中就设计出完美的架构。一个好的实践是:先用一个简单的线性 Workflow 串联几个 Skill 跑通核心流程,然后逐步识别出需要分支判断的地方引入 Router,再将其中膨胀的节点重构为 Sub-agent。这种迭代演进的方式,能让你在理解问题域的同时,演化出最适合的 Agent 系统结构。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询