Agent工程三层架构：Graph、Loop、Harness实战指南

做 Agent 工程这几年我踩过最大的坑就是把 Graph、Loop、Harness 三个概念混在一锅粥里写。早期项目里工作流靠 if-else 堆循环逻辑散落在各个回调函数里上线前还要为插件加载失败、状态序列化循环引用这类问题加班排查。后来我下决心把一个项目拆成三层Graph 管编排Loop 管执行Harness 管接入整套系统才真正从“能跑”变成“好维护、好部署、好扩容”。这篇文章想把我在这套三层架构上的完整理解、代码实现和生产经验一次说清楚。内容包括每一层到底解决什么问题、它们之间的边界怎么划分、一个最小可运行的 Agent 怎么从零写出来以及我在生产环境里遇到过的那些报错到底怎么排查。适合正在做 Agent 项目、或者准备把原型工程化的朋友尤其是被“循环写飞”、“插件加载失败”、“并发一高就崩”这类问题折磨过的人。1. 为什么 Agent 工程需要三层架构1.1 三个词到底在说什么先给一个直白的类比。Graph 是剧本规定了一个 Agent 从接收问题到输出答案要走哪些节点、每个节点之间怎么跳转Loop 是演员的排练循环让同一个流程可以反复执行直到满足终止条件Harness 是舞台、灯光和音响负责把演员和剧本装进去对外提供入口对内管理道具和边界。技术上说Graph 解决的是“流程怎么组织”的问题把原本散落在代码里的顺序判断、分支跳转抽成一张可读、可改、可测试的图Loop 解决的是“Agent 怎么持续行动”的问题感知、推理、调用工具、观察结果、再推理这个循环是 Agent 区别于普通 API 调用链的核心Harness 解决的是“这个 Agent 怎么在真实环境里跑起来”的问题工具注入、插件加载、权限隔离、并发调度都归它管。1.2 分层的三个实际收益分层不是架构洁癖而是生产环境逼出来的。第一个收益是可测试性。图结构独立之后我可以不启动模型、不调用任何外部工具直接用一个 mock 的节点函数把整张图的流转路径跑一遍。这在改流程时几乎是救命级的体验改一条边只需要改数据不需要动业务代码。第二个收益是可运维性。Loop 独立之后终止条件、超时控制、异常重试、token 上限这些“运行策略”可以被统一管理。生产环境里 Agent 常常跑偏如果没有一个集中的循环控制器你只能眼看着它在一个死循环里消耗 token。第三个收益是可复用性。Harness 把“接入方式”和“业务逻辑”隔离开了。同一个 Agent 内核今天可以挂在 CLI 上明天可以挂在 HTTP 服务上后天可以并到另一个工作流里当子模块。我做的项目里最少的一次改动只花了十分钟就把一个命令行 Agent 变成了一个 Web API就是因为当时分层做得干净。1.3 这套分层适合什么场景不是所有项目都需要完整的三层。写个演示 Demo一个脚本就够了但只要是准备长期迭代、多人协作、要上生产的项目我建议从一开始就按这个边界拆代码。尤其适合这几类场景需要接多个工具、需要多轮人机交互、需要离线部署、需要并发服务多个用户、或者后续会往平台化方向演进的 Agent 项目。反过来如果只是一个固定流程的简单问答强行拆三层反而增加成本自己权衡就好。2. Graph把 Agent 的“剧本”变成可执行结构2.1 节点、边、状态图的三个基本概念Graph 的核心抽象只有三个节点、边、状态。节点是一个“处理单元”接收当前状态返回处理结果。比如“意图识别节点”接收用户消息产出意图字段“工具调用节点”接收参数发起 API 请求“回答生成节点”接收检索结果生成最终回复。边是节点之间的流转规则。简单场景里边是固定的A 执行完必然走 B复杂场景里边上挂条件由上一个节点的输出动态决定下一步跳到哪里。这一步其实就是把原来写在代码里的if xxx: do_a else: do_b显式化成数据。状态是整个图共用的“黑板”通常是一个字典。所有节点都能读写它用户输入、中间结果、上下文历史、错误信息都放在里面。状态设计直接影响排障难度我后面会专门讲。实际落地时我不建议一上来就引入重量级框架。先画一张两个节点的图跑通了再慢慢加分支。一个最小 Graph 执行器四五十行 Python 就能写出来先理解本质再决定要不要引入更成熟的框架。2.2 条件分支和并行从“线性链”到“真实工作流”真实业务不会是一根直线一定会有“如果是代码问题就走代码助手节点如果是文档问题就走知识库检索节点”这类分叉。条件分支的通常做法是上一个节点在执行完后向状态里写入next_node字段图执行器读取这个字段决定下一个节点是谁。并行是另一个常见需求典型场景是“同时检索多个数据源”。实现思路是图执行器遇到并行节点时把多个分支挂到线程池里并发执行等全部完成再合并回主流程。这里有个容易踩的坑多个并行分支同时读写同一个状态字典会产生竞态。我的做法是给每个并行分支一个独立的状态切片执行完再合并绝不让他们直接改公共状态。2.3 画图时最容易踩的三个坑第一个坑是“节点粒度太粗”。有人把一整段业务逻辑塞进一个节点里图看起来只有三四个节点实际上每个节点里塞了几百行代码。图的意义就完全丧失了出了问题一样难查。我自己的标准是一个节点只做一件事能在一屏内看懂它的输入和输出。第二个坑是“状态字段满天飞”。图跑几轮之后状态字典里什么都有各节点各写各的后期根本没法追溯。后来我要求所有节点必须声明自己读哪些字段、写哪些字段图执行器在每次流转时记录这些操作出问题可以直接回放。第三个坑是“没有兜底边”。模型判断的下一步不总是可靠的一定要给图增加一个“未匹配则走兜底节点”的规则。没有兜底一个脏数据就能让整张图直接断掉用户看到的就是一个卡死的 Agent。2.4 图的边界多大的“剧本”才算合理图如果把所有细节都画进去复杂度会爆炸如果只画主干又失去了编排的意义。我的经验是把“业务步骤”画进图把“实现细节”留在节点里。比如“调用翻译工具”是一个节点但具体调哪个接口、传什么参数属于节点内部的事不进图。这样图表达的永远是“流程怎么走”而不是“每一行代码在干什么”维护起来才舒服。3. LoopAgent 的“心跳”与执行引擎3.1 一次循环到底在循环什么一个典型的 Agent 循环长这样拿用户输入让模型判断当前该做什么如果需要调用工具就发工具请求拿回结果把结果塞回上下文再让模型根据新信息继续决策直到模型认为已经可以回答用户退出循环。这个循环的本质是“行动—观察—再决策”的闭环。很多人误以为调用一次大模型就是 Agent不对没有循环的模型调用只是单次推理。Agent 之所以比普通 API 调用“聪明”正是因为在循环里它能看到自己行动的后果并根据后果调整下一步。实现上循环控制器的职责有四个顺序调度节点、维护状态、检查终止条件、处理异常。你在外面看它就是一个while但内部要考虑的问题比想象中多得多。3.2 循环的终止条件和防跑飞写循环最容易出事的地方就是终止条件。模型在复杂任务里很容易陷入“工具调完又调永远觉得信息不够”的状态。我见过最夸张的一次一个测试 Agent 在无人值守状态下连续调了六十多次工具把当天的预算跑掉大半。所以循环必须有三道硬约束最大轮数、最大 token 数、单次循环超时。任何一个先触发就强行终止并走“总结已获得信息 给出部分回答”的兜底逻辑。这个兜底很重要不能裸奔着抛异常给用户。3.3 循环状态与管理共享状态怎么不出错循环每一轮都要读写状态状态里通常包含历史消息列表、当前累计的中间结果、已调用过的工具记录。这里最容易出的问题是状态越滚越大最后把上下文塞爆。解决办法是定期做摘要压缩把早期历史消息概括成一段摘要再保留最近几轮的完整消息。还有一点要提醒不要在循环内部使用“全局单例”存状态。并发场景下两个用户会话会互相串数据这是生产环境最典型的低级事故。每个会话必须持有独立的状态实例循环执行完之后可以序列化存到 Redis 这类外部存储里方便断点续跑。3.4 循环和图怎么配合图负责复杂流程编排循环负责执行。配合方式有两种。第一种是“图驱动循环”整个 Agent 就是一张大图Loop 只是图执行器的一部分每次循环相当于重新走一遍图的若干节点。第二种是“循环驱动图”主流程是一个循环图被当作循环里的一个工具节点使用比如循环里有一个“子流程节点”它内部再跑一张子图。两种都有人用我自己的习惯是主流程用循环复杂子任务用图两者嵌套但不搅在一起。4. HarnessAgent 的“运行容器”和“生产边界”4.1 Harness 和 Agent 的区别到底在哪很多人把 Harness 和 Agent 混着叫其实分工完全不同。Agent 的核心是“决策逻辑”它在循环里决定下一步做什么而 Harness 是“承载 Agent 的环境”提供 Agent 运行所需的一切外部条件。你可以把 Agent 理解成公司里的项目组Harness 是整栋办公楼。项目组会思考、会做方案但办公楼负责供电、供水、消防、门禁、会议室预定。没有办公楼项目组只能露天办公风一吹就散没有 HarnessAgent 没有工具可用、没有权限边界、没有并发调度、没有插件管理只能在脚本里裸奔。我常跟团队说一句话Agent 决定“做什么”Harness 决定“能做什么、能做成什么样”。安全隔离、并发配额、工具白名单、插件生命周期这些都属于能做的范畴都应该在 Harness 层解决而不是散落在业务代码里。4.2 Harness 要提供的五项能力第一个是工具注入。Agent 需要的所有能力比如搜索、计算、调用业务 API都应该由 Harness 以接口形式注入进来。工具与 Agent 之间通过统一协议通信换工具不需要改 Agent 逻辑。第二个是插件管理。业务能力以插件形式扩展Harness 负责在启动时扫描插件目录、校验插件清单、把插件注册到可用列表。市面上不少 Harness 工具都在做这件事实际效果差别很大但底层思路是一致的。第三个是生命周期管理。Agent 什么状态是初始化、什么时候运行、什么时候暂停、什么时候销毁释放资源要有清晰的状态机。没有生命周期管理跑完的会话占着内存不走并发一高就崩。第四个是安全边界。工具权限要收敛Agent 只能拿到它该拿的权限。比如一个做内容生成的 Agent就不应该拥有删除数据库的权限。哪怕是内部工具权限隔离也要做因为 Agent 的行为本身就不可完全预测。第五个是并发调度。单个 Agent 实例在循环模式下是串行消耗资源的要服务多个用户就必须在 Harness 层做并发控制。这块下面单独展开。4.3 AI Agent 怎么扛并发从串行循环到生产者消费者Agent 扛并发的难点不在计算而在大模型调用慢、工具调用慢一条会话链路上的等待时间很长。如果每个请求都开一个重量级 Agent 实例资源消耗会很难看。我用的模式是“生产者-消费者 有界线程池”。请求进来先落队列Worker 按固定并发数消费队列里的请求每个 Worker 内部维护一个会话循环。并发数不是越大越好我的经验是并发大小主要看下游模型 API 和工具 API 的承受能力通常先压测确定一个安全值再留 20% 余量。另一个关键是“状态外部化”。Agent 会话运行中会在循环里不断更新状态如果进程重启状态就丢了。我把中间状态定期快照到 Redis并给每个会话分配唯一 ID客户端带着会话 ID 就能从最近快照继续执行。这套设计让 Agent 服务可以横向扩容节点挂了另一个节点接管同一个会话用户体感几乎无中断。再补充一点并发场景下一定要做超时和熔断。某个工具接口变慢会导致整个循环被拖死。我在 Harness 层给每个外部调用都包了超时超时次数超过阈值就自动熔断改走降级逻辑而不是让用户一直等。4.4 内网和离线环境下部署需要多确认的事企业里很多 Agent 项目要部署到内网服务器和公网隔离。离线部署的核心工作本质上是把“外部依赖”全部搬进去。我做过几次内网部署每次必查五样东西模型权重或模型 API 地址是不是内网可达依赖库是不是已经打成离线包工具插件需要的资源文件是否在本地向量库和 Embedding 模型是否已本地化密钥和证书是否提前注入。缺一样启动的时候就会报错而且这类报错往往指向不明排查起来很费时间。Skill 类的功能模块部署到内网也有一个常见坑插件加载顺序依赖外网某些目录离线之后就找不到内容了。我的建议是任何 Skill 或插件模块都先做一次“依赖自检”启动时把所有依赖声明打印出来缺了什么一目了然。比启动到一半才报错要省事得多。5. 从最小 Graph 到完整 Harness手写实现5.1 第一步图引擎先写一个极简图引擎。核心是节点注册、边注册和按状态跳转的执行逻辑。from typing import Callable, Dict, Any class GraphEngine: def __init__(self): self.nodes: Dict[str, Callable[[Dict], Dict]] {} self.entry None def register(self, name: str, fn: Callable[[Dict], Dict]): self.nodes[name] fn def set_entry(self, name: str): self.entry name def step(self, node_name: str, state: Dict) - str: # 执行节点节点返回更新后的 state并约定写入 next 字段作为跳转目标 new_state self.nodes[node_name](state) state.update(new_state) return state.get(next, END) def run(self, initial_state: Dict) - Dict: state initial_state current self.entry while current ! END: current self.step(current, state) return state注意这个实现里next字段由每个节点自己决定引擎不关心业务细节。生产环境可以在这个基础上加条件边、并行节点和子图支持但核心结构保持不变。5.2 第二步循环控制器图引擎只能线性走一遍Agent 需要的是“反复执行直到条件满足”。所以我写了独立的循环控制器来约束整个过程。class LoopController: def __init__(self, graph: GraphEngine, max_steps: int 10, max_wait: float 30.0): self.graph graph self.max_steps max_steps self.max_wait max_wait def run(self, initial_state: Dict) - Dict: state initial_state steps 0 while True: steps 1 if steps self.max_steps: state[final_answer] 达到最大轮数基于当前信息作答 break next_node state.get(next, START) if next_node END: break # 卡超时保护防止递归调用把进程卡死 state self._run_with_timeout(next_node, state) return state def _run_with_timeout(self, node: str, state: Dict) - Dict: # 生产环境建议用线程或协程实现真正的超时控制 return self.graph.step(node, state)我没在这个例子里直接引入asyncio但生产中建议用协程或线程池包一层真正的超时机制。循环控制器的核心价值在于终止条件是它管的、超时是它管的、重试策略也是它管的业务节点根本不用关心这些。5.3 第三步Harness 接入层最后写一个 Harness把图、循环和外部工具粘在一起。对外只暴露一个run接口。class ToolRegistry: def __init__(self): self.tools {} def register(self, name: str, fn: Callable): self.tools[name] fn def call(self, name: str, **kwargs): if name not in self.tools: raise KeyError(ftool {name} not found) return self.tools[name](**kwargs) class AgentHarness: def __init__(self, loop: LoopController, registry: ToolRegistry): self.loop loop self.registry registry def run(self, user_input: str, session_id: str None) - Dict: initial_state { user_input: user_input, history: [], tool_results: [], final_answer: None, } result self.loop.run(initial_state) return {session_id: session_id, answer: result.get(final_answer)}到这里三层已经完全分开业务节点只关心自己做什么循环控制器只关心跑多久Harness 只关心怎么接入。后续要加 HTTP 接口只需要在 Harness 外面再包一层三个核心类都不用动。5.4 加上并发和可观测性生产环境里我会在 Harness 上层增加一个 Worker 池。请求进来时先分配会话 ID再提交到队列Worker 消费队列并调用 AgentHarness 的run方法。同时我会把三个关键日志点埋进代码里节点进入、节点离开、循环终止。每个日志都带上会话 ID。这套日志在排障时价值极高。之前有一次生产问题用户说“回答很奇怪”我拉出日志发现某个工具节点反复进入了五次一眼就定位到是终止条件没拦住循环。6. 生产环境常见问题与排查技巧实录6.1 插件加载失败Harness failed to load plugins这类报错几乎是内网部署和自建 Harness 的标配。表面上是“插件加载失败”实际原因通常集中在三处插件目录路径不一致、插件依赖缺失、插件清单字段写错。我排查的顺序是先看一眼 Harness 的加载日志里插件扫描到了哪个目录确认路径对得上再手动在命令行里 import 插件的入口文件看是不是缺了依赖最后核对插件清单里的入口字段和实际文件名是否一致。大多数情况下是清单字段和文件名对不上大小写错一个字符都会栽。6.2 序列化循环引用self referencing loop detected 是啥意思这个报错多出现于把 Agent 状态序列化并存储的场景。比如你想把会话状态存进 Redis结果某个对象内部出现了循环引用序列化器就拒绝干活了。状态对象里最常见的原因是 agent 实例里挂了工具注册表工具注册表又反向引用了 agent。解决办法有三个方向第一状态里只放可序列化的基础数据和列表不放对象引用第二给状态对象加自定义序列化逻辑把需要持久化的字段挑出来单独处理第三在序列化配置里忽略导航属性。我建议优先做第一条最省心也顺便治好了状态越滚越大的毛病。6.3 执行被终止agent execution terminated due to error这个报错信息很笼统看到它先不要慌它不是告诉你失败原因而是告诉你“Agent 执行被拦截了”。常见的触发原因有循环超过最大轮数被终止、某个工具调用抛了异常被 Harness 捕获后终止、权限检查未通过被安全模块拦截。排查时我直接看两个地方终止前最后一次成功的节点日志以及终止时 Harness 记录的 reason 字段。如果是权限拦截日志里会标明哪个工具触发了哪个策略如果是超时终止看是不是循环条件设置太紧。这类问题大部分是配置问题改参数就能解决真正需要改代码的反而少。6.4 状态错乱和并发下的典型坑并发场景最常见的问题是“状态串了”。表现形式是用户 A 提问回答里出现了用户 B 的历史记录。根因几乎都是状态存储在全局变量里或者 session 维度没切干净。排查技巧是在状态里加一个不可变的 session_id 字段每次节点执行时校验不一致就立刻报错这样能把问题暴露在最早发生的地方而不是等用户反馈。另一个坑是“并发数一高单个请求变慢十倍”。原因多半是某个共享工具接口被并发请求打爆了。我给共享工具加了独立连接池和限流器并且把慢调用隔离到单独的线程池不让它拖垮主循环。6.5 常见问题速查表现象最可能原因优先排查动作插件加载失败插件清单路径或字段错误核对清单与实际文件名状态存不进数据表对象循环引用去掉对象引用只存基础数据Agent 一直不结束终止条件没覆盖某些分支检查循环最大轮数和兜底边并发下回答串号状态存储在全局变量校验 session_id 是否隔离某工具调用让全流程变慢缺少超时和隔离给外部调用加独立超时内网部署启动报错外部依赖没搬进去启动前做依赖自检我再补一个小技巧所有运行时报错在 Harness 层统一包装一次带上会话 ID、节点名和操作名。原始报错信息作为cause字段保留。这样线上日志看起来非常整齐定位问题时能少花一半时间。做 Agent 工程这几年我最大的感受是Agent 的能力上限由模型决定但工程下限由架构决定。Graph、Loop、Harness 这套三层拆分帮我解决的不只是代码组织问题更是“出了问题能不能快速定位、来了新需求敢不敢动手改”的问题。如果你正在做一个会长期演进的 Agent 项目真心建议照这个方向把边界划出来。哪怕先不引入任何框架就用我上面这套几十行的结构起步后面再逐步替换和扩展也不会走弯路。