尧图网页设计 
很多人把 Agent 开发当成“大模型 API 调用 提示词拼接”但真正跑过一个月以上项目的人都知道事情远没那么简单。模型会升级、工具会变更、任务会中途失败真正的 Agent 工程重心不在模型本身而在这套“把模型、工具、状态、业务流程焊接在一起”的运行时外壳上。这个外壳业内越来越习惯叫它Harness。无论是 Anthropic 在长时任务设计里强调的“状态持久化与检查点”还是 Google AX 所代表的“声明式调度”本质上都是在回答同一个问题怎么让 Agent 不是一次性的“灵光一现”而是能扛得住真实业务负载的工程系统。这篇文章我想从 Harness 的本质拆起结合 Anthropic 的长时任务设计思路和 Google AX 的声明式调度实践聊聊 Agent 护城河到底在哪里最后给出一份可落地的最小 Harness 实操案例。无论你是刚入行的 AI 应用开发还是已经在跑多 Agent 系统这篇都值得花十分钟看完。1. 拆解 Agent Harness为什么它成了新的护城河1.1 Harness 到底是什么先给个最朴素的定义Harness 是包在 Agent 外面的一整套运行时基础设施。它不负责“思考”但负责让“思考”能落地。你可以把它想象成赛车的外壳和传动系统——引擎大模型固然重要但没有变速箱、悬挂、刹车和稳定系统引擎再猛也上不了赛场。Harness 里通常包含这几部分与模型的交互层处理模型 API 调用、重试、流式返回、上下文打包、多轮对话记忆的维护。工具注册与执行层Agent 要用哪些工具工具的入参出参协议是什么怎么并发调用怎么处理超时。状态管理层记录当前任务进行到哪一步了中间变量存在哪任务中断后能不能恢复。调度与编排层一个 Agent 实例跑哪些步骤多个 Agent 之间怎么协同谁先执行谁后执行。安全与审计层模型输出到底要不要直接执行敏感操作要不要二次确认每一步操作有没有留痕。没有 Harness 的 Agent往往是用一个 Python 脚本把 prompt 发给模型然后把模型返回的 JSON 解析出来再调下一个函数。这能跑通 Demo但一旦任务复杂到需要几十步操作、涉及外部系统状态变更就会迅速失控。1.2 Harness 和 Agent 框架的区别很多人会把 Harness 和 LangChain、LlamaIndex 这类 Agent 框架混在一起。在我看来两者解决的问题层级不一样。框架解决的是“我怎么快速写出一个 Agent”它提供现成的链、代理、记忆组件让你少写样板代码。而 Harness 更偏底层它解决的是“这个 Agent 怎么稳定地跑在业务环境里”它关注的是生命周期、故障恢复、资源调度这些偏运维侧的问题。打个比方框架像是施工队的预制板帮你把墙砌得快一点Harness 则是建筑里的水电管廊和承重结构平时看不见但整栋楼能不能安全运营取决于它。所以你会发现Claude Code、DeepSeek Harness 这类项目虽然以“框架”之名流传但它们真正做的其实是 Harness 的事接管终端交互、管理文件读写工具、维护任务状态、在模型调用失败时自动重试。模型可以随时换但 Harness 是和你的业务深度耦合的这才构成了护城河。1.3 Harness 的三个核心能力基于我自己的实践判断一个 Harness 健不健壮主要看三件事第一状态管理能力。任务中断后能不能从断点恢复而不是从头再来。尤其是长时任务一步宕机全盘重跑的成本是不可接受的。第二工具治理能力。工具不是越多越好而是能不能收拢到一个统一协议下支持权限控制、审计和回滚。谁在什么时候调了哪个工具改了哪些数据都得能追溯。第三调度灵活度。是简单线性执行还是能根据中间结果动态决定下一步需要人工介入时能不能暂停确认后能不能继续这些看似细节决定了系统能不能上生产。这三件事都绕不开模型之外的大量工程细节而这正是 Harness 的价值所在也是每个团队真正需要沉淀的资产。2. Anthropic 长时任务设计教给我们的四件事Anthropic 在 Agent 工程上一直强调长时任务Long-running Tasks的设计这背后其实是一套很成熟的分布式系统思维。我把它拆成四个核心要点你可以在自己的 Harness 里直接对号入座。2.1 状态是长时任务的第一公民短任务可以无状态长时任务不行。一个要跑十分钟甚至几小时的 Agent 任务如果状态只存在内存变量里任何一个进程重启都会前功尽弃。Anthropic 的实践给我的启发是把 Agent 的每一次关键决策、每一个中间结果都显式地写入持久化存储。不光是最终的答案还包括任务进度、已执行的工具调用、当前的上下文摘要。我自己在项目里会用一张task_runs表来记录状态字段大致是字段作用task_id任务唯一标识step_index当前执行步骤statuspending / running / paused / done / failedinput_snapshot当前输入的序列化快照output_snapshot上一步输出的序列化快照error_info最近一次错误信息updated_at最后更新时间每个步骤完成后先写库再进入下一步。这样哪怕进程被杀重新拉起后也能根据 task_id 恢复到中断前的 step_index。2.2 检查点与断点续跑很多长时任务的失败不是模型的问题而是外部系统不稳定——网络抖动、第三方 API 超时、数据库连接断开。Anthropic 的设计里特别强调“容错”而不是“不犯错”。对于 Harness 来说要做两件事定期打检查点Checkpoint。每隔 N 步或者每次调用外部工具后把状态固化下来。实现幂等的断点续跑。重新执行时已经成功的步骤不要重复执行已经产生副作用的工具调用要能识别并跳过。比如一个 Agent 任务要依次调用“创建订单 → 扣减库存 → 发送通知”如果扣减库存成功后网络断掉了重启后不应该再扣一次库存。怎么解决在 Harness 里给每个工具调用维护一个唯一call_id执行前先查一下这个 call_id 是否已经成功过如果成功过就直接拿缓存结果。这一招和数据库事务的“记录先写日志再执行操作”很像本质上就是要为每一次外部副作用留下痕迹。2.3 工具调用要可审计、可回滚长时任务的另一个风险是跑得越久中间产生的副作用越多一旦发现方向错了可能已经改了十几个状态。所以 Anthropic 的设计里非常看重工具调用的可审计性。我的建议是Harness 里必须有一个完整调用链日志格式类似{ tool_name: database.update, call_id: call_8f3a2b, arguments: {table: inventory, id: 1001, delta: -1}, result: {success: true, affected_rows: 1}, timestamp: 2025-04-20T03:12:45Z, task_id: task_42 }这样一旦任务结果不符合预期你可以倒查每一步定位是哪一次工具调用引入了错误。更进一步如果工具支持补偿操作比如“扣减库存”对应的“增加库存”还可以做定向回滚。2.4 上下文管理和记忆分层长时任务最头疼的问题之一是上下文窗口不够用。Anthropic 给出的方向不是“拼命加 token”而是选择性遗忘和分层记忆。Harness 里可以维护三部分记忆工作记忆当前步骤必需的上下文比如正在处理的文件内容只保留当前相关片段。项目记忆任务开始以来沉淀下来的关键结论、已选方案、用户偏好用摘要形式保存。长期记忆跨任务复用的知识库比如企业内部的代码规范、数据库表结构、历史问题解法。每次喂给模型的 prompt由 Harness 负责从这三层里组装出来而不是把历史全量拼进去。这样任务无论跑多久上下文都能控制在一个合理范围内。3. Google AX 的声明式调度把 Agent 当成基础设施来管如果说 Anthropic 的长时任务设计解决的是“一个 Agent 怎么跑得久”那 Google AX 的声明式调度解决的则是“一堆 Agent 怎么编排得稳”。3.1 声明式 vs 命令式传统命令式编排里你要写一堆if...else和循环描述“先做 A再做 B如果 C 则做 D”。流程一变代码就得跟着改而且很难看得出来任务全貌。声明式调度则相反你只需要描述“最终想要什么状态”具体怎么做由调度器决定。比如 Kubernetes 里你写一个 Deployment YAML声明“我要三个副本”K8s 就会自动去创建和维持这 3 个副本中间某个副本挂了它会自动拉起。Google AX 的调度模型本质上是把 K8s 这套声明式理念搬到了 Agent 领域。你不再写“调用 A 再调用 B”的脚本而是声明一个任务 DAG有向无环图定义节点依赖、重试策略、并发上限调度器负责执行并保证最终收敛到目标状态。3.2 AX 调度模型的核心字段用 YAML 描述一个 Agent 任务大致会是这样apiVersion: ax/v1 kind: AgentWorkflow metadata: name: order-service-agent spec: steps: - name: parse_order agent: order_parser dependsOn: [] retry: maxAttempts: 3 backoff: exponential timeout: 30s - name: check_stock agent: inventory_checker dependsOn: [parse_order] timeout: 15s - name: confirm_order agent: order_confirmator dependsOn: [check_stock] requiresApproval: true几个字段我实际用下来觉得特别关键dependsOn声明依赖关系调度器会自动安排执行顺序。retry失败后的重试策略指数退避是标配。timeout防止某个 Agent 卡死拖垮整个流程。requiresApproval敏感动作执行前需要人工确认。声明式的好处是你可以随时修改 YAML 里的参数比如把并发数从 1 调到 5调度器会自动按新配置重新调度不需要改一行代码。这对运营中的 Agent 系统来说太重要了。3.3 与 Harness 的配合方式Google AX 管的是“任务怎么编排”Harness 管的是“单个任务怎么执行”两者天然是分层配合关系。我设想的简化架构是触发事件 → AX 调度器声明式 DAG 编排 → 分配任务到 Harness 实例 → Harness 执行步骤 → 反馈结果给 AX → 继续下一节点当 Harness 实例执行失败时AX 调度器可以根据重试策略决定是重启同一实例、换一个实例还是终止整个工作流。Harness 里的检查点数据就是调度器做决策的重要依据。这套设计一旦落地Agent 系统就从“一个人写死流程”变成了“平台自动运维”这也是为什么我说 Harness 和声明式调度是两条腿缺一不可。3.4 常见误区别把声明式调度当成万能药声明式调度也不是银弹。如果任务本身是强状态交互、无法拆成 DAG硬上声明式反而会复杂化。比如一个 Agent 需要“根据用户实时反馈动态调整策略”这属于循环依赖不适合用静态 DAG 描述。这种场景还是得在 Harness 里写事件驱动逻辑或者把动态决策封装成某个 Agent 节点内部的行为而不是强求整个流程声明化。我踩过的坑是一开始想把所有流程都改造成 YAML结果发现团队光维护 YAML 文档就耗费了大量时间而且因为过度抽象排查问题时还得在代码和配置之间来回跳。合理的边界是稳定且线性的流程优先声明式动态且探索性的逻辑保留在 Harness 代码里。4. 实操从零搭一个最小可运行的长时 Agent Harness理论说再多不如跑一个最小系统。下面这套代码足够你在本地验证长时任务、检查点恢复这些核心机制。4.1 设计目标我们要做一个极简 Harness核心能力包括支持多步骤任务的定义。每个步骤执行前后自动保存状态。模拟任务中途崩溃重启后能从断点恢复。工具调用统一走注册表记录 audit log。4.2 项目结构minimal_harness/ ├── harness.py # 核心运行时 ├── tasks.py # 任务定义 ├── state_store.py # 状态持久化本地 JSON └── run.py # 启动入口这里用 JSON 文件模拟持久化存储实际生产环境可以换成 Redis、PostgreSQL 或者云上的对象存储。4.3 核心代码实现先写状态存储这是长时任务的底座# state_store.py import json import os import time class StateStore: def __init__(self, path./state): self.path path os.makedirs(path, exist_okTrue) def _task_file(self, task_id): return os.path.join(self.path, f{task_id}.json) def save(self, task_id, step_index, status, snapshot): data { task_id: task_id, step_index: step_index, status: status, snapshot: snapshot, updated_at: time.time() } with open(self._task_file(task_id), w) as f: json.dump(data, f, ensure_asciiFalse, indent2) def load(self, task_id): try: with open(self._task_file(task_id), r) as f: return json.load(f) except FileNotFoundError: return None然后是 Harness 核心运行时。它维护工具注册表并在步骤执行前后写入检查点# harness.py import traceback from state_store import StateStore class Harness: def __init__(self, state_store: StateStore): self.tools {} self.state_store state_store self.audit_log [] def register_tool(self, name, func): self.tools[name] func def call_tool(self, name, arguments, call_id): if name not in self.tools: raise ValueError(fTool {name} not registered) # 幂等检查同一 call_id 已成功则直接返回缓存 state self.state_store.load(ftool_{call_id}) if state and state[status] success: self.audit_log.append({call_id: call_id, skipped: True}) return state[result] try: result self.tools[name](**arguments) except Exception as e: self.state_store.save( ftool_{call_id}, 0, failed, {error: str(e), trace: traceback.format_exc()} ) raise # 执行成功后写审计日志和状态 self.state_store.save(ftool_{call_id}, 0, success, result) self.audit_log.append({ call_id: call_id, tool: name, args: arguments, result: result, ts: __import__(time).time() }) return result def run_task(self, task): task_id task[task_id] state self.state_store.load(task_id) start_index 0 if state: start_index state[step_index] 1 if state[status] done: print(f[Harness] Task {task_id} already completed, skip.) return state[snapshot] print(f[Harness] Recovered task {task_id} from step {start_index}) steps task[steps] snapshot state[snapshot] if state else {} for i in range(start_index, len(steps)): step steps[i] step_name step[name] tool_name step[tool] arguments step.get(arguments, {}) call_id f{task_id}_{step_name}_{i} print(f[Harness] Running step {i}: {step_name}) # 用户态 Hookstep_before if task.get(step_before): task[step_before](step, snapshot) result self.call_tool(tool_name, arguments, call_id) snapshot[last_output] result # 用户态 Hookstep_after if task.get(step_after): task[step_after](step, snapshot) self.state_store.save(task_id, i, running, snapshot) self.state_store.save(task_id, len(steps) - 1, done, snapshot) print(f[Harness] Task {task_id} completed.) return snapshot任务定义文件里用 Python 函数模拟外部工具并用状态存储模拟“任务跑一半崩溃”# tasks.py import time from harness import Harness from state_store import StateStore store StateStore() harness Harness(store) # 注册两个外部工具 harness.register_tool # 不可这么写走单独注册逻辑下面简化 def _fake(): pass这里需要注意上面代码里harness.register_tool不能直接装饰我会改成显式harness.register_tool(...)def create_order(order_id, amount): print(f[Tool] create_order({order_id}, {amount})) return {status: created, order_id: order_id} def deduct_stock(product_id, delta): print(f[Tool] deduct_stock({product_id}, {delta})) return {status: deducted, product_id: product_id, delta: delta} def send_notification(user_id, message): print(f[Tool] send_notification({user_id}, {message})) return {status: sent, user_id: user_id} if __name__ __main__: h Harness(store) h.register_tool(create_order, create_order) h.register_tool(deduct_stock, deduct_stock) h.register_tool(send_notification, send_notification) task { task_id: demo_task_001, steps: [ {name: step1_create, tool: create_order, arguments: {order_id: A1001, amount: 99}}, {name: step2_deduct, tool: deduct_stock, arguments: {product_id: SKU-X, delta: 1}}, {name: step3_notify, tool: send_notification, arguments: {user_id: user_42, message: order confirmed}} ] } # 模拟第一次执行第2步后进程崩溃 # 这里为了演示在真实运行里直接跑 h.run_task(task) # 为了模拟崩溃我们手动在第二步执行后退出 # 实际演示时 result h.run_task(task) print(Final result:, result)如果你想要更“真实”的崩溃恢复演示可以在任务里加一个 step_after hook在第二个步骤执行后直接os._exit(1)def crash_after_step(step, snapshot): if step[name] step2_deduct: print([Debug] Simulating crash...) import os os._exit(1) task[step_after] crash_after_step这样第一次运行会在第 2 步之后崩溃第二次再运行h.run_task(task)时Harness 会从恢复点step_index2 之后的步骤即 step3继续执行而不是从头跑一遍。4.4 状态恢复流程演示用上面的代码跑一个完整流程你会看到这样的输出第一次运行[Harness] Running step 0: step1_create [Tool] create_order(A1001, 99) [Harness] Running step 1: step2_deduct [Tool] deduct_stock(SKU-X, 1) [Debug] Simulating crash...第二次运行[Harness] Recovered task demo_task_001 from step 2 [Harness] Running step 2: step3_notify [Tool] send_notification(user_42, order confirmed) [Harness] Task demo_task_001 completed.就这么简单核心机制已经可用。实际生产里你只需要把 StateStore 换成 Redis/数据库实现把工具调用换成真实的外部 API把任务定义换成 YAML 配置再叠加 Google AX 那种声明式调度层就是一个五脏俱全的 Agent 运行时。4.5 参数和并发设计建议上面这套最小实现是单线程串行执行的。生产环境的 Harness 一定要考虑并发问题我建议几个参数提前设计好最大并发数给每个 Agent 实例限制max_concurrency避免一次性拉起太多工具调用打爆下游系统。超时时间每个工具调用都要有timeout我一般设 30 秒重试两次超过就标记失败并触发替代路径。速率限制针对外部 API建议在 Harness 里做令牌桶限流而不是依赖调用方自觉。这些参数都可以用声明式配置暴露出来比如 YAML 里的global: {maxConcurrency: 5, defaultTimeout: 30}这样运营同学调参时不需要改代码。5. 常见问题与排查技巧实录写这套 Harness 和调度系统的过程中我踩了不少坑挑几个典型的分享给你省得你再走一遍弯路。5.1 长时任务中途挂掉恢复后重复执行了工具这个是最常见的坑。恢复逻辑如果只按 step_index 推进很可能把“上一次已经执行成功但还没来得及保存状态”的步骤重新执行一遍。解决办法就是前面说的 call_id 幂等机制。每个工具调用都带上全局唯一的 call_id执行前先查状态库。我曾经在一个订单系统里吃过亏重复扣了两次库存从那以后所有写操作工具都强制要求幂等键。排查技巧看审计日志里有没有skipped: true的记录如果有说明幂等机制生效了如果没有说明你的工具还没接入幂等检查赶紧补。5.2 上下文越跑越长模型开始胡说八道长时任务跑到后面经常出现“早期记忆丢失”和“上下文污染”问题。我的排查办法是在 Harness 里给每个步骤记录 token 消耗超过设定阈值就主动触发摘要压缩。具体做法是当累计 token 达到窗口的 70% 时把早期的对话历史用一次独立的 summarize 调用压缩成 500 字以内的摘要替换掉原始内容。这一步看似“丢信息”但实际效果远好于硬塞一堆无关历史导致模型注意力涣散。5.3 调度流程卡死节点既不成功也不失败声明式调度最怕的就是事件“悬空”——某个 Agent 节点跑了两小时还在 running既不超时也不退出。我的建议是三条所有 Agent 节点必须有显式timeout不要依赖默认值。调度器要定期扫描 running 状态的任务超过心跳时间的强制标记失败。每次状态更新都要带updated_at作为心跳依据。我在代码里的StateStore已经保存了updated_at生产环境直接用这个字段来判断是否失联。5.4 YAML 配置改错导致大规模误调度声明式配置虽然方便但权限控制一定要做好。我踩过最惨的一次运营同事改并发参数时不小心把maxConcurrency: 1写成了maxConcurreny: 100拼写错误结果配置校验没过但系统没有拦住把下游数据库压垮了。所以 Harness 在加载声明式配置时一定要做严格的 Schema 校验未知字段直接报错而不是忽略。可以用 JSON Schema 或者 Pydantic 这类工具做看起来是小事关键时刻能救命。5.5 Agent 卡在模型调用上一直报连接错误热词里出现过的unable to connect to anthropic services这类问题在 Harness 里也很常见。核心处理思路是重试要有指数退避且必须与任务恢复机制联动。我的做法是在 Harness 里对模型调用做三层重试第一层立即重试一次解决瞬时网络抖动。第二层等待 1 秒、2 秒、4 秒的指数退避重试最多 5 次。第三层重试全部失败后把任务标记为 paused而不是 failed。paused 状态的任务可以等网络恢复后手动或者由调度器自动恢复。这样做的好处是一次模型 API 的长时间故障不会导致整个长时任务回滚最多是暂停等待。6. 从护城河到工程实践一点个人体会回到标题那句话Harness 是 Agent 的新护城河。我在实际项目里越来越认同这个判断。模型的能力迭代很快今天用这个模型明天可能就换另一个更强更便宜的。但你的业务流程、你的工具链、你的状态设计、你的调度策略全都沉淀在 Harness 这一层。换个模型prompt 改一改就能跑但换 Harness等于把所有业务逻辑和运维经验推倒重来。这就是护城河的真正含义。最后再分享一个小技巧不要让 Harness 变成第二套业务系统。我见过有人把各种业务规则硬编码进 Agent 运行时结果 Harness 里塞满了 if-else比业务代码还复杂。好的做法是Harness 只做通用能力——状态管理、工具调用、审计日志、调度执行具体业务规则尽量下沉到工具函数和配置里。这样你的 Harness 才能保持轻盈成为真正可以复用的平台层。Agent 这块地模型只是入场券Harness 才是真正的基建。早一天把基建打牢后面就早一天省心。 Agent Harness：大模型应用的新护城河与长时任务工程实践
很多人把 Agent 开发当成“大模型 API 调用 提示词拼接”但真正跑过一个月以上项目的人都知道事情远没那么简单。模型会升级、工具会变更、任务会中途失败真正的 Agent 工程重心不在模型本身而在这套“把模型、工具、状态、业务流程焊接在一起”的运行时外壳上。这个外壳业内越来越习惯叫它Harness。无论是 Anthropic 在长时任务设计里强调的“状态持久化与检查点”还是 Google AX 所代表的“声明式调度”本质上都是在回答同一个问题怎么让 Agent 不是一次性的“灵光一现”而是能扛得住真实业务负载的工程系统。这篇文章我想从 Harness 的本质拆起结合 Anthropic 的长时任务设计思路和 Google AX 的声明式调度实践聊聊 Agent 护城河到底在哪里最后给出一份可落地的最小 Harness 实操案例。无论你是刚入行的 AI 应用开发还是已经在跑多 Agent 系统这篇都值得花十分钟看完。1. 拆解 Agent Harness为什么它成了新的护城河1.1 Harness 到底是什么先给个最朴素的定义Harness 是包在 Agent 外面的一整套运行时基础设施。它不负责“思考”但负责让“思考”能落地。你可以把它想象成赛车的外壳和传动系统——引擎大模型固然重要但没有变速箱、悬挂、刹车和稳定系统引擎再猛也上不了赛场。Harness 里通常包含这几部分与模型的交互层处理模型 API 调用、重试、流式返回、上下文打包、多轮对话记忆的维护。工具注册与执行层Agent 要用哪些工具工具的入参出参协议是什么怎么并发调用怎么处理超时。状态管理层记录当前任务进行到哪一步了中间变量存在哪任务中断后能不能恢复。调度与编排层一个 Agent 实例跑哪些步骤多个 Agent 之间怎么协同谁先执行谁后执行。安全与审计层模型输出到底要不要直接执行敏感操作要不要二次确认每一步操作有没有留痕。没有 Harness 的 Agent往往是用一个 Python 脚本把 prompt 发给模型然后把模型返回的 JSON 解析出来再调下一个函数。这能跑通 Demo但一旦任务复杂到需要几十步操作、涉及外部系统状态变更就会迅速失控。1.2 Harness 和 Agent 框架的区别很多人会把 Harness 和 LangChain、LlamaIndex 这类 Agent 框架混在一起。在我看来两者解决的问题层级不一样。框架解决的是“我怎么快速写出一个 Agent”它提供现成的链、代理、记忆组件让你少写样板代码。而 Harness 更偏底层它解决的是“这个 Agent 怎么稳定地跑在业务环境里”它关注的是生命周期、故障恢复、资源调度这些偏运维侧的问题。打个比方框架像是施工队的预制板帮你把墙砌得快一点Harness 则是建筑里的水电管廊和承重结构平时看不见但整栋楼能不能安全运营取决于它。所以你会发现Claude Code、DeepSeek Harness 这类项目虽然以“框架”之名流传但它们真正做的其实是 Harness 的事接管终端交互、管理文件读写工具、维护任务状态、在模型调用失败时自动重试。模型可以随时换但 Harness 是和你的业务深度耦合的这才构成了护城河。1.3 Harness 的三个核心能力基于我自己的实践判断一个 Harness 健不健壮主要看三件事第一状态管理能力。任务中断后能不能从断点恢复而不是从头再来。尤其是长时任务一步宕机全盘重跑的成本是不可接受的。第二工具治理能力。工具不是越多越好而是能不能收拢到一个统一协议下支持权限控制、审计和回滚。谁在什么时候调了哪个工具改了哪些数据都得能追溯。第三调度灵活度。是简单线性执行还是能根据中间结果动态决定下一步需要人工介入时能不能暂停确认后能不能继续这些看似细节决定了系统能不能上生产。这三件事都绕不开模型之外的大量工程细节而这正是 Harness 的价值所在也是每个团队真正需要沉淀的资产。2. Anthropic 长时任务设计教给我们的四件事Anthropic 在 Agent 工程上一直强调长时任务Long-running Tasks的设计这背后其实是一套很成熟的分布式系统思维。我把它拆成四个核心要点你可以在自己的 Harness 里直接对号入座。2.1 状态是长时任务的第一公民短任务可以无状态长时任务不行。一个要跑十分钟甚至几小时的 Agent 任务如果状态只存在内存变量里任何一个进程重启都会前功尽弃。Anthropic 的实践给我的启发是把 Agent 的每一次关键决策、每一个中间结果都显式地写入持久化存储。不光是最终的答案还包括任务进度、已执行的工具调用、当前的上下文摘要。我自己在项目里会用一张task_runs表来记录状态字段大致是字段作用task_id任务唯一标识step_index当前执行步骤statuspending / running / paused / done / failedinput_snapshot当前输入的序列化快照output_snapshot上一步输出的序列化快照error_info最近一次错误信息updated_at最后更新时间每个步骤完成后先写库再进入下一步。这样哪怕进程被杀重新拉起后也能根据 task_id 恢复到中断前的 step_index。2.2 检查点与断点续跑很多长时任务的失败不是模型的问题而是外部系统不稳定——网络抖动、第三方 API 超时、数据库连接断开。Anthropic 的设计里特别强调“容错”而不是“不犯错”。对于 Harness 来说要做两件事定期打检查点Checkpoint。每隔 N 步或者每次调用外部工具后把状态固化下来。实现幂等的断点续跑。重新执行时已经成功的步骤不要重复执行已经产生副作用的工具调用要能识别并跳过。比如一个 Agent 任务要依次调用“创建订单 → 扣减库存 → 发送通知”如果扣减库存成功后网络断掉了重启后不应该再扣一次库存。怎么解决在 Harness 里给每个工具调用维护一个唯一call_id执行前先查一下这个 call_id 是否已经成功过如果成功过就直接拿缓存结果。这一招和数据库事务的“记录先写日志再执行操作”很像本质上就是要为每一次外部副作用留下痕迹。2.3 工具调用要可审计、可回滚长时任务的另一个风险是跑得越久中间产生的副作用越多一旦发现方向错了可能已经改了十几个状态。所以 Anthropic 的设计里非常看重工具调用的可审计性。我的建议是Harness 里必须有一个完整调用链日志格式类似{ tool_name: database.update, call_id: call_8f3a2b, arguments: {table: inventory, id: 1001, delta: -1}, result: {success: true, affected_rows: 1}, timestamp: 2025-04-20T03:12:45Z, task_id: task_42 }这样一旦任务结果不符合预期你可以倒查每一步定位是哪一次工具调用引入了错误。更进一步如果工具支持补偿操作比如“扣减库存”对应的“增加库存”还可以做定向回滚。2.4 上下文管理和记忆分层长时任务最头疼的问题之一是上下文窗口不够用。Anthropic 给出的方向不是“拼命加 token”而是选择性遗忘和分层记忆。Harness 里可以维护三部分记忆工作记忆当前步骤必需的上下文比如正在处理的文件内容只保留当前相关片段。项目记忆任务开始以来沉淀下来的关键结论、已选方案、用户偏好用摘要形式保存。长期记忆跨任务复用的知识库比如企业内部的代码规范、数据库表结构、历史问题解法。每次喂给模型的 prompt由 Harness 负责从这三层里组装出来而不是把历史全量拼进去。这样任务无论跑多久上下文都能控制在一个合理范围内。3. Google AX 的声明式调度把 Agent 当成基础设施来管如果说 Anthropic 的长时任务设计解决的是“一个 Agent 怎么跑得久”那 Google AX 的声明式调度解决的则是“一堆 Agent 怎么编排得稳”。3.1 声明式 vs 命令式传统命令式编排里你要写一堆if...else和循环描述“先做 A再做 B如果 C 则做 D”。流程一变代码就得跟着改而且很难看得出来任务全貌。声明式调度则相反你只需要描述“最终想要什么状态”具体怎么做由调度器决定。比如 Kubernetes 里你写一个 Deployment YAML声明“我要三个副本”K8s 就会自动去创建和维持这 3 个副本中间某个副本挂了它会自动拉起。Google AX 的调度模型本质上是把 K8s 这套声明式理念搬到了 Agent 领域。你不再写“调用 A 再调用 B”的脚本而是声明一个任务 DAG有向无环图定义节点依赖、重试策略、并发上限调度器负责执行并保证最终收敛到目标状态。3.2 AX 调度模型的核心字段用 YAML 描述一个 Agent 任务大致会是这样apiVersion: ax/v1 kind: AgentWorkflow metadata: name: order-service-agent spec: steps: - name: parse_order agent: order_parser dependsOn: [] retry: maxAttempts: 3 backoff: exponential timeout: 30s - name: check_stock agent: inventory_checker dependsOn: [parse_order] timeout: 15s - name: confirm_order agent: order_confirmator dependsOn: [check_stock] requiresApproval: true几个字段我实际用下来觉得特别关键dependsOn声明依赖关系调度器会自动安排执行顺序。retry失败后的重试策略指数退避是标配。timeout防止某个 Agent 卡死拖垮整个流程。requiresApproval敏感动作执行前需要人工确认。声明式的好处是你可以随时修改 YAML 里的参数比如把并发数从 1 调到 5调度器会自动按新配置重新调度不需要改一行代码。这对运营中的 Agent 系统来说太重要了。3.3 与 Harness 的配合方式Google AX 管的是“任务怎么编排”Harness 管的是“单个任务怎么执行”两者天然是分层配合关系。我设想的简化架构是触发事件 → AX 调度器声明式 DAG 编排 → 分配任务到 Harness 实例 → Harness 执行步骤 → 反馈结果给 AX → 继续下一节点当 Harness 实例执行失败时AX 调度器可以根据重试策略决定是重启同一实例、换一个实例还是终止整个工作流。Harness 里的检查点数据就是调度器做决策的重要依据。这套设计一旦落地Agent 系统就从“一个人写死流程”变成了“平台自动运维”这也是为什么我说 Harness 和声明式调度是两条腿缺一不可。3.4 常见误区别把声明式调度当成万能药声明式调度也不是银弹。如果任务本身是强状态交互、无法拆成 DAG硬上声明式反而会复杂化。比如一个 Agent 需要“根据用户实时反馈动态调整策略”这属于循环依赖不适合用静态 DAG 描述。这种场景还是得在 Harness 里写事件驱动逻辑或者把动态决策封装成某个 Agent 节点内部的行为而不是强求整个流程声明化。我踩过的坑是一开始想把所有流程都改造成 YAML结果发现团队光维护 YAML 文档就耗费了大量时间而且因为过度抽象排查问题时还得在代码和配置之间来回跳。合理的边界是稳定且线性的流程优先声明式动态且探索性的逻辑保留在 Harness 代码里。4. 实操从零搭一个最小可运行的长时 Agent Harness理论说再多不如跑一个最小系统。下面这套代码足够你在本地验证长时任务、检查点恢复这些核心机制。4.1 设计目标我们要做一个极简 Harness核心能力包括支持多步骤任务的定义。每个步骤执行前后自动保存状态。模拟任务中途崩溃重启后能从断点恢复。工具调用统一走注册表记录 audit log。4.2 项目结构minimal_harness/ ├── harness.py # 核心运行时 ├── tasks.py # 任务定义 ├── state_store.py # 状态持久化本地 JSON └── run.py # 启动入口这里用 JSON 文件模拟持久化存储实际生产环境可以换成 Redis、PostgreSQL 或者云上的对象存储。4.3 核心代码实现先写状态存储这是长时任务的底座# state_store.py import json import os import time class StateStore: def __init__(self, path./state): self.path path os.makedirs(path, exist_okTrue) def _task_file(self, task_id): return os.path.join(self.path, f{task_id}.json) def save(self, task_id, step_index, status, snapshot): data { task_id: task_id, step_index: step_index, status: status, snapshot: snapshot, updated_at: time.time() } with open(self._task_file(task_id), w) as f: json.dump(data, f, ensure_asciiFalse, indent2) def load(self, task_id): try: with open(self._task_file(task_id), r) as f: return json.load(f) except FileNotFoundError: return None然后是 Harness 核心运行时。它维护工具注册表并在步骤执行前后写入检查点# harness.py import traceback from state_store import StateStore class Harness: def __init__(self, state_store: StateStore): self.tools {} self.state_store state_store self.audit_log [] def register_tool(self, name, func): self.tools[name] func def call_tool(self, name, arguments, call_id): if name not in self.tools: raise ValueError(fTool {name} not registered) # 幂等检查同一 call_id 已成功则直接返回缓存 state self.state_store.load(ftool_{call_id}) if state and state[status] success: self.audit_log.append({call_id: call_id, skipped: True}) return state[result] try: result self.tools[name](**arguments) except Exception as e: self.state_store.save( ftool_{call_id}, 0, failed, {error: str(e), trace: traceback.format_exc()} ) raise # 执行成功后写审计日志和状态 self.state_store.save(ftool_{call_id}, 0, success, result) self.audit_log.append({ call_id: call_id, tool: name, args: arguments, result: result, ts: __import__(time).time() }) return result def run_task(self, task): task_id task[task_id] state self.state_store.load(task_id) start_index 0 if state: start_index state[step_index] 1 if state[status] done: print(f[Harness] Task {task_id} already completed, skip.) return state[snapshot] print(f[Harness] Recovered task {task_id} from step {start_index}) steps task[steps] snapshot state[snapshot] if state else {} for i in range(start_index, len(steps)): step steps[i] step_name step[name] tool_name step[tool] arguments step.get(arguments, {}) call_id f{task_id}_{step_name}_{i} print(f[Harness] Running step {i}: {step_name}) # 用户态 Hookstep_before if task.get(step_before): task[step_before](step, snapshot) result self.call_tool(tool_name, arguments, call_id) snapshot[last_output] result # 用户态 Hookstep_after if task.get(step_after): task[step_after](step, snapshot) self.state_store.save(task_id, i, running, snapshot) self.state_store.save(task_id, len(steps) - 1, done, snapshot) print(f[Harness] Task {task_id} completed.) return snapshot任务定义文件里用 Python 函数模拟外部工具并用状态存储模拟“任务跑一半崩溃”# tasks.py import time from harness import Harness from state_store import StateStore store StateStore() harness Harness(store) # 注册两个外部工具 harness.register_tool # 不可这么写走单独注册逻辑下面简化 def _fake(): pass这里需要注意上面代码里harness.register_tool不能直接装饰我会改成显式harness.register_tool(...)def create_order(order_id, amount): print(f[Tool] create_order({order_id}, {amount})) return {status: created, order_id: order_id} def deduct_stock(product_id, delta): print(f[Tool] deduct_stock({product_id}, {delta})) return {status: deducted, product_id: product_id, delta: delta} def send_notification(user_id, message): print(f[Tool] send_notification({user_id}, {message})) return {status: sent, user_id: user_id} if __name__ __main__: h Harness(store) h.register_tool(create_order, create_order) h.register_tool(deduct_stock, deduct_stock) h.register_tool(send_notification, send_notification) task { task_id: demo_task_001, steps: [ {name: step1_create, tool: create_order, arguments: {order_id: A1001, amount: 99}}, {name: step2_deduct, tool: deduct_stock, arguments: {product_id: SKU-X, delta: 1}}, {name: step3_notify, tool: send_notification, arguments: {user_id: user_42, message: order confirmed}} ] } # 模拟第一次执行第2步后进程崩溃 # 这里为了演示在真实运行里直接跑 h.run_task(task) # 为了模拟崩溃我们手动在第二步执行后退出 # 实际演示时 result h.run_task(task) print(Final result:, result)如果你想要更“真实”的崩溃恢复演示可以在任务里加一个 step_after hook在第二个步骤执行后直接os._exit(1)def crash_after_step(step, snapshot): if step[name] step2_deduct: print([Debug] Simulating crash...) import os os._exit(1) task[step_after] crash_after_step这样第一次运行会在第 2 步之后崩溃第二次再运行h.run_task(task)时Harness 会从恢复点step_index2 之后的步骤即 step3继续执行而不是从头跑一遍。4.4 状态恢复流程演示用上面的代码跑一个完整流程你会看到这样的输出第一次运行[Harness] Running step 0: step1_create [Tool] create_order(A1001, 99) [Harness] Running step 1: step2_deduct [Tool] deduct_stock(SKU-X, 1) [Debug] Simulating crash...第二次运行[Harness] Recovered task demo_task_001 from step 2 [Harness] Running step 2: step3_notify [Tool] send_notification(user_42, order confirmed) [Harness] Task demo_task_001 completed.就这么简单核心机制已经可用。实际生产里你只需要把 StateStore 换成 Redis/数据库实现把工具调用换成真实的外部 API把任务定义换成 YAML 配置再叠加 Google AX 那种声明式调度层就是一个五脏俱全的 Agent 运行时。4.5 参数和并发设计建议上面这套最小实现是单线程串行执行的。生产环境的 Harness 一定要考虑并发问题我建议几个参数提前设计好最大并发数给每个 Agent 实例限制max_concurrency避免一次性拉起太多工具调用打爆下游系统。超时时间每个工具调用都要有timeout我一般设 30 秒重试两次超过就标记失败并触发替代路径。速率限制针对外部 API建议在 Harness 里做令牌桶限流而不是依赖调用方自觉。这些参数都可以用声明式配置暴露出来比如 YAML 里的global: {maxConcurrency: 5, defaultTimeout: 30}这样运营同学调参时不需要改代码。5. 常见问题与排查技巧实录写这套 Harness 和调度系统的过程中我踩了不少坑挑几个典型的分享给你省得你再走一遍弯路。5.1 长时任务中途挂掉恢复后重复执行了工具这个是最常见的坑。恢复逻辑如果只按 step_index 推进很可能把“上一次已经执行成功但还没来得及保存状态”的步骤重新执行一遍。解决办法就是前面说的 call_id 幂等机制。每个工具调用都带上全局唯一的 call_id执行前先查状态库。我曾经在一个订单系统里吃过亏重复扣了两次库存从那以后所有写操作工具都强制要求幂等键。排查技巧看审计日志里有没有skipped: true的记录如果有说明幂等机制生效了如果没有说明你的工具还没接入幂等检查赶紧补。5.2 上下文越跑越长模型开始胡说八道长时任务跑到后面经常出现“早期记忆丢失”和“上下文污染”问题。我的排查办法是在 Harness 里给每个步骤记录 token 消耗超过设定阈值就主动触发摘要压缩。具体做法是当累计 token 达到窗口的 70% 时把早期的对话历史用一次独立的 summarize 调用压缩成 500 字以内的摘要替换掉原始内容。这一步看似“丢信息”但实际效果远好于硬塞一堆无关历史导致模型注意力涣散。5.3 调度流程卡死节点既不成功也不失败声明式调度最怕的就是事件“悬空”——某个 Agent 节点跑了两小时还在 running既不超时也不退出。我的建议是三条所有 Agent 节点必须有显式timeout不要依赖默认值。调度器要定期扫描 running 状态的任务超过心跳时间的强制标记失败。每次状态更新都要带updated_at作为心跳依据。我在代码里的StateStore已经保存了updated_at生产环境直接用这个字段来判断是否失联。5.4 YAML 配置改错导致大规模误调度声明式配置虽然方便但权限控制一定要做好。我踩过最惨的一次运营同事改并发参数时不小心把maxConcurrency: 1写成了maxConcurreny: 100拼写错误结果配置校验没过但系统没有拦住把下游数据库压垮了。所以 Harness 在加载声明式配置时一定要做严格的 Schema 校验未知字段直接报错而不是忽略。可以用 JSON Schema 或者 Pydantic 这类工具做看起来是小事关键时刻能救命。5.5 Agent 卡在模型调用上一直报连接错误热词里出现过的unable to connect to anthropic services这类问题在 Harness 里也很常见。核心处理思路是重试要有指数退避且必须与任务恢复机制联动。我的做法是在 Harness 里对模型调用做三层重试第一层立即重试一次解决瞬时网络抖动。第二层等待 1 秒、2 秒、4 秒的指数退避重试最多 5 次。第三层重试全部失败后把任务标记为 paused而不是 failed。paused 状态的任务可以等网络恢复后手动或者由调度器自动恢复。这样做的好处是一次模型 API 的长时间故障不会导致整个长时任务回滚最多是暂停等待。6. 从护城河到工程实践一点个人体会回到标题那句话Harness 是 Agent 的新护城河。我在实际项目里越来越认同这个判断。模型的能力迭代很快今天用这个模型明天可能就换另一个更强更便宜的。但你的业务流程、你的工具链、你的状态设计、你的调度策略全都沉淀在 Harness 这一层。换个模型prompt 改一改就能跑但换 Harness等于把所有业务逻辑和运维经验推倒重来。这就是护城河的真正含义。最后再分享一个小技巧不要让 Harness 变成第二套业务系统。我见过有人把各种业务规则硬编码进 Agent 运行时结果 Harness 里塞满了 if-else比业务代码还复杂。好的做法是Harness 只做通用能力——状态管理、工具调用、审计日志、调度执行具体业务规则尽量下沉到工具函数和配置里。这样你的 Harness 才能保持轻盈成为真正可以复用的平台层。Agent 这块地模型只是入场券Harness 才是真正的基建。早一天把基建打牢后面就早一天省心。