Agent、Loop、Graph三段递进：从单次调用到复杂编排的智能体开发路线

这次我们直接聊一个很多人学 Agent 开发时都会卡住的问题Agent、Loop、Graph这三个词到底什么关系网上教程各讲各的有人教你写工具调用有人讲 ReAct 循环还有人一上来就上 LangGraph 状态图看完更晕。如果你正处于这个阶段这篇文章就是一份“三段递进地图”。我把 Agent 开发拆成三个递进层次先理解单体 Agent再理解让 Agent 反复思考的 Loop最后理解把多个 Loop 和分支编排成图的 Graph。每一层解决什么问题、代码长什么样、什么时候该从上一层走到下一层一次说清楚。阅读完你至少能拿到两样东西一张清晰的 Agent 开发学习路线图以及一套从最小示例到带循环、带人工审批节点的可落地代码模板。1. Agent、Loop、Graph 三段递进地图速览先给全局视角。这三个词不是并列关系而是递进关系理解成三个工程阶段更合适。概念一句话定位核心解决的问题典型实现复杂度Agent能调用工具、自主完成任务的智能体让模型不止会聊天还能行动OpenAI Function Calling、Anthropic Tool Use低Loop让 Agent 反复执行的循环机制让模型能多轮推理、反思、纠错ReAct 循环、Agent Loop中Graph把多个节点和循环编排成状态图把复杂任务拆成节点实现可控、可回溯、可人工介入LangGraph、自定义状态机高依赖关系也很明确Graph 里面可以包含 LoopLoop 里面要运行 Agent。学习路线就是从内往外打先把单次 Agent 调用跑通再套上循环再升级成图。2. 第一段Agent——单次推理的最小单元2.1 Agent 解决什么问题很多初学者把 Agent 理解成“会聊天的机器人”这个理解不够工程化。从代码角度看Agent 的本质是把大模型的推理能力和外部工具结合起来完成一个具体动作。比如一个能查天气的 Agent它的核心逻辑是用户输入“北京明天天气怎么样”。模型判断需要调用天气查询工具。程序执行工具拿到天气数据。模型把数据整理成自然语言回复用户。这里面“调用哪个工具、传什么参数”的决策由模型完成而“实际执行工具”的动作由程序完成。Agent 就是这层决策与执行的封装。2.2 一个最小 Agent 的结构先不引入任何框架看最朴素的实现思路。假设你接了一个支持工具调用的模型接口典型的请求流程如下import json import openai client openai.OpenAI() def get_weather(city: str) - str: 模拟天气查询工具 return f{city} 今天晴气温 22 度。 tools [ { type: function, function: { name: get_weather, description: 查询指定城市的天气, parameters: { type: object, properties: { city: {type: string, description: 城市名} }, required: [city] } } } ] def run_agent_once(user_input: str) - str: messages [{role: user, content: user_input}] response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolstools ) message response.choices[0].message if message.tool_calls: for tool_call in message.tool_calls: args json.loads(tool_call.function.arguments) if tool_call.function.name get_weather: result get_weather(args[city]) messages.append({ role: tool, tool_call_id: tool_call.id, content: result }) response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolstools ) return response.choices[0].message.content return message.content注意这个示例中我调用了两次模型接口第一次让模型决策是否调用工具第二次把工具结果喂回模型生成最终回答。这就是一次完整的 Agent 调用。从例子可以看出Agent 本身是“单轮决策”。用户问一句Agent 执行一轮工具调用给一个答案。如果任务简单到这里就结束了。3. 第二段Loop——让 Agent 反复工作的循环3.1 什么是 Agent Loop单次 Agent 能力有限。考虑一个任务用户问“帮我分析这份财报里的风险并整理成表格”。Agent 需要先读取文件再总结内容再判断是否漏了数据可能还要再读一次文件。这一系列动作无法靠一次模型调用完成。这里就需要 Loop循环。Agent Loop 的思路是让 Agent 在一个循环里反复执行“思考 - 行动 - 观察”的过程直到某个退出条件满足。这个模式在学术界叫 ReActReasoning Acting在工程界常被称为 Agent Loop 或 Agent Runner。核心流程如下while 未达到退出条件: 模型生成下一步动作 如果是工具调用: 执行工具 把结果拼回上下文 如果是最终回答: 跳出循环3.2 ReAct 循环的代码骨架下面是一个不依赖框架的 ReAct 循环骨架只保留了核心控制流import json from openai import OpenAI client OpenAI() TOOLS { get_weather: lambda city: f{city} 今天晴22 度, calculate: lambda expr: str(eval(expr)), # 仅用于教学示例生产环境不要用 eval } TOOL_SCHEMAS [ { type: function, function: { name: get_weather, description: 查询城市天气, parameters: { type: object, properties: { city: {type: string} }, required: [city] } } }, { type: function, function: { name: calculate, description: 计算数学表达式, parameters: { type: object, properties: { expr: {type: string} }, required: [expr] } } } ] def run_agent_loop(user_input: str, max_iters: int 5) - str: messages [{role: user, content: user_input}] for step in range(max_iters): response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolsTOOL_SCHEMAS ) message response.choices[0].message messages.append(message) if not message.tool_calls: return message.content for tool_call in message.tool_calls: name tool_call.function.name args json.loads(tool_call.function.arguments) result TOOLS[name](**args) messages.append({ role: tool, tool_call_id: tool_call.id, content: result }) return 达到最大迭代次数停止循环。这个循环的关键点有两个上下文拼接每次把模型回复、工具结果都追加到 messages 里模型才能“看到”自己之前做了什么。退出条件要么模型不再请求工具调用、直接给最终回答要么触达 max_iters 上限。从这里可以看出Loop 解决的是“多轮协作”问题。Agent 不再是问一句答一句而是可以连续调用多个工具、根据中间结果调整下一步。3.3 循环什么时候该停Loop 不是无限转圈工程上必须设计退出条件否则会产生大量无效调用既费钱又费时间。常见的停止策略有停止方式说明适用场景模型主动停止模型不再返回 tool_calls给出 final answer大多数 ReAct 任务最大轮次限制设置 max_iters防止死循环所有生产场景必须设置人工中断长时间任务需要人工确认后再继续敏感操作、高风险动作错误熔断工具连续报错 N 次后停止接口不稳定时到这里第二阶段就清楚了Agent 负责决策Loop 负责让决策持续发生。但 Loop 也有问题它是一个单线循环任务一旦分叉比如“先查天气再规划路线天气不好就改室内方案”用 while 循环写会越来越乱。这就引出第三段 Graph。4. 第三段Graph——把多个 Loop 和分支编排成图4.1 为什么需要 Graph把复杂的 Agent 任务都写在一个 while 循环里会有什么问题逻辑分叉难以表达。天气好走 A 分支天气差走 B 分支用 if 嵌套在循环里代码会越来越难看。循环嵌套循环时难以追踪状态。无法可视化整个任务流程排查问题全靠读代码。无法在任务中间插入人工审批节点。Graph 的思路是把任务抽象成一张有向图节点是处理步骤边是状态转移条件。Agent 在节点上执行Loop 在环上发生整个任务流程变成一张可读、可扩展的图。这正是近期 Agent 框架纷纷引入 Graph 层的原因。LangGraph 就是这一类实现的代表Spring AI Alibaba 的后续版本也开始强调 graph 编排能力。4.2 节点、边、状态Graph 三要素状态State贯穿整个流程的数据结构所有节点都能读、能改。节点Node一个处理单元接收状态、处理、返回更新后的状态。边Edge决定下一步走哪个节点可以带条件判断。一个典型的 Agent Graph 流程可能是开始 - 规划节点 - 执行工具节点 - 判断节点 | 条件不满足 - 返回执行工具节点形成环 条件满足 - 生成最终答案 - 结束4.3 human-in-the-loop 在 Graph 里的位置human-in-the-loop 是 Agent 开发中经常被提到的机制也是 Graph 比 Loop 更好落地它的原因。在一个单纯的 Loop 中想让循环暂停、等人工确认需要自己维护一个外部标志位还要处理并发问题。在 Graph 中人工节点本身就是图里的一个节点流程走到该节点时暂停收到人工确认后再继续。下面是一个示意from typing import TypedDict from langgraph.graph import StateGraph, END class AgentState(TypedDict): task: str plan: str approved: bool result: str def plan_node(state: AgentState) - dict: # 模拟模型生成计划 return {plan: step1 - step2 - step3} def execute_node(state: AgentState) - dict: # 模拟执行计划 return {result: executed state[plan]} def human_approve_node(state: AgentState) - dict: # 生产环境这里会阻塞等待人工审批结果 # 这里用输入简化实际需要对接审批界面或接口 approved input(是否批准该计划输入 yes 继续: ) return {approved: approved yes} def after_approve_node(state: AgentState) - dict: return {result: state[result] , approved by human} def should_continue(state: AgentState) - str: if state[approved]: return execute return replan def build_graph(): graph StateGraph(AgentState) graph.add_node(plan, plan_node) graph.add_node(execute, execute_node) graph.add_node(human_approve, human_approve_node) graph.add_node(after_approve, after_approve_node) graph.set_entry_point(plan) graph.add_edge(plan, human_approve) graph.add_conditional_edges( human_approve, should_continue, {execute: execute, replan: plan} ) graph.add_edge(execute, after_approve) graph.add_edge(after_approve, END) return graph.compile()这是一个简化示例实际项目中 human_approve_node 会对接审批接口而不是阻塞在控制台。重点在于Graph 把流程控制权从代码的嵌套逻辑中解放出来变成数据驱动的状态流转。5. 三段递进地图Agent 开发学习路线把这三点串起来就是一张 Agent 开发的“三段递进地图”阶段学习目标推荐练习下一步第一阶段Agent理解工具调用、理解模型如何决策写一个带天气查询工具的最小 Agent给 Agent 套循环第二阶段Loop理解 ReAct 循环、理解上下文拼接写一个能连续调用两个工具的循环引入图编排第三阶段Graph理解状态、节点、边、分支、人工介入用 LangGraph 重写上面的循环落地到真实业务这同一条路线也可以理解为 Agent 开发的复杂度和工程化程度递进。如果是个人小工具做到 Loop 可能就够用了如果是生产级多步骤任务Graph 几乎是必须的因为可维护性和可观测性差别很大。6. 环境准备与前置条件下面进入实操环节。如果你要完整跑通本文代码需要准备以下环境环境项要求说明Python3.10 及以上Agent 开发主流版本模型接口OpenAI 兼容接口也可以是本地部署的模型服务只要支持 tools 参数依赖库openai、langgraph按实际框架安装操作系统Windows / macOS / Linux 均可本文代码均为跨平台硬件CPU 即可运行GPU 影响推理速度框架本身不是重计算组件安装依赖pip install openai langgraph如果使用本地模型服务比如通过 vLLM 或 Ollama 暴露一个 OpenAI 兼容的 API只需要修改 OpenAI client 的 base_urlclient OpenAI( base_urlhttp://127.0.0.1:8000/v1, # 以本地服务实际地址为准 api_keyEMPTY # 本地服务一般不需要真实 key )这里注意模型是否支持 tools 参数取决于模型本身。本地小模型对工具调用的支持不稳定建议先确认模型文档里写了支持 function calling再开始测试。7. 从 Agent 到 Loop 的本地实操7.1 最小 Agent 测试先运行第 2 节的最小 Agent 示例。输入一个需要调用工具的问题比如“北京今天天气怎么样”。观察预期输出程序先调用一次模型接口返回 tool_calls。程序执行本地天气函数。程序把工具结果拼回 messages。程序再调用模型接口返回最终答案。这里重点观察模型是否识别出“需要调用工具”以及返回的工具参数是否正确。失败时优先检查模型是否支持 tools 参数。tool schema 是否写规范。本地服务时工具调用格式是否与 OpenAI 兼容。7.2 加一层 Loop 变成 ReAct再运行第 3 节的 ReAct 循环输入一个需要多步计算的复合问题例如“计算 12 * 8然后加上 4最后告诉我结果”。这个任务至少需要两次工具调用。Loop 的代码会自动完成模型: 调用 calculate(12 * 8) 工具: 96 模型: 调用 calculate(96 4) 工具: 100 模型: 最终答案是 100这是 Loop 和单次 Agent 最直观的差异单次 Agent 只处理一次工具调用Loop 会持续处理到任务完成。注意示例里用了 eval 执行表达式这仅用于演示生产环境绝对不要直接用 eval换成 ast 解析或改用外部计算服务。7.3 验证观察要点跑 Loop 时建议重点观察以下指标观察项关注点模型调用次数是否符合预期是否出现无意义重复调用上下文长度每轮调用都会追加消息任务越长token 消耗越大工具结果质量工具报错时模型是否能正确理解并重试退出时机模型是准确判断“任务完成”还是永远在调用工具如果模型一直不退出先检查 max_iters 是否设置再检查工具 description 是否写得足够清晰。很多循环不可控的问题根源不是模型能力而是工具描述模糊。8. 从 Loop 到 Graph 的编排实操8.1 用 LangGraph 实现一个简单状态图LangGraph 是目前实践 Agent Graph 比较主流的方案。它的核心思想就是用 StateGraph 构建一张图然后调用 compile 得到可执行的 app。一个最简单的执行流程如下from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END class State(TypedDict): query: str count: int def start_node(state: State) - dict: print(--- start ---) return {count: state[count] 1} def finish_node(state: State) - dict: print(--- finish ---) return {query: state[query]} def build(): graph StateGraph(State) graph.add_node(start, start_node) graph.add_node(finish, finish_node) graph.set_entry_point(start) graph.add_edge(start, finish) graph.add_edge(finish, END) return graph.compile() app build() result app.invoke({query: test, count: 0}) print(result)这段代码展示了最基础的 Graph 流程节点、边、状态传递。注意 State 里定义的字段节点函数返回的 dict 会更新到全局状态中。8.2 在 Graph 里实现循环Graph 里的循环本质是让某个节点通过条件边指向它之前的节点。还是那句Loop 是包含在 Graph 内部的。一个带重试循环的图如下from typing import TypedDict from langgraph.graph import StateGraph, END class RetryState(TypedDict): attempt: int success: bool def do_work(state: RetryState) - dict: attempt state[attempt] 1 print(fattempt {attempt}) # 模拟失败条件实际按任务结果判断 success attempt 3 return {attempt: attempt, success: success} def route_after_work(state: RetryState) - str: if state[success]: return end return retry def build_retry_graph(): graph StateGraph(RetryState) graph.add_node(do_work, do_work) graph.set_entry_point(do_work) graph.add_conditional_edges( do_work, route_after_work, {retry: do_work, end: END} ) return graph.compile() app build_retry_graph() result app.invoke({attempt: 0, success: False}) print(final:, result)这里graph.add_conditional_edges是循环的关键根据状态决定下一步是回到do_work还是结束。8.3 人工审批节点回到第 4.3 节的 human-in-the-loop 示例。在 LangGraph 中标准做法是把节点设计成阻塞等待人工确认。生产环境通常会配合一个存储层把任务挂起前端审批通过后再继续执行。实现思路参考# 伪代码用于理解流程 class TaskStore: def save(self, task_id, status, state): ... def load(self, task_id): ... def update(self, task_id, status, state): ... def human_approve_node(state): task_id state[task_id] # 把任务状态存储为 pending_review task_store.save(task_id, pending_review, state) # 阻塞等待审批接口回调或轮询数据库状态 while True: task task_store.load(task_id) if task[status] approved: return {approved: True} if task[status] rejected: return {approved: False} time.sleep(1)不同版本的 LangGraph 也提供了 interrupts 机制具体 API 以官方文档为准。核心思想不变人工节点是 Graph 里的一个节点状态流转可以被暂停、恢复。9. 接口 API 与批量任务接入9.1 把 Graph 封装成 API 服务不管你的 Agent 是基于 Loop 还是 Graph落地到业务里通常要封装成 API。常见的做法是使用 FastAPI 把编排逻辑包起来from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class TaskRequest(BaseModel): task: str user_id: str class TaskResponse(BaseModel): task_id: str status: str app.post(/agent/task, response_modelTaskResponse) def create_task(req: TaskRequest): task_id ftask_{req.user_id}_{int(time.time())} # 提交到后台任务队列异步执行 return TaskResponse(task_idtask_id, statuspending) app.get(/agent/task/{task_id}) def get_task(task_id: str): # 查询任务状态与结果 return {task_id: task_id, status: done, result: ...}这里要注意Agent 任务一般比较慢尤其涉及多次模型调用不适合同步等结果。API 设计建议采用“提交任务 轮询结果”或 WebSocket 推送。9.2 批量任务设计批量跑 Agent 任务时不建议无脑并发请求模型接口。原因模型服务有速率限制并发过高会报错。多个任务共享同一个上下文逻辑时状态容易串。出错后需要重试批量框架要自备队列和失败处理。一个简单的批量任务结构可以用目录表示inputs/ task_001.json task_002.json outputs/ task_001.json task_002.json failed/ task_003.json每个任务文件里保存输入参数跑完写结果失败单独记录重试。配合消息队列比如 Redis、RabbitMQ就能做成生产级批量任务系统。10. 性能与资源占用观察Agent 类应用和重计算应用不同它的瓶颈通常不在 GPU 显存而在模型推理延迟和 API 调用次数。需要重点观察四个指标指标说明优化方向每任务模型调用次数Loop 越多调用越多成本越高优化工具描述减少无效循环端到端延迟用户提交任务到拿到结果的耗时并行工具调用、流式输出上下文 token 消耗每次循环都追加消息长任务消耗很大定期压缩摘要、滑动窗口裁剪失败重试次数工具不稳定时循环会反复重试设置最大重试次数快速熔断如果你用本地模型部署 Agent显存占用以模型本身的推理显存为主。7B 模型通常需要约 6-8G 显存70B 模型需要多张卡具体以实际模型版本为准。Graph 编排层本身基本不占显存它消耗的是 CPU 内存和任务状态存储。降低资源消耗的方法有给模型调用加缓存相同参数不下发重复请求。工具结果做摘要避免长文本反复传回模型。在 Loop 里限制最大迭代次数。条件允许时多个独立工具调用合并成一次并行请求。11. 常见问题与排查方法问题现象可能原因排查方式解决方案模型不调用工具模型不支持 tools或工具描述不清查看模型文档测试一个最简单的工具调用换支持 function calling 的模型或重写工具描述工具调用参数格式错误schema 定义和模型返回不一致打印模型返回的原始 tool_calls严格按 OpenAI 格式定义参数不要省略类型Loop 永远不退出缺少最大轮次限制检查循环代码是否有 max_iters强制设置轮次上限上下文越来越长导致超限循环中消息无限追加观察 messages 长度做消息裁剪或摘要压缩任务进程卡死人工审批节点没有恢复机制检查审批回调是否触发加入超时返回默认值或做断线恢复并发批量任务频繁报错触达模型服务 rate limit查看后端返回状态码 429加限流、退避重试条件边不生效路由函数返回值和映射表 key 不匹配打印路由函数返回值检查 conditional_edges 的映射字典本地模型工具结果不稳定小模型对结构输出敏感多次测试观察失败模式降低任务复杂度或换用更大参数模型12. 最佳实践与使用建议结合三段递进地图给出实际工程建议。优先跑通最小闭环。不要一上来就搭 Graph。先把单体 Agent 在目标模型上跑通确认工具调用稳定再加循环。很多人卡了很久最后发现是模型不支持 tools 参数属于前置条件没满足。每层都要设护栏。Agent 层给 timeoutLoop 层给 max_itersGraph 层给人工恢复机制。层层设防任务才不会失控。保持状态可观测。不管是手动维护 messages还是用 LangGraph 维护 State都要打印每个节点的输入输出。Agent 任务时延高失败排查困难日志是你最重要的调试工具。工具描述写清楚。模型决策依赖工具描述。同一个工具描述模糊和描述精确循环次数可能差很多。建议把参数示例写进描述里。控制工具权限边界。Agent 能调用工具意味着它有执行能力。生产环境要限制工具作用范围不允许任意命令执行、不允许删除文件、不允许无授权拉取敏感数据。至少要做一个“操作确认”节点高风险动作进入人工审批。注意数据合规。涉及用户隐私和版权素材时必须先确认授权再喂给模型。如果调用云端模型注意数据出境和隐私政策如果涉及人脸、声音等个人信息处理必须获得合法授权。保留一套最小可运行配置。把模型、工具、循环次数、超时时间写成配置文件不要散落在代码里。这样出问题时可以快速切换模型或参数定位。13. 总结与下一步Agent、Loop、Graph 是一条递进路线不是三个割裂概念。单体 Agent 解决单次决策Loop 解决多轮迭代Graph 解决复杂任务的编排与人工介入。如果你正在学习 Agent 开发建议按这条路线推进用 OpenAI 兼容接口写一个最小 Agent确认模型能调用工具。把单次调用改造成 while 循环做一个能连续调用多个工具的 ReAct Agent。把循环改造成 LangGraph 状态图加上分支和人工审批节点。封装成 API接入批量任务队列再上生产。最容易踩的坑是跨层跳级还没搞清楚工具调用就直接学框架结果分不清报错来自模型、循环还是框架本身。按顺序打通每一步踩过的坑都会成为下一层的基础。下一步建议研究这三个方向长任务内存管理、多 Agent 协作编排、工具调用的流式输出。这三块都是基于“三段递进地图”往生产级能力延伸的必经环节。