Agent Harness：从能跑到稳定跑，长时任务与声明式调度的工程实践

最近在群里聊 Agent 工程发现一个很有意思的现象大家讨论的焦点已经从“哪个模型更强”慢慢变成了“怎么把 Agent 稳定地跑起来”。纯写 prompt 的时代早过去了现在大家关注的是长时任务怎么不中断、工具调用怎么可控、系统挂了怎么恢复。这个领域有个词越来越高频——Harness。我自己的感受是Harness 正在变成 Agent 项目真正拉开差距的护城河。它不解决“模型聪明不聪明”的问题它解决的是 Agent 能不能从 demo 走向生产、从跑一次变成天天跑的问题。这篇文章就想从 Anthropic 长时任务的设计思路到 Google AX 的声明式调度思想把 Harness 这条主线拆开聊一聊同时也把我实际踩过的坑、总结过的方案一块放出来。1. Harness 到底在解决什么问题1.1 从“能跑”到“能稳定跑”单个 Agent 演示起来非常惊艳让它完成一个多步骤任务、调用几个工具、给出一个合理结果看起来一切都很顺。但真正放到业务里事情就变味了任务可能跑十分钟、半小时甚至跨天中间要调外部 API要读写数据库要操作浏览器网络抖动一下任务就断某个工具返回了一个不符合预期的结构后面的逻辑跟着崩。这时候你面临的核心问题已经不是“prompt 写得对不对”而是“整个执行过程有没有兜底”。Harness 这个词最早源自测试领域意思是“测试夹具”把被测对象架起来、接好输入输出、观察它的行为。搬到 Agent 领域Harness 就是围绕 Agent 搭建的一整套执行支撑系统。它负责拉起 Agent、注入上下文、控制工具调用权限、管理沙盒环境、记录执行日志、在异常时重试或降级。一句话概括让 Agent 的每次执行都处于可控、可观测、可恢复的状态。我见过很多从零搭 Agent 的团队一开始热情满满地写 agent loop等做到第五个功能需求时发现代码里到处都是 while 循环和 try-except。其实这已经是在手搓 harness 了只不过搓得很痛苦。与其这样不如一开始就把执行层、策略层、观测层分开设计。1.2 Harness 的三个核心组成执行、护栏、观测拆开看Harness 至少要覆盖三块东西。第一块是执行骨架。Agent 的主循环、步骤解析、工具调度、上下文拼接这些都算执行骨架。它决定了一个任务从“拿到用户请求”到“最终返回结果”之间经过了哪些阶段。执行骨架设计得好的系统每个步骤都是显式的出了问题你能精准定位到第几步。第二块是护栏。Agent 直接调用工具是有风险的尤其是删除类操作、写操作、外发请求。Harness 要在这中间做权限校验、参数校验、频率限制。有些团队会在这里挂内容过滤规则比如不允许模型输出包含敏感信息的文本。记忆安全也是一个新方向比如 A-memguard 这种针对 LLM Agent 记忆的主动防御框架本质上也是在护栏层做文章防止恶意信息污染长期记忆。护栏不是限制能力而是让 Agent 在边界范围内安全地发挥。第三块是观测。Agent 是个黑盒你不知道它为什么突然改主意、为什么反复调用同一个工具、为什么在某一步卡了十分钟。观测层通过结构化日志、token 消耗记录、工具调用轨迹、成本统计把黑盒变成白盒。没有观测的 Agent 系统出问题只能靠猜调试效率极低。能力维度没有 Harness 的 Agent有 Harness 的 Agent故障恢复任务中断只能从头跑支持断点续跑、局部重试权限控制模型拿到什么工具就能用什么按任务、按角色设权限边界可观测性只有终端输出有步骤级日志、成本、耗时并发能力靠手动编排容易资源争抢有队列、限流、幂等控制沙盒隔离直接跑在宿主机独立容器/进程失败不扩散这可能有点抽象我们可以把 Anthropic 的长时任务设计当作第一个具体案例来看它恰好把执行骨架、护栏和观测这三个问题都回答了一遍。2. Anthropic 长时任务设计给我们的启发2.1 长时任务的三个生命周期阶段Anthropic 在 Agent 工程上的一个非常突出的贡献是把 Agent 任务的结构由“一次对话”扩展成了“一个长生命周期的工作流”。常见的实现方式里一个长时任务会被拆成三个阶段解析编排阶段、执行阶段、收尾归档阶段。解析编排阶段负责把用户目标转成可执行的任务图这里面既有模型动态决策的部分也有规则约束的部分。执行阶段是真正跑工具、调模型、处理中间结果的地方。收尾归档阶段负责把执行记录落盘、更新记忆、归还资源。这跟写传统后端的思路很像。一个长时任务本质上就是一个有状态的流程实例只不过它的状态转移不是完全确定的模型中每一步可能做出不同选择。Harness 的价值就在这种不确定性之上又提供了相对确定的运行框架。我在实际项目里遇到过一个问题任务跑了一半模型突然反复要求读取同一个文件每次读到的内容都一样但它就是不肯往下走。如果只盯着模型调 prompt问题永远解决不了但从 harness 的角度看我可以在执行骨架里加一个工具调用频次检测超过阈值就自动截断当前思考链强制 agent 重新审视目标。这种机制 Anthropic 自己在实践里也很看重——他们管这类能力叫“流程约束”。2.2 会话与沙盒为什么 Agent 需要独立环境Anthropic 的 agent 设计里特别强调 session 和 sandbox。一个长时间运行的任务会对应一个或者多个 sessionsession 里保存着任务上下文、历史消息、工具调用记录。如果任务中间断了重新拉起时可以从最近的 checkpoint 恢复而不是一切归零。这个思路和数据库的 WAL 日志很像先记录意图再执行动作。沙盒则解决的是安全问题。Agent 要执行代码、读写文件、联网访问如果所有东西都直接打在宿主机上一个工具 bug 就可能毁掉整个环境。正确的做法是给每个任务分配独立的容器或虚拟环境权限按需开放。任务结束后直接销毁环境不留下脏数据。我在部署 Agent 时会遵循一条原则宿主机只跑 harnessAgent 的业务动作全部在沙盒里完成。哪怕只是跑一个简单的 Python 脚本也不要直接 subprocess 调用而是把它丢进 sandbox 执行。代价是初始化稍微慢一点但换来的是故障隔离和审计便利这个投入非常值。2.3 长时任务里最容易出错的四个点长时任务最常见的四类错误我逐个遇到过。第一类是外部服务连接不稳定。最典型的就是日志里报unable to connect to anthropic services或者是failed to connect to api.anthropic.com。很多人的第一反应是模型厂商的问题但实际排查下来很多时候是出口网络环境的问题DNS 解析超时、出口 IP 被限流、网关路由配置不正确、TLS 握手被中间设备改写。遇到这种错误正确的排查顺序是先检查网络链路本身再去看代码里的超时设置。第二类是插件加载失败。社区里经常有人报harness failed to load plugins尤其是 DeepSeek Harness 在 Linux 上装插件时报web boot: 2 entries did not activate。这种现象十有八九是配置文件格式不正确、插件依赖的 Python 版本不匹配或者是插件被丢到了错误的目录。Harness 报错信息又常常不完整这时候只能逐个插件试。第三类是权限校验失败。Agent 尝试访问一个它本不该访问的资源被 UUID 权限策略挡掉。这个错误的名字经常很长比如claude doesnt look like an anthropic model: expected a gateway model route看起来像是模型类型不对实际上往往是你配置了网关路由但请求没有正确匹配到指定模型也就是说网关层校验失败了。第四类是任务执行到一半被终止。比如agent execution terminated due to error这种情况要分两种一种是被上游主动 kill 的另一种是内存或 CPU 超限被沙盒回收的。没有好的 harness 层这类问题只能靠人肉重启。3. Google AX 的声明式调度把调度权交给配置3.1 命令式编排的痛点如果从零写一个 Agent 调度器很多人第一反应是直接写 Python一个循环按顺序执行步骤遇到失败就 try-except。这种“命令式编排”在小任务里没问题但业务复杂以后就会失控。比如任务 A 完成后要并行跑 B 和 CD 依赖 B 的结果但不想依赖 C重试策略还希望 B 和 C 不一样。你会在代码里越写越多 if-else最后没人敢改这个模块。Google AX 的声明式调度方向本质上是把“怎么做”和“做什么”分开。你不再用代码描述每个步骤的执行细节而是用一个声明式配置告诉调度器任务之间有怎样的依赖关系、每步的超时和重试策略是多少、资源上限是什么。调度器负责解释这份配置并负责并发控制、失败重试、依赖检查。声明式调度和命令式编排的核心区别在于命令式的控制流是隐藏在代码逻辑里的要看完整段代码才知道会怎么跑声明式的控制流是显式写在配置里的任何人拿一份配置文件就能知道整个任务的拓扑结构。3.2 一个 AX 风格声明式调度的示例我们可以看一下简化后的声明式任务定义核心字段就是节点、依赖、执行参数、护栏策略。比如下面这份 YAML。pipeline: - id: collect_data provider: fetcher timeout: 120s retry: 2 retry_delay: 5s - id: clean_data provider: code_executor depends_on: collect_data timeout: 60s sandbox: true - id: summarize provider: llm depends_on: clean_data timeout: 90s retry: 1 guardrails: - type: output_schema schema_path: ./schemas/summary.json - id: notify provider: webhook depends_on: summarize concurrency: 1这份配置不需要写任何业务代码调度器就能知道先抓数据数据抓完了清洗清洗完再让 LLM 总结最后发通知。每一步都定义了超时和重试clean_data要求必须跑在沙盒里summarize的结果要经过 schema 校验。用 Python 实现一个极简解释器并不复杂。核心就是一个循环不断检查哪些节点满足了依赖条件把它们丢进执行池执行完更新状态。等所有节点都完成或者某个节点失败且重试耗尽就结束。from dataclasses import dataclass, field dataclass class Node: id: str depends_on: list[str] field(default_factorylist) status: str pending # pending | running | done | failed retry: int 0 output: object None class DeclarativeScheduler: def __init__(self, nodes): self.nodes {n.id: n for n in nodes} self.running set() self.failed {} def ready_nodes(self): ready [] for node in self.nodes.values(): if node.status ! pending: continue deps [self.nodes[d] for d in node.depends_on] if all(d.status done for d in deps): ready.append(node) return ready def run(self, execute): while any(n.status pending and n.id not in self.failed for n in self.nodes.values()): for node in self.ready_nodes(): self.running.add(node.id) node.status running try: node.output execute(node.id) node.status done except Exception as exc: self.failed[node.id] exc if not self.running and not self.ready_nodes(): break return {id: n.status for id, n in self.nodes.items()}这个调度器只有几十行但它已经能表达依赖、并行、状态流转。真实生产环境里的 AX 会比这复杂很多会增加资源配额、优先级队列、血缘追踪但核心思想不变把任务拓扑当作数据来看待。3.3 声明式调度如何解决并发和重试很多人在问 “AI Agent 怎么扛并发”。答案不在模型层而在调度层。命令式编排里并发通常要靠开发者手动创建线程池或消息队列非常容易出问题。声明式调度则可以让每个节点声明独立的concurrency配置调度器根据配置去控制同时运行的实例数。比如 LLM API 有限流你可以把模型调用节点的并发上限设为 4其他节点的并发设为 10互不影响。重试策略也能做到精细化。某个步骤如果是幂等的失败后可以直接重试如果非幂等则要做幂等键处理。声明式配置里把retry和retry_delay写清楚调度器就能在正确的时间窗口内自动重试而不是毫无章法地连打。还有一个常被忽略的好处声明式配置天然适合审计和回放。你看到一份配置就知道系统当时应该怎么运行。配合 harness 的日志系统你可以把某次线上故障完整复原这在调试 Agent 这种非确定系统里极其宝贵。4. 手写一个轻量 Agent Harness4.1 设计目标与模块划分聊完理论直接上实操。我建议在动手写第一版 harness 的时候不要一上来就引入重型框架而是先做一个轻量版本。模块划分可以在早期固定下来后面再迭代。我通常把 harness 分成四个模块task 定义、executor、guard 链、observer。task 定义描述任务输入、上下文、期望输出executor 负责执行主循环guard 链是权限和限制的检查点observer 负责记录日志和指标。四个模块之间通过事件解耦这样最容易扩展。代码仓库结构大致是这个样子agent-harness/ ├── harness/ │ ├── task.py │ ├── executor.py │ ├── guards.py │ └── observer.py ├── pipelines/ │ └── demo_pipeline.yaml └── main.py第一版不需要特别复杂关键是两个点让执行过程中的每一步都有 hook让每一个关键状态都能被外部观察到。4.2 核心代码任务执行器与护栏下面这个类是 executor 的核心骨架。它做了几件事把任务分成多个 step每个 step 执行前先过 guards执行后把结果发给 observer捕获异常并按照策略重试。import time import uuid from typing import Callable, Optional class HarnessExecutionError(RuntimeError): pass class Step: def __init__(self, step_id: str, action: Callable, retry: int 2): self.step_id step_id self.action action self.retry retry self.attempts 0 def execute(self, context: dict): while True: try: self.attempts 1 result self.action(context) return result except Exception as exc: if self.attempts self.retry: raise HarnessExecutionError( fstep{self.step_id} failed after {self.attempts} attempts ) from exc time.sleep(2 ** self.attempts) class Harness: def __init__(self, observerNone): self.observer observer self.guards [] def add_guard(self, guard: Callable): self.guards.append(guard) def run(self, steps: list[Step], context: dict): execution_id uuid.uuid4().hex[:8] context[execution_id] execution_id if self.observer: self.observer.on_start(execution_id, context) for step in steps: for guard in self.guards: guard(step, context) output step.execute(context) context[step.step_id] output if self.observer: self.observer.on_step_done(execution_id, step.step_id, output) if self.observer: self.observer.on_finish(execution_id, context) return context这里面的 guard 怎么写我常用的一个 guard 是“受限操作拦截”。比如步骤里调用了file.delete但当前任务上下文没有标记允许删除文件这个 guard 就直接 fail-fast不让 action 执行。def guard_no_destructive_ops(step: Step, context: dict): allowed context.get(allowed_ops, set()) if step.step_id.startswith(delete_) and delete not in allowed: raise PermissionError(foperation not allowed: {step.step_id})有了这个简单的 guard 链路Agent 的能力边界就不是靠模型自觉了而是靠 harness 强制约束。我在生产环境会加很多类似的 guard例如禁止外发 IP 请求、限制单次工具调用输出长度、内存使用告警等。4.3 接上 Anthropic 风格 API 与 AX 声明式调度executor 本身不关心模型是哪家的它只负责执行 steps。真正接模型的地方在 action 里。举个例子一个 step 可以是call_llm它构建一个请求体发送到 Anthropic 风格的 API 端点上。def call_llm(context: dict): client context[client] messages context[messages] resp client.messages.create( modelcontext[model], max_tokenscontext.get(max_tokens, 2048), messagesmessages, ) return resp.content然后把声明式配置解析成 steps。解析 YAML 的代码很简单下面是核心映射逻辑import yaml def parser_pipeline_to_steps(config_path: str, action_registry: dict): raw yaml.safe_load(open(config_path)) steps [] for item in raw[pipeline]: action action_registry[item[provider]] steps.append(Step( step_iditem[id], actionaction, retryitem.get(retry, 0), )) return steps这样配置管配置、代码管代码。想调整任务流程不需要改一行业务代码只改 YAML 文件就够了。4.4 本地模型场景把 Harness 指向自部署服务有个趋势值得注意越来越多团队开始把 Harness 和本地模型服务相结合。比如 DeepSeek Harness 系列很多人折腾的其实是“如何让本地模型具备生产级别的工程能力”。当你把 API base 指向本地服务时最大的变化是网络延迟降低、数据不出内网、可以自定义中间层。我在本地部署实践里得到的经验是本地模型跑 Agent最需要的恰恰是 harness 层。开源模型的调用接口五花八门有的兼容 OpenAI 风格有的自己做了一套有的干脆只支持原生协议。Harness 可以在这些接口之上做一层统一抽象让上层任务流程完全不感知模型的差异。配置里只需指定 providermodel: provider: local base_url: http://127.0.0.1:8000 model_name: deepseek-harness-0.1.5“0.1.5 安装失败”这类问题多数出在依赖冲突上和 harness 设计本身没有关系。先把环境看住再谈能力编排。5. 常见问题与排查技巧实录5.1 连接失败类unable to connect 的排查顺序这类报错几乎人人会遇到。unable to connect to anthropic services、failed to connect to api.anthropic.com是同一个问题的两种表达。遇到后先别慌按下面的顺序排查。第一步看 DNS。在你的运行环境里执行dig api.anthropic.com或者nslookup api.anthropic.com确认域名解析出来的 IP 是正常范围。如果 DNS 超时问题大概率在网络配置。第二步看出口链路。curl -v https://api.anthropic.com/v1/messages观察 TLS 握手和 HTTP 响应。如果这里能通但代码里报连接失败问题多出在代码里的超时配得太短或者用了错误的 base URL。第三步看网关路由。一些企业环境会有专有的 API 网关如果你配置里写的 endpoint 和网关路由不匹配就会出现“模型返回看起来不对劲”的情况比如expected a gateway model route。这种报错不是模型的问题是路由没有对齐。提示超时设置不要一刀切。Agent 的步步调用往往比普通接口慢建议把 connect timeout 和 read timeout 分开设置connect 给 10 秒read 给 60 秒以上。一次性把 read timeout 设太短长时任务必挂。5.2 插件加载失败failed to load pluginsharness failed to load plugins是社区里反馈很多的一个问题。插件系统是 harness 灵活性的来源但也是混乱的源头。我遇到过的失败原因主要有三种。第一种是插件自带依赖和主环境冲突。比如插件要求pydantic2.0但主环境是pydantic2。这种情况最好给插件单独建虚拟环境或者用依赖隔离机制。第二种是插件配置文件的格式问题YAML 里缩进错了一位就可能导致整个插件无法激活。第三种是插件目录权限问题harness 以非 root 身份运行但插件目录没有读权限会静默失败。DeepSeek Harness 在 Linux 上安装失败常见现象是报web boot: 2 entries did not activate。我处理过几次思路就是逐个把插件移到临时目录二分法定位到是哪一条 entry 出了问题再处理它的依赖。插件系统这种地方没有捷径可走。5.3 任务执行被终止与沙盒更新失败agent execution terminated due to error这个报错看起来像是 Agent 逻辑出错了其实很多是环境层的问题。我遇到过一个场景Agent 在沙盒里跑数据清洗数据量突然翻倍内存涨到了沙盒上限整个容器被直接 kill。从业务日志看一切正常没有任何异常堆栈。后来才发现是沙盒资源配额不够。解决方法是把沙盒的 memory 上限调大同时在 harness 里增加事前检查在数据量超过阈值时先扩容。还有一次是报“显示更新agent沙盒”失败翻译成人话就是harness 尝试更新运行时沙盒环境但容器镜像拉不下来或者磁盘空间不够。这种情况要检查两件事镜像仓库能不能访问以及宿主机的/var/lib/containers分区还有多少空间。磁盘满是最容易被忽略的原因。5.4 高并发下 Agent 的扛法队列、限流、幂等Web 高并发靠加机器解决Agent 高并发不能这么玩因为你可能同时往里打几千个长时任务每个任务还都调用模型 API后端马上就会限流。扛并发的关键在 harness 层做好四件事。队列是必须的。所有任务先入队再由 worker 按优先级拉取。限流策略要区分模型 API 的限流和工具调用的限流这两者的 rate limit 往往不一样。幂等是防止重复执行的基石给每个任务生成一个 task_id在外部系统里用 task_id 去重。最后是背压机制当任务堆积超过阈值时harness 要主动拒绝新任务而不是无限排队。我把一些常见报错和排查思路整理成了一个表现象最可能的原因第一排查项unable to connect to anthropic services出口链路或网关配置curl 验证网络链路failed to load plugins依赖冲突或配置格式错误逐个禁用插件定位expected a gateway model route网关路由未匹配检查 base_url 与路由表agent execution terminated due to error沙盒资源耗尽或主动终止查看沙盒退出前日志更新 agent 沙盒失败镜像拉取失败或磁盘不足检查磁盘空间与镜像仓库6. 一点个人体会做 Agent 工程这么久我最大的体会是模型能力会快速拉平而 harness 才是决定项目上限的东西。一副好牌交给一个没有 harness 的系统跑三天就会千疮百孔一套平庸的模型组合配上扎实的 harness反而能稳定输出价值。最近我在新项目里已经开始把声明式调度作为默认模式所有任务先用配置描述再交给执行器去跑。改动业务流程时团队里任何一个人都能 review 那份配置而不是去翻几百行 Python 代码。这个习惯一旦养成你会发现自己不再害怕 Agent 的“不可控”因为你已经把不确定性限制在了一个可观测、可回滚的框架里。如果你还在手动堆代码管理 Agent 任务建议从这个周末开始用一个空目录加一份 YAML 配置搭一个最小 harness 骨架出来。不用等大厂出正式方案先把护城河挖起来再说。