Agent 系统设计：LLM 的固有缺陷与 Harness 工程实践

import re

def search_and_replace(file_path: str, old_text: str, new_text: str) -> dict:
    """Agent-friendly 的代码编辑工具。

    Agent 只需要提供：
    1. old_text: 要被替换的原文（必须在文件中唯一匹配）
    2. new_text: 替换后的新文本

    定位工作由工具完成，不需要 Agent 提供行号。
    """
    with open(file_path, 'r') as f:
        content = f.read()

    # 精确匹配——Agent 不需要数行号
    occurrences = content.count(old_text)

    if occurrences == 0:
        # Agent-friendly error: 告诉 Agent 怎么修，而不只是说"失败了"
        # 尝试模糊匹配，给出修复建议
        lines = content.split('\n')
        candidates = []
        for i, line in enumerate(lines):
            if any(word in line for word in old_text.split()[:3]):
                candidates.append(f"  Line {i+1}: {line.strip()}")

        suggestion = "\n".join(candidates[:3]) if candidates else "No similar lines found."
        return {
            "success": False,
            "error": f"old_text not found in {file_path}. "
                     f"Did you mean one of these?\n{suggestion}\n"
                     f"Re-read the file and provide the exact text to replace."
        }

    if occurrences > 1:
        return {
            "success": False,
            "error": f"old_text matches {occurrences} locations in {file_path}. "
                     f"Provide more surrounding context to make the match unique."
        }

    # 唯一匹配——执行替换
    new_content = content.replace(old_text, new_text, 1)
    with open(file_path, 'w') as f:
        f.write(new_content)

    return {"success": True, "message": f"Replaced 1 occurrence in {file_path}"}

# 工具定义的 JSON Schema —— Agent 只能在这个范围内操作
tools = [
    {
        "type": "function",
        "function": {
            "name": "search_files",
            "description": "Search for files matching a glob pattern",
            "parameters": {
                "type": "object",
                "properties": {
                    "pattern": {
                        "type": "string",
                        "description": "Glob pattern, e.g. '**/*.py'"
                    },
                    "path": {
                        "type": "string",
                        "description": "Directory to search in",
                        "default": "."
                    }
                },
                "required": ["pattern"],
                # strict: true 阻止 Agent 添加未定义的参数
                "additionalProperties": False
            },
            "strict": True  # 严格模式：Agent 不能发明新参数
        }
    }
]

def validate_and_execute(tool_name: str, params: dict) -> dict:
    """校验工具参数，返回 agent-friendly 错误信息。"""

    if tool_name == "schedule_event":
        # 校验日期格式
        if "start_date" in params:
            if not re.match(r'\d{4}-\d{2}-\d{2}', params["start_date"]):
                return {
                    "error": "start_date must be YYYY-MM-DD format. "
                             "Example: 2025-09-01. "
                             "You provided: " + repr(params["start_date"]),
                    "retry": True  # 明确告诉 Agent 可以重试
                }

        # 校验必填字段
        if "project_id" not in params:
            return {
                "error": "project_id is required. "
                         "Use list_projects tool first to find available project IDs.",
                "retry": True
            }

    # 校验通过，执行工具
    return execute_tool(tool_name, params)

def verify_outcome(agent_claim: str, actual_state: dict) -> bool:
    """Outcome > Narration: 验证环境状态，不信 Agent 的自述。"""

    if agent_claim == "Tests pass":
        # 不信 Agent 的话，自己跑测试
        result = subprocess.run(["pytest", "--tb=short"], capture_output=True)
        return result.returncode == 0

    if agent_claim == "File created":
        return os.path.exists(actual_state["expected_path"])

    if agent_claim == "Bug fixed":
        # 跑完整测试套件 + lint
        test_ok = subprocess.run(["pytest"], capture_output=True).returncode == 0
        lint_ok = subprocess.run(["ruff", "check", "."], capture_output=True).returncode == 0
        return test_ok and lint_ok

class ContextManager:
    """上下文管理器：自动压缩 + 分层加载。"""

    def __init__(self, max_tokens: int = 200_000, compact_threshold: float = 0.4):
        self.max_tokens = max_tokens
        self.compact_threshold = compact_threshold  # OpenAI 发现 40% 是拐点
        self.messages = []

    def should_compact(self) -> bool:
        """超过阈值就触发压缩。"""
        current = self.estimate_tokens()
        return current / self.max_tokens > self.compact_threshold

    def compact(self):
        """压缩策略：优先清理旧的工具输出。"""
        # Step 1: 清理旧的工具输出（最大的 token 消耗源）
        for msg in self.messages[:-10]:  # 保留最近 10 条
            if msg.get("role") == "tool":
                msg["content"] = self._summarize(msg["content"])

        # Step 2: 如果还是太大，摘要早期对话
        if self.should_compact():
            early = self.messages[:len(self.messages)//2]
            summary = self._summarize_conversation(early)
            self.messages = [
                {"role": "system", "content": f"Earlier conversation summary:\n{summary}"}
            ] + self.messages[len(self.messages)//2:]

    def _summarize(self, content: str) -> str:
        """截断长工具输出，保留关键信息。"""
        if len(content) > 2000:
            return content[:1000] + "\n...[truncated]...\n" + content[-500:]
        return content

# Claude 打 Pokemon 的实践：通过 NOTES.md 跨 context reset 保持精确状态
class ExternalMemory:
    def __init__(self, path="NOTES.md"):
        self.path = path

    def save_progress(self, task_id: str, status: str, findings: list[str]):
        """每完成一步就写文件——context reset 后仍可恢复。"""
        with open(self.path, 'a') as f:
            f.write(f"\n## Task {task_id}: {status}\n")
            for finding in findings:
                f.write(f"- {finding}\n")

    def load_progress(self) -> str:
        """新会话开始时，先读取之前的进度。"""
        if os.path.exists(self.path):
            with open(self.path) as f:
                return f.read()
        return ""

Progressive Disclosure 策略
├── L0: 全局地图（文件树 + 模块摘要）     < 2K tokens   ← 始终在 context 里
├── L1: 当前任务相关文件列表              < 5K tokens   ← 任务开始时加载
└── L2: 具体代码片段（只读相关函数）       < 30K tokens  ← 按需检索

# 自定义 linter 规则：强制分层架构
# OpenAI 的做法：每个 domain 内代码只能 "forward" 依赖
# 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 (
            f"ERROR: {importing_file} (layer: {importer_layer}) "
            f"cannot import from {imported_module} (layer: {imported_layer}).\n"
            f"FIX: Move shared types to the 'types' layer, "
            f"or use the Provider interface for cross-cutting concerns.\n"
            f"SEE: docs/architecture/layering.md"
        )  # ← 这条错误信息就是给 Agent 看的修复指令
    return None

repo/
├── ARCHITECTURE.md        # 高层架构概览（< 500 行）
├── docs/
│   ├── architecture/
│   │   ├── layering.md    # 分层规则
│   │   └── providers.md   # 跨切面接口
│   └── decisions/
│       ├── 001-use-react.md   # ADR: 为什么用 React
│       └── 002-auth-flow.md   # ADR: 认证流程
└── CLAUDE.md              # Agent 专用指令

class AgentHarness:
    """LangChain 风格的中间件管道。"""

    def __init__(self):
        self.middlewares = [
            LocalContextMiddleware(),    # 映射工作环境
            LoopDetectionMiddleware(),   # 防止重复失败
            ReasoningSandwichMiddleware(), # 推理三明治
            PreCompletionChecklistMiddleware(), # 完成前检查
        ]

    async def run(self, task: str) -> str:
        context = {"task": task, "messages": []}
        for mw in self.middlewares:
            context = await mw.process(context)
        return context["result"]


class LocalContextMiddleware:
    """映射工作环境，避免 Agent 浪费时间探索。"""

    async def process(self, ctx: dict) -> dict:
        # 自动注入：目录结构、可用工具、Python 版本等
        env_info = {
            "cwd": os.getcwd(),
            "python_version": sys.version,
            "available_tools": list_tools(),
            "directory_structure": get_tree(".", max_depth=2),
        }
        ctx["messages"].insert(0, {
            "role": "system",
            "content": f"Environment:\n{json.dumps(env_info, indent=2)}"
        })
        return ctx


class LoopDetectionMiddleware:
    """检测并阻断重复失败的循环。"""

    def __init__(self, max_retries: int = 3):
        self.action_history = []
        self.max_retries = max_retries

    async def process(self, ctx: dict) -> dict:
        last_action = ctx.get("last_action")
        if last_action:
            same_count = sum(1 for a in self.action_history[-5:] if a == last_action)
            if same_count >= self.max_retries:
                ctx["messages"].append({
                    "role": "system",
                    "content": f"LOOP DETECTED: You've tried '{last_action}' "
                               f"{same_count} times and it keeps failing. "
                               f"Try a completely different approach."
                })
            self.action_history.append(last_action)
        return ctx


class ReasoningSandwichMiddleware:
    """推理三明治：关键决策用高推理，执行用低推理。

    LangChain 发现：xhigh 推理全程用反而得分低（53.9%，超时太多），
    high 推理得 63.6%。最佳策略是 xhigh-high-xhigh（三明治）：
    - 规划阶段：xhigh reasoning（仔细想）
    - 执行阶段：high reasoning（快速做）
    - 验证阶段：xhigh reasoning（仔细查）
    """

    async def process(self, ctx: dict) -> dict:
        phase = ctx.get("phase", "plan")
        if phase in ("plan", "verify"):
            ctx["reasoning_effort"] = "high"    # 关键决策用高推理
        else:
            ctx["reasoning_effort"] = "medium"  # 执行阶段用快速推理
        return ctx

async def agent_loop(user_input: str, tools: list, max_iterations: int = 50):
    """Agent 核心循环——用户输入到最终回复。

    来源：OpenAI Codex Agent Loop 架构。
    关键：每次 API 请求都带完整对话历史（stateless），
    支持 Zero Data Retention 合规。
    """
    messages = build_initial_messages(user_input)  # system + tools + user

    for i in range(max_iterations):
        # 1. 调用模型（streaming）
        response = await client.responses.create(
            model="claude-sonnet-4-6",
            input=messages,
            tools=tools,
            stream=True
        )

        # 2. 检查终止条件
        if response.stop_reason == "end_turn":
            # Agent 认为任务完成，返回最终消息
            return response.content

        if response.stop_reason == "tool_use":
            # 3. 提取工具调用
            tool_calls = extract_tool_calls(response)

            for call in tool_calls:
                # 4. 执行工具（在沙箱中）
                result = await execute_in_sandbox(call.name, call.arguments)

                # 5. 追加结果到消息历史
                messages.append({"role": "assistant", "content": response.content})
                messages.append({
                    "role": "tool",
                    "tool_use_id": call.id,
                    "content": truncate(result, max_tokens=25_000)  # 截断长输出
                })

            # 6. 回到步骤 1（循环继续）
            continue

    # 达到最大迭代次数
    return "Max iterations reached. Progress saved to NOTES.md."

请求结构（按优先级递减）：
┌─────────────────────────────────────────┐
│ Server System Message (最高优先级)       │  ← 静态，缓存友好
│ Tool Definitions                        │  ← 静态，缓存友好
│ Client Instructions (CLAUDE.md 等)      │  ← 较静态，缓存友好
├─────────────────────────────────────────┤
│ Message History                         │  ← 变化的部分
│   ├── User Message 1                    │     只追加不修改
│   ├── Assistant Response 1              │     以保留前缀
│   ├── Tool Result 1                     │
│   ├── ...                               │
│   └── User Message N (最新)             │
└─────────────────────────────────────────┘

缓存命中条件：前缀完全一致。
所以：静态内容在前，变化内容追加在后。

# 强制 Agent 输出结构化的思考-行动-观察三元组
step_schema = {
    "type": "object",
    "properties": {
        "thought": {
            "type": "string",
            "description": "你的推理过程：当前状态分析 + 下一步计划"
        },
        "action": {
            "type": "string",
            "enum": ["search_code", "read_file", "edit_file", "run_test",
                     "run_lint", "commit", "ask_human", "done"],
            "description": "要执行的动作"
        },
        "action_input": {
            "type": "object",
            "description": "动作的参数"
        },
        "confidence": {
            "type": "number",
            "minimum": 0, "maximum": 1,
            "description": "对当前方向的信心分数"
        }
    },
    "required": ["thought", "action", "action_input", "confidence"]
}

# 当信心分数低于阈值时，强制 Agent 回到 Plan 阶段
if step["confidence"] < 0.3:
    messages.append({
        "role": "system",
        "content": "LOW CONFIDENCE DETECTED. Re-read the exec-plan and "
                   "reconsider your approach before continuing."
    })

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

    async def process(self, ctx: dict) -> dict:
        if ctx.get("agent_wants_to_finish"):
            checklist = [
                ("tests_pass", "Run ALL tests, not just the ones you wrote"),
                ("lint_clean", "Run linter on changed files"),
                ("spec_match", "Re-read the original task spec and verify each requirement"),
                ("no_debug_code", "Remove any debug prints or temporary code"),
                ("edge_cases", "Test at least 2 edge cases"),
            ]

            failed = []
            for check_id, instruction in checklist:
                result = await self.run_check(check_id, ctx)
                if not result["pass"]:
                    failed.append(f"FAILED: {instruction}\n  Reason: {result['reason']}")

            if failed:
                ctx["agent_wants_to_finish"] = False
                ctx["messages"].append({
                    "role": "system",
                    "content": "COMPLETION BLOCKED. Fix these issues:\n" +
                               "\n".join(failed)
                })
        return ctx

async def agent_code_review(pr_diff: str, spec: str) -> dict:
    """独立 Reviewer Agent：全新 context，没有 Writer 的记忆。"""

    review_prompt = f"""Review this code change against the spec.

SPEC: {spec}

DIFF:
{pr_diff}

Check for:
1. Does the change actually satisfy the spec? (not just "looks reasonable")
2. Are there edge cases not handled?
3. Are there security issues (injection, XSS, etc.)?
4. Are there performance issues?

Be specific. If you find issues, show the exact line and explain why."""

    # 关键：这是一个全新的 context window
    # Reviewer 不知道 Writer 的推理过程，只看代码本身
    response = await client.messages.create(
        model="claude-sonnet-4-6",
        messages=[{"role": "user", "content": review_prompt}],
        max_tokens=4096
    )

    return parse_review(response.content)

# 一个设计良好的搜索工具——遵循所有 10 条原则
search_tool = {
    "type": "function",
    "function": {
        "name": "search_codebase",  # 原则 1: 动作 + 对象
        "description": (
            "Search for code patterns in the repository. "
            "Returns matching file paths and line numbers. "
            "Use this instead of reading entire files — it's faster and "
            "uses less context."  # 告诉 Agent 什么时候用这个工具
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "pattern": {
                    "type": "string",
                    "description": (
                        "Regex pattern to search for. "
                        "Examples: 'def process_payment', 'class.*Handler', "
                        "'TODO|FIXME|HACK'"  # 原则 8: 提供输入示例
                    )
                },
                "file_type": {
                    "type": "string",
                    "description": "Filter by file extension",
                    "enum": ["py", "ts", "js", "go", "rs", "java", "all"],
                    "default": "all"  # 原则 2: 可选参数有默认值
                },
                "max_results": {
                    "type": "integer",
                    "description": "Max results to return (default 20, max 100)",
                    "default": 20,
                    "maximum": 100  # 原则 7: 控制输出大小
                }
            },
            "required": ["pattern"],  # 原则 2: 最少必填参数
            "additionalProperties": False
        },
        "strict": True
    }
}

async def plan_then_execute(task: str):
    """规划用 Opus，执行用 Sonnet。"""

    # Phase 1: 规划（高推理模型）
    plan = await client.messages.create(
        model="claude-opus-4-6",  # 强推理
        messages=[{"role": "user", "content": f"Plan steps to: {task}"}],
        max_tokens=2048
    )
    steps = parse_plan(plan.content)

    # Phase 2: 逐步执行（快速模型）
    results = []
    for step in steps:
        result = await client.messages.create(
            model="claude-sonnet-4-6",  # 快速执行
            messages=[{
                "role": "user",
                "content": f"Execute this step:\n{step}\n\nContext:\n{results[-3:]}"
            }],
            tools=tools
        )
        results.append(result)

    return results

import langsmith

@langsmith.traceable(name="bug_fix_agent")
async def fix_bug(issue: str):
    """每个工具调用、LLM 调用、状态转换都被 trace。"""

    # 这些调用会自动记录到 LangSmith trace
    files = await search_code(issue)       # trace: search_code
    context = await read_files(files[:3])   # trace: read_files
    fix = await generate_fix(context)       # trace: generate_fix (LLM call)
    result = await apply_and_test(fix)      # trace: apply_and_test

    return result

# 事后分析：
# - 哪些步骤耗时最长？
# - 哪些工具调用失败了？
# - Agent 是否进入了无效循环？
# - Token 消耗分布如何？

# Step 1: 让 Agent 写测试（明确禁止写实现）
claude "Write failing tests for the payment retry logic.
       DO NOT write any implementation code."

# Step 2: 提交测试
git add tests/ && git commit -m "Add payment retry tests"

# Step 3: 让 Agent 写实现（明确禁止改测试）
claude "Write code to make all tests pass.
       DO NOT modify any test files."

async def bug_fix_pipeline(issue_url: str):
    """端到端 Bug 修复 Agent——综合运用所有设计原则。"""

    # ===== 阶段 1: 理解 =====
    issue = await fetch_issue(issue_url)

    # 用 code search 工具，不让 Agent 猜路径 (§1 精确计算)
    related_files = await search_codebase(
        pattern=extract_keywords(issue.body),
        file_type=detect_language(issue)
    )

    # 复现 bug——验证确实失败 (§2 幻觉防御)
    repro_result = await run_tests(filter=issue.labels)
    assert not repro_result.all_pass, "Bug not reproduced, investigate further"

    # ===== 阶段 2: 定位 =====
    # 只加载相关代码，最小化 context 占用 (§3 上下文)
    context = await read_files(
        related_files[:5],  # 最多 5 个文件
        sections="relevant_functions_only"  # 只读相关函数
    )

    # 生成假设，写入外部计划 (§6 规划)
    plan = await generate_plan(issue, context)
    await save_to_file("exec-plan.md", plan)

    # ===== 阶段 3: 修复 =====
    # 用 SEARCH_AND_REPLACE，不用行号 (§1)
    fix = await generate_fix(plan, context)
    await apply_search_and_replace(fix.file, fix.old_text, fix.new_text)

    # 逐步验证 (§4 Harness)
    test_result = await run_tests()
    if not test_result.all_pass:
        # 不继续下一步，立即修复
        await retry_fix(test_result.failures, plan)

    lint_result = await run_lint()
    assert lint_result.clean, f"Lint violations: {lint_result.issues}"

    # ===== 阶段 4: 验证 =====
    # Outcome > Narration: 验证环境状态 (§10 评估)
    final_tests = await run_tests(full_suite=True)
    assert final_tests.all_pass, "Full test suite must pass"

    # ===== 阶段 5: 提交 =====
    pr = await create_pr(issue, fix)

    # 独立 Reviewer Agent (§7 自我偏见)
    review = await agent_code_review(pr.diff, issue.body)

    if review.approved:
        return pr.url
    elif review.round < 2:
        # 根据 review 反馈修改，再次提交
        await address_review_comments(review.comments)
        return await resubmit_pr(pr)
    else:
        # 2 轮不过 → 升级给人类 (§11 安全)
        await escalate_to_human(pr, review)