使用LangGraph构建AI智能体应用工作流

大型语言模型正推动新一代应用的诞生，不过这些应用早已超越了简单的“提问-回答”模式。随着系统日益复杂，智能体式工作流（agentic workflows）已成为标准配置——它让LLM借助预定义组件和显式状态管理，编排结构化、多步骤的流程。换言之，智能体式工作流遵循一套固定且一致的步骤序列，不依赖执行中的临时调整，而是强调可靠性、透明度和可控性。LLM的决策被严格约束在设定边界内，每个阶段都有章可循、可复现。在本书后续章节中，我们会基于这些原则继续探讨更自主、更具适应性的智能体架构。

5.1 理解智能体式工作流与智能体

由LLM驱动的智能体系统通常走两条路：智能体式工作流（agentic workflows）和智能体（agents）。这两种模式决定了应用如何运作，如图5.1所示。这两个术语经常被混用——但它们之间的差异其实很大。所以在继续深入之前，有必要先把这两个概念厘清：

智能体式工作流（简称工作流）—— 用一套预设的固定步骤引导应用运行。LLM的任务是在预定义选项里做选择，帮助系统完成任务并管理全局流程。

智能体（agent）—— 语言模型在此不仅是执行任务，它还会推理、决策，根据可用工具和不断变化的情况动态决定下一步如何走。这里的“工具”（tool），通常是一个能返回数据或处理数据的函数。

图5.1 工作流与智能体。工作流让LLM从固定选项里选择下一步，比如把请求路由到SQL数据库或REST API，结合结果生成答案。而智能体会动态挑选和组合工具来实现目标。

两种模式都依赖LLM驱动应用，但工作流走的是结构化、可预测的路径，智能体却能随着新信息和目标实时调整。下面先深入探讨工作流。

5.1.1 工作流

工作流让LLM从有限选项中挑选下一步，通常实现Controller-Worker或Router这类模式，如图5.2所示。

图5.2 常见工作流模式。Controller-Worker模式中，控制器用LLM按顺序给worker派任务。Router模式里，LLM只根据上下文把任务导向合适的worker。

Controller-Worker模式里，控制器按顺序给worker发任务；Router模式则只是将任务定向到对应的处理器（或worker）。本章重点在工作流，我们会把之前用LangChain搭建的Web研究应用改造成基于LangGraph的智能体式工作流。

5.1.2 智能体

基于LLM的智能体利用语言模型感知数据、推理、决定行动，最终实现目标。高级智能体还能记住过去的交互，动态构建工作流，甚至从反馈中学习。与固化“提示-响应”系统不同，智能体基于实时数据和可用工具生成新流程。本书第5部分（第11到14章）会介绍多工具智能体以及更复杂的多智能体系统。

5.1.3 什么时候使用基于智能体的架构

基于LLM的工作流和智能体关系紧密，经常重叠，没有一条绝对清晰的分界线。当你的应用需要把复杂任务拆成小步骤、根据前一步结果决策、访问外部工具或数据源、或者在长时间交互中维持上下文时，这两种方式都很有用。但只有真正需要显式状态管理和动态控制时，才值得引入智能体式工作流或智能体——也就是说，要在它们带来的能力与额外复杂度之间做好权衡。

提示　如果想更深入理解工作流、智能体以及何时使用，推荐阅读Anthropic的文章“Building Effective Agents”（https://mng.bz/lZ0o）。

5.1.4 智能体开发框架

现在有多款框架可用于构建智能体系统，每个都有不同的侧重点和取舍：

AutoGPT —— 强调完全自治、目标驱动，几乎不需要人工监督，但任务一致性可能不太好。

CrewAI —— 通过可共享的模板支持协作式多智能体系统，适合专门化团队。

LangGraph —— 通过基于图的执行方式构建有状态、可持久化的智能体式工作流。处理复杂应用很强大，但技术门槛较高。

LlamaIndex —— 知识检索方面突出，但覆盖面比通用智能体框架窄。

Microsoft Autogen —— 高度可定制的多智能体对话，学习曲线较陡。

Microsoft Semantic Kernel —— 更强调记忆与规划，和Azure服务集成良好。

n8n与LangFlow —— 可视化界面、集成能力强，对非开发者友好，但高级推理能力需要额外组件。

OpenAI Agent SDK与Google Agent Development Kit（ADK） —— 提供更精简的API，用于高效开发多智能体系统和智能体式工作流。

5.2 LangGraph 基础

LangGraph构建在LangChain之上，用于管理更复杂的智能体式工作流——包括分支路径、有状态处理，以及步骤间清晰的过渡。它是一个构建基于图的、有状态的、多步骤AI应用的框架。在LangGraph里，“节点”代表单独的任务，比如生成文本、调用API或分析数据；“边”定义连接这些任务的路径；“状态”则是在节点间流动、每一步都会更新的信息。当你需要做决策、管理状态，或者处理复杂的智能体式工作流时，这种方式比传统链更合适。

注意　LangGraph不是LangChain的替代品，而是扩展。可以把LangChain理解为各种构建积木，而LangGraph是蓝图，用来将这些部件连接成复杂系统。LangChain提供LLM、embeddings、retrievers等组件，LangGraph则帮你把这些组件组织成结构化、有状态的工作流。

想高效使用LangGraph，得先理解几个关键概念：图的核心组成部分（节点和边）、状态如何在图中流动，以及条件边如何控制图的行为。这些正是下一节要展开的基础构件。

5.3 从 LangChain 链迁移到 LangGraph

LangChain里那种简单、线性的链，对直接、单一路径的任务足够，但应用一复杂，局限就暴露了。典型的LangChain写法长这样：

chain = (prompt_template | llm | output_parser)

当任务需要分裂成不同路径、根据新信息反复执行某几步，或者要在多步过程中维护状态时，这种结构就显得吃力。多个过程需要并行时也一样不理想。

LangGraph通过更好的状态管理、条件分支和对循环工作流的支持解决了这些问题。显式状态管理让你能一致地定义并追踪整个工作流中的数据，对记忆和推理至关重要。条件分支允许智能体根据前一步结果选择不同路径，让决策更自然。循环工作流则让智能体能反复执行某项任务，直到条件满足，非常适合不断优化结果。

LangGraph也让复杂工作流更容易理解和调试。它的图结构让数据流向清晰可见，追踪或修复问题时轻松很多。

这种基于图的方式很适合多步骤推理、任务规划、长对话中的上下文管理、研究任务协调、业务流程自动化等用例。随着应用越来越复杂，LangGraph这种智能体架构的优势会越来越明显——它提供了构建智能多步骤系统所需的控制力和灵活性，让系统能自行适应并做出决策。

5.4 LangGraph 核心组件

LangGraph为构建有状态、多步骤AI应用提供了强大的框架。图5.3展示了构成LangGraph应用基础的核心组件。

图5.3 LangGraph核心组件。一个强类型状态（本例由ResearchState建模）在整个工作流中流动。节点通常是Python函数（例如def select_assistant），执行任务；边在节点间创建有向数据流，有时还带条件路径。

每个LangGraph应用的核心都是一个状态对象——在我们例子里叫ResearchState——它为整个工作流定义了一个清晰、强类型的状态。这个状态通常用Python的TypedDict定义，确保组件间传递的数据结构良好，并能做类型检查。

在LangGraph里，每个节点都是一个处理单元，负责生成搜索查询、调用外部API、摘要结果、转换数据等任务。这些节点通常以Python函数实现。节点之间的边决定了数据的有向流动，也就是信息如何在图中穿行。

LangGraph一个强大的特性是条件边（conditional edges），你可以基于运行时状态定义动态执行路径。再结合入口点（entry points）和结束条件（end conditions），就能完全控制图从哪里开始、如何推进、何时结束。接下来的小节会逐步讲解如何定义并连接这些组件，让你能构建出处理复杂工作流与自适应决策的系统。

5.4.1 StateGraph 结构

LangGraph的核心工具之一是StateGraph类，用它来定义描述应用工作流的图。例如：

from langgraph.graph import StateGraph
from typing import TypedDict
class ResearchState(TypedDict):
#1
input_query: str
intermediate_result: str
final_output: str
graph = StateGraph(ResearchState) #2
#1 定义状态结构
#2 创建图

5.4.2 状态管理与类型定义

状态管理是LangGraph应用的核心。链式方法依赖隐式或弱类型状态，而LangGraph强制使用显式、强类型的状态，让工作流更稳健、更可预测。下面是一个扩展版的ResearchState，包含更多细节：

from typing import TypedDict, Optional, List
class ResearchState(TypedDict):
user_question: str
assistant_info: Optional[dict]
search_queries: Optional[List[dict]]
search_results: Optional[List[dict]]
research_summary: Optional[str]
final_report: Optional[str]

每个节点接收当前状态，返回一些更新内容，这些更新会被合并进整体状态：

def process_node(state: dict) -> dict:
result = do_something(state["input_data"]) #1
return {"output_data": result} #2
#1 处理状态
#2 返回状态更新

5.4.3 节点函数与边的定义

节点代表处理步骤，每个节点是一个函数：接收当前状态，返回状态更新。例如：

def generate_search_queries(state: dict) -> dict:
"""根据用户问题生成搜索查询。"""
question = state["user_question"]
queries = llm_generate_queries(question)
return {"search_queries": queries}
graph.add_node("generate_queries", generate_search_queries) #1
#1 向图中添加节点

边定义了节点之间允许的迁移。一个简单的线性边写起来像这样：

graph.add_edge("generate_queries", "perform_searches")

而条件边会用一个函数，根据当前状态决定下一个节点是什么：

def should_refine_queries(state: dict) -> str:
if len(state["search_results"]) < 2:
return "refine_queries"
else:
return "summarize_results"
graph.add_conditional_edge("perform_searches", should_refine_queries)

5.4.4 入口点与结束条件

每张图都需要一个起点和清晰的结束条件，比如：

graph.set_entry_point("parse_question") #1
from langgraph.graph import END
graph.add_edge("write_final_report", END) #2
#1 设置入口点
#2 定义图结束

接下来，我们进入一个LangGraph的实际应用示例，在那里本节介绍的概念——强类型状态、节点函数、边、入口点和结束条件——会一起汇聚成一个真实工作流。

5.5 将 Web 研究助手改造成一个 AI 智能体

为了演示LangGraph怎么工作，下面把第4章那个基于LangChain的Web研究助手改造成一个基于智能体的系统。升级之后，应用能评估网页结果摘要的相关性：如果不到50%是相关的，就把流程重新导回“生成新搜索查询”这一步；如果相关摘要足够多，就继续写最终报告。要在纯LangChain里实现这种动态控制会很复杂，这正是转向LangGraph智能体方式的原因。这个案例研究会一步步带你看完整个改造过程，突出显式状态管理和模块化设计的好处。

5.5.1 原始 LangChain 实现概览

原来的Web研究助手用的是LangChain的顺序链，流程包括以下几个步骤：

• 根据用户问题选择合适的研究助手
• 生成搜索查询
• 执行Web搜索并收集URL
• 抓取并摘要每个搜索结果
• 编译最终研究报告

每一步的输出都作为下一步的输入，从下面这段原始实现代码摘录里看得出来。

代码清单5.1 Web研究助手的原始实现

assistant_instructions_chain = (
{'user_question': RunnablePassthrough()}
| ASSISTANT_SELECTION_PROMPT_TEMPLATE
| get_llm()
| StrOutputParser()
| to_obj
)
web_searches_chain = (
# ... 输入处理 ...
| WEB_SEARCH_PROMPT_TEMPLATE
| get_llm()
| StrOutputParser()
| to_obj
)
web_research_chain = ( #1
assistant_instructions_chain
| web_searches_chain
| search_and_summarization_chain.map()
| RunnableLambda(lambda x: # ... 处理结果...)
| RESEARCH_REPORT_PROMPT_TEMPLATE
| get_llm()
| StrOutputParser()
)
#1 最终研究链

这种方式虽然能用，但局限很明显：

• 流程刚性、线性，很难根据中间结果动态调整。比如想实现“如果摘要中不到50%相关，就回去重新生成搜索查询”的条件逻辑，用这种方式写起来很笨重。
• 错误处理困难，没有显式状态，很难追踪和管理失败。
• 状态没有被显式管理，多步骤之间维护上下文变得复杂。
• 一旦流程复杂，调试很痛苦，搞不清链中哪一部分出错了，也不知道原因。

5.5.2 识别要转换的组件

要把这个Web研究助手迁移到LangGraph，首先得识别出那些将被转换为节点的关键组件。每个节点负责流程中的某个特定部分：

Assistant Selector —— 根据用户问题决定使用哪种研究助手
Query Generator —— 基于用户输入创建搜索查询
Web Searcher —— 根据查询执行搜索并收集URL
Content Summarizer —— 抓取网页内容并生成摘要
Relevance Evaluator —— 评估这些摘要是否足够相关，决定继续还是重新生成搜索查询
Report Writer —— 用相关摘要编写最终研究报告

和简单的线性流程不同，这里引入了一个条件分支元素。评估摘要相关性之后，有两种可能：如果足够相关，就进Report Writer；如果不足，就返回Query Generator重新生成搜索查询。这个决策基于一个预定义阈值（比如不到50%的摘要相关），而且可以重复最多三轮，避免无限循环。

整个流程控制由一个条件路由函数route_based_on_relevance管理。它检查搜索结果的相关系数和当前迭代次数。如果相关性不足，且还没达到最大迭代次数，就生成新查询并重复搜索与评估；如果已经达到最大迭代次数，就无论相关性如何，直接用当前结果写报告。

对于每个组件，要定义三件事：

• 输入状态 —— 节点运行所需的数据
• 处理逻辑 —— 节点执行的任务
• 状态更新 —— 节点返回给整体状态的更新信息

这种模块化、条件化的做法，让系统既灵活又有适应性；而用LangChain的线性链来实现这些能力，会显得非常笨重。接下来，正式进入改造过程。

5.5.3 逐步转换过程

现在，一步步把一个LangChain应用转换成LangGraph应用。下面展示的是实际代码的简化版本，完整版可在GitHub仓库中找到。

第1步：定义状态

第一步，设计会在整张图中流动的状态结构。一个定义良好的状态，有助于在所有节点之间持续追踪数据。在这个例子中，用嵌套类型来建模一个复合状态，如代码清单5.2所示。这种状态结构清晰地定义了每个阶段可以访问哪些数据，减少歧义，简化调试。

代码清单5.2 基于LangGraph的研究助手的状态类型

from typing import List, Dict, Any, TypedDict, Optional

class AssistantInfo(TypedDict): #1
assistant_type: str
assistant_instructions: str
user_question: str

class SearchQuery(TypedDict): #1
search_query: str
user_question: str

class SearchResult(TypedDict): #1
result_url: str
search_query: str
user_question: str
is_fallback: Optional[bool]

class SearchSummary(TypedDict): #1
summary: str
result_url: str
user_question: str
is_fallback: Optional[bool]

class ResearchReport(TypedDict): #1
report: str

class ResearchState(TypedDict): #2
user_question: str
assistant_info: Optional[AssistantInfo]
search_queries: Optional[List[SearchQuery]]
search_results: Optional[List[SearchResult]]
search_summaries: Optional[List[SearchSummary]]
research_summary: Optional[str]
final_report: Optional[str]
used_fallback_search: Optional[bool]
relevance_evaluation: Optional[Dict[str, Any]]
should_regenerate_queries: Optional[bool]
iteration_count: Optional[int]
#1 用于状态处理的类型字典
#2 图状态

第2步：把组件转换成节点函数

接下来，把每个组件转换成一个节点函数。每个函数接收当前状态、执行处理逻辑，并返回更新后的状态信息，如下一个代码清单所示。这里展示的是简化版，完整实现可以在GitHub仓库中查看。

代码清单5.3 节点函数

def select_assistant(state: dict) -> dict:
"""选择合适的研究助手。"""
user_question = state["user_question"]
prompt = ASSISTANT_SELECTION_PROMPT_TEMPLATE.format(user_question=user_question)
response = get_llm().invoke(prompt)
assistant_info = parse_assistant_info(response.content) #1
return {"assistant_info": assistant_info} #2

def generate_search_queries(state: dict) -> dict:
"""根据问题生成搜索查询。"""
assistant_info = state["assistant_info"]
user_question = state["user_question"]
prompt = WEB_SEARCH_PROMPT_TEMPLATE.format( #3
assistant_instructions=assistant_info["assistant_instructions"],
user_question=user_question,
num_search_queries=3
)
response = get_llm().invoke(prompt)
search_queries = parse_search_queries(response.content) #4
return {"search_queries": search_queries} #5
#1 解析响应提取助手信息
#2 返回状态更新
#3 使用LLM创建查询
#4 解析响应得到搜索查询
#5 返回状态更新

其余节点函数也都遵循同样的模式。接下来定义整张图的结构。

第3步：定义图结构

节点函数都准备好了，就可以创建这张图，并定义节点之间如何连接，从而确定执行顺序与数据流向，如代码清单5.4所示。与简单的线性链不同，这个版本的图新增了一个“相关性评估”节点，以及一条条件边，用于根据搜索结果的相关性动态改变执行路径。

代码清单5.4 图结构

from langgraph.graph import StateGraph, END

graph = StateGraph(ResearchState) #1

graph.add_node("select_assistant", select_assistant) #2
graph.add_node("generate_search_queries", generate_search_queries) #2
graph.add_node("perform_web_searches", perform_web_searches) #2
graph.add_node("summarize_search_results", summarize_search_results) #2
graph.add_node("evaluate_search_relevance", evaluate_search_relevance) #2
graph.add_node("write_research_report", write_research_report) #2

def route_based_on_relevance(state): #3
iteration_count = state.get("iteration_count", 0) + 1
state["iteration_count"] = iteration_count
if iteration_count >= 3:
return "write_research_report"
if state.get("should_regenerate_queries", False):
return "generate_search_queries"
return "write_research_report"

graph.add_edge("select_assistant", "generate_search_queries") #4
graph.add_edge("generate_search_queries", "perform_web_searches") #4
graph.add_edge("perform_web_searches", "summarize_search_results") #4
graph.add_edge("summarize_search_results", "evaluate_search_relevance") #4
graph.add_edge("write_research_report", END) #4

graph.add_conditional_edges( #5
"evaluate_search_relevance",
route_based_on_relevance,
{"generate_search_queries": "generate_search_queries",
"write_research_report": "write_research_report"}
)

graph.set_entry_point("select_assistant") #6
#1 创建图
#2 添加节点
#3 定义条件相关性评估
#4 定义节点之间的边
#5 定义条件边
#6 设置入口点

新增的Relevance Evaluator节点会检查摘要结果中是否有足够比例的内容是相关的。如果少于50%的结果满足标准，图就会把流程重新导向Query Generator，优化搜索。如果摘要已经足够，或者已经达到最多三轮迭代，系统继续前进写最终报告。这种条件流控制，相比LangChain那种刚性的线性链，是一个显著的增强——系统能根据中间结果动态调整。

第4步：编译并运行图

定义完图之后，就可以像代码清单5.5那样，用一个初始状态来编译并运行它。这一步需要设置一个初始状态对象，填好所有必需字段，还包括用于控制条件流程的额外参数，比如should_regenerate_queries和iteration_count。

代码清单5.5 运行图

app = graph.compile() #1

initial_state = { #2
"user_question": "What can you tell me about Astorga's roman spas?",
"assistant_info": None,
"search_queries": None,
"search_results": None,
"search_summaries": None,
"research_summary": None,
"final_report": None,
"used_fallback_search": False,
"relevance_evaluation": None,
"should_regenerate_queries": None,
"iteration_count": 0
}

result = app.invoke(initial_state) #3
final_report = result["final_report"] #4
#1 编译图
#2 创建初始状态
#3 运行图
#4 提取最终报告

通过引入条件边和相关性评估，这个逐步转换把原本刚性、线性的链，变成了一个灵活、有状态、可适应的智能体式工作流。现在，系统能评估自己的结果，必要时通过优化搜索查询自我调整，确保最终报告建立在足够相关的信息之上。若用纯LangChain实现这种适应性，会非常笨重——这也说明了对于复杂LLM应用，LangGraph是更合理的选择。

5.5.4 代码对比与实现收益

这个案例研究展示了：与传统的LangChain链相比，LangGraph在复杂的多步骤AI应用中，如何显著增强灵活性、可控性和适应性。能够基于运行时评估结果实现条件流程控制，使LangGraph成为构建智能、具备上下文感知能力的智能体系统的强大工具。具体来说，LangGraph方案带来了以下重要收益：

显式状态管理 —— 状态被清晰定义，并在每个节点之间传递，使数据处理透明且可靠。

模块化组件 —— 每个节点只负责单一任务，让测试、调试和维护都更简单。

清晰的流程控制 —— 图结构把执行顺序和数据流向清楚地表达出来，复杂流程更容易追踪和理解。

更容易调试 —— 节点和边都清晰定义后，很容易定位出错位置以及触发错误的数据。

增强的错误处理能力 —— 每个节点可以实现各自独立的错误处理策略，而不影响系统其他部分。

条件流程控制 —— 基于相关性评估引入条件边之后，应用能动态改变执行路径：如果结果不足，就继续优化搜索查询；如果结果足够，就进入报告编写。这种适应能力让应用对中间结果做出智能响应，而这在纯LangChain中很难实现。

未来可扩展性更强 —— 添加或修改节点时，对整体系统改动很小，因此升级和扩展新能力更顺畅。

小结

• 智能体式工作流按预定义步骤顺序执行。智能体根据中间结果或错误动态选择工具并调整路径。
• LangGraph使用有向图来构建工作流。节点表示处理函数，边定义迁移关系，条件边根据运行时状态决定执行路径。研究助手就是一个例子：可以根据前几次搜索得到的内容质量，决定是继续搜索更多来源，还是开始整理结果。
• 状态管理用于追踪工作流各步骤之间的数据。节点从类型化状态对象中读写信息。对每个节点而言，状态是不可变的，但在整张图中会不断累积。
• 条件边根据运行时条件决定执行走向。比如，研究工作流可能在检索内容不足时回去继续搜索；如果来源足够，则继续进入综合整理阶段。
• StateGraph类用于定义整个工作流结构。节点函数执行离散任务（如搜索、解析、摘要），边按照定义的逻辑连接它们。
• 把线性链转换成LangGraph图之后，原本耦合的流程被拆成离散节点。这让调试、测试以及给工作流增加新能力都更简单。
• LangGraph工作流会保留执行历史和中间状态。你可以检查推理路径、从某个检查点重新播放工作流，或者基于先前决策分叉执行。在简单的LangChain链中实现这些能力非常困难。
• 可以用Python的TypedDict定义状态以获得强类型约束，例如：class ResearchState(TypedDict): question: str; search_queries: list[str]; results: list[dict]。这样确保节点间流过的数据是经过类型检查的。
• 使用graph = StateGraph(ResearchState)创建图，其中ResearchState就是你定义的强类型状态。这会强制所有节点接收并返回兼容的状态结构。
• 通过graph.add_node("node_name", node_function)添加节点，其中node_function是接收状态并返回状态更新的Python函数。节点函数在可能的情况下应尽量保持纯函数特性。
• 使用graph.add_edge("source_node", "destination_node")连接节点，构建有向数据流。这确定了节点之间的执行顺序。
• 通过graph.set_entry_point("first_node")定义入口点，或使用START常量标记执行起点。第一个节点会接收初始状态。
• 通过graph.add_edge("final_node", END)标记终点，表示工作流结束。执行到END时，系统停止并返回最终状态。
• 执行前需要先用app = graph.compile()编译图。这一步校验图结构并生成可执行应用。
• 节点函数接收当前状态作为输入，并返回部分状态更新，而不是整体替换整个状态。只需返回想更新的字段，例如：return {"search_queries": queries}。
• 可以通过graph.add_conditional_edges("source_node", router_function, {"option1": "node1", "option2": "node2"})添加条件边。路由函数需要返回一个字符串，必须匹配其中某个选项。
• 条件边的路由函数必须返回与后续节点名称匹配的字符串。可以使用if逻辑决定路由，例如：return "search_more" if len(results) < 3 else "write_report"。
• LangGraph是对LangChain的扩展，而不是替代。可以把LangChain中的组件（LLM、retriever、embedding等）当作构建积木，在LangGraph节点中使用，构建复杂工作流。