Harness 是什么

"""
最小可用 Harness 骨架。
刻意把六层压缩成 40 行，只为展示「Harness 到底在做什么」。
生产环境每一行都会膨胀成好几十行，但骨架不变。
"""
from dataclasses import dataclass, field
@dataclass
class Budget:
    max_turns: int = 12            # 轮数上限
    max_tokens: int = 120_000      # 累计 token 上限（成本闸）
    max_seconds: float = 300.0     # 挂钟时间上限
    turns: int = 0
    tokens: int = 0
    trace: list = field(default_factory=list)   # 全链路可观测
    def spent(self) -> bool:
        return self.turns >= self.max_turns
class Tool:
    """工具 = 契约。描述、schema、是否幂等、是否有副作用，四者必须齐全。"""
    def __init__(self, name, description, schema, handler, side_effect=False):
        self.name, self.description = name, description
        self.schema, self.handler = schema, handler
        self.side_effect = side_effect      # 有副作用的工具执行前必须确认
TOOLS = {
    "search": Tool("search", "按关键词检索站内文章，返回 [{title,url,snippet}]",
                   {"q": "string"}, lambda **kw: [{"title": "示例", "url": "…"}],
                   side_effect=False),
    "write_file": Tool("write_file", "把内容写入指定路径，会覆盖同名文件",
                       {"path": "string", "content": "string"},
                       lambda **kw: "written", side_effect=True),
}
def build_context(task: str, history: list, observations: list) -> list:
    """
    第 ③ 层的核心：决定模型看到什么。这里用最朴素的三段式，
    真实系统要在这里做预算分配、摘要压缩与状态外置。
    """
    tool_docs = "\n".join(f"- {t.name}: {t.description}" for t in TOOLS.values())
    return [
        {"role": "system", "content": f"你是任务执行器。可用工具：\n{tool_docs}"},
        {"role": "user", "content": f"任务：{task}"},
        *history,
        *observations,
    ]
def classify_error(exc: Exception) -> str:
    """错误分类决定回喂方式，这是第 ②③ 层交界处最容易被跳过的一步"""
    name = type(exc).__name__
    if name in ("TimeoutError",):
        return "TIMEOUT：可以原样重试一次，超时通常是瞬时的"
    if name in ("ValueError", "KeyError"):
        return "BAD_ARGS：参数有问题，必须改参数后重试，重试原参数无意义"
    if name in ("PermissionError",):
        return "FORBIDDEN：没有权限，重试无用，应上报请求授权"
    return "UNKNOWN：记录并交由上层决定"
def run(task: str, budget: Budget) -> dict:
    history, observations = [], []
    while not budget.spent():
        budget.turns += 1
        # 1. 组装上下文（决定模型看到什么）
        messages = build_context(task, history, observations)
        # 2. 调用模型，要求结构化输出
        plan = call_model(messages, temperature=0)      # 占位
        budget.trace.append({"turn": budget.turns, "plan": plan})
        # 3. 终止判断：模型认为完成
        if plan.get("action") == "finish":
            return {"status": "done", "result": plan.get("answer"),
                    "turns": budget.turns, "trace": budget.trace}
        # 4. 执行工具，带错误分类
        tool = TOOLS.get(plan.get("tool"))
        if tool is None:
            observations.append({"role": "user",
                                 "content": f"工具 {plan.get('tool')} 不存在，请从可用列表中选择。"})
            continue
        if tool.side_effect and not confirm(f"即将执行有副作用的操作 {tool.name}，确认？"):
            observations.append({"role": "user", "content": "用户拒绝了该操作，请换方案或结束。"})
            continue
        try:
            result = tool.handler(**plan.get("args", {}))
            observations.append({"role": "tool", "content": str(result)[:2000]})
        except Exception as e:                            # 关键：错误也回喂
            observations.append({"role": "user",
                                 "content": f"工具执行失败[{classify_error(e)}]：{e}"})
    return {"status": "budget_exhausted", "turns": budget.turns, "trace": budget.trace}