
最近在AI Agent开发圈子里一个话题引发了广泛讨论Agent的工程范式是否正在从传统的“循环Loop”转向更复杂的“图Graph”伴随着这个讨论一些关于Anthropic内部文档的传言也甚嚣尘上。无论传言真假一个不争的事实是以LangGraph、Snap Graph Builder等为代表的图工程框架正在迅速崛起成为构建复杂、可编排AI Agent的新宠。对于开发者而言理解从Loop到Graph的演进不仅是跟上技术潮流更是解锁下一代智能应用能力的关键。本文将为你系统梳理AI Agent工程范式的演进之路深入对比Loop与Graph架构的核心差异与优劣并通过一个完整的LangGraph实战项目手把手教你如何构建一个具备记忆、工具调用和条件分支的智能Agent。无论你是刚接触Agent概念的新手还是正在寻找更优架构的进阶开发者都能从中获得清晰的认知和可直接复用的代码。1. 背景与核心概念为什么Agent工程范式在演进在深入技术细节之前我们首先要厘清几个核心概念什么是AI Agent传统的Loop工程存在哪些局限Graph工程又为何被寄予厚望AI Agent智能体通常指能够感知环境、自主决策并执行行动以实现目标的软件实体。在大语言模型LLM的加持下现代AI Agent的核心是一个“大脑”LLM它能够理解用户指令、规划步骤、调用工具如搜索、计算、写代码并持续迭代直至完成任务。传统的Loop工程循环工程是早期Agent实现的主流范式。其核心思想非常简单构建一个“感知-思考-行动”的循环。感知接收用户输入或环境状态。思考LLM根据当前信息和历史决定下一步做什么例如直接回答或调用某个工具。行动执行决策如生成文本或调用工具API。循环将行动结果作为新的输入回到“思考”步骤直到任务完成或满足终止条件。这种模式直观易懂易于实现一个简单的while循环配合if-else逻辑就能搭建起来。然而当任务复杂度上升时Loop模式的弊端凸显状态管理混乱所有历史对话、工具调用结果、中间状态都线性堆积难以结构化管理和提取关键信息。流程僵化循环是线性的难以处理需要并行执行、条件分支或者有复杂依赖关系的任务。可调试性差当Agent“卡住”或出错时很难定位是循环中的哪一步、哪个状态出了问题。难以实现“人类在环”在关键决策点引入人工审核或指导的机制在简单的循环中很难优雅地嵌入。新兴的Graph工程图工程正是为了解决这些痛点。它将Agent的执行流程抽象为一个有向图。节点Node代表一个执行单元可以是一个LLM调用、一个工具函数、一个条件判断甚至是一个人机交互接口。边Edge定义了节点之间的流转逻辑决定了执行完一个节点后下一步该去往哪个节点。边可以是有条件的根据上一个节点的输出结果动态选择路径。Graph模式的优势在于显式化工作流整个Agent的决策和执行流程被可视化为一张图一目了然。灵活编排轻松实现顺序、并行、分支、循环等复杂逻辑。结构化状态图有一个全局的、结构化的“状态”对象在不同节点间传递和修改状态管理变得清晰。易于调试与监控可以跟踪执行路径查看每个节点的输入输出快速定位问题。原生支持Human-in-the-loop可以专门设计一个“人工审核”节点当图执行到该节点时暂停等待人工输入后再继续。所谓的“从Loop走向Graph”本质是从一种隐式、线性的控制流转向一种显式、可编排、可视化的控制流。这标志着AI Agent工程从“能用”走向“好用”、“可靠”和“可维护”。2. 环境准备与版本说明接下来我们将通过一个实战项目来感受Graph工程的魅力。我们将使用LangGraph这是一个由LangChain团队推出的、专门用于构建多智能体工作流的框架。它基于Pydantic进行强类型状态管理设计理念非常清晰。环境要求操作系统Windows 10/11, macOS, 或 Linux (本文示例在macOS/Linux环境下测试)。Python版本 3.8 (推荐3.9或3.10)。主要依赖langgraph: 核心图编排框架。langchain-openai: 用于接入OpenAI的LLM我们将使用GPT-3.5-turbo作为“大脑”。你也可以替换为其他LangChain支持的模型。python-dotenv: 管理环境变量安全存储API Key。项目初始化创建项目目录并进入mkdir langgraph-agent-demo cd langgraph-agent-demo创建虚拟环境推荐python -m venv venv source venv/bin/activate # Linux/macOS # venv\Scripts\activate # Windows安装依赖包pip install langgraph langchain-openai python-dotenv创建环境变量文件.env并填入你的OpenAI API KeyOPENAI_API_KEY你的-api-key-here重要切勿将API Key直接硬编码在代码中。.env文件应被加入.gitignore。版本说明本文基于以下主要库版本编写不同版本间API可能有细微差异请以官方文档为准。langgraph: 0.0.40langchain-openai: 0.0.5langchain-core: 0.1.0如果你的项目已有其他依赖请注意版本兼容性。3. 核心概念与LangGraph基础拆解在写代码前需要理解LangGraph中的几个核心抽象。3.1 状态State在LangGraph中图运行时维护一个全局的、共享的状态对象。这个状态通常是一个Pydantic模型定义了整个工作流中需要传递的所有数据。例如可能包含messages对话历史question用户问题intermediate_steps工具调用结果等字段。3.2 节点Node节点是一个函数它接收当前State作为输入对State进行修改或返回一个更新后的State并可选地指定下一个要执行的节点。节点是工作流中的“工作单元”。3.3 边Edge边决定了执行流的方向。在LangGraph中边通常通过条件函数或路由逻辑来实现。最常用的是conditional_edge它根据某个条件判断下一步该走哪条路。3.4 图Graph图是节点和边的集合。你创建图添加节点定义边然后将其编译成一个可执行的“运行时”。3.5 一个最简单的LangGraph心智模型想象一个客服Agent的工作流节点A接收问题将用户问题放入状态。节点B分类LLM判断问题属于“技术故障”还是“账户咨询”。条件边如果是“技术故障”前往节点C调用故障库工具如果是“账户咨询”前往节点D调用账户信息工具。节点C/D调用相应工具将结果写回状态。节点E生成回答LLM综合工具结果和问题生成最终回复给用户。这个流程用Loop写会充满复杂的if-else嵌套而用Graph表示则是一张清晰的可视化流程图。4. 实战构建一个具备工具调用与分支能力的Graph Agent我们的目标是构建一个“研究助手”Agent。它的功能是根据用户提出的复杂问题例如“特斯拉和比亚迪在电动汽车市场的竞争格局如何”能自动决定是否需要联网搜索并整合搜索到的信息生成一份结构化的简短报告。4.1 定义状态State首先我们定义工作流中需要流转的数据结构。在项目根目录创建state.py。# state.py from typing import List, Optional, Any, Dict from typing_extensions import TypedDict from langchain_core.messages import BaseMessage class AgentState(TypedDict): Agent工作流的全局状态定义。 # 用户输入的原始问题 input: str # 完整的对话历史消息列表 messages: List[BaseMessage] # 是否需要执行搜索由LLM判断 needs_search: Optional[bool] # 存储搜索工具返回的结果 search_results: Optional[List[Dict[str, Any]]] # 存储最终生成的答案 final_answer: Optional[str]我们使用TypedDict来定义状态这为代码提供了良好的类型提示。状态中包含输入、对话历史、决策标志、中间结果和最终输出。4.2 创建工具与LLM接下来我们创建Agent所需的“工具”和“大脑”。创建tools_and_llm.py。注意为了示例我们用一个模拟的搜索函数代替真实的网络搜索API。# tools_and_llm.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain.tools import tool from langchain_core.messages import HumanMessage, SystemMessage # 加载环境变量 load_dotenv() # 初始化LLM llm ChatOpenAI( modelgpt-3.5-turbo, temperature0, # 降低随机性使决策更稳定 api_keyos.getenv(OPENAI_API_KEY) ) # 定义一个模拟的搜索工具 tool def web_search_tool(query: str) - list: 模拟网络搜索工具。在实际应用中这里应替换为真实的SerpAPI、Google Search API等调用。 返回一个包含搜索结果字典的列表。 print(f[工具调用] 模拟搜索查询: {query}) # 这里是模拟数据 mock_results [ {title: 特斯拉2023年全球交付量报告, snippet: 特斯拉在2023年实现了..., source: mock-source-1}, {title: 比亚迪新能源汽车市场战略分析, snippet: 比亚迪通过垂直整合供应链..., source: mock-source-2}, {title: EV市场竞争特斯拉 vs 比亚迪, snippet: 两者在电池技术、自动驾驶和定价上展开激烈竞争..., source: mock-source-3}, ] return mock_results # 系统提示词用于引导Agent的行为 SYSTEM_PROMPT 你是一个专业的研究助手。你的任务是分析用户的问题并决定是否需要通过搜索来获取最新信息以更好地回答问题。 请严格遵循以下规则 1. 仔细分析用户的问题。 2. 如果问题涉及实时数据、最新事件、具体公司财报、技术对比等需要最新外部知识的内容则回答“需要搜索”。 3. 如果问题是一般性概念、定义、无需外部信息的逻辑推理等则回答“无需搜索”。 4. 只输出“需要搜索”或“无需搜索”不要添加任何其他解释。4.3 构建Graph节点现在我们来创建Graph中的各个节点。每个节点都是一个函数接收AgentState返回一个对State的更新。创建nodes.py。# nodes.py from .state import AgentState from .tools_and_llm import llm, web_search_tool, SYSTEM_PROMPT from langchain_core.messages import SystemMessage, HumanMessage, AIMessage def decide_search_node(state: AgentState) - AgentState: 节点1决策是否需要搜索。 print(--- 进入 [决策节点] ---) # 准备给LLM的消息 decision_messages [ SystemMessage(contentSYSTEM_PROMPT), HumanMessage(contentstate[input]) ] # 调用LLM进行决策 decision_response llm.invoke(decision_messages) decision_text decision_response.content.strip() # 解析决策 needs_search 需要搜索 in decision_text print(f决策结果: {decision_text} - needs_search{needs_search}) # 更新状态 new_state state.copy() new_state[needs_search] needs_search # 将这次交互也记录到历史中 new_state[messages] state[messages] [HumanMessage(contentstate[input]), AIMessage(contentdecision_text)] return new_state def search_node(state: AgentState) - AgentState: 节点2执行搜索。 if not state.get(needs_search): print(--- 跳过 [搜索节点] (无需搜索) ---) return state print(--- 进入 [搜索节点] ---) query state[input] try: results web_search_tool.invoke({query: query}) print(f搜索完成获得 {len(results)} 条结果。) except Exception as e: print(f搜索工具调用失败: {e}) results [] new_state state.copy() new_state[search_results] results return new_state def generate_answer_node(state: AgentState) - AgentState: 节点3生成最终答案。 print(--- 进入 [生成答案节点] ---) # 准备生成答案的上下文 context if state.get(needs_search) and state.get(search_results): context 以下是根据你的问题搜索到的相关信息\n for i, res in enumerate(state[search_results]): context f{i1}. {res[title]}: {res[snippet]}\n else: context 未进行网络搜索将基于模型已有知识回答。\n prompt f 请基于以下上下文和问题生成一个简洁、结构化的回答。 上下文 {context} 用户原始问题 {state[input]} 请生成回答 answer_messages [HumanMessage(contentprompt)] answer_response llm.invoke(answer_messages) final_answer answer_response.content new_state state.copy() new_state[final_answer] final_answer # 记录最终交互 new_state[messages] state[messages] [HumanMessage(contentprompt), AIMessage(contentfinal_answer)] return new_state4.4 定义图与边工作流这是Graph工程的核心将节点连接起来并定义流转逻辑。创建graph_definition.py。# graph_definition.py from langgraph.graph import StateGraph, END from .state import AgentState from .nodes import decide_search_node, search_node, generate_answer_node def route_after_decision(state: AgentState) - str: 条件路由函数根据‘是否需要搜索’决定下一步。 if state.get(needs_search): return search # 前往搜索节点 else: return generate_answer # 跳过搜索直接生成答案 # 1. 创建一个图并指定其状态结构 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(decide, decide_search_node) # 决策节点 workflow.add_node(search, search_node) # 搜索节点 workflow.add_node(generate_answer, generate_answer_node) # 答案生成节点 # 3. 设置入口点 workflow.set_entry_point(decide) # 4. 添加边定义节点间的流向 # 从决策节点出发根据条件路由到不同节点 workflow.add_conditional_edges( decide, route_after_decision, # 条件判断函数 { search: search, # 如果返回search则前往search节点 generate_answer: generate_answer # 如果返回generate_answer则前往generate_answer节点 } ) # 从搜索节点出来无条件前往答案生成节点 workflow.add_edge(search, generate_answer) # 答案生成节点是终点连接到END workflow.add_edge(generate_answer, END) # 5. 编译图生成可执行对象 app workflow.compile()4.5 主程序与运行验证最后我们创建一个主程序来运行这个Graph Agent。创建main.py。# main.py from graph_definition import app from state import AgentState def run_agent(question: str): 运行研究助手Agent。 print(f\n 开始处理问题 ) print(f用户问题: {question}) # 初始化状态 initial_state: AgentState { input: question, messages: [], needs_search: None, search_results: None, final_answer: None } # 执行图 print(\n[执行图工作流]) final_state app.invoke(initial_state) # 输出结果 print(\n 最终结果 ) print(final_state[final_answer]) print(\n) return final_state if __name__ __main__: # 测试用例1需要搜索的问题 question1 特斯拉和比亚迪在2023年的电动汽车销量对比如何 run_agent(question1) # 测试用例2无需搜索的问题 question2 请解释一下什么是机器学习 run_agent(question2)4.6 运行与结果分析在项目根目录下运行程序python main.py预期输出示例 开始处理问题 用户问题: 特斯拉和比亚迪在2023年的电动汽车销量对比如何 [执行图工作流] --- 进入 [决策节点] --- 决策结果: 需要搜索 - needs_searchTrue --- 进入 [搜索节点] --- [工具调用] 模拟搜索查询: 特斯拉和比亚迪在2023年的电动汽车销量对比如何 搜索完成获得 3 条结果。 --- 进入 [生成答案节点] --- 最终结果 根据搜索到的信息以下是特斯拉和比亚迪在2023年电动汽车销量的简要对比 **特斯拉 (Tesla):** - 2023年全球共交付约181万辆电动汽车。 - 主要增长动力来自Model Y和Model 3尤其是在中国和欧洲市场。 ... **比亚迪 (BYD):** - 2023年全年新能源汽车销量超过302万辆其中纯电动乘用车约占一半。 - 凭借强大的垂直整合供应链和丰富的产品矩阵从海豚到仰望在中国市场占据主导地位。 ... 开始处理问题 用户问题: 请解释一下什么是机器学习 [执行图工作流] --- 进入 [决策节点] --- 决策结果: 无需搜索 - needs_searchFalse --- 跳过 [搜索节点] (无需搜索) --- --- 进入 [生成答案节点] --- 最终结果 机器学习是人工智能的一个核心分支它致力于研究如何让计算机系统无需显式编程就能通过经验数据自动改进其性能。 ... 从输出可以清晰看到Graph的执行路径对于问题1它走了decide - search - generate_answer路径对于问题2它走了decide - generate_answer路径跳过了搜索节点。这正是Graph编排强大之处的直观体现。5. 常见问题与排查思路在构建和运行LangGraph Agent时你可能会遇到以下典型问题。问题现象可能原因排查思路与解决方案ModuleNotFoundError: No module named ‘langgraph’依赖未正确安装。1. 确认虚拟环境已激活。2. 运行pip list | grep langgraph检查是否安装。3. 重新安装pip install langgraph langchain-openai。openai.AuthenticationErrorOpenAI API Key 错误或未设置。1. 检查.env文件是否存在格式是否正确OPENAI_API_KEYsk-...。2. 在代码中打印os.getenv(“OPENAI_API_KEY”)的前几位确认已加载。3. 确保Key有余额且未过期。图编译或执行时报类型错误状态State的结构与节点输入输出不匹配。1. 检查AgentState的TypedDict定义确保所有节点访问的键都已声明。2. 确保每个节点函数都接收并返回AgentState。3. 使用print(state.keys())在节点内调试状态内容。Agent始终选择“需要搜索”或“无需搜索”系统提示词SYSTEM_PROMPT不够精确或LLM温度参数过高。1. 优化SYSTEM_PROMPT给出更具体、更清晰的判断示例。2. 将LLM的temperature参数设为0减少随机性。3. 在decide_search_node中打印出LLM的完整响应分析其推理过程。工具调用失败或超时网络问题、工具API变更、模拟工具函数有bug。1. 如果是真实API检查网络连接和API密钥权限。2. 在工具函数内部添加try-except和详细日志。3. 对于模拟工具确保返回的数据结构符合预期例如是list类型。执行流未按预期进行如该搜索却没搜条件边conditional_edge的路由函数逻辑有误或状态更新不正确。1. 仔细检查route_after_decision函数确保其返回值与add_conditional_edges中定义的映射键完全一致。2. 在决策节点后打印state[‘needs_search’]的值确认状态已被正确更新。‘Graph’ object has no attribute ‘invoke’可能使用了旧版本的LangGraph API或者图对象未正确编译。1. 确认使用的是StateGraph和workflow.compile()来创建可调用对象app。2. 检查LangGraph版本本文示例基于较新的稳定API。6. 最佳实践与工程建议将Graph工程应用于生产环境或复杂项目时遵循以下最佳实践可以避免很多坑。6.1 状态设计要精简且强类型只存储必要的状态状态是全局共享的避免在其中存储过大的中间数据如整个网页内容。可以存储摘要或引用ID。使用Pydantic模型虽然示例用了TypedDict但对于更复杂的验证推荐使用Pydantic的BaseModel来定义State可以利用其数据验证和序列化能力。区分对话历史与内部状态像messages这样的对话历史最好单独管理与Agent的内部控制状态如needs_search分离使结构更清晰。6.2 节点设计遵循单一职责每个节点只做一件事。例如一个节点专门做“决策”一个节点专门做“搜索”一个节点专门做“格式化输出”。这提高了代码的可测试性和可复用性。节点函数应保持纯净尽量不产生副作用如直接修改外部数据库。如需副作用应通过状态传递或显式调用服务。6.3 充分利用可视化与调试工具LangGraph的一个巨大优势是可视化。使用workflow.get_graph().draw_mermaid()可以将你的图导出为Mermaid图表直观地审视整个工作流。在开发时大量使用print日志或集成logging模块记录每个节点的输入/输出和状态变化这对于调试复杂流程至关重要。6.4 实现稳健的错误处理与回退在图级别或节点级别设置异常处理。LangGraph支持“中断”和“检查点”机制可以在节点失败时暂停流程并允许人工干预或重试。为关键节点如LLM调用、外部API调用设计重试逻辑和超时机制。在条件边中始终考虑“默认路径”避免因为意外状态导致图执行卡死。6.5 性能与成本优化缓存对于频繁调用且结果不变的LLM请求或工具调用例如对相同问题的决策可以考虑引入缓存机制。异步执行如果节点间没有严格的先后依赖可以考虑使用LangGraph的异步支持来并行执行缩短整体响应时间。LLM调用优化精心设计提示词使用函数调用Tool Calling让LLM返回结构化JSON便于程序解析减少不必要的大段文本生成。6.6 向更复杂的模式演进子图对于特别复杂的节点可以将其本身实现为一个子图实现层次化、模块化的设计。Human-in-the-loop在图中的关键决策点插入“人工审核节点”。该节点可以暂停图执行通过一个回调如发送邮件、生成工单通知真人待真人输入后再恢复图执行。这是构建可靠生产级Agent的关键。多Agent协作你可以定义多个具有不同专长的Agent如“研究员”、“写手”、“校对员”并用一个主图来编排它们之间的协作实现更复杂的任务分解与解决。从简单的循环到有向图AI Agent的工程范式升级本质上是应对复杂性增长的必然选择。Graph提供的显式编排、清晰状态和可视化能力使得开发、调试和维护复杂Agent工作流变得可行。通过本文的实战你已经掌握了使用LangGraph构建Graph Agent的基本技能。接下来可以尝试将模拟搜索工具替换为真实的SerpAPI或 Tavily Search API引入更多工具如计算器、代码执行器或者设计一个包含人工审核节点的客服流程图。