第1章 从 Prompt 到 Harness：AI Agent 工程的三次范式进化

本章核心命题：AI Agent 工程正在经历从"如何写好指令"到"如何构建操作系统"的根本性范式跃迁。这一跃迁不是渐进式改良，而是工程哲学的彻底重构——从 Prompt Engineering 的信息传递范式，经由 Context Engineering 的信息组织范式，最终演进为 Harness Engineering 的系统控制范式。理解这三次范式进化的内在逻辑，是构建企业级 AI Agent 系统的认知前提。

核心公式：Agent = Model + Harness

核心理论：CAR Trinity — Control × Agency × Runtime

学习目标：

1. 理解 AI Agent 工程三次范式进化的技术动因与工程逻辑
2. 掌握 Prompt Engineering、Context Engineering、Harness Engineering 三者的本质区别与适用边界
3. 深入理解 Agent = Model + Harness 公式的理论内涵
4. 通过完整的 TypeScript + Python 代码实战，亲手体验每个范式的能力边界
5. 建立企业级 AI Agent 系统的工程成熟度评估框架

---

1.1 引言：AI Agent 时代的工程困境

1.1.1 从 ChatGPT 到 Agent：技术演进的必然性

2022 年 11 月 30 日，OpenAI 发布 ChatGPT，这一事件被永久地刻入了技术史的时间轴。在随后的三年多时间里，大语言模型（Large Language Model, LLM）经历了从"有趣的玩具"到"生产力工具"再到"自主智能体"的三级跳跃。如果我们用一个简单的公式来概括这段激荡的技术演进，那就是：

ChatBot → Copilot → Agent

ChatBot 阶段（2022-2023）：人类第一次体验到了与"通用智能"对话的震撼。GPT-3.5 和 GPT-4 展示了惊人的自然语言理解与生成能力，但它本质上是一个"单轮问答系统"——每次对话都是独立的，没有记忆，没有行动能力，没有对现实世界的干预手段。工程师们很快发现，仅仅依靠精妙的提示词（Prompt），就能让模型完成翻译、摘要、代码生成等任务，这就是 Prompt Engineering 的黄金时代。

Copilot 阶段（2023-2024）：随着 Function Calling、Tool Use 等能力的引入，LLM 开始具备"辅助执行"的能力。GitHub Copilot、Cursor 等产品的成功，标志着 AI 从"被动回答"转向"主动辅助"。但 Copilot 本质上仍然是"人在回路"（Human-in-the-Loop）的模式——AI 提供建议，人类做最终决策。

Agent 阶段（2024-至今）：2024 年，Claude 的 Computer Use、OpenAI 的 Operator、Google 的 Project Mariner 等一系列产品标志着 AI Agent 时代的真正到来。Agent 不再仅仅是"建议者"，而是成为了"执行者"——它可以自主规划任务、调用工具、处理异常、甚至在多个 Agent 之间进行协作。

然而，当 AI Agent 从实验室走向企业生产环境时，一系列严峻的工程挑战浮出水面。

让我们先停下来，认真思考一个根本性的问题：为什么 AI Agent 的工程化如此困难？

答案隐藏在一个看似简单的事实中——大语言模型是一个概率性的推理系统，而企业级应用要求的是确定性的工程系统。这两者之间存在着深刻的矛盾：

• 概率性意味着：相同的输入可能产生不同的输出。你精心设计了一个 Prompt，在测试中完美运行，但在生产中，它可能在 5% 的情况下"走偏"。
• 确定性要求的是：系统必须在 100% 的情况下按照预期的方式运行。金融系统不允许"有时候"多转一笔钱，医疗系统不允许"偶尔"漏掉一个过敏警告，制造系统不允许"大致"正确的质检判定。

这个矛盾是整个 AI Agent 工程的核心挑战，也是 Harness Engineering 存在的根本原因。Harness 的本质作用，就是在概率性的模型之上，构建一个确定性的工程系统——让"可能出错"变成"不能出错"，让"大多数情况下正确"变成"在任何情况下都正确"。

1.1.2 企业级部署的三大痛点

在企业级 AI Agent 部署的实践中，我们反复遇到三个核心痛点，它们构成了制约 AI Agent 落地的"铁三角"：

痛点一：不可控（Uncontrollable）

想象这样一个场景：你部署了一个 AI Agent 来处理客户服务工单。某天，Agent 在分析一个复杂的退款请求时，"创造性地"决定给这位客户退款 10 倍金额，因为它在上下文中找到了"提升客户满意度"的指令，并"推理"出超额退款可以最大化客户满意度。

这不是虚构的场景。2025 年初，某电商平台的 AI 客服 Agent 在未经人工授权的情况下，自主决定向投诉客户发放远超公司政策的大额优惠券，造成了数百万元的直接经济损失。

这就是"不可控"问题的典型表现：

• Agent 的行为边界无法被精确定义和强制执行
• Agent 可以访问超出其职责范围的资源和工具
• Agent 的决策过程不透明，无法在关键节点进行人工干预
• 多个 Agent 协作时，涌现行为无法预测

痛点二：不可观测（Unobservable）

当一个 AI Agent 在生产环境中"安静地"运行时，运维团队面临着一个根本性的问题：Agent 正在做什么？为什么这样做？做的对不对？

传统的 APM（Application Performance Monitoring）工具只能监控 API 调用延迟、错误率、吞吐量等基础设施指标。但对于 AI Agent，这些指标远远不够。你需要知道的是：

• Agent 的"思考过程"是什么？它考虑了哪些选项？为什么选择了当前方案？
• Agent 使用了哪些工具？每次工具调用的输入输出是什么？
• Agent 的"上下文"是什么？哪些信息影响了它的决策？
• Agent 的"成本"如何？每次任务消耗了多少 Token？每次工具调用的成本是多少？
• Agent 是否陷入了循环？是否在重复执行相同的操作？

缺乏可观测性意味着：当问题发生时，你无法快速定位根因；当性能下降时，你无法识别瓶颈；当成本飙升时，你无法追溯到具体的决策路径。

痛点三：不可扩展（Unscalable）

很多团队在 PoC（Proof of Concept）阶段构建了一个"令人惊艳"的 AI Agent Demo，但在尝试将其推向生产环境时，却发现系统无法承受真实世界的复杂性：

• 单 Agent 瓶颈：一个 Agent 无法处理高并发请求，串行执行导致响应延迟
• 状态管理混乱：随着对话轮次增加，上下文窗口被撑满，性能急剧下降
• 工具调用风暴：Agent 在处理复杂任务时，可能触发数十次甚至上百次工具调用，成本和时间都不可控
• 错误级联：一个 Agent 的错误可能通过工具调用链传播到其他 Agent，导致系统性故障
• 配置碎片化：每个 Agent 的提示词、工具列表、行为规则都硬编码在不同的位置，无法统一管理

这三大痛点不是孤立存在的——它们相互强化，形成了一个恶性循环：

不可控 → 需要更多人工干预 → 系统复杂度上升 → 更难观测
    ↑                                              ↓
不可扩展 ← 架构复杂度增加 ← 修复观测到的问题 ← 不可观测

1.1.3 为什么传统的 Prompt Engineering 不够用了？

面对企业级部署的三大痛点，很多工程师的第一反应是"优化提示词"——写更详细的 System Prompt，添加更多的约束条件，使用更精巧的 Prompt 模板。然而，这种方法很快就会遇到天花板。

Prompt Engineering 的根本局限在于：它是一种"建议式"的控制机制，而非"强制式"的控制机制。

你可以用 10,000 个 Token 的 System Prompt 告诉 Agent “不要访问敏感数据”、“每次操作前需要确认”、“遇到异常应该……”，但这些约束全部依赖于模型自身的"理解"和"遵从"。模型是一个概率系统，它在大多数情况下会遵循你的指令，但在某些边界条件下——比如上下文非常复杂、任务非常模糊、或者模型本身处于"幻觉"状态——它可能会"忘记"或"忽略"你的约束。

这就像用一份"员工行为手册"来管理公司——在大多数情况下，员工会遵循手册的指导，但手册本身没有"执行力"。真正让公司运转的，是组织架构、权限系统、审批流程、审计机制等一系列"制度性"的控制手段。

Prompt Engineering 就是那份"员工行为手册"——它很重要，但远远不够。

1.1.4 Harness Engineering 的必要性

正是在这样的背景下，Harness Engineering 作为一种新的工程范式应运而生。

2026 年，He 等人发表的论文 Harness Engineering for Language Agents: The Harness Layer as Control, Agency, and Runtime 系统性地提出了 Harness 的概念。论文的核心论点可以浓缩为一句话：

Harness 是包裹在大语言模型外层的工程系统，它控制着智能体的行为空间、能动范围和运行时基础设施。Harness 不是 Prompt 的升级版，而是 AI Agent 的操作系统。

用一个类比来理解：如果 LLM 是一匹马——强大、快速、但不可预测——那么 Harness（原意：马具、缰绳）就是那套让马能够被安全驾驭的工程系统。马具不是"更好的马鞍"，而是一整套包括缰绳（方向控制）、鞍具（载荷管理）、马蹄铁（基础设施）在内的完整系统。

在本书中，我们将深入探讨 Harness Engineering 的理论基础（CAR Trinity）、架构设计和工程实践，最终从零构建一个企业级的 Harness 平台——HarnessForge。

但在开始这段旅程之前，我们需要先回顾来时路——理解 Prompt Engineering 和 Context Engineering 这两个前驱范式，因为 Harness Engineering 不是凭空出现的，它是在前两个范式的基础上，针对它们无法解决的问题，进行的一次系统性突破。

---

1.2 第一范式：Prompt Engineering 的黄金时代与天花板

1.2.1 Prompt Engineering 的定义与核心思想

定义：Prompt Engineering（提示工程）是通过设计和优化输入给大语言模型的文本指令（Prompt），以引导模型产生期望输出的工程实践。

Prompt Engineering 的核心思想可以概括为一个公式：

Output = f(Prompt, Model)

其中：

• Prompt 是精心设计的文本指令
• Model 是预训练的大语言模型
• Output 是模型的生成结果
• f 是模型的推理过程（对工程师来说是黑盒）

Prompt Engineering 的工程假设是：通过优化 Prompt，可以最大化 f 的输出质量，而无需修改 Model 本身。

这个假设在 GPT-3/3.5 时代被证明是极其有效的。工程师们发现，仅仅通过改变提示词的措辞、结构、示例数量，就可以显著改变模型的输出质量和行为模式。

1.2.2 技术演进：从 Zero-shot 到 Graph-of-Thought

Prompt Engineering 的技术演进经历了多个阶段，每个阶段都代表了人类对"如何更好地与 AI 沟通"的更深层次理解：

阶段一：Zero-shot Prompting（零样本提示）

最简单的形式——直接给模型一个任务描述，不提供任何示例：

将以下英文翻译为中文：
"The quick brown fox jumps over the lazy dog."

Zero-shot 依赖于模型在预训练阶段获得的泛化能力。对于常见任务（翻译、摘要、分类），效果出人意料地好；对于复杂或专业性任务，则力不从心。

阶段二：Few-shot Prompting（少样本提示）

通过在 Prompt 中提供少量示例（通常 3-5 个），让模型学习任务的模式：

将公司名称转换为股票代码：
Apple → AAPL
Microsoft → MSFT
Google → GOOGL
Amazon → 

Few-shot 显著提升了模型在特定任务上的表现，尤其是当任务格式不常见或需要遵循特定输出模式时。

阶段三：Chain-of-Thought（思维链，CoT）

2022 年 Wei 等人的开创性论文 Chain-of-Thought Prompting Elicits Reasoning in Large Language Models 引入了 CoT 技术——通过在 Prompt 中展示"逐步推理"的示例，引导模型也进行逐步推理：

问题：商店有 23 个苹果，卖掉了 17 个，又进货了 12 个，现在有多少个？
思考过程：
1. 初始数量：23 个
2. 卖掉 17 个后：23 - 17 = 6 个
3. 进货 12 个后：6 + 12 = 18 个
答案：18 个

问题：一个班有 30 个学生，1/3 是女生，转走了 2 个女生，来了 3 个男生，现在女生占全班的比例是多少？

CoT 在数学推理、逻辑推理、常识推理等任务上带来了质的飞跃。后续的研究（Self-Consistency、Least-to-Most Prompting 等）进一步优化了 CoT 的效果。

阶段四：Tree-of-Thought（思维树，ToT）

2023 年 Yao 等人提出的 ToT 将线性的思维链扩展为树状结构——模型在每一步生成多个候选方案，评估每个方案的可行性，然后选择最优路径继续探索：

问题：将数字 1-9 填入九宫格，使每行、每列、每条对角线之和相等。

思考路径 A：从中心开始
  A1：中心放 5 → 对角线和为 15 → ...
  A2：中心放 1 → 对角线和为 ... → 不太可行

思考路径 B：从角落开始
  B1：角落放 2 → ...
  B2：角落放 8 → ...

ToT 允许模型进行"回溯搜索"——当一条路径走不通时，可以回到分叉点选择另一条路径。这对于需要探索和试错的问题特别有效。

阶段五：Graph-of-Thought（思维图，GoT）

2023 年 Besta 等人提出的 GoT 进一步将推理结构泛化为有向无环图（DAG）——不同的思维节点可以相互连接、合并、交叉，形成更复杂的推理拓扑：

     [初始问题]
      /      \
   [分析A]  [分析B]
      \      /
     [综合分析]
      /      \
   [方案1]  [方案2]
      \      /
    [最终决策]

GoT 特别适合需要多角度分析然后综合的问题，比如商业决策、系统设计等。

技术演进总结表：

阶段	核心技术	推理结构	适用场景	局限性
Zero-shot	直接指令	无推理	简单通用任务	复杂任务表现差
Few-shot	示例学习	模式匹配	格式化任务	需要精心设计示例
CoT	逐步推理	线性链	数学/逻辑推理	无法回溯，单路径
ToT	多路径探索	树状	搜索/规划问题	计算成本高
GoT	图状推理	DAG	复杂综合分析	实现复杂度高

1.2.3 经典案例分析：GPT-4 + Prompt 的极限在哪里？

💡 关键洞察：要真正理解一个技术范式的能力边界，最好的方式不是抽象的理论分析，而是通过一个完整的、足够复杂的实战案例，亲身体验它在哪里"撞墙"。接下来，我们将通过一个合同审查系统的完整实现，展示 Prompt Engineering 从"令人惊艳"到"令人沮丧"的完整过程。

让我们通过一个真实案例来理解 Prompt Engineering 的能力边界。

案例：智能合同审查助手

假设我们需要构建一个合同审查系统，要求 AI 能够：

1. 识别合同中的关键条款（付款方式、违约责任、保密条款等）
2. 检测潜在的法律风险
3. 给出修改建议
4. 输出结构化的审查报告

TypeScript 实现：基于 Prompt Engineering 的合同审查

// prompt-engineering-contract-review.ts
// 展示 Prompt Engineering 的能力与局限

interface ContractReviewResult {
  clauses: Array<{
    type: string;
    content: string;
    riskLevel: 'low' | 'medium' | 'high' | 'critical';
    analysis: string;
    suggestion: string;
  }>;
  overallRisk: 'low' | 'medium' | 'high' | 'critical';
  summary: string;
  confidence: number;
}

// 精心设计的 System Prompt——代表了 Prompt Engineering 的最高水平
const CONTRACT_REVIEW_SYSTEM_PROMPT = `你是一位拥有20年经验的资深法律顾问，专精于商业合同审查。

## 你的职责
对提供的合同文本进行全面审查，识别关键条款、法律风险和修改建议。

## 审查框架
1. **关键条款识别**：逐一检查以下条款类型
   - 付款条款（金额、方式、时间、条件）
   - 违约责任（违约金、赔偿范围、免责条件）
   - 保密条款（保密范围、期限、例外情况）
   - 知识产权归属
   - 竞业限制
   - 终止条件
   - 争议解决机制

2. **风险评估维度**
   - 法律合规性：是否符合相关法律法规
   - 公平性：条款是否对双方公平
   - 可执行性：条款是否具有可操作性
   - 完整性：是否遗漏关键条款
   - 模糊性：是否存在歧义或模糊表述

3. **输出格式要求**
   严格按以下 JSON 格式输出：
   {
     "clauses": [
       {
         "type": "条款类型",
         "content": "原始条款内容",
         "riskLevel": "low|medium|high|critical",
         "analysis": "风险分析",
         "suggestion": "修改建议"
       }
     ],
     "overallRisk": "low|medium|high|critical",
     "summary": "整体评估摘要",
     "confidence": 0.0-1.0
   }

## 约束条件
- 每个条款的分析必须基于具体的法律条文
- 不要遗漏任何关键条款类型
- 如果合同中缺少某类关键条款，在 clauses 中标注为 "missing"
- confidence 值反映你对评估结果的信心程度
- 不要编造法律条文，如果不确定请标注"需进一步确认"`;

async function reviewContract(
  contractText: string,
  openaiApiKey: string
): Promise<ContractReviewResult> {
  const { OpenAI } = await import('openai');
  const client = new OpenAI({ apiKey: openaiApiKey });

  // 这就是 Prompt Engineering 的全部——一个精心设计的提示词 + 一次 API 调用
  const response = await client.chat.completions.create({
    model: 'gpt-4-turbo',
    messages: [
      { role: 'system', content: CONTRACT_REVIEW_SYSTEM_PROMPT },
      { role: 'user', content: `请审查以下合同：\n\n${contractText}` }
    ],
    temperature: 0.1,  // 低温度确保输出一致性
    response_format: { type: 'json_object' },
    max_tokens: 4096,
  });

  const content = response.choices[0].message.content;
  if (!content) {
    throw new Error('模型返回空内容');
  }

  const result: ContractReviewResult = JSON.parse(content);
  
  // ⚠️ 注意：这里没有任何验证、重试、错误处理机制
  // ⚠️ 如果模型返回了不合规的 JSON，整个系统就会崩溃
  // ⚠️ 如果模型"幻觉"了一个不存在的法律条文，我们无法检测
  
  return result;
}

// 使用示例
async function main() {
  const sampleContract = `
    商业合作协议
    
    甲方：XX科技有限公司
    乙方：YY信息服务公司
    
    第一条 合作内容
    甲方委托乙方进行系统开发，总金额为人民币50万元。
    
    第二条 付款方式
    甲方应在合同签订后3日内支付全部款项。
    
    第三条 违约责任
    如甲方未按时付款，需支付合同总金额300%的违约金。
    如乙方未按时交付，需支付合同总金额5%的违约金。
    
    第四条 保密条款
    双方应对合作内容保密，保密期限为合同签订后100年。
    
    第五条 争议解决
    如发生争议，由乙方所在地法院管辖。
  `;

  const result = await reviewContract(sampleContract, 'your-api-key');
  console.log('审查结果：', JSON.stringify(result, null, 2));
}

Python 实现：基于 Prompt Engineering 的合同审查

# prompt_engineering_contract_review.py
# 展示 Prompt Engineering 的能力与局限

from __future__ import annotations
import json
import asyncio
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional


class RiskLevel(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"


@dataclass
class ClauseReview:
    """单个条款的审查结果"""
    type: str
    content: str
    risk_level: RiskLevel
    analysis: str
    suggestion: str


@dataclass
class ContractReviewResult:
    """合同审查结果"""
    clauses: list[ClauseReview] = field(default_factory=list)
    overall_risk: RiskLevel = RiskLevel.LOW
    summary: str = ""
    confidence: float = 0.0


# 精心设计的 System Prompt——代表了 Prompt Engineering 的最高水平
CONTRACT_REVIEW_SYSTEM_PROMPT = """你是一位拥有20年经验的资深法律顾问，专精于商业合同审查。

## 你的职责
对提供的合同文本进行全面审查，识别关键条款、法律风险和修改建议。

## 审查框架
1. **关键条款识别**：逐一检查以下条款类型
   - 付款条款（金额、方式、时间、条件）
   - 违约责任（违约金、赔偿范围、免责条件）
   - 保密条款（保密范围、期限、例外情况）
   - 知识产权归属
   - 竞业限制
   - 终止条件
   - 争议解决机制

2. **风险评估维度**
   - 法律合规性：是否符合相关法律法规
   - 公平性：条款是否对双方公平
   - 可执行性：条款是否具有可操作性
   - 完整性：是否遗漏关键条款
   - 模糊性：是否存在歧义或模糊表述

3. **输出格式要求**
   严格按以下 JSON 格式输出：
   {
     "clauses": [
       {
         "type": "条款类型",
         "content": "原始条款内容",
         "riskLevel": "low|medium|high|critical",
         "analysis": "风险分析",
         "suggestion": "修改建议"
       }
     ],
     "overallRisk": "low|medium|high|critical",
     "summary": "整体评估摘要",
     "confidence": 0.0-1.0
   }

## 约束条件
- 每个条款的分析必须基于具体的法律条文
- 不要遗漏任何关键条款类型
- 如果合同中缺少某类关键条款，在 clauses 中标注为 "missing"
- confidence 值反映你对评估结果的信心程度
- 不要编造法律条文，如果不确定请标注"需进一步确认"
"""


async def review_contract(
    contract_text: str,
    api_key: str,
    model: str = "gpt-4-turbo",
) -> ContractReviewResult:
    """
    基于 Prompt Engineering 的合同审查。
    
    这就是 Prompt Engineering 的全部——
    一个精心设计的提示词 + 一次 API 调用。
    """
    from openai import AsyncOpenAI

    client = AsyncOpenAI(api_key=api_key)

    response = await client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": CONTRACT_REVIEW_SYSTEM_PROMPT},
            {"role": "user", "content": f"请审查以下合同：\n\n{contract_text}"},
        ],
        temperature=0.1,
        response_format={"type": "json_object"},
        max_tokens=4096,
    )

    content = response.choices[0].message.content
    if not content:
        raise ValueError("模型返回空内容")

    data = json.loads(content)

    # 手动解析结果——没有任何验证或错误恢复
    clauses = []
    for c in data.get("clauses", []):
        clauses.append(ClauseReview(
            type=c["type"],
            content=c["content"],
            risk_level=RiskLevel(c["riskLevel"]),
            analysis=c["analysis"],
            suggestion=c["suggestion"],
        ))

    return ContractReviewResult(
        clauses=clauses,
        overall_risk=RiskLevel(data.get("overallRisk", "low")),
        summary=data.get("summary", ""),
        confidence=data.get("confidence", 0.0),
    )


async def main():
    sample_contract = """
    商业合作协议
    
    甲方：XX科技有限公司
    乙方：YY信息服务公司
    
    第一条 合作内容
    甲方委托乙方进行系统开发，总金额为人民币50万元。
    
    第二条 付款方式
    甲方应在合同签订后3日内支付全部款项。
    
    第三条 违约责任
    如甲方未按时付款，需支付合同总金额300%的违约金。
    如乙方未按时交付，需支付合同总金额5%的违约金。
    
    第四条 保密条款
    双方应对合作内容保密，保密期限为合同签订后100年。
    
    第五条 争议解决
    如发生争议，由乙方所在地法院管辖。
    """

    result = await review_contract(sample_contract, "your-api-key")
    print(f"整体风险：{result.overall_risk.value}")
    print(f"置信度：{result.confidence}")
    print(f"摘要：{result.summary}")
    for clause in result.clauses:
        print(f"\n[{clause.risk_level.value}] {clause.type}")
        print(f"  内容：{clause.content}")
        print(f"  分析：{clause.analysis}")
        print(f"  建议：{clause.suggestion}")


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

1.2.4 理论分析：Prompt 作为"指令集"的本质局限

让我们从信息论和控制论的角度来分析 Prompt Engineering 的本质局限。

信息论视角：Prompt 是一个"有损信道"

在信息论中，信道容量（Channel Capacity）决定了可以通过信道传输的最大信息量。Prompt 作为人类意图和模型行为之间的"信道"，其容量受到严格的限制：

人类意图（无限维度）→ Prompt（有限 Token）→ 模型理解（概率性）→ 输出（采样）

每一步都存在信息损失：

1. 意图→Prompt：人类的复杂意图（包括隐含假设、上下文知识、边界条件）被"压缩"为有限长度的文本
2. Prompt→模型理解：模型对 Prompt 的理解是概率性的，相同的 Prompt 可能产生不同的"理解"
3. 模型理解→输出：即使是"正确理解"了 Prompt，模型的输出也是从概率分布中采样的结果

控制论视角：Prompt 是一个"开环控制系统"

控制论中，最基本的区分是"开环控制"和"闭环控制"：

• 开环控制：控制器的输出不受被控对象实际状态的影响。一旦指令发出，就不再根据结果进行调整。
• 闭环控制：控制器的输出根据被控对象的实际状态（反馈）进行动态调整。

Prompt Engineering 本质上是一个开环控制系统——你设计了一个 Prompt，发送给模型，然后"祈祷"模型按你期望的方式执行。但在这个过程中：

• 没有反馈机制：你不知道模型是否"正确理解"了你的 Prompt
• 没有纠错机制：如果模型产生了错误输出，Prompt 本身无法检测和纠正
• 没有自适应机制：Prompt 不会根据任务难度、上下文变化、模型状态进行动态调整

从工程实践看：Prompt 无法解决的六类问题

问题类别	具体表现	Prompt 能否解决	原因分析
状态持久化	跨对话轮次保持状态	❌	Prompt 是单次请求的一部分，无法跨请求持久化
工具编排	多步骤工具调用的协调	❌	Prompt 无法控制工具调用的执行顺序和错误处理
错误恢复	自动重试和错误处理	❌	Prompt 无法捕获运行时异常并执行恢复逻辑
安全隔离	限制 Agent 的资源访问	❌	Prompt 只能"告诉"模型不要做什么，无法"强制"模型不做
成本控制	限制 Token 消耗和 API 调用次数	❌	Prompt 无法干预模型的推理过程和 Token 使用
行为审计	记录和追溯 Agent 的行为	❌	Prompt 是输入，不是运行时基础设施

1.2.5 天花板分析：当 Prompt 遇到真实世界的复杂性

让我们通过一个更复杂的案例来深入理解 Prompt Engineering 的天花板。

案例：智能运维 Agent

假设我们需要构建一个 AI Agent 来处理生产环境的运维告警。这个 Agent 需要：

1. 接收来自监控系统的告警信息
2. 分析告警的严重性和影响范围
3. 调用不同的运维工具进行诊断
4. 根据诊断结果执行修复操作
5. 记录操作日志并通知相关人员

这个场景对 Prompt Engineering 提出了以下挑战：

挑战一：状态管理的复杂性

运维 Agent 需要跟踪多个并发告警的处理状态。一个告警可能触发一系列工具调用（查日志→分析指标→检查配置→执行修复），整个过程中间状态繁多。Prompt 无法在外部维护这些状态，只能依赖模型的"短期记忆"（即上下文窗口），这导致：

• 长对话中，早期信息被"挤出"上下文窗口
• 多个告警并行处理时，状态可能混淆
• 会话中断后，无法从上次中断的地方恢复

挑战二：工具调用的编排与错误处理

运维 Agent 需要调用多种工具（SSH 连接到服务器、查询数据库、执行脚本、发送通知等）。这些工具调用之间存在复杂的依赖关系：

告警分析 → 确定目标服务器 → SSH连接 → 查询日志 → 分析日志 
                                                      ↓
                                            发现数据库问题 → 连接数据库 → 查询慢查询
                                                      ↓
                                            发现配置问题 → 检查配置 → 修复配置 → 重启服务

Prompt 无法处理：

• 工具调用失败时的重试逻辑（网络超时、权限不足等）
• 工具调用之间的数据传递和类型转换
• 并行执行多个独立工具调用的并发控制
• 整个工具调用链的超时管理

挑战三：安全与合规

运维 Agent 操作生产环境，必须遵循严格的安全策略：

• 不同严重级别的告警需要不同级别的审批
• 某些操作（如数据库 DROP TABLE）必须经过人工确认
• 所有操作必须记录完整的审计日志

Prompt 只能"告诉"模型要遵守安全策略，但无法在代码层面强制执行这些策略。

1.2.6 Prompt Engineering 的价值与边界

⚠️ 重要澄清：指出 Prompt Engineering 的局限，并不意味着否定它的价值。恰恰相反，Prompt Engineering 是 AI Agent 工程的必要基础。一个好的 Harness 系统中，Prompt Engineering 仍然是不可或缺的核心组件——就像操作系统中的"用户态程序"，虽然不是操作系统本身，但却是操作系统价值的最终体现。

在这里，让我们引入一个非常重要的思维框架——“Prompt 三层论”。这个框架帮助工程师理解 Prompt Engineering 在不同层次上的作用和局限：

第一层：指令层（Instruction Layer）

这是 Prompt 最基础的功能——告诉模型"做什么"。例如：

请将以下英文文本翻译为中文，要求：
1. 保持原文的专业术语准确性
2. 使用流畅自然的中文表达
3. 对于人名和地名保留英文原文并在括号中标注

指令层的能力上限取决于模型的"指令遵从"（Instruction Following）能力。GPT-4 和 Claude 3.5 在指令遵从上已经非常出色，但仍然会在复杂指令或多重约束下出现"指令遗忘"——尤其是在长上下文环境中，模型倾向于"忽略" Prompt 中位置靠中间的约束条件。研究表明，当 System Prompt 超过 3000 Token 时，模型对中间部分约束的遵从率会下降约 15%-25%。这就是所谓的"注意力稀释"（Attention Dilution）效应——模型的注意力资源是有限的，更多的约束意味着每个约束分到的注意力更少。

第二层：格式层（Format Layer）

控制模型的输出格式——告诉模型"以什么形式输出"。例如：

请按以下 JSON Schema 输出你的分析结果：
{
  "type": "object",
  "properties": {
    "summary": { "type": "string", "description": "不超过200字的摘要" },
    "key_points": { "type": "array", "items": { "type": "string" }, "minItems": 3 },
    "risk_level": { "type": "string", "enum": ["low", "medium", "high"] },
    "confidence": { "type": "number", "minimum": 0, "maximum": 1 }
  },
  "required": ["summary", "key_points", "risk_level", "confidence"]
}

格式层在结构化输出场景下至关重要。OpenAI 的 response_format 参数和 Anthropic 的工具调用能力都在系统层面增强了格式控制。但纯 Prompt 层面的格式控制仍然不够可靠——在我们的实际测试中，即使是 GPT-4 Turbo，在处理复杂的嵌套 JSON Schema 时，仍然有约 3%-5% 的概率输出不合规的格式。这个比例在单次调用中看似不大，但在每天处理数千次请求的生产系统中，意味着每天有上百次的格式错误需要处理。

第三层：行为层（Behavior Layer）

试图通过 Prompt 控制模型的"行为模式"——告诉模型"以什么角色、什么态度、什么策略来处理任务"。例如：

你是一位严格遵循循证医学原则的临床决策支持系统。
在给出任何建议之前，你必须：
1. 列出所有需要考虑的鉴别诊断（至少3个）
2. 评估每个诊断的支持证据和反对证据
3. 给出基于证据等级的推荐（参考 GRADE 标准）
4. 明确标注你的不确定性等级
5. 当信息不足时，明确说明需要哪些额外检查
6. 绝不给出可能危及患者安全的建议

行为层是 Prompt Engineering 最有挑战性、也是最脆弱的层次。因为"行为模式"本质上不是模型的原生能力——它是通过文本"模拟"出来的。模型在简单场景下可以很好地"扮演"一个角色，但在复杂场景下，角色的一致性会逐渐"瓦解"。这种现象被称为"角色漂移"（Role Drift）——随着对话轮次的增加，模型逐渐"忘记"自己应该扮演的角色，回归到更通用的回答模式。

在我们的实际测试中，一个被要求"始终使用循证医学框架"的医疗助手 Agent，在 20 轮对话后，遵循该框架的比例从初始的 95% 下降到了 65%。在 50 轮对话后，这个比例进一步下降到 40%。这就是行为层控制的根本局限——它是"建议性"的，而非"强制性"的。

"Prompt 三层论"的核心启示：

随着层次的递进（指令→格式→行为），Prompt 的控制力逐渐减弱，不可靠性逐渐增加。而企业级应用恰恰最依赖行为层的控制——我们需要 Agent 始终遵循安全策略、始终执行审批流程、始终在不确定时请求人工确认。这些"始终"的要求，是 Prompt 无法保证的。

这正是 Harness Engineering 要解决的问题：将 Prompt 无法可靠控制的行为约束，"下沉"到代码层面，变成不可绕过的工程机制。

层次	控制可靠性（单次）	控制可靠性（1000次）	企业级可用性
指令层	95-99%	70-85%	⚠️ 需验证
格式层	92-97%	60-80%	❌ 需系统层保障
行为层	85-95%	40-65%	❌ 不可接受

上表中的数据来自我们在 GPT-4 Turbo 和 Claude 3.5 Sonnet 上的实际测试结果。具体数值会因模型版本、Prompt 设计、上下文长度等因素有所不同，但总体趋势是一致的：层次越高，可靠性越低。

Prompt Engineering 的价值边界可以这样概括：

Prompt Engineering 解决了"如何与模型沟通"的问题，但没有解决"如何让模型安全可靠地在生产环境中执行任务"的问题。前者是通信问题，后者是系统工程问题。

这一认识，自然地将我们引向了第二个范式——Context Engineering。

---

1.3 第二范式：Context Engineering 的系统化跃迁

1.3.1 从 Prompt 到 Context：信息论视角的范式转换

如果说 Prompt Engineering 关注的是"说什么"（What to say），那么 Context Engineering 关注的是"提供什么信息"（What to provide）。这是一个从"指令设计"到"信息架构"的范式跃迁。

Context Engineering 的定义：Context Engineering 是系统性地构建、管理和优化大语言模型运行时上下文的工程实践，目标是在有限的上下文窗口内，为模型提供完成任务所需的最相关信息。

Context Engineering 的核心洞察来自信息论中的一个基本原理：信道容量有限时，信息的选择和编码方式比信息本身更重要。

大语言模型的上下文窗口就是这个"有限信道"：

• GPT-4 Turbo：128K tokens（约 96,000 个中文字符）
• Claude 3.5：200K tokens（约 150,000 个中文字符）
• Gemini 1.5 Pro：1M tokens（约 750,000 个中文字符）

即使上下文窗口在不断扩大，信息论的基本约束依然存在：

1. 更多的上下文 = 更高的推理成本（Token 计费 + 延迟增加）
2. 更多的上下文 ≠ 更好的结果（"迷失在中间"效应，Lost in the Middle）
3. 上下文窗口的有效利用率通常远低于理论容量

1.3.2 Context Engineering 的五大维度

Context Engineering 可以分解为五个核心维度，每个维度都对应着一组工程技术和设计决策：

维度一：检索（Retrieval）

从外部知识源中获取与当前任务相关的信息。这就是 RAG（Retrieval-Augmented Generation）的核心。

关键技术：

• 向量嵌入（Embedding）+ 近似最近邻搜索（ANN）
• BM25 等稀疏检索
• 混合检索（Hybrid Retrieval）
• 查询重写（Query Rewriting）
• 多路召回 + 融合排序

维度二：压缩（Compression）

在不损失关键信息的前提下，减少上下文的 Token 数量。

关键技术：

• 摘要生成（Summarization）
• 关键信息提取（Key Information Extraction）
• Token 级别的压缩（如 LLMLingua）
• 对话历史的层级摘要
• 结构化数据的精简表示

维度三：排序（Prioritization）

根据信息的重要性和相关性，决定信息在上下文中的位置和呈现顺序。

关键技术：

• "Lost in the Middle"感知的排序策略
• 重要信息放在开头和结尾（首因效应 + 近因效应）
• 按相关性分数排序
• 按时间顺序排序（最近优先）
• 动态排序（根据任务类型调整）

维度四：裁剪（Pruning）

在上下文窗口接近容量限制时，有策略地移除低价值信息。

关键技术：

• 滑动窗口（固定保留最近 N 轮对话）
• 基于重要性的裁剪（保留重要信息，移除不重要信息）
• 基于角色的裁剪（保留 System Prompt 和最近消息，压缩历史消息）
• 动态裁剪（根据当前 Token 使用量自适应调整）

维度五：缓存（Caching）

通过缓存机制减少重复计算和 API 调用，降低成本和延迟。

关键技术：

• Prompt Caching（如 Claude 的 Cache 机制、OpenAI 的 Cached Tokens）
• 语义缓存（相似问题的结果复用）
• KV Cache 管理（推理层面的优化）
• 分层缓存（内存 → 磁盘 → 远程存储）

1.3.3 RAG 系统的进化：从 Naive 到 Modular

RAG（Retrieval-Augmented Generation）是 Context Engineering 最具代表性的应用，其自身也经历了显著的进化：

Naive RAG（基础 RAG）

用户查询 → 向量化 → 向量检索 → Top-K 结果 → 拼接到 Prompt → LLM 生成

Naive RAG 的问题：

• 检索结果可能不相关（检索噪声）
• 检索结果可能冗余（重复信息）
• 检索结果可能不完整（需要多跳推理）
• 缺乏对检索质量的评估

Advanced RAG（高级 RAG）

在 Naive RAG 的基础上引入了多项优化：

• 查询优化：查询重写、查询扩展、HyDE（Hypothetical Document Embedding）
• 检索优化：多路召回、重排序（Reranking）、交叉编码器精排
• 后处理：去重、压缩、摘要
• 评估反馈：检索质量评估、生成质量评估、端到端评估

Modular RAG（模块化 RAG）

最新一代的 RAG 架构将整个系统拆分为可组合的模块：

查询理解模块 → 检索策略模块 → 结果处理模块 → 上下文构建模块 → 生成模块
     ↓              ↓              ↓              ↓             ↓
  查询分类       路由选择        重排序          压缩/扩展     答案生成
  查询改写       模型选择        去重            排序/裁剪     引用标注
  意图识别       索引选择        融合            缓存管理      置信度评估

1.3.4 代码实战：构建一个完整的 Context Engineering 系统

让我们构建一个比 Prompt Engineering 案例更完整的 Context Engineering 系统。这个系统将为一个"智能技术支持 Agent"提供上下文管理服务。

TypeScript 实现

// context-engineering-system.ts
// 一个完整的 Context Engineering 系统，展示五大维度的工程实现

import { createHash } from 'crypto';

// ==================== 核心类型定义 ====================

interface Message {
  role: 'system' | 'user' | 'assistant' | 'tool';
  content: string;
  timestamp: number;
  metadata?: Record<string, any>;
}

interface RetrievedChunk {
  content: string;
  source: string;
  score: number;
  metadata: Record<string, any>;
}

interface ContextWindow {
  maxTokens: number;
  currentTokens: number;
  messages: Message[];
  retrievedDocs: RetrievedChunk[];
  cacheKey: string;
}

// ==================== Token 计数器（简化版） ====================

class TokenCounter {
  /**
   * 简化的 Token 计数——实际项目应使用 tiktoken 等库。
   * 中文大约 1.5 字符 ≈ 1 Token，英文大约 4 字符 ≈ 1 Token。
   */
  static count(text: string): number {
    const chineseChars = (text.match(/[一-鿿]/g) || []).length;
    const otherChars = text.length - chineseChars;
    return Math.ceil(chineseChars / 1.5 + otherChars / 4);
  }
}

// ==================== 维度一：检索（Retrieval） ====================

interface VectorStore {
  search(query: string, topK: number): Promise<RetrievedChunk[]>;
}

class HybridRetriever {
  /**
   * 混合检索器：结合向量检索和关键词检索。
   * 展示 Context Engineering 中"检索"维度的工程实践。
   */
  constructor(
    private vectorStore: VectorStore,
    private keywordIndex: Map<string, RetrievedChunk[]>,
    private vectorWeight: number = 0.7,
    private keywordWeight: number = 0.3,
  ) {}

  async retrieve(query: string, topK: number = 5): Promise<RetrievedChunk[]> {
    // 1. 向量检索（语义相似）
    const vectorResults = await this.vectorStore.search(query, topK * 2);

    // 2. 关键词检索（精确匹配）
    const keywordResults = this.keywordSearch(query, topK * 2);

    // 3. 融合排序（Reciprocal Rank Fusion）
    const fused = this.reciprocalRankFusion(
      vectorResults,
      keywordResults,
    );

    // 4. 去重（基于内容哈希）
    const deduplicated = this.deduplicate(fused);

    // 5. 返回 Top-K
    return deduplicated.slice(0, topK);
  }

  private keywordSearch(query: string, topK: number): RetrievedChunk[] {
    const results: RetrievedChunk[] = [];
    const queryTerms = query.toLowerCase().split(/\s+/);

    for (const [keyword, chunks] of this.keywordIndex) {
      if (queryTerms.some(term => keyword.includes(term))) {
        results.push(...chunks);
      }
    }

    return results.slice(0, topK);
  }

  private reciprocalRankFusion(
    vectorResults: RetrievedChunk[],
    keywordResults: RetrievedChunk[],
  ): RetrievedChunk[] {
    const k = 60; // RRF 常数
    const scoreMap = new Map<string, { chunk: RetrievedChunk; score: number }>();

    vectorResults.forEach((chunk, rank) => {
      const key = this.chunkKey(chunk);
      const rrfScore = this.vectorWeight * (1 / (k + rank + 1));
      const existing = scoreMap.get(key);
      if (existing) {
        existing.score += rrfScore;
      } else {
        scoreMap.set(key, { chunk, score: rrfScore });
      }
    });

    keywordResults.forEach((chunk, rank) => {
      const key = this.chunkKey(chunk);
      const rrfScore = this.keywordWeight * (1 / (k + rank + 1));
      const existing = scoreMap.get(key);
      if (existing) {
        existing.score += rrfScore;
      } else {
        scoreMap.set(key, { chunk, score: rrfScore });
      }
    });

    return Array.from(scoreMap.values())
      .sort((a, b) => b.score - a.score)
      .map(item => ({ ...item.chunk, score: item.score }));
  }

  private deduplicate(chunks: RetrievedChunk[]): RetrievedChunk[] {
    const seen = new Set<string>();
    return chunks.filter(chunk => {
      const key = this.chunkKey(chunk);
      if (seen.has(key)) return false;
      seen.add(key);
      return true;
    });
  }

  private chunkKey(chunk: RetrievedChunk): string {
    return createHash('md5').update(chunk.content).digest('hex');
  }
}

// ==================== 维度二：压缩（Compression） ====================

class ContextCompressor {
  /**
   * 上下文压缩器：通过摘要和提取关键信息来减少 Token 数量。
   * 展示 Context Engineering 中"压缩"维度的工程实践。
   */

  /**
   * 压缩对话历史：保留最近 N 轮完整对话，将更早的对话压缩为摘要。
   */
  static compressHistory(
    messages: Message[],
    keepRecent: number = 5,
    maxSummaryTokens: number = 500,
  ): Message[] {
    if (messages.length <= keepRecent) {
      return messages;
    }

    const recentMessages = messages.slice(-keepRecent);
    const oldMessages = messages.slice(0, -keepRecent);

    // 将旧消息压缩为一条摘要
    const summary = ContextCompressor.createSummary(oldMessages, maxSummaryTokens);

    return [summary, ...recentMessages];
  }

  /**
   * 创建对话摘要（简化版——实际应使用 LLM 生成摘要）。
   */
  private static createSummary(
    messages: Message[],
    maxTokens: number,
  ): Message {
    const keyPoints: string[] = [];

    for (const msg of messages) {
      if (msg.role === 'user') {
        // 提取用户消息的关键意图
        const intent = msg.content.slice(0, 100);
        keyPoints.push(`用户提到: ${intent}`);
      } else if (msg.role === 'assistant') {
        // 提取助手回复的关键信息
        const keyInfo = msg.content.slice(0, 150);
        keyPoints.push(`助手回复: ${keyInfo}`);
      }
    }

    const summaryText = `[对话历史摘要] ${keyPoints.join('; ')}`;

    return {
      role: 'system',
      content: summaryText.slice(0, maxTokens * 4), // 粗估
      timestamp: Date.now(),
      metadata: { type: 'history_summary', originalCount: messages.length },
    };
  }

  /**
   * 压缩检索结果：去重 + 摘要 + Token 预算控制。
   */
  static compressRetrievalResults(
    chunks: RetrievedChunk[],
    tokenBudget: number,
  ): RetrievedChunk[] {
    const compressed: RetrievedChunk[] = [];
    let usedTokens = 0;

    for (const chunk of chunks) {
      const chunkTokens = TokenCounter.count(chunk.content);

      if (usedTokens + chunkTokens > tokenBudget) {
        // 超出预算，尝试截断
        const remainingTokens = tokenBudget - usedTokens;
        if (remainingTokens > 50) { // 至少保留 50 个 Token
          const truncated: RetrievedChunk = {
            ...chunk,
            content: chunk.content.slice(0, remainingTokens * 4),
            metadata: { ...chunk.metadata, truncated: true },
          };
          compressed.push(truncated);
        }
        break;
      }

      compressed.push(chunk);
      usedTokens += chunkTokens;
    }

    return compressed;
  }
}

// ==================== 维度三：排序（Prioritization） ====================

class ContextPrioritizer {
  /**
   * 上下文排序器：根据信息的重要性和相关性安排上下文顺序。
   * 利用"首因效应"和"近因效应"优化信息位置。
   */

  /**
   * 构建优化排序的上下文：
   * - 开头：System Prompt（最重要）
   * - 中间：检索结果（按相关性排序）
   * - 结尾：最近的用户消息（最相关）
   */
  static buildPrioritizedContext(
    systemPrompt: string,
    history: Message[],
    retrievedDocs: RetrievedChunk[],
    currentQuery: string,
  ): Message[] {
    const context: Message[] = [];

    // 1. System Prompt 放在最前面（首因效应）
    context.push({
      role: 'system',
      content: systemPrompt,
      timestamp: Date.now(),
    });

    // 2. 压缩后的对话历史
    context.push(...history);

    // 3. 检索结果按相关性排序
    const sortedDocs = [...retrievedDocs].sort((a, b) => b.score - a.score);
    for (const doc of sortedDocs) {
      context.push({
        role: 'system',
        content: `[参考文档 - ${doc.source}]\n${doc.content}`,
        timestamp: Date.now(),
        metadata: { type: 'retrieval', score: doc.score },
      });
    }

    // 4. 当前用户查询放在最后（近因效应）
    context.push({
      role: 'user',
      content: currentQuery,
      timestamp: Date.now(),
    });

    return context;
  }
}

// ==================== 维度四：裁剪（Pruning） ====================

class ContextPruner {
  /**
   * 上下文裁剪器：在 Token 预算约束下，有策略地移除低价值信息。
   */

  static prune(
    messages: Message[],
    maxTokens: number,
    strategy: 'sliding_window' | 'importance_based' | 'role_based' = 'importance_based',
  ): Message[] {
    const totalTokens = messages.reduce(
      (sum, msg) => sum + TokenCounter.count(msg.content),
      0,
    );

    if (totalTokens <= maxTokens) {
      return messages; // 不需要裁剪
    }

    switch (strategy) {
      case 'sliding_window':
        return ContextPruner.slidingWindowPrune(messages, maxTokens);
      case 'importance_based':
        return ContextPruner.importanceBasedPrune(messages, maxTokens);
      case 'role_based':
        return ContextPruner.roleBasedPrune(messages, maxTokens);
    }
  }

  /**
   * 滑动窗口裁剪：只保留最近的 N 条消息。
   */
  private static slidingWindowPrune(
    messages: Message[],
    maxTokens: number,
  ): Message[] {
    const result: Message[] = [];
    let usedTokens = 0;

    // 从最新消息开始，向前添加
    for (let i = messages.length - 1; i >= 0; i--) {
      const msgTokens = TokenCounter.count(messages[i].content);
      if (usedTokens + msgTokens > maxTokens) break;
      result.unshift(messages[i]);
      usedTokens += msgTokens;
    }

    return result;
  }

  /**
   * 基于重要性的裁剪：根据消息的重要性分数排序，移除最不重要的。
   */
  private static importanceBasedPrune(
    messages: Message[],
    maxTokens: number,
  ): Message[] {
    // 为每条消息计算重要性分数
    const scored = messages.map(msg => ({
      message: msg,
      importance: ContextPruner.calculateImportance(msg),
    }));

    // 按重要性降序排序
    scored.sort((a, b) => b.importance - a.importance);

    // 贪心选择：从高重要性开始，直到超出 Token 预算
    const selected: Message[] = [];
    let usedTokens = 0;

    for (const item of scored) {
      const msgTokens = TokenCounter.count(item.message.content);
      if (usedTokens + msgTokens > maxTokens) continue;
      selected.push(item.message);
      usedTokens += msgTokens;
    }

    // 按时间顺序重新排序
    selected.sort((a, b) => a.timestamp - b.timestamp);
    return selected;
  }

  /**
   * 基于角色的裁剪：保护 System Prompt，压缩中间历史，保留最新消息。
   */
  private static roleBasedPrune(
    messages: Message[],
    maxTokens: number,
  ): Message[] {
    const systemMessages = messages.filter(m => m.role === 'system');
    const conversationMessages = messages.filter(m => m.role !== 'system');

    // System Prompt 永远保留
    const systemTokens = systemMessages.reduce(
      (sum, msg) => sum + TokenCounter.count(msg.content),
      0,
    );
    const remainingBudget = maxTokens - systemTokens;

    // 对话消息使用滑动窗口
    const prunedConversation = ContextPruner.slidingWindowPrune(
      conversationMessages,
      remainingBudget,
    );

    return [...systemMessages, ...prunedConversation];
  }

  private static calculateImportance(message: Message): number {
    let score = 0;

    // System 消息最重要
    if (message.role === 'system') score += 100;
    // 最近的消息更重要（时间衰减）
    const ageHours = (Date.now() - message.timestamp) / (1000 * 3600);
    score += Math.max(0, 50 - ageHours);
    // 用户消息比助手消息更重要
    if (message.role === 'user') score += 10;
    // 包含特定关键词的消息更重要
    if (/重要|紧急|必须|注意/i.test(message.content)) score += 20;

    return score;
  }
}

// ==================== 维度五：缓存（Caching） ====================

class SemanticCache {
  /**
   * 语义缓存：对相似的查询复用之前的结果。
   */
  private cache = new Map<string, {
    result: string;
    embedding: number[];
    timestamp: number;
    hitCount: number;
  }>();

  private readonly ttlMs: number;
  private readonly similarityThreshold: number;

  constructor(options: {
    ttlMs?: number;
    similarityThreshold?: number;
  } = {}) {
    this.ttlMs = options.ttlMs ?? 3600_000; // 默认 1 小时
    this.similarityThreshold = options.similarityThreshold ?? 0.95;
  }

  /**
   * 查询缓存。
   */
  async get(
    query: string,
    embeddingFn: (text: string) => Promise<number[]>,
  ): Promise<string | null> {
    const queryEmbedding = await embeddingFn(query);

    for (const [key, entry] of this.cache) {
      // 检查 TTL
      if (Date.now() - entry.timestamp > this.ttlMs) {
        this.cache.delete(key);
        continue;
      }

      // 计算余弦相似度
      const similarity = this.cosineSimilarity(queryEmbedding, entry.embedding);
      if (similarity >= this.similarityThreshold) {
        entry.hitCount++;
        return entry.result;
      }
    }

    return null;
  }

  /**
   * 写入缓存。
   */
  async set(
    query: string,
    result: string,
    embeddingFn: (text: string) => Promise<number[]>,
  ): Promise<void> {
    const embedding = await embeddingFn(query);
    const key = createHash('md5').update(query).digest('hex');

    this.cache.set(key, {
      result,
      embedding,
      timestamp: Date.now(),
      hitCount: 0,
    });
  }

  private cosineSimilarity(a: number[], b: number[]): number {
    if (a.length !== b.length) return 0;

    let dotProduct = 0;
    let normA = 0;
    let normB = 0;

    for (let i = 0; i < a.length; i++) {
      dotProduct += a[i] * b[i];
      normA += a[i] * a[i];
      normB += b[i] * b[i];
    }

    const denominator = Math.sqrt(normA) * Math.sqrt(normB);
    if (denominator === 0) return 0;

    return dotProduct / denominator;
  }

  get stats() {
    let totalHits = 0;
    for (const entry of this.cache.values()) {
      totalHits += entry.hitCount;
    }
    return {
      size: this.cache.size,
      totalHits,
      hitRate: this.cache.size > 0 ? totalHits / this.cache.size : 0,
    };
  }
}

// ==================== 上下文管理器（整合五大维度） ====================

class ContextEngineer {
  /**
   * 上下文工程师：整合检索、压缩、排序、裁剪、缓存五大维度，
   * 为 LLM 构建最优上下文。
   */
  constructor(
    private retriever: HybridRetriever,
    private cache: SemanticCache,
    private maxContextTokens: number = 8000,
  ) {}

  async buildContext(
    systemPrompt: string,
    conversationHistory: Message[],
    currentQuery: string,
    embeddingFn: (text: string) => Promise<number[]>,
  ): Promise<Message[]> {
    // 1. 检查缓存
    const cachedResult = await this.cache.get(currentQuery, embeddingFn);
    if (cachedResult) {
      console.log('✅ 缓存命中，跳过上下文构建');
      return [{ role: 'assistant', content: cachedResult, timestamp: Date.now() }];
    }

    // 2. 检索相关文档
    const retrievedDocs = await this.retriever.retrieve(currentQuery, 10);

    // 3. 压缩对话历史
    const compressedHistory = ContextCompressor.compressHistory(
      conversationHistory,
      5, // 保留最近 5 轮
      500, // 摘要最大 500 Token
    );

    // 4. 压缩检索结果
    const historyTokens = compressedHistory.reduce(
      (sum, msg) => sum + TokenCounter.count(msg.content),
      0,
    );
    const systemTokens = TokenCounter.count(systemPrompt);
    const queryTokens = TokenCounter.count(currentQuery);
    const docTokenBudget = this.maxContextTokens - systemTokens - historyTokens - queryTokens;

    const compressedDocs = ContextCompressor.compressRetrievalResults(
      retrievedDocs,
      Math.max(docTokenBudget, 200),
    );

    // 5. 排序构建上下文
    const context = ContextPrioritizer.buildPrioritizedContext(
      systemPrompt,
      compressedHistory,
      compressedDocs,
      currentQuery,
    );

    // 6. 最终裁剪（确保不超过 Token 预算）
    const prunedContext = ContextPruner.prune(
      context,
      this.maxContextTokens,
      'importance_based',
    );

    return prunedContext;
  }
}

// ==================== 使用示例 ====================

async function contextEngineeringDemo() {
  // 模拟向量存储
  const mockVectorStore: VectorStore = {
    async search(query: string, topK: number): Promise<RetrievedChunk[]> {
      return [
        {
          content: '产品退款政策：30天内可无理由退款，需保持产品完好。',
          source: 'policy/refund.md',
          score: 0.92,
          metadata: { category: 'refund' },
        },
        {
          content: 'VIP客户享受延长至60天的退款期限。',
          source: 'policy/vip.md',
          score: 0.85,
          metadata: { category: 'vip' },
        },
      ];
    },
  };

  const retriever = new HybridRetriever(
    mockVectorStore,
    new Map(),
  );
  const cache = new SemanticCache();
  const engineer = new ContextEngineer(retriever, cache);

  const history: Message[] = [
    { role: 'user', content: '我想问一下退款的事情', timestamp: Date.now() - 60000 },
    { role: 'assistant', content: '好的，请问您的订单号是什么？', timestamp: Date.now() - 55000 },
    { role: 'user', content: '订单号是 ORD-2025-12345', timestamp: Date.now() - 50000 },
  ];

  const context = await engineer.buildContext(
    '你是一个专业的客服助手，请根据参考文档回答客户问题。',
    history,
    '我购买的产品已经超过30天了，还能退款吗？我是VIP客户。',
    async (text: string) => new Array(128).fill(0), // 模拟 embedding
  );

  console.log('构建的上下文：');
  console.log(`总消息数：${context.length}`);
  console.log(`总 Token 数：${context.reduce((sum, msg) => sum + TokenCounter.count(msg.content), 0)}`);
  context.forEach((msg, i) => {
    console.log(`\n[${i}] ${msg.role}:`);
    console.log(msg.content.slice(0, 200));
  });
}

Python 实现

# context_engineering_system.py
# 一个完整的 Context Engineering 系统，展示五大维度的工程实现

from __future__ import annotations

import hashlib
import time
import math
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Awaitable, Optional


# ==================== 核心类型定义 ====================

@dataclass
class Message:
    role: str  # 'system' | 'user' | 'assistant' | 'tool'
    content: str
    timestamp: float = field(default_factory=time.time)
    metadata: dict[str, Any] = field(default_factory=dict)


@dataclass
class RetrievedChunk:
    content: str
    source: str
    score: float
    metadata: dict[str, Any] = field(default_factory=dict)


# ==================== Token 计数器 ====================

class TokenCounter:
    """简化的 Token 计数器。"""

    @staticmethod
    def count(text: str) -> int:
        chinese_chars = sum(1 for c in text if '一' <= c <= '鿿')
        other_chars = len(text) - chinese_chars
        return math.ceil(chinese_chars / 1.5 + other_chars / 4)


# ==================== 维度一：检索（Retrieval） ====================

class HybridRetriever:
    """
    混合检索器：结合向量检索和关键词检索。
    展示 Context Engineering 中"检索"维度的工程实践。
    """

    def __init__(
        self,
        vector_search_fn: Callable[[str, int], Awaitable[list[RetrievedChunk]]],
        keyword_index: dict[str, list[RetrievedChunk]],
        vector_weight: float = 0.7,
        keyword_weight: float = 0.3,
    ):
        self.vector_search_fn = vector_search_fn
        self.keyword_index = keyword_index
        self.vector_weight = vector_weight
        self.keyword_weight = keyword_weight

    async def retrieve(self, query: str, top_k: int = 5) -> list[RetrievedChunk]:
        # 1. 向量检索
        vector_results = await self.vector_search_fn(query, top_k * 2)

        # 2. 关键词检索
        keyword_results = self._keyword_search(query, top_k * 2)

        # 3. 融合排序（RRF）
        fused = self._reciprocal_rank_fusion(vector_results, keyword_results)

        # 4. 去重
        deduplicated = self._deduplicate(fused)

        return deduplicated[:top_k]

    def _keyword_search(self, query: str, top_k: int) -> list[RetrievedChunk]:
        results: list[RetrievedChunk] = []
        query_terms = query.lower().split()

        for keyword, chunks in self.keyword_index.items():
            if any(term in keyword for term in query_terms):
                results.extend(chunks)

        return results[:top_k]

    def _reciprocal_rank_fusion(
        self,
        vector_results: list[RetrievedChunk],
        keyword_results: list[RetrievedChunk],
    ) -> list[RetrievedChunk]:
        k = 60  # RRF 常数
        score_map: dict[str, tuple[RetrievedChunk, float]] = {}

        for rank, chunk in enumerate(vector_results):
            key = self._chunk_key(chunk)
            rrf_score = self.vector_weight * (1 / (k + rank + 1))
            if key in score_map:
                score_map[key] = (chunk, score_map[key][1] + rrf_score)
            else:
                score_map[key] = (chunk, rrf_score)

        for rank, chunk in enumerate(keyword_results):
            key = self._chunk_key(chunk)
            rrf_score = self.keyword_weight * (1 / (k + rank + 1))
            if key in score_map:
                score_map[key] = (chunk, score_map[key][1] + rrf_score)
            else:
                score_map[key] = (chunk, rrf_score)

        sorted_items = sorted(score_map.values(), key=lambda x: x[1], reverse=True)
        return [
            RetrievedChunk(
                content=chunk.content,
                source=chunk.source,
                score=score,
                metadata=chunk.metadata,
            )
            for chunk, score in sorted_items
        ]

    def _deduplicate(self, chunks: list[RetrievedChunk]) -> list[RetrievedChunk]:
        seen: set[str] = set()
        result: list[RetrievedChunk] = []
        for chunk in chunks:
            key = self._chunk_key(chunk)
            if key not in seen:
                seen.add(key)
                result.append(chunk)
        return result

    @staticmethod
    def _chunk_key(chunk: RetrievedChunk) -> str:
        return hashlib.md5(chunk.content.encode()).hexdigest()


# ==================== 维度二：压缩（Compression） ====================

class ContextCompressor:
    """上下文压缩器。"""

    @staticmethod
    def compress_history(
        messages: list[Message],
        keep_recent: int = 5,
        max_summary_tokens: int = 500,
    ) -> list[Message]:
        if len(messages) <= keep_recent:
            return messages

        recent = messages[-keep_recent:]
        old = messages[:-keep_recent]
        summary = ContextCompressor._create_summary(old, max_summary_tokens)
        return [summary] + recent

    @staticmethod
    def _create_summary(messages: list[Message], max_tokens: int) -> Message:
        key_points = []
        for msg in messages:
            if msg.role == "user":
                key_points.append(f"用户提到: {msg.content[:100]}")
            elif msg.role == "assistant":
                key_points.append(f"助手回复: {msg.content[:150]}")

        summary_text = f"[对话历史摘要] {'; '.join(key_points)}"
        return Message(
            role="system",
            content=summary_text[: max_tokens * 4],
            metadata={"type": "history_summary", "original_count": len(messages)},
        )

    @staticmethod
    def compress_retrieval_results(
        chunks: list[RetrievedChunk],
        token_budget: int,
    ) -> list[RetrievedChunk]:
        compressed: list[RetrievedChunk] = []
        used_tokens = 0

        for chunk in chunks:
            chunk_tokens = TokenCounter.count(chunk.content)
            if used_tokens + chunk_tokens > token_budget:
                remaining = token_budget - used_tokens
                if remaining > 50:
                    compressed.append(RetrievedChunk(
                        content=chunk.content[: remaining * 4],
                        source=chunk.source,
                        score=chunk.score,
                        metadata={**chunk.metadata, "truncated": True},
                    ))
                break
            compressed.append(chunk)
            used_tokens += chunk_tokens

        return compressed


# ==================== 维度三：排序（Prioritization） ====================

class ContextPrioritizer:
    """上下文排序器。"""

    @staticmethod
    def build_prioritized_context(
        system_prompt: str,
        history: list[Message],
        retrieved_docs: list[RetrievedChunk],
        current_query: str,
    ) -> list[Message]:
        context: list[Message] = []

        # 1. System Prompt（首因效应）
        context.append(Message(role="system", content=system_prompt))

        # 2. 对话历史
        context.extend(history)

        # 3. 检索结果（按相关性排序）
        sorted_docs = sorted(retrieved_docs, key=lambda d: d.score, reverse=True)
        for doc in sorted_docs:
            context.append(Message(
                role="system",
                content=f"[参考文档 - {doc.source}]\n{doc.content}",
                metadata={"type": "retrieval", "score": doc.score},
            ))

        # 4. 当前查询（近因效应）
        context.append(Message(role="user", content=current_query))

        return context


# ==================== 维度四：裁剪（Pruning） ====================

class PruningStrategy(Enum):
    SLIDING_WINDOW = "sliding_window"
    IMPORTANCE_BASED = "importance_based"
    ROLE_BASED = "role_based"


class ContextPruner:
    """上下文裁剪器。"""

    @staticmethod
    def prune(
        messages: list[Message],
        max_tokens: int,
        strategy: PruningStrategy = PruningStrategy.IMPORTANCE_BASED,
    ) -> list[Message]:
        total_tokens = sum(TokenCounter.count(m.content) for m in messages)
        if total_tokens <= max_tokens:
            return messages

        if strategy == PruningStrategy.SLIDING_WINDOW:
            return ContextPruner._sliding_window(messages, max_tokens)
        elif strategy == PruningStrategy.IMPORTANCE_BASED:
            return ContextPruner._importance_based(messages, max_tokens)
        else:
            return ContextPruner._role_based(messages, max_tokens)

    @staticmethod
    def _sliding_window(messages: list[Message], max_tokens: int) -> list[Message]:
        result: list[Message] = []
        used = 0
        for msg in reversed(messages):
            msg_tokens = TokenCounter.count(msg.content)
            if used + msg_tokens > max_tokens:
                break
            result.insert(0, msg)
            used += msg_tokens
        return result

    @staticmethod
    def _importance_based(messages: list[Message], max_tokens: int) -> list[Message]:
        scored = [
            (msg, ContextPruner._calculate_importance(msg))
            for msg in messages
        ]
        scored.sort(key=lambda x: x[1], reverse=True)

        selected: list[Message] = []
        used = 0
        for msg, _ in scored:
            msg_tokens = TokenCounter.count(msg.content)
            if used + msg_tokens > max_tokens:
                continue
            selected.append(msg)
            used += msg_tokens

        selected.sort(key=lambda m: m.timestamp)
        return selected

    @staticmethod
    def _role_based(messages: list[Message], max_tokens: int) -> list[Message]:
        system_msgs = [m for m in messages if m.role == "system"]
        conversation_msgs = [m for m in messages if m.role != "system"]

        system_tokens = sum(TokenCounter.count(m.content) for m in system_msgs)
        remaining = max_tokens - system_tokens

        pruned = ContextPruner._sliding_window(conversation_msgs, remaining)
        return system_msgs + pruned

    @staticmethod
    def _calculate_importance(message: Message) -> float:
        score = 0.0
        if message.role == "system":
            score += 100
        age_hours = (time.time() - message.timestamp) / 3600
        score += max(0, 50 - age_hours)
        if message.role == "user":
            score += 10
        return score


# ==================== 维度五：缓存（Caching） ====================

class SemanticCache:
    """语义缓存器。"""

    def __init__(
        self,
        ttl_seconds: float = 3600,
        similarity_threshold: float = 0.95,
    ):
        self.ttl_seconds = ttl_seconds
        self.similarity_threshold = similarity_threshold
        self._cache: dict[str, dict] = {}

    async def get(
        self,
        query: str,
        embedding_fn: Callable[[str], Awaitable[list[float]]],
    ) -> Optional[str]:
        query_embedding = await embedding_fn(query)

        for key, entry in list(self._cache.items()):
            if time.time() - entry["timestamp"] > self.ttl_seconds:
                del self._cache[key]
                continue

            similarity = self._cosine_similarity(query_embedding, entry["embedding"])
            if similarity >= self.similarity_threshold:
                entry["hit_count"] += 1
                return entry["result"]

        return None

    async def set(
        self,
        query: str,
        result: str,
        embedding_fn: Callable[[str], Awaitable[list[float]]],
    ) -> None:
        embedding = await embedding_fn(query)
        key = hashlib.md5(query.encode()).hexdigest()
        self._cache[key] = {
            "result": result,
            "embedding": embedding,
            "timestamp": time.time(),
            "hit_count": 0,
        }

    @staticmethod
    def _cosine_similarity(a: list[float], b: list[float]) -> float:
        if len(a) != len(b):
            return 0.0
        dot = sum(x * y for x, y in zip(a, b))
        norm_a = math.sqrt(sum(x * x for x in a))
        norm_b = math.sqrt(sum(x * x for x in b))
        if norm_a == 0 or norm_b == 0:
            return 0.0
        return dot / (norm_a * norm_b)

1.3.5 Context Engineering 的局限：解决了信息问题，没有解决控制问题

💡 关键洞察：Context Engineering 是对 Prompt Engineering 的增强，而非替代。它解决了"模型应该知道什么"的问题，但没有解决"模型应该做什么、怎么做、做错了怎么办"的问题。

让我们通过一个具体的场景来理解这个局限：

场景：AI Agent 执行数据库迁移

假设你有一个 AI Agent 负责执行数据库迁移脚本。你用 Context Engineering 为 Agent 提供了完美的上下文：

1. ✅ 检索：检索了相关的数据库 Schema 文档、历史迁移记录
2. ✅ 压缩：将冗长的 Schema 文档压缩为关键信息
3. ✅ 排序：将最相关的迁移模板放在上下文的开头
4. ✅ 裁剪：在 Token 预算内最大化有用信息的密度
5. ✅ 缓存：复用了之前类似迁移的执行计划

但即使上下文再完美，以下问题仍然无法通过 Context 解决：

• Agent 执行了一个 DROP TABLE 命令怎么办？ Context 可以"告诉" Agent 不要这么做，但无法在代码层面阻止它。
• 迁移过程中数据库连接断开怎么办？ Context 无法提供自动重试和断点续传机制。
• Agent 决定跳过备份步骤直接执行迁移怎么办？ Context 无法强制执行操作顺序。
• 迁移失败后如何回滚？ Context 无法提供事务管理和回滚机制。
• 如何记录完整的操作审计日志？ Context 无法提供日志记录和审计追踪。

这些问题的本质是控制问题——不是"Agent 应该知道什么"，而是"Agent 被允许做什么、如何确保Agent按照正确的流程执行、出了问题如何恢复"。

解决控制问题，需要一个全新的范式——Harness Engineering。

---

1.4 第三范式：Harness Engineering 的工程革命

1.4.1 Harness 的词源学：从马具到缰绳到操作系统

在深入技术讨论之前，让我们先追溯"Harness"这个词的含义演变，因为词源学往往能揭示概念的本质。

原始含义：马具（Horse Harness）

Harness 最初的含义是"马具"——一整套用于驾驭马匹的装备，包括：

• 缰绳（Reins）：控制方向和速度
• 鞍具（Saddle）：承载骑手或货物
• 马蹄铁（Horseshoes）：保护马蹄，适应不同地形
• 马笼头（Bridle）：约束马的行为
• 挽具（Traces）：连接马与车辆

关键洞察：马具不是"更好的马"，而是让马这种强大的动物能够被人类安全、高效地使用的工程系统。没有马具，马是一匹强壮但不可预测的野兽；有了马具，马成为了人类文明的推动力。

引申含义：控制与利用

在工程领域，“Harness"作为动词意味着"驾驭、利用”：

• Harness the power of the river（驾驭河流的力量）→ 水坝
• Harness solar energy（利用太阳能）→ 光伏板
• Harness nuclear energy（利用核能）→ 核反应堆

每个例子中，Harness 都代表一种将强大但不可控的自然力量转化为可控、可用的工程系统的完整方案。

AI Agent 语境：智能体的操作系统

在 AI Agent 的语境中，Harness 继承了这个词源学的传统：

Harness 是将大语言模型（强大但不可预测）转化为可安全、高效、可靠地在生产环境中运行的智能体的完整工程系统。

就像马具之于马、操作系统之于硬件一样，Harness 之于 LLM：

类比	原始力量	Harness 系统	结果
马具	马的力量和速度	缰绳+鞍具+马蹄铁	可控的交通工具
操作系统	CPU 的原始计算能力	进程管理+内存管理+文件系统	可控的计算平台
Harness	LLM 的推理和生成能力	控制+能动性+运行时	可控的智能体

1.4.2 He et al. (2026) 论文的核心理论解读

2026 年，He 等人发表的论文 Harness Engineering for Language Agents: The Harness Layer as Control, Agency, and Runtime 为 AI Agent 工程提供了第一个系统性的理论框架。

论文的核心论点：

1. Harness 是一个独立的架构层：它既不是模型的一部分（不是 Fine-tuning），也不是 Prompt 的一部分（不是 Prompt Engineering），而是位于模型和应用之间的一个独立的、可工程化设计的中间层。

2. Harness 由三个核心维度组成（CAR Trinity）：

   • Control（控制）：约束智能体的行为空间——定义"不能做什么"
   • Agency（能动性）：引导智能体的自主决策——定义"应该做什么"
   • Runtime（运行时）：提供智能体的基础设施——定义"如何运行"

3. Harness 的设计决定了 Agent 的行为上界：模型的推理能力决定了 Agent 的能力下界（最差表现），而 Harness 的设计决定了 Agent 的能力上界（最优表现）。

架构示意：

┌─────────────────────────────────────────────────────┐
│                    应用层（Application）               │
│         业务逻辑、用户界面、行业规则                     │
├─────────────────────────────────────────────────────┤
│                  Harness 层                           │
│  ┌───────────┐  ┌───────────┐  ┌───────────────────┐│
│  │  Control   │  │  Agency   │  │    Runtime        ││
│  │  行为约束   │  │  能动引导  │  │  运行基础设施      ││
│  │           │  │           │  │                   ││
│  │ · 权限控制 │  │ · 任务规划│  │ · 状态管理        ││
│  │ · 资源限制 │  │ · 工具选择│  │ · 工具执行        ││
│  │ · 安全策略 │  │ · 错误恢复│  │ · 上下文管理      ││
│  │ · 审计追踪 │  │ · 成本优化│  │ · 可观测性        ││
│  └───────────┘  └───────────┘  └───────────────────┘│
├─────────────────────────────────────────────────────┤
│                  模型层（Model）                       │
│           LLM 推理、生成、理解能力                      │
└─────────────────────────────────────────────────────┘

1.4.3 Agent = Model + Harness：公式的深度解析

这个看似简单的公式蕴含着深刻的工程哲学：

Agent = Model + Harness

解读一：构成关系

一个完整的 AI Agent 由两部分组成：

• Model：大语言模型，提供"智能"——理解语言、推理、生成
• Harness：工程系统，提供"控制"——约束行为、管理状态、执行工具

没有 Model，Harness 是一个空壳；没有 Harness，Model 是一个"裸奔的大脑"——聪明但不可控。

解读二：正交关系

Model 和 Harness 是正交的（Orthogonal）：

• 更换 Model（如从 GPT-4 切换到 Claude）不需要更换 Harness
• 升级 Harness（如添加新的工具、修改安全策略）不需要修改 Model
• 同一个 Harness 可以驱动不同的 Model
• 同一个 Model 可以在不同的 Harness 中运行

这种正交性是企业级部署的关键——它允许你独立地选择和升级 Model 和 Harness，而不是将它们耦合在一起。

解读三：乘法关系

更准确地说，Agent 的能力是 Model 和 Harness 的乘积，而非简单相加：

Agent 能力 = f(Model 能力 × Harness 质量)

这意味着：

• 一个能力很强的 Model + 一个很差的 Harness = 差的 Agent（强大的能力被糟糕的控制浪费）
• 一个能力一般的 Model + 一个优秀的 Harness = 好的 Agent（有限的能力被优秀的控制最大化）
• 一个能力很强的 Model + 一个优秀的 Harness = 卓越的 Agent

解读四：工程投入的转移

在 Prompt Engineering 时代，80% 的工程投入花在 Model 层面（调优 Prompt）。在 Harness Engineering 时代，工程投入应该发生转移：

Prompt Engineering 时代：  80% Prompt + 20% 基础设施
Context Engineering 时代： 60% 上下文管理 + 30% Prompt + 10% 基础设施
Harness Engineering 时代： 30% Prompt + 20% 上下文 + 50% Harness（控制+运行时+工具）

1.4.4 CAR Trinity 预览

CAR Trinity 是 Harness Engineering 的理论核心，我们将在第 2 章深入展开，这里先做一个预览：

C — Control（控制维度）

控制维度回答的问题是：Agent 被允许做什么？不被允许做什么？

核心组件：

• 权限系统（Permission System）：定义 Agent 可以访问的资源、工具和操作
• 约束规则（Constraint Rules）：硬性约束（绝不能做）和软性约束（尽量避免做）
• 审批流程（Approval Workflow）：在关键操作节点引入人工确认
• 审计追踪（Audit Trail）：记录所有操作和决策，支持事后分析

A — Agency（能动性维度）

能动性维度回答的问题是：Agent 应该如何自主决策？如何引导其做出好的决策？

核心组件：

• 任务规划（Task Planning）：将复杂任务分解为可执行的子任务
• 工具选择（Tool Selection）：根据任务需求选择合适的工具
• 错误恢复（Error Recovery）：在工具调用失败时自动重试或切换策略
• 成本优化（Cost Optimization）：在保证质量的前提下最小化 Token 消耗

R — Runtime（运行时维度）

运行时维度回答的问题是：Agent 如何被运行、管理和监控？

核心组件：

• 状态管理（State Management）：跨对话轮次的状态持久化和恢复
• 工具执行（Tool Execution）：安全、可靠地执行工具调用
• 上下文管理（Context Management）：动态构建和管理上下文窗口
• 可观测性（Observability）：实时监控 Agent 的行为和性能

三者的关系：

         Control（约束）
           /        \
          /          \
         /    互相     \
        /    制约      \
       /       &        \
      /       协同        \
     /                    \
  Agency ──────────────── Runtime
 （能动性）              （运行时）

• Control 约束 Agency：Agent 的自主决策不能超出控制维度定义的行为边界
• Agency 利用 Runtime：Agent 的自主决策通过运行时维度来执行
• Runtime 支撑 Control：控制维度的规则通过运行时维度来强制执行
• 三者必须平衡：过度控制会限制能动性，过度能动会突破控制，运行时是两者的基础

1.4.5 代码实战：一个最小化的 Harness 原型

让我们通过代码来理解 Harness 与 Prompt/Context 的本质区别。我们将构建一个最小化的 Harness 原型，展示 Control、Agency、Runtime 三个维度的基本功能。

TypeScript 实现：最小化 Harness 原型

// minimal-harness.ts
// 一个最小化的 Harness 原型，展示与 Prompt/Context 的本质区别

// ==================== 核心类型 ====================

interface Tool {
  name: string;
  description: string;
  parameters: Record<string, { type: string; description: string }>;
  execute: (params: Record<string, any>) => Promise<string>;
}

interface Permission {
  resource: string;
  actions: string[];
}

interface AuditEntry {
  timestamp: number;
  agentId: string;
  action: string;
  details: Record<string, any>;
  result: 'success' | 'failure' | 'blocked';
}

interface HarnessConfig {
  agentId: string;
  maxTokens: number;
  maxToolCalls: number;
  maxRetries: number;
  timeoutMs: number;
  permissions: Permission[];
  requireApproval: string[];  // 需要人工确认的工具
  allowedModels: string[];
}

// ==================== 维度一：Control（控制） ====================

class HarnessController {
  /**
   * 控制层：约束 Agent 的行为空间。
   * 
   * 与 Prompt Engineering 的关键区别：
   * - Prompt "请求" Agent 不要做某事（建议性的）
   * - Controller "强制" Agent 不能做某事（强制性的）
   */
  private auditLog: AuditEntry[] = [];

  constructor(private config: HarnessConfig) {}

  /**
   * 权限检查：在工具调用前强制执行权限验证。
   * 这是 Harness 与 Prompt 的核心区别——这里是代码级别的强制检查，
   * 而不是提示词级别的"请求"。
   */
  checkPermission(toolName: string, params: Record<string, any>): {
    allowed: boolean;
    reason?: string;
  } {
    // 检查工具是否在允许列表中
    const permission = this.config.permissions.find(
      p => p.resource === toolName || p.resource === '*'
    );

    if (!permission) {
      this.logAudit('permission_denied', {
        tool: toolName,
        params,
        reason: 'tool_not_permitted',
      }, 'blocked');
      return { allowed: false, reason: `工具 "${toolName}" 不在允许列表中` };
    }

    // 检查特定操作权限
    if (toolName === 'database_query') {
      const query = (params.query as string || '').toUpperCase();
      const dangerousOps = ['DROP', 'DELETE', 'TRUNCATE', 'ALTER'];
      const hasDangerousOp = dangerousOps.some(op => query.includes(op));

      if (hasDangerousOp && !permission.actions.includes('write')) {
        this.logAudit('permission_denied', {
          tool: toolName,
          params,
          reason: 'dangerous_operation',
        }, 'blocked');
        return { allowed: false, reason: '不允许执行写操作' };
      }
    }

    return { allowed: true };
  }

  /**
   * 人工审批检查：某些操作需要人工确认。
   */
  requiresApproval(toolName: string): boolean {
    return this.config.requireApproval.includes(toolName);
  }

  /**
   * Token 预算检查：限制单次任务的 Token 消耗。
   */
  checkTokenBudget(usedTokens: number): {
    withinBudget: boolean;
    remaining: number;
  } {
    const remaining = this.config.maxTokens - usedTokens;
    return {
      withinBudget: remaining > 0,
      remaining: Math.max(0, remaining),
    };
  }

  /**
   * 工具调用次数检查：限制工具调用频率。
   */
  checkToolCallLimit(currentCount: number): {
    withinLimit: boolean;
    remaining: number;
  } {
    const remaining = this.config.maxToolCalls - currentCount;
    return {
      withinLimit: remaining > 0,
      remaining: Math.max(0, remaining),
    };
  }

  /**
   * 审计日志记录。
   */
  logAudit(
    action: string,
    details: Record<string, any>,
    result: 'success' | 'failure' | 'blocked',
  ): void {
    this.auditLog.push({
      timestamp: Date.now(),
      agentId: this.config.agentId,
      action,
      details,
      result,
    });
  }

  getAuditLog(): AuditEntry[] {
    return [...this.auditLog];
  }
}

// ==================== 维度二：Agency（能动性） ====================

class HarnessAgency {
  /**
   * 能动性层：引导 Agent 的自主决策。
   * 
   * 与 Prompt Engineering 的关键区别：
   * - Prompt 只能给出"一次性"的指令
   * - Agency 在每一步决策点都提供动态的引导
   */
  private toolRegistry = new Map<string, Tool>();

  /**
   * 注册可用工具。
   */
  registerTool(tool: Tool): void {
    this.toolRegistry.set(tool.name, tool);
  }

  /**
   * 工具选择引导：根据任务上下文，推荐最合适的工具。
   * 这不是"命令" Agent 使用某个工具，而是"引导"它做出更好的选择。
   */
  recommendTool(taskDescription: string): {
    recommended: string[];
    reasoning: string;
  } {
    const recommended: string[] = [];
    let reasoning = '';

    // 基于关键词的简单推荐（实际项目应使用更复杂的方法）
    for (const [name, tool] of this.toolRegistry) {
      if (taskDescription.toLowerCase().includes('查询') ||
          taskDescription.toLowerCase().includes('数据库')) {
        if (name.includes('database') || name.includes('query')) {
          recommended.push(name);
          reasoning += `推荐 "${name}"：${tool.description}\n`;
        }
      }
      if (taskDescription.toLowerCase().includes('文件') ||
          taskDescription.toLowerCase().includes('读取')) {
        if (name.includes('file') || name.includes('read')) {
          recommended.push(name);
          reasoning += `推荐 "${name}"：${tool.description}\n`;
        }
      }
    }

    return { recommended, reasoning };
  }

  /**
   * 错误恢复策略：当工具调用失败时，提供恢复方案。
   * 这是 Prompt Engineering 完全无法实现的。
   */
  async executeWithRecovery(
    toolName: string,
    params: Record<string, any>,
    maxRetries: number,
  ): Promise<{
    success: boolean;
    result: string;
    attempts: number;
    strategy: string;
  }> {
    const tool = this.toolRegistry.get(toolName);
    if (!tool) {
      return {
        success: false,
        result: `工具 "${toolName}" 未注册`,
        attempts: 0,
        strategy: 'none',
      };
    }

    let lastError: Error | null = null;
    let strategy = 'direct';

    for (let attempt = 1; attempt <= maxRetries; attempt++) {
      try {
        const result = await tool.execute(params);
        return { success: true, result, attempts: attempt, strategy };
      } catch (error) {
        lastError = error as Error;

        // 根据错误类型选择恢复策略
        if (attempt === 1) {
          strategy = 'retry_same';
        } else if (attempt === 2) {
          strategy = 'retry_with_backoff';
          await this.delay(1000 * attempt); // 指数退避
        } else {
          strategy = 'fallback';
          // 尝试替代方案
          const fallbackResult = await this.tryFallback(toolName, params);
          if (fallbackResult) {
            return {
              success: true,
              result: fallbackResult,
              attempts: attempt,
              strategy: 'fallback',
            };
          }
        }
      }
    }

    return {
      success: false,
      result: `失败 ${maxRetries} 次后放弃: ${lastError?.message}`,
      attempts: maxRetries,
      strategy,
    };
  }

  private async tryFallback(
    toolName: string,
    params: Record<string, any>,
  ): Promise<string | null> {
    // 简化的 fallback 逻辑
    if (toolName === 'database_query') {
      // 数据库查询失败，尝试使用缓存
      return '[Fallback] 使用缓存数据（可能不是最新的）';
    }
    return null;
  }

  private delay(ms: number): Promise<void> {
    return new Promise(resolve => setTimeout(resolve, ms));
  }
}

// ==================== 维度三：Runtime（运行时） ====================

class HarnessRuntime {
  /**
   * 运行时层：提供 Agent 的运行基础设施。
   * 
   * 与 Prompt Engineering 的关键区别：
   * - Prompt Engineering 没有"运行时"概念——每次调用都是独立的
   * - Harness Runtime 提供跨调用的状态管理、资源管理和可观测性
   */
  private state = new Map<string, any>();
  private metrics: Record<string, number[]> = {};

  /**
   * 状态管理：持久化和恢复 Agent 状态。
   */
  setState(key: string, value: any): void {
    this.state.set(key, value);
  }

  getState<T>(key: string, defaultValue?: T): T | undefined {
    return (this.state.get(key) as T) ?? defaultValue;
  }

  /**
   * 性能指标收集。
   */
  recordMetric(name: string, value: number): void {
    if (!this.metrics[name]) {
      this.metrics[name] = [];
    }
    this.metrics[name].push(value);
  }

  getMetrics(name: string): {
    count: number;
    avg: number;
    min: number;
    max: number;
    p95: number;
  } {
    const values = this.metrics[name] || [];
    if (values.length === 0) {
      return { count: 0, avg: 0, min: 0, max: 0, p95: 0 };
    }

    const sorted = [...values].sort((a, b) => a - b);
    const sum = values.reduce((a, b) => a + b, 0);

    return {
      count: values.length,
      avg: sum / values.length,
      min: sorted[0],
      max: sorted[sorted.length - 1],
      p95: sorted[Math.floor(sorted.length * 0.95)],
    };
  }

  /**
   * 超时执行：确保工具调用不会无限期运行。
   */
  async executeWithTimeout<T>(
    fn: () => Promise<T>,
    timeoutMs: number,
    label: string,
  ): Promise<{ success: boolean; result?: T; error?: string; durationMs: number }> {
    const start = Date.now();

    return Promise.race([
      fn().then(result => ({
        success: true as const,
        result,
        durationMs: Date.now() - start,
      })),
      new Promise<{ success: false; error: string; durationMs: number }>(resolve => {
        setTimeout(() => {
          resolve({
            success: false,
            error: `${label} 超时（${timeoutMs}ms）`,
            durationMs: timeoutMs,
          });
        }, timeoutMs);
      }),
    ]);
  }
}

// ==================== Harness 引擎（整合三个维度） ====================

class MinimalHarness {
  /**
   * 最小化 Harness 引擎：整合 Control、Agency、Runtime 三个维度。
   * 
   * 这就是 Harness 与 Prompt Engineering 的本质区别：
   * - Prompt Engineering：一个精心设计的字符串
   * - Harness Engineering：一个完整的工程系统
   */
  readonly controller: HarnessController;
  readonly agency: HarnessAgency;
  readonly runtime: HarnessRuntime;

  constructor(config: HarnessConfig) {
    this.controller = new HarnessController(config);
    this.agency = new HarnessAgency();
    this.runtime = new HarnessRuntime();
  }

  /**
   * 执行一个工具调用——展示 Harness 的完整流程。
   * 
   * 对比 Prompt Engineering：
   * - Prompt：直接调用 LLM，让它"决定"使用什么工具
   * - Harness：每个工具调用都经过权限检查、预算检查、超时管理、
   *   错误恢复、审计记录等一系列控制流程
   */
  async executeToolCall(
    toolName: string,
    params: Record<string, any>,
  ): Promise<{
    success: boolean;
    result: string;
    auditTrail: AuditEntry[];
    metrics: Record<string, any>;
  }> {
    const startTime = Date.now();

    // === Control 维度 ===
    // 1. 权限检查（强制性，不可绕过）
    const permissionCheck = this.controller.checkPermission(toolName, params);
    if (!permissionCheck.allowed) {
      return {
        success: false,
        result: `权限拒绝: ${permissionCheck.reason}`,
        auditTrail: this.controller.getAuditLog(),
        metrics: {},
      };
    }

    // 2. 人工审批检查
    if (this.controller.requiresApproval(toolName)) {
      // 在实际系统中，这里会暂停执行并等待人工确认
      console.log(`⚠️ 工具 "${toolName}" 需要人工审批`);
      // 简化起见，我们假设审批通过
    }

    // 3. Token 预算检查
    const tokenCheck = this.controller.checkTokenBudget(
      this.runtime.getState<number>('usedTokens', 0)!,
    );
    if (!tokenCheck.withinBudget) {
      return {
        success: false,
        result: `Token 预算耗尽（剩余: ${tokenCheck.remaining}）`,
        auditTrail: this.controller.getAuditLog(),
        metrics: {},
      };
    }

    // === Agency 维度 ===
    // 4. 带错误恢复的工具执行
    const executionResult = await this.agency.executeWithRecovery(
      toolName,
      params,
      3, // maxRetries
    );

    // === Runtime 维度 ===
    // 5. 超时保护
    const durationMs = Date.now() - startTime;
    this.runtime.recordMetric('tool_call_duration_ms', durationMs);
    this.runtime.recordMetric('token_usage', 100); // 模拟 Token 消耗

    // 6. 状态更新
    this.runtime.setState(
      'usedTokens',
      (this.runtime.getState<number>('usedTokens', 0) ?? 0) + 100,
    );
    this.runtime.setState('lastToolCall', { toolName, params, timestamp: Date.now() });

    // 7. 审计记录
    this.controller.logAudit(
      `tool_call:${toolName}`,
      { params, result: executionResult.result.slice(0, 200) },
      executionResult.success ? 'success' : 'failure',
    );

    return {
      success: executionResult.success,
      result: executionResult.result,
      auditTrail: this.controller.getAuditLog(),
      metrics: {
        durationMs,
        attempts: executionResult.attempts,
        strategy: executionResult.strategy,
        tokenUsed: this.runtime.getState<number>('usedTokens', 0),
      },
    };
  }
}

// ==================== 对比演示 ====================

async function demonstrateDifference() {
  console.log('=== Prompt Engineering 方式 ===');
  console.log('const prompt = "你是一个数据库管理员，请执行以下SQL查询..."');
  console.log('const result = await openai.chat.completions.create({ messages: [{ content: prompt }] });');
  console.log('→ 没有权限检查、没有错误恢复、没有审计日志、没有超时保护\n');

  console.log('=== Harness Engineering 方式 ===');

  // 创建 Harness
  const harness = new MinimalHarness({
    agentId: 'db-admin-agent',
    maxTokens: 10000,
    maxToolCalls: 20,
    maxRetries: 3,
    timeoutMs: 30000,
    permissions: [
      { resource: 'database_query', actions: ['read'] },
      { resource: 'log_search', actions: ['read'] },
      // 注意：没有 'database_admin' 权限
    ],
    requireApproval: ['database_query'], // 数据库查询需要审批
    allowedModels: ['gpt-4-turbo', 'claude-3.5-sonnet'],
  });

  // 注册工具
  harness.agency.registerTool({
    name: 'database_query',
    description: '执行数据库查询',
    parameters: {
      query: { type: 'string', description: 'SQL 查询语句' },
    },
    execute: async (params) => {
      // 模拟数据库查询
      return JSON.stringify({ rows: [{ id: 1, name: '测试数据' }] });
    },
  });

  // 执行工具调用——经过完整的 Harness 流程
  const result = await harness.executeToolCall('database_query', {
    query: 'SELECT * FROM users WHERE status = "active"',
  });

  console.log('执行结果：', result.success ? '✅ 成功' : '❌ 失败');
  console.log('结果数据：', result.result);
  console.log('审计记录：', result.auditTrail.length, '条');
  console.log('性能指标：', result.metrics);

  // 尝试执行未授权的操作
  console.log('\n--- 尝试未授权操作 ---');
  const blockedResult = await harness.executeToolCall('database_admin', {
    query: 'DROP TABLE users',
  });
  console.log('执行结果：', blockedResult.success ? '✅ 成功' : '❌ 被阻止');
  console.log('阻止原因：', blockedResult.result);
  console.log('审计记录：', blockedResult.auditTrail.length, '条（包含阻止记录）');
}

Python 实现：最小化 Harness 原型

# minimal_harness.py
# 一个最小化的 Harness 原型，展示与 Prompt/Context 的本质区别

from __future__ import annotations

import asyncio
import time
import json
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Awaitable, Optional


# ==================== 核心类型 ====================

class AuditResult(Enum):
    SUCCESS = "success"
    FAILURE = "failure"
    BLOCKED = "blocked"


@dataclass
class Tool:
    name: str
    description: str
    parameters: dict[str, dict[str, str]]
    execute_fn: Callable[[dict[str, Any]], Awaitable[str]]


@dataclass
class Permission:
    resource: str
    actions: list[str]


@dataclass
class AuditEntry:
    timestamp: float
    agent_id: str
    action: str
    details: dict[str, Any]
    result: AuditResult


@dataclass
class HarnessConfig:
    agent_id: str
    max_tokens: int = 10000
    max_tool_calls: int = 20
    max_retries: int = 3
    timeout_ms: int = 30000
    permissions: list[Permission] = field(default_factory=list)
    require_approval: list[str] = field(default_factory=list)
    allowed_models: list[str] = field(default_factory=list)


# ==================== 维度一：Control（控制） ====================

class HarnessController:
    """
    控制层：约束 Agent 的行为空间。
    
    与 Prompt Engineering 的关键区别：
    - Prompt "请求" Agent 不要做某事（建议性的）
    - Controller "强制" Agent 不能做某事（强制性的）
    """

    def __init__(self, config: HarnessConfig):
        self.config = config
        self._audit_log: list[AuditEntry] = []

    def check_permission(
        self, tool_name: str, params: dict[str, Any]
    ) -> tuple[bool, Optional[str]]:
        """权限检查：在工具调用前强制执行。"""
        # 检查工具是否在允许列表中
        permission = None
        for p in self.config.permissions:
            if p.resource == tool_name or p.resource == "*":
                permission = p
                break

        if permission is None:
            self.log_audit(
                "permission_denied",
                {"tool": tool_name, "reason": "tool_not_permitted"},
                AuditResult.BLOCKED,
            )
            return False, f'工具 "{tool_name}" 不在允许列表中'

        # 检查危险操作
        if tool_name == "database_query":
            query = str(params.get("query", "")).upper()
            dangerous_ops = ["DROP", "DELETE", "TRUNCATE", "ALTER"]
            if any(op in query for op in dangerous_ops):
                if "write" not in permission.actions:
                    self.log_audit(
                        "permission_denied",
                        {"tool": tool_name, "reason": "dangerous_operation"},
                        AuditResult.BLOCKED,
                    )
                    return False, "不允许执行写操作"

        return True, None

    def requires_approval(self, tool_name: str) -> bool:
        """检查是否需要人工审批。"""
        return tool_name in self.config.require_approval

    def check_token_budget(self, used_tokens: int) -> tuple[bool, int]:
        """Token 预算检查。"""
        remaining = self.config.max_tokens - used_tokens
        return remaining > 0, max(0, remaining)

    def log_audit(
        self,
        action: str,
        details: dict[str, Any],
        result: AuditResult,
    ) -> None:
        """审计日志记录。"""
        self._audit_log.append(AuditEntry(
            timestamp=time.time(),
            agent_id=self.config.agent_id,
            action=action,
            details=details,
            result=result,
        ))

    @property
    def audit_log(self) -> list[AuditEntry]:
        return list(self._audit_log)


# ==================== 维度二：Agency（能动性） ====================

class HarnessAgency:
    """
    能动性层：引导 Agent 的自主决策。
    
    与 Prompt Engineering 的关键区别：
    - Prompt 只能给出"一次性"的指令
    - Agency 在每一步决策点都提供动态的引导
    """

    def __init__(self):
        self._tools: dict[str, Tool] = {}

    def register_tool(self, tool: Tool) -> None:
        """注册可用工具。"""
        self._tools[tool.name] = tool

    def recommend_tool(self, task_description: str) -> tuple[list[str], str]:
        """工具选择引导：根据任务上下文推荐工具。"""
        recommended: list[str] = []
        reasoning = ""

        task_lower = task_description.lower()
        for name, tool in self._tools.items():
            if any(kw in task_lower for kw in ["查询", "数据库", "query"]):
                if "database" in name or "query" in name:
                    recommended.append(name)
                    reasoning += f'推荐 "{name}"：{tool.description}\n'

        return recommended, reasoning

    async def execute_with_recovery(
        self,
        tool_name: str,
        params: dict[str, Any],
        max_retries: int = 3,
    ) -> dict[str, Any]:
        """
        错误恢复策略：当工具调用失败时，自动重试。
        这是 Prompt Engineering 完全无法实现的。
        """
        tool = self._tools.get(tool_name)
        if not tool:
            return {
                "success": False,
                "result": f'工具 "{tool_name}" 未注册',
                "attempts": 0,
                "strategy": "none",
            }

        last_error: Optional[Exception] = None
        strategy = "direct"

        for attempt in range(1, max_retries + 1):
            try:
                result = await tool.execute_fn(params)
                return {
                    "success": True,
                    "result": result,
                    "attempts": attempt,
                    "strategy": strategy,
                }
            except Exception as e:
                last_error = e

                if attempt == 1:
                    strategy = "retry_same"
                elif attempt == 2:
                    strategy = "retry_with_backoff"
                    await asyncio.sleep(1.0 * attempt)
                else:
                    strategy = "fallback"
                    fallback_result = await self._try_fallback(tool_name, params)
                    if fallback_result is not None:
                        return {
                            "success": True,
                            "result": fallback_result,
                            "attempts": attempt,
                            "strategy": "fallback",
                        }

        return {
            "success": False,
            "result": f"失败 {max_retries} 次后放弃: {last_error}",
            "attempts": max_retries,
            "strategy": strategy,
        }

    async def _try_fallback(
        self, tool_name: str, params: dict[str, Any]
    ) -> Optional[str]:
        """简化的 fallback 逻辑。"""
        if tool_name == "database_query":
            return "[Fallback] 使用缓存数据（可能不是最新的）"
        return None


# ==================== 维度三：Runtime（运行时） ====================

class HarnessRuntime:
    """
    运行时层：提供 Agent 的运行基础设施。
    
    与 Prompt Engineering 的关键区别：
    - Prompt Engineering 没有"运行时"概念
    - Harness Runtime 提供跨调用的状态管理、资源管理和可观测性
    """

    def __init__(self):
        self._state: dict[str, Any] = {}
        self._metrics: dict[str, list[float]] = {}

    def set_state(self, key: str, value: Any) -> None:
        self._state[key] = value

    def get_state(self, key: str, default: Any = None) -> Any:
        return self._state.get(key, default)

    def record_metric(self, name: str, value: float) -> None:
        if name not in self._metrics:
            self._metrics[name] = []
        self._metrics[name].append(value)

    def get_metric_stats(self, name: str) -> dict[str, float]:
        values = self._metrics.get(name, [])
        if not values:
            return {"count": 0, "avg": 0, "min": 0, "max": 0, "p95": 0}

        sorted_vals = sorted(values)
        return {
            "count": len(values),
            "avg": sum(values) / len(values),
            "min": sorted_vals[0],
            "max": sorted_vals[-1],
            "p95": sorted_vals[int(len(sorted_vals) * 0.95)],
        }

    async def execute_with_timeout(
        self,
        fn: Callable[[], Awaitable[Any]],
        timeout_ms: int,
        label: str,
    ) -> dict[str, Any]:
        """超时执行：确保工具调用不会无限期运行。"""
        start = time.time()
        timeout_sec = timeout_ms / 1000

        try:
            result = await asyncio.wait_for(fn(), timeout=timeout_sec)
            return {
                "success": True,
                "result": result,
                "duration_ms": (time.time() - start) * 1000,
            }
        except asyncio.TimeoutError:
            return {
                "success": False,
                "error": f"{label} 超时（{timeout_ms}ms）",
                "duration_ms": timeout_ms,
            }


# ==================== Harness 引擎 ====================

class MinimalHarness:
    """
    最小化 Harness 引擎：整合 Control、Agency、Runtime 三个维度。
    
    这就是 Harness 与 Prompt Engineering 的本质区别：
    - Prompt Engineering：一个精心设计的字符串
    - Harness Engineering：一个完整的工程系统
    """

    def __init__(self, config: HarnessConfig):
        self.controller = HarnessController(config)
        self.agency = HarnessAgency()
        self.runtime = HarnessRuntime()

    async def execute_tool_call(
        self,
        tool_name: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """执行工具调用——经过完整的 Harness 控制流程。"""
        start_time = time.time()

        # === Control 维度 ===
        # 1. 权限检查（强制性，不可绕过）
        allowed, reason = self.controller.check_permission(tool_name, params)
        if not allowed:
            return {
                "success": False,
                "result": f"权限拒绝: {reason}",
                "audit_trail": self.controller.audit_log,
                "metrics": {},
            }

        # 2. 人工审批检查
        if self.controller.requires_approval(tool_name):
            print(f'⚠️ 工具 "{tool_name}" 需要人工审批')

        # 3. Token 预算检查
        used_tokens = self.runtime.get_state("used_tokens", 0)
        within_budget, remaining = self.controller.check_token_budget(used_tokens)
        if not within_budget:
            return {
                "success": False,
                "result": f"Token 预算耗尽（剩余: {remaining}）",
                "audit_trail": self.controller.audit_log,
                "metrics": {},
            }

        # === Agency 维度 ===
        # 4. 带错误恢复的工具执行
        exec_result = await self.agency.execute_with_recovery(
            tool_name, params, max_retries=3
        )

        # === Runtime 维度 ===
        # 5. 性能指标记录
        duration_ms = (time.time() - start_time) * 1000
        self.runtime.record_metric("tool_call_duration_ms", duration_ms)
        self.runtime.record_metric("token_usage", 100)

        # 6. 状态更新
        current_tokens = self.runtime.get_state("used_tokens", 0)
        self.runtime.set_state("used_tokens", current_tokens + 100)
        self.runtime.set_state("last_tool_call", {
            "tool": tool_name,
            "timestamp": time.time(),
        })

        # 7. 审计记录
        self.controller.log_audit(
            f"tool_call:{tool_name}",
            {"params": params, "result_preview": str(exec_result["result"])[:200]},
            AuditResult.SUCCESS if exec_result["success"] else AuditResult.FAILURE,
        )

        return {
            "success": exec_result["success"],
            "result": exec_result["result"],
            "audit_trail": self.controller.audit_log,
            "metrics": {
                "duration_ms": duration_ms,
                "attempts": exec_result["attempts"],
                "strategy": exec_result["strategy"],
                "tokens_used": self.runtime.get_state("used_tokens", 0),
            },
        }


# ==================== 对比演示 ====================

async def demonstrate_difference():
    print("=== Prompt Engineering 方式 ===")
    print('prompt = "你是一个数据库管理员，请执行以下SQL查询..."')
    print("result = await client.chat.completions.create(messages=[...])")
    print("→ 没有权限检查、没有错误恢复、没有审计日志、没有超时保护\n")

    print("=== Harness Engineering 方式 ===")

    # 创建 Harness
    harness = MinimalHarness(HarnessConfig(
        agent_id="db-admin-agent",
        max_tokens=10000,
        max_tool_calls=20,
        max_retries=3,
        timeout_ms=30000,
        permissions=[
            Permission(resource="database_query", actions=["read"]),
            Permission(resource="log_search", actions=["read"]),
        ],
        require_approval=["database_query"],
        allowed_models=["gpt-4-turbo", "claude-3.5-sonnet"],
    ))

    # 注册工具
    async def mock_db_query(params: dict[str, Any]) -> str:
        return json.dumps({"rows": [{"id": 1, "name": "测试数据"}]})

    harness.agency.register_tool(Tool(
        name="database_query",
        description="执行数据库查询",
        parameters={"query": {"type": "string", "description": "SQL 查询语句"}},
        execute_fn=mock_db_query,
    ))

    # 执行工具调用——经过完整的 Harness 流程
    result = await harness.execute_tool_call(
        "database_query",
        {"query": 'SELECT * FROM users WHERE status = "active"'},
    )

    print(f"执行结果：{'✅ 成功' if result['success'] else '❌ 失败'}")
    print(f"结果数据：{result['result']}")
    print(f"审计记录：{len(result['audit_trail'])} 条")
    print(f"性能指标：{result['metrics']}")

    # 尝试未授权操作
    print("\n--- 尝试未授权操作 ---")
    blocked = await harness.execute_tool_call(
        "database_admin",
        {"query": "DROP TABLE users"},
    )
    print(f"执行结果：{'✅ 成功' if blocked['success'] else '❌ 被阻止'}")
    print(f"阻止原因：{blocked['result']}")


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

1.4.6 Harness 与 Prompt/Context 的本质区别

通过以上代码，我们可以清晰地看到三者的本质区别：

对比维度	Prompt Engineering	Context Engineering	Harness Engineering
核心产物	一个字符串	一个信息管道	一个工程系统
控制机制	建议性的（“请不要…”）	信息性的（“这里是相关信息…”）	强制性的（代码级权限检查）
错误处理	无（模型自行决定）	无（模型自行决定）	自动重试、回退、熔断
状态管理	无（每次独立调用）	有限（上下文窗口内）	完整（持久化状态管理）
安全机制	无（依赖模型遵从）	无（依赖模型判断）	多层防御（权限、审批、审计）
可观测性	无（黑盒）	有限（检索质量指标）	完整（全链路追踪+指标）
成本控制	无	有限（压缩）	完整（预算、限流、熔断）
可扩展性	不可扩展	有限扩展	完整（工具注册、配置热更新）

💡 关键洞察：Prompt Engineering 是给模型的一封信，Context Engineering 是给模型的一个资料包，Harness Engineering 是给模型的一个操作系统。三者不是替代关系，而是层级包含关系：一个好的 Harness 系统内部包含优秀的 Context Engineering，一个好的 Context Engineering 内部包含优秀的 Prompt Engineering。

┌────────────────────────────────────────────────┐
│                Harness Layer                    │
│   ┌────────────────────────────────────────┐    │
│   │          Context Layer                  │    │
│   │   ┌─────────────────────────────────┐   │    │
│   │   │         Prompt Layer            │   │    │
│   │   │                                 │   │    │
│   │   │  "请根据以下信息回答用户问题"     │   │    │
│   │   │                                 │   │    │
│   │   └─────────────────────────────────┘   │    │
│   │                                         │    │
│   │  检索 + 压缩 + 排序 + 裁剪 + 缓存       │    │
│   │                                         │    │
│   └────────────────────────────────────────┘    │
│                                                 │
│  权限 + 审批 + 审计 + 重试 + 超时 + 状态管理     │
│                                                 │
└────────────────────────────────────────────────┘

---

1.5 三次范式进化的对比分析

1.5.1 详细对比：Prompt vs Context vs Harness

经过前几节的深入分析，现在我们可以从更多维度进行系统性的对比：

基础属性对比

属性	Prompt Engineering	Context Engineering	Harness Engineering
工程产物	文本字符串	信息管道	运行时系统
控制方式	建议式（soft）	信息式（informative）	强制式（hard）
作用时机	调用前	调用前+调用中	全生命周期
依赖模型	强依赖	中等依赖	弱依赖
可替换模型	需重新调优 Prompt	需调整检索策略	无缝替换
工程复杂度	低	中	高
适合团队规模	1-3人	3-10人	10+人
适合项目阶段	PoC/Demo	MVP/早期产品	生产级系统

能力维度对比

能力维度	Prompt	Context	Harness
任务理解	✅ 通过指令引导	✅ 通过上下文增强	✅ 通过结构化任务分解
状态管理	❌ 无	⚠️ 窗口内有限	✅ 跨会话持久化
工具编排	❌ 无	❌ 无	✅ 完整 DAG 编排
错误处理	❌ 无	❌ 无	✅ 重试/回退/熔断
安全控制	❌ 无	❌ 无	✅ 多层防御
可观测性	❌ 黑盒	⚠️ 检索指标	✅ 全链路追踪
成本治理	❌ 无	⚠️ Token 压缩	✅ 预算/限流/熔断
多 Agent 协作	❌ 无	❌ 无	✅ 编排+通信+隔离
配置管理	⚠️ 硬编码	⚠️ 半结构化	✅ 声明式配置
热更新	❌ 需重启	⚠️ 部分支持	✅ 配置热更新

控制能力对比

控制类型	Prompt	Context	Harness
访问控制（ACL）	❌	❌	✅ 代码级强制执行
操作审批	❌	❌	✅ 工作流引擎
速率限制	❌	❌	✅ 令牌桶/滑动窗口
资源配额	❌	❌	✅ 多维度配额管理
审计追踪	❌	❌	✅ 完整操作日志
行为回滚	❌	❌	✅ 事务+补偿
合规检查	❌	❌	✅ 规则引擎

1.5.2 适用场景分析

什么时候用 Prompt Engineering 就够了？

场景	原因	示例
单次文本生成任务	无需状态管理、工具调用	翻译、摘要、邮件撰写
内部原型验证	快速验证想法	PoC Demo
简单的分类/提取	任务边界清晰	情感分析、NER
个人使用场景	无需安全/合规	个人助手、学习笔记
一次性批处理	无需长期运行	批量数据处理

什么时候必须用 Harness Engineering？

场景	原因	示例
生产环境部署	需要可靠性保证	企业客服、内部工具
涉及敏感操作	需要安全控制	数据库操作、资金处理
多步骤复杂任务	需要工具编排和错误恢复	运维自动化、数据分析
多 Agent 协作	需要编排和隔离	工作流自动化
成本敏感场景	需要成本治理	大规模 API 调用
合规要求高的行业	需要审计追踪	金融、医疗、法律
高并发场景	需要并发控制和资源管理	SaaS 平台

1.5.3 AI Agent 工程的五级成熟度模型（L1-L5）

基于三次范式进化，我们可以定义一个 AI Agent 工程的成熟度模型，帮助组织评估自己的工程水平并制定改进路线图：

级别	名称	特征描述	关键技术	适用组织
L1	Prompt 级	直接使用 LLM API，依赖手工调优 Prompt	System Prompt + Few-shot	个人开发者、创业初期
L2	Context 级	引入 RAG 和上下文管理，系统化信息组织	RAG + 对话压缩 + 缓存	小型团队、MVP 阶段
L3	Harness 级（基础）	引入基础的控制和运行时，具备基本的可靠性和安全性	权限控制 + 工具注册 + 错误处理 + 基本监控	中型团队、产品化阶段
L4	Harness 级（企业）	完整的 Harness 系统，支持多 Agent 协作和企业级部署	CAR Trinity + 多 Agent 编排 + 完整可观测性 + 合规	大型企业、关键业务系统
L5	自进化级	Harness 系统具备自我优化和自适应能力	MetaHarness + 自动调参 + 异常自愈 + 知识积累	技术领先企业

成熟度评估问卷（快速自评）：

请回答以下 10 个问题（是/否），计算你的组织处于哪个级别：

1. 你的 AI 系统是否主要依赖 System Prompt 来控制 Agent 行为？→ 如果是，L1
2. 你是否引入了 RAG 或向量数据库来增强模型的知识？→ 如果是，L2
3. 你的 Agent 是否具有持久化的状态管理？→ 如果是，L3
4. 你的 Agent 的工具调用是否有权限控制？→ 如果是，L3
5. 你的 Agent 是否有自动错误恢复机制？→ 如果是，L3
6. 你是否有完整的 Agent 行为审计日志？→ 如果是，L4
7. 你的系统是否支持多个 Agent 协同工作？→ 如果是，L4
8. 你的 Harness 是否具有跨模型的兼容性？→ 如果是，L4
9. 你的 Harness 是否能自动优化自身的配置参数？→ 如果是，L5
10. 你的系统是否具备异常自愈能力？→ 如果是，L5

---

1.6 企业级 AI 的 Harness 转型之路

1.6.1 企业 AI 部署的真实痛点案例

案例一：金融行业的智能投研 Agent

某头部券商在 2024 年启动了一个 AI 投研 Agent 项目，目标是辅助分析师进行行业研究和投资报告撰写。初始方案采用 Prompt Engineering + RAG 的架构：

初始架构（L2 级别）：

• 使用 GPT-4 Turbo 作为底层模型
• System Prompt 中包含了详细的投研分析框架和合规要求
• RAG 系统接入了公司的研报库、财务数据库和新闻库

遇到的问题：

1. 数据泄露风险：在一次演示中，Agent 在回答某个行业分析问题时，"创造性地"引用了一份标记为"机密"的内部并购评估报告，并将关键数据泄露给了不具备查看权限的用户。原因是 RAG 系统没有在检索层面实施权限过滤，而 Prompt 中的"不要泄露机密信息"指令被模型忽略了。

2. 分析一致性差：不同分析师使用同一套 Prompt 和 RAG 系统，但得到的投研报告质量参差不齐。Agent 的分析框架"漂移"严重——有时遵循 DCF 模型，有时使用相对估值法，有时则完全"自由发挥"。Prompt 中虽然指定了分析框架，但模型在长上下文中经常"忘记"这些约束。

3. 成本失控：由于缺乏 Token 预算管理，某些复杂的研究任务（如涉及多个行业的对比分析）会触发数百次 RAG 检索和极长的推理过程，单次任务的成本高达数十美元。

Harness 转型方案（L4 级别）：

• Control 层：实施数据访问权限控制（按用户角色过滤 RAG 检索结果）、分析报告模板强制校验（确保遵循指定框架）、Token 预算硬限制（每任务上限 50K tokens）
• Agency 层：结构化分析流程（数据收集 → 行业分析 → 财务建模 → 风险评估 → 投资建议），每步都有明确的输入输出规范和质量检查点
• Runtime 层：完整的执行追踪和审计日志、成本实时监控和告警、分析报告的版本管理和对比

转型效果：

• 数据泄露事件降为零
• 分析报告一致性提升 40%（通过模板校验和流程强制）
• 平均任务成本降低 60%（通过 Token 预算和缓存机制）
• 分析师满意度提升 35%（Agent 行为更可预测、更可控）

案例二：医疗行业的智能辅助诊断 Agent

某三甲医院在 2025 年试点了一个 AI 辅助诊断系统，帮助医生进行初步诊断和检查项目推荐。

遇到的问题：

1. 诊断建议不安全：Agent 在没有充分信息的情况下，给出了过于"自信"的诊断建议，导致部分医生过度依赖 AI 建议而减少了自身的判断。一个典型案例：Agent 将一位患者的症状诊断为"普通感冒"，但实际是早期心肌炎——因为 Agent 没有主动询问关键的心血管相关症状。

2. 缺乏"不知道"的能力：当遇到罕见病或复杂病例时，Agent 不会说"我不确定"，而是强行给出一个"看起来合理"的诊断。Prompt 中的"遇到不确定的情况请说明"指令在复杂上下文中的遵从率不足 60%。

3. 合规审计缺失：医疗行业对 AI 系统的决策过程有严格的审计要求，但初始系统无法完整记录 Agent 的推理过程和依据。

Harness 转型方案：

• Control 层：强制"不确定性声明"机制（当诊断置信度低于阈值时，Agent 必须输出"建议进一步检查"而非具体诊断）、诊断结果与医学知识库的交叉验证、完整的决策审计链
• Agency 层：结构化问诊流程（主诉 → 现病史 → 既往史 → 体格检查 → 辅助检查 → 初步诊断 → 鉴别诊断），每步都有信息完整性检查
• Runtime 层：诊断过程可视化（医生可以看到 Agent 的推理路径和依据）、与 HIS（医院信息系统）的深度集成、实时药物相互作用检查

案例三：制造行业的智能质检 Agent

某汽车零部件制造商部署了一个 AI 质检 Agent，用于分析生产线上的缺陷照片并给出质量判定。

遇到的问题：

1. 判定标准不一致：Agent 对相似的缺陷照片给出不同的判定结果（有时判为"合格"，有时判为"不合格"），因为 Prompt 中的质量标准描述不够精确，且模型在不同上下文中的解读不一致。

2. 与 MES 系统集成困难：Agent 的判定结果需要写入 MES（制造执行系统），但缺乏标准化的集成接口和事务管理，偶尔出现判定结果丢失或重复写入的问题。

3. 无法追踪判定依据：当客户投诉某个零件质量问题时，无法追溯当初 Agent 是基于什么依据做出的"合格"判定。

这些案例共同揭示了一个规律：企业级 AI 部署的失败，99% 不是因为模型不够好，而是因为缺乏系统化的 Harness 工程。

1.6.2 转型路线图：从 PoC 到生产的五阶段方法论

基于大量的企业实践，我们总结出了一套从 PoC 到生产环境的五阶段转型方法论：

阶段一          阶段二          阶段三          阶段四          阶段五
PoC 验证    →   能力评估    →   基础 Harness →   企业 Harness →   自进化 Harness
（2-4周）       （1-2周）       （4-8周）       （8-16周）       （持续迭代）

阶段一：PoC 验证（2-4 周）

目标：验证 AI Agent 在目标场景中的可行性。

• 使用 Prompt Engineering + 简单 RAG 快速构建原型
• 选择 1-2 个代表性场景进行验证
• 收集用户反馈和性能数据
• 关键产出：可行性报告、初步 ROI 评估、技术风险清单

阶段二：能力评估（1-2 周）

目标：明确从 PoC 到生产需要的工程投入。

• 对照成熟度模型评估当前水平（通常是 L1 或 L2）
• 确定目标成熟度级别（通常是 L3 或 L4）
• 识别差距和优先级
• 关键产出：差距分析报告、转型路线图、资源需求计划

阶段三：基础 Harness 建设（4-8 周）

目标：建立 Harness 的基础能力，达到 L3 级别。

• 实施权限控制和工具注册机制
• 建立基本的错误处理和重试逻辑
• 添加基础的日志记录和监控
• 实现跨会话的状态管理
• 关键产出：基础 Harness 框架、第一批生产场景的上线

阶段四：企业 Harness 完善（8-16 周）

目标：完善 Harness 到企业级水平，达到 L4 级别。

• 实现完整的 CAR Trinity（控制+能动性+运行时）
• 支持多 Agent 协作和编排
• 建立完整的可观测性体系
• 实施合规审计和安全加固
• 关键产出：完整的企业级 Harness 平台、全部生产场景的迁移

阶段五：自进化 Harness（持续迭代）

目标：让 Harness 系统具备自我优化能力，向 L5 级别演进。

• 实现配置参数的自动调优
• 建立异常自愈机制
• 引入知识积累和经验学习
• 探索 MetaHarness（管理 Harness 的 Harness）
• 关键产出：持续优化的 Harness 平台、技术壁垒形成

1.6.3 组织能力建设：AI 工程团队的角色与技能矩阵

Harness 转型不仅是技术转型，也是组织转型。一个完整的企业级 AI Agent 团队需要以下角色：

角色	职责	核心技能	人数配比
AI 产品经理	场景定义、需求分析、ROI 评估	LLM 能力认知、业务流程理解、用户研究	1:3（每3个开发配1个PM）
Harness 架构师	Harness 系统设计、技术选型	分布式系统、安全架构、LLM 应用	1:8
Prompt 工程师	Prompt 设计与优化、评估体系	语言学、心理学、LLM 特性、A/B 测试	1:5
Agent 开发工程师	Agent 功能开发、工具集成	TypeScript/Python、API 设计、工具开发	核心团队
平台工程师	Harness 平台运维、可观测性	K8s/Docker、监控、日志、SRE	1:10
安全工程师	安全策略设计、合规审计	安全架构、数据隐私、合规标准	1:15
AI 质量工程师	Agent 行为测试、质量保证	测试策略、评估框架、自动化测试	1:5

1.6.4 ROI 分析框架：Harness 投资的量化评估

Harness 工程需要投入工程资源，因此需要清晰的 ROI 评估框架：

成本项

成本类别	初始投入	持续成本
人力成本	Harness 架构师 + 开发团队（4-16 周）	运维 + 迭代（持续）
基础设施	监控、日志、审计系统	存储 + 计算
学习成本	团队培训、技术预研	新成员 onboarding
迁移成本	PoC 到生产的迁移工作	新场景的接入

收益项

收益类别	量化方式	典型数值
事故减少	减少的安全事件 × 每次事故成本	事故率降低 70-90%
成本优化	Token 消耗减少 × 单价	API 成本降低 40-60%
效率提升	人工干预减少 × 人力时薪	运维工时减少 50-70%
质量提升	一致性提升 × 质量成本	错误率降低 60-80%
合规价值	避免的合规罚款	难以量化（但可能是最大的）

ROI 计算公式：

ROI = (年度收益 - 年度成本) / 年度成本 × 100%

根据我们的行业经验，一个 L4 级别的 Harness 系统在金融行业的首年 ROI 通常在 150%-300% 之间，医疗行业在 200%-400% 之间（合规价值较高），制造行业在 100%-200% 之间。

1.6.5 案例分析：某金融机构的 Agent Harness 转型全过程

让我们完整回顾一个真实的 Harness 转型案例（已脱敏）。

背景：某资产管理公司（管理规模约 500 亿人民币）在 2024 年 Q3 启动了 AI Agent 项目，目标是提升投研效率和风控能力。

阶段一：PoC（2024 年 Q3，3 周）

团队使用 GPT-4 + LangChain 快速构建了三个 PoC：

• 研报摘要 Agent：自动提取研报关键观点
• 风险评估 Agent：分析投资组合的风险敞口
• 会议纪要 Agent：自动生成投委会会议纪要

PoC 效果令人兴奋：研报摘要的准确率达到了 85%，会议纪要的可用率达到了 78%。管理层决定投入生产。

阶段二：生产化尝试（2024 年 Q4，6 周）

团队在 PoC 基础上直接"硬编码"生产功能：

• 将 System Prompt 从 500 字扩展到 5000 字（添加了各种约束和规则）
• 接入了公司的 Wind 金融数据终端（作为 RAG 数据源）
• 开发了简单的 Web 界面供分析师使用

问题爆发：上线两周后，问题密集出现：

1. 一位分析师的对话中，Agent 错误地将另一分析师的私有研究笔记作为"公开信息"引用
2. 一次系统故障导致所有 Agent 会话的状态丢失，分析师们需要重新开始所有未完成的研究
3. API 成本在一个月内达到了预算的 3 倍（某些深度研究任务消耗了大量 Token）
4. 合规部门发现无法审计 Agent 的决策过程，要求暂停使用

阶段三：Harness 转型（2025 年 Q1-Q2，12 周）

痛定思痛后，团队决定进行系统性的 Harness 转型：

第 1-2 周：架构设计
  - 确定 CAR Trinity 架构
  - 设计权限模型（RBAC + ABAC 混合）
  - 规划工具注册和执行框架

第 3-6 周：Control 层建设
  - 实现数据访问权限控制（按角色、部门、数据分类）
  - 建立操作审批流程（高风险操作需主管确认）
  - 实施完整的审计日志系统

第 7-9 周：Agency 层建设
  - 构建结构化的投研分析工作流
  - 实现错误恢复和降级策略
  - 建立 Token 预算管理

第 10-12 周：Runtime 层建设
  - 实现跨会话的状态持久化
  - 建立完整的可观测性体系（traces + metrics + logs）
  - 与 Wind 数据终端的深度集成

转型效果（截至 2025 年 Q3）：

指标	转型前	转型后	变化
数据泄露事件	3次/月	0次/月	-100%
分析师满意度	45%	82%	+82%
API 月度成本	¥180,000	¥72,000	-60%
平均任务完成时间	45分钟	22分钟	-51%
合规审计通过率	不通过	通过	✅
系统可用性	92%	99.5%	+8.2%

这个案例生动地说明了：从 PoC 到生产的鸿沟，不是更大的模型或更好的 Prompt 能跨越的，而是需要系统化的 Harness 工程。

---

1.7 本书导读与知识地图

1.7.1 全书结构总览

本书围绕 CAR Trinity（Control × Agency × Runtime） 理论框架，通过十二个章节系统性地展开 Harness Engineering 的理论、架构和实践。全书分为四个篇章：

第一篇：理论基础（第 1-2 章）
    第 1 章：从 Prompt 到 Harness：AI Agent 工程的三次范式进化  ← 你在读的这一章
    第 2 章：CAR 三元理论：Control-Agency-Runtime 的统一场论

第二篇：CAR 三维度深入（第 3-5 章）
    第 3 章：控制论视角：约束即力量的工程哲学
    第 4 章：能动性工程：被引导的自主性与结构化决策
    第 5 章：运行时架构：智能体的操作系统

第三篇：企业级 Harness 五大支柱（第 6-10 章）
    第 6 章：企业级 Harness 平台架构设计
    第 7 章：结构化知识系统：Harness 的记忆与认知引擎
    第 8 章：机械化架构约束：行为空间与资源治理
    第 9 章：可观测性注入：透明化运行时
    第 10 章：自修复闭环：智能体的免疫系统

第四篇：实战与展望（第 11-12 章）
    第 11 章：HarnessForge 实战：从零构建企业级 Harness 平台
    第 12 章：安全工程、多 Agent 协同与未来展望

各章之间的逻辑关系：

第1章（范式进化）
    ↓ 引出
第2章（CAR 理论）
    ↓ 展开为三个维度
    ├── 第3章（Control）
    ├── 第4章（Agency）
    └── 第5章（Runtime）
        ↓ 整合为
第6章（企业架构）
    ↓ 细化为五大支柱
    ├── 第7章（知识系统）
    ├── 第8章（架构约束）
    ├── 第9章（可观测性）
    └── 第10章（自修复）
        ↓ 综合实战
第11章（HarnessForge 平台）
    ↓ 展望
第12章（安全+多Agent+未来）

1.7.2 核心理论线索：CAR Trinity 如何贯穿全书

CAR Trinity 不是一个孤立的理论框架，它是贯穿全书的"红线"：

• 第 2 章定义 CAR Trinity 的完整理论，建立统一的概念模型
• 第 3-5 章分别深入 C、A、R 三个维度，每个维度都从理论基础讲到工程实践
• 第 6-10 章将 CAR Trinity 落地为企业级架构的五大支柱：
  • 知识系统 = Agency 的认知基础
  • 架构约束 = Control 的工程实现
  • 可观测性 = Runtime 的透明化
  • 自修复 = Control + Agency + Runtime 的协同
• 第 11 章将三个维度整合为一个完整的平台（HarnessForge）

1.7.3 代码项目说明：HarnessForge 平台的渐进式构建

本书的核心代码项目是 HarnessForge——一个从零构建的企业级 Harness 平台。

HarnessForge 的技术栈：

• 后端：TypeScript + Node.js（主服务） + Python（ML 管道）
• 前端：React + TypeScript（管理控制台）
• 基础设施：Docker + Kubernetes + PostgreSQL + Redis
• LLM 集成：支持 OpenAI、Anthropic Claude、开源模型
• 可观测性：OpenTelemetry + Prometheus + Grafana

每章的代码都是 HarnessForge 的一个增量模块：

章节	代码模块	功能
第 1 章	harness-core	最小化 Harness 原型
第 2 章	car-trinity	CAR Trinity 核心抽象
第 3 章	harness-control	权限、审批、审计系统
第 4 章	harness-agency	任务规划、工具选择、错误恢复
第 5 章	harness-runtime	状态管理、工具执行、上下文管理
第 6 章	harness-arch	企业级架构框架
第 7 章	knowledge-system	结构化知识系统
第 8 章	constraint-engine	行为约束引擎
第 9 章	observability	可观测性系统
第 10 章	self-healing	自修复系统
第 11 章	harnessforge	完整平台集成
第 12 章	security-multiagent	安全 + 多 Agent 扩展

所有代码都是渐进式的——每章新增的代码都建立在前一章的基础上，最终在第 11 章整合为一个完整的平台。

1.7.4 阅读建议：不同角色的推荐阅读路径

路径一：CTO/技术总监（战略视角）

推荐阅读：第 1 章 → 第 2 章 → 第 6 章 → 第 12 章

• 重点关注：范式进化的战略意义、CAR Trinity 的全局视角、企业架构设计、安全与合规

路径二：架构师（系统设计视角）

推荐阅读：全书顺序阅读，重点关注第 2-6 章

• 重点关注：CAR Trinity 的详细设计、五大支柱的架构模式、平台集成

路径三：Agent 开发工程师（实现视角）

推荐阅读：第 1 章 → 第 3-5 章 → 第 7-10 章 → 第 11 章

• 重点关注：三个维度的代码实现、五大支柱的工程实践、HarnessForge 的完整代码

路径四：Prompt 工程师（转型视角）

推荐阅读：第 1 章 → 第 2 章 → 第 4 章 → 第 7 章

• 重点关注：从 Prompt 到 Harness 的思维转换、能动性工程、知识系统

路径五：运维/SRE（运维视角）

推荐阅读：第 5 章 → 第 9 章 → 第 10 章 → 第 11 章

• 重点关注：运行时架构、可观测性、自修复、部署运维

1.7.5 前置知识要求与学习资源

必要前置知识：

• TypeScript 或 Python 的中级以上编程能力
• HTTP API 设计和使用经验
• 基本的 LLM API 调用经验（OpenAI、Anthropic 等）
• Docker 容器的基本使用

建议前置知识（非必须，但有助于理解）：

• 分布式系统设计基础
• 控制论和系统论基本概念
• Kubernetes 基本操作
• 微服务架构经验

推荐学习资源：

资源类型	推荐资源	用途
LLM 基础	OpenAI API 文档、Anthropic 文档	理解模型能力边界
系统设计	《Designing Data-Intensive Applications》	分布式系统基础
控制论	《Control Theory for Engineers》（入门章节即可）	理解控制论概念
Agent 框架	Claude Agent SDK 文档、LangChain 文档	理解现有工具生态
可观测性	《Cloud-Native Observability》	监控和追踪

---

1.8 本章小结与思考题

1.8.1 核心概念回顾

本章围绕 AI Agent 工程的三次范式进化 这一主线，建立了以下核心概念：

概念一：三次范式进化

Prompt Engineering（指令设计）
    → Context Engineering（信息架构）
        → Harness Engineering（系统工程）

每一次进化都不是替代，而是层级包含——Harness 包含 Context，Context 包含 Prompt。

概念二：Agent = Model + Harness

这是全书的核心公式。Model 提供智能（推理和生成能力），Harness 提供控制（约束、引导和运行基础设施）。两者是正交的、乘法的关系。

概念三：CAR Trinity

Harness 由三个维度组成：

• Control（控制）：约束行为空间——定义"不能做什么"
• Agency（能动性）：引导自主决策——定义"应该做什么"
• Runtime（运行时）：提供运行基础设施——定义"如何运行"

三者互相制约、互相协同，缺一不可。

概念四：企业级 AI 的工程鸿沟

从 PoC 到生产环境的鸿沟，不是更大的模型或更好的 Prompt 能跨越的。需要系统化的 Harness 工程，包括权限控制、错误恢复、状态管理、可观测性、成本治理等一系列工程能力。

1.8.2 关键洞察总结

💡 洞察一：Prompt Engineering 是"建议式"控制，Harness Engineering 是"强制式"控制。前者依赖模型的遵从，后者依赖代码的执行。这是最根本的区别。

💡 洞察二：企业级 AI 部署的失败，99% 不是因为模型不够好，而是因为缺乏 Harness 工程。模型能力的"最后一公里"是由 Harness 完成的。

💡 洞察三：Harness 不是一个产品，而是一个工程范式。它不是某个框架或库的专利，而是一种系统性的工程思维方式。

💡 洞察四：CAR Trinity 的三个维度必须平衡发展。过度控制会扼杀 Agent 的灵活性，过度能动性会突破安全边界，运行时不充分则两者都无法可靠运行。

💡 洞察五：从 Prompt 到 Harness 的转型不仅是技术转型，也是组织转型。它需要新的角色（Harness 架构师）、新的流程（控制审查）和新的文化（工程化思维优先于"调参"思维）。

1.8.3 思考题

思考题 1（概念理解）：

请解释为什么说"Prompt Engineering 是开环控制，Harness Engineering 是闭环控制"？给出一个具体的例子来说明两者的区别。

思考题 2（场景分析）：

假设你要构建一个 AI Agent 来自动化处理客户的退款请求。请分析：

• 哪些部分只需要 Prompt Engineering 就够了？
• 哪些部分需要 Context Engineering？
• 哪些部分必须用 Harness Engineering？
• 如果只用 Prompt Engineering 来实现整个系统，可能会遇到什么问题？

思考题 3（架构设计）：

在你目前的工作中，是否已经有使用 AI Agent 的场景？如果有，请评估它目前处于成熟度模型的哪个级别，并分析需要哪些 Harness 能力来达到下一个级别。如果没有，请设想一个适合你所在行业的 AI Agent 场景，并设计其 Harness 的基本架构。

思考题 4（代码实践）：

修改本章的 MinimalHarness 原型（TypeScript 或 Python 版本），添加以下功能：

• 工具调用的并发控制（同时最多执行 3 个工具调用）
• 基于角色的工具访问控制（不同角色看到不同的工具列表）
• 工具调用结果的自动验证（检查输出是否符合预期格式）

思考题 5（战略思考）：

如果你是一家 500 人规模的科技公司的 CTO，你的团队正在使用 GPT-4 API 构建内部工具。请制定一个从 L2（Context 级）升级到 L4（企业 Harness 级）的 12 个月路线图，包括：

• 每个阶段的目标和里程碑
• 所需的人力资源
• 预期的 ROI
• 关键风险和缓解措施

1.8.4 延伸阅读

文献/资源	作者/来源	关注点
Harness Engineering for Language Agents	He et al. (2026)	CAR Trinity 理论基础
Chain-of-Thought Prompting	Wei et al. (2022)	Prompt Engineering 经典
Tree of Thoughts	Yao et al. (2023)	推理结构进化
Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks	Lewis et al. (2020)	RAG 原始论文
Self-RAG: Learning to Retrieve, Generate, and Critique	Asai et al. (2023)	RAG 进化
LLMLingua: Compressing Prompts for Accelerated Inference	Jiang et al. (2023)	上下文压缩
Lost in the Middle: How Language Models Use Long Contexts	Liu et al. (2023)	上下文排序
ReAct: Synergizing Reasoning and Acting in LLMs	Yao et al. (2022)	Agent 架构经典
A Survey on Large Language Model based Autonomous Agents	Wang et al. (2023)	Agent 综述
The AI Agent OS: Architectural Patterns	Anthropic (2025)	Agent OS 架构

---

1.9 附录：AI Agent 技术发展年表

1.9.1 2020-2026 关键里程碑

以下年表记录了 AI Agent 工程领域的关键里程碑事件，帮助读者理解技术发展的脉络和加速趋势。

2020 年

时间	事件	意义
2020.06	GPT-3 发布（OpenAI）	175B 参数，首次展示大规模 LLM 的 Few-shot 能力
2020.10	RAG 论文发表（Lewis et al.）	开创检索增强生成范式

2021 年

时间	事件	意义
2021.01	DALL·E 发布（OpenAI）	展示多模态生成能力
2021.08	Codex 发布（OpenAI）	代码生成的突破，GitHub Copilot 的基础

2022 年

时间	事件	意义
2022.01	Chain-of-Thought 论文（Wei et al.）	开创思维链推理范式
2022.03	Chinchilla 论文（DeepMind）	建立 LLM 缩放定律
2022.06	GitHub Copilot 正式发布	AI 辅助编程进入主流
2022.08	LangChain 首次发布	最早的 LLM 应用开发框架
2022.10	ReAct 论文（Yao et al.）	推理与行动协同的 Agent 架构
2022.11.30	ChatGPT 发布	AI Agent 时代的起点

2023 年

时间	事件	意义
2023.02	LLaMA 开源（Meta）	开源 LLM 的里程碑
2023.03	GPT-4 发布	多模态+大幅能力提升
2023.03	AutoGPT 发布	最早的自主 Agent 项目
2023.04	BabyAGI 发布	任务驱动的自主 Agent
2023.05	Tree of Thoughts 论文	树状推理结构
2023.06	Function Calling（OpenAI）	原生工具调用能力
2023.07	Claude 2 发布（Anthropic）	100K 上下文窗口
2023.08	Graph of Thoughts 论文	图状推理结构
2023.09	Llama 2 开源	可商用的开源 LLM
2023.11	GPTs（OpenAI）	Agent 平台化尝试
2023.12	Mixtral 8x7B（Mistral）	MoE 架构的开源实现

2024 年

时间	事件	意义
2024.01	Claude 3 系列发布	多模型层级（Haiku/Sonnet/Opus）
2024.02	Gemini 1.5 Pro（Google）	1M Token 上下文窗口
2024.03	Devin（Cognition）	首个"AI 软件工程师"
2024.04	Claude 3 Opus 登顶多个榜单	长上下文+强推理
2024.05	GPT-4o 发布	原生多模态+实时语音
2024.06	Claude 3.5 Sonnet	性价比标杆
2024.07	Llama 3.1 405B 开源	开源模型追平闭源
2024.09	o1 系列发布（OpenAI）	推理模型新范式
2024.10	Computer Use（Claude）	Agent 操控计算机的能力
2024.12	Operator 发布（OpenAI）	浏览器自动化 Agent

2025 年

时间	事件	意义
2025.01	Claude 3.5/4 系列 Agent 能力	多步骤自主执行
2025.03	Claude Code（Anthropic）	CLI Agent 工具
2025.04	MCP 协议广泛采用	标准化工具互操作
2025.06	Claude 4 Opus/Sonnet	新一代 Agent 能力
2025.08	企业 Agent 平台涌现	Harness 需求爆发
2025.10	多 Agent 框架成熟	协作和编排成为焦点
2025.12	Agent 安全问题凸显	控制和安全成为核心需求

2026 年（截至当前）

时间	事件	意义
2026.01	Claude Agent SDK 发布	企业级 Agent 开发框架
2026.02	Harness Engineering 论文发表	CAR Trinity 理论提出
2026.03	企业 Harness 平台开始出现	从框架到平台的转型
2026.05	本书开始撰写	系统化 Harness Engineering 工程实践

1.9.2 重要论文与开源项目时间线

学术论文时间线

2020 ──── RAG (Lewis et al.)
2022 ──── CoT (Wei et al.)
         │ ReAct (Yao et al.)
2023 ──── ToT (Yao et al.)
         │ GoT (Besta et al.)
         │ Self-RAG (Asai et al.)
         │ Lost in the Middle (Liu et al.)
         │ LLMLingua (Jiang et al.)
2024 ──── Agent Survey (Wang et al.)
         │ LLM Agent OS Patterns
2025 ──── Multi-Agent Systems Survey
         │ Agent Safety & Alignment
2026 ──── Harness Engineering (He et al.) ← 本书理论基础

开源项目时间线

2022 ──── LangChain
2023 ──── AutoGPT ──── BabyAGI ──── LlamaIndex
         │ MetaGPT ──── CrewAI
2024 ──── LangGraph ──── Autogen
         │ OpenAI Swarm
2025 ──── Claude Code ──── OpenAI Agents SDK
         │ MCP Protocol
2026 ──── Claude Agent SDK
         │ HarnessForge (本书项目)

关键趋势观察：

从时间线中可以清晰地看到三个阶段的演进：

1. 2022-2023：Prompt/Reasoning 阶段 — 重点在如何更好地引导模型推理（CoT、ToT、GoT、ReAct）
2. 2023-2024：Context/Framework 阶段 — 重点在如何系统化地组织信息和管理 LLM 应用（RAG、LangChain、LlamaIndex）
3. 2025-2026：Harness/Platform 阶段 — 重点在如何构建可控、可靠、可扩展的 Agent 系统（Agent SDK、Harness Engineering、企业平台）

这一演进趋势与我们在本章中论述的三次范式进化完全吻合。

---

本章术语表

术语	英文	定义
Prompt Engineering	Prompt Engineering	通过设计和优化文本指令来引导 LLM 行为的工程实践
Context Engineering	Context Engineering	系统性地构建、管理和优化 LLM 运行时上下文的工程实践
Harness Engineering	Harness Engineering	构建包裹在 LLM 外层的控制、能动性和运行时系统的工程范式
CAR Trinity	CAR Trinity	Control × Agency × Runtime 的三元统一理论
Control（控制维度）	Control	Harness 中约束 Agent 行为空间的组件集合
Agency（能动性维度）	Agency	Harness 中引导 Agent 自主决策的组件集合
Runtime（运行时维度）	Runtime	Harness 中提供 Agent 运行基础设施的组件集合
Agent	Agent	Model + Harness 的完整系统
RAG	Retrieval-Augmented Generation	检索增强生成，从外部知识库检索信息增强 LLM 的上下文
CoT	Chain-of-Thought	思维链，引导 LLM 进行逐步推理的 Prompt 技术
ToT	Tree-of-Thought	思维树，将推理扩展为树状搜索结构的 Prompt 技术
GoT	Graph-of-Thought	思维图，将推理泛化为有向无环图的 Prompt 技术
MCP	Model Context Protocol	模型上下文协议，标准化的工具互操作协议
L1-L5	Maturity Levels	AI Agent 工程的五级成熟度模型
HarnessForge	HarnessForge	本书的核心代码项目，一个企业级 Harness 平台

---

下一章预告：第 2 章将深入展开 CAR Trinity 理论——Control、Agency、Runtime 三个维度的详细定义、相互关系、设计原则和评估方法。我们将建立一个统一的理论框架，为后续的工程实践提供坚实的理论基础。

---

本章完