1. 从“一问一答”到“任务驱动”为什么我们需要工作流编排如果你最近在折腾AI应用尤其是想搞点能“自己动起来”的智能体Agent那你肯定对ChatBot那种“你问我答”的交互模式感到不满足了。我们想要的是一个能理解复杂指令然后像老练的项目经理一样自己拆解任务、调用工具、协调步骤最终交付结果的“智能员工”。这个从简单的对话Chat到复杂的、有向无环的任务图Task Graph的跨越就是Agent系统从玩具走向实用的关键一步也是今天我们要深入聊的“编排与工作流”。回想一下你用LangChain或者类似框架搭的第一个Agent很可能就是一个ReActReasoning Acting模式的单步思考执行循环。你问“今天北京天气怎么样”它先“思考”要调用天气API然后执行调用最后把结果组织成自然语言回复给你。这个过程是线性的、一次性的。但现实世界的需求远不止于此。比如一个电商客服Agent接到用户指令“我想买一件适合周末郊游的男士冲锋衣预算800以内要防水好的顺便看看同品牌有没有搭配的徒步鞋推荐。” 这个需求里包含了多个子任务理解用户需求郊游、男士、预算、防水、查询商品库、进行商品筛选和对比、跨品类衣服和鞋子的关联推荐。这些任务之间有依赖关系先找到衣服才能基于品牌找鞋子也有并行可能筛选衣服的同时可以获取鞋子品类信息。如果还用单步循环硬怼代码会变成一团乱麻状态管理、错误处理、流程控制都会让人头疼。这就是工作流编排登场的时刻。它的核心价值在于将复杂的业务逻辑可视化、模块化、可管理化。通过定义一个个节点Node和连接边Edge我们构建出一张任务执行图。每个节点负责一个明确的原子操作比如“调用商品搜索API”、“调用LLM进行商品摘要”、“调用支付风控接口”节点之间的边则定义了数据流和控制流比如“当A节点成功完成后将它的输出作为输入触发B和C节点并行执行”。LangGraph、n8n、Dify工作流、Flowable等工具都是为了解决这个问题而生的。它们把我们从手写复杂状态机和回调地狱中解放出来让我们能更专注于业务逻辑本身。所以当我们谈《从零实现Agent系统》的“编排与工作流”时我们不是在讨论一个可选的炫技功能而是在解决Agent能否处理真实场景复杂需求的工程基石问题。接下来我会结合LangGraph这个目前非常流行的库带你从概念到实战一步步拆解如何将你的Chat式Agent升级为具备强大编排能力的任务驱动型系统。2. 理解核心范式LangGraph的三要素与有向无环图DAG在动手写代码之前我们必须先统一思想理解工作流编排背后的核心计算模型——有向无环图以及LangGraph是如何对这个模型进行抽象和实现的。这能帮你避免后期陷入“为什么我的流程跑飞了”的困惑。2.1 什么是有向无环图DAG你可以把DAG想象成一个项目的任务清单图。每个任务是一个“节点”任务之间的依赖关系是“边”。关键有两点1.有向边是有方向的从任务A指向任务B意味着A必须在B之前完成。2.无环你不可能沿着箭头方向走最后又回到起点。这意味着不能有循环依赖否则就死锁了永远执行不完。在Agent工作流中节点可以是任何执行单元调用一次大语言模型LLM、执行一段Python函数、查询一次数据库、调用一个外部API。边则定义了执行路径“如果节点A的输出包含关键词‘需要审核’则走向人工审核节点B否则直接走向自动归档节点C。” 这种基于条件的路由Conditional Routing是让工作流“智能”起来的关键。2.2 LangGraph的核心抽象State、Node、EdgeLangGraph的设计非常精炼它用三个核心概念就构建了整个编排世界1. State状态这是工作流的“记忆中枢”和“数据总线”。它是一个字典或Pydantic模型在整个图执行过程中流动、被各个节点读写。State里通常包含用户的原始输入input、LLM的对话历史messages、中间计算结果intermediate_results、流程控制标志next_step等。在LangGraph中你会定义一个StateGraph并明确指定这个State的结构。所有节点都接收这个完整的State作为输入并返回一个更新后的State或其中的一部分。这种设计使得数据传递变得显式和规范。2. Node节点节点就是一个普通的Python函数或可调用对象它接收当前的State执行一些操作然后返回一个更新后的State。关键点在于节点应该保持“单一职责”。例如llm_node: 专门负责调用LLM将State中的messages历史发给模型并把返回的新消息追加到State中。tool_node: 专门负责调用某个工具比如计算器、搜索引擎将结果存入State的intermediate_results。router_node: 专门负责分析State内容决定下一步该走哪条边。3. Edge边边定义了节点之间的执行顺序。分为两种普通边add_edge无条件地从源节点指向目标节点。A做完就一定做B。条件边add_conditional_edges这是LangGraph的灵魂。它允许你根据State的内容动态决定下一步走向哪个节点。你需要提供一个路由函数routing_function它检查State并返回下一个要执行的节点名称。通过组合State、Node和Conditional Edge你可以构建出极其灵活的工作流比如循环直到某个条件满足才退出、分支根据不同情况走不同处理路径、并行虽然LangGraph默认是顺序执行但可以通过子图或外部协调实现并行语义。2.3 LangGraph vs LangChain不是替代是进化很多人会混淆LangGraph和LangChain。简单来说LangChain是一个构建LLM应用的全功能工具箱它早期也包含了一些简单的链Chain和代理Agent的序列化执行能力。而LangGraph则是专注于复杂、有状态、带循环和条件分支的工作流编排引擎。它更底层、更灵活能够编排任何东西不仅仅是LLM调用。你可以把LangChain里的各种组件LLM、Tool、Retriever作为Node用LangGraph来编排它们从而构建出比传统LangChain Agent更强大、更可控的智能体。所以它们的关系是互补而非对立。LangGraph解决了LangChain在复杂流程编排上的短板。3. 实战构建一个智能内容创作工作流光说不练假把式。让我们来设计并实现一个相对复杂的智能内容创作Agent工作流。这个工作流的任务是根据用户一个模糊的主题请求例如“写一篇关于Python异步编程的博客”自动完成从大纲生成、章节细化、到图片建议最后整合成文的完整过程。3.1 定义工作流State与整体架构首先我们需要规划整个流程需要哪些数据也就是定义State。我们使用TypedDict来获得更好的类型提示。from typing import TypedDict, List, Annotated import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage # 1. 定义State这是我们工作流的共享内存 class ContentCreationState(TypedDict): # 用户输入 user_request: str # 工作流各阶段产出 blog_topic: str outline: List[str] expanded_sections: List[dict] # 每个元素是 {section_title: str, content: str} image_suggestions: List[str] final_blog: str # 控制流标志 current_step: str needs_revision: bool我们的工作流设计如下节点解析需求接收用户请求明确博客主题。节点生成大纲基于主题生成一份博客大纲章节列表。条件边判断大纲是否需要人工修订如果是跳转到“人工修订”节点否则继续。节点细化章节并行或顺序地为大纲中的每个章节生成详细内容。节点生成图片建议基于文章内容为每个主要章节建议配图。节点整合成文将所有章节内容和图片建议整合成一篇格式优美的Markdown文章。结束。3.2 实现各个功能节点每个节点都是一个纯函数。我们使用LangChain的ChatOpenAI作为LLM但你也可以替换成任何模型。llm ChatOpenAI(modelgpt-4o, temperature0.7) def parse_request(state: ContentCreationState) - ContentCreationState: 节点1解析用户请求明确主题 user_request state[user_request] prompt f 用户提出了以下内容创作请求{user_request} 请从中提炼出一个具体、明确的博客文章主题标题要求主题清晰、有吸引力且范围适中适合写一篇1500字左右的博客。 只返回这个主题标题不要有其他任何解释。 message [HumanMessage(contentprompt)] response llm.invoke(message) topic response.content.strip() return {blog_topic: topic, current_step: topic_extracted} def generate_outline(state: ContentCreationState) - ContentCreationState: 节点2根据主题生成大纲 topic state[blog_topic] prompt f 博客主题是{topic} 请为这个主题生成一份详细的博客大纲。要求 1. 列出5-7个主要章节标题。 2. 每个章节标题应该是一个动宾短语概括该节核心内容。 3. 大纲应逻辑连贯涵盖引言、核心内容、案例分析如果适用、总结等部分。 请以Python列表的格式返回例如[引言Python异步编程为何重要, 核心概念asyncio与事件循环, ...] message [HumanMessage(contentprompt)] response llm.invoke(message) # 简单处理实际应用中可能需要更复杂的解析来确保返回的是列表 import ast try: outline_list ast.literal_eval(response.content) except: # 如果解析失败按行分割并清理 outline_list [line.strip(- *).strip() for line in response.content.split(\n) if line.strip()] return {outline: outline_list, current_step: outline_generated} def human_revision(state: ContentCreationState) - ContentCreationState: 节点3人工修订大纲模拟 # 在实际应用中这里可以连接到一个用户界面让用户修改大纲。 # 此处我们模拟用户确认通过。 print(f【人工审核点】当前大纲为{state[outline]}) print(模拟用户审核后点击通过。) return {needs_revision: False, current_step: outline_revised} def expand_sections(state: ContentCreationState) - ContentCreationState: 节点4为大纲中的每个章节生成详细内容 topic state[blog_topic] outline state[outline] expanded [] for i, section_title in enumerate(outline): prompt f 我们正在撰写博客{topic} 当前需要细化的章节是{section_title} 请为这个章节撰写详细内容字数在300-500字左右。内容应专业、易懂并包含具体的代码示例或比喻如果适用。 返回格式直接返回该章节的完整文本内容。 message [HumanMessage(contentprompt)] response llm.invoke(message) expanded.append({section_title: section_title, content: response.content}) print(f 已生成章节: {section_title}) return {expanded_sections: expanded, current_step: sections_expanded} def suggest_images(state: ContentCreationState) - ContentCreationState: 节点5为文章生成配图建议 topic state[blog_topic] sections state[expanded_sections] all_content \n.join([s[content] for s in sections]) prompt f 博客主题{topic} 博客主要内容{all_content[:2000]}... 内容截断 请为这篇博客文章生成3-5个配图建议。每个建议描述图片应该展现什么场景或概念。 例如“一张展示事件循环工作流程的示意图。” 请以Python列表的格式返回。 message [HumanMessage(contentprompt)] response llm.invoke(message) import ast try: suggestions ast.literal_eval(response.content) except: suggestions [s.strip() for s in response.content.split(\n) if s.strip()] return {image_suggestions: suggestions, current_step: images_suggested} def assemble_final_blog(state: ContentCreationState) - ContentCreationState: 节点6整合所有内容生成最终Markdown topic state[blog_topic] sections state[expanded_sections] images state[image_suggestions] markdown_parts [f# {topic}\n\n] for sec in sections: markdown_parts.append(f## {sec[section_title]}\n\n) markdown_parts.append(f{sec[content]}\n\n) markdown_parts.append(## 配图建议\n\n) for idx, img_desc in enumerate(images): markdown_parts.append(f{idx1}. {img_desc}\n) final_md .join(markdown_parts) return {final_blog: final_md, current_step: final_assembled}3.3 构建图并设置条件路由这是将节点连接起来赋予工作流“智能”的关键步骤。# 初始化状态图指定我们的State类型 workflow StateGraph(ContentCreationState) # 2. 添加节点 workflow.add_node(parse_request, parse_request) workflow.add_node(generate_outline, generate_outline) workflow.add_node(human_revision, human_revision) # 人工干预节点 workflow.add_node(expand_sections, expand_sections) workflow.add_node(suggest_images, suggest_images) workflow.add_node(assemble_final, assemble_final_blog) # 3. 设置入口点 workflow.set_entry_point(parse_request) # 4. 添加普通边无条件顺序执行 workflow.add_edge(parse_request, generate_outline) workflow.add_edge(expand_sections, suggest_images) workflow.add_edge(suggest_images, assemble_final) # 5. 添加条件边关键 def route_after_outline(state: ContentCreationState) - str: 大纲生成后的路由函数决定是否需要人工修订 # 这里可以设置更复杂的逻辑比如检查大纲质量、长度等。 # 为了演示我们假设随机决定或根据某个规则。 # 我们简单模拟如果大纲长度小于3则认为需要修订。 if len(state.get(outline, [])) 3: return human_revision # 跳转到人工修订节点 else: return expand_sections # 否则继续执行章节细化 # 将条件边从 generate_outline 节点引出 workflow.add_conditional_edges( generate_outline, route_after_outline, { human_revision: human_revision, # 如果返回human_revision则去该节点 expand_sections: expand_sections # 如果返回expand_sections则去该节点 } ) # 6. 从人工修订节点出来后应该去章节细化节点 workflow.add_edge(human_revision, expand_sections) # 7. 设置终点 workflow.add_edge(assemble_final, END) # 8. 编译图 app workflow.compile()3.4 运行与可视化工作流现在我们可以运行这个工作流并查看其执行过程。# 定义初始状态 initial_state: ContentCreationState { user_request: 写一篇给中级开发者的Python异步编程入门指南要包含asyncio的实际用例。, blog_topic: , outline: [], expanded_sections: [], image_suggestions: [], final_blog: , current_step: , needs_revision: False } # 执行工作流 final_state app.invoke(initial_state) print(\n *50) print(工作流执行完成) print(f最终主题{final_state[blog_topic]}) print(f生成大纲条目数{len(final_state[outline])}) print(f生成章节数{len(final_state[expanded_sections])}) print(f图片建议数{len(final_state[image_suggestions])}) print(\n最终博客前500字符预览) print(final_state[final_blog][:500] ...)为了更直观地理解我们构建的图LangGraph提供了可视化方法需要安装graphviz。# 将图结构导出为PNG图片可选需要安装graphviz try: from IPython.display import Image, display display(Image(app.get_graph().draw_mermaid_png())) except: # 非Jupyter环境可以打印Mermaid文本 print(app.get_graph().draw_mermaid())这张图会清晰地展示出parse_request-generate_outline- (条件分支) -human_revision或expand_sections-suggest_images-assemble_final- END 的完整路径。可视化是调试复杂工作流的利器。4. 进阶技巧与避坑指南让工作流真正可靠构建出能跑通的工作流只是第一步。要让它在生产环境中可靠运行还需要考虑很多工程细节。下面是我在实际项目中踩过坑后总结的经验。4.1 状态State设计的艺术平衡灵活与清晰State是你的全局变量设计不好会成为维护的噩梦。切忌“大而全”的State不要把所有可能用到的数据都塞进一个State字典。这会导致节点职责不清难以调试。应该按功能域分组甚至可以考虑使用嵌套的Pydantic模型来获得更好的类型检查和文档。使用Annotated进行状态更新合并这是LangGraph的一个高级特性。当你多个节点可能并发修改State的不同部分时虽然LangGraph默认顺序执行但模式设计时可考虑未来扩展可以使用Annotated来声明如何合并更新。例如from typing import TypedDict, Annotated from langgraph.graph import add_messages class State(TypedDict): messages: Annotated[list, add_messages] # 专门用于消息列表的合并 data: str # 普通字段后写入的会覆盖先写入的这样当多个节点返回{messages: [new_message]}时这些消息会被追加到原有的messages列表而不是覆盖。这对于管理对话历史至关重要。为State设置初始值和校验使用Pydantic的BaseModel来定义State可以利用其数据验证和默认值功能避免节点读到None导致崩溃。4.2 错误处理与持久化工作流不是“一锤子买卖”线上的工作流可能运行几分钟甚至几小时必须考虑异常和中断。节点级别的Try-Catch在每个节点函数内部进行细致的异常捕获。不是简单打印日志而是要把错误信息以一种结构化的方式比如{error: str, step: node_name}写入State。这样你可以设计一个专门的error_handler节点根据错误类型决定是重试、转人工还是优雅失败。设置检查点Checkpoint与持久化LangGraph内置了检查点机制这是其核心优势之一。通过配置一个Checkpointer工作流在每次状态变更后都可以被持久化到数据库如SQLite、Postgres。这意味着服务重启无影响如果服务器崩溃可以从最后一个检查点恢复执行。实现“暂停-继续”对于需要人工审批的长流程可以将状态保存等人工操作完成后再根据检查点ID恢复执行。调试与审计你可以查看任意一次执行的历史状态序列就像看回放录像一样。from langgraph.checkpoint.sqlite import SqliteSaver memory SqliteSaver.from_conn_string(:memory:) # 使用内存数据库生产环境需用文件或网络DB app workflow.compile(checkpointermemory) # 执行时会返回一个线程ID config {configurable: {thread_id: user_123_session_1}} initial_state {...} # 第一次执行 result1 app.invoke(initial_state, config) # 假设流程在此暂停... 之后可以继续 result2 app.invoke({user_input: 批准}, config) # 从上次暂停的地方继续超时控制对于调用外部API或运行时间不确定的节点一定要设置超时。可以在节点函数内用asyncio.wait_for或者在部署时通过进程管理工具如Supervisor来限制。4.3 条件路由Conditional Edges的陷阱确保路由函数稳定可靠路由函数是工作流的“决策大脑”它必须绝对稳定。路由函数应保持纯净它只应读取State进行计算返回下一个节点名。切忌在路由函数中修改State或执行有副作用的操作如调用API。LangGraph的设计哲学是状态修改应在Node中完成。返回的节点名必须是字符串常量路由函数返回的必须是你在add_conditional_edges的映射字典里明确定义的键。一个常见错误是动态生成节点名这会导致图找不到目标节点而崩溃。如果逻辑复杂可以返回一个标志然后在映射字典里用这个标志对应到多个可能的节点。做好兜底路由总是为路由函数设置一个默认的、安全的出口比如指向一个error_handler节点或END。避免因为State中某个意外字段导致路由函数返回None或非法值。4.4 调试与监控给工作流装上“仪表盘”当工作流复杂后仅靠打印日志很难定位问题。利用LangGraph的追踪Tracing如果你使用LangSmithLangChain的监控平台LangGraph的每次调用、每个节点的输入输出都会自动记录。你可以清晰地看到State是如何一步步变化的哪个节点耗时最长哪个节点出错了。这是不可或缺的调试和优化工具。在State中植入追踪ID在初始State中就放入一个唯一的trace_id如UUID这个ID会贯穿整个工作流。在所有对外部服务的调用数据库、API以及日志中都带上这个trace_id。这样无论问题出在哪个环节你都可以通过这个ID在日志系统中串联起完整的执行链路。可视化不是一次性的不要只在设计时看图。考虑在管理后台动态渲染当前执行到哪个节点用不同颜色标记“执行中”、“成功”、“失败”的状态。这能极大提升运维效率。5. 超越基础复杂模式与架构思考当你掌握了单工作流的构建后可以开始思考更复杂的架构模式以应对真正的企业级场景。5.1 子图Subgraph与模块化管理复杂度一个庞大的工作流图会难以理解和维护。LangGraph允许你将一部分节点和边打包成一个子图这个子图对外表现得就像一个普通的节点。这是软件工程中“分而治之”思想的体现。何时使用子图当你有一组节点共同完成一个逻辑上独立的功能时比如“用户身份验证与授权”、“支付处理流程”、“生成周报数据聚合”等。好处复用性定义好的支付子图可以被订单处理工作流和退款处理工作流同时调用。可维护性修改支付逻辑时只需改动子图内部不影响外部调用者。简化主图主工作流图变得非常清晰只有几个高级别的子图节点。实现方式在LangGraph中你可以先构建一个独立的StateGraph并编译它然后将这个编译好的图对象通过add_node加入到另一个主图中。主图调用子图时会传入完整的State子图在其内部执行完毕后返回更新后的State。5.2 并行与异步执行提升效率LangGraph的默认执行引擎是顺序的、同步的。但在很多场景下节点之间没有依赖可以并行执行以缩短总耗时。LangGraph的“伪并行”虽然一个图内的节点是顺序执行但你可以通过设计利用外部机制实现并行。例如在expand_sections节点中你可以使用asyncio.gather并发调用多个LLM来生成不同章节的内容。但要注意这要求你的节点函数本身是异步的并且你使用的LLM等客户端支持异步调用。更彻底的并行多个工作流实例对于完全独立的任务如处理1000个用户的请求最有效的方式是启动1000个独立的工作流实例由你的Web框架如FastAPI或任务队列如Celery去并行处理。LangGraph的检查点机制能保证每个实例的状态独立且可持久化。协调并行任务如果并行任务之间有最终的数据聚合需求比如Map-Reduce模式你可以设计一个主工作流它负责派发子任务生成多个子工作流实例ID然后等待所有子任务完成通过轮询数据库或消息队列最后进行结果聚合。这通常需要结合外部存储来实现。5.3 与现有系统集成工作流作为服务最终你的Agent工作流需要嵌入到现有的应用生态中。暴露为API端点使用FastAPI、Flask等框架将编译好的LangGraphapp包装成一个REST API。请求体包含初始StateAPI返回一个执行线程ID。你可以提供另外的端点来查询状态、继续执行或强制终止。事件驱动让工作流监听消息队列如RabbitMQ、Kafka的事件。当收到“新订单”事件时自动触发订单处理工作流当收到“用户反馈”事件时触发客服跟进工作流。LangGraph本身不提供消息队列集成但这可以在你的API层或专门的消息消费者中实现。作为后台任务在Django、Spring等全栈框架中将耗时的工作流执行丢给Celery、RQ或框架自带的异步任务系统。关键是要妥善保存thread_id以便在需要时能够通过LangGraph的检查点恢复或查询状态。从简单的Chat到复杂的任务图Agent系统的能力边界被极大地拓展了。编排与工作流不是银弹它引入了额外的复杂度但它是管理复杂性的必要工具。通过LangGraph这样的框架我们获得了描述、可视化、执行和调试复杂AI业务流程的标准方法。记住好的工作流设计始于清晰的业务逻辑分解和稳健的状态管理。先用手绘出任务的节点和边想清楚“什么情况下该走哪条路”然后再开始编码。当你习惯了这种思维模式你会发现构建能处理真实世界复杂性的智能体不再是一个遥不可及的梦想而是一个可以逐步实现的工程目标。 LangGraph工作流编排：从DAG原理到Agent系统实战
1. 从“一问一答”到“任务驱动”为什么我们需要工作流编排如果你最近在折腾AI应用尤其是想搞点能“自己动起来”的智能体Agent那你肯定对ChatBot那种“你问我答”的交互模式感到不满足了。我们想要的是一个能理解复杂指令然后像老练的项目经理一样自己拆解任务、调用工具、协调步骤最终交付结果的“智能员工”。这个从简单的对话Chat到复杂的、有向无环的任务图Task Graph的跨越就是Agent系统从玩具走向实用的关键一步也是今天我们要深入聊的“编排与工作流”。回想一下你用LangChain或者类似框架搭的第一个Agent很可能就是一个ReActReasoning Acting模式的单步思考执行循环。你问“今天北京天气怎么样”它先“思考”要调用天气API然后执行调用最后把结果组织成自然语言回复给你。这个过程是线性的、一次性的。但现实世界的需求远不止于此。比如一个电商客服Agent接到用户指令“我想买一件适合周末郊游的男士冲锋衣预算800以内要防水好的顺便看看同品牌有没有搭配的徒步鞋推荐。” 这个需求里包含了多个子任务理解用户需求郊游、男士、预算、防水、查询商品库、进行商品筛选和对比、跨品类衣服和鞋子的关联推荐。这些任务之间有依赖关系先找到衣服才能基于品牌找鞋子也有并行可能筛选衣服的同时可以获取鞋子品类信息。如果还用单步循环硬怼代码会变成一团乱麻状态管理、错误处理、流程控制都会让人头疼。这就是工作流编排登场的时刻。它的核心价值在于将复杂的业务逻辑可视化、模块化、可管理化。通过定义一个个节点Node和连接边Edge我们构建出一张任务执行图。每个节点负责一个明确的原子操作比如“调用商品搜索API”、“调用LLM进行商品摘要”、“调用支付风控接口”节点之间的边则定义了数据流和控制流比如“当A节点成功完成后将它的输出作为输入触发B和C节点并行执行”。LangGraph、n8n、Dify工作流、Flowable等工具都是为了解决这个问题而生的。它们把我们从手写复杂状态机和回调地狱中解放出来让我们能更专注于业务逻辑本身。所以当我们谈《从零实现Agent系统》的“编排与工作流”时我们不是在讨论一个可选的炫技功能而是在解决Agent能否处理真实场景复杂需求的工程基石问题。接下来我会结合LangGraph这个目前非常流行的库带你从概念到实战一步步拆解如何将你的Chat式Agent升级为具备强大编排能力的任务驱动型系统。2. 理解核心范式LangGraph的三要素与有向无环图DAG在动手写代码之前我们必须先统一思想理解工作流编排背后的核心计算模型——有向无环图以及LangGraph是如何对这个模型进行抽象和实现的。这能帮你避免后期陷入“为什么我的流程跑飞了”的困惑。2.1 什么是有向无环图DAG你可以把DAG想象成一个项目的任务清单图。每个任务是一个“节点”任务之间的依赖关系是“边”。关键有两点1.有向边是有方向的从任务A指向任务B意味着A必须在B之前完成。2.无环你不可能沿着箭头方向走最后又回到起点。这意味着不能有循环依赖否则就死锁了永远执行不完。在Agent工作流中节点可以是任何执行单元调用一次大语言模型LLM、执行一段Python函数、查询一次数据库、调用一个外部API。边则定义了执行路径“如果节点A的输出包含关键词‘需要审核’则走向人工审核节点B否则直接走向自动归档节点C。” 这种基于条件的路由Conditional Routing是让工作流“智能”起来的关键。2.2 LangGraph的核心抽象State、Node、EdgeLangGraph的设计非常精炼它用三个核心概念就构建了整个编排世界1. State状态这是工作流的“记忆中枢”和“数据总线”。它是一个字典或Pydantic模型在整个图执行过程中流动、被各个节点读写。State里通常包含用户的原始输入input、LLM的对话历史messages、中间计算结果intermediate_results、流程控制标志next_step等。在LangGraph中你会定义一个StateGraph并明确指定这个State的结构。所有节点都接收这个完整的State作为输入并返回一个更新后的State或其中的一部分。这种设计使得数据传递变得显式和规范。2. Node节点节点就是一个普通的Python函数或可调用对象它接收当前的State执行一些操作然后返回一个更新后的State。关键点在于节点应该保持“单一职责”。例如llm_node: 专门负责调用LLM将State中的messages历史发给模型并把返回的新消息追加到State中。tool_node: 专门负责调用某个工具比如计算器、搜索引擎将结果存入State的intermediate_results。router_node: 专门负责分析State内容决定下一步该走哪条边。3. Edge边边定义了节点之间的执行顺序。分为两种普通边add_edge无条件地从源节点指向目标节点。A做完就一定做B。条件边add_conditional_edges这是LangGraph的灵魂。它允许你根据State的内容动态决定下一步走向哪个节点。你需要提供一个路由函数routing_function它检查State并返回下一个要执行的节点名称。通过组合State、Node和Conditional Edge你可以构建出极其灵活的工作流比如循环直到某个条件满足才退出、分支根据不同情况走不同处理路径、并行虽然LangGraph默认是顺序执行但可以通过子图或外部协调实现并行语义。2.3 LangGraph vs LangChain不是替代是进化很多人会混淆LangGraph和LangChain。简单来说LangChain是一个构建LLM应用的全功能工具箱它早期也包含了一些简单的链Chain和代理Agent的序列化执行能力。而LangGraph则是专注于复杂、有状态、带循环和条件分支的工作流编排引擎。它更底层、更灵活能够编排任何东西不仅仅是LLM调用。你可以把LangChain里的各种组件LLM、Tool、Retriever作为Node用LangGraph来编排它们从而构建出比传统LangChain Agent更强大、更可控的智能体。所以它们的关系是互补而非对立。LangGraph解决了LangChain在复杂流程编排上的短板。3. 实战构建一个智能内容创作工作流光说不练假把式。让我们来设计并实现一个相对复杂的智能内容创作Agent工作流。这个工作流的任务是根据用户一个模糊的主题请求例如“写一篇关于Python异步编程的博客”自动完成从大纲生成、章节细化、到图片建议最后整合成文的完整过程。3.1 定义工作流State与整体架构首先我们需要规划整个流程需要哪些数据也就是定义State。我们使用TypedDict来获得更好的类型提示。from typing import TypedDict, List, Annotated import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage # 1. 定义State这是我们工作流的共享内存 class ContentCreationState(TypedDict): # 用户输入 user_request: str # 工作流各阶段产出 blog_topic: str outline: List[str] expanded_sections: List[dict] # 每个元素是 {section_title: str, content: str} image_suggestions: List[str] final_blog: str # 控制流标志 current_step: str needs_revision: bool我们的工作流设计如下节点解析需求接收用户请求明确博客主题。节点生成大纲基于主题生成一份博客大纲章节列表。条件边判断大纲是否需要人工修订如果是跳转到“人工修订”节点否则继续。节点细化章节并行或顺序地为大纲中的每个章节生成详细内容。节点生成图片建议基于文章内容为每个主要章节建议配图。节点整合成文将所有章节内容和图片建议整合成一篇格式优美的Markdown文章。结束。3.2 实现各个功能节点每个节点都是一个纯函数。我们使用LangChain的ChatOpenAI作为LLM但你也可以替换成任何模型。llm ChatOpenAI(modelgpt-4o, temperature0.7) def parse_request(state: ContentCreationState) - ContentCreationState: 节点1解析用户请求明确主题 user_request state[user_request] prompt f 用户提出了以下内容创作请求{user_request} 请从中提炼出一个具体、明确的博客文章主题标题要求主题清晰、有吸引力且范围适中适合写一篇1500字左右的博客。 只返回这个主题标题不要有其他任何解释。 message [HumanMessage(contentprompt)] response llm.invoke(message) topic response.content.strip() return {blog_topic: topic, current_step: topic_extracted} def generate_outline(state: ContentCreationState) - ContentCreationState: 节点2根据主题生成大纲 topic state[blog_topic] prompt f 博客主题是{topic} 请为这个主题生成一份详细的博客大纲。要求 1. 列出5-7个主要章节标题。 2. 每个章节标题应该是一个动宾短语概括该节核心内容。 3. 大纲应逻辑连贯涵盖引言、核心内容、案例分析如果适用、总结等部分。 请以Python列表的格式返回例如[引言Python异步编程为何重要, 核心概念asyncio与事件循环, ...] message [HumanMessage(contentprompt)] response llm.invoke(message) # 简单处理实际应用中可能需要更复杂的解析来确保返回的是列表 import ast try: outline_list ast.literal_eval(response.content) except: # 如果解析失败按行分割并清理 outline_list [line.strip(- *).strip() for line in response.content.split(\n) if line.strip()] return {outline: outline_list, current_step: outline_generated} def human_revision(state: ContentCreationState) - ContentCreationState: 节点3人工修订大纲模拟 # 在实际应用中这里可以连接到一个用户界面让用户修改大纲。 # 此处我们模拟用户确认通过。 print(f【人工审核点】当前大纲为{state[outline]}) print(模拟用户审核后点击通过。) return {needs_revision: False, current_step: outline_revised} def expand_sections(state: ContentCreationState) - ContentCreationState: 节点4为大纲中的每个章节生成详细内容 topic state[blog_topic] outline state[outline] expanded [] for i, section_title in enumerate(outline): prompt f 我们正在撰写博客{topic} 当前需要细化的章节是{section_title} 请为这个章节撰写详细内容字数在300-500字左右。内容应专业、易懂并包含具体的代码示例或比喻如果适用。 返回格式直接返回该章节的完整文本内容。 message [HumanMessage(contentprompt)] response llm.invoke(message) expanded.append({section_title: section_title, content: response.content}) print(f 已生成章节: {section_title}) return {expanded_sections: expanded, current_step: sections_expanded} def suggest_images(state: ContentCreationState) - ContentCreationState: 节点5为文章生成配图建议 topic state[blog_topic] sections state[expanded_sections] all_content \n.join([s[content] for s in sections]) prompt f 博客主题{topic} 博客主要内容{all_content[:2000]}... 内容截断 请为这篇博客文章生成3-5个配图建议。每个建议描述图片应该展现什么场景或概念。 例如“一张展示事件循环工作流程的示意图。” 请以Python列表的格式返回。 message [HumanMessage(contentprompt)] response llm.invoke(message) import ast try: suggestions ast.literal_eval(response.content) except: suggestions [s.strip() for s in response.content.split(\n) if s.strip()] return {image_suggestions: suggestions, current_step: images_suggested} def assemble_final_blog(state: ContentCreationState) - ContentCreationState: 节点6整合所有内容生成最终Markdown topic state[blog_topic] sections state[expanded_sections] images state[image_suggestions] markdown_parts [f# {topic}\n\n] for sec in sections: markdown_parts.append(f## {sec[section_title]}\n\n) markdown_parts.append(f{sec[content]}\n\n) markdown_parts.append(## 配图建议\n\n) for idx, img_desc in enumerate(images): markdown_parts.append(f{idx1}. {img_desc}\n) final_md .join(markdown_parts) return {final_blog: final_md, current_step: final_assembled}3.3 构建图并设置条件路由这是将节点连接起来赋予工作流“智能”的关键步骤。# 初始化状态图指定我们的State类型 workflow StateGraph(ContentCreationState) # 2. 添加节点 workflow.add_node(parse_request, parse_request) workflow.add_node(generate_outline, generate_outline) workflow.add_node(human_revision, human_revision) # 人工干预节点 workflow.add_node(expand_sections, expand_sections) workflow.add_node(suggest_images, suggest_images) workflow.add_node(assemble_final, assemble_final_blog) # 3. 设置入口点 workflow.set_entry_point(parse_request) # 4. 添加普通边无条件顺序执行 workflow.add_edge(parse_request, generate_outline) workflow.add_edge(expand_sections, suggest_images) workflow.add_edge(suggest_images, assemble_final) # 5. 添加条件边关键 def route_after_outline(state: ContentCreationState) - str: 大纲生成后的路由函数决定是否需要人工修订 # 这里可以设置更复杂的逻辑比如检查大纲质量、长度等。 # 为了演示我们假设随机决定或根据某个规则。 # 我们简单模拟如果大纲长度小于3则认为需要修订。 if len(state.get(outline, [])) 3: return human_revision # 跳转到人工修订节点 else: return expand_sections # 否则继续执行章节细化 # 将条件边从 generate_outline 节点引出 workflow.add_conditional_edges( generate_outline, route_after_outline, { human_revision: human_revision, # 如果返回human_revision则去该节点 expand_sections: expand_sections # 如果返回expand_sections则去该节点 } ) # 6. 从人工修订节点出来后应该去章节细化节点 workflow.add_edge(human_revision, expand_sections) # 7. 设置终点 workflow.add_edge(assemble_final, END) # 8. 编译图 app workflow.compile()3.4 运行与可视化工作流现在我们可以运行这个工作流并查看其执行过程。# 定义初始状态 initial_state: ContentCreationState { user_request: 写一篇给中级开发者的Python异步编程入门指南要包含asyncio的实际用例。, blog_topic: , outline: [], expanded_sections: [], image_suggestions: [], final_blog: , current_step: , needs_revision: False } # 执行工作流 final_state app.invoke(initial_state) print(\n *50) print(工作流执行完成) print(f最终主题{final_state[blog_topic]}) print(f生成大纲条目数{len(final_state[outline])}) print(f生成章节数{len(final_state[expanded_sections])}) print(f图片建议数{len(final_state[image_suggestions])}) print(\n最终博客前500字符预览) print(final_state[final_blog][:500] ...)为了更直观地理解我们构建的图LangGraph提供了可视化方法需要安装graphviz。# 将图结构导出为PNG图片可选需要安装graphviz try: from IPython.display import Image, display display(Image(app.get_graph().draw_mermaid_png())) except: # 非Jupyter环境可以打印Mermaid文本 print(app.get_graph().draw_mermaid())这张图会清晰地展示出parse_request-generate_outline- (条件分支) -human_revision或expand_sections-suggest_images-assemble_final- END 的完整路径。可视化是调试复杂工作流的利器。4. 进阶技巧与避坑指南让工作流真正可靠构建出能跑通的工作流只是第一步。要让它在生产环境中可靠运行还需要考虑很多工程细节。下面是我在实际项目中踩过坑后总结的经验。4.1 状态State设计的艺术平衡灵活与清晰State是你的全局变量设计不好会成为维护的噩梦。切忌“大而全”的State不要把所有可能用到的数据都塞进一个State字典。这会导致节点职责不清难以调试。应该按功能域分组甚至可以考虑使用嵌套的Pydantic模型来获得更好的类型检查和文档。使用Annotated进行状态更新合并这是LangGraph的一个高级特性。当你多个节点可能并发修改State的不同部分时虽然LangGraph默认顺序执行但模式设计时可考虑未来扩展可以使用Annotated来声明如何合并更新。例如from typing import TypedDict, Annotated from langgraph.graph import add_messages class State(TypedDict): messages: Annotated[list, add_messages] # 专门用于消息列表的合并 data: str # 普通字段后写入的会覆盖先写入的这样当多个节点返回{messages: [new_message]}时这些消息会被追加到原有的messages列表而不是覆盖。这对于管理对话历史至关重要。为State设置初始值和校验使用Pydantic的BaseModel来定义State可以利用其数据验证和默认值功能避免节点读到None导致崩溃。4.2 错误处理与持久化工作流不是“一锤子买卖”线上的工作流可能运行几分钟甚至几小时必须考虑异常和中断。节点级别的Try-Catch在每个节点函数内部进行细致的异常捕获。不是简单打印日志而是要把错误信息以一种结构化的方式比如{error: str, step: node_name}写入State。这样你可以设计一个专门的error_handler节点根据错误类型决定是重试、转人工还是优雅失败。设置检查点Checkpoint与持久化LangGraph内置了检查点机制这是其核心优势之一。通过配置一个Checkpointer工作流在每次状态变更后都可以被持久化到数据库如SQLite、Postgres。这意味着服务重启无影响如果服务器崩溃可以从最后一个检查点恢复执行。实现“暂停-继续”对于需要人工审批的长流程可以将状态保存等人工操作完成后再根据检查点ID恢复执行。调试与审计你可以查看任意一次执行的历史状态序列就像看回放录像一样。from langgraph.checkpoint.sqlite import SqliteSaver memory SqliteSaver.from_conn_string(:memory:) # 使用内存数据库生产环境需用文件或网络DB app workflow.compile(checkpointermemory) # 执行时会返回一个线程ID config {configurable: {thread_id: user_123_session_1}} initial_state {...} # 第一次执行 result1 app.invoke(initial_state, config) # 假设流程在此暂停... 之后可以继续 result2 app.invoke({user_input: 批准}, config) # 从上次暂停的地方继续超时控制对于调用外部API或运行时间不确定的节点一定要设置超时。可以在节点函数内用asyncio.wait_for或者在部署时通过进程管理工具如Supervisor来限制。4.3 条件路由Conditional Edges的陷阱确保路由函数稳定可靠路由函数是工作流的“决策大脑”它必须绝对稳定。路由函数应保持纯净它只应读取State进行计算返回下一个节点名。切忌在路由函数中修改State或执行有副作用的操作如调用API。LangGraph的设计哲学是状态修改应在Node中完成。返回的节点名必须是字符串常量路由函数返回的必须是你在add_conditional_edges的映射字典里明确定义的键。一个常见错误是动态生成节点名这会导致图找不到目标节点而崩溃。如果逻辑复杂可以返回一个标志然后在映射字典里用这个标志对应到多个可能的节点。做好兜底路由总是为路由函数设置一个默认的、安全的出口比如指向一个error_handler节点或END。避免因为State中某个意外字段导致路由函数返回None或非法值。4.4 调试与监控给工作流装上“仪表盘”当工作流复杂后仅靠打印日志很难定位问题。利用LangGraph的追踪Tracing如果你使用LangSmithLangChain的监控平台LangGraph的每次调用、每个节点的输入输出都会自动记录。你可以清晰地看到State是如何一步步变化的哪个节点耗时最长哪个节点出错了。这是不可或缺的调试和优化工具。在State中植入追踪ID在初始State中就放入一个唯一的trace_id如UUID这个ID会贯穿整个工作流。在所有对外部服务的调用数据库、API以及日志中都带上这个trace_id。这样无论问题出在哪个环节你都可以通过这个ID在日志系统中串联起完整的执行链路。可视化不是一次性的不要只在设计时看图。考虑在管理后台动态渲染当前执行到哪个节点用不同颜色标记“执行中”、“成功”、“失败”的状态。这能极大提升运维效率。5. 超越基础复杂模式与架构思考当你掌握了单工作流的构建后可以开始思考更复杂的架构模式以应对真正的企业级场景。5.1 子图Subgraph与模块化管理复杂度一个庞大的工作流图会难以理解和维护。LangGraph允许你将一部分节点和边打包成一个子图这个子图对外表现得就像一个普通的节点。这是软件工程中“分而治之”思想的体现。何时使用子图当你有一组节点共同完成一个逻辑上独立的功能时比如“用户身份验证与授权”、“支付处理流程”、“生成周报数据聚合”等。好处复用性定义好的支付子图可以被订单处理工作流和退款处理工作流同时调用。可维护性修改支付逻辑时只需改动子图内部不影响外部调用者。简化主图主工作流图变得非常清晰只有几个高级别的子图节点。实现方式在LangGraph中你可以先构建一个独立的StateGraph并编译它然后将这个编译好的图对象通过add_node加入到另一个主图中。主图调用子图时会传入完整的State子图在其内部执行完毕后返回更新后的State。5.2 并行与异步执行提升效率LangGraph的默认执行引擎是顺序的、同步的。但在很多场景下节点之间没有依赖可以并行执行以缩短总耗时。LangGraph的“伪并行”虽然一个图内的节点是顺序执行但你可以通过设计利用外部机制实现并行。例如在expand_sections节点中你可以使用asyncio.gather并发调用多个LLM来生成不同章节的内容。但要注意这要求你的节点函数本身是异步的并且你使用的LLM等客户端支持异步调用。更彻底的并行多个工作流实例对于完全独立的任务如处理1000个用户的请求最有效的方式是启动1000个独立的工作流实例由你的Web框架如FastAPI或任务队列如Celery去并行处理。LangGraph的检查点机制能保证每个实例的状态独立且可持久化。协调并行任务如果并行任务之间有最终的数据聚合需求比如Map-Reduce模式你可以设计一个主工作流它负责派发子任务生成多个子工作流实例ID然后等待所有子任务完成通过轮询数据库或消息队列最后进行结果聚合。这通常需要结合外部存储来实现。5.3 与现有系统集成工作流作为服务最终你的Agent工作流需要嵌入到现有的应用生态中。暴露为API端点使用FastAPI、Flask等框架将编译好的LangGraphapp包装成一个REST API。请求体包含初始StateAPI返回一个执行线程ID。你可以提供另外的端点来查询状态、继续执行或强制终止。事件驱动让工作流监听消息队列如RabbitMQ、Kafka的事件。当收到“新订单”事件时自动触发订单处理工作流当收到“用户反馈”事件时触发客服跟进工作流。LangGraph本身不提供消息队列集成但这可以在你的API层或专门的消息消费者中实现。作为后台任务在Django、Spring等全栈框架中将耗时的工作流执行丢给Celery、RQ或框架自带的异步任务系统。关键是要妥善保存thread_id以便在需要时能够通过LangGraph的检查点恢复或查询状态。从简单的Chat到复杂的任务图Agent系统的能力边界被极大地拓展了。编排与工作流不是银弹它引入了额外的复杂度但它是管理复杂性的必要工具。通过LangGraph这样的框架我们获得了描述、可视化、执行和调试复杂AI业务流程的标准方法。记住好的工作流设计始于清晰的业务逻辑分解和稳健的状态管理。先用手绘出任务的节点和边想清楚“什么情况下该走哪条路”然后再开始编码。当你习惯了这种思维模式你会发现构建能处理真实世界复杂性的智能体不再是一个遥不可及的梦想而是一个可以逐步实现的工程目标。