LangGraph实战：构建具备条件路由与循环执行能力的智能Agent

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 模式：思考与行动的完美循环

ReAct（Reason + Act）是驱动智能 Agent 的核心范式，它完美诠释了“条件路由”与“工具调用”是如何协同工作的。

1. Reason（思考）：LLM 基于当前任务状态（State）进行思考，分析现状，决定下一步需要做什么。它可能决定需要一项新信息，也可能认为已经可以得出结论。
2. Act（行动）：如果思考后认为需要行动，LLM 就会选择并调用一个合适的工具。这个“调用”是一个结构化动作。
3. 观察（Observe）：工具执行的结果被写回到 State 中。
4. 循环：基于新的、包含了工具执行结果的 State，LLM 再次进入“思考”阶段。如此循环，直到 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 f"https://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(model="gpt-4-turbo-preview", temperature=0.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 "revision"

3.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 总迭代次数：2

4. 高级模式：基于 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(model="gpt-3.5-turbo", temperature=0) # 格式化消息 messages = router_prompt.format_messages( user_question=state["user_request"], collected_info=state.get("collected_info", "无"), current_step=state.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 自己说出“最终答案”外，必须设置硬性终止条件：
  1. 最大迭代次数：如我们例子中的iteration_count。
  2. 超时机制：在app.invoke时设置超时。
  3. 递归限制：编译图时使用app = workflow.compile(recursion_limit=100)。
• 设置检查点：对于长时间运行的工作流，考虑将中间状态持久化（如存入数据库）。这样即使进程中断，也能从上一个检查点恢复，而不是从头开始。

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 从“玩具”迈向“工具”，乃至成为“爆款产品”核心引擎的关键一步。记住，一个好的工作流设计，逻辑清晰胜过技巧堆砌。先从一个小而完整的循环开始，逐步增加复杂度和智能，你会更稳地走向成功。