1. 项目概述从单步执行到循环思考的跃迁如果你已经跟着上一篇内容用 LangGraph 搭出了一个能自动生成图文内容的 Agent 雏形那么恭喜你已经迈出了从零到一的关键一步。但那个 Agent 更像一个听话的“流水线工人”我们给它一个明确的指令比如“写一篇关于咖啡的短文并配图”它就会按部就班地执行“写作 - 配图”的固定流程。然而现实世界中的需求往往复杂多变充满了“如果...那么...”的决策点。比如用户可能说“帮我分析一下这个产品的优缺点如果优点多于缺点就生成一个宣传海报否则生成一个改进建议清单。” 这时固定的线性流程就失灵了。这正是我们本篇要解决的核心问题为 Agent 注入“思考循环”的能力。这不仅仅是让 Agent 能重复做某件事而是让它能根据中间结果动态地决定下一步做什么、做多久、以及何时停止。这背后依赖两个核心机制条件路由和工具调用。条件路由让 Agent 拥有了“判断力”而工具调用则是它执行判断、完成任务的具体“手脚”。通过 LangGraph 将这两者有机结合我们就能构建出能够应对复杂、动态场景的智能体这也是打造一个真正“爆款”级跨平台图文 Agent 的必经之路。本篇我们将深入 LangGraph 的实战拆解如何构建一个具备条件判断和循环执行能力的智能工作流。2. 核心概念拆解条件路由、工具调用与 ReAct 模式在动手写代码之前我们必须把地基打牢。理解这三个概念是构建高级 Agent 的基石。2.1 条件路由Agent 的决策神经网络你可以把条件路由想象成 Agent 大脑中的“决策树”或“流程图”。它不是一个简单的if-else语句而是基于 LangGraph 的State状态和Graph图结构实现的一种动态路径选择机制。状态State是决策的依据在 LangGraph 中一个字典或 Pydantic 模型贯穿整个工作流的始终它记录了当前任务的所有上下文信息比如用户的输入、LLM 的回复、工具执行的结果、循环次数等。条件路由函数的核心工作就是检查当前 State 中的某些关键字段的值。边Edges是可能的路径在图中我们定义多个不同的节点如“分析节点”、“写作节点”、“绘图节点”。条件路由决定了从当前节点出发下一步应该走向哪一个节点。如何工作LangGraph 允许我们为一个节点定义多个出口边每条边对应一个条件函数。系统会按顺序评估这些条件函数第一个返回True的条件所指向的节点将成为下一个执行的目标。一个常见的误区是试图在条件函数里做复杂的逻辑计算。实际上它的职责应该尽可能单一基于 State 中的某个明确标志做判断。复杂的逻辑判断应该前置到某个专门的“判断节点”通常由 LLM 驱动中由该节点将判断结果写入 State供后续的条件路由函数读取。2.2 工具调用Agent 与外部世界的交互之手工具调用是 Agent 能力的延伸。LLM 本身是“思想家”但它无法直接搜索网络、查询数据库、调用绘图 API 或发送邮件。工具Tools就是为 LLM 封装好的、可供其调用的具体函数。工具的定义在 LangChain/LangGraph 生态中一个工具通常是一个用tool装饰器修饰的 Python 函数或者是一个结构化的BaseTool子类。关键是要有清晰的名称name、描述description和参数模式args_schema。描述至关重要LLM 完全依赖描述来决定在什么情况下调用这个工具。绑定与调用我们将一系列工具绑定到一个 LLM 对象上例如通过bind_tools方法得到一个支持工具调用的 LLM。当这个 LLM 认为需要借助外部能力时它会在回复中输出一个特殊的结构化信息如ToolCall而不是普通文本。LangGraph 的运行时环境会捕获这个信息找到对应的工具函数并执行然后将执行结果以特定格式放回 State 中供 LLM 在下一轮思考时使用。工具的设计哲学工具应该保持“单一职责”和“高内聚”。一个工具只做一件事并把它做好。例如search_web工具只负责搜索并返回摘要generate_image工具只负责调用文生图 API。避免创建“瑞士军刀”式的巨型工具。2.3 ReAct 模式思考与行动的完美循环ReActReason Act是驱动智能 Agent 的核心范式它完美诠释了“条件路由”与“工具调用”是如何协同工作的。Reason思考LLM 基于当前任务状态State进行思考分析现状决定下一步需要做什么。它可能决定需要一项新信息也可能认为已经可以得出结论。Act行动如果思考后认为需要行动LLM 就会选择并调用一个合适的工具。这个“调用”是一个结构化动作。观察Observe工具执行的结果被写回到 State 中。循环基于新的、包含了工具执行结果的 StateLLM 再次进入“思考”阶段。如此循环直到 LLM 认为任务已经完成输出最终答案。在 LangGraph 中我们通常用一个节点来实现“思考-行动”这个组合步骤。这个节点里运行着绑定了工具的 LLM。LLM 的输出会被解析如果是工具调用就执行工具并更新状态然后通过条件路由让工作流再次回到这个“思考-行动”节点形成循环如果是最终答案就通过条件路由导向结束节点。注意一个设计良好的 ReAct 循环必须有一个明确的终止条件。否则 Agent 可能会陷入无限循环不断调用工具而不产出结果。终止条件通常由 LLM 在思考后决定例如输出一个final_answer的特殊标记并通过条件路由来实现。3. 实战构建具备审核循环的图文生成 Agent理论说得再多不如一行代码。让我们升级上一篇的简单图文流水线构建一个更智能的 Agent它不仅能生成图文还会对生成的内容进行自我审核。如果审核不通过它会尝试修改或重新生成直到满足条件或达到最大重试次数。3.1 定义增强版状态与工具首先我们需要一个更强大的状态模型来承载循环过程中的信息。from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): Agent 工作流的状态 # 消息历史用于LLM上下文 messages: Annotated[List, add_messages] # 用户原始输入 user_request: str # 当前生成的文本内容 current_content: str # 当前生成的图片URL或描述 current_image: str # 审核结果 approval_status: str # 例如”approved“, “needs_revision”, “rejected” # 审核意见 feedback: str # 循环/重试次数 iteration_count: int接下来定义几个核心工具。这里我们使用模拟工具来保持示例简洁在实际项目中你需要替换为真实的 API 调用。from langchain.tools import tool from langchain_core.messages import ToolMessage import random tool def generate_content(topic: str) - str: 根据给定主题生成一篇简短的营销文案或博客段落。 # 模拟LLM生成内容。实际应调用如ChatOpenAI等。 templates [ f探索{topic}的非凡世界我们发现其独特的魅力在于..., f在当今市场{topic}正引领着一场革新。我们的分析表明..., f你是否厌倦了平庸的{topic}这款产品将彻底改变你的体验... ] return random.choice(templates) tool def generate_image(prompt: str) - str: 根据文本描述生成一张图片返回图片的URL或文件路径。 # 模拟调用DALL-E、SD等API。实际应返回真实的URL。 image_id random.randint(1000, 9999) return fhttps://example.com/generated_image_{image_id}.png tool def content_reviewer(content: str) - dict: 审核一段文本内容。检查其是否积极、符合营销语气、无明显错误。 返回一个包含‘status’和‘feedback’字段的字典。 # 模拟一个审核逻辑。实际应用中这里可以调用另一个LLM进行专项审核。 if len(content) 30: status needs_revision feedback 内容过于简短请扩充细节以增强说服力。 elif 糟糕 in content or 差劲 in content: status rejected feedback 内容包含负面词汇不适合营销材料。 else: # 随机模拟一个通过或需要微调的情况 if random.random() 0.7: status approved feedback 内容优秀可以直接使用。 else: status needs_revision feedback 内容尚可但建议加入更多吸引眼球的形容词或用户痛点描述。 return {status: status, feedback: feedback}3.2 构建 LangGraph 节点与条件路由现在我们来构建图的工作节点。核心在于一个“创作与审核”循环。from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage import json # 初始化LLM并绑定工具 llm ChatOpenAI(modelgpt-4-turbo-preview, temperature0.7) llm_with_tools llm.bind_tools([generate_content, generate_image, content_reviewer]) def content_generation_node(state: AgentState): 节点1内容生成节点。根据用户请求生成文本和图片。 print(f[节点1] 第{state[iteration_count]1}次尝试生成内容...) user_request state[user_request] # 1. 调用LLM让其决定是否需要生成内容并可能直接调用工具 # 这里简化我们直接调用工具。更复杂的实现中可以让LLM来协调。 new_content generate_content.invoke({topic: user_request}) image_prompt f一幅关于{user_request}的、高质量、吸引人的矢量插画 new_image_url generate_image.invoke({prompt: image_prompt}) # 更新状态 state[current_content] new_content state[current_image] new_image_url # 初始化工审核状态 state[approval_status] pending state[feedback] return state def review_node(state: AgentState): 节点2审核节点。调用审核工具评估当前生成的内容。 print(f[节点2] 审核生成的内容...) content_to_review state[current_content] review_result content_reviewer.invoke({content: content_to_review}) state[approval_status] review_result[status] state[feedback] review_result[feedback] state[iteration_count] 1 print(f[节点2] 审核结果{review_result[status]}。反馈{review_result[feedback]}) return state def revision_node(state: AgentState): 节点3修订节点。基于审核反馈重新生成或修改内容。 print(f[节点3] 根据反馈进行修订...) feedback state[feedback] old_content state[current_content] user_request state[user_request] # 在实际应用中这里应该调用LLM将旧内容和反馈作为提示词生成修订版。 # 此处为模拟。 revised_content f[修订版] {old_content} 已根据反馈‘{feedback}’进行优化 # 也可能需要根据反馈重新生成图片 revised_image_prompt f根据反馈‘{feedback}’调整风格{user_request} revised_image_url generate_image.invoke({prompt: revised_image_prompt}) state[current_content] revised_content state[current_image] revised_image_url return state def finalize_node(state: AgentState): 节点4终审节点。准备最终输出。 print(f[节点4] 任务完成准备最终输出。) # 这里可以整理最终状态格式化输出等。 # 我们简单地在状态中标记完成。 state[approval_status] final_approved return state接下来是条件路由逻辑这是实现循环的关键。def should_continue(state: AgentState) - str: 条件路由函数根据审核状态决定下一步是结束、修订还是重新审核 返回下一个节点的名称。 status state[approval_status] iteration state[iteration_count] if status approved: # 审核通过进入最终节点 return finalize elif status rejected or iteration 3: # 设置最大重试次数为3 # 被拒绝或重试过多也进入最终节点可能是失败结果 print(f条件路由状态‘{status}’或迭代{iteration}次结束流程。) return finalize else: # status needs_revision # 需要修订进入修订节点 return revision3.3 组装工作流并测试将节点和边组装起来形成完整的工作流图。# 创建图构建器 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(generate, content_generation_node) workflow.add_node(review, review_node) workflow.add_node(revise, revision_node) workflow.add_node(finalize, finalize_node) # 设置入口边 workflow.set_entry_point(generate) workflow.add_edge(generate, review) # 生成后必然进入审核 # 设置条件边从审核节点出来根据条件路由决定去向 workflow.add_conditional_edges( review, should_continue, # 条件判断函数 { finalize: finalize, # 如果返回”finalize“则跳转到finalize节点 revision: revise, # 如果返回”revision“则跳转到revise节点 } ) # 设置修订后的边修订完成后应回到审核节点再次审核 workflow.add_edge(revise, review) # 设置最终边 workflow.add_edge(finalize, END) # 编译图 app workflow.compile()现在让我们运行这个具备循环能力的 Agent。# 初始化状态 initial_state { messages: [HumanMessage(content请为我们的新款智能咖啡杯创作推广图文)], user_request: 新款智能咖啡杯, current_content: , current_image: , approval_status: , feedback: , iteration_count: 0, } # 执行工作流 print(开始执行智能图文生成Agent工作流...) final_state app.invoke(initial_state, config{recursion_limit”: 50}) # 设置递归上限防止死循环 print(\n 工作流执行结束 ) print(f最终内容{final_state[current_content][:100]}...) print(f最终图片{final_state[current_image]}) print(f审核状态{final_state[approval_status]}) print(f总迭代次数{final_state[iteration_count]})运行上述代码你可能会看到类似如下的输出它清晰地展示了 Agent 的思考循环过程开始执行智能图文生成Agent工作流... [节点1] 第1次尝试生成内容... [节点2] 审核生成的内容... [节点2] 审核结果needs_revision。反馈内容尚可但建议加入更多吸引眼球的形容词或用户痛点描述。 [节点3] 根据反馈进行修订... [节点2] 审核生成的内容... [节点2] 审核结果approved。反馈内容优秀可以直接使用。 [节点4] 任务完成准备最终输出。 工作流执行结束 最终内容[修订版] 探索新款智能咖啡杯的非凡世界我们发现其独特的魅力在于... 已根据反馈‘内容尚可...’进行优化... 最终图片https://example.com/generated_image_7421.png 审核状态final_approved 总迭代次数24. 高级模式基于 LLM 决策的动态路由上面的例子中条件路由 (should_continue) 是基于一个简单的规则审核状态和迭代次数。但在更复杂的场景下决策逻辑本身可能就需要 LLM 的参与。例如Agent 可能需要自行判断“用户的问题是否已完全解答”或者“当前收集的信息是否足够做出决策”。这时我们可以创建一个专门的“路由决策节点”。这个节点里运行着一个 LLM它的任务就是分析当前 State然后输出下一个应该执行的节点名称。LangGraph 的add_conditional_edges可以很好地支持这种模式。from langchain_core.prompts import ChatPromptTemplate def llm_router_node(state: AgentState): 基于LLM的智能路由决策节点 # 构建给LLM的提示词让其根据当前状态做决策 router_prompt ChatPromptTemplate.from_messages([ (system, 你是一个工作流调度器。请根据当前的任务状态决定下一步应该执行哪个操作。你只能回复以下选项之一[generate_image, search_web, final_answer]), (human, 当前任务状态 用户问题{user_question} 已收集信息{collected_info} 当前步骤{current_step} 请决定下一步。 ) ]) # 假设我们有一个专用的路由LLM router_llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 格式化消息 messages router_prompt.format_messages( user_questionstate[user_request], collected_infostate.get(collected_info, 无), current_stepstate.get(current_step, start) ) # 获取LLM的决策 decision router_llm.invoke(messages).content.strip().lower() # 将决策写入状态供条件路由函数读取 state[next_node_decision] decision return state def route_based_on_llm_decision(state: AgentState) - str: 条件路由函数读取LLM的决策返回节点名 # 直接从状态中取出LLM的决策 decision state.get(next_node_decision, final_answer) # 确保决策是有效的节点名 if decision in [generate_image, search_web, final_answer]: return decision else: return final_answer # 默认安全出口然后在构建图时将add_conditional_edges的决策函数指向route_based_on_llm_decision。这样工作流的走向就完全由 LLM 的实时分析来动态控制了实现了更高层次的自主性。5. 避坑指南与性能优化在实际开发中你会遇到许多教程里不会提到的问题。以下是我从多个项目中总结出的关键经验。5.1 状态管理的常见陷阱状态字段未初始化在条件路由函数中访问一个不存在的状态字段会导致运行时错误。务必在初始状态或前序节点中为所有可能被访问的字段设置默认值如空字符串、空列表、0等。状态污染在节点函数中直接修改传入的state字典是安全的因为 LangGraph 默认使用dict的副本机制。但如果你使用了复杂的嵌套对象需要注意深拷贝问题。最佳实践是始终通过返回一个新的字典或使用state.update()来更新状态避免原地修改复杂对象。工具调用结果的处理工具执行后返回的结果需要正确地包装成ToolMessage并添加到state[‘messages’]中这样 LLM 才能在下一轮看到它。LangGraph 的ToolNode或tools_condition能帮你自动化这部分但手动处理时很容易遗漏。5.2 控制循环与超时防止无限循环这是 ReAct 模式最大的风险。除了依靠 LLM 自己说出“最终答案”外必须设置硬性终止条件最大迭代次数如我们例子中的iteration_count。超时机制在app.invoke时设置超时。递归限制编译图时使用app workflow.compile(recursion_limit100)。设置检查点对于长时间运行的工作流考虑将中间状态持久化如存入数据库。这样即使进程中断也能从上一个检查点恢复而不是从头开始。5.3 工具设计的黄金法则描述要精准工具的描述是 LLM 是否调用它的唯一依据。避免模糊的描述如“处理数据”而应使用“根据用户ID从MySQL users表中查询用户名和邮箱”。失败处理要优雅工具函数内部必须有完善的try-except异常处理。失败时应返回明确的错误信息如{error: API unavailable}而不是抛出异常导致整个工作流崩溃。LLM 需要看到错误信息才能决定下一步如重试或切换工具。考虑工具的成本与延迟将耗时短、成本低的工具如本地计算、缓存查询放在前面。像调用昂贵文生图 API 这样的工具最好在内容最终确认后再调用避免在修订循环中反复生成图片造成不必要的开销。5.4 调试与监控打印是关键在每个节点的开始和结束处打印状态的关键字段如我们示例中的print语句这是追踪工作流执行路径最直接的方法。可视化你的图LangGraph 提供了workflow.get_graph().draw_mermaid_png()功能需要安装pygraphviz生成工作流的可视化图片对于理解复杂流程和向他人解释非常有帮助。使用 LangSmith如果你在开发生产级应用强烈建议集成 LangSmith。它能记录每一次 LLM 调用、工具执行和状态流转提供完整的轨迹追踪、延迟分析和成本统计是调试和优化 Agent 的终极利器。构建一个具备思考循环的 Agent就像在教一个数字员工如何独立完成一项多步骤、有审核、可回溯的任务。LangGraph 提供的条件路由和工具调用机制为我们搭建这个“数字大脑”的决策与执行中枢提供了清晰的范式。从固定流水线到动态工作流的转变正是你的 Agent 从“玩具”迈向“工具”乃至成为“爆款产品”核心引擎的关键一步。记住一个好的工作流设计逻辑清晰胜过技巧堆砌。先从一个小而完整的循环开始逐步增加复杂度和智能你会更稳地走向成功。
