如果你真正动手做过 Agent 项目大概率会撞上同一个感受Demo 跑得飞起一到生产就翻车。今天能用的对话流程明天换个工具就崩明明只是加一个“查天气”的功能却要把整个主流程的代码翻一遍。我意识到问题不在模型智商而在工程结构。后来我花了两三个月把项目拆成三层——Harness、Loop、Graph——很多反复出现的疑难杂症才真正消失了。这篇文章我想完整讲一遍这三层架构包括每一层负责什么、层与层之间怎么配合、生产落地时具体怎么做以及我在重构过程中踩过的坑。内容不偏理论基本都是可以直接复用的设计思路和代码骨架。适合正在做 Agent 应用、或者准备从单体 Agent 代码往工程化方向重构的开发者和技术负责人。1. 为什么 Agent 工程需要三层架构从 Demo 到生产的那道坎1.1 单体 Agent 代码是怎么一步步失控的很多人写 Agent 的第一个版本就是在一个 Python 文件里把大模型 API、工具函数、Prompt 模板、状态缓存、重试逻辑全部堆在一起。第一版几十行确实够用但随着工具增多、业务分支变复杂这个文件会膨胀到两三千行。此时最典型的问题有三个第一职责完全耦合。模型调用逻辑和业务判断逻辑粘在一起改 Prompt 要小心翼翼不然可能影响工具调用的解析加一个新工具要改主流程否则模型绕不过去。第二无法单独测试。整段逻辑跑起来才知道对不对单元测试根本无从下手因为每一步都在依赖全局状态。第三崩溃恢复能力为零。一旦某个工具调用超时或者返回脏数据整个 Agent 会话就废掉用户只能重新开一个对话。我见过很多项目死在“能演示”到“能稳定跑”的鸿沟上。三层架构就是为这道鸿沟准备的。1.2 三层架构的职责边界一句话说清这三层我能用最直白的话给你捋清楚Harness 是 Agent 的“躯体”负责运行环境、工具注册、插件加载、安全拦截、模型适配。说白了就是让 Agent 跑得起来、跑得安全的那层壳。Loop 是 Agent 的“神经回路”负责在一个任务内部反复执行“思考-行动-观察”的循环。大模型每一步该调什么工具、看到结果后下一步怎么决策都在这一层完成。Graph 是 Agent 的“路径规划”负责多个步骤之间、多个 Agent 之间的编排。任务先干什么后干什么、什么情况下走哪个分支、哪些子任务可以并行都由 Graph 层决定。打个比方Graph 是流水线的传动带和分拣口Loop 是某一个工位上工人反复完成的操作循环Harness 是整座车间的照明、通风、安全护栏和电力系统。三者不互相替代而是各管一段。1.3 三个设计原则控制反转、可观测、可恢复我在重构时给自己定了三条硬性原则缺一条后面都会痛苦。控制反转要求 Agent 内部逻辑不直接 new 工具对象、不直接发起网络请求而是通过 Harness 提供的接口发出意图由 Harness 决定调用哪个真实实现。这样换工具、加限制、做 mock 都集中在 Harness 一层不用动业务逻辑。可观测意味着每一层都要输出结构化日志。接口调用人、调用工具名、消耗 token、循环次数、图节点状态全部带链路 ID 记录下来。没有这一步后面排查问题基本靠猜。可恢复是指任何一次工具调用失败都不应该让整个会话报废。Loop 层要捕获异常并反馈给模型让它重新决策Graph 层要能重试节点Harness 要负责把崩溃的会话恢复到最近一个稳定状态。2. Harness 层Agent 的躯体也是安全与扩展的边界2.1 什么算一个“够用”的 HarnessHarness 这个词在 Agent 生态里出现频率越来越高但很多人把它和“Agent 框架”混为一谈。我理解的区别是框架解决“怎么写 Agent”Harness 解决“怎么安全稳定地跑 Agent”。一个够用的 Harness 至少要包含四个模块模型适配器统一封装不同大模型 APIOpenAI 兼容接口、DeepSeek、Claude 等让上层 Loop 不必关心底层是哪个模型。工具注册中心维护一个工具名到实现函数的映射表同时声明每个工具的参数格式和权限级别。插件加载器从指定目录按约定动态加载技能包实现“加一个能力不用改主代码”。生命周期管理负责 Agent 启动、会话创建、超时回收、优雅退出。2.2 Harness 不是 Agent 框架别再混淆了我们拿主流 Agent 框架来做对比容易理解。框架一般给你一个 Agent 类和一套抽象工具你继承后写 Prompt 就能出活。而 Harness 更像一个容器它不关心你的业务逻辑是什么只负责提供“模型进来、工具出去、日志留下、异常拦住”的通道。我用一个真实经历说明为什么需要区分。早期我在项目里给 Agent 加“上传文件解析”能力时直接在 Agent 类里写了一个函数。后来权限收紧要求解析前必须做文件类型白名单校验结果改动牵一发动全身。重构到 Harness 之后这个校验逻辑被抽成 Harness 层面的拦截器所有工具调用统一过一遍拦截安全策略变成可配置项不再和业务纠缠在一起。2.3 从零搭一个极简 Harness插件化设计下面是我常用的 Harness 骨架用 Python 写一个最小版本核心思路是“能力即插件”。# harness.py import importlib import json import yaml from pathlib import Path from typing import Dict, Callable, Any class ToolRegistry: 工具注册中心管理所有 Agent 可用的工具 def __init__(self): self._tools: Dict[str, Dict[str, Any]] {} def register(self, name: str, func: Callable, schema: dict, permission: str read): self._tools[name] { func: func, schema: schema, permission: permission, } def call(self, name: str, **kwargs): if name not in self._tools: raise KeyError(fTool {name} not registered) tool self._tools[name] # 安全拦截可以放在这里校验参数、检查权限、记录审计日志 return tool[func](**kwargs) class PluginLoader: 插件加载器扫描 plugins 目录并按约定加载工具 def __init__(self, registry: ToolRegistry, plugin_dir: str): self.registry registry self.plugin_dir Path(plugin_dir) def load_all(self): for manifest_path in self.plugin_dir.glob(*/plugin.yaml): with open(manifest_path, r, encodingutf-8) as f: manifest yaml.safe_load(f) module_path manifest[entry] module_name fplugins.{manifest[name]}.{Path(module_path).stem} module importlib.import_module(module_name) for tool in manifest[tools]: func getattr(module, tool[handler]) self.registry.register(tool[name], func, tool[schema], tool.get(permission, read)) class ModelAdapter: 模型适配器统一模型接口底层可替换 def __init__(self, base_url: str, api_key: str, model: str): # 这里以 OpenAI 兼容接口为例DeepSeek、通义等都能用这个方式接入 from openai import OpenAI self.client OpenAI(base_urlbase_url, api_keyapi_key) self.model model def chat(self, messages, **kwargs): return self.client.chat.completions.create(modelself.model, messagesmessages, **kwargs) class Harness: Agent 运行外壳把插件、模型、会话状态组合在一起 def __init__(self, config_path: str): with open(config_path, r, encodingutf-8) as f: self.config yaml.safe_load(f) self.registry ToolRegistry() self.plugin_loader PluginLoader(self.registry, self.config[plugin_dir]) self.plugin_loader.load_all() self.model ModelAdapter( base_urlself.config[model][base_url], api_keyself.config[model][api_key], modelself.config[model][name], ) self.session_state {} def create_session(self, session_id: str): self.session_state[session_id] {history: [], tool_logs: []} def get_state(self, session_id: str): return self.session_state.get(session_id)核心在于你把“有哪些工具”和“怎么调模型”全部收口到 Harness 内部。上面这段代码看起来简单但已经可以支撑一个最小可用的 Agent 外壳加载 plugins 目录下的技能包模型走统一适配层会话状态集中管理。配置文件大概长这样plugin_dir: ./plugins model: name: deepseek-chat base_url: https://api.deepseek.com api_key: ${DEEPSEEK_API_KEY}2.4 DeepSeek 等模型接入 Harness 时的真实适配经验现在国产大模型的 API 大多兼容 OpenAI 协议所以 Harness 的适配层不需要为每家单独写客户端。只要把 base_url 换成对应的服务地址即可。我在接 DeepSeek 时遇到过一个坑它的敏感内容拦截策略和 OpenAI 不太一样某些 Prompt 会触发空回复甚至报错。Harness 层必须捕获这种异常返回给 Loop 层让模型换一种措辞重新组织答案而不是直接把异常抛给用户。我把模型调用包装成下面的模式def safe_chat(self, messages, max_retries3): for attempt in range(max_retries): try: resp self.client.chat.completions.create(modelself.model, messagesmessages) if not resp.choices or not resp.choices[0].message.content: raise ValueError(Empty response, may be blocked by safety policy) return resp.choices[0].message.content except Exception: # 退避重试让上层 Loop 有时间调整 Prompt time.sleep(2 ** attempt) raise RuntimeError(Model chat failed after retries)另一个经验是插件加载失败的容错。有段时间我经常看到类似“harness failed to load plugins web boot: 1 entry did not activate”的报错查下来基本都是插件入口的加载时序问题插件模块在 boot 阶段依赖了尚不存在的全局对象。解决方式是在 loader 里捕获单插件异常记录错误后继续加载其他插件不要让一个坏插件拖垮整个 Harness。3. Loop 层Agent 的神经回路也是智能的关键引擎3.1 ReAct 循环到底在转什么Loop 层是很多人理解的“Agent 本体”因为它对应的是 ReAct 模式的核心循环。这个循环可以概括为四步模型根据当前状态产出下一步思考然后决定调用某个工具Harness 执行工具并返回结果模型观察结果更新认知再次进入下一步思考。周而复始直到达成目标或者触发终止条件。这就类似视频处理里 ffmpeg 的 loop 滤镜把一个片段反复循环播放。但 Agent 的 Loop 必须带退出机制否则就成了死循环。所以 Loop engineering 的核心不是“让模型多转几圈”而是“让模型在该停的时候停下来”。3.2 Loop engineering 的核心参数与终止条件我把 Loop 层最关键的设计参数整理成了表格方便你对照自己的项目调整参数作用建议取值max_steps单轮任务最大决策次数10~20视任务复杂度而定stop_condition用户定义的任务完成判定自定义函数或模型自评context_compress_threshold触发上下文压缩的 token 阈值模型窗口的 50%~70%retry_when_tool_error工具报错后是否允许模型换策略重试是但最多 2~3 次temperature模型决策采样温度工具调用场景 0~0.3有个容易忽略的点不要只靠模型自己判断“任务完成了”。模型的自评很容易产生幻觉尤其在多步骤任务里它会在中间某一步误以为目标已达成。所以我通常加一层规则校验比如必须调用指定工具拿到最终结果后循环才允许结束。3.3 用伪代码写一个带反思的 Loop下面是一个典型 Loop 骨架加入了反思Reflexion机制当任务失败或偏离方向时让模型自己生成一段修正总结下次迭代时带上这段总结重新决策。# loop.py class AgentLoop: def __init__(self, harness, max_steps15): self.harness harness self.max_steps max_steps self.history [] def run(self, task: str, session_id: str) - str: messages [ {role: system, content: 你是一个能调用外部工具的 Agent...}, {role: user, content: task}, ] reflection for step in range(self.max_steps): # 把反思结果附加到上下文 if reflection: messages.append({role: user, content: f反思{reflection}}) response self.harness.model.chat(messages) thought, action self.parse_decision(response) self.history.append({step: step, thought: thought, action: action}) if action.get(type) finish: return action.get(answer) if action.get(type) tool: try: observation self.harness.registry.call( action[tool_name], **action[args] ) except Exception as e: observation f工具调用失败{e} reflection f上次调用 {action[tool_name]} 失败原因是 {e}。请换一种方式。 messages.append({role: assistant, content: response}) messages.append({role: tool, content: str(observation)}) return self.force_summary(messages)注意我在工具报错失败后没有直接终止而是记录 reflection 让模型下一轮修正。这招实测非常管用很多第一轮参数传错的问题第二轮模型就能自己纠正过来。3.4 循环中的状态管理与上下文压缩Loop 层最痛苦的工程问题是上下文爆炸。模型窗口有限工具返回结果一长几千 token 就没了。我的实践分三步解决第一步每次工具调用结果只保留截断版本比如 500 字以内超出部分丢弃或写入外部存储。第二步消息列表定期做压缩把早期的对话改写为摘要。第三步压缩后的摘要和最近的原始消息一起送进模型。def compress_if_needed(self, messages, max_tokens8000): total sum(estimate_tokens(m[content]) for m in messages) if total max_tokens: return messages # 把最旧的一半消息合并为摘要 old_part messages[: len(messages) // 2] recent_part messages[len(messages) // 2:] summary self.harness.model.chat([ {role: system, content: 请用三句话概括以下对话的已完成信息和关键事实}, {role: user, content: json.dumps(old_part, ensure_asciiFalse)}, ]) return [{role: system, content: f历史摘要{summary}}] recent_part这里有一个经验摘要不能只记录“聊了什么”要记录“已经完成哪些动作、得到哪些事实”。否则模型会忘记之前已经调用过工具导致重复调用同一个查询。4. Graph 层Agent 的路径规划调度一切的编排骨架4.1 从单 Loop 到图编排什么时候必须升级单个 Loop 解决“做一件事”Graph 解决“做一件事的流程里有很多步骤、分支、并行、人工审批”。我把必须上 Graph 的信号列一下任务不是一次对话能完成的包含多个独立阶段比如“先收集信息再分析再生成报告”。存在条件分支根据中间结果决定下一步走哪条路径。需要并行执行多个任务比如同时搜索多个数据源。中间步骤需要人工确认例如付款、发布、删除操作。需要多个专用 Agent 协作而不是一个通用 Agent 包办所有事。没到这些信号之前我建议你先别引入 Graph 框架。因为图编排的抽象成本不低简单任务强行上 Graph 反而是过度设计。4.2 Graph 和 Loop、Harness 是怎么配合的Graph 层不直接调模型、也不直接调工具它只做一 件事维护节点状态和流转。一个节点的内部逻辑可以是一个 Loop比如让 Agent 自主完成“资料调研”这个环节也可以是一个单纯的 LLM 调用甚至可以是一个人工审批页面。Harness 在这些节点下面提供公共能力模型接口、工具调用、上下文存储、权限控制。所以我常说 Graph 是“脑”Loop 是“神经”Harness 是“身体”。这个类比帮我和团队沟通时省了很多解释成本。4.3 一个最小 Graph Engine 的骨架市面上有 LangGraph、Coze 等工作流引擎但如果只想理解原理一个最小执行器就够了# graph.py class Node: def __init__(self, name, handler, next_mapNone): self.name name self.handler handler # 接收 inputs 和 context返回 (outputs, next_node) self.next_map next_map or {} class Graph: def __init__(self): self.nodes {} def add_node(self, node: Node): self.nodes[node.name] node def run(self, entry_node: str, initial_inputs: dict): current entry_node inputs initial_inputs context {} while current: node self.nodes[current] outputs, next_node node.handler(inputs, context) context[current] outputs current node.next_map.get(next_node, next_node) inputs outputs if current END: return context return context配合具体节点def collect_inputs(inputs, context): # 这里可以放一个 Loop Agent也可以直接调模型 return {collected: 原始信息...}, analyze def analyze(inputs, context): return {analysis: 分析结论...}, report def report(inputs, context): result f报告{context[collect_inputs][collected]} / {inputs[analysis]} return {report: result}, END实际生产里每个节点的 handler 就是完整封装的一个执行单元内部可以包含 Harness 调用和 Loop 循环。这样业务逻辑被拆成了零件测试和维护都清晰很多。4.4 Graph 四类高频模式与跨领域图思想的相通之处我整理了自己项目里最常用的四类 Graph 模式顺序链最简单A 节点完成进 B 节点适合流水线型任务。条件分支根据判断节点输出决定下一步走左侧还是右侧。并行扇出与汇聚一个节点同时发起多个子任务全部完成或部分完成后汇聚。人工审批任务运行到特定节点时挂起等待审批结果通过后继续。有意思的是这种“局部子任务各自处理再汇总到全局结论”的思路和脑功能网络分析中的 local-to-global 思想很接近。脑科学的图模型也是先构建局部脑区子图再整合成全局连接模式。Agent 的 Graph 编排同样遵循局部自治与全局协作的平衡。另外可以提一下商图quotient graph概念当你把一组内部结构复杂的重复子流程看作一个整体时整个流程图就被大大简化了。我在做 Agent 流程回归测试时会把“用户提问→信息检索→内容生成→反馈优化”这一整块收缩成一个抽象节点然后用简化后的商图验证整体路径是否符合预期。这个思路特别适合大型流程图的可视化审查。4.5 画图可以别把图画成蜘蛛网我见过最失败的一个 Graph 设计节点有 40 多个边连得密密麻麻根本看不出主流程。后面我定了几条规矩主路径控制在 5 个节点以内复杂分支用子图包装。任何节点只能有一个明确的输入来源避免多人同时拉数据导致状态混乱。每个节点必须定义超时和失败出口不能死等一个子任务。工具层面如果你喜欢可视化建图可以找支持 graph builder 的 Agent 编排工具把节点拖拽出来生成配置。但记住图形化只是辅助核心还是节点间的数据契约要清晰——每个节点输入什么、输出什么必须在 Graph 定义里写清楚。5. 生产实践三层架构落地的全链路细节5.1 可观测性怎么知道 Agent 在干嘛没做可观测性之前生产出了问题只能看用户反馈猜。重构后我把可观测性按三层拆开Harness 层记录每次模型请求的 token 数、用时、模型名、插件加载清单。Loop 层记录每一步的思考内容、决策动作、工具名和参数、工具返回摘要。Graph 层记录节点进入/退出时间、节点状态、分支走向、并行子任务结果。用 trace_id 贯穿三层一条用户请求从 Graph 节点到 Loop 循环再到 Harness 调用全程可串联。我这里分享一个常用格式把结构化日志写成 JSON 行后续接 ELK 或 Loki 都方便{trace_id: abc123, layer: loop, step: 3, action: tool_call, tool: weather_query, args: {city: 北京}, cost_tokens: 386}会话回放也很关键。我会定期把会话状态做快照存到对象存储。出问题时直接把快照灌回本地环境复跑一次就能定位根因。这个方法省了我大量排查时间。5.2 安全防线工具权限、输出过滤与审计Agent 的安全问题不能靠模型自觉要靠在 Harness 层拦截。我实践下来至少有四道防线工具白名单Agent 能用的工具必须在 Harness 注册表里存在且权限等级要匹配会话级别。参数校验工具调用前校验参数格式和取值范围防止模型生成越界参数。危险操作拦截涉及删除、写入、支付的动作强制走 Graph 的“人工审批”节点。输出过滤模型返回内容过敏感词和策略过滤器再展示给用户。我踩过最痛的一个坑是模型在一个内部工具里生成了删除数据的请求幸好有白名单拦截否则后果严重。从那以后我把“最小权限原则”写进 Harness 的代码评审规范里。5.3 评估与回归没有 eval 就不要上生产很多人把 Agent 做完就上线只测了几条路径。这种项目我基本可以断定会在线上出幺蛾子。一套可用的评估体系至少包含三类指标任务成功率最终结果是否满足用户预期需要人工或裁判模型打分。过程效率平均循环步数、工具调用次数、token 消耗步数异常增多往往说明 Prompt 有歧义。鲁棒性对输入变体、错别字、工具返回异常、网络超时的容忍度。我习惯把评估用例分成三档正常路径、边界路径、故障注入路径。故障注入就是故意让某个工具返回报错看 Loop 层能否恢复。实测下来能在故障注入测试中存活下来的 Agent上线后稳定性会高很多。5.4 性能优化少花钱、少等几秒的具体手法Agent 生产环境最大的成本往往是模型调用。我有几个省钱又省时间的经验用模型路由简单节点用小模型复杂推理用大模型。DeepSeek 这类模型性价比高可以作为默认主力遇到特别复杂的推理再切换更强大的模型。工具结果缓存相同参数的查询结果缓存十分钟避免重复调用。并行节点并发执行Graph 层里互不依赖的节点用 asyncio 并发跑整体耗时能缩短一半。另外提一句LLM 输出 JSON 解析很容易出问题哪怕让模型“只输出 JSON 不要其他内容”它偶尔还是会带 Markdown 代码块。我建议在 Harness 层做一次 “JSON 容错解析器”把 json 包裹、前后多余文本全部去掉再解析。我项目里因为这个解析器工具调用的成功率从 78% 提到了 95%。6. 问题排查实战我踩过的七个坑6.1 最典型的 Agent 故障现场我把实际操作中反复踩过的七类问题列出来每条都是真实生产案例。死循环Agent 在某个决策点上反复调用同一个工具参数稍有变化但逻辑完全一样。原因是 stop_condition 没有覆盖“重复尝试无进展”的情况。我最后加了一个“连续同动作次数”计数器超过 3 次就直接强制终止并向上报告。JSON 解析失败模型生成的工具调用参数不合法常见于参数里出现单引号、换行、注释。解决办法就是上文说的 JSON 容错解析器以及 Prompt 里给出严格的参数示例。上下文爆掉长会话运行到一半token 数超了模型窗口。靠压缩机制减少了近 60% 的异常中断。工具幻觉模型调用了一个实际上不存在的工具名。原因是注册表里的工具描述和用户 Prompt 的意图模糊地带太多。我在 Prompt 里显式列出工具清单并加了一条工具选择规则“如果没有匹配工具直接告诉用户能力边界”。Graph 节点超时某个子 Agent 卡了很久不返回导致整个图挂起。后来每个节点都配超时时间超时后走失败分支或返回哨兵值。插件热加载状态丢失更新插件后新版本函数使用的内存缓存全部丢失用户感知就是“功能偶尔失效”。我把插件状态迁移到独立的 Redis 或文件存储插件只读状态不再持有内存缓存。评估通过但生产翻车测试集覆盖不到真实用户的长尾输入。现在我会每个月拿线上日志的失败样例回灌到评估集保持测试库跟着生产一起进化。6.2 快速定位问题的方法与兜底策略排查 Agent 问题你别一上来就改代码。我的标准流程是先看链路日志确认问题出在哪一层如果 Harness 日志显示模型调用失败那就是适配或网络问题如果 Loop 日志显示反复调用同一工具那就是决策逻辑问题如果 Graph 日志显示某个节点没走完那就是流程编排问题。定位到层之后再借助会话回放复现。我强烈建议所有 Agent 项目上线前都搭一个“回放调试”工具把线上 trace 文件重放到本地沙箱比断点调试高效得多。6.3 附常见故障速查表现象可能原因快速排查最终方案Agent 反复调用同一工具循环终止条件不完善查看 Loop 日志中动作序列增加重复动作计数器工具参数解析失败LLM 输出 JSON 不干净查看原始模型输出Harness 层做 JSON 容错解析上下文超限工具结果未截断查看历史消息 token 统计压缩机制结果截断调用了不存在的工具工具描述与需求意图模糊查看工具名和注册表显式工具清单选择规则图节点长时间无响应子 Agent 或工具未设超时查看 Graph 节点耗时节点超时失败分支插件更新后状态异常插件内部状态在内存中查看进程是否重启状态外部化存储线上 badcase 持续出现评估集覆盖不全回放失败样例每月回灌线上日志到评估集数据序列化报循环引用Graph 节点状态里引用了父节点对象查看异常堆栈中的序列化函数只存顶层字段避免引用整个 context还有一个不太起眼但很常见的坑上下文压缩时如果把“已经完成的动作”摘要丢掉了模型会在后续步骤里重复执行。我的经验是压缩摘要必须包含三类信息已获取的关键事实、已调用的工具及结果概要、当前任务进度的判断。不满足这三项的摘要宁可多留一些原始文本也不要压缩成一句干巴巴的“用户问过天气”。结尾这套 Harness、Loop、Graph 三层架构我实践下来最大的体会是它并没有增加多少代码量却让项目的认知负担降了一个量级。以前改一个功能要在几千行单体代码里找位置现在只需要明确改动落在哪一层然后动手改那一个模块。插件、模型、工具、流程都变成了可以独立替换的零件。如果你现在正被单体 Agent 代码的复杂度折磨我建议你先别急着换框架打开你的主文件把模型调用、工具定义、流程控制分别抽到三个目录里。不需要一次到位先抽出边界再慢慢颗粒化。等你的项目开始出现“加一个工具要改三处代码”或者“一个节点不稳整条流程都挂”这类信号时三层架构就是你下一步应该走的方向。最后再分享一个小技巧每次给 Agent 加新能力都先在 Harness 层写一个对应的故障注入测试——故意让工具返回错误看 Loop 能不能自救。这比任何架构理论都更能帮你提前发现问题。