Agent工程核心拆解:Harness、Loop与Graph三层架构实战指南
做Agent工程这两年我最大的感受是大部分翻车事故都不是模型不够聪明而是外围那层工程架构没扛住。记忆丢失、工具乱调、循环停不下来、上下文爆炸这些问题根子其实都落在 Harness、Loop、Graph 这三层架构上。这套三层架构不是某个框架的专利而是所有生产级 Agent 系统都躲不开的骨架Harness 负责管好外部资源和生命周期Loop 负责让模型反复思考行动Graph 负责把复杂任务拆成可编排的节点。这篇文章适合两类人看一类是正在从“写个 Prompt 调用模型”往“做正经 Agent 项目”转型的开发者另一类是已经在用 LangGraph、CrewAI、自研框架但总觉得系统不可控、不好排查问题的工程师。我会把这层架构拆开讲配合真实可落地的代码示例、参数设计逻辑和踩坑记录尽量让大家看完就能往自己的项目里搬。1. 为什么 Agent 工程需要三层架构先别急着看代码想清楚分层这件事后面能少走很多弯路。Agent 系统的复杂度跟传统 CRUD 完全不是一个量级模型会产生不确定输出工具会失败用户会中途改需求上下文会膨胀。如果不做结构性分层所有逻辑堆在一起调试的时候根本分不清是模型错了、工具错了还是编排错了。1.1 没有分层时踩过的坑我最早做 Agent 也是单文件脚本加一个大循环所有状态塞在一个 dict 里工具调用、模型推理、结果处理全部挤在一起。当时做了个客服工单 Agent功能跑起来没问题一到生产就暴露了某个工具调用超时后异常直接打断主循环后续节点全没执行模型上一轮留下的中间状态污染了下一轮判断想重放某个失败请求连日志都串不起来。那种感觉就像把所有业务逻辑写在一个几千行的函数里每次改需求都像在拆炸弹。后来才意识到Agent 工程的问题本质不是“模型聪明不聪明”而是“系统边界清不清楚”。Harness、Loop、Graph 这三层其实是把 Agent 运行时的核心问题分开治理谁来提供环境、谁来驱动思考、谁来组织任务。1.2 三层的职责边界与协作关系我习惯用一张表来记这三层的边界开发时心里始终装着它就不会乱层级核心问题主要组成典型故障HarnessAgent 运行的外部环境和生命周期配置加载、模型客户端、工具注册表、权限控制、审计日志、超时重试插件没启动、工具权限过大、配置丢失、token 预算失控Loop模型“推理-行动-观察”的迭代过程Thought/Action/Observation 循环、终止条件、历史状态、去重机制死循环、上下文爆炸、重复调用工具、成本超支Graph任务拓扑与依赖编排节点、边、条件边、并行分支、状态对象、检查点环、数据竞争、状态不同步、分支无法汇合三个层级不是孤立存在的。Harness 创建并驱动一个或多个 Agent 实例每个 Agent 内部跑着 Loop而面对复杂任务时Graph 把这些 Loop 和普通业务节点编排成一张有向图。Graph 是编排骨架Loop 是决策引擎Harness 是外围护甲。搞清楚每个问题应该在哪一层解决是做好 Agent 工程的第一步。2. HarnessAgent 的外部骨架与运行环境2.1 Harness 到底在管什么Harness 这个词来自工程学里的“线束”就是那一捆把电源、信号、接口整理好的线缆。Agent 里的 Harness 也是这个意思它把模型连接、工具接口、配置项、错误处理、日志审计这些“线”整理到一个稳定的外壳里让 Agent 本体只需要专注于推理和决策。很多开发者对 Harness 的理解是“一个启动类”其实它管的事远比启动多。我实践下来Harness 至少要承担这几件事加载配置和密钥、初始化模型客户端、维护工具注册表包括工具描述和入参校验、做权限拦截哪些工具当前会话能用、统一处理超时与重试、累加 token 用量、记录全链路审计日志、提供优雅关闭和会话恢复能力。举个例子同样一个“查物流”工具Harness 层可以做一层“口罩”记录调用方是谁、调用了几次、返回了多少字节然后在工具返回异常时包装成统一错误码而不是让原始异常直接炸到模型面前。这些事如果放到业务节点里每个节点都要重复做而且容易漏。2.2 Harness 与 Agent 的区别壳与脑很多初学者分不清 Harness 和 Agent甚至把两者混为一谈。我在工程里习惯这样定义Agent 是决策单元它接收观察、选择动作、生成回应表现的是“策略”Harness 是承载 Agent 的运行壳它提供工具、限制权限、控制生命周期表现的是“治理”。可以类比成一个餐厅Agent 是厨师负责决定先切菜还是先炒菜Harness 是厨房本身决定水电气是否通畅、食材哪些可用、油烟怎么排、火警怎么处理。如果让厨师自己去修水管那他就是不是一个合格的厨师了。同样的道理如果让 Agent 自己决定调用哪些工具、每次调用等待多久、token 用了多少那决策负担太重且很容易绕过治理规则。所以在代码结构上我严格要求Harness 不写任何业务决策逻辑业务节点也不直接碰模型客户端的底层细节。两者通过接口衔接Agent 只知道自己能调用哪些工具不需要知道工具背后是 HTTP 还是本地脚本。2.3 一个最小 Harness 的代码骨架纸上谈兵没意思我直接给一个最小可用的 Harness 骨架语言用 Python方便大家复制改造import json import time import uuid class Harness: def __init__(self, config): self.config config self.tool_registry {} self.session_store {} self.model_client self._init_model_client(config[model]) self.trace_id uuid.uuid4().hex def _init_model_client(self, model_config): # 实际项目里这里会初始化 OpenAI/DeepSeek/本地模型网关等客户端 # 关键点是统一出口便于在 Harness 层做 token 统计和超时控制 return ModelGateway(model_config) def register_tool(self, name, description, handler, required_permissionsNone): self.tool_registry[name] { handler: handler, description: description, permissions: required_permissions or [], } def run(self, user_input, session_idNone): # 1. 恢复或创建会话 session self._get_session(session_id) # 2. 校验输入注入会话上下文 state { task: user_input, session: session, trace_id: self.trace_id, tool_results: [], final_answer: None, } # 3. 交给执行引擎Loop / GraphHarness 自己不碰业务 executor self.config.get(executor) # 可以是 Loop也可以是 Graph result executor.execute(state, harnessself) # 4. 统一审计 self._audit(state, result) return result def call_tool(self, tool_name, arguments, context): if tool_name not in self.tool_registry: raise ToolNotFoundError(tool_name) # 这里可以统一做权限校验、超时控制、调用审计、异常包装 handler self.tool_registry[tool_name][handler] return handler(**arguments)核心思路就一句话Harness 只做环境和治理不写业务判断。ModelGateway 可以在 Harness 初始化时注入 token 计数器每轮对话结束后把增量 token 累加到 state 里超时和重试逻辑也统一放在 call_tool 里而不是让每个工具实现者自己处理。2.4 生产级 Harness 的四个治理点骨架只是入门生产环境真正要命的是下面四个治理点第一工具白名单与权限收敛。很多框架默认把全部工具暴露给 Agent这非常危险。我见过一个 Agent 因为拿到了“删除文件”工具一次误调用直接清掉了测试环境的关键数据。正确做法是 Harness 根据当前会话的角色和任务动态生成工具可见列表会话 A 能看的工具会话 B 不一定能看。第二超时与重试策略。工具调用必须设 timeout我常用的基线是普通查询 10 秒生成类任务 30 秒超过直接熔断。重试最多 2 次采用退避间隔 1 秒、2 秒。不要无限重试否则下游服务会被 Agent 拖垮。第三Token 预算熔断。Harness 层要维护一个累计 token 计数器预算耗尽时强制停止并返回“预算不足请缩小问题范围”。计算方式很简单单轮预算 平均每步 token 数 × 最大步数 × 1.3 的冗余系数宁可预留多一点也好过中途被硬切。第四审计与追踪。每个工具调用的入参需要脱敏、返回状态、耗时、token 增量都必须落到结构化日志里链路 ID 贯穿 Harness、Loop、Graph。否则线上出问题你根本不知道 Agent 到底干了什么。3. LoopAgent“思考-行动”的主循环3.1 ReAct 循环的简化模型Loop 这层解决的是“Agent 如何一步步逼近目标”的问题。目前主流范式还是 ReActThought思考→ Action行动→ Observation观察→ 再思考如此往复。模型先根据当前状态决定要调什么工具工具返回结果后模型再决定下一步是继续行动还是输出最终答案。这听起来简单但工程化以后有很多细节。比如 Thought 该怎么引导Action 应该用结构化的 JSON 还是 function callingObservation 太长怎么办历史里保留多少轮这些都会直接影响模型收敛效果。我用一个客服 Agent 举例用户问“我的订单怎么还没发货”模型先 Thought“用户想查订单状态需要先获取订单号”然后 Action“调用 get_order_id”Observation 返回订单号再 Action“调用 track_delivery”拿到物流轨迹后最终组织成人话返回用户。3.2 Loop 的终止条件设计Loop 最难的其实不是循环本身而是“什么时候该停下来”。很多 Agent 翻车就翻在这里模型觉得自己已经完成任务了直接输出最终答案但实际上流程只走了一半或者模型在一个问题上反复绕圈怎么都出不来。我现在的做法是绝不只靠模型自然语言里的“done”来判断终止。至少要叠加这样几层硬性条件达到显式任务完成标志比如状态机里“全部子任务状态为 completed”。超过最大步数max_steps常见设置为 5~10复杂任务可以放宽到 15。Token 预算耗尽由 Harness 层统一统计并强制返回。工具连续失败达到阈值比如连续 3 次工具异常直接转人工或返回错误。用户主动取消。终止条件必须写在代码里而不是写在 Prompt 里。如果你只是让模型“想清楚再回答”它大概率会在没有完成时提前收工或者在能停下来时继续胡诌。我在生产里吃过太多次这种亏后来干脆给每个任务类型显式定义一个验收函数模型输出 final_answer 之前必须匹配验收逻辑。3.3 带保护的 Loop 实现我写一个带基础保护的 Loop 示例方便理解这层到底该做什么class AgentLoop: def __init__(self, max_steps10, max_tokens12000): self.max_steps max_steps self.max_tokens max_tokens self.observed_hashes set() def execute(self, state, harness): steps 0 total_tokens 0 history [] while steps self.max_steps and total_tokens self.max_tokens: # 1. 让模型基于当前历史和工具结果生成下一步 response harness.model_client.respond( messageshistory, toolsharness.get_visible_tools(state), trace_idstate[trace_id], ) steps 1 total_tokens response.usage.total_tokens history.append(response.message) # 2. 如果模型输出了最终答案且通过验收就结束 if response.is_final_answer and self._validate_answer(response.content, state): state[final_answer] response.content break # 3. 如果模型要调用工具 if response.tool_calls: for tool_call in response.tool_calls: observation harness.call_tool( tool_call.name, tool_call.arguments, contextstate, ) history.append({ role: tool, tool_call_id: tool_call.id, content: observation, }) # 4. 去重保护如果同样的观察出现多次说明在空转 obs_sig self._signature(history[-1]) if obs_sig in self.observed_hashes: state[loop_abort] repeated_observation break self.observed_hashes.add(obs_sig) if steps self.max_steps: state[loop_abort] max_steps_exceeded elif total_tokens self.max_tokens: state[loop_abort] token_budget_exceeded return state关键点在于第 4 步的去重保护。很多死循环不是模型死机而是模型每轮都会调用同一个工具、拿到同一个结果然后又做出同样的决策。这时候如果你只看步数可能要等到 max_steps 才停加一个观察签名去重连续重复两次基本就能判定空转了直接退出省成本。3.4 Loop 失控的常见特征与成本控制Loop 失控有明显的先兆我建议把下面几条写进告警规则同一工具的同一参数组合被调用超过 2 次却没产生新状态。单步 response 的 token 数量超过历史最高值的 2 倍。单步响应越来越长但动作一直没变。模型反复要求“更多上下文”但从不执行操作。成本控制方面核心是让“最大步数”动态化而不是写死一个常数。我的做法是先估算平均每步成本再结合本次任务剩余预算计算允许的步数上限。比如平均每步 1500 tokens总预算 9000 tokens那就把 max_steps 设为 5保留 1500 tokens 给最终生成。用公式表达就是max_steps min(default_max, (total_budget - reserved_tokens) / avg_step_tokens)这里 reserved_tokens 是给最终答案预留的输出空间建议不低于 500 tokens。这样做的好处是如果任务很复杂但预算不够Agent 会更快放弃而不是闷头跑到一半被掐断。4. Graph把复杂任务画成有向图4.1 从顺序链到有向图的跃迁当任务只有一个 Loop 就能解决时Graph 是多余的。但真实业务很少这么简单一个工单 Agent 可能要同时查订单、查库存、查物流还要根据不同的用户意图走完全不同的处理流程。如果所有逻辑都塞在同一个 Loop 里模型每一步都要从十几个工具里选一个既慢又容易错。Graph 这层做的事情就是把“接下来该执行哪个逻辑单元”从模型手里拿回来一部分交给显式的拓扑结构来控制。顺序链只能表达“A 后接 B”有向图能表达分支、并行、回退、循环。你可以把 Graph 理解成一张地铁图站是节点线是边不同线路可以换乘某条线封了可以绕路而不是只能从头坐到尾。特别说明一下我讲的 Graph 不特指 LangGraph。LangGraph 是一个优秀的实现但 Graph 本身是一种抽象自己写也不复杂。理解了节点、边、状态这老三样任何框架对你都只是提效工具。4.2 Agent 工程里的 Graph 核心要素Graph 里最重要的四个要素是节点Node、边Edge、状态State、调度器Executor。节点是执行单元可以是一个普通函数、一次 LLM 调用也可以是一个完整的 Loop。我习惯把所有“会改变核心状态”的动作都抽象成节点比如意图识别、工具查询、数据二次处理、生成回复。边分两种顺序边和条件边。顺序边就是“上一个完成后自动进入下一个”条件边需要读状态根据状态的值决定走向哪个分支。条件边是 Graph 表达力的关键没有它就只能画直线。举个例子意图识别节点识别出用户想“退货”就走退货流程识别出“查物流”就走物流查询流程。状态是全局共享的上下文对象但有个设计原则节点之间尽量不要直接修改彼此的字段。每个节点只负责读取自己需要的前置字段写入自己拥有命名空间的输出字段。这能有效防止并行分支写冲突。调度器负责按拓扑顺序执行节点。理想情况下它应该支持并行分支但在生产环境里我建议初期先按拓扑排序顺序执行等机制稳定了再上并行否则排查问题会很痛苦。4.3 用简单 GraphBuilder 实现条件分支下面给一个足够写小型 Agent 的 GraphBuilder 示例没有依赖外部框架class GraphBuilder: def __init__(self): self.nodes {} self.edges [] # (from, to, condition) def add_node(self, name, handler): self.nodes[name] handler def add_edge(self, from_node, to_node, conditionNone): self.edges.append((from_node, to_node, condition)) def _next_nodes(self, node_name, state): next_nodes [] for from_node, to_node, condition in self.edges: if from_node node_name: if condition is None or condition(state): next_nodes.append(to_node) return next_nodes def execute(self, start_node, state): current_nodes [start_node] executed set() while current_nodes: next_nodes [] for node_name in current_nodes: if node_name in executed: continue executed.add(node_name) handler self.nodes[node_name] result handler(state) if result: state.update(result) # 找到下一个满足条件的节点 next_nodes.extend(self._next_nodes(node_name, state)) current_nodes next_nodes return state调度逻辑不复杂从 start_node 开始执行节点后记录返回值到 state然后扫描当前节点的所有出边找到满足条件的下游节点继续执行。这里为了示例简化成顺序执行没有处理并行但已经能支撑大部分业务分支流程。比如我想实现“工单先分类是 FAQ 就走检索是复杂问题就走重写查询人工复核”def classify(state): # 调用模型或规则返回 intent state[intent] model_classify(state[task]) return state def retrieve_faq(state): state[faq_answer] search_faq(state[task]) return state def complex_handle(state): loop AgentLoop(max_steps5) state loop.execute(state, harness) return state g GraphBuilder() g.add_node(classify, classify) g.add_node(faq, retrieve_faq) g.add_node(complex, complex_handle) g.add_edge(classify, faq, conditionlambda s: s[intent] faq) g.add_edge(classify, complex, conditionlambda s: s[intent] ! faq) state g.execute(classify, initial_state)注意这里“complex”节点内部又是一个 Loop。Graph 是宏观编排Loop 是微观迭代两者嵌套使用非常自然。这也是我标题里“Harness、Loop、Graph”三层架构能成立的原因Harness 管环境Loop 管单任务的思考过程Graph 管多任务之间的协作拓扑。4.4 状态管理与回退Graph 是 Loop 的容器Graph 真正难的点不在“把图画出来”而在“状态怎么管”。我在生产里的经验是状态对象必须不可变或半不可变每个节点执行前先快照节点失败时能回退到上一个稳定状态。为什么需要快照因为 Agent 的每一步都可能产生不可预期的副作用。比如“写工单备注”节点写了一半下一个“发送通知”节点失败你如果把整个流程回滚重跑有可能给用户发了两次通知。正确做法是节点操作严格区分“可重试”和“不可重试”可重试节点失败后可以重新执行不可重试节点失败后转入人工审批节点不做自动重放。Checkpoint检查点是实现回退的基础设施。每执行完一个节点把当前 state 的版本号和关键字段存到数据库或 Redis后续要排查或重放时直接找对应版本就行。Graph 层面我还会给 state 加一个单调递增的 version 字段并行分支写冲突时后写入的分支要检查 version不一致就放弃而不是覆盖。5. 三层架构的生产落地与联动实践5.1 完整示例工单 Agent 的三层协作用一个具体场景把三层串起来用户提交“我的订单还没发货帮我查一下并催催”。Harness 先接住这个请求从请求头解析用户身份加载该用户的权限配置初始化 trace_id从 session_store 恢复上一次会话状态。Harness 会看到这个用户只有“查订单、查物流、发催单通知”三个工具可用没有“改价格”“删订单”等敏感工具。接下来 Harness 把状态交给 Graph。Graph 第一步是“意图分类”节点识别出这是“查物流催发货”的组合意图于是走分支先查订单再查物流然后进入“催发货 Loop”节点。这个 Loop 节点内部执行多轮迭代第一轮调用物流查询工具发现物流 3 天没更新第二轮调用催单通知工具第三轮等待通知结果并确认是否发送成功。Loop 跑完之后Graph 进入“汇总答案”节点把查询结果和催单结果整理成用户能看懂的话。最后 Harness 统一把这次调用的工具日志、token 消耗、节点耗时写入审计库然后把最终答案返回给上层应用。整个过程里Harness 不关心业务细节Graph 不关心模型怎么思考Loop 不关心外部工具权限各层各司其职。5.2 可观测性设计我见过太多团队把 Agent 丢上线然后出问题只能靠“重新跑一遍碰运气”这就是没做可观测性。三层架构的可观测性要分层设计Harness 层记录每个会话的 trace_id、工具调用入参出参脱敏、耗时、token 累计、错误码。Loop 层记录每一步的 thought 摘要、action 名称、observation 长度、步数、当前 token 消耗。Graph 层记录节点开始/结束时间、条件边命中分支、状态快照变化、异常节点和重试次数。我常用的日志结构是 JSON 一行一条{ts: 1730000000, trace_id: abc123, layer: loop, event: step_end, step: 3, tool_call: track_delivery, tokens: 1520, status: success}排查问题时直接按 trace_id 聚合能还原 Agent 每一步的完整行为。注意脱敏工具原始返回里可能带用户手机号、地址等敏感信息落日志前要做字段过滤否则合规上会出事。5.3 部署形态与运维要点部署形态上Harness 可以作为一个独立服务也可以作为库嵌入你的主服务。我倾向于独立服务因为 Agent 的负载特性和普通 Web 接口不同长耗时、高 token 消耗、突发调用多独立出来好做资源隔离和扩缩容。Loop 和 Graph 要尽量保持无状态运行中的状态全部放在 Redis 或数据库这样即使 Harness 实例崩溃重启也能从最近一个 Checkpoint 恢复任务而不是让用户重新描述一遍问题。模型客户端走统一网关Harness 对接内网模型网关不在业务节点里直接拼模型地址。插件动态加载要注意入口校验与版本兼容环境变量尽量收敛到配置中心不要散落在各节点代码里。我遇到过的教训是把 Loop 的历史消息直接存内存一重启就丢用户等半天后只能收到一条“会话已过期请重试”。后来改成每轮结束把 history 和 state 快照一起持久化体验立刻好了很多。6. 常见问题与排查技巧实录6.1 高频问题速查表下面是这段时间被问得最多的几类问题我整理成速查表基本覆盖生产环境 80% 的 Agent 故障现象根因解决方案Loop 不停跑费用暴涨终止条件缺失或模型重复输出加 max_steps、token 预算、观察去重工具调用失败后整体中断节点未做异常隔离Harness 统一异常包装失败转重试或人工Graph 状态乱A 分支覆盖 B 分支并行分支写同一个 key每个分支写独立命名空间加 version 检查上下文爆炸模型开始胡言乱语历史全部塞给模型只保留关键 Observation压缩中间过程插件加载报错入口不激活插件入口类未注册/扫描路径错误检查入口注解、依赖版本、插件 HOME 路径响应越来越慢单步 token 过多或者循环过长动态步数控制单步 token 上限6.2 一次真实的插件加载失败排查有一次在生产环境部署一个基于 LLM 的 Harness 项目启动时一直报 “Harness failed to load plugins: web boot: 1 entry did not activate”意思是有个 Web 插件的入口没有激活。当时第一反应是插件代码写错了但单独运行插件又一切正常。排查过程走了不少弯路最后发现是插件被安装到了两个目录Harness 的插件扫描器只扫描了其中一个目录结果入口类虽然存在但注册逻辑没有在扫描路径里执行。后来我把插件目录统一收敛到 manifest 指定的路径清理了旧目录重启就正常了。这个案例的通用启示是插件加载问题先看扫描目录配置和入口注册机制再看依赖版本最后才怀疑业务代码。Harness 这类外壳项目最坑的就是隐式约定太多一个文件放错位置运行时不会报编译错而是报一个“入口未激活”这种让人摸不着头脑的错误。6.3 我踩过的几个坑和补救习惯最后分享几个踩过几次坑之后养成的习惯现在基本成了我的默认规范。第一不要把所有东西塞进 Harness。Harness 是壳不是业务逻辑仓库。一旦你开始往 Harness 里写“如果意图是A就走A流程”这样的代码说明你把 Graph 的活抢了。这会让系统越来越难扩展。第二给 Loop 一个“看得见的终止信号”。让模型在最终回答前必须输出一个结构化标记比如 final_answer 标签Harness 解析到标记才真正结束而不是靠模型自然语言里的“我可以回答这个问题了”去判断。没有这个信号你永远无法确定模型是真结束还是假装结束。第三Graph 节点尽量幂等、无状态。每个节点只依赖传入的 state 参数不读全局变量不做隐式副作用这样回放、重试、回退都会非常安全。你会发现一旦做到了这一点调试 Agent 的过程从“猜”变成了“看日志”。最后分享一个我一直在用的习惯在 Harness 层加一个“心跳事件总线”所有的 Loop 和 Graph 节点都向总线汇报进度。前期会多一点代码但生产出事时能快速定位到是第几步、哪个节点、哪次工具调用出了问题。Agent 工程没有银弹能救命的都是这些朴素的治理手段希望这篇三层的拆解能让你在实际项目里少走弯路。