1. 长任务为什么总是翻车从Agent的运行时困境说起做过Agent开发的人都有一个共同体会让Agent跑一个短任务比如查个天气、写段摘要、做个简单计算基本都能应付。但一旦任务链条拉长到几十步甚至上百步翻车率就直线上升。这不是模型能力不够而是运行时管理出了问题。我最早接触Agent是在一个自动化代码审查的场景里当时用的是一个基于ReAct范式的框架单步推理效果很好但让它连续审查一个包含30多个文件的代码仓库时跑到第15个文件左右就开始出现重复读取、遗漏文件、甚至把之前已经审查过的内容重新审查一遍的情况。后来排查发现根本原因不是模型变笨了而是上下文窗口被中间结果撑爆了早期的关键指令被挤到了窗口边缘模型实际上是在“半失忆”状态下继续工作。这就是长任务的核心矛盾任务越长需要的上下文越多但上下文窗口是有限的。Harness这个概念本质上就是围绕这个矛盾展开的一整套工程解法。它不是一个具体的框架或工具而是一种运行时治理的思路——通过外部的状态管理、任务分解、检查点机制让Agent在长任务中保持“记忆连贯”和“行为可控”。热词里提到的“harness和agent区别”其实问到了点子上。Agent是执行者Harness是执行者身上的那套“安全带导航仪行车记录仪”。Agent负责思考和行动Harness负责确保这些思考和行动在长任务中不跑偏、不断片、不失控。你可以把Agent想象成一个能力很强但记性有限的员工Harness就是他的工作手册、进度看板和交接班制度。这套东西适合谁来参考如果你正在做Agent应用开发尤其是涉及多步骤、长流程、需要跨会话保持状态的场景比如自动化测试、代码迁移、文档批量处理、复杂数据管道编排那Harness的思路就是你必须掌握的。如果你只是做单轮问答或者简单工具调用那暂时用不上但了解一下也没坏处因为任务复杂度一旦上去这些问题是绕不开的。2. Harness的核心设计思路把“记忆”从模型里搬出来2.1 为什么不能全靠模型自己记很多人第一反应是现在模型上下文窗口不是越来越大吗128K、200K甚至1M的都有直接把所有中间结果塞进去不就行了我一开始也这么想实测下来发现两个问题。第一上下文越长模型对中间信息的注意力越稀释。这不是我瞎说有实验数据支撑当上下文超过一定长度后模型对位于中间位置的信息召回率会明显下降业界管这叫“lost in the middle”现象。你把30个文件的审查记录全塞进去模型对第15个文件的记忆可能还不如对第1个和第30个清楚。第二成本问题。每次推理都带着全部历史上下文token消耗是线性增长的。一个100步的任务如果每步都带全部历史总token消耗是O(n²)级别的。这在生产环境里是不可接受的。所以Harness的第一个核心设计就是把状态管理从模型内部搬到外部。模型只负责当前步骤的推理和决策历史状态由外部系统维护需要的时候再按需注入。这就像你不需要记住整个项目所有的代码只需要记住当前正在改的那个文件和相关的接口定义就行。2.2 任务分解与状态机的引入把状态搬出来之后下一个问题就是怎么组织这些状态我的经验是用状态机来管理任务进度是最稳妥的方案。具体来说把一个长任务拆成若干个阶段每个阶段有明确的输入、输出和完成条件。Agent在每个阶段内部可以自由推理但阶段之间的切换由Harness控制。这样做的好处是即使某个阶段内部出现了推理偏差也不会影响到其他阶段因为阶段之间的边界是硬性的。举个例子我之前做过一个批量数据清洗的Agent任务是把5000条脏数据清洗成结构化格式。如果让Agent一口气跑完跑到第2000条左右就开始出现格式混乱。后来改成状态机模式每100条为一个批次每个批次完成后由Harness做一次格式校验校验通过才进入下一批。这样即使某一批出了问题也只影响那100条不会污染全局。状态机的设计要点是每个状态的进入条件和退出条件必须可验证。不能是“Agent觉得做完了”就切换而是要有客观的校验标准。比如“文件已写入且行数大于0”这种可编程检查的条件。2.3 检查点机制让任务可以断点续跑长任务另一个头疼的问题是跑到一半挂了怎么办网络抖动、API限流、进程崩溃任何一个小问题都可能让几个小时的成果付诸东流。Harness的检查点机制就是解决这个问题的。核心思路很简单定期把当前状态持久化到外部存储任务重启时从最近的检查点恢复而不是从头开始。但这里有个细节很容易被忽略检查点不能只存“进度到哪了”还要存“当前的关键上下文”。比如你跑到第50步检查点里不仅要记录“已完成50步”还要记录“第50步产生的关键输出是什么”、“下一步的输入依赖是什么”。否则恢复的时候虽然知道进度但不知道上下文还是得从头推理。我一般会在检查点里存三类信息进度标记当前处于哪个阶段、第几步、关键状态最近N步的输入输出摘要、环境快照如果任务依赖外部环境比如文件系统状态、数据库连接信息等。存储介质用SQLite或者Redis都行看任务规模。注意检查点的写入频率要权衡。太频繁会影响性能太稀疏会丢失太多进度。我的经验是每完成一个“可验证的子任务”就写一次比如每处理完一个文件、每完成一个批次。3. 核心细节解析Harness的四个关键组件3.1 上下文管理器按需注入而非全量携带上下文管理器是Harness里最核心的组件它决定了每一步推理时模型能看到什么信息。设计原则是只给当前步骤需要的信息不给无关的历史包袱。具体实现上我通常会把上下文分成几个层级全局指令层任务的总目标、约束条件、输出格式要求。这部分始终保留但要做精简一般控制在500 token以内。阶段上下文层当前阶段的目标、已完成步骤的摘要、当前步骤的输入。这部分随阶段切换而更新。即时信息层当前步骤直接依赖的数据、工具返回结果、上一步的输出。这部分每步都变。关键技巧是摘要压缩。已完成步骤的详细记录不需要全部保留用模型生成一段简短摘要就行。比如“已完成文件A的审查发现3个问题缺少错误处理、变量命名不规范、存在硬编码”这样的摘要比保留完整的审查对话记录节省90%以上的token。我实测过一个对比一个50步的代码审查任务全量携带上下文需要约80K token用摘要压缩后只需要约12K token而且任务完成质量没有明显下降。3.2 工具调用网关统一入口与权限控制Agent在长任务中会频繁调用各种工具读写文件、执行命令、调用API、查询数据库。如果没有统一管理很容易出现权限混乱、调用冲突、错误处理不一致的问题。Harness的工具调用网关做三件事第一统一注册和发现。所有工具在网关注册Agent通过网关调用不直接接触底层实现。这样换工具实现的时候不影响Agent逻辑。第二权限和频率控制。哪些工具可以调用、每分钟最多调用几次、单次调用的超时时间都在网关层配置。我踩过一个坑Agent在循环里疯狂调用文件读取工具把磁盘IO打满了。后来在网关加了频率限制同一个工具在10秒内最多调用5次问题解决。第三错误统一处理。工具调用失败时网关负责重试、降级、或者返回结构化错误信息给Agent。不要让Agent直接面对原始的异常堆栈那样它会懵。# 工具网关的简化实现示例 class ToolGateway: def __init__(self): self.tools {} self.rate_limits {} self.call_counts {} def register(self, name, func, max_calls_per_minute60): self.tools[name] func self.rate_limits[name] max_calls_per_minute def call(self, name, **kwargs): if name not in self.tools: return {error: fTool {name} not registered} # 频率检查 current_minute int(time.time() / 60) key f{name}:{current_minute} self.call_counts[key] self.call_counts.get(key, 0) 1 if self.call_counts[key] self.rate_limits[name]: return {error: Rate limit exceeded, retry later} try: result self.tools[name](**kwargs) return {success: True, data: result} except Exception as e: return {success: False, error: str(e)}3.3 进度追踪器让“做到哪了”一目了然进度追踪器解决的是“Agent不知道自己做到哪了”的问题。在长任务中Agent很容易迷失方向反复做已经做过的事或者跳过必要的步骤。我的做法是维护一个任务清单每个子任务有明确的状态待处理、进行中、已完成、失败。Agent每步开始前先看清单确认当前应该处理哪个子任务。完成后更新状态。这个清单要持久化不能只放在内存里。我一般用JSON文件或者SQLite表来存结构大概是这样字段说明task_id子任务唯一标识description子任务描述statuspending/in_progress/done/failedinput_ref输入数据引用output_ref输出数据引用retry_count重试次数created_at创建时间updated_at更新时间Agent每完成一步就更新对应记录。Harness在每步开始前检查清单如果发现所有子任务都完成了就触发结束流程如果有失败的根据重试策略决定是重试还是跳过。3.4 异常恢复器分类处理而非一刀切长任务中出异常是常态关键是分类处理。我把异常分成三类可重试异常网络超时、API限流、临时性资源不可用。这类异常直接重试一般重试3次每次间隔指数退避。可跳过异常某个子任务失败但不影响整体目标比如批量处理中某一条数据格式错误。记录错误跳过继续。致命异常权限不足、关键依赖缺失、输出格式严重偏离。这类异常直接终止任务保存检查点通知人工介入。分类的逻辑要写在Harness里不要让Agent自己判断。Agent的判断力在长任务中不可靠它可能会把致命异常当成可重试异常反复重试浪费资源。4. 实操过程从零搭建一个长任务Harness4.1 环境准备与基础框架选型先说环境。Python 3.10以上主要依赖就几个pydantic做数据校验sqlite3做状态存储tenacity做重试控制。如果你用的Agent框架本身有状态管理模块比如LangGraph的checkpointer可以直接用但理解底层原理还是有必要的。框架选型上我的建议是不要一上来就上重型框架。很多框架把Harness的逻辑封装得太深出了问题很难排查。我一般先用轻量级方案跑通确认需求后再考虑是否引入框架。基础目录结构大概这样harness_demo/ ├── main.py # 入口 ├── harness/ │ ├── __init__.py │ ├── context.py # 上下文管理器 │ ├── gateway.py # 工具网关 │ ├── tracker.py # 进度追踪 │ └── recovery.py # 异常恢复 ├── tools/ │ ├── file_ops.py │ └── api_calls.py └── data/ └── state.db # SQLite状态库4.2 状态存储层的搭建状态存储用SQLite就够了轻量、零配置、支持并发读。建三张表任务表、子任务表、检查点表。CREATE TABLE tasks ( task_id TEXT PRIMARY KEY, goal TEXT NOT NULL, status TEXT DEFAULT pending, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE subtasks ( subtask_id TEXT PRIMARY KEY, task_id TEXT NOT NULL, description TEXT, status TEXT DEFAULT pending, input_ref TEXT, output_ref TEXT, retry_count INTEGER DEFAULT 0, FOREIGN KEY (task_id) REFERENCES tasks(task_id) ); CREATE TABLE checkpoints ( checkpoint_id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL, step_number INTEGER, context_snapshot TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (task_id) REFERENCES tasks(task_id) );写入检查点的时候context_snapshot存的是JSON序列化后的上下文摘要。恢复的时候反序列化回来注入到Agent的prompt里。4.3 主循环的实现逻辑主循环是Harness的心脏控制着整个任务的推进。逻辑不复杂但细节很多。def run_task(task_id, agent, max_steps200): task load_task(task_id) step 0 while step max_steps: # 1. 检查是否完成 if all_subtasks_done(task_id): mark_task_done(task_id) break # 2. 获取下一个待处理子任务 subtask get_next_pending_subtask(task_id) if subtask is None: # 没有待处理的但有失败的触发恢复逻辑 handle_failed_subtasks(task_id) continue # 3. 构建上下文 context build_context(task_id, subtask) # 4. 调用Agent推理 try: action agent.reason(context) except Exception as e: handle_agent_error(task_id, subtask, e) continue # 5. 执行动作 result execute_action(action, gateway) # 6. 更新状态 update_subtask_status(subtask, result) # 7. 写检查点 if step % 5 0: save_checkpoint(task_id, step, context) step 1 return load_task(task_id)这个循环里第2步的“获取下一个待处理子任务”是关键。我一般用优先级队列失败的子任务优先级最高先处理失败的然后是待处理的最后是进行中的可能是上次中断留下的。4.4 上下文构建的具体策略build_context函数决定了Agent每步能看到什么。我的实现分三块def build_context(task_id, subtask): # 全局指令精简版任务目标 global_instruction get_global_instruction(task_id) # 阶段摘要已完成子任务的摘要 completed_summary get_completed_summary(task_id, last_n10) # 当前任务详情 current_detail { subtask_id: subtask.subtask_id, description: subtask.description, input: load_input(subtask.input_ref), constraints: get_constraints(subtask) } return { global: global_instruction, history_summary: completed_summary, current: current_detail }get_completed_summary只取最近10个已完成子任务的摘要更早的用一句话概括。这样上下文长度可控同时保留了必要的连贯性。4.5 异常恢复的实操配置异常恢复的配置我一般放在一个YAML文件里方便调整recovery: retry: max_attempts: 3 backoff_base: 2 backoff_max: 60 skip: enabled: true max_skip_ratio: 0.1 # 最多跳过10%的子任务 fatal: notify: true save_checkpoint: truemax_skip_ratio这个参数很重要。如果跳过的子任务超过10%说明任务本身可能有问题继续跑下去没有意义应该终止并人工检查。5. 常见问题与排查技巧实录5.1 Agent陷入循环怎么办这是长任务中最常见的问题。Agent反复执行同一个动作或者在不同状态之间来回跳转。排查思路先看日志确认循环的模式。如果是动作级循环反复调用同一个工具大概率是工具返回的结果不符合Agent预期它以为没成功所以一直重试。检查工具返回格式是否和prompt里描述的一致。如果是状态级循环在几个子任务之间来回跳说明任务分解有问题子任务之间的依赖关系没理清。这时候需要重新设计状态机确保每个子任务有明确的单向推进路径。我的兜底方案是加一个循环检测器记录最近10步的动作序列如果发现重复模式强制中断并触发人工检查。5.2 上下文丢失导致行为异常表现是Agent突然“忘记”了之前的约束开始输出不符合格式要求的内容。原因通常是上下文构建时把关键指令挤掉了。排查方法在每步推理前把实际传给模型的prompt打印出来检查全局指令是否还在、阶段摘要是否完整。我遇到过因为摘要生成时把约束条件也摘要掉了导致Agent后面完全无视输出格式要求。解决方法是全局指令不参与摘要压缩始终原样保留。摘要只压缩历史动作和中间结果不压缩约束条件。5.3 检查点恢复后状态不一致恢复后Agent的行为和中断前对不上比如重复处理已经完成的子任务。这通常是检查点写入时状态没同步。检查点的写入要和状态更新在同一个事务里。我一般用SQLite的事务机制更新子任务状态和写检查点在一个BEGIN...COMMIT块里完成。这样要么都成功要么都回滚不会出现状态和检查点不一致的情况。5.4 工具调用超时拖垮整个任务某个工具调用卡住整个任务就停在那里。解决方案是给每个工具调用设超时超时后返回结构化错误让Agent决定是重试还是跳过。超时时间怎么定我的经验是文件操作5秒API调用30秒复杂计算60秒。超过这个时间大概率是出了问题继续等没有意义。问题类型典型表现排查方向解决手段动作循环反复调用同一工具工具返回格式统一返回结构加循环检测状态循环子任务间来回跳任务依赖关系重新设计状态机上下文丢失无视约束条件prompt内容检查全局指令不压缩检查点不一致恢复后重复处理事务同步状态和检查点同事务写入工具超时任务卡住不动工具性能设超时返回结构化错误5.5 长任务中的成本控制长任务跑起来token消耗很快不加控制的话成本会失控。几个实用的控制手段摘要压缩是最有效的能把历史上下文的token消耗降低80%以上。按需注入也很关键不是每步都需要全部上下文只给当前子任务相关的信息。缓存机制可以用上相同的工具调用结果缓存起来避免重复调用。我实测过一个50步的任务不做任何优化时总token消耗约120K做了摘要压缩和按需注入后降到约25K成本降到原来的五分之一任务完成质量基本不变。5.6 并发场景下的Harness适配热词里有人问“ai agent怎么扛并发”这确实是生产环境必须面对的问题。单个Agent的长任务已经够复杂了多个Agent并发跑的时候Harness需要额外考虑资源隔离和状态隔离。资源隔离方面每个Agent实例要有独立的工具网关实例避免频率限制互相干扰。状态隔离方面每个任务有独立的task_id状态存储按task_id分区互不影响。如果并发量很大SQLite可能扛不住换成PostgreSQL或者Redis。但大多数场景下单机跑几个并发任务SQLite完全够用。6. 一些踩坑之后的经验之谈Harness这个东西看别人写的感觉很简单自己上手做的时候坑是一个接一个。我最大的体会是不要试图让Agent自己管理自己。Agent的推理能力很强但它的“元认知”能力很弱它不知道自己做到哪了、不知道自己是不是在循环、不知道自己是不是该停下来。这些判断必须由Harness来做而且要做得足够“笨”——用简单的规则、明确的阈值、可验证的条件而不是让Agent自己去“感觉”。另一个体会是检查点的粒度比频率更重要。一开始我每步都写检查点结果IO成了瓶颈。后来改成按子任务写每个子任务完成时写一次性能上去了恢复粒度也够用。因为子任务本身就是最小的可验证单元恢复到子任务边界是最合理的。还有一点日志要记全但不要记在上下文里。Agent的每一步推理、每一次工具调用、每一个状态变更都要记日志但日志是给人看的不要塞进Agent的上下文。上下文里只放Agent当前需要的信息日志放外部文件或者数据库排查问题的时候去查。最后说一个容易被忽略的点Harness本身也需要测试。很多人只测试Agent的行为不测试Harness的逻辑。结果Agent没问题Harness的状态机有bug任务照样跑不通。我的做法是给Harness写单元测试模拟各种异常场景工具超时、检查点写入失败、状态更新冲突确保Harness在各种边界条件下都能正确响应。这套东西搭起来之后Agent跑长任务的稳定性会有质的提升。我之前那个代码审查的Agent从最初跑30个文件就崩到后来稳定跑200个文件不出错靠的就是Harness这一层的治理。模型没换prompt没大改就是加了状态管理、检查点、异常恢复这些工程手段。所以说Agent的可靠性问题很多时候不是模型的问题是工程的问题。
Harness工程实战：如何让AI Agent稳定跑完长任务
1. 长任务为什么总是翻车从Agent的运行时困境说起做过Agent开发的人都有一个共同体会让Agent跑一个短任务比如查个天气、写段摘要、做个简单计算基本都能应付。但一旦任务链条拉长到几十步甚至上百步翻车率就直线上升。这不是模型能力不够而是运行时管理出了问题。我最早接触Agent是在一个自动化代码审查的场景里当时用的是一个基于ReAct范式的框架单步推理效果很好但让它连续审查一个包含30多个文件的代码仓库时跑到第15个文件左右就开始出现重复读取、遗漏文件、甚至把之前已经审查过的内容重新审查一遍的情况。后来排查发现根本原因不是模型变笨了而是上下文窗口被中间结果撑爆了早期的关键指令被挤到了窗口边缘模型实际上是在“半失忆”状态下继续工作。这就是长任务的核心矛盾任务越长需要的上下文越多但上下文窗口是有限的。Harness这个概念本质上就是围绕这个矛盾展开的一整套工程解法。它不是一个具体的框架或工具而是一种运行时治理的思路——通过外部的状态管理、任务分解、检查点机制让Agent在长任务中保持“记忆连贯”和“行为可控”。热词里提到的“harness和agent区别”其实问到了点子上。Agent是执行者Harness是执行者身上的那套“安全带导航仪行车记录仪”。Agent负责思考和行动Harness负责确保这些思考和行动在长任务中不跑偏、不断片、不失控。你可以把Agent想象成一个能力很强但记性有限的员工Harness就是他的工作手册、进度看板和交接班制度。这套东西适合谁来参考如果你正在做Agent应用开发尤其是涉及多步骤、长流程、需要跨会话保持状态的场景比如自动化测试、代码迁移、文档批量处理、复杂数据管道编排那Harness的思路就是你必须掌握的。如果你只是做单轮问答或者简单工具调用那暂时用不上但了解一下也没坏处因为任务复杂度一旦上去这些问题是绕不开的。2. Harness的核心设计思路把“记忆”从模型里搬出来2.1 为什么不能全靠模型自己记很多人第一反应是现在模型上下文窗口不是越来越大吗128K、200K甚至1M的都有直接把所有中间结果塞进去不就行了我一开始也这么想实测下来发现两个问题。第一上下文越长模型对中间信息的注意力越稀释。这不是我瞎说有实验数据支撑当上下文超过一定长度后模型对位于中间位置的信息召回率会明显下降业界管这叫“lost in the middle”现象。你把30个文件的审查记录全塞进去模型对第15个文件的记忆可能还不如对第1个和第30个清楚。第二成本问题。每次推理都带着全部历史上下文token消耗是线性增长的。一个100步的任务如果每步都带全部历史总token消耗是O(n²)级别的。这在生产环境里是不可接受的。所以Harness的第一个核心设计就是把状态管理从模型内部搬到外部。模型只负责当前步骤的推理和决策历史状态由外部系统维护需要的时候再按需注入。这就像你不需要记住整个项目所有的代码只需要记住当前正在改的那个文件和相关的接口定义就行。2.2 任务分解与状态机的引入把状态搬出来之后下一个问题就是怎么组织这些状态我的经验是用状态机来管理任务进度是最稳妥的方案。具体来说把一个长任务拆成若干个阶段每个阶段有明确的输入、输出和完成条件。Agent在每个阶段内部可以自由推理但阶段之间的切换由Harness控制。这样做的好处是即使某个阶段内部出现了推理偏差也不会影响到其他阶段因为阶段之间的边界是硬性的。举个例子我之前做过一个批量数据清洗的Agent任务是把5000条脏数据清洗成结构化格式。如果让Agent一口气跑完跑到第2000条左右就开始出现格式混乱。后来改成状态机模式每100条为一个批次每个批次完成后由Harness做一次格式校验校验通过才进入下一批。这样即使某一批出了问题也只影响那100条不会污染全局。状态机的设计要点是每个状态的进入条件和退出条件必须可验证。不能是“Agent觉得做完了”就切换而是要有客观的校验标准。比如“文件已写入且行数大于0”这种可编程检查的条件。2.3 检查点机制让任务可以断点续跑长任务另一个头疼的问题是跑到一半挂了怎么办网络抖动、API限流、进程崩溃任何一个小问题都可能让几个小时的成果付诸东流。Harness的检查点机制就是解决这个问题的。核心思路很简单定期把当前状态持久化到外部存储任务重启时从最近的检查点恢复而不是从头开始。但这里有个细节很容易被忽略检查点不能只存“进度到哪了”还要存“当前的关键上下文”。比如你跑到第50步检查点里不仅要记录“已完成50步”还要记录“第50步产生的关键输出是什么”、“下一步的输入依赖是什么”。否则恢复的时候虽然知道进度但不知道上下文还是得从头推理。我一般会在检查点里存三类信息进度标记当前处于哪个阶段、第几步、关键状态最近N步的输入输出摘要、环境快照如果任务依赖外部环境比如文件系统状态、数据库连接信息等。存储介质用SQLite或者Redis都行看任务规模。注意检查点的写入频率要权衡。太频繁会影响性能太稀疏会丢失太多进度。我的经验是每完成一个“可验证的子任务”就写一次比如每处理完一个文件、每完成一个批次。3. 核心细节解析Harness的四个关键组件3.1 上下文管理器按需注入而非全量携带上下文管理器是Harness里最核心的组件它决定了每一步推理时模型能看到什么信息。设计原则是只给当前步骤需要的信息不给无关的历史包袱。具体实现上我通常会把上下文分成几个层级全局指令层任务的总目标、约束条件、输出格式要求。这部分始终保留但要做精简一般控制在500 token以内。阶段上下文层当前阶段的目标、已完成步骤的摘要、当前步骤的输入。这部分随阶段切换而更新。即时信息层当前步骤直接依赖的数据、工具返回结果、上一步的输出。这部分每步都变。关键技巧是摘要压缩。已完成步骤的详细记录不需要全部保留用模型生成一段简短摘要就行。比如“已完成文件A的审查发现3个问题缺少错误处理、变量命名不规范、存在硬编码”这样的摘要比保留完整的审查对话记录节省90%以上的token。我实测过一个对比一个50步的代码审查任务全量携带上下文需要约80K token用摘要压缩后只需要约12K token而且任务完成质量没有明显下降。3.2 工具调用网关统一入口与权限控制Agent在长任务中会频繁调用各种工具读写文件、执行命令、调用API、查询数据库。如果没有统一管理很容易出现权限混乱、调用冲突、错误处理不一致的问题。Harness的工具调用网关做三件事第一统一注册和发现。所有工具在网关注册Agent通过网关调用不直接接触底层实现。这样换工具实现的时候不影响Agent逻辑。第二权限和频率控制。哪些工具可以调用、每分钟最多调用几次、单次调用的超时时间都在网关层配置。我踩过一个坑Agent在循环里疯狂调用文件读取工具把磁盘IO打满了。后来在网关加了频率限制同一个工具在10秒内最多调用5次问题解决。第三错误统一处理。工具调用失败时网关负责重试、降级、或者返回结构化错误信息给Agent。不要让Agent直接面对原始的异常堆栈那样它会懵。# 工具网关的简化实现示例 class ToolGateway: def __init__(self): self.tools {} self.rate_limits {} self.call_counts {} def register(self, name, func, max_calls_per_minute60): self.tools[name] func self.rate_limits[name] max_calls_per_minute def call(self, name, **kwargs): if name not in self.tools: return {error: fTool {name} not registered} # 频率检查 current_minute int(time.time() / 60) key f{name}:{current_minute} self.call_counts[key] self.call_counts.get(key, 0) 1 if self.call_counts[key] self.rate_limits[name]: return {error: Rate limit exceeded, retry later} try: result self.tools[name](**kwargs) return {success: True, data: result} except Exception as e: return {success: False, error: str(e)}3.3 进度追踪器让“做到哪了”一目了然进度追踪器解决的是“Agent不知道自己做到哪了”的问题。在长任务中Agent很容易迷失方向反复做已经做过的事或者跳过必要的步骤。我的做法是维护一个任务清单每个子任务有明确的状态待处理、进行中、已完成、失败。Agent每步开始前先看清单确认当前应该处理哪个子任务。完成后更新状态。这个清单要持久化不能只放在内存里。我一般用JSON文件或者SQLite表来存结构大概是这样字段说明task_id子任务唯一标识description子任务描述statuspending/in_progress/done/failedinput_ref输入数据引用output_ref输出数据引用retry_count重试次数created_at创建时间updated_at更新时间Agent每完成一步就更新对应记录。Harness在每步开始前检查清单如果发现所有子任务都完成了就触发结束流程如果有失败的根据重试策略决定是重试还是跳过。3.4 异常恢复器分类处理而非一刀切长任务中出异常是常态关键是分类处理。我把异常分成三类可重试异常网络超时、API限流、临时性资源不可用。这类异常直接重试一般重试3次每次间隔指数退避。可跳过异常某个子任务失败但不影响整体目标比如批量处理中某一条数据格式错误。记录错误跳过继续。致命异常权限不足、关键依赖缺失、输出格式严重偏离。这类异常直接终止任务保存检查点通知人工介入。分类的逻辑要写在Harness里不要让Agent自己判断。Agent的判断力在长任务中不可靠它可能会把致命异常当成可重试异常反复重试浪费资源。4. 实操过程从零搭建一个长任务Harness4.1 环境准备与基础框架选型先说环境。Python 3.10以上主要依赖就几个pydantic做数据校验sqlite3做状态存储tenacity做重试控制。如果你用的Agent框架本身有状态管理模块比如LangGraph的checkpointer可以直接用但理解底层原理还是有必要的。框架选型上我的建议是不要一上来就上重型框架。很多框架把Harness的逻辑封装得太深出了问题很难排查。我一般先用轻量级方案跑通确认需求后再考虑是否引入框架。基础目录结构大概这样harness_demo/ ├── main.py # 入口 ├── harness/ │ ├── __init__.py │ ├── context.py # 上下文管理器 │ ├── gateway.py # 工具网关 │ ├── tracker.py # 进度追踪 │ └── recovery.py # 异常恢复 ├── tools/ │ ├── file_ops.py │ └── api_calls.py └── data/ └── state.db # SQLite状态库4.2 状态存储层的搭建状态存储用SQLite就够了轻量、零配置、支持并发读。建三张表任务表、子任务表、检查点表。CREATE TABLE tasks ( task_id TEXT PRIMARY KEY, goal TEXT NOT NULL, status TEXT DEFAULT pending, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE subtasks ( subtask_id TEXT PRIMARY KEY, task_id TEXT NOT NULL, description TEXT, status TEXT DEFAULT pending, input_ref TEXT, output_ref TEXT, retry_count INTEGER DEFAULT 0, FOREIGN KEY (task_id) REFERENCES tasks(task_id) ); CREATE TABLE checkpoints ( checkpoint_id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL, step_number INTEGER, context_snapshot TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (task_id) REFERENCES tasks(task_id) );写入检查点的时候context_snapshot存的是JSON序列化后的上下文摘要。恢复的时候反序列化回来注入到Agent的prompt里。4.3 主循环的实现逻辑主循环是Harness的心脏控制着整个任务的推进。逻辑不复杂但细节很多。def run_task(task_id, agent, max_steps200): task load_task(task_id) step 0 while step max_steps: # 1. 检查是否完成 if all_subtasks_done(task_id): mark_task_done(task_id) break # 2. 获取下一个待处理子任务 subtask get_next_pending_subtask(task_id) if subtask is None: # 没有待处理的但有失败的触发恢复逻辑 handle_failed_subtasks(task_id) continue # 3. 构建上下文 context build_context(task_id, subtask) # 4. 调用Agent推理 try: action agent.reason(context) except Exception as e: handle_agent_error(task_id, subtask, e) continue # 5. 执行动作 result execute_action(action, gateway) # 6. 更新状态 update_subtask_status(subtask, result) # 7. 写检查点 if step % 5 0: save_checkpoint(task_id, step, context) step 1 return load_task(task_id)这个循环里第2步的“获取下一个待处理子任务”是关键。我一般用优先级队列失败的子任务优先级最高先处理失败的然后是待处理的最后是进行中的可能是上次中断留下的。4.4 上下文构建的具体策略build_context函数决定了Agent每步能看到什么。我的实现分三块def build_context(task_id, subtask): # 全局指令精简版任务目标 global_instruction get_global_instruction(task_id) # 阶段摘要已完成子任务的摘要 completed_summary get_completed_summary(task_id, last_n10) # 当前任务详情 current_detail { subtask_id: subtask.subtask_id, description: subtask.description, input: load_input(subtask.input_ref), constraints: get_constraints(subtask) } return { global: global_instruction, history_summary: completed_summary, current: current_detail }get_completed_summary只取最近10个已完成子任务的摘要更早的用一句话概括。这样上下文长度可控同时保留了必要的连贯性。4.5 异常恢复的实操配置异常恢复的配置我一般放在一个YAML文件里方便调整recovery: retry: max_attempts: 3 backoff_base: 2 backoff_max: 60 skip: enabled: true max_skip_ratio: 0.1 # 最多跳过10%的子任务 fatal: notify: true save_checkpoint: truemax_skip_ratio这个参数很重要。如果跳过的子任务超过10%说明任务本身可能有问题继续跑下去没有意义应该终止并人工检查。5. 常见问题与排查技巧实录5.1 Agent陷入循环怎么办这是长任务中最常见的问题。Agent反复执行同一个动作或者在不同状态之间来回跳转。排查思路先看日志确认循环的模式。如果是动作级循环反复调用同一个工具大概率是工具返回的结果不符合Agent预期它以为没成功所以一直重试。检查工具返回格式是否和prompt里描述的一致。如果是状态级循环在几个子任务之间来回跳说明任务分解有问题子任务之间的依赖关系没理清。这时候需要重新设计状态机确保每个子任务有明确的单向推进路径。我的兜底方案是加一个循环检测器记录最近10步的动作序列如果发现重复模式强制中断并触发人工检查。5.2 上下文丢失导致行为异常表现是Agent突然“忘记”了之前的约束开始输出不符合格式要求的内容。原因通常是上下文构建时把关键指令挤掉了。排查方法在每步推理前把实际传给模型的prompt打印出来检查全局指令是否还在、阶段摘要是否完整。我遇到过因为摘要生成时把约束条件也摘要掉了导致Agent后面完全无视输出格式要求。解决方法是全局指令不参与摘要压缩始终原样保留。摘要只压缩历史动作和中间结果不压缩约束条件。5.3 检查点恢复后状态不一致恢复后Agent的行为和中断前对不上比如重复处理已经完成的子任务。这通常是检查点写入时状态没同步。检查点的写入要和状态更新在同一个事务里。我一般用SQLite的事务机制更新子任务状态和写检查点在一个BEGIN...COMMIT块里完成。这样要么都成功要么都回滚不会出现状态和检查点不一致的情况。5.4 工具调用超时拖垮整个任务某个工具调用卡住整个任务就停在那里。解决方案是给每个工具调用设超时超时后返回结构化错误让Agent决定是重试还是跳过。超时时间怎么定我的经验是文件操作5秒API调用30秒复杂计算60秒。超过这个时间大概率是出了问题继续等没有意义。问题类型典型表现排查方向解决手段动作循环反复调用同一工具工具返回格式统一返回结构加循环检测状态循环子任务间来回跳任务依赖关系重新设计状态机上下文丢失无视约束条件prompt内容检查全局指令不压缩检查点不一致恢复后重复处理事务同步状态和检查点同事务写入工具超时任务卡住不动工具性能设超时返回结构化错误5.5 长任务中的成本控制长任务跑起来token消耗很快不加控制的话成本会失控。几个实用的控制手段摘要压缩是最有效的能把历史上下文的token消耗降低80%以上。按需注入也很关键不是每步都需要全部上下文只给当前子任务相关的信息。缓存机制可以用上相同的工具调用结果缓存起来避免重复调用。我实测过一个50步的任务不做任何优化时总token消耗约120K做了摘要压缩和按需注入后降到约25K成本降到原来的五分之一任务完成质量基本不变。5.6 并发场景下的Harness适配热词里有人问“ai agent怎么扛并发”这确实是生产环境必须面对的问题。单个Agent的长任务已经够复杂了多个Agent并发跑的时候Harness需要额外考虑资源隔离和状态隔离。资源隔离方面每个Agent实例要有独立的工具网关实例避免频率限制互相干扰。状态隔离方面每个任务有独立的task_id状态存储按task_id分区互不影响。如果并发量很大SQLite可能扛不住换成PostgreSQL或者Redis。但大多数场景下单机跑几个并发任务SQLite完全够用。6. 一些踩坑之后的经验之谈Harness这个东西看别人写的感觉很简单自己上手做的时候坑是一个接一个。我最大的体会是不要试图让Agent自己管理自己。Agent的推理能力很强但它的“元认知”能力很弱它不知道自己做到哪了、不知道自己是不是在循环、不知道自己是不是该停下来。这些判断必须由Harness来做而且要做得足够“笨”——用简单的规则、明确的阈值、可验证的条件而不是让Agent自己去“感觉”。另一个体会是检查点的粒度比频率更重要。一开始我每步都写检查点结果IO成了瓶颈。后来改成按子任务写每个子任务完成时写一次性能上去了恢复粒度也够用。因为子任务本身就是最小的可验证单元恢复到子任务边界是最合理的。还有一点日志要记全但不要记在上下文里。Agent的每一步推理、每一次工具调用、每一个状态变更都要记日志但日志是给人看的不要塞进Agent的上下文。上下文里只放Agent当前需要的信息日志放外部文件或者数据库排查问题的时候去查。最后说一个容易被忽略的点Harness本身也需要测试。很多人只测试Agent的行为不测试Harness的逻辑。结果Agent没问题Harness的状态机有bug任务照样跑不通。我的做法是给Harness写单元测试模拟各种异常场景工具超时、检查点写入失败、状态更新冲突确保Harness在各种边界条件下都能正确响应。这套东西搭起来之后Agent跑长任务的稳定性会有质的提升。我之前那个代码审查的Agent从最初跑30个文件就崩到后来稳定跑200个文件不出错靠的就是Harness这一层的治理。模型没换prompt没大改就是加了状态管理、检查点、异常恢复这些工程手段。所以说Agent的可靠性问题很多时候不是模型的问题是工程的问题。