Harness Engineering：LangChain 不换模型、只改环境，从 Top 30 到 Top 5

Agent Request
  → LocalContextMiddleware      # 映射工作环境（before_model）
  → LoopDetectionMiddleware     # 检测重复失败（wrap_tool_call）
  → ReasoningSandwichMiddleware # 控制推理预算（modify_model_request）
  → PreCompletionChecklistMiddleware # 完成前检查（after_model）
Agent Response

class LocalContextMiddleware(AgentMiddleware):
    """在 Agent 启动时映射工作环境，避免浪费 token 探索。"""

    @before_model
    async def inject_environment(self, ctx):
        env_info = {
            # 目录结构
            "cwd": os.getcwd(),
            "tree": run_cmd("find . -maxdepth 2 -type f | head -100"),

            # 可用工具
            "python": run_cmd("which python3 && python3 --version"),
            "node": run_cmd("which node && node --version"),
            "git": run_cmd("git rev-parse --is-inside-work-tree 2>/dev/null"),

            # 项目信息
            "package_files": glob.glob("**/package.json") + glob.glob("**/pyproject.toml"),
            "test_commands": detect_test_runner(),
        }

        ctx.messages.insert(0, {
            "role": "system",
            "content": f"# Environment Map (pre-loaded)\n{format_env(env_info)}\n"
                       f"Do NOT waste time exploring. Use this map."
        })

class LoopDetectionMiddleware(AgentMiddleware):
    """检测并打断 Agent 对同一文件的重复编辑。"""

    def __init__(self, max_edits_per_file: int = 5):
        self.edit_counts: dict[str, int] = {}
        self.max_edits = max_edits_per_file

    @wrap_tool_call
    async def check_loop(self, tool_call, ctx):
        # 跟踪每个文件的编辑次数
        if tool_call.name in ("edit_file", "write_file"):
            file_path = tool_call.args.get("path", "")
            self.edit_counts[file_path] = self.edit_counts.get(file_path, 0) + 1

            if self.edit_counts[file_path] >= self.max_edits:
                # 注入"退后重新思考"的指令
                ctx.messages.append({
                    "role": "system",
                    "content": (
                        f"LOOP DETECTED: You've edited '{file_path}' "
                        f"{self.edit_counts[file_path]} times. "
                        f"This pattern suggests your current approach is not working.\n\n"
                        f"STOP editing this file. Instead:\n"
                        f"1. Re-read the original task spec\n"
                        f"2. Consider a fundamentally different approach\n"
                        f"3. Check if you misunderstood the requirements"
                    )
                })

        # 继续执行工具调用
        return await tool_call.execute()

class ReasoningSandwichMiddleware(AgentMiddleware):
    """推理三明治：规划和验证用 xhigh，执行用 high。

    核心洞察：最大化推理并不等于最优。
    Terminal Bench 上全程 xhigh 得分 53.9%（超时太多），
    全程 high 得分 63.6%。
    三明治策略 xhigh-high-xhigh 得分 66.5%——
    关键在于把推理预算花在刀刃上。
    """

    @modify_model_request
    async def set_reasoning_budget(self, request, ctx):
        phase = self._detect_phase(ctx)

        if phase == "planning":
            # 规划阶段：充分理解问题
            # Terminal Bench 的任务可以非常复杂
            request.reasoning_effort = "xhigh"

        elif phase == "implementation":
            # 执行阶段：不需要最大推理，节省时间和 token
            request.reasoning_effort = "high"

        elif phase == "verification":
            # 验证阶段：在提交前捕获错误
            request.reasoning_effort = "xhigh"

        return request

    def _detect_phase(self, ctx) -> str:
        """基于对话历史推断当前阶段。"""
        messages = ctx.messages
        tool_calls = [m for m in messages if m.get("role") == "tool"]

        if len(tool_calls) == 0:
            return "planning"        # 还没调用任何工具

        last_msgs = messages[-5:]
        has_test_run = any("test" in str(m) for m in last_msgs)
        has_verification = any("verify" in str(m) or "check" in str(m) for m in last_msgs)

        if has_test_run or has_verification:
            return "verification"    # 在跑测试或检查

        return "implementation"      # 在写代码

class PreCompletionChecklistMiddleware(AgentMiddleware):
    """在 Agent 宣布完成之前，强制执行验证检查清单。

    名称来源：Ralph Wiggum Loop —— "perpetually confused,
    always making mistakes, but never stopping."
    核心：Agent 不能自己判断'做完了'，必须过检查清单。
    """

    @after_model
    async def enforce_checklist(self, response, ctx):
        if not self._agent_wants_to_finish(response):
            return response  # 还在工作，不拦截

        # Agent 想结束了——拦截并强制检查
        checklist = [
            "Did you run the ACTUAL tests (not just re-read your code)?",
            "Did you verify against the ORIGINAL task spec (not your interpretation)?",
            "Did you test edge cases, not just the happy path?",
            "Did you check that file paths match the spec exactly?",
            "Are there any debug prints or temporary code left?",
        ]

        # 注入检查清单，强制 Agent 继续
        ctx.messages.append({
            "role": "system",
            "content": (
                "COMPLETION BLOCKED. Before you can finish, verify:\n\n"
                + "\n".join(f"- [ ] {item}" for item in checklist)
                + "\n\nRun verification NOW. Do not just re-read your code."
            )
        })

        # 返回 None 或修改后的 response，迫使 Agent 继续
        return None  # Agent 被强制进入下一轮

1. Plan   — 理解任务，分解子问题
2. Build  — 实现方案，包括写测试
3. Verify — 对照任务 spec 验证（不是重读代码！）
4. Fix    — 修复验证中发现的问题

最终 Prompt = 用户自定义提示（可选）
            + BASE_AGENT_PROMPT（核心指令）
            + 各中间件注入的指令

第 1 轮：跑 benchmark（52.8%）
  → 收集 LangSmith traces（每个 Agent 动作都记录：延迟、token 数、成本）
  → 用 "Trace Analyzer Skill" 自动分析失败模式
  → 发现：过早自我批准 + Doom Loop 是最大问题

第 2 轮：加入 PreCompletionChecklist + LoopDetection
  → 跑 benchmark（提升到 ~58%）
  → 分析新的 trace，发现环境发现错误

第 3 轮：加入 LocalContext
  → 跑 benchmark（提升到 ~61%）
  → 发现推理预算分配不优

第 4 轮：加入推理三明治
  → 全程 high = 63.6%，三明治 = 66.5%
  → Top 5 ✓

# OpenAI 的做法：强制分层架构
# Types -> Config -> Repo -> Service -> Runtime -> UI

LAYER_ORDER = ["types", "config", "repo", "service", "runtime", "ui"]

def check_import(importing_file: str, imported_module: str) -> str | None:
    importer_layer = get_layer(importing_file)
    imported_layer = get_layer(imported_module)

    if LAYER_ORDER.index(imported_layer) < LAYER_ORDER.index(importer_layer):
        return (
            # 这条错误信息就是给 Agent 看的修复指令
            f"ARCHITECTURE VIOLATION: {importing_file} ({importer_layer}) "
            f"cannot import from {imported_module} ({imported_layer}).\n"
            f"\n"
            f"ALLOWED: Code can only depend forward: "
            f"Types → Config → Repo → Service → Runtime → UI\n"
            f"\n"
            f"FIX OPTIONS:\n"
            f"1. Move shared types to the 'types' layer\n"
            f"2. Use a Provider interface for cross-cutting concerns\n"
            f"3. See: docs/architecture/layering.md for examples\n"
        )
    return None

AGENTS.md（~100 行，始终注入 context）
  ├── 指向 docs/design/         — 设计文档
  ├── 指向 docs/architecture/   — 架构文档
  ├── 指向 docs/quality/        — 质量评估
  └── 指向 docs/plans/          — 执行计划

具体知识存在 docs/ 目录下，按需加载。

# 跑一轮 benchmark，收集 traces
deepagents run --benchmark your-eval --trace langsmith

# LangChain 的方法：用 Agent 分析 Agent 的 trace
from langsmith import Client

client = Client()
runs = client.list_runs(project_name="my-agent-eval", is_root=True)

# 按失败原因分类
failure_categories = {}
for run in runs:
    if run.status == "error" or not run.outputs.get("pass"):
        category = classify_failure(run.trace)  # 人工或 LLM 分类
        failure_categories.setdefault(category, []).append(run)

# 找到最高频的失败模式
for cat, runs in sorted(failure_categories.items(), key=lambda x: -len(x[1])):
    print(f"{cat}: {len(runs)} failures")

# 每次只加一层，测量效果
middlewares = [
    LocalContextMiddleware(),   # 先解决环境发现问题
]

# 跑 benchmark → 看提升了多少 → 分析新的 top 失败模式
# 然后加下一层

middlewares.append(LoopDetectionMiddleware(max_edits=5))
# 再跑 benchmark → 测量 → 分析...

middlewares.append(ReasoningSandwichMiddleware())
middlewares.append(PreCompletionChecklistMiddleware())

class RippableMiddleware(AgentMiddleware):
    """每个中间件都应该可以被单独移除，不影响其他部分。"""

    def __init__(self, enabled: bool = True):
        self.enabled = enabled

    @before_model
    async def process(self, ctx):
        if not self.enabled:
            return  # 完全跳过

        # ... 实际逻辑