智能体的工作流(Workflow)怎么搭建?节点编排、条件分支、异步执行如何设计?

　引言

　　在开发智能体应用时，很多人会遇到这样的困境：大语言模型很强，能回答问题、能写代码、能做总结，但一旦任务涉及多个步骤——比如“先查天气，再根据天气推荐活动，然后预订餐厅，最后发送确认邮件”——单次模型调用就完全不够用了。你需要的不是“一个模型”，而是一个能串联多个工具、根据中间结果动态决策、并且能并行处理多个任务的流程系统。

　　这就是智能体工作流要解决的问题。如果你正在开发AI应用，尤其是那些涉及多步推理、多工具调用、动态决策的场景，却缺乏完善的工作流编排能力，那你的智能体注定只能停留在“玩具”阶段。

　　搭好工作流之后，你的智能体将具备以下能力：步骤之间自动衔接、根据条件选择不同路径、多个子任务并行执行提升效率、执行过程中支持人工介入和审批。整个系统从“一问一答”升级为“自动化任务处理流水线”。

　　本文将从零开始，逐步带你搭建一个完整的智能体工作流系统。预计阅读和动手实践时间约3至4小时，难度为中等偏上，建议你具备一定的编程基础和流程思维。我们会以LangGraph为主框架来展开，因为它是目前构建智能体工作流最成熟、最灵活的工具之一。

　　准备阶段

　　动手搭建之前，先准备好以下内容：

　　所需工具/材料：

　　Python 3.10以上环境

　　LangGraph库(pip install langgraph)

　　LangChain核心库(pip install langchain langchain-openai)

　　一个LLM的API密钥(OpenAI、Anthropic或兼容接口的本地模型均可)

　　需要调用的外部工具API密钥(如天气API、地图API等，根据你的业务场景准备)

　　可选：LangSmith用于调试和追踪(强烈推荐，能大幅提升开发效率)

　　前置条件：

　　熟悉Python异步编程基础(async/await)

　　理解大语言模型的基本调用方式

　　了解“图”数据结构的基本概念(节点、边、有向图)

　　已明确你要编排的具体业务流程(至少知道有哪些步骤、哪些条件分支)

　　环境/状态准备建议：

　　创建一个新的Python虚拟环境

　　提前准备好所有外部服务的API密钥，存入环境变量或.env文件

　　安装依赖：pip install langgraph langchain langchain-openai python-dotenv

　　建议在开发过程中开启LangSmith追踪(免费额度足够开发使用)，方便可视化调试工作流

　　详细步骤

　　步骤一：理解工作流的核心概念——状态、节点、边

　　在动手写代码之前，先吃透三个核心概念：

　　状态(State) ：工作流是“有记忆”的。状态就是一个在所有节点之间共享的字典对象，每个节点可以读取状态中的已有信息，也可以向状态中写入新信息。状态贯穿整个工作流生命周期。

　　节点(Node) ：工作流中的每一个步骤。节点可以是：调用一次大语言模型、执行一个工具函数、调用一个外部API、运行一段自定义逻辑，甚至是一个人工审批环节。

　　边(Edge) ：节点之间的连接关系。边决定了工作流的执行顺序。普通的边是“顺序执行”——A完成后一定去B。条件边则是“根据状态决定”——A完成后，根据状态中的某个值判断去B还是去C。

　　操作结果： 完成这一步后，你应该能用一句话描述工作流的本质：一个由节点组成、由状态驱动、由边连接的有向图。

　　常见问题提示： 初学者最容易混淆“状态”和“上下文”。在LangGraph中，状态是全局的、可变的、在节点间传递的数据容器;上下文通常指LLM调用的输入输出。两者有重叠但不完全相同，建议从状态视角设计工作流。

　　步骤二：定义工作流的状态结构

　　用LangGraph定义一个状态类。状态需要明确指定“每次更新时是覆盖还是追加”，这直接关系到后续节点能否正确处理数据。

　　python

　　f rom typing import TypedDict, List, Annotated

　　import operator

　　class AgentState(TypedDict):

　　# 用户输入，通常只在起始时设置一次

　　user_input: str

　　# 对话消息历史，使用operator.add实现追加而非覆盖

　　messages: Annotated[List[dict], operator.add]

　　# 当前执行步骤记录

　　current_step: str

　　# 中间结果存放处

　　intermediate_results: dict

　　# 最终输出

　　final_output: str

　　# 错误标记

　　has_error: bool

　　状态设计的原则：只存放节点之间需要共享的数据，不要把所有临时变量都塞进状态。过大的状态对象会降低可读性和调试效率。

　　操作结果： 完成这一步后，你的工作流有了统一的数据结构，所有节点将基于这个结构读写数据。

　　常见问题提示： Annotated[List, operator.add]是LangGraph中常用的追加模式。如果你希望某字段被覆盖而非追加，就不要使用operator.add，直接定义为普通类型即可。

　　步骤三：设计并实现各个节点

　　节点是工作流的基本执行单元。每个节点接收状态，执行逻辑，返回更新后的状态。

　　以下是一个典型的多节点工作流示例，包含：入口节点(接收用户输入)、LLM推理节点(调用大模型做决策)、工具调用节点(执行具体操作)和输出节点(生成最终回复)。

　　python

　　f rom langgraph.graph import StateGraph, END

　　f rom langchain_openai import ChatOpenAI

　　llm = ChatOpenAI(model="gpt-4")

　　# 节点1：入口节点——接收输入并初始化

　　def entry_node(state: AgentState):

　　return {

　　"current_step": "entry",

　　"messages": [{"role": "user", "content": state["user_input"]}]

　　}

　　# 节点2：LLM推理节点——调用模型分析意图

　　def llm_decision_node(state: AgentState):

　　response = llm.invoke(state["messages"])

　　# 解析模型的决策结果(例如判断是否需要调用工具)

　　return {

　　"current_step": "decision",

　　"messages": [{"role": "assistant", "content": response.content}],

　　"intermediate_results": {"decision": response.content}

　　}

　　# 节点3：工具执行节点——根据决策调用对应工具

　　def tool_execution_node(state: AgentState):

　　decision = state["intermediate_results"].get("decision", "")

　　if "查天气" in decision:

　　# 调用天气API

　　result = call_weather_api(state["user_input"])

　　elif "搜资料" in decision:

　　# 调用搜索API

　　result = call_search_api(state["user_input"])

　　else:

　　result = "未匹配到任何工具"

　　return {

　　"current_step": "tool_execution",

　　"intermediate_results": {"tool_result": result}

　　}

　　# 节点4：输出节点——生成最终回复

　　def output_node(state: AgentState):

　　result = state["intermediate_results"].get("tool_result", "无结果")

　　final = llm.invoke([{"role": "system", "content": f"根据工具结果{result}，生成友好的最终回复"}])

　　return {

　　"current_step": "output",

　　"final_output": final.content

　　}

　　操作结果： 完成这一步后，你拥有了工作流的全部功能节点，每个节点职责单一、逻辑清晰。

　　常见问题提示： 每个节点的返回值只会被合并到状态中，而不是覆盖整个状态。所以你只需要返回发生变化的字段即可，未返回的字段保持不变。这个“增量更新”机制是LangGraph的核心设计之一。

　　步骤四：编排节点——用边定义执行顺序

　　节点写好了，接下来用“边”把它们串联起来。LangGraph使用add_edge和add_conditional_edges来定义流程。

　　python

　　# 创建工作流图

　　workflow = StateGraph(AgentState)

　　# 添加所有节点

　　workflow.add_node("entry", entry_node)

　　workflow.add_node("decision", llm_decision_node)

　　workflow.add_node("tool", tool_execution_node)

　　workflow.add_node("output", output_node)

　　# 定义起点

　　workflow.set_entry_point("entry")

　　# 顺序边：entry完成后一定执行decision

　　workflow.add_edge("entry", "decision")

　　# 条件边：decision完成后，根据状态决定去tool还是直接去output

　　def route_after_decision(state: AgentState):

　　decision = state["intermediate_results"].get("decision", "")

　　if "需要工具" in decision o r  "查" in decision o r  "搜" in decision:

　　return "tool"

　　else:

　　return "output"

　　workflow.add_conditional_edges(

　　"decision",

　　route_after_decision,

　　{

　　"tool": "tool",

　　"output": "output"

　　}

　　)

　　# 顺序边：tool完成后去output

　　workflow.add_edge("tool", "output")

　　# 定义终点

　　workflow.add_edge("output", END)

　　# 编译工作流

　　app = workflow.compile()

　　操作结果： 完成这一步后，你的工作流不再是孤立的节点集合，而是一个完整的、可执行的流程图。

　　常见问题提示： 条件边的路由映射必须与返回值严格对应。如果route_after_decision返回了"unknown"，但映射字典中没有"unknown"这个键，工作流会抛出异常。建议在路由函数中设置默认分支。

　　步骤五：运行工作流并观察执行轨迹

　　编译完成后，就可以运行工作流了。运行方式支持同步和异步两种模式。

　　同步执行适用于简单场景：

　　python

　　initial_state = {

　　"user_input": "今天北京天气怎么样?",

　　"current_step": "",

　　"intermediate_results": {},

　　"final_output": "",

　　"has_error": False

　　}

　　# 执行工作流

　　final_state = app.invoke(initial_state)

　　print(final_state["final_output"])

　　异步执行适用于需要调用外部API、等待用户输入或并行处理多个子任务的场景(下一节会深入展开)：

　　python

　　async def run_async():

　　final_state = await app.ainvoke(initial_state)

　　print(final_state["final_output"])

　　LangGraph还支持流式输出(stream方法)，可以实时观察每个节点的执行进度和中间输出。

　　操作结果： 完成这一步后，你的智能体工作流能够完整运行，并能根据中间决策动态选择执行路径。

　　常见问题提示： 初次运行失败时，务必开启LangSmith追踪或打印每个节点的状态变化。90%的bug都来自“节点间数据格式不匹配”或“条件分支的路由逻辑写错了”。

　　步骤六：实现异步执行与并行节点

　　异步执行是智能体工作流从“简单”走向“实用”的关键能力。当工作流中有多个互不依赖的子任务时，并行执行能大幅缩短总耗时。

　　LangGraph原生支持节点级的异步执行。如果你的节点函数被定义为async def，LangGraph会在执行时自动调度。

　　示例：将“调用两个不同的API”改为并行执行

　　python

　　import asyncio

　　async def async_tool_1(state: AgentState):

　　# 模拟耗时API调用

　　await asyncio.sleep(2)

　　return {"intermediate_results": {"result_1": "API 1结果"}}

　　async def async_tool_2(state: AgentState):

　　await asyncio.sleep(2)

　　return {"intermediate_results": {"result_2": "API 2结果"}}

　　# 注意：这两个节点在图中需要设计为并行执行

　　# 方法是让它们在图中没有先后依赖关系，即从同一个父节点出发

　　workflow.add_node("api1", async_tool_1)

　　workflow.add_node("api2", async_tool_2)

　　workflow.add_edge("entry", "api1")

　　workflow.add_edge("entry", "api2") # 从entry同时指向两个节点，它们会被并发执行

　　workflow.add_edge("api1", "merge")

　　workflow.add_edge("api2", "merge") # 两个都完成后才进入merge

　　操作结果： 完成这一步后，你的工作流具备并行处理能力，多任务场景下效率提升50%以上。

　　常见问题提示： 并行节点的结果合并需要特别设计。如果两个节点都试图修改状态中的同一个字段，可能出现“后写入覆盖先写入”的问题。建议并行节点写入不同的字段，在合并节点中统一整合。

　　步骤七：添加检查点与状态持久化(可选但推荐)

　　智能体工作流在实际应用中往往需要支持中断和恢复——例如执行到某一步需要等待用户审批、等待外部回调或等待定时触发。LangGraph提供了检查点(Checkpoint)机制，能够保存和恢复工作流的状态。

　　python

　　f rom langgraph.checkpoint import MemorySaver

　　# 创建带有内存检查点的工作流

　　memory = MemorySaver()

　　app = workflow.compile(checkpointer=memory)

　　# 执行时可以指定thread_id，用于后续恢复

　　config = {"configurable": {"thread_id": "task_12345"}}

　　final_state = app.invoke(initial_state, config=config)

　　# 下次执行时使用相同的thread_id即可恢复

　　resumed_state = app.invoke(update_state, config=config)

　　在生产环境中，MemorySaver应替换为SqliteSaver或PostgresSaver，实现持久化存储，避免服务重启后丢失所有执行状态。

　　操作结果： 完成这一步后，你的工作流支持长时间运行、支持人工介入、支持随时中断和恢复。

　　常见问题提示： 检查点保存的是整个状态，如果状态中包含不可序列化的对象(如数据库连接、文件句柄)，保存会失败。确保状态中的所有数据都是JSON可序列化的。

　　整体流程回顾

　　用简化版流程串一遍所有步骤：

　　第一步：理解概念 —— 搞清楚状态、节点、边是什么 → 第二步：定义状态 —— 用TypedDict明确数据结构 → 第三步：实现节点 —— 为每个步骤写一个函数 → 第四步：编排节点 —— 用add_edge和add_conditional_edges连接 → 第五步：编译运行 —— compile()后执行invoke() → 第六步：异步优化 —— 将独立任务改为并行执行 → 第七步：检查点 —— 添加持久化支持断点续跑。

　　关键节点标注：

　　条件分支函数(路由函数) —— 这里写错了，整个流程就走偏了，务必写单元测试验证

　　状态合并逻辑 —— 注意追加与覆盖的区别，字段一旦定义好就不要随意变更类型

　　异步节点的异常处理 —— 并行任务中一个失败了，其他任务可能仍在运行，需要设计统一的错误收集和降级策略

　　高级技巧

　　技巧一：人机回环(Human-in-the-Loop)。 在某些关键决策点——如大额支付、敏感数据操作、内容审核等——可以让工作流暂停并等待人工审批。LangGraph的interrupt机制可以实现这个能力，配合检查点使用效果更佳。

　　技巧二：子图嵌套。 当工作流过于复杂时，可以将一组节点打包成一个子图，然后在主图中作为单个节点引用。这能显著提升大型工作流的可维护性。

　　技巧三：流式输出与进度反馈。 如果你的智能体需要较长时间处理任务，不要等到全部完成再返回。利用astream_events接口，在每个节点完成时向前端推送进度消息，用户体验会好很多。

　　技巧四：错误重试与降级。 为每个节点配置独立的错误处理策略——重试几次、超时时间多长、失败后是终止还是跳过。LangGraph支持在节点层面添加retry_policy参数。

　　总结与下一步

　　智能体工作流是AI应用从“玩具”走向“工具”的关键一步。没有工作流的智能体只能做单轮问答，有了工作流，它才能完成“规划-执行-验证-调整”的完整闭环。

　　回顾全文的核心：状态是工作流的血液，节点是骨骼，边是神经。状态驱动决策，节点执行操作，边传递信息并控制流向。掌握了这三者的设计和编排，你就掌握了搭建任何复杂度智能体工作流的能力。

　　现在就从你的实际业务场景出发，画一张流程图，把每个步骤写成一个节点，把每个判断写成一个条件分支——你的第一个专业级智能体工作流就在眼前了。

　　如果你的团队目前缺乏AI工作流开发的人力，或者你想快速验证一个复杂的流程方案但缺少技术实现，一品威客网汇聚了全国各地的AI开发者和技术专家。你可以在任务大厅发布智能体工作流开发需求，系统将根据你的描述智能匹配合适的服务商;也可以在人才大厅浏览AI工程师和LangGraph专家的资料，查看他们的项目经验和用户评价;还可以在服务大厅查看各类商铺的案例参考，了解不同团队的实际交付质量。一品威客网覆盖设计、开发、AI应用等300多项细分服务，日均处理海量需求对接。通过雇主攻略学习如何高效发布任务、筛选靠谱服务商，借助V客优享改变你的工作方式——用人而不养人，足不出户就能获得专业的AI开发技术支持。一品威客汇聚百万服务商，提供文化创意与技术服务，热门标签频道让你快速找到所需服务，热门搜索词助你精准定位，给你优质的网站体验。无论你是要找人搭建完整的工作流系统，还是需要某个环节的技术咨询，一品威客都能帮你快速对接专业人才，让AI开发少走弯路。

　　FAQs

　　问：LangGraph和LangChain是什么关系?我必须要用它们吗?

　　LangChain是一个大语言模型应用开发框架，提供模型调用、工具集成、提示词模板等基础能力。LangGraph是专门用于构建“有状态、多步骤、可循环”智能体工作流的库，它构建在LangChain之上。你当然可以用纯代码和异步框架自己实现工作流，但LangGraph提供了状态管理、条件分支、检查点、可视化调试等开箱即用的能力，能节省你至少70%的开发时间。对于生产级项目，强烈建议使用LangGraph。

　　问：条件分支能嵌套多层吗?比如“先判断A，再判断B”?

　　可以。LangGraph的条件边可以串联使用：节点A → 条件边1 → 节点B 或 节点C → 节点B内再次使用条件边分发给不同子节点。实际上，通过在节点函数内部使用if-elif也可以实现简单的分支逻辑，但用条件边的好处是整个路由关系在图中显式可见，便于调试和维护。如果分支逻辑非常复杂(超过3层)，建议将条件判断封装到一个独立的“路由节点”中，保持其他节点逻辑的纯净。

　　问：工作流执行到一半失败了，已经产生的中间结果会丢失吗?

　　这取决于你是否启用了检查点机制。如果没有检查点，整个状态会随着程序崩溃而丢失。如果启用了MemorySaver或SqliteSaver，执行到每个节点时状态都会被自动保存。失败后，你可以用相同的thread_id重新运行工作流，它会从上一次检查点位置继续执行，而无需重新开始。但要注意：被中断的节点内部的执行进度(例如循环中的第5次迭代)是无法恢复的，只能重新执行该节点。

　　问：多个智能体工作流之间能否相互调用或组合?

　　可以。LangGraph支持“子图”机制——你将一个已编译的工作流作为节点嵌入到另一个更大的工作流中，实现工作流的复用和组合。此外，不同工作流之间也可以通过共享状态中的某些字段(如任务队列、共享数据库表)来实现间接协作。如果你的多个智能体需要共享记忆或知识库，可以设计一个中央状态服务，所有工作流通过读写该服务来交换信息。