Harness 运行时：AI Agent 的"执行引擎"——从上下文管理到自适应进化的底层架构深度解析

前言：被忽视的中间层

2026年的今天，大模型的能力边界已经被反复讨论——上下文窗口从128K扩展到1M，推理速度从分钟级压缩到秒级，多模态从文本延伸到了视觉、音频甚至视频。几乎每个月都有新的"最强模型"诞生，每一次发布都伴随着基准测试分数的刷新。

然而，当我们真正去使用这些模型构建生产级 AI Agent 系统时，一个残酷的事实浮现出来：同样的模型权重，换一个 Harness，运行效果能差出十几个百分点。

Harness，中文里很难找到精确的对应词汇。如果勉强翻译，可以叫"运行时框架"或"执行引擎"。它处于模型与外部世界之间，负责把模型的"思考"转化为可执行的动作，把外部反馈编织进模型的"记忆"，把错误恢复机制串联成连贯的体验。

这层中间件，长期被视为"胶水代码"——不够优雅，不够有技术含量，不值得深度投入。但2026年的实践告诉我们，Harness 才是决定 Agent 系统质量的天花板。它决定了模型看到什么、能做什么、错了怎么兜底、成功了如何复用。

本文将系统性地解析 Agent Harness 的底层架构，从上下文管理、工具调用、错误恢复、记忆系统到自适应进化，配合完整的代码实现，帮助读者真正理解为什么 Harness 才是 AI Agent 工程化的核心战场。

一、问题本质：为什么 Harness 如此关键

1.1 模型能力 ≠ Agent 效果

让我们用一个具体场景来说明。假设你需要用 Claude Code 这样的终端 Agent 修复一个 GitHub issue：

1. 你把 issue 描述扔给模型
2. 模型开始阅读代码、分析问题
3. 模型执行 git 操作、编写代码
4. 模型运行测试、发现失败
5. 模型分析失败原因、调整代码
6. 再次测试……

这个循环可能持续几十轮。在这个过程中，模型本身的"聪明程度"只是成功的一个因素。真正决定成败的，是：

• 上下文窗口管理：哪些信息该保留、哪些该丢弃、保留的顺序如何
• 工具定义质量：工具的描述是否清晰、参数 schema 是否无歧义
• 错误恢复机制：测试失败后模型知道该怎么分析吗，还是直接放弃
• 记忆复用能力：曾经踩过的坑，这次是否避免了重蹈
• 执行状态连续性：长时间任务中断后能否无缝恢复

这些全部由 Harness 负责。模型只是一个"强大的推理引擎"，Harness 才是把推理结果转化为真实价值的"执行系统"。

1.2 真实数据：Harness 差异带来的性能鸿沟

SWE-bench 是一个评估 AI Agent 解决真实软件工程问题能力的标准测试集。2025年底，Claude Code 首次在这个测试集上突破了 72.5% 的 solve rate，引发了社区的广泛关注。

但鲜为人知的是，同一个 Claude 模型，在不同的 Harness 配置下，SWE-bench 的得分波动范围超过 20 个百分点：

这不是因为模型变了，而是 Harness 实现了更精细的上下文管理、更智能的失败恢复策略、以及跨任务记忆复用。

1.3 Harness 的技术边界

Harness 不是万能的。它需要处理的问题包括：

上下文窗口的稀缺性：即使是最新的 1M token 上下文窗口，在处理大型代码库时仍然不够用。Harness 需要智能地决定保留什么、压缩什么、丢弃什么。

工具调用的可靠性：当模型决定调用一个工具时，Harness 需要确保调用的语义正确、参数完整、超时合理、结果可解析。

长时间运行的稳定性：一个可能持续数小时的代码修复任务，中间的任何环节出错都需要能恢复，而不是从头开始。

多轮对话的连贯性：Agent 不同于单轮对话，需要在几十甚至上百轮交互中保持目标一致性。

接下来，我们深入 Harness 的每一个核心模块。

二、上下文管理：让有限窗口发挥无限价值

2.1 上下文管理的核心挑战

大模型的上下文窗口是一个固定容量的"工作记忆"。当处理真实项目时，这个容量很快就会被填满：

• 一个中等规模的代码库可能有数十万行代码
• 一次代码修复过程可能产生数千行日志
• 错误堆栈、测试输出、代码 diff……每一项都在消耗上下文

Harness 的第一个职责，就是在这个有限空间内，动态维护一个"最有价值的上下文状态"。

2.2 上下文分层策略

一个成熟的 Harness 通常将上下文分为以下几层：

第一层：系统指令层（System Prompt）

这一层包含 Agent 的角色定义、行为规范、工具定义。这一层的内容相对稳定，但也会根据任务类型动态调整。

class SystemPromptManager:
    """系统提示词管理器，支持动态组合"""

    def __init__(self):
        self.base_instructions = """
你是一个专业的软件工程师，擅长代码分析、问题诊断和修复。
你的工作风格是：先理解问题，再制定计划，最后精确执行。
遇到不确定的问题时，你会先做假设，然后通过实验验证假设。
"""
        self.task_specific_rules = []

    def build_system_prompt(self, task_type: str, tools: list[str]) -> str:
        """构建完整的系统提示词"""
        rules = "\n".join(self.task_specific_rules)
        tool_defs = self._render_tool_definitions(tools)

        return f"""{self.base_instructions}

## 任务类型
{task_type}

## 行为规则
{rules}

## 可用工具
{tool_defs}

## 输出格式
当你需要使用工具时，必须输出以下格式：
<tool_call>
name: <工具名称>
arguments: <JSON格式的参数>
</tool_call>

当你完成任务时，输出：
<done>
summary: <任务总结>
result: <最终结果>
</done>
"""

第二层：任务上下文层（Task Context）

这一层包含当前任务的目标、历史操作、关键发现。这一层是动态变化的，也是 Harness 管理的核心区域。

from dataclasses import dataclass, field
from typing import Any
from enum import Enum
import time

class ContextPriority(Enum):
    """上下文片段的优先级"""
    CRITICAL = 3   # 必须保留（如关键决策点）
    HIGH = 2       # 高价值（如操作结果）
    MEDIUM = 1     # 中等价值（如日志片段）
    LOW = 0        # 可丢弃（如重复信息）

@dataclass
class ContextSegment:
    """上下文片段"""
    content: str
    priority: ContextPriority
    token_count: int
    created_at: float = field(default_factory=time.time)
    source: str = ""  # 来源：code/log/error/plan
    tags: set[str] = field(default_factory=set)

class ContextWindowManager:
    """上下文窗口管理器——Harness 的核心组件"""

    def __init__(self, max_tokens: int = 100000, reserve_tokens: int = 5000):
        """
        max_tokens: 模型上下文窗口上限
        reserve_tokens: 预留给输出的空间
        """
        self.max_tokens = max_tokens
        self.reserve_tokens = reserve_tokens
        self.available_tokens = max_tokens - reserve_tokens

        # 上下文片段存储（按优先级排序）
        self.segments: list[ContextSegment] = []

        # 当前 token 计数
        self._current_tokens = 0

    def add(self, content: str, priority: ContextPriority,
            source: str = "", tags: set[str] = None) -> None:
        """添加新的上下文片段"""
        # 粗略估算 token 数（中文约 1.5 tokens/字，英文约 4 chars/token）
        estimated_tokens = self._estimate_tokens(content)

        segment = ContextSegment(
            content=content,
            priority=priority,
            token_count=estimated_tokens,
            source=source,
            tags=tags or set()
        )

        self.segments.append(segment)
        self._current_tokens += estimated_tokens

        # 如果超过限制，触发压缩
        if self._current_tokens > self.available_tokens:
            self._compact()

    def _estimate_tokens(self, text: str) -> int:
        """估算 token 数量"""
        chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
        other_chars = len(text) - chinese_chars
        return int(chinese_chars * 1.5 + other_chars / 4)

    def _compact(self) -> None:
        """上下文压缩：当空间不足时，智能丢弃低优先级内容"""
        # 按优先级和新鲜度排序
        self.segments.sort(key=lambda s: (
            s.priority.value,
            -s.created_at  # 新鲜度：越新越重要
        ))

        # 目标：腾出 30% 的空间
        target_tokens = self.available_tokens * 0.7
        removed_tokens = 0
        new_segments = []

        for segment in self.segments:
            if self._current_tokens - removed_tokens <= target_tokens:
                new_segments.append(segment)
            else:
                removed_tokens += segment.token_count

        self.segments = new_segments
        self._current_tokens -= removed_tokens

    def get_context_for_model(self) -> str:
        """生成发送给模型的完整上下文"""
        # 按时间顺序排列（确保连贯性）
        self.segments.sort(key=lambda s: s.created_at)

        return "\n\n".join(seg.content for seg in self.segments)

    def get_summary(self) -> dict:
        """获取当前上下文状态摘要"""
        return {
            "total_tokens": self._current_tokens,
            "available_tokens": self.available_tokens,
            "usage_ratio": self._current_tokens / self.available_tokens,
            "segment_count": len(self.segments),
            "segments_by_priority": {
                p.name: len([s for s in self.segments if s.priority == p])
                for p in ContextPriority
            }
        }

2.3 代码库的智能索引

当 Agent 需要处理大型代码库时，完整加载所有代码不现实。一个成熟的 Harness 会建立代码索引，按需加载：

import subprocess
import json
from pathlib import Path
from dataclasses import dataclass

@dataclass
class FileIndex:
    """文件索引条目"""
    path: str
    file_type: str
    size: int
    summary: str  # AI 生成的摘要
    key_definitions: list[str]  # 关键函数/类定义

class CodebaseIndexer:
    """代码库索引器——让 Agent 按需加载代码"""

    def __init__(self, repo_path: str):
        self.repo_path = Path(repo_path)
        self.index: dict[str, FileIndex] = {}
        self._build_index()

    def _build_index(self) -> None:
        """构建代码库索引"""
        # 使用 tree + wc 快速获取文件结构
        result = subprocess.run(
            ["find", str(self.repo_path), "-type", "f",
             "-name", "*.py", "-o", "-name", "*.go",
             "-o", "-name", "*.ts", "-o", "-name", "*.js"],
            capture_output=True, text=True
        )

        for file_path in result.stdout.strip().split("\n"):
            if not file_path:
                continue

            path = Path(file_path)
            try:
                size = path.stat().st_size
                self.index[str(path)] = FileIndex(
                    path=str(path),
                    file_type=path.suffix,
                    size=size,
                    summary="",  # 后续由 LLM 生成摘要
                    key_definitions=[]
                )
            except Exception:
                continue

    def get_relevant_files(self, query: str, top_k: int = 10) -> list[str]:
        """
        基于查询返回最相关的文件列表。
        实际生产中这里会使用 embedding 相似度匹配，
        这里简化使用关键词匹配演示。
        """
        query_keywords = set(query.lower().split())
        scored = []

        for path, idx in self.index.items():
            # 简单评分：路径名 + 文件类型匹配
            path_lower = path.lower()
            score = sum(1 for kw in query_keywords if kw in path_lower)
            if score > 0:
                scored.append((score, path))

        scored.sort(reverse=True)
        return [path for _, path in scored[:top_k]]

    def load_file_content(self, path: str, max_lines: int = 500) -> str:
        """加载单个文件的内容（带行号，便于引用）"""
        try:
            with open(path, 'r', encoding='utf-8') as f:
                lines = f.readlines()

            # 大文件只加载前后部分
            if len(lines) > max_lines:
                half = max_lines // 2
                content_lines = (
                    lines[:half] +
                    [f"\n... [{len(lines) - max_lines} lines omitted] ...\n\n"] +
                    lines[-half:]
                )
            else:
                content_lines = lines

            numbered = [f"{i+1:4d} | {line}" for i, line in enumerate(content_lines)]
            return f"// File: {path}\n" + "".join(numbered)

        except Exception as e:
            return f"// Failed to load {path}: {e}"

三、工具调用：让模型"行动"而非"空谈"

3.1 工具定义的艺术

工具（Tools）是 Agent 与外部世界交互的接口。工具定义的质量直接决定了 Agent 的行动能力。

一个工具的描述需要包含：

1. 功能描述：这个工具做什么（模型据此决定何时调用）
2. 参数 schema：每个参数的类型、含义、是否必需
3. 返回值说明：成功和失败时分别返回什么
4. 使用示例：帮助模型理解正确的调用方式

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

@dataclass
class ToolDefinition:
    """工具定义"""
    name: str
    description: str
    parameters: dict  # JSON Schema 格式
    handler: Callable[..., Any]
    examples: list[dict] = None

    def to_schema(self) -> dict:
        """导出为 MCP 协议格式的工具 schema"""
        return {
            "name": self.name,
            "description": self.description,
            "inputSchema": {
                "type": "object",
                "properties": self.parameters.get("properties", {}),
                "required": self.parameters.get("required", [])
            }
        }

# 示例：定义一个"搜索代码"工具
search_code_tool = ToolDefinition(
    name="search_code",
    description="""在指定的代码库中搜索包含特定模式的内容。
支持正则表达式，可以搜索函数名、变量名、注释等。
适用于：当你不确定某个功能在哪里实现，或者需要找到包含特定关键词的代码时使用。""",
    parameters={
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "搜索查询，可以是函数名、变量名、或正则表达式"
            },
            "file_pattern": {
                "type": "string",
                "description": "文件匹配模式，如 '*.py'、'src/**/*.go'，默认为 '*'（搜索所有文件）"
            },
            "max_results": {
                "type": "integer",
                "description": "最多返回多少条结果，默认为 20"
            }
        },
        "required": ["query"]
    },
    handler=lambda query, file_pattern="*", max_results=20: search_code(query, file_pattern, max_results),
    examples=[
        {
            "description": "搜索处理用户认证的函数",
            "call": {"query": "def authenticate", "file_pattern": "*.py"}
        },
        {
            "description": "搜索包含 'error' 关键词的日志语句",
            "call": {"query": "logger.error", "file_pattern": "*.go", "max_results": 10}
        }
    ]
)

# 示例：定义一个"执行命令"工具
execute_command_tool = ToolDefinition(
    name="execute_command",
    description="""在终端中执行 shell 命令。
适用于：运行测试、构建项目、git 操作、文件操作等。
注意：
- 这个工具会等待命令完成，超时时间为 60 秒
- STDERR 会和 STDOUT 一起返回
- 如果命令需要交互式输入，请使用 execute_command 中的 stdin 参数
- 危险命令（如 rm -rf）会被记录但不会阻止执行""",
    parameters={
        "type": "object",
        "properties": {
            "command": {
                "type": "string",
                "description": "要执行的命令（完整的 shell 命令字符串）"
            },
            "cwd": {
                "type": "string",
                "description": "命令执行的工作目录，默认为项目根目录"
            },
            "timeout": {
                "type": "integer",
                "description": "超时时间（秒），默认为 60"
            }
        },
        "required": ["command"]
    },
    handler=lambda command, cwd=None, timeout=60: execute_cmd(command, cwd, timeout),
    examples=[
        {
            "description": "运行项目的测试套件",
            "call": {"command": "pytest tests/ -v", "timeout": 120}
        },
        {
            "description": "查看 git 状态",
            "call": {"command": "git status"}
        }
    ]
)

3.2 工具调用的生命周期

当模型决定调用工具时，Harness 需要管理完整的调用生命周期：

import asyncio
from typing import Any, Optional
from dataclasses import dataclass
from enum import Enum
import time

class ToolCallStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"
    TIMEOUT = "timeout"

@dataclass
class ToolCall:
    """工具调用记录"""
    call_id: str
    tool_name: str
    arguments: dict
    status: ToolCallStatus = ToolCallStatus.PENDING
    result: Any = None
    error: str = None
    started_at: float = 0
    completed_at: float = 0
    retries: int = 0

class ToolExecutor:
    """工具执行器——管理工具调用的完整生命周期"""

    def __init__(self, tools: list[ToolDefinition]):
        self.tools = {t.name: t for t in tools}
        self.call_history: list[ToolCall] = []

    async def execute(
        self,
        tool_name: str,
        arguments: dict,
        call_id: str = None,
        max_retries: int = 3
    ) -> ToolCall:
        """执行工具调用，带重试机制"""
        import uuid
        call_id = call_id or str(uuid.uuid4())[:8]

        call = ToolCall(
            call_id=call_id,
            tool_name=tool_name,
            arguments=arguments
        )

        if tool_name not in self.tools:
            call.status = ToolCallStatus.FAILED
            call.error = f"Unknown tool: {tool_name}"
            return call

        tool = self.tools[tool_name]
        call.status = ToolCallStatus.RUNNING
        call.started_at = time.time()

        try:
            # 支持同步和异步 handler
            handler = tool.handler
            if asyncio.iscoroutinefunction(handler):
                result = await asyncio.wait_for(
                    handler(**arguments),
                    timeout=arguments.get("timeout", 60)
                )
            else:
                result = handler(**arguments)

            call.status = ToolCallStatus.SUCCESS
            call.result = result

        except asyncio.TimeoutError:
            call.status = ToolCallStatus.TIMEOUT
            call.error = f"Tool execution timed out after {arguments.get('timeout', 60)}s"

        except Exception as e:
            if call.retries < max_retries:
                # 重试逻辑
                call.retries += 1
                await asyncio.sleep(2 ** call.retries)  # 指数退避
                return await self.execute(tool_name, arguments, call_id, max_retries)
            else:
                call.status = ToolCallStatus.FAILED
                call.error = str(e)

        call.completed_at = time.time()
        self.call_history.append(call)
        return call

    def format_tool_result(self, call: ToolCall) -> str:
        """将工具调用结果格式化为字符串，供模型理解"""
        if call.status == ToolCallStatus.SUCCESS:
            result_str = str(call.result)
            # 截断过长的输出
            if len(result_str) > 3000:
                result_str = result_str[:3000] + f"\n... [truncated, {len(result_str)-3000} chars omitted]"
            return f"[Tool: {call.tool_name}] Success\n{result_str}"
        elif call.status == ToolCallStatus.TIMEOUT:
            return f"[Tool: {call.tool_name}] Timeout: {call.error}"
        else:
            return f"[Tool: {call.tool_name}] Error: {call.error}"

四、错误恢复：让 Agent 从失败中学习

4.1 错误分类与应对策略

Agent 在执行任务时遇到的错误，大致可以分为三类，每类需要不同的恢复策略：

第一类：工具执行错误

工具调用失败（超时、网络错误、参数错误等）。这类错误通常可以通过重试解决。

def handle_tool_error(call: ToolCall, max_retries: int = 3) -> str:
    """生成针对工具执行错误的恢复提示"""
    if call.status == ToolCallStatus.TIMEOUT:
        return f"""工具 {call.tool_name} 执行超时（{call.arguments.get('timeout', 60)}s）。
可能的原因：
1. 命令执行时间过长，可以增加 timeout 参数
2. 系统负载高，重试可能成功
3. 命令在等待某些永远不会发生的事件

建议操作：检查命令本身是否正确，如果是合法的长时间任务，增加 timeout 后重试。"""

    elif "not found" in str(call.error).lower():
        return f"""工具 {call.tool_name} 执行失败：命令不存在。
可能的原因：
1. 命令拼写错误
2. 缺少必要的依赖工具
3. PATH 环境变量问题

建议操作：检查命令是否正确安装，或者使用其他工具达到同样目的。"""

    elif call.retries >= max_retries:
        return f"""工具 {call.tool_name} 重试 {max_retries} 次后仍然失败。
错误信息：{call.error}

建议操作：考虑换一个思路解决这个问题。当前的错误可能表明这个方向走不通。"""

    return f"工具 {call.tool_name} 执行失败：{call.error}，正在重试..."

第二类：业务逻辑错误

代码执行失败、测试不通过、API 返回错误状态等。这类错误需要更深入的分析。

def analyze_business_error(error_output: str, context: str) -> str:
    """分析业务逻辑错误，生成诊断提示"""
    # 提取错误类型
    error_type = classify_error(error_output)

    analysis_prompts = {
        "test_failure": f"""测试失败分析：

错误输出：
{error_output}

上下文：
{context}

分析思路：
1. 首先理解哪个测试失败了——从错误信息中找到 TEST_NAME
2. 理解测试的预期行为——查看测试代码，搞清楚这个测试在验证什么
3. 定位失败原因——通常在 "AssertionError" 或 "Error" 之后的部分
4. 制定修复方案——错误可能是：
   - 代码逻辑错误（需要修改实现）
   - 测试本身有问题（需要修改测试）
   - 环境问题（需要修复测试环境）

请先仔细阅读测试代码，理解它在验证什么，再看实际的错误信息。""",

        "compilation_error": f"""编译错误分析：

错误输出：
{error_output}

分析思路：
1. 编译器会报告第一个错误，后面的错误可能是连锁反应
2. 优先修复第一个错误
3. 常见编译错误类型：
   - 语法错误：缺少分号、括号不匹配、关键字拼写错误
   - 类型错误：类型不匹配、参数数量不对
   - 引用错误：使用了未定义的变量或函数

请仔细阅读编译器的错误信息，定位到具体的文件和行号。""",

        "runtime_error": f"""运行时错误分析：

错误输出：
{error_output}

分析思路：
1. 查看错误类型（TypeError、ValueError、NullPointerException 等）
2. 查看错误发生的调用栈，找到导致错误的关键代码行
3. 分析错误原因：
   - 变量未初始化
   - 类型不匹配
   - 边界条件未处理
   - 并发问题

请从调用栈的底部开始分析，通常最后一行才是真正出问题的地方。"""
    }

    return analysis_prompts.get(error_type, f"遇到错误：{error_output}")

def classify_error(output: str) -> str:
    """分类错误类型"""
    output_lower = output.lower()
    if "test" in output_lower and ("fail" in output_lower or "error" in output_lower):
        return "test_failure"
    elif any(x in output_lower for x in ["compilation error", "syntax error", "error:"]):
        return "compilation_error"
    elif any(x in output_lower for x in ["exception", "traceback", "stack trace"]):
        return "runtime_error"
    return "unknown"

第三类：目标迷失

Agent 在执行过程中偏离了原始目标，或者反复尝试同样的失败路径。

class GoalTracker:
    """目标追踪器——检测并纠正目标迷失"""

    def __init__(self, original_goal: str):
        self.original_goal = original_goal
        self.steps: list[dict] = []
        self.consecutive_failures = 0

    def record_step(self, action: str, result: str, success: bool) -> None:
        """记录执行步骤"""
        self.steps.append({
            "action": action,
            "result_preview": result[:200] if result else "",
            "success": success,
            "timestamp": time.time()
        })

        if not success:
            self.consecutive_failures += 1
        else:
            self.consecutive_failures = 0

    def should_reassess(self) -> bool:
        """判断是否需要重新评估任务方向"""
        # 连续失败超过 5 次，需要重新思考
        if self.consecutive_failures >= 5:
            return True

        # 步骤数量异常多（正常任务不应超过 50 步）
        if len(self.steps) > 50:
            return True

        return False

    def generate_reassess_prompt(self) -> str:
        """生成重新评估提示"""
        steps_summary = "\n".join(
            f"{i+1}. [{'✓' if s['success'] else '✗'}] {s['action']}"
            for i, s in enumerate(self.steps[-10:])  # 最近 10 步
        )

        return f"""你已经执行了 {len(self.steps)} 步操作，但尚未完成任务。

## 原始目标
{self.original_goal}

## 最近的操作记录
{steps_summary}

## 连续失败次数
{self.consecutive_failures}

你可能已经陷入了"局部最优"——在一个方向上反复尝试，但始终无法成功。

请暂停当前的操作，重新思考：
1. 这些失败是否有共同的模式？——如果是，说明方向本身有问题
2. 是否遗漏了什么关键信息？——可能需要补充上下文
3. 是否有更简单的解决路径？——有时候绕路反而更快
4. 任务是否真的可行？——如果缺少必要的权限或依赖，可能需要先解决前置条件

请花一点时间整理思路，然后给出一个明确的新计划。不要重复已经失败的策略。"""

五、记忆系统：让 Agent 记住"不该重蹈的覆辙"

5.1 记忆系统的分层架构

人类解决问题时，会自然地复用过去的经验。Agent 也需要类似的记忆能力。一个完整的 Agent 记忆系统通常分为三层：

感觉记忆（Working Memory）：当前任务内的上下文，已经在前文的 ContextWindowManager 中讨论过。

短时记忆（Session Memory）：当前会话内的经验教训，比如"这个 API 曾经返回 403，所以需要先刷新 token"。

长时记忆（Long-term Memory）：跨会话积累的知识，比如"在这个代码库里，所有的配置都在 config.yaml 中"。

from dataclasses import dataclass, field
from typing import Any
import time
import json
from pathlib import Path

@dataclass
class MemoryEntry:
    """记忆条目"""
    content: str
    tags: list[str]
    created_at: float = field(default_factory=time.time)
    last_accessed: float = field(default_factory=time.time)
    access_count: int = 0
    source: str = "agent"  # agent / user / auto
    relevance_scores: dict[str, float] = field(default_factory=dict)

class AgentMemory:
    """Agent 记忆系统——跨任务知识积累"""

    def __init__(self, storage_path: str = "~/.agent_memory"):
        self.storage_path = Path(storage_path).expanduser()
        self.storage_path.mkdir(parents=True, exist_ok=True)

        self.short_term: list[MemoryEntry] = []  # 当前会话
        self.long_term: list[MemoryEntry] = []   # 持久化存储

        self._load_long_term_memory()

    def remember(self, content: str, tags: list[str],
                 source: str = "agent") -> None:
        """将信息存入记忆"""
        entry = MemoryEntry(
            content=content,
            tags=tags,
            source=source
        )
        self.short_term.append(entry)

        # 如果是重要知识，同步写入长时记忆
        if any(tag in tags for tag in ["important", "lesson", "pattern", "fix"]):
            self.long_term.append(entry)
            self._save_long_term_memory()

    def recall(self, query: str, max_results: int = 5) -> list[MemoryEntry]:
        """检索相关记忆"""
        query_lower = query.lower()
        query_keywords = set(query_lower.split())

        all_memories = self.short_term + self.long_term
        scored = []

        for entry in all_memories:
            # 基于标签匹配
            tag_match = sum(1 for tag in entry.tags if tag in query_keywords)

            # 基于内容关键词匹配
            content_words = set(entry.content.lower().split())
            content_match = len(query_keywords & content_words)

            score = tag_match * 3 + content_match  # 标签权重更高

            if score > 0:
                entry.access_count += 1
                entry.last_accessed = time.time()
                scored.append((score, entry))

        scored.sort(reverse=True)
        return [entry for _, entry in scored[:max_results]]

    def _load_long_term_memory(self) -> None:
        """从磁盘加载长时记忆"""
        memory_file = self.storage_path / "long_term.json"
        if memory_file.exists():
            try:
                with open(memory_file, 'r', encoding='utf-8') as f:
                    data = json.load(f)
                    self.long_term = [MemoryEntry(**e) for e in data]
            except Exception:
                self.long_term = []

    def _save_long_term_memory(self) -> None:
        """持久化长时记忆"""
        memory_file = self.storage_path / "long_term.json"
        # 只保留最近 100 条最重要的记忆
        important = sorted(
            self.long_term,
            key=lambda e: (e.access_count, e.last_accessed),
            reverse=True
        )[:100]

        with open(memory_file, 'w', encoding='utf-8') as f:
            json.dump([vars(e) for e in important], f, ensure_ascii=False, indent=2)

    def format_memory_for_context(self, memories: list[MemoryEntry]) -> str:
        """将记忆格式化为上下文片段"""
        if not memories:
            return ""

        formatted = ["\n## 相关历史经验\n"]
        for entry in memories:
            formatted.append(
                f"- **[{'×' + str(entry.access_count) if entry.access_count > 1 else 'NEW'}]** "
                f"{entry.content}\n  "
                f"标签: {', '.join(entry.tags)}"
            )
        return "\n".join(formatted)

5.2 自动从失败中提取经验

一个成熟的 Harness 会自动从失败中提取经验，存入记忆：

def extract_lesson_from_failure(
    task: str,
    attempted_approaches: list[str],
    failure_reason: str
) -> MemoryEntry:
    """从失败中提取可复用的经验教训"""
    lesson_content = f"""任务：{task}
失败的尝试：{', '.join(attempted_approaches)}
失败原因：{failure_reason}

教训：不要重复同样的失败路径。"""
    return MemoryEntry(
        content=lesson_content,
        tags=["lesson", "failure", "pattern"],
        source="auto"
    )

六、执行状态管理：让长时间任务"断点续传"

6.1 为什么需要状态持久化

一个复杂的代码修复任务可能持续数小时。在此期间，可能发生：

• 网络中断导致会话丢失
• 开发者切换到其他任务
• Agent 被重启
• 模型上下文窗口耗尽

如果没有状态持久化，所有进度都会丢失。Harness 需要支持"断点续传"——保存执行状态，随时可以恢复。

import pickle
import uuid
from dataclasses import dataclass, field, asdict
from typing import Any, Optional
import time

@dataclass
class ExecutionState:
    """执行状态快照"""
    state_id: str
    task_description: str
    current_step: int
    context_snapshot: str
    memory_snapshot: dict
    tool_call_history: list[dict]
    recent_results: list[str]
    created_at: float = field(default_factory=time.time)
    updated_at: float = field(default_factory=time.time)
    metadata: dict = field(default_factory=dict)

class StateManager:
    """状态管理器——支持任务中断恢复"""

    def __init__(self, state_dir: str = "~/.agent_state"):
        self.state_dir = Path(state_dir).expanduser()
        self.state_dir.mkdir(parents=True, exist_ok=True)

        self.current_state: Optional[ExecutionState] = None

    def save_checkpoint(
        self,
        task: str,
        context_mgr: ContextWindowManager,
        memory: AgentMemory,
        executor: ToolExecutor,
        step: int,
        metadata: dict = None
    ) -> str:
        """保存执行状态快照"""
        state_id = str(uuid.uuid4())[:12]

        state = ExecutionState(
            state_id=state_id,
            task_description=task,
            current_step=step,
            context_snapshot=context_mgr.get_context_for_model(),
            memory_snapshot={
                "short_term": [vars(e) for e in memory.short_term],
                "long_term_access_count": {str(i): e.access_count
                    for i, e in enumerate(memory.long_term)}
            },
            tool_call_history=[asdict(c) for c in executor.call_history[-20:]],
            recent_results=[r for r in context_mgr.segments[-5:]],
            metadata=metadata or {}
        )

        state_file = self.state_dir / f"state_{state_id}.pkl"
        with open(state_file, 'wb') as f:
            pickle.dump(state, f)

        self.current_state = state
        return state_id

    def restore_checkpoint(self, state_id: str) -> ExecutionState:
        """恢复执行状态"""
        state_file = self.state_dir / f"state_{state_id}.pkl"
        if not state_file.exists():
            raise FileNotFoundError(f"Checkpoint {state_id} not found")

        with open(state_file, 'rb') as f:
            state = pickle.load(f)

        self.current_state = state
        return state

    def list_checkpoints(self) -> list[dict]:
        """列出所有保存的检查点"""
        checkpoints = []
        for state_file in self.state_dir.glob("state_*.pkl"):
            try:
                with open(state_file, 'rb') as f:
                    state = pickle.load(f)
                checkpoints.append({
                    "state_id": state.state_id,
                    "task": state.task_description[:50],
                    "step": state.current_step,
                    "created_at": state.created_at,
                    "file": str(state_file)
                })
            except Exception:
                continue

        return sorted(checkpoints, key=lambda x: x["created_at"], reverse=True)

6.2 恢复后的"热身"机制

恢复状态后，Agent 需要一个"热身"过程，重新建立对当前状态的认知：

def generate_warmup_prompt(restored_state: ExecutionState) -> str:
    """生成恢复后的热身提示"""
    recent_actions = "\n".join(
        f"- {h['tool_name']}: {h['arguments']}"
        for h in restored_state.tool_call_history[-5:]
    )

    return f"""## 任务恢复

你正在恢复一个之前中断的任务。以下是你之前的进度：

**任务目标**：{restored_state.task_description}

**当前进度**：第 {restored_state.current_step} 步

**最近的操作**：
{recent_actions}

**最近的结果摘要**：
{chr(10).join(restored_state.recent_results[-3:])}

## 恢复后的第一件事

请不要立即继续操作。首先：
1. 理解当前状态——回顾上下文快照，搞清楚现在卡在哪里
2. 确认目标不变——任务目标是否仍然有效
3. 制定下一步计划——基于已有进度，明确接下来该做什么

只有在完全理解当前状态后，才继续执行。避免重复已经失败的尝试。"""

七、自适应 Harness：让执行策略"自己进化"

7.1 核心思路

最前沿的 Harness 研究方向是"自适应"——让 Harness 本身能够根据任务表现动态调整执行策略。这包括：

• 动态工具选择：根据任务类型决定使用哪些工具
• 策略路由：不同类型的任务使用不同的执行策略
• 自我评估：让 Harness 评估自己的表现，并据此调整

from dataclasses import dataclass
from typing import Callable
from enum import Enum

class ExecutionStrategy(Enum):
    """执行策略枚举"""
    CONSERVATIVE = "conservative"   # 保守：每步验证，小步快跑
    AGGRESSIVE = "aggressive"       # 激进：多步并行，快速试错
    PLANNING = "planning"           # 规划优先：先制定计划再执行
    EXPLORATORY = "exploratory"     # 探索：先理解代码库再行动

@dataclass
class StrategyPerformance:
    """策略表现追踪"""
    strategy: ExecutionStrategy
    success_count: int = 0
    failure_count: int = 0
    avg_steps_to_success: float = 0
    total_tokens_used: int = 0

    @property
    def success_rate(self) -> float:
        total = self.success_count + self.failure_count
        return self.success_count / total if total > 0 else 0.0

class AdaptiveHarness:
    """自适应 Harness——根据任务类型和历史表现动态选择策略"""

    def __init__(self):
        self.strategies = {
            ExecutionStrategy.CONSERVATIVE: self._conservative_strategy,
            ExecutionStrategy.AGGRESSIVE: self._aggressive_strategy,
            ExecutionStrategy.PLANNING: self._planning_strategy,
            ExecutionStrategy.EXPLORATORY: self._exploratory_strategy,
        }

        self.strategy_performance: dict[ExecutionStrategy, StrategyPerformance] = {
            s: StrategyPerformance(strategy=s)
            for s in ExecutionStrategy
        }

    def select_strategy(self, task_type: str, task_complexity: float) -> ExecutionStrategy:
        """
        根据任务类型和复杂度选择最佳策略。

        策略选择规则：
        - 简单任务 → CONSERVATIVE（快速完成）
        - 复杂任务 → PLANNING（先规划再执行）
        - 代码库探索 → EXPLORATORY（先理解再行动）
        - 快速试错场景 → AGGRESSIVE（多路径并行尝试）
        """
        # 基于任务类型的启发式选择
        if "explore" in task_type.lower() or "understand" in task_type.lower():
            base_strategy = ExecutionStrategy.EXPLORATORY
        elif "fix" in task_type.lower() or "debug" in task_type.lower():
            base_strategy = ExecutionStrategy.PLANNING
        else:
            base_strategy = ExecutionStrategy.CONSERVATIVE

        # 根据历史表现调整
        base_perf = self.strategy_performance[base_strategy]

        # 如果基础策略历史成功率低于 40%，尝试其他策略
        if base_perf.success_rate < 0.4:
            candidates = [s for s in ExecutionStrategy if s != base_strategy]
            best_alt = max(
                candidates,
                key=lambda s: self.strategy_performance[s].success_rate
            )
            if self.strategy_performance[best_alt].success_rate > base_perf.success_rate:
                base_strategy = best_alt

        # 复杂度调整：高复杂度任务倾向于 PLANNING
        if task_complexity > 0.7 and base_strategy == ExecutionStrategy.CONSERVATIVE:
            base_strategy = ExecutionStrategy.PLANNING

        return base_strategy

    def record_outcome(
        self,
        strategy: ExecutionStrategy,
        success: bool,
        steps: int,
        tokens_used: int
    ) -> None:
        """记录策略执行结果，用于后续优化"""
        perf = self.strategy_performance[strategy]

        if success:
            perf.success_count += 1
            # 增量更新平均步数
            n = perf.success_count
            perf.avg_steps_to_success = (
                (perf.avg_steps_to_success * (n - 1) + steps) / n
            )
        else:
            perf.failure_count += 1

        perf.total_tokens_used += tokens_used

    def _conservative_strategy(self) -> dict:
        """保守策略：每步验证，确保质量"""
        return {
            "max_steps_per_iteration": 1,
            "verify_each_step": True,
            "parallel_tools": False,
            "checkpoint_frequency": 3,
        }

    def _aggressive_strategy(self) -> dict:
        """激进策略：多步并行，快速试错"""
        return {
            "max_steps_per_iteration": 5,
            "verify_each_step": False,
            "parallel_tools": True,
            "checkpoint_frequency": 10,
        }

    def _planning_strategy(self) -> dict:
        """规划优先：先制定详细计划再执行"""
        return {
            "require_plan_before_action": True,
            "plan_verification": True,
            "max_plan_steps": 20,
            "checkpoint_frequency": 5,
        }

    def _exploratory_strategy(self) -> dict:
        """探索策略：先全面理解再行动"""
        return {
            "initial_exploration_steps": 10,
            "codebase_indexing": True,
            "require_understanding_summary": True,
            "checkpoint_frequency": 5,
        }

    def get_strategy_report(self) -> str:
        """生成策略表现报告"""
        lines = ["## 策略表现报告\n"]
        for strategy, perf in self.strategy_performance.items():
            total = perf.success_count + perf.failure_count
            if total == 0:
                continue
            lines.append(
                f"- **{strategy.value}**: "
                f"成功率 {perf.success_rate:.1%} ({perf.success_count}/{total}), "
                f"平均 {perf.avg_steps_to_success:.1f} 步完成, "
                f"总消耗 {perf.total_tokens_used:,} tokens"
            )
        return "\n".join(lines)

八、实战：构建一个完整的 Agent Harness

8.1 整体架构

把以上所有模块组合起来，我们得到一个完整的 Agent Harness：

import asyncio
from typing import Optional

class AgentHarness:
    """
    完整的 Agent Harness 实现

    整合了：
    - 上下文管理（ContextWindowManager）
    - 工具执行（ToolExecutor）
    - 记忆系统（AgentMemory）
    - 状态管理（StateManager）
    - 自适应策略（AdaptiveHarness）
    - 错误恢复（GoalTracker）
    """

    def __init__(
        self,
        model_client,  # 模型客户端（支持 OpenAI / Anthropic 等格式）
        tools: list[ToolDefinition],
        config: dict = None
    ):
        self.model = model_client
        self.config = config or {}

        # 核心组件初始化
        max_tokens = self.config.get("max_context_tokens", 100000)
        self.context = ContextWindowManager(max_tokens=max_tokens)
        self.tools = ToolExecutor(tools)
        self.memory = AgentMemory()
        self.state_manager = StateManager()
        self.adaptive = AdaptiveHarness()
        self.goal_tracker: Optional[GoalTracker] = None

        # 配置
        self.max_iterations = self.config.get("max_iterations", 100)
        self.auto_checkpoint = self.config.get("auto_checkpoint", True)
        self.checkpoint_interval = self.config.get("checkpoint_interval", 10)

    async def run(self, task: str) -> dict:
        """运行 Agent 完成指定任务"""
        print(f"[Harness] Starting task: {task[:80]}...")

        # 初始化目标追踪
        self.goal_tracker = GoalTracker(task)
        self.context.add(task, ContextPriority.CRITICAL, source="task", tags={"task"})

        # 策略选择
        strategy = self.adaptive.select_strategy(task, task_complexity=0.6)
        strategy_config = self.adaptive.strategies[strategy]()
        print(f"[Harness] Using strategy: {strategy.value}")

        for iteration in range(self.max_iterations):
            # 检查点保存
            if self.auto_checkpoint and iteration % self.checkpoint_interval == 0:
                state_id = self.state_manager.save_checkpoint(
                    task=task,
                    context_mgr=self.context,
                    memory=self.memory,
                    executor=self.tools,
                    step=iteration
                )
                print(f"[Harness] Checkpoint saved: {state_id}")

            # 检查目标迷失
            if self.goal_tracker.should_reassess():
                reassess = self.goal_tracker.generate_reassess_prompt()
                self.context.add(reassign, ContextPriority.CRITICAL, source="system")
                print("[Harness] Goal reassessment triggered")

            # 构建 prompt
            prompt = self._build_prompt(strategy_config)

            # 调用模型
            response = await self.model.generate(prompt)

            # 解析响应
            if response.type == "tool_call":
                # 执行工具
                call = await self.tools.execute(
                    response.tool_name,
                    response.arguments
                )
                result_str = self.tools.format_tool_result(call)
                self.context.add(result_str, ContextPriority.HIGH, source="tool")

                self.goal_tracker.record_step(
                    response.tool_name, result_str, call.status == ToolCallStatus.SUCCESS
                )

                # 自动记忆重要发现
                if call.status == ToolCallStatus.SUCCESS:
                    self.memory.remember(
                        f"工具 {call.tool_name} 在此任务中成功执行，结果：{result_str[:100]}",
                        tags=["tool_success", call.tool_name]
                    )

            elif response.type == "done":
                # 任务完成
                print(f"[Harness] Task completed in {iteration + 1} iterations")
                self.adaptive.record_outcome(strategy, True, iteration, self.context._current_tokens)
                return {"success": True, "result": response.content}

            elif response.type == "text":
                # 普通文本响应，添加到上下文
                self.context.add(response.content, ContextPriority.MEDIUM, source="model")

            else:
                self.context.add(str(response), ContextPriority.LOW, source="unknown")

        # 达到最大迭代次数
        print(f"[Harness] Max iterations ({self.max_iterations}) reached")
        self.adaptive.record_outcome(strategy, False, self.max_iterations, self.context._current_tokens)
        return {"success": False, "reason": "max_iterations_reached"}

    def _build_prompt(self, strategy_config: dict) -> str:
        """构建发送给模型的 prompt"""
        parts = []

        # 系统提示
        parts.append(self._build_system_prompt())

        # 任务相关记忆
        recent_memories = self.memory.recall(
            self.goal_tracker.original_goal, max_results=3
        )
        if recent_memories:
            parts.append(self.memory.format_memory_for_context(recent_memories))

        # 当前上下文
        parts.append("\n## 当前状态\n")
        parts.append(self.context.get_context_for_model())

        # 策略特定指令
        if strategy_config.get("require_plan_before_action"):
            parts.append("\n请先制定详细的执行计划，再开始操作。")

        if strategy_config.get("initial_exploration_steps"):
            parts.append("\n请先探索代码库，理解整体结构后再开始修复。")

        return "\n\n".join(parts)

    def _build_system_prompt(self) -> str:
        """构建系统提示词"""
        tool_schemas = [t.to_schema() for t in self.tools.tools.values()]
        return f"""你是一个专业的 AI Agent，在 {self.config.get('task_domain', 'software engineering')} 领域执行任务。

## 核心原则
1. **先理解，再行动**：在执行任何操作前，确保理解了问题的本质
2. **小步快跑**：每次操作后验证结果，不要一次性做太大改动
3. **透明思考**：在调用工具前，先解释你的思路
4. **持续学习**：记录重要的发现和教训，供后续使用

## 工具使用规范
你可以通过调用工具与外部世界交互。每个工具都有明确的功能描述和参数定义。

{json.dumps(tool_schemas, ensure_ascii=False, indent=2)}

## 输出格式
调用工具：<tool_call>...<tool_call>
完成任务：<done>...<done>
"""

8.2 使用示例

async def main():
    # 初始化模型客户端（以 Anthropic 为例）
    model = AnthropicClient(api_key="sk-...")

    # 注册工具
    tools = [
        search_code_tool,
        execute_command_tool,
        read_file_tool,
        write_file_tool,
    ]

    # 初始化 Harness
    harness = AgentHarness(
        model_client=model,
        tools=tools,
        config={
            "max_context_tokens": 100000,
            "max_iterations": 100,
            "auto_checkpoint": True,
            "checkpoint_interval": 10,
            "task_domain": "code repair"
        }
    )

    # 运行任务
    result = await harness.run(
        "修复 GitHub issue #1234：用户上传大文件时内存溢出"
    )

    print(result)

if __name__ == "__main__":
    asyncio.run(main())

九、性能优化与最佳实践

9.1 上下文压缩的艺术

上下文压缩不是简单地截断。好的压缩策略需要：

结构化摘要：不保留原始日志，而是生成结构化的摘要：

def summarize_log_batch(logs: list[str]) -> str:
    """将一批日志压缩为结构化摘要"""
    error_lines = [l for l in logs if "error" in l.lower() or "fail" in l.lower()]
    success_operations = [l for l in logs if "success" in l.lower() or "completed" in l.lower()]

    summary_parts = []

    if error_lines:
        summary_parts.append(f"发现 {len(error_lines)} 条错误记录")
        # 保留前 3 条错误
        for err in error_lines[:3]:
            summary_parts.append(f"  - {err[:150]}")

    if success_operations:
        summary_parts.append(f"成功完成 {len(success_operations)} 项操作")

    summary_parts.append(f"日志批次总计 {len(logs)} 行")

    return "\n".join(summary_parts)

增量压缩：不是每次超限才压缩，而是在达到 70% 容量时就开始主动压缩，避免突然的大规模丢弃：

class ProactiveContextManager(ContextWindowManager):
    """带主动压缩功能的上下文管理器"""

    def add(self, content: str, priority: ContextPriority, source: str = "", tags: set = None):
        super().add(content, priority, source, tags)

        # 超过 70% 容量时主动压缩
        if self._current_tokens > self.available_tokens * 0.7:
            # 使用 summarizer 压缩低优先级内容
            self._smart_compact()

    def _smart_compact(self):
        """智能压缩：生成摘要而非直接丢弃"""
        low_priority = [s for s in self.segments
                       if s.priority in (ContextPriority.LOW, ContextPriority.MEDIUM)
                       and s.source in ("log", "tool_output")]

        if not low_priority:
            return

        # 按来源分组
        by_source = {}
        for seg in low_priority:
            by_source.setdefault(seg.source, []).append(seg.content)

        # 对每组生成摘要
        new_segments = [s for s in self.segments if s not in low_priority]
        for source, contents in by_source.items():
            summary = summarize_log_batch(contents)
            new_segments.append(ContextSegment(
                content=f"[压缩摘要 - {source}]\n{summary}",
                priority=ContextPriority.HIGH,  # 摘要保留较高优先级
                token_count=self._estimate_tokens(summary),
                source=source,
                tags={"compressed", source}
            ))

        self.segments = new_segments

9.2 工具并行化

对于相互独立的工具调用，可以并行执行以加速：

async def execute_parallel(
    self,
    tool_calls: list[tuple[str, dict]],
    max_concurrent: int = 3
) -> list[ToolCall]:
    """并行执行多个工具调用"""
    semaphore = asyncio.Semaphore(max_concurrent)

    async def bounded_execute(tool_name: str, args: dict):
        async with semaphore:
            return await self.execute(tool_name, args)

    tasks = [
        bounded_execute(name, args)
        for name, args in tool_calls
    ]

    return await asyncio.gather(*tasks)

9.3 工具调用超时设计

合理的超时设计能防止 Agent 被单个慢操作卡死：

class TieredTimeoutExecutor(ToolExecutor):
    """分层超时执行器——不同类型的工具有不同超时配置"""

    DEFAULT_TIMEOUTS = {
        "search_code": 30,
        "execute_command": 60,
        "read_file": 10,
        "write_file": 30,
        "api_call": 20,
    }

    async def execute(self, tool_name: str, arguments: dict, **kwargs) -> ToolCall:
        timeout = arguments.get(
            "timeout",
            self.DEFAULT_TIMEOUTS.get(tool_name, 60)
        )
        return await super().execute(tool_name, arguments, timeout=timeout, **kwargs)

十、未来展望：Harness 的演进方向

10.1 自我优化的 Harness

2026年最激动人心的研究方向是"让 Harness 自己学会优化自己"：

• 基于强化学习的策略调整：Harness 的各个参数（压缩时机、重试次数、策略选择）作为可学习的参数，通过历史任务表现持续优化
• 工具自动发现：Harness 能根据任务需求，自动组合和生成新的工具，而不是依赖预先定义的工具集
• 跨任务知识迁移：在一个项目中学习到的模式，能否迁移到另一个项目

10.2 多 Agent 协作的 Harness

当一个 Agent 无法独立完成任务时，Harness 需要支持多 Agent 协作：

• 任务分解：将复杂任务分配给不同的专业 Agent
• 结果聚合：整合多个 Agent 的输出，形成完整解决方案
• 冲突解决：当不同 Agent 的建议冲突时，Harness 需要有能力判断哪个更可靠

10.3 安全与可控性

随着 Agent 能力的增强，Harness 的安全边界变得越来越重要：

• 权限分级：不同的工具调用需要不同级别的确认
• 操作审计：所有操作都需要完整的审计日志
• 回滚机制：任何破坏性操作都应该可以快速回滚

结语：Harness 是 Agent 工程化的主战场

回到最初的问题：为什么同样的模型，换一个 Harness，效果能差出十几个百分点？

因为模型提供的是"智能"，Harness 提供的是"执行力"。没有 Harness，模型只是一个强大的推理引擎，无法稳定地完成真实任务。有了好的 Harness，模型的能力才能被充分释放。

2026年的 AI Agent 领域，正在经历一个范式转变：人们开始意识到，模型能力的边际收益在递减，而 Harness 工程的边际收益在递增。当所有人都能访问同样的模型时，Harness 的质量就成了决定性的竞争优势。

这意味着，对于今天的工程师来说，学习如何构建高质量的 Agent Harness，是和学会使用大模型本身一样重要的技能。本文所讨论的上下文管理、工具调用、错误恢复、记忆系统、状态持久化和自适应策略，构成了 Harness 工程的核心知识体系。

掌握这些，你就不只是"在使用 AI"，而是真正在"构建 AI-native 的系统"。这才是 2026 年工程师最核心的竞争力。

相关标签：AI Agent|Harness|运行时|上下文管理|工具调用|错误恢复|记忆系统|状态管理|自适应策略|Agent架构