Harness Engineering：当模型能力不再是瓶颈，工程系统的设计质量才是

from dataclasses import dataclass, field
from typing import Callable
import json

@dataclass
class HarnessState:
    """Harness维护的全局状态，跨越Agent的多次调用"""
    todos: list[dict] = field(default_factory=list)
    files: dict[str, str] = field(default_factory=dict)  # 虚拟文件系统
    token_budget: int = 128_000
    tokens_used: int = 0
    verification_log: list[dict] = field(default_factory=list)

class Harness:
    """
    Harness = 模型之外的一切。
    它不负责'想'，负责让'想'的结果变得可控、可验证、可恢复。
    """

    def __init__(self, model_client, tools: dict, system_prompt: str,
                 compaction_threshold: float = 0.4):
        self.model = model_client
        self.tools = tools
        self.system_prompt = system_prompt
        self.state = HarnessState()
        # 上下文使用率超过40%时触发压缩——这是Smart Zone的上限
        self.compaction_threshold = compaction_threshold

    # ---------- 上下文管理 ----------
    def _assemble_context(self, task: str, history: list[dict]) -> list[dict]:
        """
        动态组装上下文：不是把历史全塞进去，而是根据当前任务
        决定'此刻应该看到什么'。
        """
        messages = [{"role": "system", "content": self.system_prompt}]

        # 常驻：当前TODO列表（精简版）
        active_todos = [
            {"id": t["id"], "status": t["status"], "desc": t["desc"][:80]}
            for t in self.state.todos if t["status"] != "completed"
        ]
        if active_todos:
            messages.append({
                "role": "system",
                "content": f"当前任务清单：\n{json.dumps(active_todos, ensure_ascii=False)}"
            })

        # 按需：与当前任务相关的文件内容
        relevant_files = self._retrieve_relevant_files(task)
        if relevant_files:
            messages.append({
                "role": "system",
                "content": f"相关文件：\n{relevant_files}"
            })

        # 历史：最近N轮对话（经过压缩的）
        messages.extend(history[-6:])
        return messages

    def _maybe_compact(self, history: list[dict]) -> list[dict]:
        """
        上下文使用率接近阈值时，压缩早期对话为摘要。
        这是防止'Lost in the Middle'和成本爆炸的关键机制。
        """
        usage = self.state.tokens_used / self.state.token_budget
        if usage < self.compaction_threshold or len(history) < 10:
            return history

        # 保留最近4轮完整对话，其余压缩为摘要
        recent = history[-4:]
        older = history[:-4]
        summary = self._summarize(older)  # 调用模型生成摘要
        compressed = [{"role": "system", "content": f"[历史摘要] {summary}"}]
        compressed.extend(recent)

        self.state.tokens_used = self._estimate_tokens(compressed)
        return compressed

    # ---------- 子代理委派 ----------
    def delegate(self, sub_task: str, sub_agent_prompt: str) -> str:
        """
        将独立的多步骤任务交给子代理执行。
        关键设计：子代理的中间过程不进入主Agent的上下文，
        只有最终结果返回。这防止了上下文污染。
        """
        sub_history = []
        sub_context = [
            {"role": "system", "content": sub_agent_prompt},
            {"role": "user", "content": sub_task}
        ]

        for _ in range(15):  # 子代理有自己的步数上限
            response = self.model.chat(sub_context)
            if response.tool_calls:
                result = self._execute_tool(response.tool_calls[0])
                sub_context.append({"role": "tool", "content": result})
            else:
                # 子代理给出最终答案
                return response.content

        return "[子代理达到步数上限，未完成任务]"

    # ---------- 验证循环 ----------
    def verify(self, action: str, result: str) -> dict:
        """
        验证循环：不让Agent的声明直接成为最终状态。
        每一类关键操作都有对应的验证规则。
        """
        checks = []

        # 规则1：文件操作后验证文件是否真的存在且非空
        if action.startswith("write_file"):
            path = action.split(":")[1].strip()
            content = self.state.files.get(path, "")
            checks.append({
                "name": "file_exists",
                "passed": bool(content),
                "detail": f"{path} {'已写入' if content else '为空或不存在'}"
            })

        # 规则2：代码执行后验证是否返回了预期信号
        if action == "run_tests":
            checks.append({
                "name": "tests_passed",
                "passed": "FAIL" not in result.upper(),
                "detail": result[:200]
            })

        # 规则3：任何声明'完成'的操作必须通过至少一项独立验证
        if "完成" in result or "done" in result.lower():
            checks.append({
                "name": "independent_verification",
                "passed": self._cross_check(),
                "detail": "需要至少一项独立验证确认完成声明"
            })

        all_passed = all(c["passed"] for c in checks)
        self.state.verification_log.append({
            "action": action, "checks": checks, "passed": all_passed
        })

        return {"passed": all_passed, "checks": checks}

    # ---------- 主循环 ----------
    def run(self, task: str, max_steps: int = 30):
        """
        Agent Harness的主循环。
        注意：循环中的大部分步骤不是'调用模型'，而是
        上下文组装、压缩、工具执行、验证和状态更新。
        """
        history = [{"role": "user", "content": task}]
        self.state.todos = self._initial_plan(task)

        for step in range(max_steps):
            # 1. 组装上下文（不是全量历史）
            context = self._assemble_context(task, history)

            # 2. 调用模型（这是循环中唯一'调用模型'的步骤）
            response = self.model.chat(context)

            # 3. 如果有工具调用，执行并验证
            if response.tool_calls:
                for tc in response.tool_calls:
                    result = self._execute_tool(tc)
                    verification = self.verify(tc.name, result)

                    if not verification["passed"]:
                        # 验证失败 → 将失败信息注入下一轮上下文
                        failure_msg = (
                            f"操作 {tc.name} 的验证未通过：\n"
                            f"{json.dumps(verification['checks'], ensure_ascii=False)}\n"
                            f"请分析原因并修正。"
                        )
                        history.append({"role": "system", "content": failure_msg})
                    else:
                        history.append({"role": "tool", "content": result})

            else:
                # 模型给出最终回答
                history.append({"role": "assistant", "content": response.content})
                break

            # 4. 更新状态，必要时压缩上下文
            self._update_todos(response)
            history = self._maybe_compact(history)

        return self._final_report()

    def _execute_tool(self, tool_call) -> str:
        tool_fn = self.tools.get(tool_call.name)
        if not tool_fn:
            return f"未知工具：{tool_call.name}"
        try:
            return str(tool_fn(**tool_call.args))
        except Exception as e:
            return f"执行失败：{type(e).__name__}: {e}"

    def _summarize(self, messages):
        # 实际实现会调用模型生成结构化摘要
        return "[早期对话已压缩]"

    def _estimate_tokens(self, messages):
        return sum(len(m.get("content", "")) // 3 for m in messages)

    def _retrieve_relevant_files(self, task):
        return ""

    def _initial_plan(self, task):
        return []

    def _update_todos(self, response):
        pass

    def _cross_check(self):
        return True

    def _final_report(self):
        return {
            "todos": self.state.todos,
            "verification_log": self.state.verification_log
        }