如果你接触过几个 Agent 项目,大概率见过这样的代码:一个while True里放着几十行提示词拼接,再调用大模型,拿到结果又带着上一步的输出继续拼接。项目前期跑得欢,等工具一多、任务一变复杂,这种写法就会变成灾难——你根本分不清是“模型没有理解任务”还是“工具返回了脏数据”,更不用说做并发和灰度了。我经历过这个阶段之后,开始把 Agent 工程拆成三层:Harness、Loop、Graph。简单说,Harness 管环境与资源,Loop 管思考与行动的主循环,Graph 管任务与数据的编排。这个拆分解决了团队协作、可测试、运维和扩展的大部分问题。下面我会把这三层的边界、接口和落地要点一五一十讲清楚,并结合生产环境的实际案例,分享一些常规文档里不会写的坑。
1. Harness:Agent 的运行时外壳,不是一层普通的 API 封装
我第一次跟团队说“我们要做个 Harness”的时候,有人以为就是写一个统一的 LLM 调用封装库。这种理解很常见,但会把 Harness 做薄了。Harness 的职责远不是封装 API,而是 Agent 运行的整个外部支撑系统:环境隔离、资源权限、工具注册、状态存取、观测审计,都在这一层。你可以把 Harness 想象成“Agent 的驾驶舱和车本身”,Loop 是驾驶员的思考方式,Graph 是导航路线。车不好,路不好,驾驶员技术再好也容易翻车。
1.1 Harness 的职责边界:哪些事必须在 Harness 层解决,哪些必须留给 Loop
先框定边界,否则后面容易互相越权。Harness 层必须处理这几件事:
- 环境与运行时:Agent 跑在什么环境里?是本机子进程、容器,还是远程沙箱?工具执行时要对文件系统、网络、进程做哪些限制?
- 模型接入与切换:兼容不同厂商、不同尺寸的模型,统一成同一套接口,方便灰度切换。
- 工具注册与鉴权:每个工具的名称、参数 schema、权限级别、调用频控、是否需人工确认。
- 会话与记忆存储:对话历史、任务状态、长期记忆持久化在哪,如何读取。
- 可观测与成本计量:记录每一次 LLM 请求和工具调用的耗时、token、费用。
而“下一步该调用哪个工具”“这个结果意味着什么”“任务是否完成了”这些推理决策,必须留在 Loop 层。我见过不少项目把工具权限判断写在 Loop 里,结果每加一个工具就要改循环逻辑,改多了循环就成了屎山。正确的做法是:Loop 只声明意图,比如“我要调用 search_web 搜索某个关键词”,Harness 层负责检查这个 Agent 有没有权限调用、参数合不合法、需不需要限流,然后执行并返回结构化的结果。
1.2 为什么 Harness 和 Loop 的混淆是绝大多数 Agent 项目失控的根源
举个例子,早期我写过一个内部问答 Agent。当时的“伪 Harness”就是在循环里直接写openai.ChatCompletion.create,工具调用也直接在一个if tool_name == "search":的巨分支里实现。后来需要新增一个“查日历”工具,我不仅要改工具函数,还要改提示词拼接逻辑、改结果处理逻辑,还要担心旧会话里模型会不会误调用。测试覆盖非常脆弱,因为循环里塞了太多与“思考”无关的细节。
把 Harness 独立出来之后,情况完全不同。工具变成了一个注册表:
@tool.register( name="search_web", permissions=["read"], rate_limit=10, ) def search_web(query: str) -> list[dict]: ...Loop 层永远只拿到这样的接口:tools = harness.available_tools(),然后模型基于工具描述决定调用哪个。Harness 内部即使换了搜索服务、加了缓存、改了限流方式,Loop 和提示词一句都不用动。这才是分层该有的效果。所谓“职责单一”,不是靠代码规范管住的,是靠接口边界管住的。
1.3 生产级 Harness 的检查清单:环境隔离、密钥管理、超时和审计
我在生产环境落地 Harness 时,有一张检查清单,每次新项目都会过一遍:
| 检查项 | 生产要求 | 常见反模式 |
|---|---|---|
| 环境隔离 | 工具跑在容器/受限子进程,限制文件写入和网络访问 | 直接在宿主机执行任意命令 |
| 密钥管理 | 密钥放在环境变量或密钥系统,注入到运行时,不进入提示词 | 把 api_key 拼进 system prompt |
| 超时控制 | 每次 LLM 调用和工具调用都有独立超时(如 30s/60s) | 一个全局 timeout 覆盖所有调用 |
| 工具权限 | 工具分类为只读、写、高危,高危需人工审批 | Agent 能调用所有注册工具 |
| 审计日志 | 记录请求 id、模型、工具、耗时、token、结果摘要 | 只记录最终答案,中间过程全部丢失 |
| 多模型适配 | 统一接口,可一键切模型 | 每个模型一套接入代码 |
这看起来琐碎,但每一条都在生产事故里救过我。比如密钥管理,早期我把 API key 放在环境变量里,然后为了让模型“了解自己的身份”,顺手把 key 放进了 system prompt。结果是日志系统把整个 prompt 打到了日志平台,等于密钥泄露。后来日志平台的安全团队找上门来,我才意识到 Harness 的日志脱敏必须做在最底层,任何包含密钥的字段都不能进明文日志。
2. Loop:Agent 的思维循环,能不能停下来比能不能转起来重要
如果说 Harness 是车,Loop 就是驾驶员的思考流程。一个 Agent 没有 Loop,就是一个一次性的问答接口;有了 Loop,它才能根据工具结果不断调整动作。但很多人实现的 Loop 就是while True,这在 Demo 里能跑,生产环境根本不敢用。真正可用的 Loop 要能回答三个问题:当前状态是什么?下一步做什么?什么时候结束?
2.1 经典四阶段循环:观察、规划、行动、反思
我常用的 Loop 结构是四阶段,跟强化学习里的“感知-决策-行动-学习”有相似之处,但更工程化:
- 观察(Observe):从 Harness 拉取当前会话、工具返回、系统提醒等所有可观测信号,整理成结构化的输入。
- 规划(Plan):让模型基于观察结果输出一个结构化的决策,通常是一个 intent 或 plan object,而不是自由文本。
- 行动(Act):如果决策是调用工具,就交给 Harness 执行;如果是最终答案,就输出并结束。
- 反思(Reflect):把执行后的结果和预期对比,决定继续、调整还是终止。
对应到伪代码大概是这样:
class AgentLoop: def __init__(self, harness: Harness, max_steps=20): self.harness = harness self.max_steps = max_steps async def run(self, session_id, user_message): state = await self.harness.load_state(session_id) for step in range(self.max_steps): observation = await self.observe(state, user_message) plan = await self.plan(observation) if plan.final_answer: return plan.final_answer result = await self.harness.execute_tool(plan.tool_name, plan.args, session_id) reflection = await self.reflect(observation, plan, result) self.save_state(session_id, step, plan, result, reflection)这里的关键不是具体 API,而是每一步都有一个可检查的中间状态。出了问题,你能准确说出“在第几步调了哪个工具,返回了什么,模型又决定做什么”。没有这种中间状态,AI 项目调试就是猜谜。
2.2 终止条件:三套保险丝,防止 Agent 空转和烧钱
Loop 设计里最容易被忽略的就是终止条件。我在生产环境见过的失控,几乎都跟“停不下来”有关。一个搜索类 Agent 可能会为了一个模糊的问题反复变换关键词,一个代码类 Agent 可能会一直尝试修复同一个编译错误。所以我会同时装三套保险丝:
- 最大步数:单个 Loop 限制在 10~20 步,超过就终止并返回“未完成”状态。
- 收敛检测:如果连续 3 次循环产生的动作和结果高度相似,说明已经空转,强制中断并请求用户澄清。
- 预算上限:累计 token 数量或金额达到阈值后停止,必要时允许用户续费继续。
很多框架默认只给max_iterations,这远远不够。因为模型很擅长“绕圈”,它换个说法又试一次,步数上限往往拦不住。所以我在终止条件里一定会加“重复动作检测”:
seen_actions = [] for step in ...: if plan.tool_name in seen_actions[-3:] and plan.args == seen_actions[-3:][-1]: logger.warning("repeated action detected, abort") break seen_actions.append(plan.tool_name)这种检测看起来简单,但它能避免大量无意义的 token 消耗。我有一次线上事故就是模型不断调用同一个查询接口但是参数微小变化,导致单用户成本飙到正常情况的一百多倍。
2.3 记忆与上下文管理:窗口、摘要、外置记忆的实用策略
Loop 每一步都会产生新的观察结果,如果全部塞进上下文,两三轮后 Context 就会爆炸。不仅费钱,模型的注意力也会被大量无关工具输出稀释,导致“越跑越笨”。
我常用“滚动窗口 + 自动摘要 + 外置记忆”的组合:
- 滚动窗口:保留最近 K 轮的完整消息,K 通常取 5~8。
- 自动摘要:每 M 轮(比如 5 轮)把前面 M 轮的历史压缩成一段摘要,存到外置记忆里。
- 结构化记忆:把用户偏好、任务目标、已完成步骤、待办事项分开存储,而不是混在对话历史里。
- 按需检索:在观察阶段,只把当前相关的最新摘要和目标注入 prompt。
一个常见的实现是每隔 5 步调用一次模型做摘要:
if step % 5 == 0: summary = await self.summarize_history(state["recent_history"]) state["long_term_memory"].append(summary) state["recent_history"] = []加了这个机制后,长会话的稳定性提升非常明显。我有个客服类 Agent,之前上线后第三天就开始答非所问,因为上下文里积累了太多历史工具输出。加了自动摘要后,连续跑几个月的会话都能保持目标清晰。
2.4 子任务 Loop:当一个问题内部需要自己循环多次
复杂任务往往不是单层 Loop 能搞定的。比如 Agent 要写代码并测试,它可能进入一个“写代码 -> 运行 -> 看报错 -> 修改”的内层循环,而这个内层循环又嵌在“理解需求 -> 设计方案 -> 编码 -> 测试 -> 提交”的外层流程里。
如果直接把内层循环和外层循环写在一起,状态管理会非常恶心,回退和重试也会牵连整个外层流程。我现在的做法是:把子任务封装成一次独立的 Loop 实例,通过工具暴露给上层。
比如一个“写代码”工具,内部会自己跑 5 轮edit -> execute -> observe,最终返回一个结构化结果:
{ "status": "completed", "final_code": "...", "test_summary": "...", "attempts": 5 }上层 Loop 只看到这个结果,不需要关心内层循环的每一步。这种封装有几个好处:内层循环的 token 预算可以单独控制;内层失败不会污染外层状态;内层循环可以异步执行或重试,复用性更强。代价是你要多写一层接口,但这个成本在中后期会回本。
3. Graph:把任务编织成图,而不是链式调用
很多 Agent 框架只做了单 Agent 循环,但真实业务里,一个用户请求往往要拆成多个步骤、多个角色、多个数据源。这时候还是“一条链”的方式组织,流程一长就没法维护了。Graph 的意义在于把流程变成有向图,节点是操作或子 Agent,边是数据流和控制流。它能表达并行、分支、汇聚、重试,也能配合工具做可视化。
3.1 为什么复杂的 Agent 流程要建模成图,而不是一条链
链式调用看起来简单,但在复杂场景下有一堆问题。比如“用户想了解某个领域的近况”这个任务,链式实现是:先搜索,再依次阅读十个链接,最后总结。十篇文章只能串行读,效率低;如果第三篇文章超时,整个流程重跑,代价大;如果用户在流程中改了需求,链式代码很难动态调整。
图模型允许你把“搜索”作为源节点,后面接十个“阅读”节点并行执行,最后再接一个“汇总”节点。任何一个阅读节点失败,只需要重试那一个节点,不会影响整个流程。更重要的是,图可以可视化,业务同事能直接看到“现在跑到哪一步了”。
3.2 图节点的原子性设计:一个节点只做一件事
做 Graph 第一年的我,把“搜索+阅读+总结”放在一个节点里,结果出问题时,日志显示节点耗时 20s,但不知道是卡在搜索还是卡在总结上。后来我把节点拆成三个:
search_node:输入 query,输出 url 列表;read_node:输入 url,输出正文内容;summarize_node:输入多篇正文,输出摘要。
拆完以后,每次重试和报警都精准多了。所以每个节点最好满足三个条件:单一职责、输入输出是 JSON 可序列化的、不依赖上一次运行的进程内存。满足这三点,节点才能被独立调度、并行执行和自动重试。这也是我在图引擎设计里最看重的一条原则。
3.3 动态图:让 Agent 自己决定下一步分支
静态图适合流程固定的场景,但 Agent 的核心价值之一就是应对不确定性。所以生产环境通常需要动态图能力。动态图有两种常见形态:
- 分支节点:节点根据自身输出决定走哪条边。例如“检查结果是否通过”,输出
pass或fail,图引擎据此路由到不同下游节点。 - 路由节点:由一个 LLM 节点根据当前任务状态,从候选操作里选一条路径继续。比如“是继续资料,还是直接开始写报告”。
更高级的动态图是让 Agent 在运行时动态添加新节点或新边。这种能力强,但风险也大,一不小心就会出现环和失控调用。我给团队的规则是:动态添加的节点只能来自预定义的白名单,不能执行任意代码;每一次动态扩展都必须在审计日志里留下完整记录。否则问题排查会变成一场噩梦。
3.4 在实际生产中选择图引擎的标尺
市面上有各种图执行引擎、编排工具,我选型时主要看这几项:
| 维度 | 我的偏好 | 原因 |
|---|---|---|
| 表达能力 | 支持并行、条件分支、子图、动态边 | 覆盖绝大多数业务流程 |
| 部署依赖 | 尽量轻,最好纯代码可跑 | 不想为了一个流程框引入重型中间件 |
| 可观测性 | 节点级 tracing、重试日志 | 快速定位失败节点 |
| 学习成本 | 普通后端能上手 | 避免只有架构师能维护 |
| 可视化 | 有最好,但前期不强求 | 可视化是给沟通用的,不是给系统用的 |
我的实际建议是:初期使用代码原生方式实现图,用 if、队列、回调这些基础能力搭建自己的小图引擎。当你发现团队需要在多个业务线复用流程、需要频繁调整先后顺序时,再引入可视化编排工具。不要一开始就上重武器,否则项目会被框架约束,而不是框架服务于项目。
4. 三层架构在生产环境中的落地顺序与协作约定
很多团队拿到三层架构之后,不知道怎么开始改。这很正常,因为这不是改一两个文件的问题,而是整个工程结构的调整。我推荐一个演进路径:先稳定 Harness 接口,再约束 Loop 结构,最后引入 Graph 编排。别试图一天重构完,逐步替换会稳得多。
4.1 从 Demo 到工程:把单文件重构为三层时,第一步做什么
第一步不是写代码,而是划清接口契约。我通常先定义一个统一的AgentMessage数据结构,作为三层的通用语言:
{ "role": "user | assistant | tool | system", "content": "文本或结构化内容", "metadata": { "session_id": "xxx", "step_id": 5, "node_id": "search_node", "cost_tokens": 1234 } }只要所有中间产物都遵循这个结构,Harness 返回的、Loop 决策的、Graph 节点之间传递的数据就能互相衔接。然后才是代码拆分。我个人的拆分顺序是:
- 把工具调用从循环里抽出来,放进 Harness 的
execute_tool。 - 把循环体改造成状态机,每一步的输入输出都变成
AgentMessage。 - 当出现第二个业务流程时,再把 Loop 包到 Graph 节点里。
这样做的好处是每一步都有独立的验证点。我见过一个团队,一上来就引入图编排框架,结果原代码全部推翻,连基本问答都跑不通,这就是步子迈太大。
4.2 并发:Agent 系统扛住并发压力的关键不是模型,而是循环和图的调度
关于“AI Agent 怎么扛并发”,很多人的第一反应是“加机器、加线程”。但 Agent 的瓶颈往往不是计算资源,而是每个请求的处理模式。如果你用同步while循环处理一个请求,一个请求要经历 10 次 LLM 调用、每次延迟 2 秒,整个过程占用一个线程 20 秒。100 个并发就要 100 个线程,还很容易被下游限流。
更合适的模式是把 Loop 做成异步状态机,循环可以暂停和恢复。每次 LLM 调用返回后,状态被写入 Redis,等待下一次事件触发继续执行。这样并发数只受 LLM API 的吞吐限制,不受线程数限制。
Graph 层的并发调度也要注意。并行节点不是越好越多,因为每个并行分支都会消耗 token,而且可能触发 API 限流。我给并行分支设置最大并发数,比如同时最多 5 个阅读节点。还要设计 fan-in 聚合的背压机制,当某一分支慢时,其他分支的结果先缓存,不阻塞整体流程。
4.3 可观测性与调试:给每一层埋点,否则线上问题无解
三层架构最有价值的点在于,你可以把问题快速定位到某一层。如果没有埋点,一切定位都靠猜。我要求每一层至少输出一条结构化日志:
- Harness 层日志:工具名、参数摘要、耗时、错误、token 消耗、权限检查结果。
- Loop 层日志:step、阶段(observe/plan/act/reflect)、命中的终止条件、当前 context 估算长度。
- Graph 层日志:节点名、输入摘要、输出摘要、重试次数、上游接点。
每个请求分配一个trace_id,贯穿所有日志。这样排查问题时可以用一次grep trace_id拉出完整的执行链路。我分享一条经验:如果发现“工具返回的结果不对”,优先查 Harness 日志,因为那是工具层的问题;如果发现“结果对了但模型总是重复某个动作”,优先查 Loop 日志;如果发现“任务跳转到了错误的子 Agent”,优先查 Graph 日志。有了清晰的日志,问题就会自动归位到对应的层。
4.4 灰度发布与回滚策略:Agent 系统的发布,不能只回滚代码
Agent 的行为由模型版本、提示词、工具集合、Graph 流程共同决定,所以灰度发布要带上“行为组合包”。比如一次发布可能修改了summarize_node的提示词,同时换了更便宜的模型。如果只监控响应成功率,可能看不出问题,因为 Agent 经常“文本输出正常但结论变差”。我建议至少监控三类指标:
- 任务完成率:最终是否产出了用户可接受的答案。
- 步数分布:平均步数是否突然变长。
- 成本指标:单会话 token 是否激增。
灰度策略上,先内部用户,再 5% 流量,逐步放量到 30%、100%。会话型 Agent 在灰度时要在 session 元数据里记录发布版本号。出问题时,除了回滚代码,还要考虑已有会话的状态是否兼容。如果新版本与旧版本的状态结构不兼容,建议直接清空会话,让用户重新发起,不要硬迁移。
5. 生产环境中我踩过的坑,以及如何排查定位
这一节我挑几个代表性很强的坑讲,它们不是一次性的错误,而是反复出现在不同 Agent 项目里的共性问题,希望你能避开。
5.1 坑一:Agent 递归调用自己,形成无限循环风暴
现象:一个主 Agent 遇到无法回答的问题时,会调用“子 Agent”工具,而子 Agent 内部又因为同样的问题回调主 Agent,两个人互相踢皮球,请求量指数级上涨。某天线上直接打到 LLM 服务限流,账单也涨得离谱。
排查过程:我在日志系统里用trace_id拉出完整调用链,发现同一个会话的调用深度超过 40 层。当时我们的工具注册表只记录了“新增了子 Agent 工具”,没有限制调用深度。
解决方式:在 Harness 层统一增加递归深度限制,任何 Agent 工具调用前检查当前深度,超过 2 层直接拒绝并返回“请明确问题”。同时 Loop 层也加上了“某个工具连续被同一动作调用 3 次以上就强制结束”。从那以后,这类循环风暴再没发生过。
5.2 坑二:长会话中 Agent 越来越笨,用户目标被淹没
现象:客服型 Agent 在会话进行到 20 轮左右时,开始频繁建议用户做一些和原始需求不相关的事。比如用户想退货,Agent 却不停推荐相似商品。
排查过程:我打印了第 20 轮时系统实际发给模型的 prompt,发现最前面的 system 指令还在,但中间夹了大量工具返回的商品信息,模型注意力被这些噪声带跑了。用户原始目标被淹没在几千 token 后面。
解决方式:把“用户核心目标”单独提取出来,固定放在 prompt 的最前面,并每 5 轮通过一个轻量模型做目标一致性校验。引入自动摘要机制,把早期历史压缩成简短记忆。最后还加了一个“复述目标”动作:如果模型连续 3 轮没有引用目标字段,就触发一次系统提醒。这个方法让长会话的稳定性有了质的提升。
5.3 坑三:Harness 的工具权限过大,Agent 修改了服务器配置
现象:测试环境里,Agent 为了“验证部署是否成功”,调用了一个 shell 工具去执行命令,顺手把配置文件改了。幸好是在测试环境,否则后果严重。
排查过程:查看 Harness 审计日志,发现工具execute_command被注册给了 Loop,并且没有任何权限限制。模型只是按用户指令执行了命令,但“用户指令”其实来自一个恶意构造的输入。
解决方式:从那以后,我把所有工具按权限分为三级:只读、可写、高危。高危工具(shell、文件删除、远程调用)必须显式开启,且需要用户确认才能执行。提示词里不会再出现高危工具的全部参数,而是由 Harness 在确认后注入。经验就是:永远不要相信模型会自发地遵守安全规则,权限判断必须在 Harness 层用代码强制。
5.4 坑四:Graph 并发节点共享可变状态导致写入冲突
现象:Graph 里有两个并行节点,都要往同一个 session state 里写入各自的结果。最终 state 里只剩一个节点的结果,另一个节点数据丢失,下游节点拿到错误输入。
排查过程:对比两个节点的输出和最终状态,发现最后写入的覆盖了先前的。再查代码,发现并行节点操作的是同一个 dict。
解决方式:规定 Graph 节点只能写自己名下的命名空间,比如state["results"]["read_a"]和state["results"]["read_b"],不允许直接写state["result"]。多个节点需要聚合时,用单独的 reduce 节点合并。如果实在有并发写同一个 key 的需求,就要定义冲突解决策略,比如“求和”或“取置信度最高”。
5.5 坑五:成本失控,一个简单的 Debug 任务烧掉了大量 Token
现象:一个开发辅助 Agent,面对一段编译错误,连续调用了 15 次“修改文件”和“运行测试”工具,每次只是微调,成本是预期的一百倍。
排查过程:查看 Loop 日志,发现模型每次失败后都尝试新的修改方式,但这些方式从语义上高度相似。步数上限是 20,所以没有触发终止,但成本已经爆了。
解决方式:三个措施:一是引入重复动作检测,连续多次相同动作时终止;二是增加单步成本预算,每步累积 token 超过阈值就请求用户确认是否继续;三是给“反思阶段”增加一个明确动作——如果连续两次失败,必须向用户求助,而不是闷头硬试。
这些坑看起来独立,其实都指向同一个问题:如果没有在三层架构里提前定好边界和保险,Agent 的自由发挥就会变成生产事故。而把这些经验沉淀成代码里的检查项之后,三层架构才真正从一个“概念”变成“工程方法”。我对团队反复说的一句话是:三层架构不会让你的 Agent 立即变聪明,但它能让你在 Agent 不聪明的时候,还能科学地找出问题并解决问题。这个习惯,值得每个 Agent 项目从第一天就养起来。