一个语言模型，每次 API 调用都是独立事件。不记得上一轮说了什么不知道三分钟前调过什么工具，更不会在进程崩溃后从断点恢复。它没有真正的主动性——本质上是被一个 while 循环推着走的。

这个 while 循环。加上它周围的一整套基础设施，就是 Harness。

OpenAI 在 年 月 正式把它命名为 "Harness Engineering"，并定义为独立学科。在那之前，Codex 的团队已经了 万行生产代码——零手写——靠的不只是模型有多强。很大程度上是 Harness 足够可靠。

至于公式很简单，Agent = Model + Harness。但做起来不简单，

为什么 Demo 跑通了生产环境一塌糊涂

痛点一：成功率骤降

先看一个真实数字：一个 步的工作流。每步 % 成功率，整条链路跑通的概率只有 %。提到 %，也只有 %，

痛点二：错误累计导致灾难

Agent 不是那种"要么对要么错"的程序。其实，它在每一步都面临多个决策分支，每个分支都在累积不确定性。单步的错误率看似可控，乘起来就是灾难。其实，

痛点三：缺乏自动化自愈能力

Demo 看起来好。是因为你盯着它跑，出了问题你手动纠正、重启上下文、换个问法重来。没有人盯着的时候，它需要一个自动化的"你"——一个知道什么时候重试、什么时候放弃、什么时候换方案的运行时程序。

这就是 Harness 要解决的问题。

年的一条主要结论来自 LangChain 自己的实践：同一个模型。Harness 改进后Terminal Bench 排名从第 跳到第，分数从 % 提到 % . Harness 设计的差异可以拉开 分的任务完成率差距，模型没变。

架构拆解这方面。一个 Harness 由哪些层组成

年 月的 arXiv 论文《Architectural Design Decisions in Harnesses》对 个开源 Agent 程序做了实证分析，结论是 % 的程序采用 Agent Loop 作为主要执行模式。但 Loop 只是骨架，真正的差异在上面长出的五层结构。

从第一层来看，推理-行动循环

痛点四：推理与行动脱节导致“脑补”错误

ReAct 循环是 Harness 的心脏。它的结构简单到只有四步循环：

• Thought → Action → Observation → Thought…怎么说呢，

This loop originates from Yao et al.'s ReAct paper and forces model to verify its thoughts with real-world data instead of hallucinating.

The loop solves two fatal issues of pure CoT:

• The reasoning process is isolated from external world;wrong assumptions cannot be corrected.
• The model tends to “hallucinate” facts instead of querying for verification.

说到第二层。工具调用协议

痛点五：工具爆炸式增长带来的决策噪声和成本膨胀

• 工具数量： Vercel 在 年 的实验表明，将 Text-to-SQL Agent 从 - 个专用工具砍到 个，成功率从 % 跳到 %，速度快了 倍，token 消耗减少 %。
• Scheam 设计： Anthropic 在 “Building Effective Agents” 中提出 ACI 原则。要像 UI 那样严谨地写工具描述，包括示例、边界条件和强制格式。
• MCP 标准化： MCP 把 N×M 的集成问题降为 N+M，实现即插即用；怎么说呢，截至 年 月。MCP SDK 下载量超过,万次。
• A2A 协议补足 Agent 间通信，使环境更加完整。老实说，

第三层的观点是。上下文管理

痛点六：长上下文导致噪声淹没关键信息 & Token 成本线性增长

• 静态提示词前置： 程序身份、通用规则和工具 Schema 固定在上下文开头，以最大化 Prefix Caching 收益。
• 会话历史压缩： 每 N 轮进行一次摘要压缩；压缩策略直接影响记忆质量，需要在“保留关键信息”和“降低噪声”之间平衡。
• # 工具输出卸载： 大量搜索结果或文件内容存入向量库/SQLite。仅在需要时检索，从而保持对话窗口轻量化。

至于第四层，状态持久化

Eureka 痛点七：崩溃后丢失全部进度 —— 没有持久化就等于没有生产价值。

• # Event Sourcing： 每个事件以 JSONL 行追加写入磁盘。即使进程在第 X 行崩溃，第 X‑1 行之前的数据仍然安全可恢复。
• # Ralph Loop： 当上下文接近极限时通过 Stop Hook 将当前状态写盘并注入干净上下文继续执行，实现跨越 Context Boundary 的无缝恢复。
• 再看beyond，状态永远驻留磁盘。上下文仅作缓存使用，这也是许多大公司正式投入生产的关键原则之一。

至于第五层，错误恢复

Eureka 痛点八：传统二元错误模型无法覆盖 LLM 特有的“软错误”。

• # 重试：  指数退避 + 随机抖动，降低 -% 的重试风暴；话说回来，适用于网络超时、API 限流等瞬时故障。
• # 降级：  当主模型不可用或输出质量低于阈值时切换至备选模型 → 本地小模型 → 缓存结果，实现服务连续性。
• # 熔断：  连续 N 次失败后打开熔断器。暂停调用 M 秒再半开探测，以防雪崩效应蔓延至上游程序。
• # 重规划：  当工具返回结果与当前计划冲突时回到 Planner 重新生成计划，而不是盲目继续错误方法。
`
• # 人工升级：  所有自动恢复耗尽后暂停 Agent 并通知人工干预。这是最终安全网，也是合规要求常见条款之一。`

从零建立的观点是，用 Python 搭建一个生产骨架

Coding Goal:不是建立通用框架。而是提供“一键跑通”的参考实现，让你可以直接 copy / modify / extend。全程仅依赖标准库 + OpenAI SDK。

1. 主要循环 — Agent 主 while 循环
2. 工具注册表 — 定义/验证/分发
3. 上下文管理 — 提示词组装 & 压缩
4. 状态持久化 — JSONL 日志 + 崩溃恢复
5. 权限门控 — 工具调用前安全检查
6. 错误恢复 — 指数退避 + 熔断器

至于模块一，工具注册表

import json
from typing import Any,Callable
from dataclasses import dataclass,field
@dataclass
class Tool:
""" 一个工具的完整定义：模型看到的描述 + 运行时函数 """
name这方面，str
description: str
parameters: dict # JSON Schema
从fn来看，Callable
requires_approval: bool = False
def to_openai_schema -> dict:
return {
"type"这方面，"function","function": {
至于"name"，self.name,"description": self.description,"parameters": self.parameters,},}
def validate_args -> dict:
""" 简易 Schema 校验，仅演示版 """
required = self.parameters.get
properties = self.parameters.get
for key in required:
if key not in args:
raise ValueError
for key。value in args.items:
if key in properties:
expected = properties.get
if expected == "integer" and not isinstance:
raise TypeError.__name__}")
return args
class ToolRegistry:
""" 动态注册 / 查询 / 执行 """
def __init__:
self._tools: dict = {}
def register:
self._tools = tool
def get_schemas -> list:
return
def execute -> str:
tool = self._tools.get
if tool is None:
return f"错误：未找到工具 '{name}'"
再看try，validated = tool.validate_args
result = tool.fn
return str
except as e:
return f"参数错误：{e}"
except Exception as e:
return f"执行错误：{e}"

模块二的观点是，上下文管理器

from dataclasses import dataclass,field
from typing import Any
@dataclass
class Message:
说到role，str # system / user / assistant / tool
content: str
tool_call_id: str | None = None
tool_calls: list | None = None
@dataclass
class ContextManager:
""" 管理对话历史 & 自动压缩 """
system_prompt: str
messages: list = field
compact_at: int = 30 # 超过此轮数触发压缩
keep_recent: int = 10 # 压缩后保留最近 N 条
def build_prompt -> list:
payload =
for msg in self.messages:
entry = {"role": msg.role,"content": msg.content}
if msg.tool_call_id:
entry = msg.tool_call_id
if msg.tool_calls:
entry = msg.tool_calls
payload.append
return payload
def add:
self.messages.append)
def maybe_compact -> int:
""" 若历史过长，用 LLM 摘要压缩 """
if len <= self.compact_at:
            return 0
        old = self.messages
        recent = self.messages
        summary_text = "
".join
        compact_prompt = (
            "

⚠️ 痛点九：“压缩太激进会删掉关键步骤”，务必在触发前检测是否出现卡顿或重复操作！​​​​​​​​​​​​​

模块三的观点是，状态持久化

import json
import os
from datetime import datetime
class Journal:
""" JSONL 日志，每行立即落盘，可用于崩溃恢复 """
def __init__:
self.filepath = filepath
# 初始 line 为已有行数或0，以免重复覆盖
self._line_count = sum) if os.path.exists else 0
def append:
safe_payload = {k:v for k,v in payload.items if k not in }
with open as f:
line_obj = {
再看"line"，self._line_count,"timestamp": datetime.now.isoformat,"type": event_type,**safe_payload,}
f.write+"
")
f.flush
os.fsync)
self._line_count += 1
# ---------- 恢复相关 ----------
def replay -> list:
if not os.path.exists:
return
events=
with open as f:
for line in f:
line=line.strip
if line:return events.append)
return events
def restore_messages -> list:
msgs=
for ev in self.replay:
if ev=="user_message":
msgs.append
elif ev=="assistant_message":
m={"role":"assistant"。"content":ev.get}
if "tool_calls" in ev:m=ev
msgs.append
elif ev=="tool_result":
msgs.append({
说到"role"，"tool","tool_call_id":ev,"content":ev
})
return msgs

再看模块四，权限门控

import json
from enum import Enum
class Decision:
ALLOW ="allow"
DENY ="deny"
ASK ="ask"
class PermissionGate:
""" 工具执行前审查 """
DENY_LIST={
: Decision.ASK,: Decision.ASK,: Decision.ASK,: Decision.ALLOW,: Decision.ALLOW,}
@classmethod
def check->Decision:
for,dec in cls.DENY_LIST.items:
if pattern==name:
if pattern=="write_file":
path=args.get
if any for d in ):
return Decision.DENY
return dec
return Decision.ALLOW
@staticmethod
def request_approval->bool:
summary=json.dumps
print？")
resp=input.strip.lower
return resp=="y"

至于模块五，错误恢复

import time,re random
from functools import wraps
from typing import Any
def retry_with_backoff(max_retries:int=5,base_delay:float=1.0,max_delay:float=30.0,retryable:= ):
""" 指数退避 + 随机抖动 """
def decorator:
@wraps
def wrapper:
last_error=None
for attempt in range:
try这方面。return fn
except retryable as e :
last_error=e
if attemptAny :
if self.state=='open':
if time.time-self.last_failure_time>self.recovery_timeout :
self.state='half_open'
else : raise RuntimeError
try :
result=f
if self.state=='closed':self.failure_count=0
elif self.state=='half_open':self.state='closed';self.failure_count=0
return result
except Exception :
self.failure_count+=1;说起来，self.last_failure_time=time.time
if self.failure_count>=self.failure_threshold:self.state='open'
raise

完整组装 —— 把所有模块串起来

import json
import openai
from openai import OpenAI
class AgentHarness:
""" 可直接部署到生产环境的最小可运行引擎 """
MAX_ITERATIONS=50
def __init__(self,system_prompt:str,tools:ToolRegistry,journal:Journal。model:str='gpt-4o'):
self.ctx=ContextManager
self.tools=tools
self.journal=journal
self.model=model
self.client=OpenAI
self.breaker=CircuitBreaker
# ---------- 主入口 ----------
def run->str :
# 恢复历史
restored=self.journal.restore_messages
if restored :
for m in restored:self.ctx.add
self.journal.append
 # 写入使用者任务并记录日志
self.ctx.add
self.journal.append
for iteration in range:
try : response=self._call_model
except RuntimeError :return '错误：模型调用失败，任务中止。'
choice=response.choices
msg=choice.message
# ----- 完整回答 -----
if msg.content and not msg.tool_calls :
# 保存 assistant 最终回复
self.ctx.add
self.journal.append
return msg.content
# ----- 模型请求调用工具 -----
# 保存含 tool_calls 的 assistant 消息
tc_list=
# 添加至上下文 & 日志
self.ctx.add
self.journal.append('assistant_message',{
'content':msg.content or ''。'tool_calls':tc_list})
# 按顺序执行每个 tool_call
for tc in :
decision=PermissionGate.check)
if decision==Decision.DENY :
result=f'权限拒绝：工具 {tc.function.name} 被阻止'
elif decision==Decision.ASK :
approved=PermissionGate.request_approval)
result=self.tools.execute) \
if approved else f'人工拒绝：{tc.function.name}'
else :
result=self.tools.execute)
# 写入 Tool 返回 & 日志
self.ctx.add
self.journal.append('tool_result',{
'tool_name':tc.function.name,'tool_call_id':tc.id,'content':result}))
# 如有必要进行上下文压缩
self.ctx.maybe_compact
return '达到最大迭代次数，任务未完成。'


@retrywithbackoff)
def callmodel:
return self.breaker.call(
self.client.chat.completions.create,model=self.model,messages=self.ctx.buildprompt),tools=self.tools.getschemas,tool_choice='auto')

@retrywithbackoff)
def quicksummarize->str :
resp=self.breaker.call(
self.client.chat.completions.create。model='gpt-4o-mini',messages=
)
return resp.choices.message.content or ''

‬‮‮‪‪‭‭‭       ⁠⁠⁠⁠⁠⁠⁠⁠&&&&&


. 工具是 Harness 主要对象之一 
Datadog “State of AI Engineering ” 报告显示，在生产环境中% 的 LLM 输入 token 是程序提示词——也就是工具定义、行为规则和输出格式。如果 Schema 写得糟糕，即便换更强大的模型也难以提高成功率。把时间花在高质量  上往往能带来比 Prompt 调优更显著 ROI。

 . Planner 与 Generator 分离 
Anthropic 在 年 月发布《Effective Harnesses》里把同级实例拆成 Planner 与 Generator 两个角色，不让同一实例既做计划又执行。从实验来看，两者协作比单体模型提高约%。原因是自我评估不可靠，需要外部审判。. 状态持久化必须从 Day‑Zero 开始 
Demo 阶段常跳过日志落盘，用 `print` 或内存变量凑合。怎么说呢，一旦进入生产，“一次意外 Crash 就把所有进度抹掉”。第一天就实现 Event Sourcing 能省去后期迁移成本。. 分层降级防止单点失效 
推荐至少实现 “Retry → Fallback”。如果仍然频繁出错，再加熔断器与 Re‑plan；最极端情况下才引入 Human‑in‑‑loop。

---