1. 传统 Agent 的 while 循环为什么大家都在写却都在踩坑如果你翻开任何一个入门级的 Agent 项目大概率会看到一段熟悉的代码while not task_completed:里面是调用大模型生成下一步 → 执行工具 → 检查结果 → 继续循环。这种写法很直观也确实能跑通简单场景但很多人没意识到这个while循环正是 Agent 后来所有痛苦的根源。我最早也是这么写的直到一次线上事故彻底让我决定拆掉它。那次事故说起来也不复杂一个用户把任务设成了持续收集信息直到你认为足够为止我的 Agent 就真的在循环里来回调用工具把几千块 token 烧光了而且因为它是循环结构外部根本没法安全地中断它。事后我翻代码发现整个执行过程就是个没人看管的while没有超时、没有步数上限、没有状态回溯只要大模型一句Yes, continue就能永远跑下去。也就是从那一刻开始我意识到这种循环驱动的 Agent 架构本质上把控制权全交给了大模型的随机性。1.1 循环架构的本质与典型实现不夸张地说市面上 80% 的 Agent 框架的核心都是一个循环你给 Agent 一个目标它反复经历思考、行动、观察这个过程直到它认为自己完成了。在代码层面这就是一个典型的命令式循环messages [{role: user, content: user_task}] while True: response llm(messages) action parse_action(response) result execute_tool(action) messages.append({role: assistant, content: response}) messages.append({role: tool, content: result}) if done in response: break这种写法好在哪里好在一个循环就能表达不断推进的概念正好贴合了大模型一轮一轮生成 token 的习惯。每轮调用大模型把工具结果塞回上下文让模型看到新信息后继续决策天然地形成反馈回路。对 demo 和小工具来说这就是最快能跑起来的方案。但它是能用不是好用。循环里藏着两个隐含假设第一大模型每一轮都会做出正确决策第二这个过程一定会在有限步内收敛。真实世界这两个假设几乎都不成立尤其是任务越复杂模型就越容易在某个分支里反复试探表现出来就是来回调用同一个工具、输出几乎一样的文本完全陷入空转。1.2 循环架构的四个隐藏问题第一个问题是** token 浪费与成本失控**。while循环每转一圈就要把历史消息全部重新发送给大模型上下文窗口会持续膨胀。哪怕任务本身很简单只要模型多犹豫几次开销就会指数级上升。我记得有一次只是让 Agent 整理一份表格它硬是因为提取格式不对反复重试了 12 次最后我一看账单光那一次任务的 token 成本就能买好几杯咖啡。第二个问题是死循环问题极难排查。while循环没有步数上限这种天然的保护机制一旦模型连续几次输出同样的内容程序就会无限空转而且后台日志几乎看不出异常——因为每一轮都是正常的LLM 调用和工具执行只是没有进展。你只能靠外部手动杀掉进程事后分析这一大坨日志更是痛苦。第三个问题是不可观测、不可控制。你想暂停一个运行中的 Agent 吗想在某一轮之后手动介入修改计划吗循环架构下这些操作要么完全做不到要么得靠全局变量做一堆丑陋的if条件判断。外部用户发出一个停止请求你甚至不知道这个请求要插到循环的哪个位置才有效更别谈对单个步骤做回放和调试。第四个问题是并发能力天然受限。如果你想让多个任务并行处理或者一个任务里同时执行多个独立子任务while循环真的无能为力。它天生是一条线走到黑的结构分支只能靠模型在回复里虚拟地提一嘴然后循环再挨个去做完全没有真正的并行可言。我见过不少团队在这个阶段硬上架构最后只能靠把整个 Agent 复制 N 份来硬扛并发既浪费资源又难以同步状态。所以问题的关键不是我讨厌while这个语法而是循环结构把任务的推进逻辑和大模型的随机决策强耦合在了一起。你没法干预推进逻辑就只能被模型带着走。2. 拆掉循环状态机与图编排的核心思路既然循环这么让人头疼那把它拆掉换成什么我的答案很简单换成显式的状态流转也就是用状态机或图DAG来驱动 Agent 的每一步。所谓拆掉 while 循环不是说代码里不能出现for或while而是把循环这个隐形的控制流变成一套可看见、可干涉、可分析的执行图。这个转变的本质是把 Agent 的执行从一个大黑盒反复问大模型变成一系列明确的节点按照明确的边依次执行。每个节点只做一件事比如调用一次大模型生成计划、执行一次文件搜索、校验上一阶段的输出格式。每个节点执行完后根据它的输出决定下一步走向这个走向要么写死在代码里要么由模型在受限选项里选择而不是让它自由发挥。这样做的好处立竿见影你终于可以在每一步之间插入你自己的业务逻辑可以在某个节点执行前做权限检查可以在某个节点输出不符合预期时直接短路让整个任务失败而不是继续空转。循环架构里这一切都得靠在循环体内加一堆条件判断来实现往往搅成一团图架构里这些逻辑天然就是节点和边的属性。2.1 从循环问到状态流转我一开始接触这个思路时也觉得有点抽象这不就是把while换成了switch吗后来我真正动手写才明白区别在哪。在循环架构里Agent 的下一步完全由大模型说了算它想干什么你不知道它说要干嘛你只能照做而在状态流转架构里Agent 的下一步是预先定义的有限集合大模型只能在集合里挑甚至很多时候根本不需要大模型来决定——比如解析结果失败就重试这条边就是纯代码逻辑跟模型没半毛钱关系。举一个很具体的例子你要做一个调研市场并写报告的 Agent。循环架构下你会让模型自己决定先去哪查、查完怎么汇总最后什么时候写报告状态流转架构下你预先定义五个节点明确目标、搜索引擎检索、抓取网页内容、内容分析、生成报告然后定义边——目标不明确就回到第一个节点检索结果为空就换一组关键词再检索最多重试三次。大模型在每个节点里只负责生成当前步骤的文本比如在抓取网页内容节点里它只负责提取正文的核心要点不负责决定要不要再搜一轮。听起来好像限制了模型的能力对吧但我实测发现限制反而提升了稳定性。因为大模型在开放自由环境下做计划的能力远不如在明确边界内做单步动作的能力。你让它决定下一步搜索什么关键词它大概率会给你一个宽泛甚至跑偏的词但你在节点里给它基于刚才的分析提炼三个更精确的搜索词这个明确指令时它的表现会稳定很多。这就是把控制权从模型幻觉手里夺回来交还给确定性的代码逻辑。2.2 为什么状态机比 while 更适合 Agent我认为有四个核心理由每个都能直击循环架构的命门。第一是可暂停可恢复。状态机的每一步都落在节点上节点之间的状态是可以持久化的。这意味着你可以把整个 Agent 的执行进度存到数据库里进程重启后能从上次断掉的节点继续跑。循环架构下你没法做到这一点因为那个记录执行到哪一步的变量只是 Python 内存里的一行代码重启即消失。对需要跑几分钟甚至几小时的重型 Agent 来说可恢复就是刚需谁也不想任务执行到一半因为一台服务器重启就从头再来。第二是可并行可合并。图结构天然支持多个节点同时执行。比如搜索资料这个节点可以拆成三个搜索节点分别搜不同平台它们之间没有依赖关系就能并发跑。执行完后再合并到一个汇总节点。这在 while 架构里几乎不可能实现你只能串行一个个搜完再手动合并。我后面会专门讲并发这块这里先提一句状态机架构才是真正能用分布式能力来扛 Agent 并发的基础。第三是可回退可重试。你可以在某条边上定义失败动作比如这个节点失败就回退到上一个节点修改参数重试。这比循环架构里的反复问模型下一步怎么办要可靠得多因为回退的路径是程序员预先定义的不是模型临场发挥的。模型只需要负责在当前节点把事做好不需要理解整个任务的前因后果。第四是可观测可调试。有了明确的节点和边你就可以给每个节点打日志、设断点、统计单独耗时和 token 消耗。任务卡住了你可以直接看到卡在哪个节点上、为什么卡住而不是对着while里一行行print猜。这个优势在复杂任务上简直是降维打击——我后来排查线上问题基本是打开执行图看一眼就能定位省下来的时间不是一点半点。3. 实操落地怎么把循环拆成节点 边理论讲得再漂亮不如直接跑一个最小实现。下面我会带着你从头搭一个无 while 循环的 Agent 执行引擎核心只有三个概念节点Node、边Edge和执行器Executor。我不依赖任何重量级框架用纯 Python你甚至可以直接把这个实现复制到自己的项目里改造。我的设计原则很简单每个节点就是一个可执行单元它接收上下文产出结果然后返回下一步该去哪个节点。整个执行过程中不允许出现等一下问模型下一步干啥这种无限等待所有跳转都由边的规则决定。至于那些必须反复迭代的流程比如搜索结果不够好需要换个关键词再试我把它建模成一个重试边而不是一个死循环。3.1 最小实现节点、边、执行器先定义节点。一个节点有两个核心方法process负责任务执行decide负责选择下一步。为了让逻辑清晰我把节点设计成执行 决策分离的形态这样即使以后要在大模型和纯代码逻辑之间切换改动也很小。from dataclasses import dataclass, field from typing import Dict, Any, Optional, List from enum import Enum class NodeStatus(str, Enum): SUCCESS success FAILED failed SKIPPED skipped dataclass class Context: task: str data: Dict[str, Any] field(default_factorydict) iteration: int 0 max_iteration: int 5 class Node: def process(self, ctx: Context) - Any: raise NotImplementedError def decide(self, result: Any, ctx: Context) - str: # 返回下一个节点的名字或 END return END然后定义边。边的作用就是把节点之间的跳转规则和一个名字绑定在一起便于管理。这里我偷了个懒直接用字典来存边但对于更复杂的项目建议用一个Edge类把跳转条件和执行参数都存起来。class Graph: def __init__(self): self.nodes: Dict[str, Node] {} self.edges: Dict[str, Dict[str, str]] {} # node_name - {condition: next_node} def add_node(self, name: str, node: Node): self.nodes[name] node def add_edge(self, from_node: str, condition: str, to_node: str): self.edges.setdefault(from_node, {})[condition] to_node def get_next(self, from_node: str, condition: str) - Optional[str]: return self.edges.get(from_node, {}).get(condition)最后是关键的执行器。它不用while但用了一个for循环限定最大步数本质上是把无限循环替换成了有限循环 显式跳转。这一步可以说是整个架构的核心转变你不再依靠模型的自我判断来决定何时结束而是依靠图上预先定义的边来结束。def run(graph: Graph, ctx: Context, start_node: str start): current start_node for step in range(ctx.max_iteration): if current END: return ctx node graph.nodes[current] result node.process(ctx) condition success if result else failed next_node graph.get_next(current, condition) if next_node is None: # 没有定义下一步默认结束 current END else: current next_node # 超过最大步数可以记录日志或抛出异常 raise TimeoutError(fGraph exceeded max step {ctx.max_iteration})有同学会问这不是还是有for嘛对但这个for和while有决定性的区别它每一步走哪个节点是由边的条件决定的不是由模型的自由意志决定的。while循环里模型让循环继续程序就继续图执行器里模型只负责单个节点的产出是否跳转、跳去哪是预先设计好的业务规则。所以这个for只是一个安全熔断器不是架构的核心控制机制。3.2 用队列代替循环的细节如果你的场景里确实需要动态添加任务比如从网页里找出所有链接再逐个抓取那么用while很自然用一个队列也能做到而且更可控。这个技巧我很推荐给初学图编排的人用队列实现遍历逻辑而不是用循环实现无脑重试。from collections import deque class CrawlNode(Node): def process(self, ctx: Context) - List[str]: # 从页面里提取链接返回新发现的 URL 列表 return [http://example.com/page2, http://example.com/page3] def decide(self, result: List[str], ctx: Context) - str: # 如果有新链接继续到 fetch 节点否则结束 return fetch if result else END在执行器层面可以维护一个待处理队列每个 URL 都作为一个任务项流入图里处理完一个再弹出一个。这个结构天然避免了无限递归也能通过队列长度来限制最大待办量一旦超限就停止扩展这正是爬虫里常见的种子队列思路。放到 Agent 里也一样当模型说要继续探索时它只是把新任务塞进队列而不是在原地打转。还有一个细节节点要尽量做成幂等的。也就是说同一个节点对同一份输入执行两次结果应该一样。这很重要因为图架构允许你随时回退重试某个节点如果节点有副作用比如发了封邮件、改了数据库重试就会重复操作造成事故。我的习惯是所有对外部系统有写入操作的节点都要定义一个idempotency_key执行前先查重执行过就直接跳过。3.3 处理循环需求的替代方案重试、回溯、迭代节点有些任务天然需要循环比如重试直到成功为止。我的做法不是禁止循环而是把这些循环封装成具有明确终止条件的节点。比如重试我写一个RetryNode内部用一个for循环重试 N 次每次重试之间做指数退避重试次数耗尽就返回failed。它作为图中的一个节点存在对外只有一个输入一个输出不会让整个执行流陷入无限循环。这个节点的重试逻辑完全透明你可以看到它重试了几次、每次的 token 成本是多少。再比如回溯在某些工作流里当前节点失败后需要回到上一个节点重新调整参数再试。这其实在图上就是一个回到上游的边。我通常会加一个attempt字段到 Context 里每回溯一次attempt 1超过限定次数就彻底失败。这样就能精确控制任务不会来回跳没完。class ResearchNode(Node): def process(self, ctx: Context) - Any: # 复杂执行逻辑 ctx.iteration 1 if ctx.iteration 3: raise RuntimeError(too many iterations) return {analysis: ok}你看到没有这里我仍然用了递增计数这种循环的核心思想但它被封装在了节点内部而不是作为整个 Agent 的主循环。主循环只有图执行器那一个所有循环都是显式的、有边界的、可观测的。这套思路如果你的项目是用 LangChain 或 LlamaIndex 这类框架同样适用因为它们底层就是图结构只是它们默认给你用了循环的 AgentExecutor你要做的往往是把它替换成自定义的图执行逻辑。4. 常见问题与排查技巧实录这一部分我把自己在实际迁移过程中踩过的坑以及和团队一起摸索出来的排查方法完整整理出来。如果你正打算把while循环的 Agent 改成图架构下面这些几乎你都会遇到。4.1 死循环怎么避免我在前面反复强调显式边界目的就是对付死循环。但即使换成了图死循环依然可能出现只不过它变成了图上成环的边。比如你定义了A - B - A这样的边如果执行器没有限制步数它照样能无限转圈。我的避坑经验有三条第一所有端口一定要有全局步数上限我这个示例里是max_iteration线上环境一般设 100太高的任务多半是图设计有问题。第二每个节点尽量记录已访问次数当某个节点被访问超过预设次数比如 5 次时强制让它走失败分支而不是继续走成功分支。第三环路必须有下降趋势意思是从 A 到 B 再到 A 时任务的某种关键指标比如搜索深度、结果条数必须比上次更好如果指标没有变好立刻终止。if ctx.iteration 5 and previous_score current_score: raise StopIteration(no improvement, stop loop)这种方法尤其适合那种迭代改进类的任务比如让 Agent 反复打磨一篇文章如果没有一个文本质量评分作为下降趋势的锚点它就会永远改来改去最后甚至改出一堆废话。有了评分进步不大就主动收手效果会好非常多。另外要警惕大模型自带的对话惯性。在循环架构里模型发现自己答不上来时往往会继续生成毫无意义的让我再想想本质是一种幻觉式填充。在图架构里你要在节点输出端做结构化校验。比如定义好节点应该输出 JSON如果解析失败就直接走失败分支重试或者终止而不是把失败的文本原封不动传下去。这个习惯可以帮你挡掉很多模型带来的隐性死循环。4.2 状态丢失怎么办图执行器的状态分散在 Context 和各个内部节点一旦丢失整个任务就可能跑偏。我遇到最多的问题是真的把Context对象深入某个子函数里改了一通结果因为 Python 的引用共享或异步并发的竞态状态被覆盖或漏写。解决办法是一个很老套但超级有效的习惯把 Context 设计成纯数据结构只在执行器层修改它。节点内部只能读 Context要改数据的话通过返回值传到执行器由执行器统一合并。这样做的好处是状态流转路径非常清晰不会出现三个节点同时改同一个字段这种灾难。如果你不得不保留内部状态一定要加锁或者用不可变类型 新变量代替原地修改。第二个状态问题就是我在前面说的持久化。线上 Agent 跑一半崩了重新拉起后要能续跑。我的方案是给 Context 增加一个serialize和deserialize方法把整个执行进度存到数据库节点和边的定义本身是纯代码不需要持久化只要保存当前节点名和 Context 内容就够。恢复的时候先跳到上次的节点继续执行而不是从头来。这一步看似简单却让 Agent 可靠性上了一个大台阶也是循环架构完全做不到的。4.3 并发场景下的坑当你把 Agent 拆成图之后第二个自然会想到的事就是并发。很多团队问我ai agent 怎么扛并发我的答案是先别急着上 Kubernetes先把你的 Agent 执行器的并发模型搞清楚。我踩过一个大坑用 Python 直接跑多线程调用多个串行节点以为节点之间互不依赖其实是共享了同一个大模型 API 的配额结果一并发所有请求都撞到限流。所以正确的做法是把节点的执行和调度解耦。调度只有一份负责维护每个任务在哪个节点节点执行可以丢给一个线程池或进程池但要限制并发上限并且每个并发任务带独立的 Context绝不允许共享可变状态。另一个并发坑是并行子任务的结果合并。比如你让三个搜索节点并行它们各自返回一堆结果合并的时候如果没有去重策略最终报告里会有大量重复内容。我建议在合并节点里显式处理这些情况用散列或相似度过滤一遍再交给模型去总结。这样才能保证并发带来的收益不是垃圾数据而是真正的高质量信息。至于怎么选择执行器我实践下来简单场景用同步的图执行器就够也就是我上面写的那个run函数配合队列就能实现最基本的并发。如果任务更重再考虑用 asyncio 或者 Celery 来承载节点执行。但核心架构不变节点和边始终是业务逻辑的主体异步只是执行层的实现细节。不要让并发框架绑架你的架构。4.4 调试与可观测性最后谈谈调试。图架构最大的红利就是调试起来真的爽。每次执行结束我可以完整导出执行轨迹包括每个节点的名称、输入、输出、耗时、token 数、跳转的目标。把这些数据画成一张图虽然这里不画 mermaid但我自己用 JSON 记录任何一环不对劲一眼就能揪出来。我强烈建议在开发环境里给每个节点加这么一段代码import time import logging def run_node_with_log(graph, node_name, ctx): start time.perf_counter() try: result graph.nodes[node_name].process(ctx) logging.info(f[{node_name}] success, elapsed{time.perf_counter()-start:.3f}s) return result except Exception as e: logging.error(f[{node_name}] failed: {e}, elapsed{time.perf_counter()-start:.3f}s) raise日志里除了时间还要记录关键参数比如当前节点收到的输入摘要、模型返回的原始文本长度。这样出了 bug你能明确判断是模型输出有问题还是你的代码逻辑有问题还是外部 API 超时。这三类问题在循环架构中经常混在一起而在图架构中天然被隔离到不同节点上排查难度完全不一样。5. 我对这个架构的最终评价与几点延伸想法说实话拆掉 while 循环这个决定是我做 Agent 开发以来回报率最高的架构调整之一。它没有引入什么高深理论只是把一个大黑盒拆成了若干个小盒子加几条明线但效果立竿见影token 成本降低了大概 30%死循环基本绝迹线上问题定位时间从小时级降到分钟级而且因为每个节点可控我甚至开始敢让 Agent 去执行一些之前完全不敢放给它跑的写操作——因为每一步都有权限校验和回退机制兜底。我最开始也担心为这个架构多写的那一大堆节点类是不是过度设计后来发现完全不是。当你面对的是一个稳定运行的生产系统那些看起来多余的边界条件、状态校验、重试策略全都是必须存在的东西。而 while 循环恰恰把这些东西全都抹平了让你只能在事后追着日志叹气。如果你现在还在用 while 循环写 Agent我建议你可以先不做全面重构只挑一个经常出问题的场景把它的核心逻辑改成图编排试试。比如就把反复搜索直到满意改成搜索节点 评价节点 重试边这个极小的图你立刻就能感受到差别。等你习惯了这种节点 边的思考方式再回头去看那些复杂的 Agent 框架你会发现它们其实都是这套思想的工业化封装到那时你再决定要不要引入重框架思路会清晰得多。最后再分享一个小技巧无论你用什么实现一定要在 agent 的核心执行路径上强制要求每个节点显式声明自己的输出格式和成功条件哪怕多写两行代码。很多异常和死循环本质上都是因为节点认为自己成功了但下游觉得它失败了这种语义不一致。只要每个节点输出都是严格结构化的图编排的威力才能真正发挥出来。我自己在实战里反复吃到这个甜头也劝你尽早把它变成你的默认习惯。