Harness工程实战指南：Lilian Weng的AI自我改进理论与落地实践

class Planner:
    def __init__(self, model_client, max_depth=5):
        self.model = model_client
        self.max_depth = max_depth

    def plan(self, task: str, context: dict) -> list[Task]:
        prompt = f"""
        任务: {task}
        上下文: {json.dumps(context)}

        将此任务分解为子任务。要求:
        1. 每个子任务必须可独立验证
        2. 标注子任务间的依赖关系
        3. 为每个子任务提供回滚策略
        4. 最多{self.max_depth}层嵌套
        """
        response = self.model.complete(prompt)
        return self._parse_tasks(response)

    def replan(self, failed_task: Task, error: str) -> list[Task]:
        """当子任务失败时，动态重新规划"""
        prompt = f"子任务'{failed_task.name}'失败: {error}。请提供替代方案。"
        response = self.model.complete(prompt)
        return self._parse_tasks(response)

class ToolCaller:
    def __init__(self, skills_dir: str):
        self.skills = self._load_skills(skills_dir)

    def _load_skills(self, dir: str) -> dict:
        """从Markdown文件加载技能定义"""
        skills = {}
        for path in Path(dir).glob("*.md"):
            skill = self._parse_skill(path.read_text())
            skills[skill.name] = skill
        return skills

    def execute(self, skill_name: str, params: dict) -> Result:
        skill = self.skills[skill_name]
        # 安全检查
        self._check_permissions(skill, params)
        # 沙箱执行
        with Sandbox() as sb:
            result = sb.run(skill.code, params)
        return result

class Evaluator:
    def __init__(self, model_client, provenance_tracker):
        self.model = model_client
        self.provenance = provenance_tracker

    def evaluate(self, output: str, task: Task) -> EvalResult:
        # 1. 基础质量检查
        quality = self._check_quality(output, task)

        # 2. 来源验证 - 确保输出基于真实工具结果
        provenance = self.provenance.verify(output)

        if not provenance.verified:
            return EvalResult(
                passed=False,
                reason=f"输出来源不可验证: {provenance.unverified_claims}"
            )

        # 3. 测试验证 - 如果有测试用例
        if task.test_cases:
            test_results = self._run_tests(output, task.test_cases)
            if not test_results.all_passed:
                return EvalResult(
                    passed=False,
                    reason=f"测试失败: {test_results.failures}"
                )

        return EvalResult(passed=True, quality_score=quality.score)

import json
from pathlib import Path
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class Task:
    name: str
    description: str
    dependencies: list[str] = field(default_factory=list)
    rollback: Optional[str] = None

class MinimalHarness:
    """最小可用Harness实现"""

    def __init__(self, model_client, workspace: str = ".agent_workspace"):
        self.model = model_client
        self.workspace = Path(workspace)
        self.workspace.mkdir(exist_ok=True)
        self.memory = self._load_memory()
        self.history = []

    def run(self, user_request: str) -> str:
        """主执行循环"""
        # 1. 规划
        tasks = self._plan(user_request)
        results = {}

        for task in tasks:
            # 2. 检查依赖
            if not all(d in results for d in task.dependencies):
                continue

            # 3. 构建上下文
            context = self._build_context(user_request, task, results)

            # 4. 执行
            output = self.model.complete(context)

            # 5. 评估
            eval_result = self._evaluate(output, task)
            if not eval_result["passed"]:
                # 6. 恢复
                if task.rollback:
                    self._rollback(task, results)
                # 重新规划
                tasks = self._replan(task, eval_result["reason"]) + tasks
                continue

            results[task.name] = output
            self.history.append({
                "task": task.name,
                "output": output,
                "source": "model_generation",
                "verified": True
            })

        # 7. 综合结果
        final = self._synthesize(results)
        self._save_memory()
        return final

    def _build_context(self, request, task, results):
        """构建上下文：任务 + 相关记忆 + 前序结果"""
        relevant_memory = self._retrieve_memory(task.description)
        prior_results = {k: v[:200] for k, v in results.items()
                        if k in task.dependencies}

        return f"""
请求: {request}
当前子任务: {task.description}
前序结果: {json.dumps(prior_results, ensure_ascii=False)}
相关记忆: {relevant_memory}
"""

    def _retrieve_memory(self, query: str) -> str:
        """从文件系统检索相关记忆"""
        memory_file = self.workspace / "memory.md"
        if memory_file.exists():
            # 简单的关键词匹配（生产环境应使用向量检索）
            content = memory_file.read_text()
            lines = content.split("
")
            relevant = [l for l in lines if any(
                w in l for w in query.split()[:5]
            )]
            return "
".join(relevant[:10])
        return ""

    def _save_memory(self):
        """将本次会话的关键信息保存到记忆"""
        memory_file = self.workspace / "memory.md"
        with open(memory_file, "a") as f:
            for entry in self.history[-5:]:  # 只保存最近5条
                f.write(f"- [{entry['source']}] {entry['task']}: {entry['output'][:100]}
")

    # _plan, _evaluate, _replan, _rollback, _synthesize 省略...