智能体(Agent)和 AI 闭环系统,是这两年最容易被高估也最容易做浅的两个词。不少团队能把一个智能体 Demo 跑通,让它在大模型对话框里回答几个问题,或者调用一次工具;但一旦要求它“自主驱动”往前推进——没有人工一步步点击,自己接收任务、执行、检查结果、根据反馈调整策略——系统就开始乱套。这篇文章要讲的不是某个具体智能体框架的安装教程,而是把“循环工程”当成一套可复用的设计方法,拆解一个自主驱动的 AI 闭环系统需要哪些模块、哪些参数、哪些终止条件,以及从单任务到批量落地时真正要盯住的问题。适合正在做智能体应用开发、准备把智能体接入业务系统、或者被“大模型只能做单轮问答”卡住的人。最值得先理解的一点是:闭环不是让 AI 一直转圈,而是让 AI 在一个有边界、有校验、有反馈、能终止的循环里可控地完成复杂任务。
1. 为什么智能体应用经常停在“能跑但不好用”
1.1 大多数智能体 Demo 只做了“单轮推理”
很多智能体项目的演示结构是:用户输入问题,模型读取提示词,系统从知识库检索相关内容,模型生成回答,然后返回。这个流程看起来像智能体,其实更接近一个“带检索的问答机器人”。比如“基于企业知识库做问答”,用户问一句,它答一句,中间没有任务状态,没有环境变化,也没有结果校验。一旦需求变成“帮我处理完这 50 个工单,把结果写回表格,并且自动标记异常项”,原来的链路就撑不住了。
单轮推理解决的是“根据给定信息生成一个合理回答”。闭环解决的是“在没有人工连续干预的情况下,把任务从初始状态推进到完成状态”。区别在于是否有状态、是否有循环、是否有可校验的中间产物。这也是为什么很多智能体项目代码量不大,但接业务时复杂度突然上升。对话演示里的偶然正确,不能直接等同于业务系统里的稳定可靠。
我会建议团队在写代码之前先明确一个判断:你到底要做“智能问答”,还是要做“自主执行系统”。这两件事的架构差很多。前者可以没有循环,后者必须有循环。
1.2 “循环工程”的核心是给循环加上边界
提到 AI 闭环,不少人首先想到的是“让 AI 自己反复执行、自己评价自己,直到满意为止”。这个想法很危险。没有边界的循环不是工程系统,而是故障系统。真正做循环工程,要回答五个问题:循环的入口是什么?出口是什么?状态怎么记录?什么条件终止?异常怎么跳出?如果这些在代码里没有明确体现,那系统跑起来就是碰运气。
我一般会先把这五个问题写在设计文档里,再写代码。如果只回答出“入口是用户提问”,说明还没想清楚。入口可能是文件变化、定时任务、数据库新增记录、接口回调。出口可能是任务完成、等待用户补充、达到最大步骤数、发生不可恢复错误。工程上关心的是“这个循环在什么情况下一定会停”,而不是“它理论上能跑多聪明”。
用过 Dify、扣子这类低代码平台的人,对“工作流”应该不陌生:拖一个节点,接一个模型,再把结果传给下一个节点。平台帮你隐藏了状态管理和循环控制。但自主驱动的智能体闭环,通常比固定工作流更灵活,也更容易失控。所以需要把循环机制显式设计出来,而不是依赖演示环境里的“刚好能跑”。低代码平台适合快速验证,自己搭循环适合理解边界、做精细控制。两者不是二选一,而是不同阶段用不同工具。
2. 最小闭环系统的骨架与首个业务场景
2.1 选场景:先做“工单自动分类与初稿回复”
我不建议一上来就做一个“全自动办公助手”或“多智能体协作平台”。目标太宽,你会分不清失败是模型问题、流程问题还是历史上下文问题。我会先选一个边界相对清晰的场景:工单自动分类与初步回复。
输入是工单标题和描述。输出是分类、紧急度、匹配的知识库条目、初稿回复。如果信息不足,则标记“需要人工补充”。如果知识库没有匹配内容,则进入“待人工确认”状态。这个场景好在所有步骤都可被结构化验证:分类是否在预设列表内、紧急度是否为枚举值、回复中是否包含必填信息、知识库引用是否存在。相比“写一段营销文案”这类开放任务,工单场景更容易验证闭环是否真的闭环。验证通过后再迁移到其他业务。
这个选择还有一个好处:它天然适合“人工介入”。工单系统本来就有审核、转派、关闭等状态,你可以在自动闭环的末端留一个waiting_user状态,让系统先处理能处理的部分,再把需要人确认的部分交给人。这样既训练了闭环能力,又不至于因为一次误操作造成业务事故。
2.2 核心循环:用伪代码搭出第一个闭环
下面是一个不绑定具体框架的伪代码,意图是展示闭环的最小结构。不要把它当成可直接上生产的代码,而是用它理解循环里的五个核心动作:初始化、决策、执行、验证、写回。
# 闭环最小骨架示意 state = init_state(task) for step in range(max_steps): log_state(state) decision = llm_decide(state, tools, available_actions) if decision.action == "finish": state["output"] = decision.output state["status"] = "success" break if decision.action == "need_user": state["status"] = "waiting_user" break result = execute_tool(decision.action, decision.params) if not validate_result(result): state["errors"].append(build_feedback(decision, result)) state["retry_count"] += 1 if state["retry_count"] >= max_retries: state["status"] = "failed" break continue state.update(result) remember(state, result) else: state["status"] = "exceed_max_steps" state["output"] = None notify_and_dump(state)这里可以看到,循环并不是while True,而是有 step 上限。每一步都会log_state。出错时把错误信息追加到 state 的 errors 里,作为下一轮模型的额外输入。这是反馈机制的关键:如果只是打印错误却不把错误内容传回模型,模型下一轮还是盲目试错。
很多第一次写这类系统的人会问:把 errors 拼进提示词里,会不会让上下文越来越长?会。所以伪代码里 errors 是“最近一次错误”,而不是“所有历史错误”。或者可以做一个结构化摘要:错误类型、出错动作、期望输出、实际输出、下一步建议。让模型看到这些,比看到一整段历史日志更容易修正。
2.3 结果校验:把错误反馈回模型,才叫“闭环”
闭环和普通顺序流程的区别,不只是代码里有 for 循环,而是“模型决策的结果会被校验,校验失败的反馈会重新进入模型”。我见过很多系统的做法是:模型调用工具成功就算成功,完全不检查输出是否满足业务要求。结果模型返回一段无法解析的 JSON,或者把结果写到错误目录,系统却显示“执行成功”。
这种问题很隐蔽。因为模型 API 调用本身没有报错,模型也没有说“我不会”,但产物是错的。所以校验器不能只检查“有没有返回”,还要检查“返回内容是否符合业务约束”。
举一个具体例子:如果模型必须输出{"category": "网络故障", "urgent": "high"},校验器就检查 category 是否在枚举值列表里,urgent 是否在列表里,字段是否有缺失。如果校验失败,就把结构化错误返回给模型,同时保留原始模型输出作为对照。规则校验比让模型自己判断“刚才做得好不好”可靠得多。模型自我评价只能作为辅助信号,不能作为闭环唯一的反馈来源。
这个思路放到低代码平台里也一样:平台节点之间的条件分支可以做规则校验,但你的业务规则要提前定义清楚。平台不会替你想明白“哪些输出是可接受的”。
3. 闭环系统的模块拆解:感知、决策、执行、记忆
3.1 环境准备:版本、模型和运行方式先锁死
闭环系统对运行环境的要求比单次问答高。我会建议先把这些条件固定下来:Python 版本、虚拟环境、依赖清单、模型调用方式、日志目录和任务状态存储。原始材料没有给具体版本,所以我不会写死版本号;你在落地时最好先确认依赖之间的兼容性。特别是 agentscope、langgraph、dify 这类工具更新频繁,版本差异可能直接导致 API 不兼容。
如果你的机器没有独立显卡,或者显存不高,可以考虑用 API 模型或小尺寸模型,但要把上下文长度和循环步数降下来。低配置机器不是不能跑智能体闭环,而是不能期待高并发和超长上下文同时稳定。把最大步数调到 5 到 8,把温度调到 0.2 左右,先用一个最小样例验证,再逐步增加复杂度。
还有一个容易被忽略的是日志目录和输出目录。很多任务卡住不是因为模型不行,而是因为程序没有权限写入输出目录,或者输出文件名冲突。先确认目录存在、路径正确、磁盘空间充足,再开始调模型。这个顺序能省掉一大半低级问题。
3.2 感知层:任务从哪来,能不能重放
闭环系统要能接收任务,并且要能回答两个问题:这个任务有没有唯一标识?同一个任务重复投递时,会不会产生重复处理?如果任务来自文件监控,要处理文件重命名和重复读取;如果来自数据库,要有状态字段;如果来自消息队列,要具备手动重投机制。不能重放的任务,在失败后很难排查。
| 输入来源 | 触发方式 | 适合场景 | 要重点检查的事 |
|---|---|---|---|
| 文件目录 | 定时扫描 / 监听事件 | 批量文档处理、导入 | 文件是否被占用、是否重复处理 |
| 数据库 | 轮询新增记录 / 定时任务 | 工单、订单、业务记录 | 状态更新是否原子、是否重复消费 |
| 消息队列 | 订阅 / 消费 | 异步任务、通知 | 消费确认、重试、死信 |
| 接口回调 | Webhook / API | 跨系统联动 | 签名校验、超时、幂等键 |
表里的每一行都对应具体工程问题。比如数据库轮询,间隔太短会频繁扫描,间隔太长会有延迟;文件监听要考虑文件写入尚未完成的情况。感知层不解决这些问题,后面的智能判断再准,也只会处理一堆重复或半截任务。
很多时候,感知层出问题看起来像“AI 智能体不稳定”。其实不是。是消息队列没有做确认,同一个任务被消费了三遍,模型就处理了三遍,产生了重复输出。这类问题在日志里看动作记录能发现,但更好的办法是设计阶段就给每个任务一个唯一 ID,并在每次调用工具时带上它。
3.3 决策层:模型参数和上下文管理
决策层是闭环里最像“AI”的部分,但往往也是坑最多的地方。模型参数要按用途区分:分类和结构化抽取用低温度,比如 0 到 0.3;开放式写作或头脑风暴可以高一些,但闭环系统通常不建议超过 0.7,否则同一个任务两次执行结果差异很大,校验器和后续工具会不稳定。
| 参数 | 作用 | 闭环系统里的建议 |
|---|---|---|
| temperature | 输出随机性 | 结构化任务用 0.1-0.3 |
| max_tokens | 单次回复长度上限 | 按任务类型设置,别给太长 |
| timeout_seconds | 单次模型调用超时 | 必须有,避免调用卡住 |
| tool_choice / function_call | 是否强制模型选择工具 | 有工具调用时打开 |
| response_format | 输出格式控制 | 需要 JSON 时启用 |
上下文管理比调参更影响闭环稳定性。每一轮模型输入不能无限追加历史。我会采用“三明治”式上下文:固定系统提示词在最上层,中间是当前任务状态、最近一次工具结果、最近一次验证错误,最下层是必须保留的几个关键字段。历史对话摘要单独保存,不要全部塞进模型输入。上下文窗口不是越大越好,窗口满时会产生截断和优先级混乱,最终导致模型忘记核心目标。
举个例子,一个工单任务如果执行了 10 步,把每一步的原始返回都放进下一轮输入,可能超过 2 万 token。模型看了前面 8 步的详细日志,真正关键的“当前状态”反而被淹没。更合理的做法是用一个state字典保存结构化信息,每一步只把字典的关键字段刷新给模型。
3.4 执行层:工具调用的超时、幂等与权限
工具调用是模型与现实世界交互的通道。最容易出现的问题是:模型认为它调用了工具,但工具实际没有执行;工具执行了,但返回结果没有正确写回状态;工具重复执行了多次,产生重复数据。工程上需要给每个工具调用加入超时、幂等和结果校验。
幂等尤其关键。比如“创建工单”这个操作,如果模型连续调用两次,可能创建两个重复工单。解决思路是让幂等键参与:每个任务有task_id,调用创建类工具时携带task_id + action + step作为幂等键,后端如果发现同一键已存在就直接返回已有结果。写操作不能依赖“模型自己注意别重复调用”,必须在工具层兜底。
另外,模型能调用的工具要遵循最小权限原则。不能让一个只负责查数据的智能体去写数据库。权限设计在闭环里不只是安全要求,还能减少模型误操作造成的故障。如果一个工具返回权限不足,应该把错误信息结构化返回给模型,让它选择其他可用动作,而不是直接把任务标记为失败。这里可以设计一层“可允许动作白名单”,模型只能在白名单范围内选择。
3.5 记忆层:短期上下文和长期知识库分开
记忆层是智能体闭环与普通脚本最大的区别。普通脚本也保存状态,但智能体要能“记住”之前的决策结果,并在后续步骤中使用。记忆可以分成两层:短期记忆就是当前任务的运行状态,随着任务结束而结束;长期记忆是把有价值的结论写入知识库或存储系统,供以后的任务检索。
企业知识库常见做法是把文档向量化后放入向量数据库,例如 Chroma、Milvus、Elasticsearch 等。选型前先看数据量、文档结构、检索条件和部署环境,不要别人用什么你就用什么。但要注意,向量数据库解决的是“语义检索”,不是“业务事务存储”。任务状态、日志、幂等键、审计记录这些结构化数据,应该放在普通数据库或日志系统里。把中间日志塞进向量库,会导致检索结果噪音变大,也会浪费成本。长期记忆里只沉淀结论、规范、已验证的答案和明确的业务规则。
我一般会在闭环跑通之后再做长期记忆。一开始只用一个字典文件保存任务 state 就够了。等状态数据结构稳定了,再考虑加向量库和知识库联动。否则知识库里的文档质量和向量化策略没验证,长期记忆反而会成为错误信息的源头。
4. 从单任务到批量:生产化前必须补的能力
4.1 先跑单条,再谈并发
如果在单任务状态下系统还不能稳定输出,就不要开批量。我一般会先把任务数量设置为 1,打开完整日志,观察这几点:输入正确解析了吗?模型决策合理吗?工具返回符合预期吗?校验通过了吗?最终输出放到哪了?如果这五步都正常,再拿几条真实样本跑。批量任务暴露的问题通常是并发资源、接口限流、输出命名冲突和部分失败处理,而不是模型能力。不要混淆两类问题。
批量跑的时候,也不要一上来把并发数拉到最大。你可以从并发 2 或 3 开始,观察接口响应时间、内存占用、日志清晰度。如果模型 API 有每分钟调用限制,并发不是越高越好,超过限制会产生大量限流错误,反而拖慢整体速度。任务系统的吞吐瓶颈可能不在模型推理,而在知识库检索或工具 API 响应。
还有输出文件命名的问题。单任务时你可以直接写output.json,批量时必须加入task_id或时间戳,否则任务互相覆盖。这个看起来是小事,但批量任务失败重跑时,命名冲突会造成数据对不上,排查起来很费劲。
4.2 任务状态、重试和死信队列
每个任务都应该有一个生命周期状态,而不是只用“成功/失败”两个词概括。我在实际项目中至少会保留这些状态:pending表示等待处理,running表示正在执行,success表示正常完成,failed表示失败但可以重试,dead表示超过重试上限,waiting_user表示需要人类补充信息。用状态机管理任务,日志里才能回答“任务现在到哪一步了”。
| 状态 | 含义 | 下一步可能流向 |
|---|---|---|
| pending | 等待调度 | running |
| running | 执行中 | success / failed / dead / waiting_user |
| success | 正常完成 | 无或归档 |
| failed | 失败,可重试 | running(重试) / dead |
| dead | 超过重试上限 | 人工处理 |
| waiting_user | 等待人工或外部输入 | 收到输入后回 running |
重试时必须考虑幂等。模型调用 API 超时后重试,通常不会产生副作用;但工具调用如果第一次已经创建了记录,第二次重试就可能重复。所以失败重试不能简单写retry(),要区分“可安全重试”和“需要人工确认”的失败。重试次数我一般设置 2 到 3 次,超过后进死信队列。
死信队列不是一个高级概念,它的作用就是把所有处理不掉的任务集中到一个地方,方便统一查看、修复数据、手动补跑或后续恢复。没有死信队列的批量任务,最后一定会出现“任务消失了却没人知道”的情况。你在设计状态表时,甚至可以把dead状态对应的处理人字段也留出来,方便人工认领。
4.3 日志和可观测性:没有日志的闭环等于黑盒
闭环系统是异步、多步骤、自动触发执行的,如果每一步不写结构化日志,出问题时几乎没办法查。最基本的日志字段包括:task_id、step、action、action_params、action_result、model_name、prompt_token、completion_token、elapsed_ms、status、error_message。这些字段最好以 JSON 格式写入,方便后续检索和统计。
可观测性不只是 log,还包括三个问题:这个任务目前状态是什么?运行到第几步?最近一次错误是什么?如果一个任务进入循环 20 步,没有日志你只会看到 token 消耗一直在涨,却不知道它为什么反复执行。我建议在循环入口处打一条task_started,在每次 action 前后各打一条,在状态写回后打一条状态快照。
不要觉得日志太多,等出问题时你会发现每条日志都有用。还可以统计每轮步骤数、调用成功率、平均延迟、最终状态分布,用来评估闭环是否整体健康。批量跑完后,我会习惯性扫一眼状态分布表:成功多少、失败多少、死信多少、等待用户多少。这个分布能直接告诉你系统离“稳定可用”还有多远。
5. 最常见的五个坑与对应排查链路
5.1 死循环与重复动作
AI 闭环最常见的故障就是模型反复调用同一个工具,或者不断生成方向相近的内容。原因是模型在开放环境里可能没有“应该结束了”的意识。解决思路不是靠提示词“请认真点”,而是提前设定硬性边界:最大步骤数max_steps、单动作最大重试次数max_retries、连续相同动作摘要对比。循环里记录每个动作的参数,当相同动作出现多次并有相同结果时,标记为“疑似重复”,触发提前终止或人工确认。
如果出现任务卡住,排查顺序可以这样走:先看是否还有日志输出。再看当前动作是什么、参数是什么、和上一轮是否相同。然后看模型调用是否超时。接着看工具执行是否无响应。最后看状态更新逻辑是否漏了。多数时候不是模型“装傻”,而是某一个中间步骤没有正确返回,导致模型拿不到新状态,只能反复猜测。
5.2 上下文污染与循环漂移
上下文污染指模型输入里混入了错误信息、过时日志或噪音数据,导致后续决策被带偏。比如前两步工具调用失败了,但如果把失败的完整报错一直保留在上下文里,模型可能以为系统有问题,开始不断尝试修复而不是继续主线任务。我的做法是:模型看到的上下文只包含简洁状态、最近一次错误的结构化摘要、需要下一步决策的字段,不包含大量原始日志。原始日志留给工程师查,不给模型。
循环漂移指任务在执行过程中逐渐偏离原始目标。比如原始目标是“提取发票信息并写入表格”,执行到第五步忽然变成了“给用户解释什么是发票”。要防止漂移,可以在每一轮把“当前目标”重新写入提示词,并让校验器检查输出字段和当前动作是否仍在允许范围内。对模型来说,重复提醒核心目标比只开头说一次更可靠。
我见过一个比较隐蔽的情况:模型在前一轮收到一个工具返回的“发票识别结果”,里面有一句“请根据以上结果生成说明”,结果模型把它当成了用户指令。所以工具返回结果要尽量和用户指令分开,工具输出加明确前缀,比如Tool result from invoice_ocr:。这个细节能减少很多误判。
5.3 反馈信息不够结构化
当你把校验失败信息传给模型时,要想想这段信息能不能指导模型修正。validate_result返回False只是告诉它“你错了”,它不知道怎么改。更好的格式是:哪个字段错了、期望什么、实际是什么、可能的修正方向。示例:action: create_issue, param category invalid, expected one of [network, hardware, account], got "other", please use only listed category。模型拿到这种反馈,下一轮才可能改对。
还有一种反馈需要额外注意:有些模型很“听话”,你说它错了它就道歉,但不会真正执行修正动作。所以反馈信息里不要问“你确认吗”,而是要给出“下一步应该做什么”。比如在反馈末尾加一行:You MUST retry with a valid category or set status=need_user if information is missing.这种指令式反馈比开放式询问更稳定。这属于提示词工程的一部分,但在闭环设计里,它直接影响循环收敛速度。
5.4 全自动闭环的业务边界
不是所有任务都适合全自动闭环。凡是涉及外部系统写入、费用计算、敏感内容对外发布、法律或合规判断,建议保留“人工确认”节点。自动闭环应该负责把任务推进到“待确认”状态,而不是直接执行不可逆操作。比如工单智能体可以先自动分类和生成初稿,但发送给客户前必须经人工审核;数据分析智能体可以生成报告,但对外发布前要人工确认。
设计这种边界不需要复杂技术,在决策层加一个分支即可。比如要求模型输出confidence字段,如果低于阈值,则状态转为waiting_user;或者对工具权限做分级,只读工具允许自动调用,写工具必须带requires_approval参数。关键是在系统设计阶段就决定哪些动作可以做、哪些不能做,而不是等出问题时再来补救。这种“人工介入”确实是闭环的一部分,它不影响自主性,反而让自主性变得可信。
5.5 安全与权限的最小化原则
智能体闭环被赋予的工具越多,风险面越大。模型可能因为提示词注入或上下文误导,调用一个业务上不应该调用的接口。应对方法就是权限最小化:给智能体提供“刚好够完成任务”的工具,不提供只读之外的高危能力;工具 API 的密钥单独配置,不写在提示词或代码仓库里;日志里不记录敏感字段,比如密码、token、身份证号等。
另外还要考虑审计。每个工具调用都要带task_id、操作来源标记、时间戳和调用参数,方便事后追溯。安全问题看起来是运维的事,但在自主驱动系统里,它直接决定你是否敢让智能体执行动作。如果不敢,那系统设计上就不应该放开自动执行权限。不要为了“显得智能”而把系统风险放大。
6. 一套可以用来验收闭环系统是否合格的清单
6.1 六个问题,逐项检查
设计完闭环后,可以用一套简单清单验收,不用等到上线才发现问题。
第一个问题:在给定非法输入或连续失败的情况下,系统是否会终止?对应的是max_steps、max_retries、重复动作检测。如果答案是不会,说明循环边界没做好。
第二个问题:如果任务被重复投递,系统会不会重复处理?对应的是任务唯一 ID、幂等键和任务状态流转。如果会重复处理,就不适合接入生产环境。
第三个问题:任务失败后,能否通过日志定位到具体步骤和原因?对应的是结构化日志和状态记录。如果失败后只能看到一句“处理失败”,这个系统基本没法维护。
第四个问题:模型的反馈信息是否会影响下一轮决策,而不是只输出一次?对应的是 errors 字段如何写回上下文。如果反馈没有传回模型,就没有闭环。
第五个问题:长时间运行时,上下文是否会被无限撑大?对应的是摘要、裁剪和状态精简。如果每一步都追加原始日志,系统跑不久就会失控。
第六个问题:意外中断后,任务状态是否能够恢复或人工干预?对应的是waiting_user、dead状态和重跑机制。如果中断后只能从零开始,说明状态持久化做得不够。
这六个问题都通过,这个闭环才算是“基本合格”。
6.2 从简单闭环开始,再逐步增加复杂度
我最后想强调的是一句经验之谈:在智能体项目里,复杂是慢慢长出来的,不是一开始设计出来的。先做一个边界清晰的单任务闭环,比如工单分类、日报生成、文档整理;跑稳之后,再加入工具调用、长期记忆、多智能体协作、知识库实时更新。每一步都留好日志和状态,确认稳定后再往前走。
多智能体协作尤其不要一开始就上。单智能体循环还没有终止控制时,多智能体之间的通信只会放大错误。部分开源项目在推进多智能体协作时会支持 A2A(Agent-to-Agent)之类的交互模式;如果你需要这个能力,可以先把单智能体闭环跑稳,再引入协作层,否则排查成本会成倍上涨。低代码平台可以帮你快速搭出带节点连接的智能体工作流,但底层依然需要你理解状态、校验、反馈和终止这些概念。
到了后面,真正决定一个闭环系统能不能上生产的,往往不是模型有多强,而是循环工程的基本功有没有做到位。先把单任务跑稳,再把反馈闭环做牢,最后再谈批量和知织沉淀。这个顺序走对了,智能体系统就不会停在“能跑但不好用”的状态。