1. 为什么我们需要“图”来构建智能体如果你最近在折腾大语言模型应用尤其是想搞点能自主决策、有工作流的智能体那你大概率已经听过 LangChain 和 LangGraph 这两个名字了。LangChain 像是一个功能强大的工具箱帮你把各种工具和模型连接起来。但当你真正想构建一个能“思考”、有状态、能循环往复执行任务的智能体时你可能会发现用 LangChain 的链式结构来编排复杂逻辑就像试图用一根直线去描绘一个迷宫——不是不行但会非常别扭代码会变得又臭又长状态管理更是噩梦。这就是 LangGraph 登场的原因。它不是一个替代品而是一个强大的补充和进化。LangGraph 的核心思想很简单用“图”的思维来建模智能体的工作流。什么是图在计算机科学里图是由节点和边组成的结构。节点代表一个计算单元或一个动作边代表节点之间的流转路径和条件。这种模型天然适合描述“如果满足条件A就执行任务B否则执行任务C然后根据结果再决定下一步”这类带有分支、循环和状态依赖的复杂流程。想象一下你要构建一个“跨平台爆款图文生成Agent”。它的工作流绝对不是线性的。它可能需要先分析热点然后根据热点生成文案初稿接着调用文生图模型配图再检查图文匹配度如果不匹配可能需要重新生成文案或图片最后还要适配不同平台的发布格式。这个过程充满了判断、循环和状态传递。用传统的链式调用你需要写大量的if-else和状态管理代码而用 LangGraph你可以直观地画出这个工作流的“地图”让框架来帮你驱动执行。所以理解 LangGraph首先要理解它的思维模型——图计算模型。这不仅仅是学习一个新库的API更是学习一种设计和构建复杂、有状态AI应用的新范式。接下来我们就从最核心的StateGraph开始亲手构建我们的第一个图。2. 理解 LangGraph 的核心骨架StateGraph在 LangGraph 的世界里一切工作流都始于一个StateGraph。你可以把它理解为这个“图”的蓝图或者容器。它定义了两件最重要的事整个工作流共享的状态和状态如何在节点间流转。2.1 状态的定义让信息在流程中流动在链式结构中数据往往是以临时变量的形式传递或者被封装在复杂的上下文对象里。在图中我们显式地定义一个State。这个状态是一个字典或类它规定了工作流中需要共享和传递的所有信息。让我们用即将要构建的“图文生成Agent”来举例。我们的状态可能需要包含topic: 用户输入的主题或热点关键词。copywriting: 生成的文案草稿。image_prompt: 根据文案提炼出的图片提示词。image_url: 生成的图片的存储地址或URL。platform: 目标发布平台如小红书、知乎、微博。final_output: 最终整合好的、符合平台格式的图文内容。在 LangGraph 中我们通常使用TypedDict或 Pydantic 模型来定义这个状态这样能获得更好的类型提示和验证。这是构建健壮应用的第一步。from typing import TypedDict, List, Optional from langgraph.graph import StateGraph # 定义我们的工作流状态 class AgentState(TypedDict): topic: str copywriting: Optional[str] image_prompt: Optional[str] image_url: Optional[str] platform: str final_output: Optional[str] # 我们还可以加一个记录步骤的字段方便调试 steps: List[str] # 初始化一个状态图 workflow StateGraph(AgentState)这个AgentState就是贯穿我们整个图文生成流程的“共享白板”。每个节点都可以读取和修改上面的信息。2.2 节点的力量每个都是功能模块节点是图里的“工作单元”。在 LangGraph 中一个节点就是一个函数。这个函数接收当前完整的State作为输入然后返回一个更新后的State或者说是对 State 的更新指令。关键点在于节点是模块化的。一个节点应该只负责一件明确的事情。比如analyze_topic_node: 负责分析热点丰富主题。generate_copywriting_node: 负责调用LLM生成文案。create_image_prompt_node: 负责将文案转换成适合文生图模型的提示词。generate_image_node: 负责调用文生图API。format_output_node: 负责将文案和图片整合成特定平台的发布格式。这种设计让代码无比清晰也易于测试和复用。下面我们来定义一个最简单的节点def analyze_topic_node(state: AgentState) - AgentState: 分析主题节点丰富用户输入的主题信息。 # 从状态中读取当前主题 current_topic state[“topic”] # 这里可以调用LLM或者查询知识库来丰富这个主题 # 例如将“夏日穿搭”丰富为“2024年夏季清爽通勤穿搭指南主打简约风与多巴胺配色” enriched_topic call_llm_to_enrich_topic(current_topic) # 更新状态 new_state state.copy() new_state[“topic”] enriched_topic new_state[“steps”].append(f“主题已分析并丰富为{enriched_topic}”) # 注意我们也可以直接返回一个字典用于更新原状态这是更推荐的做法。 # 但为了概念清晰这里先展示直接返回新状态的方式。 return new_state # 将节点添加到图中 workflow.add_node(“analyze_topic”, analyze_topic_node)add_node方法给这个节点起了一个名字“analyze_topic”并将其函数绑定到图上。2.3 边的魔法控制流程的走向节点定义了“做什么”边则定义了“接下来去哪”。这是图模型最精髓的部分。边分为两种普通边Edge无条件地从节点A指向节点B。意味着节点A执行完后总是执行节点B。条件边Conditional Edge根据当前State中的某个条件决定下一步走哪条路。这实现了if-else和分支逻辑。在添加边之前我们必须指定一个入口节点即工作流的起点。# 设置入口节点工作流从这里开始 workflow.set_entry_point(“analyze_topic”) # 添加一条普通边分析主题后总是去生成文案 workflow.add_edge(“analyze_topic”, “generate_copywriting”)现在我们有了一个最简单的两节点线性图analyze_topic-generate_copywriting。但这还不够智能。我们需要条件边来实现“检查图片质量不合格则重试”这样的循环逻辑。3. 构建第一个可运行的工作流图让我们把概念落地构建一个虽然简化但完整可运行的图文生成流程。这个流程包括分析主题 - 生成文案 - 创建图片提示 - 生成图片。我们暂时用模拟函数代替真实的LLM和图像API调用。3.1 定义所有节点函数我们将定义四个节点函数每个都遵循“接收状态返回状态更新”的模式。注意LangGraph 更常见的做法是让节点函数返回一个字典这个字典中的键值对会用来更新原始的state对象而不是完全替换它。这更高效也是官方推荐的方式。from typing import Annotated import operator from langgraph.graph import StateGraph, END # 重新用Annotated定义状态这是LangGraph更现代的方式 from typing_extensions import TypedDict class AgentState(TypedDict): topic: str copywriting: Optional[str] image_prompt: Optional[str] image_url: Optional[str] platform: str final_output: Optional[str] steps: Annotated[list, operator.add] # 使用operator.add作为归约器表示steps是追加的 def analyze_topic_node(state: AgentState) - dict: 模拟分析并丰富主题。 enriched_topic f“[热点分析] {state[‘topic’]} - 附趋势解读与受众分析” return {“topic”: enriched_topic, “steps”: [“已完成主题热点分析”]} def generate_copywriting_node(state: AgentState) - dict: 模拟根据丰富后的主题生成文案。 copy f“爆款文案关于{state[‘topic’]}你必须知道的3件事#热门话题” return {“copywriting”: copy, “steps”: [“已生成初版文案”]} def create_image_prompt_node(state: AgentState) - dict: 模拟根据文案提取图片提示词。 prompt f“一幅表现‘{state[‘topic’]}’概念的、具有视觉冲击力的、社交媒体风格的插画” return {“image_prompt”: prompt, “steps”: [“已创建图片提示词”]} def generate_image_node(state: AgentState) - dict: 模拟调用文生图API生成图片。 # 这里假设我们调用了一个API返回了图片URL image_url f“https://fake-image-service/generate?prompt{state[‘image_prompt’]}” return {“image_url”: image_url, “steps”: [“已生成配图”]}3.2 组装图并设置流程现在我们创建图添加节点并用边把它们按顺序连接起来。# 创建状态图 workflow_builder StateGraph(AgentState) # 添加所有节点 workflow_builder.add_node(“analyze”, analyze_topic_node) workflow_builder.add_node(“write_copy”, generate_copywriting_node) workflow_builder.add_node(“make_prompt”, create_image_prompt_node) workflow_builder.add_node(“draw_image”, generate_image_node) # 设置入口 workflow_builder.set_entry_point(“analyze”) # 添加普通边形成线性流程分析 - 写文案 - 做提示 - 画图 workflow_builder.add_edge(“analyze”, “write_copy”) workflow_builder.add_edge(“write_copy”, “make_prompt”) workflow_builder.add_edge(“make_prompt”, “draw_image”) # 最后从“画图”节点连接到结束标志 END workflow_builder.add_edge(“draw_image”, END)3.3 编译与运行图定义好后需要将其“编译”成一个可执行的对象。# 编译图 graph workflow_builder.compile() # 准备初始状态 initial_state: AgentState { “topic”: “城市露营”, “copywriting”: None, “image_prompt”: None, “image_url”: None, “platform”: “小红书”, “final_output”: None, “steps”: [] } # 运行图 final_state graph.invoke(initial_state) print(“工作流执行完毕”) print(f“最终主题{final_state[‘topic’]}”) print(f“生成文案{final_state[‘copywriting’]}”) print(f“图片提示{final_state[‘image_prompt’]}”) print(f“图片地址{final_state[‘image_url’]}”) print(f“执行步骤{final_state[‘steps’]}”)执行这段代码你会看到状态如何一步步被每个节点更新最终得到所有产出。这就是一个最基本的 LangGraph 工作流。它目前是线性的但我们已经搭建好了所有基础设施。4. 引入条件逻辑让智能体真正“智能”起来线性流程只是自动化谈不上智能。智能体需要做判断。比如生成图片后我们需要一个“质检员”节点来评估图片与文案的匹配度。如果匹配度低就返回上一步重新生成提示词甚至文案。这就需要用到条件边。我们通过一个“路由函数”来实现它。4.1 创建质检节点与路由函数首先添加一个质检节点。def quality_check_node(state: AgentState) - dict: 模拟质量检查节点。 这里我们模拟一个简单的检查如果提示词里包含‘插画’就认为合格否则不合格。 实际应用中这里可以调用一个视觉-语言模型来打分。 prompt state[“image_prompt”] if prompt and “插画” in prompt: verdict “good” message “图片与文案主题匹配度高质检通过。” else: verdict “bad” message “图片风格与文案不符需要调整。” return {“quality_check”: verdict, “steps”: [message]} # 注意我们往状态里加了一个新键 # 在定义图之后添加这个节点 workflow_builder.add_node(“check_quality”, quality_check_node)现在关键来了。我们需要修改流程在draw_image画图节点之后不直接结束而是进入check_quality质检节点。质检节点之后根据结果决定下一步。# 删除之前从 draw_image 到 END 的边 # workflow_builder.add_edge(“draw_image”, END) # 这行需要被替换 # 改为画图之后进行质检 workflow_builder.add_edge(“draw_image”, “check_quality”)接下来定义路由函数。这个函数读取质检结果并返回下一个要执行的节点名称。def route_after_quality_check(state: AgentState) - str: 根据质检结果路由。 返回下一个节点的名字。 if state.get(“quality_check”) “good”: # 质检通过前往格式化输出节点假设我们后面会加这个节点或者直接结束 # 为了示例我们先去一个叫‘format’的节点如果不存在则去END return “format” # 或者 return END else: # 质检不通过返回‘make_prompt’节点重新生成提示词 # 更复杂的逻辑可以返回‘write_copy’甚至‘analyze’ return “make_prompt” # 添加条件边 workflow_builder.add_conditional_edges( “check_quality”, # 从哪个节点出发 route_after_quality_check, # 路由函数 # 指定路由函数可能返回的所有下一跳节点这是为了图的完整性检查 { “make_prompt”: “make_prompt”, # 如果返回“make_prompt”就去“make_prompt”节点 “format”: “format”, # 如果返回“format”就去“format”节点 # 也可以直接映射到 END # END: END } ) # 我们还需要补充‘format’节点和从它到END的边 def format_output_node(state: AgentState) - dict: final_output f“【{state[‘platform’]}】{state[‘copywriting’]}\n配图{state[‘image_url’]}” return {“final_output”: final_output, “steps”: [“已格式化最终输出”]} workflow_builder.add_node(“format”, format_output_node) workflow_builder.add_edge(“format”, END)4.2 理解循环与终止现在我们的图有了循环的可能make_prompt-draw_image-check_quality- (如果bad) -make_prompt。这会产生一个无限循环吗会的如果我们的质检逻辑永远返回“bad”。在实际应用中我们必须设置循环终止条件。常见的做法是在AgentState中增加一个计数器比如retry_count。在route_after_quality_check函数中除了检查质量还要检查重试次数。如果超过阈值则路由到一个“失败处理”节点或直接结束并记录失败原因。class AgentState(TypedDict): # ... 其他字段同上 ... retry_count: int 0 # 重试计数器 def route_after_quality_check(state: AgentState) - str: MAX_RETRY 3 if state.get(“quality_check”) “good”: return “format” else: if state[“retry_count”] MAX_RETRY: return “handle_failure” # 跳转到失败处理节点 else: # 返回上游节点前我们需要让状态知道即将重试 # 注意路由函数本身不修改状态修改需要在节点中进行 # 所以这里只是返回节点名。重试计数器的增加应该在‘make_prompt’或‘draw_image’节点中判断并完成。 return “make_prompt” # 在‘make_prompt’节点中可以判断是否来自重试并更新计数器 def create_image_prompt_node(state: AgentState) - dict: # ... 原有的提示词生成逻辑 ... # 判断如果state中已有image_prompt说明是重试则计数器1 updates {“image_prompt”: new_prompt, “steps”: [“已创建/重新创建图片提示词”]} if state.get(“image_prompt”): updates[“retry_count”] state.get(“retry_count”, 0) 1 return updates通过引入条件判断和计数器我们实现了受控的循环这是构建能处理复杂、非确定性任务的智能体的关键。5. 从概念到实战设计跨平台图文Agent的完整图有了上面的基础我们可以勾勒出“跨平台爆款图文Agent”更完整的图结构。这个图会比之前的例子复杂但设计思路一脉相承。5.1 定义完整状态与节点首先扩展我们的AgentState包含更多中间产物和决策依据。class AgentState(TypedDict): # 输入 raw_input: str # 用户原始输入可能是模糊的需求 target_platforms: List[str] # 目标平台列表如 [“小红书” “知乎”] # 分析与规划 refined_brief: dict # 精炼后的需求简报包含受众、风格、关键词等 content_angle: str # 内容切入点或“钩子” # 内容生产 copywriting_drafts: List[str] # 多个文案草稿 selected_copy_index: int # 选中的文案索引 image_prompts: List[str] # 多个图片提示词 selected_image_prompt: str generated_images: List[str] # 生成的图片URL列表 selected_image_index: int # 质量与适配 platform_specific_formats: dict # key为平台名value为该平台格式化后的内容 quality_scores: dict # 各个维度的质量评分 # 流程控制 current_step: str retry_map: dict # 记录各环节重试次数如 {“copywriting”: 0, “image_gen”: 0} error_log: List[str] final_outputs: dict # 最终成果key为平台名然后设计主要节点需求澄清节点与用户交互或自我推理将模糊输入转化为明确的需求简报。创意策划节点基于简报生成多个内容切入点角度。文案生成节点针对每个角度生成多个文案变体。文案评选节点调用LLM或规则选出最佳文案。图片提示词生成节点根据选定文案生成多个图片提示词。图片生成节点调用文生图API。图片评选节点评选最佳图片。多平台适配节点针对每个目标平台将文案和图片组合成符合该平台调性、尺寸、标签规则的格式。综合质检节点检查最终成品质量如有必要触发重试可能从文案或图片环节开始。输出打包节点生成最终可交付的成果。5.2 构建复杂的边与路由这是最体现设计功力的地方。图不再是简单的链或单一路由而是一个网络。从需求澄清到创意策划是普通边。创意策划后可能并行发起文案生成和图片提示词生成如果策略允许也可能先后进行。文案评选和图片评选都是条件边如果选不出满意的可能返回上一步重生成也可能触发创意策划重新找角度。多平台适配后进入综合质检。质检不通过可能路由回文案评选、图片评选甚至创意策划具体取决于失败原因的分析这需要质检节点在状态中记录详细原因。整个流程需要嵌入多个“断路器”比如同一个环节重试超过N次则路由到降级处理节点采用保底方案或直接报错退出。# 伪代码示意复杂路由 def master_router(state: AgentState) - str: current state[“current_step”] if current “quality_check”: if state[“quality_scores”][“overall”] 0.8: return “package_output” elif state[“quality_scores”][“copywriting”] 0.6: if state[“retry_map”][“copywriting”] 2: return “degrade_copywriting” return “select_copy” elif state[“quality_scores”][“image”] 0.6: if state[“retry_map”][“image_gen”] 2: return “degrade_image” return “select_image” else: return “replan_angle” # 整体概念不行重新策划 elif current “replan_angle”: if state[“retry_map”][“planning”] 1: return “final_fallback” return “creative_planning” # ... 其他路由逻辑5.3 调试与可视化对于复杂图LangGraph 提供了强大的调试和可视化支持这是开发过程中不可或缺的。可视化from IPython.display import Image, display try: display(Image(graph.get_graph().draw_mermaid_png())) except: # 如果无法生成图片可以输出文本表示 print(graph.get_graph().draw_ascii())这能生成一张流程图让你清晰看到所有节点和边的结构对于理解复杂流程和排查问题有巨大帮助。调试与状态追踪在graph.invoke()时可以传入config参数来启用调试或者通过回调函数来监听每个节点的输入输出。更简单的方法是充分利用我们在AgentState中设计的steps或current_step字段。在每个节点执行时都把关键动作和结果记录到steps列表中。运行结束后查看这个列表就能完整追溯工作流的执行路径和决策过程。final_state graph.invoke(initial_state, config{“configurable”: {“thread_id”: “test_run_1”}}) for step in final_state[“steps”]: print(step)6. 避坑指南与核心经验在从零开始构建 LangGraph 应用时有几个坑几乎每个人都会遇到提前了解能节省大量时间。坑一状态更新不生效现象节点里修改了状态字典但下一步节点读到的还是旧值。根因没有正确理解 LangGraph 的状态更新机制。节点函数应该返回一个字典这个字典的键值对会被“合并”到总状态中而不是直接替换整个状态。对于列表等可变对象的追加操作需要使用Annotated[list, operator.add]这样的归约器来声明。解决严格按照- dict的格式编写节点函数返回你想更新的部分。对于列表追加使用归约器。坑二条件边路由函数返回的节点名未定义现象运行时报错KeyError提示某个节点不存在。根因在add_conditional_edges时路由函数返回的字符串必须在path_map参数中预先声明并且对应的节点必须已经通过add_node添加到图中。解决仔细检查路由函数所有可能的返回值并在path_map中一一映射。确保这些节点名拼写正确且已添加。坑三陷入无限循环现象程序长时间不结束或者重复执行某几个节点。根因条件边逻辑有缺陷或者缺少终止条件。例如质检节点永远返回“不合格”就会在“生成-质检”间无限循环。解决必做在状态中引入重试计数器如retry_count在路由逻辑中判断。必做设计一个“安全出口”节点比如handle_failure或final_fallback在超过重试次数或出现不可恢复错误时路由到那里。调试可视化你的图沿着循环路径检查条件逻辑。在节点中打印状态和决策依据。坑四节点函数副作用过大现象工作流行为不可预测难以调试。根因节点函数除了通过返回字典更新状态外还修改了全局变量、进行了文件读写等外部操作。这破坏了图的“纯函数”理想使得状态流难以追踪。解决尽可能让节点是纯的。所有需要传递的信息都通过State进行。如果必须有副作用如调用外部API、写入数据库将其封装在节点内并确保该操作是幂等的或做好错误处理。避免在节点间通过全局变量通信。核心经验设计先行画图再写码在动手写代码前用纸笔或绘图工具画出工作流的草图明确节点、边和状态结构。这能极大减少后期的重构成本。状态设计是灵魂花时间精心设计State。它应该是工作流的“唯一真相源”。思考清楚哪些信息需要跨节点共享哪些是临时中间变量。良好的状态设计能让节点功能更单一路由逻辑更清晰。从简单到复杂迭代不要试图一开始就构建完美的复杂图。像我们这样先构建一个能跑通的线性流程然后逐步添加条件分支、循环和并行节点。每步都测试确保理解状态的变化。拥抱可视化工具graph.get_graph().draw_mermaid_png()是你的好朋友。经常可视化检查你的图结构是否符合预期。测试测试再测试为每个节点函数编写单元测试模拟输入状态验证输出状态。对于整个图用不同的初始状态进行端到端测试特别是要测试条件分支和循环路径。LangGraph 将构建复杂AI工作流从“编写控制流代码”的泥潭中解放出来变成了更高层次的“绘制流程图”。这种思维模式的转变是它最强大的地方。当你熟悉了以状态为中心、以图为蓝本的开发方式后你会发现构建那些需要多步骤推理、具备自我修正能力的智能体不再是一件令人望而生畏的事情。