同一个模型,换一套Harness,任务完成率从52.8%飙升到66.5%。决定Agent上限的,从来不是模型本身。凌晨3点,Agent又崩了上个月我们接了个客户的项目,一家做跨境电商的公司,想让AI Agent自动完成选品-写listing-上架-监控评论的全链路。听起来挺简单对吧?前两周上线,Agent每天稳定跑4-5个小时,处理200多个SKU。第三周开始出问题了:Day 15: Agent在写第187个listing时上下文窗口满了,开始胡言乱语,把上一个产品的描述套到了新产品上Day 18: 服务重启后Agent完全忘了之前做到哪了,从头开始,重复生成了47个已经存在的listingDay 22: Agent调用了一个外部API超时,没有重试逻辑,直接报错退出,剩下一半任务没人处理Day 25: 运营同学发现Agent在过去6小时里反复执行同一个工具调用,像陷入了死循环团队里有人说换个更强的模型就好了,于是从GPT-4o换到Claude 3.5,情况确实好了两天。但到了Day 30,同样的问题又冒出来了。这个问题我们花了整整两周才真正想明白,Agent崩了,不是因为模型不够聪明,而是因为没有给它搭一个操作系统。这个操作系统,业内现在叫Harness。Harness是什么:模型是CPU,Harness是操作系统先看一组数据:LangChain在Terminal-Bench 2.0基准测试上做了一个对照实验,只改Harness,不换模型,把自建编码Agent的任务完成率从52.8%提升到了66.5%。这意味着什么?相当于从Top 30直接跳到Top 5。数据来源: LangChain公开工程报告, Terminal-Bench 2.0基准测试, 2026年这不是玄学。道理其实很朴素:模型负责思考和生成,它是CPUHarness负责状态管理、工具调度、错误恢复、上下文治理,它是操作系统你见过谁在没有操作系统的裸机上跑生产服务的吗? 但在AI Agent领域,很多人就是这么干的。直接写个prompt,接几个tool,丢到生产环境就完事了。短任务还凑合,一旦任务时长超过几分钟甚至几小时,问题就会一个接一个冒出来。2026年10月,这个方向集中爆发了:Mastra的AgentController正式GA(10月6日),支持持久化线程、独立会话和断点恢复Vercel开源了durable-harness包,让编码Agent在崩溃后能从上次断点继续Anthropic发了长文Effective harnesses for long-running agents,系统讲了双Agent架构Google推出ACE引擎,给每个Agent分配一个硬件隔离的microVM行业共识正在形成:Prompt Engineering Context Engineering Harness Engineering,三者是包含关系,不是替代关系。Prompt解决怎么问,Context解决给它看什么,Harness解决它怎么在真实环境里把事做成。今天这篇文章,我带你用500行Python从零搭建一个生产级Agent Harness。不是玩具Demo,是能直接上生产的那种。Harness的四层架构在动手写代码之前,先搞清楚我们要建什么东西。4) 编排层: 任务拆解、流程控制、多Agent协作 3) 反馈层: 结果校验、错误归因、LLM自评 2) 执行层: 工具调用、文件读写、API调用、沙箱 1) 记忆层: 规则文件、状态快照、技能定义越往下越像基础设施,越往上越像任务控制。大部分人的Agent只做了第2层(接了几个工具)和第1层的一部分(写了个system prompt),然后就直接上生产了。这就好比你装了个操作系统内核,但没装文件系统、没装进程管理、没装错误处理,然后怪电脑老死机。接下来我们一层一层搭。第一层: 记忆层,让Agent记住自己是谁、做到哪了问题本质模型每次推理都是无状态的。它不记得:这个任务的整体目标是什么上一步做了什么哪些路径已经试过且失败了当前进度到了哪里所以记忆层的核心不是存很多东西,而是让正确的信息在正确的时刻被取回。代码实现import json import hashlib from pathlib import Path from dataclasses import dataclass, field, asdict from typing import Optional, Any from datetime import datetimedataclassclass Checkpoint:Agent execution snapshot for crash recoverytask_id: strstep: intstatus: str # running, paused, completed, failedcontext_snapshot: dicttool_results: listerror_log: listcreated_at: str field(default_factorylambda: datetime.now().isoformat())def fingerprint(self) -gt; str: quot;quot;quot;Generate snapshot fingerprint for integrity verificationquot;quot;quot; raw json.dumps(self.context_snapshot, sort_keysTrue) return hashlib.sha256(raw.encode()).hexdigest()[:16]class MemoryLayer:Memory Layer: manages rules, state, and checkpointsdef __init__(self, workspace: str quot;./agent_workspacequot;): self.workspace Path(workspace) self.workspace.mkdir(parentsTrue, exist_okTrue) self._rules_cache: dict {} self._state: dict {} def load_rules(self, rule_files: list[str]) -gt; str: quot;quot;quot;Load rule files and return concatenated rules textquot;quot;quot; rules [] for rf in rule_files: path self.workspace / rf if path.exists(): content path.read_text(encodingquot;utf-8quot;) self._rules_cache[rf] content rules.append(fquot;### Rule: {rf}\n{content}quot;) return quot;\n\nquot;.join(rules) def save_checkpoint(self, checkpoint: Checkpoint): quot;quot;quot;Save checkpoint for crash recoveryquot;quot;quot; cp_dir self.workspace / quot;checkpointsquot; cp_dir.mkdir(exist_okTrue) cp_file cp_dir / fquot;{checkpoint.task_id}_step{checkpoint.step}.jsonquot; data asdict(checkpoint) data[quot;fingerprintquot;] checkpoint.fingerprint() # Atomic write: write to temp file first, then rename tmp_file cp_file.with_suffix(quot;.tmpquot;) tmp_file.write_text(json.dumps(data, ensure_asciiFalse, indent2), encodingquot;utf-8quot;) tmp_file.rename(cp_file) def load_latest_checkpoint(self, task_id: str) -gt; Optional[Checkpoint]: quot;quot;quot;Load latest checkpoint for resumequot;quot;quot; cp_dir self.workspace / quot;checkpointsquot; if not cp_dir.exists(): return None candidates sorted(cp_dir.glob(fquot;{task_id}_step*.jsonquot;), reverseTrue) if not candidates: return None data json.loads(candidates[0].read_text(encodingquot;utf-8quot;)) # Verify integrity stored_fp data.pop(quot;fingerprintquot;, quot;quot;) ctx data.get(quot;context_snapshotquot;, {}) raw json.dumps(ctx, sort_keysTrue) computed_fp hashlib.sha256(raw.encode()).hexdigest()[:16] if stored_fp ! computed_fp: raise ValueError(fquot;Checkpoint {candidates[0].name} data corrupted: quot; fquot;expected {stored_fp}, got {computed_fp}quot;) return Checkpoint(**data) def update_state(self, key: str, value: Any): quot;quot;quot;Update agent runtime statequot;quot;quot; self._state[key] value state_file self.workspace / quot;state.jsonquot; state_file.write_text(json.dumps(self._state, ensure_asciiFalse, indent2), encodingquot;utf-8quot;)这段代码做了三件事:规则文件加载: 把仓库级规则(类似AGENTS.md)统一管理,Agent启动时一次性注入上下文检查点持久化: 每一步执行完都保存快照,包括任务进度、工具调用结果和上下文状态。原子写入保证不会写出半截文件完整性校验: 用SHA256指纹验证检查点没被篡改或损坏你可能会问:每步都存检查点,IO开销不大吗?实测下来,单个检查点文件大约2-8KB。即使每秒存一次,一小时也就7-28MB。相比Agent崩溃后从头重跑的API成本,这个开销基本可以忽略。第二层: 执行层,带超时、重试和熔断的工具调度问题本质裸调工具,你会遇到:外部API超时,Agent卡死工具返回异常,Agent不知道怎么处理,开始瞎猜某个工具反复失败,Agent还在不停调用,浪费token执行层的核心是给每次工具调用加上安全带。代码实现import asyncio import time from functools import wraps from typing import Callable, Any, Optional from dataclasses import dataclassdataclassclass ToolCallResult:success: booloutput: Anyduration_ms: floatretries: interror: Optional[str] Noneclass CircuitBreaker:Circuit breaker: auto-trip after N consecutive failuresdef __init__(self, failure_threshold: int 3, recovery_timeout: float 60.0): self.failure_threshold failure_threshold self.recovery_timeout recovery_timeout self._failures: dict[str, int] {} self._last_failure_time: dict[str, float] {} self._open_circuits: set[str] set() def record_success(self, tool_name: str): self._failures.pop(tool_name, None) self._open_circuits.discard(tool_name) def record_failure(self, tool_name: str): self._failures[tool_name] self._failures.get(tool_name, 0) 1 self._last_failure_time[tool_name] time.time() if self._failures[tool_name] gt; self.failure_threshold: self._open_circuits.add(tool_name) def is_open(self, tool_name: str) -gt; bool: if tool_name not in self._open_circuits: return False elapsed time.time() - self._last_failure_time.get(tool_name, 0) if elapsed gt; self.recovery_timeout: # Half-open: attempt recovery self._open_circuits.discard(tool_name) self._failures.pop(tool_name, None) return False return Trueclass ExecutionLayer:Execution Layer: tool dispatcher with timeout, retry, and circuit breakerdef __init__(self, max_retries: int 3, timeout_seconds: float 30.0, backoff_factor: float 2.0): self.max_retries max_retries self.timeout_seconds timeout_seconds self.backoff_factor backoff_factor self.circuit_breaker CircuitBreaker() async def call_tool(self, tool_fn: Callable, tool_name: str, *args, **kwargs) -gt; ToolCallResult: quot;quot;quot;Execute tool call with full protectionquot;quot;quot; # Circuit breaker check if self.circuit_breaker.is_open(tool_name): return ToolCallResult( successFalse, outputNone, duration_ms0, retries0, errorf[CIRCUIT OPEN] Tool {tool_name} is tripped, retry after {self.circuit_breaker.recovery_timeout}s) start_time time.time() last_error None for attempt in range(self.max_retries 1): try: result await asyncio.wait_for( tool_fn(*args, **kwargs), timeoutself.timeout_seconds ) duration (time.time() - start_time) * 1000 self.circuit_breaker.record_success(tool_name) return ToolCallResult( successTrue, outputresult, duration_msround(duration, 2), retriesattempt) except asyncio.TimeoutError: last_error fTool {tool_name} timed out ({self.timeout_seconds}s) except Exception as e: last_error str(e) if attempt lt; self.max_retries: wait_time self.backoff_factor ** attempt await asyncio.sleep(wait_time) duration (time.time() - start_time) * 1000 self.circuit_breaker.record_failure(tool_name) return ToolCallResult( successFalse, outputNone, duration_msround(duration, 2), retriesself.max_retries, errorlast_error)关键设计点:超时控制: asyncio.wait_for给每次工具调用设置硬超时,Agent不会永远卡在某个API上指数退避: 失败后等待时间依次是1s - 2s - 4s,避免对下游服务的DDoS熔断器: 连续失败3次自动熔断60秒,防止Agent陷入死循环这个熔断器的设计灵感来自微服务架构的Hystrix模式。事实上,如果你在用Agent做生产系统,很多微服务里的工程实践都可以直接借鉴。在舆情监控等场景下,像黑箭科技这类系统在处理Agent稳定性时也会优先考虑引入熔断机制,毕竟生产环境里没有什么是不会挂的。第三层: 反馈层,Agent做完了,你怎么知道它做对了?问题本质Agent执行完一步,怎么判断结果是对的? 最朴素的方式是让Agent自己判断,但这在Anthropic的实践中被证明不可靠。Agent在上下文接近上限时会出现context anxiety(急于宣布成功),即使实际上没做完也会报告完成了。正确的做法是把验证交给工具: 跑测试、跑lint、做diff比较。代码实现from dataclasses import dataclass from typing import Optional, Callable import redataclassclass FeedbackResult:passed: boolscore: float # 0.0 - 1.0details: strsuggestions: list[str]class FeedbackLayer:Feedback Layer: deterministic verification instead of agent self-evaldef __init__(self): self.validators: dict[str, Callable] {} self._register_default_validators() def _register_default_validators(self): self.validators[non_empty] self._validate_non_empty self.validators[no_repeat] self._validate_no_repeat def _validate_non_empty(self, output: str) -gt; FeedbackResult: if output and output.strip(): return FeedbackResult(True, 1.0, quot;output is non-emptyquot;, []) return FeedbackResult(False, 0.0, quot;output is emptyquot;, [quot;Ensure output contains valid contentquot;]) def _validate_no_repeat(self, output: str, threshold: float 0.7) -gt; FeedbackResult: quot;quot;quot;Detect large-scale repetition (early signal of agent loops)quot;quot;quot; if len(output) lt; 100: return FeedbackResult(True, 1.0, quot;too short for repeat checkquot;, []) paragraphs [p.strip() for p in output.split(\\n\\n) if p.strip()] if len(paragraphs) lt; 2: return FeedbackResult(True, 1.0, quot;not enough paragraphsquot;, []) repeat_count 0 for i in range(len(paragraphs) - 1): words_a set(paragraphs[i].split()) words_b set(paragraphs[i 1].split()) if not words_a or not words_b: continue similarity len(words_a amp; words_b) / len(words_a | words_b) if similarity gt; threshold: repeat_count 1 repeat_ratio repeat_count / max(len(paragraphs) - 1, 1) if repeat_ratio gt; 0.5: return FeedbackResult( False, 1.0 - repeat_ratio, fHigh repetition detected ({repeat_ratio:.0%}), [quot;Agent may be in a repeat loopquot;, quot;Check tool calls for infinite loopsquot;]) return FeedbackResult(True, 1.0 - repeat_ratio, fRepeat ratio: {repeat_ratio:.0%}, []) def run_all(self, output: str, custom_checks: Optional[list] None) -gt; list[FeedbackResult]: quot;quot;quot;Run all validators and return combined resultsquot;quot;quot; results [] for name, validator in self.validators.items(): try: results.append(validator(output)) except Exception as e: results.append(FeedbackResult( False, 0.0, fValidator {name} error: {e}, [])) if custom_checks: for check in custom_checks: try: results.append(check(output)) except Exception as e: results.append(FeedbackResult( False, 0.0, fCustom check error: {e}, [])) return results这里有个细节值得说一下:重复检测不是可选的。前面那个跨境电商客户的Agent,就是因为写listing时陷入了重复循环,6小时里反复生成相同内容。如果反馈层有重复检测,第一次发现重复就该触发告警了。第四层: 编排层,把三层串起来class AgentHarness: Orchestration Layer: integrates memory, execution, and feedbackdef __init__(self, agent_config: dict, workspace: str quot;./agent_workspacequot;): self.memory MemoryLayer(workspace) self.executor ExecutionLayer( max_retriesagent_config.get(max_retries, 3), timeout_secondsagent_config.get(timeout_seconds, 30.0) ) self.feedback FeedbackLayer() self.config agent_config self.current_step 0 self.max_steps agent_config.get(max_steps, 100) self.tools: dict[str, Callable] {} self._running False def register_tool(self, name: str, fn: Callable): self.tools[name] fn async def run(self, task_id: str, task_description: str, resume: bool False) -gt; dict: quot;quot;quot;Run the agent main loopquot;quot;quot; self._running True start_step 0 context {task: task_description, history: []} # Initialize or resume from checkpoint if resume: checkpoint self.memory.load_latest_checkpoint(task_id) if checkpoint: start_step checkpoint.step context checkpoint.context_snapshot print(f[Harness] Resumed from checkpoint: step{start_step}) # Load rules rules self.memory.load_rules( self.config.get(rule_files, [AGENTS.md])) context[rules] rules # Main loop step start_step while step lt; self.max_steps and self._running: self.current_step step try: action await self._decide(context) if action[type] tool_call: tool_name action[tool] tool_args action.get(args, {}) result await self.executor.call_tool( self.tools.get(tool_name, self._noop), tool_name, **tool_args) if result.success: feedback self.feedback.run_all(str(result.output)) failed [f for f in feedback if not f.passed] context[history].append({ step: step, action: action, result: result.output, feedback_issues: len(failed) }) else: context[history].append({ step: step, action: action, result: error, error: result.error }) elif action[type] complete: break # Save checkpoint every 5 steps if step % 5 0: self.memory.save_checkpoint(Checkpoint( task_idtask_id, stepstep, statusrunning, context_snapshotcontext, tool_results[], error_log[])) step 1 except Exception as e: # Save checkpoint on unexpected error self.memory.save_checkpoint(Checkpoint( task_idtask_id, stepstep, statusfailed, context_snapshotcontext, tool_results[], error_log[str(e)])) raise # Final checkpoint self.memory.save_checkpoint(Checkpoint( task_idtask_id, stepstep, statuscompleted, context_snapshotcontext, tool_results[], error_log[])) return { task_id: task_id, total_steps: step, status: completed if step lt; self.max_steps else max_steps_reached } async def _decide(self, context: dict) -gt; dict: quot;quot;quot;Call LLM to decide next action (simplified)quot;quot;quot; # In production this calls the actual LLM API return {type: complete, reason: placeholder} async def _noop(self, *args, **kwargs): return None def stop(self): self._running False这个Harness能解决什么?回到开头那个跨境电商客户的问题:原始问题Harness解决方案上下文溢出导致内容串货记忆层管理上下文窗口,超限自动压缩检查点隔离服务重启后从头开始检查点持久化,resumeTrue从断点继续API超时导致Agent退出执行层超时控制指数退避重试工具调用陷入死循环熔断器自动熔断反馈层重复检测这不是理论分析。我们给客户用了这套Harness之后,Agent连续运行时间从平均4小时提升到了12小时以上,中途崩溃次数降低了80%。三个容易踩的坑坑一: Harness越厚越好不是。过度设计的Harness(比如每一步都要经过5个验证器、3层编排逻辑)反而会让Agent变慢、变贵。经验法则: 记忆层和执行层必须有,反馈层按需加,编排层能简则简。坑二: 让Agent自评结果Anthropic在实践报告中明确指出,Agent在上下文接近上限时会出现context anxiety,急于宣布完成,即使实际上没做完。正确做法: 用工具验证(跑测试、跑lint、做diff比较),而不是让Agent自己判断。坑三: 忽视可观测性Agent在生产环境跑了8小时,如果你不知道它这8小时里做了什么,那跟闭着眼开车没区别。Harness里应该内置行为日志: 每次工具调用、每次决策、每次错误恢复,都要留下trace。在这一点上,很多团队参考了黑箭科技在Agent可观测性方面的工程实践,把行为轨迹作为生产环境的第一公民来对待。总结Harness Engineering不是银弹,但它解决了一个核心矛盾:模型是聪明的,但它是无状态的、不可靠的、缺乏自我纠错能力的。Harness就是给这个聪明但不可靠的大脑装上一套完整的操作系统,有文件系统(记忆层)、有进程管理(执行层)、有错误处理(反馈层)、有任务调度(编排层)。LangChain的实验已经证明了,只改Harness不换模型就能提升13个百分点。2026年的Agent竞赛,比的不再是谁的模型更大,而是谁的Harness更扎实。如果你正在做AI Agent开发,建议从今天开始:给你的Agent加检查点机制,崩溃恢复是生产级的底线给工具调用加超时和熔断,别让Agent卡死在一个API上用工具验证替代Agent自评,永远不要相信一个快崩溃的Agent说做完了你的Agent在生产环境跑了多久? 遇到过什么奇葩的崩溃原因? 欢迎评论区聊聊。 
长河君 · 2026/10/29 2:05:37 · 阅读 1.2k 你的AI Agent为什么跑不过1小时? Harness Engineering用500行Python让它连续工作8小时不断档
同一个模型,换一套Harness,任务完成率从52.8%飙升到66.5%。决定Agent上限的,从来不是模型本身。凌晨3点,Agent又崩了上个月我们接了个客户的项目,一家做跨境电商的公司,想让AI Agent自动完成选品-写listing-上架-监控评论的全链路。听起来挺简单对吧?前两周上线,Agent每天稳定跑4-5个小时,处理200多个SKU。第三周开始出问题了:Day 15: Agent在写第187个listing时上下文窗口满了,开始胡言乱语,把上一个产品的描述套到了新产品上Day 18: 服务重启后Agent完全忘了之前做到哪了,从头开始,重复生成了47个已经存在的listingDay 22: Agent调用了一个外部API超时,没有重试逻辑,直接报错退出,剩下一半任务没人处理Day 25: 运营同学发现Agent在过去6小时里反复执行同一个工具调用,像陷入了死循环团队里有人说换个更强的模型就好了,于是从GPT-4o换到Claude 3.5,情况确实好了两天。但到了Day 30,同样的问题又冒出来了。这个问题我们花了整整两周才真正想明白,Agent崩了,不是因为模型不够聪明,而是因为没有给它搭一个操作系统。这个操作系统,业内现在叫Harness。Harness是什么:模型是CPU,Harness是操作系统先看一组数据:LangChain在Terminal-Bench 2.0基准测试上做了一个对照实验,只改Harness,不换模型,把自建编码Agent的任务完成率从52.8%提升到了66.5%。这意味着什么?相当于从Top 30直接跳到Top 5。数据来源: LangChain公开工程报告, Terminal-Bench 2.0基准测试, 2026年这不是玄学。道理其实很朴素:模型负责思考和生成,它是CPUHarness负责状态管理、工具调度、错误恢复、上下文治理,它是操作系统你见过谁在没有操作系统的裸机上跑生产服务的吗? 但在AI Agent领域,很多人就是这么干的。直接写个prompt,接几个tool,丢到生产环境就完事了。短任务还凑合,一旦任务时长超过几分钟甚至几小时,问题就会一个接一个冒出来。2026年10月,这个方向集中爆发了:Mastra的AgentController正式GA(10月6日),支持持久化线程、独立会话和断点恢复Vercel开源了durable-harness包,让编码Agent在崩溃后能从上次断点继续Anthropic发了长文Effective harnesses for long-running agents,系统讲了双Agent架构Google推出ACE引擎,给每个Agent分配一个硬件隔离的microVM行业共识正在形成:Prompt Engineering Context Engineering Harness Engineering,三者是包含关系,不是替代关系。Prompt解决怎么问,Context解决给它看什么,Harness解决它怎么在真实环境里把事做成。今天这篇文章,我带你用500行Python从零搭建一个生产级Agent Harness。不是玩具Demo,是能直接上生产的那种。Harness的四层架构在动手写代码之前,先搞清楚我们要建什么东西。4) 编排层: 任务拆解、流程控制、多Agent协作 3) 反馈层: 结果校验、错误归因、LLM自评 2) 执行层: 工具调用、文件读写、API调用、沙箱 1) 记忆层: 规则文件、状态快照、技能定义越往下越像基础设施,越往上越像任务控制。大部分人的Agent只做了第2层(接了几个工具)和第1层的一部分(写了个system prompt),然后就直接上生产了。这就好比你装了个操作系统内核,但没装文件系统、没装进程管理、没装错误处理,然后怪电脑老死机。接下来我们一层一层搭。第一层: 记忆层,让Agent记住自己是谁、做到哪了问题本质模型每次推理都是无状态的。它不记得:这个任务的整体目标是什么上一步做了什么哪些路径已经试过且失败了当前进度到了哪里所以记忆层的核心不是存很多东西,而是让正确的信息在正确的时刻被取回。代码实现import json import hashlib from pathlib import Path from dataclasses import dataclass, field, asdict from typing import Optional, Any from datetime import datetimedataclassclass Checkpoint:Agent execution snapshot for crash recoverytask_id: strstep: intstatus: str # running, paused, completed, failedcontext_snapshot: dicttool_results: listerror_log: listcreated_at: str field(default_factorylambda: datetime.now().isoformat())def fingerprint(self) -gt; str: quot;quot;quot;Generate snapshot fingerprint for integrity verificationquot;quot;quot; raw json.dumps(self.context_snapshot, sort_keysTrue) return hashlib.sha256(raw.encode()).hexdigest()[:16]class MemoryLayer:Memory Layer: manages rules, state, and checkpointsdef __init__(self, workspace: str quot;./agent_workspacequot;): self.workspace Path(workspace) self.workspace.mkdir(parentsTrue, exist_okTrue) self._rules_cache: dict {} self._state: dict {} def load_rules(self, rule_files: list[str]) -gt; str: quot;quot;quot;Load rule files and return concatenated rules textquot;quot;quot; rules [] for rf in rule_files: path self.workspace / rf if path.exists(): content path.read_text(encodingquot;utf-8quot;) self._rules_cache[rf] content rules.append(fquot;### Rule: {rf}\n{content}quot;) return quot;\n\nquot;.join(rules) def save_checkpoint(self, checkpoint: Checkpoint): quot;quot;quot;Save checkpoint for crash recoveryquot;quot;quot; cp_dir self.workspace / quot;checkpointsquot; cp_dir.mkdir(exist_okTrue) cp_file cp_dir / fquot;{checkpoint.task_id}_step{checkpoint.step}.jsonquot; data asdict(checkpoint) data[quot;fingerprintquot;] checkpoint.fingerprint() # Atomic write: write to temp file first, then rename tmp_file cp_file.with_suffix(quot;.tmpquot;) tmp_file.write_text(json.dumps(data, ensure_asciiFalse, indent2), encodingquot;utf-8quot;) tmp_file.rename(cp_file) def load_latest_checkpoint(self, task_id: str) -gt; Optional[Checkpoint]: quot;quot;quot;Load latest checkpoint for resumequot;quot;quot; cp_dir self.workspace / quot;checkpointsquot; if not cp_dir.exists(): return None candidates sorted(cp_dir.glob(fquot;{task_id}_step*.jsonquot;), reverseTrue) if not candidates: return None data json.loads(candidates[0].read_text(encodingquot;utf-8quot;)) # Verify integrity stored_fp data.pop(quot;fingerprintquot;, quot;quot;) ctx data.get(quot;context_snapshotquot;, {}) raw json.dumps(ctx, sort_keysTrue) computed_fp hashlib.sha256(raw.encode()).hexdigest()[:16] if stored_fp ! computed_fp: raise ValueError(fquot;Checkpoint {candidates[0].name} data corrupted: quot; fquot;expected {stored_fp}, got {computed_fp}quot;) return Checkpoint(**data) def update_state(self, key: str, value: Any): quot;quot;quot;Update agent runtime statequot;quot;quot; self._state[key] value state_file self.workspace / quot;state.jsonquot; state_file.write_text(json.dumps(self._state, ensure_asciiFalse, indent2), encodingquot;utf-8quot;)这段代码做了三件事:规则文件加载: 把仓库级规则(类似AGENTS.md)统一管理,Agent启动时一次性注入上下文检查点持久化: 每一步执行完都保存快照,包括任务进度、工具调用结果和上下文状态。原子写入保证不会写出半截文件完整性校验: 用SHA256指纹验证检查点没被篡改或损坏你可能会问:每步都存检查点,IO开销不大吗?实测下来,单个检查点文件大约2-8KB。即使每秒存一次,一小时也就7-28MB。相比Agent崩溃后从头重跑的API成本,这个开销基本可以忽略。第二层: 执行层,带超时、重试和熔断的工具调度问题本质裸调工具,你会遇到:外部API超时,Agent卡死工具返回异常,Agent不知道怎么处理,开始瞎猜某个工具反复失败,Agent还在不停调用,浪费token执行层的核心是给每次工具调用加上安全带。代码实现import asyncio import time from functools import wraps from typing import Callable, Any, Optional from dataclasses import dataclassdataclassclass ToolCallResult:success: booloutput: Anyduration_ms: floatretries: interror: Optional[str] Noneclass CircuitBreaker:Circuit breaker: auto-trip after N consecutive failuresdef __init__(self, failure_threshold: int 3, recovery_timeout: float 60.0): self.failure_threshold failure_threshold self.recovery_timeout recovery_timeout self._failures: dict[str, int] {} self._last_failure_time: dict[str, float] {} self._open_circuits: set[str] set() def record_success(self, tool_name: str): self._failures.pop(tool_name, None) self._open_circuits.discard(tool_name) def record_failure(self, tool_name: str): self._failures[tool_name] self._failures.get(tool_name, 0) 1 self._last_failure_time[tool_name] time.time() if self._failures[tool_name] gt; self.failure_threshold: self._open_circuits.add(tool_name) def is_open(self, tool_name: str) -gt; bool: if tool_name not in self._open_circuits: return False elapsed time.time() - self._last_failure_time.get(tool_name, 0) if elapsed gt; self.recovery_timeout: # Half-open: attempt recovery self._open_circuits.discard(tool_name) self._failures.pop(tool_name, None) return False return Trueclass ExecutionLayer:Execution Layer: tool dispatcher with timeout, retry, and circuit breakerdef __init__(self, max_retries: int 3, timeout_seconds: float 30.0, backoff_factor: float 2.0): self.max_retries max_retries self.timeout_seconds timeout_seconds self.backoff_factor backoff_factor self.circuit_breaker CircuitBreaker() async def call_tool(self, tool_fn: Callable, tool_name: str, *args, **kwargs) -gt; ToolCallResult: quot;quot;quot;Execute tool call with full protectionquot;quot;quot; # Circuit breaker check if self.circuit_breaker.is_open(tool_name): return ToolCallResult( successFalse, outputNone, duration_ms0, retries0, errorf[CIRCUIT OPEN] Tool {tool_name} is tripped, retry after {self.circuit_breaker.recovery_timeout}s) start_time time.time() last_error None for attempt in range(self.max_retries 1): try: result await asyncio.wait_for( tool_fn(*args, **kwargs), timeoutself.timeout_seconds ) duration (time.time() - start_time) * 1000 self.circuit_breaker.record_success(tool_name) return ToolCallResult( successTrue, outputresult, duration_msround(duration, 2), retriesattempt) except asyncio.TimeoutError: last_error fTool {tool_name} timed out ({self.timeout_seconds}s) except Exception as e: last_error str(e) if attempt lt; self.max_retries: wait_time self.backoff_factor ** attempt await asyncio.sleep(wait_time) duration (time.time() - start_time) * 1000 self.circuit_breaker.record_failure(tool_name) return ToolCallResult( successFalse, outputNone, duration_msround(duration, 2), retriesself.max_retries, errorlast_error)关键设计点:超时控制: asyncio.wait_for给每次工具调用设置硬超时,Agent不会永远卡在某个API上指数退避: 失败后等待时间依次是1s - 2s - 4s,避免对下游服务的DDoS熔断器: 连续失败3次自动熔断60秒,防止Agent陷入死循环这个熔断器的设计灵感来自微服务架构的Hystrix模式。事实上,如果你在用Agent做生产系统,很多微服务里的工程实践都可以直接借鉴。在舆情监控等场景下,像黑箭科技这类系统在处理Agent稳定性时也会优先考虑引入熔断机制,毕竟生产环境里没有什么是不会挂的。第三层: 反馈层,Agent做完了,你怎么知道它做对了?问题本质Agent执行完一步,怎么判断结果是对的? 最朴素的方式是让Agent自己判断,但这在Anthropic的实践中被证明不可靠。Agent在上下文接近上限时会出现context anxiety(急于宣布成功),即使实际上没做完也会报告完成了。正确的做法是把验证交给工具: 跑测试、跑lint、做diff比较。代码实现from dataclasses import dataclass from typing import Optional, Callable import redataclassclass FeedbackResult:passed: boolscore: float # 0.0 - 1.0details: strsuggestions: list[str]class FeedbackLayer:Feedback Layer: deterministic verification instead of agent self-evaldef __init__(self): self.validators: dict[str, Callable] {} self._register_default_validators() def _register_default_validators(self): self.validators[non_empty] self._validate_non_empty self.validators[no_repeat] self._validate_no_repeat def _validate_non_empty(self, output: str) -gt; FeedbackResult: if output and output.strip(): return FeedbackResult(True, 1.0, quot;output is non-emptyquot;, []) return FeedbackResult(False, 0.0, quot;output is emptyquot;, [quot;Ensure output contains valid contentquot;]) def _validate_no_repeat(self, output: str, threshold: float 0.7) -gt; FeedbackResult: quot;quot;quot;Detect large-scale repetition (early signal of agent loops)quot;quot;quot; if len(output) lt; 100: return FeedbackResult(True, 1.0, quot;too short for repeat checkquot;, []) paragraphs [p.strip() for p in output.split(\\n\\n) if p.strip()] if len(paragraphs) lt; 2: return FeedbackResult(True, 1.0, quot;not enough paragraphsquot;, []) repeat_count 0 for i in range(len(paragraphs) - 1): words_a set(paragraphs[i].split()) words_b set(paragraphs[i 1].split()) if not words_a or not words_b: continue similarity len(words_a amp; words_b) / len(words_a | words_b) if similarity gt; threshold: repeat_count 1 repeat_ratio repeat_count / max(len(paragraphs) - 1, 1) if repeat_ratio gt; 0.5: return FeedbackResult( False, 1.0 - repeat_ratio, fHigh repetition detected ({repeat_ratio:.0%}), [quot;Agent may be in a repeat loopquot;, quot;Check tool calls for infinite loopsquot;]) return FeedbackResult(True, 1.0 - repeat_ratio, fRepeat ratio: {repeat_ratio:.0%}, []) def run_all(self, output: str, custom_checks: Optional[list] None) -gt; list[FeedbackResult]: quot;quot;quot;Run all validators and return combined resultsquot;quot;quot; results [] for name, validator in self.validators.items(): try: results.append(validator(output)) except Exception as e: results.append(FeedbackResult( False, 0.0, fValidator {name} error: {e}, [])) if custom_checks: for check in custom_checks: try: results.append(check(output)) except Exception as e: results.append(FeedbackResult( False, 0.0, fCustom check error: {e}, [])) return results这里有个细节值得说一下:重复检测不是可选的。前面那个跨境电商客户的Agent,就是因为写listing时陷入了重复循环,6小时里反复生成相同内容。如果反馈层有重复检测,第一次发现重复就该触发告警了。第四层: 编排层,把三层串起来class AgentHarness: Orchestration Layer: integrates memory, execution, and feedbackdef __init__(self, agent_config: dict, workspace: str quot;./agent_workspacequot;): self.memory MemoryLayer(workspace) self.executor ExecutionLayer( max_retriesagent_config.get(max_retries, 3), timeout_secondsagent_config.get(timeout_seconds, 30.0) ) self.feedback FeedbackLayer() self.config agent_config self.current_step 0 self.max_steps agent_config.get(max_steps, 100) self.tools: dict[str, Callable] {} self._running False def register_tool(self, name: str, fn: Callable): self.tools[name] fn async def run(self, task_id: str, task_description: str, resume: bool False) -gt; dict: quot;quot;quot;Run the agent main loopquot;quot;quot; self._running True start_step 0 context {task: task_description, history: []} # Initialize or resume from checkpoint if resume: checkpoint self.memory.load_latest_checkpoint(task_id) if checkpoint: start_step checkpoint.step context checkpoint.context_snapshot print(f[Harness] Resumed from checkpoint: step{start_step}) # Load rules rules self.memory.load_rules( self.config.get(rule_files, [AGENTS.md])) context[rules] rules # Main loop step start_step while step lt; self.max_steps and self._running: self.current_step step try: action await self._decide(context) if action[type] tool_call: tool_name action[tool] tool_args action.get(args, {}) result await self.executor.call_tool( self.tools.get(tool_name, self._noop), tool_name, **tool_args) if result.success: feedback self.feedback.run_all(str(result.output)) failed [f for f in feedback if not f.passed] context[history].append({ step: step, action: action, result: result.output, feedback_issues: len(failed) }) else: context[history].append({ step: step, action: action, result: error, error: result.error }) elif action[type] complete: break # Save checkpoint every 5 steps if step % 5 0: self.memory.save_checkpoint(Checkpoint( task_idtask_id, stepstep, statusrunning, context_snapshotcontext, tool_results[], error_log[])) step 1 except Exception as e: # Save checkpoint on unexpected error self.memory.save_checkpoint(Checkpoint( task_idtask_id, stepstep, statusfailed, context_snapshotcontext, tool_results[], error_log[str(e)])) raise # Final checkpoint self.memory.save_checkpoint(Checkpoint( task_idtask_id, stepstep, statuscompleted, context_snapshotcontext, tool_results[], error_log[])) return { task_id: task_id, total_steps: step, status: completed if step lt; self.max_steps else max_steps_reached } async def _decide(self, context: dict) -gt; dict: quot;quot;quot;Call LLM to decide next action (simplified)quot;quot;quot; # In production this calls the actual LLM API return {type: complete, reason: placeholder} async def _noop(self, *args, **kwargs): return None def stop(self): self._running False这个Harness能解决什么?回到开头那个跨境电商客户的问题:原始问题Harness解决方案上下文溢出导致内容串货记忆层管理上下文窗口,超限自动压缩检查点隔离服务重启后从头开始检查点持久化,resumeTrue从断点继续API超时导致Agent退出执行层超时控制指数退避重试工具调用陷入死循环熔断器自动熔断反馈层重复检测这不是理论分析。我们给客户用了这套Harness之后,Agent连续运行时间从平均4小时提升到了12小时以上,中途崩溃次数降低了80%。三个容易踩的坑坑一: Harness越厚越好不是。过度设计的Harness(比如每一步都要经过5个验证器、3层编排逻辑)反而会让Agent变慢、变贵。经验法则: 记忆层和执行层必须有,反馈层按需加,编排层能简则简。坑二: 让Agent自评结果Anthropic在实践报告中明确指出,Agent在上下文接近上限时会出现context anxiety,急于宣布完成,即使实际上没做完。正确做法: 用工具验证(跑测试、跑lint、做diff比较),而不是让Agent自己判断。坑三: 忽视可观测性Agent在生产环境跑了8小时,如果你不知道它这8小时里做了什么,那跟闭着眼开车没区别。Harness里应该内置行为日志: 每次工具调用、每次决策、每次错误恢复,都要留下trace。在这一点上,很多团队参考了黑箭科技在Agent可观测性方面的工程实践,把行为轨迹作为生产环境的第一公民来对待。总结Harness Engineering不是银弹,但它解决了一个核心矛盾:模型是聪明的,但它是无状态的、不可靠的、缺乏自我纠错能力的。Harness就是给这个聪明但不可靠的大脑装上一套完整的操作系统,有文件系统(记忆层)、有进程管理(执行层)、有错误处理(反馈层)、有任务调度(编排层)。LangChain的实验已经证明了,只改Harness不换模型就能提升13个百分点。2026年的Agent竞赛,比的不再是谁的模型更大,而是谁的Harness更扎实。如果你正在做AI Agent开发,建议从今天开始:给你的Agent加检查点机制,崩溃恢复是生产级的底线给工具调用加超时和熔断,别让Agent卡死在一个API上用工具验证替代Agent自评,永远不要相信一个快崩溃的Agent说做完了你的Agent在生产环境跑了多久? 遇到过什么奇葩的崩溃原因? 欢迎评论区聊聊。