第3章 控制论视角：Harness 的控制科学基础

“一切有目的的行为都需要控制。控制不是限制，而是引导系统趋向目标的艺术。”
—— Norbert Wiener，《控制论》（1948）

---

本章导读

在 CAR Trinity（Control × Agency × Runtime）的理论框架中，Control（控制） 是 Harness 的第一维度，也是整个 Agent 系统的"方向盘"。如果说 Agency 赋予 Agent 自主决策的能力，Runtime 提供了执行的土壤，那么 Control 就是确保 Agent 在正确的轨道上运行、在偏离时能够自我纠正的核心机制。

本章从经典控制论（Cybernetics）出发，系统性地探讨如何将控制科学的原理应用于 AI Agent 的 Harness 设计。我们将看到，从 Wiener 的反馈回路到 Kalman 的状态估计，从 PID 控制器到模型预测控制（MPC），这些在工程领域经过数十年验证的理论，可以为 Agent 行为控制提供严谨的数学基础和工程方法。

学习目标

1. 理解控制论核心概念及其在 Agent 系统中的映射
2. 掌握反馈控制回路（Feedback Loop）的设计与实现
3. 学会在 Agent 中应用 PID 控制器进行行为调节
4. 理解状态机与工作流控制的设计模式
5. 掌握分层控制架构在企业级 Harness 中的应用
6. 能够设计并实现完整的 Agent 控制系统

前置知识

• 基本编程能力（TypeScript / Python）
• 了解 Agent 基本概念（参见第1章）
• 了解 CAR Trinity 框架（参见第2章）

---

3.1 控制论基础与 Agent 控制

3.1.1 从 Wiener 到 Agent：控制论的核心思想

1948年，Norbert Wiener 出版了划时代的著作《控制论：关于在动物和机器中控制和通讯的科学》，奠定了控制论（Cybernetics）的理论基础。Wiener 的核心洞察是：无论是生物体还是机器，有目的的行为都依赖于反馈（Feedback）机制。

控制论的三个核心概念可以直接映射到 Agent 系统：

控制论概念	经典工程含义	Agent 系统映射
反馈（Feedback）	系统输出回传为输入	Agent 执行结果影响后续决策
稳态（Homeostasis）	系统维持在平衡状态附近	Agent 行为约束在安全边界内
目标导向（Teleology）	系统行为趋向预定目标	Agent 输出趋向用户意图

控制论在 Agent 中的核心问题：LLM 生成的输出具有随机性和不可预测性，如何通过控制机制确保 Agent 的行为可靠地趋向目标？

这正是 Harness 控制层要解决的问题。

3.1.2 Agent 控制系统的数学模型

从控制论的角度，一个 Agent 可以被建模为一个离散时间控制系统：

系统状态：  x(k+1) = f(x(k), u(k), w(k))
观测输出：  y(k) = h(x(k), v(k))
控制输入：  u(k) = g(y(k), r(k))

其中：
  x(k) — 第 k 步的系统状态（上下文、工具状态、中间结果）
  u(k) — 控制输入（Prompt 指令、约束参数、工具选择）
  y(k) — 系统输出（Agent 的回复、工具调用结果）
  r(k) — 参考信号（用户意图、目标描述）
  w(k) — 过程噪声（LLM 随机性、环境不确定性）
  v(k) — 观测噪声（评估误差、反馈延迟）
  f    — 状态转移函数（LLM + 工具调用的组合行为）
  h    — 观测函数（输出质量评估）
  g    — 控制律（Harness 的控制策略）

这个模型揭示了一个关键事实：Agent 的控制本质上是在噪声环境下，通过调节控制输入 u(k) 使系统输出 y(k) 跟踪参考信号 r(k)。

3.1.3 开环控制 vs 闭环控制

开环控制（Open-Loop Control）：

用户输入 → [固定 Prompt 模板] → LLM → 输出

在开环控制中，Prompt 是预定义的，不根据 Agent 的实际输出进行调整。这种方式简单但脆弱——一旦 LLM 的输出偏离预期，没有任何纠正机制。

闭环控制（Closed-Loop Control）：

用户输入 → [动态 Prompt] → LLM → 输出 → [评估器] → 反馈 → [控制器] → 调整 Prompt
                ↑                                                         |
                └─────────────────────────────────────────────────────────┘

闭环控制引入了反馈机制：Agent 的输出被评估，评估结果反馈给控制器，控制器据此调整后续的输入或行为。这是 Harness 控制层的核心设计模式。

TypeScript 实现：开环 vs 闭环控制器

// === 开环控制器 ===
class OpenLoopController {
  private promptTemplate: string;

  constructor(template: string) {
    this.promptTemplate = template;
  }

  async control(input: string, llm: LLMClient): Promise<string> {
    const prompt = this.promptTemplate.replace('{input}', input);
    return await llm.generate(prompt);
    // 没有反馈，没有调整，一次性执行
  }
}

// === 闭环控制器 ===
interface ControlSignal {
  input: string;
  reference: string;       // 目标描述
  constraints: string[];   // 行为约束
}

interface FeedbackSignal {
  output: string;
  quality: number;         // 0-1 质量评分
  errors: string[];        // 检测到的问题
  suggestions: string[];   // 改进建议
}

class ClosedLoopController {
  private maxIterations: number;
  private qualityThreshold: number;
  private evaluator: OutputEvaluator;
  private history: Array<{ signal: ControlSignal; feedback: FeedbackSignal }> = [];

  constructor(config: { maxIterations: number; qualityThreshold: number }) {
    this.maxIterations = config.maxIterations;
    this.qualityThreshold = config.qualityThreshold;
    this.evaluator = new OutputEvaluator();
  }

  async control(signal: ControlSignal, llm: LLMClient): Promise<{
    output: string;
    iterations: number;
    finalQuality: number;
  }> {
    let currentPrompt = this.buildPrompt(signal, []);
    let iterations = 0;
    let output = '';
    let quality = 0;

    while (iterations < this.maxIterations) {
      // 1. 执行
      output = await llm.generate(currentPrompt);
      iterations++;

      // 2. 评估（反馈）
      const feedback = await this.evaluator.evaluate(output, signal.reference);
      quality = feedback.quality;

      this.history.push({ signal, feedback });

      // 3. 判断是否达标
      if (quality >= this.qualityThreshold) {
        break;
      }

      // 4. 调整控制输入
      currentPrompt = this.buildPrompt(signal, this.history);
    }

    return { output, iterations, finalQuality: quality };
  }

  private buildPrompt(
    signal: ControlSignal,
    history: Array<{ signal: ControlSignal; feedback: FeedbackSignal }>
  ): string {
    let prompt = `任务：${signal.input}\n目标：${signal.reference}\n`;
    prompt += `约束：\n${signal.constraints.map(c => `- ${c}`).join('\n')}\n`;

    if (history.length > 0) {
      const lastAttempt = history[history.length - 1];
      prompt += `\n上一次输出：\n${lastAttempt.signal.input}\n`;
      prompt += `评估结果（质量: ${lastAttempt.feedback.quality}）：\n`;
      prompt += `问题：${lastAttempt.feedback.errors.join(', ')}\n`;
      prompt += `建议：${lastAttempt.feedback.suggestions.join(', ')}\n`;
      prompt += `请根据以上反馈改进输出。\n`;
    }

    return prompt;
  }
}

// === 输出评估器 ===
class OutputEvaluator {
  async evaluate(output: string, reference: string): Promise<FeedbackSignal> {
    const quality = this.computeQuality(output, reference);
    const errors = this.detectErrors(output);
    const suggestions = this.generateSuggestions(output, reference);

    return { output, quality, errors, suggestions };
  }

  private computeQuality(output: string, reference: string): number {
    // 简化的质量评估：长度、关键词覆盖、格式合规
    let score = 0;

    // 长度评分（合理范围内）
    if (output.length > 100 && output.length < 5000) score += 0.3;
    else if (output.length > 50) score += 0.1;

    // 关键词覆盖率
    const refKeywords = reference.toLowerCase().split(/\s+/).filter(w => w.length > 3);
    const outputLower = output.toLowerCase();
    const covered = refKeywords.filter(kw => outputLower.includes(kw)).length;
    score += 0.4 * (covered / Math.max(refKeywords.length, 1));

    // 格式合规（有段落结构）
    if (output.includes('\n\n')) score += 0.15;
    if (output.includes('#') || output.includes('-')) score += 0.15;

    return Math.min(score, 1.0);
  }

  private detectErrors(output: string): string[] {
    const errors: string[] = [];
    if (output.length < 20) errors.push('输出过短');
    if (/ERROR|undefined|null/i.test(output)) errors.push('包含错误标记');
    return errors;
  }

  private generateSuggestions(output: string, reference: string): string[] {
    const suggestions: string[] = [];
    if (output.length < 200) suggestions.push('请提供更详细的回答');
    if (!output.includes('\n')) suggestions.push('请使用段落分隔提高可读性');
    return suggestions;
  }
}

Python 实现：闭环控制器

from dataclasses import dataclass, field
from typing import List, Optional, Callable, Awaitable
import asyncio

@dataclass
class ControlSignal:
    input: str
    reference: str
    constraints: List[str] = field(default_factory=list)

@dataclass
class FeedbackSignal:
    output: str
    quality: float
    errors: List[str] = field(default_factory=list)
    suggestions: List[str] = field(default_factory=list)

class OutputEvaluator:
    """Agent 输出评估器"""

    async def evaluate(self, output: str, reference: str) -> FeedbackSignal:
        quality = self._compute_quality(output, reference)
        errors = self._detect_errors(output)
        suggestions = self._generate_suggestions(output, reference)
        return FeedbackSignal(
            output=output, quality=quality,
            errors=errors, suggestions=suggestions
        )

    def _compute_quality(self, output: str, reference: str) -> float:
        score = 0.0
        # 长度评分
        if 100 < len(output) < 5000:
            score += 0.3
        elif len(output) > 50:
            score += 0.1

        # 关键词覆盖
        ref_keywords = [w for w in reference.lower().split() if len(w) > 3]
        output_lower = output.lower()
        covered = sum(1 for kw in ref_keywords if kw in output_lower)
        if ref_keywords:
            score += 0.4 * (covered / len(ref_keywords))

        # 格式合规
        if '\n\n' in output:
            score += 0.15
        if '#' in output or '-' in output:
            score += 0.15

        return min(score, 1.0)

    def _detect_errors(self, output: str) -> List[str]:
        errors = []
        if len(output) < 20:
            errors.append('输出过短')
        if any(kw in output.upper() for kw in ['ERROR', 'UNDEFINED', 'NULL']):
            errors.append('包含错误标记')
        return errors

    def _generate_suggestions(self, output: str, reference: str) -> List[str]:
        suggestions = []
        if len(output) < 200:
            suggestions.append('请提供更详细的回答')
        if '\n' not in output:
            suggestions.append('请使用段落分隔提高可读性')
        return suggestions


class ClosedLoopController:
    """闭环控制器：基于反馈迭代优化 Agent 输出"""

    def __init__(self, max_iterations: int = 3, quality_threshold: float = 0.8):
        self.max_iterations = max_iterations
        self.quality_threshold = quality_threshold
        self.evaluator = OutputEvaluator()
        self.history: list = []

    async def control(self, signal: ControlSignal,
                      llm_generate: Callable[[str], Awaitable[str]]) -> dict:
        current_prompt = self._build_prompt(signal, [])
        iterations = 0
        output = ''
        quality = 0.0

        while iterations < self.max_iterations:
            # 1. 执行
            output = await llm_generate(current_prompt)
            iterations += 1

            # 2. 评估（反馈）
            feedback = await self.evaluator.evaluate(output, signal.reference)
            quality = feedback.quality
            self.history.append({'signal': signal, 'feedback': feedback})

            # 3. 判断是否达标
            if quality >= self.quality_threshold:
                break

            # 4. 调整控制输入
            current_prompt = self._build_prompt(signal, self.history)

        return {
            'output': output,
            'iterations': iterations,
            'final_quality': quality
        }

    def _build_prompt(self, signal: ControlSignal, history: list) -> str:
        prompt = f"任务：{signal.input}\n目标：{signal.reference}\n"
        if signal.constraints:
            prompt += "约束：\n" + '\n'.join(f'- {c}' for c in signal.constraints) + '\n'

        if history:
            last = history[-1]
            fb = last['feedback']
            prompt += f"\n上一次输出质量：{fb.quality:.2f}\n"
            if fb.errors:
                prompt += f"问题：{', '.join(fb.errors)}\n"
            if fb.suggestions:
                prompt += f"建议：{', '.join(fb.suggestions)}\n"
            prompt += "请根据以上反馈改进输出。\n"

        return prompt

3.1.4 可控性分析：Agent 系统的可控性条件

在经典控制论中，可控性（Controllability） 是指系统是否能够通过适当的控制输入，在有限时间内从任意初始状态转移到任意目标状态。

对于 Agent 系统，可控性取决于以下因素：

1. Prompt 空间维度：控制输入（Prompt 指令）是否足够丰富以影响 Agent 行为
2. 工具集覆盖度：可用工具是否覆盖了任务所需的全部操作
3. 反馈可观测性：是否能够准确评估 Agent 的输出质量
4. 迭代预算：是否有足够的时间和 Token 预算进行迭代修正

可控性矩阵（简化版）：

因素	不可控	部分可控	完全可控
Prompt 空间	固定单一模板	有限模板组合	动态生成 + 组合
工具集	无工具	少量预定义工具	丰富工具 + 动态注册
反馈	无评估	规则评估	LLM-as-Judge + 人类反馈
迭代预算	单次调用	3-5次迭代	自适应迭代 + 预算优化

---

3.2 反馈控制回路设计

3.2.1 反馈回路的基本结构

反馈控制是 Harness 控制层的核心机制。一个完整的反馈控制回路包含以下组件：

                    ┌─────────────────────────────────────┐
                    │           参考信号 r(k)              │
                    │      (用户意图 / 任务目标)            │
                    └──────────────┬──────────────────────┘
                                   │
                                   ▼
                    ┌──────────────────────────────┐
                    │      误差计算 e(k) = r-y      │
                    └──────────────┬───────────────┘
                                   │
                                   ▼
                    ┌──────────────────────────────┐
                    │       控制器 Controller        │
                    │   (Harness 控制策略引擎)        │
                    └──────────────┬───────────────┘
                                   │ 控制输入 u(k)
                                   ▼
                    ┌──────────────────────────────┐
                    │        执行器 Actuator         │
                    │   (LLM + 工具调用引擎)          │
                    └──────────────┬───────────────┘
                                   │ 系统输出 y(k)
                                   ▼
                    ┌──────────────────────────────┐
                    │        传感器 Sensor           │
                    │  (输出评估 + 质量检测器)         │
                    └──────────────┬───────────────┘
                                   │ 反馈信号
                                   └──────────────────────────→ 误差计算

3.2.2 多层反馈架构

在企业级 Harness 中，反馈回路不是单一的，而是多层嵌套的：

第一层：Token 级反馈

• 控制粒度：单个 Token / 词的选择
• 反馈频率：每次生成
• 典型应用：敏感词过滤、格式约束

第二层：语句级反馈

• 控制粒度：单条回复的质量
• 反馈频率：每次 LLM 调用完成
• 典型应用：回答质量评估、事实性校验

第三层：任务级反馈

• 控制粒度：整个任务的完成度
• 反馈频率：任务里程碑检查点
• 典型应用：任务进度跟踪、目标对齐验证

第四层：会话级反馈

• 控制粒度：用户满意度
• 反馈频率：会话结束 / 定期
• 典型应用：用户反馈收集、长期行为优化

TypeScript 实现：多层反馈控制器

// === 多层反馈控制器 ===
enum FeedbackLevel {
  TOKEN = 'token',
  SENTENCE = 'sentence',
  TASK = 'task',
  SESSION = 'session'
}

interface FeedbackResult {
  level: FeedbackLevel;
  score: number;
  details: string;
  correctiveActions: CorrectiveAction[];
}

interface CorrectiveAction {
  type: 'retry' | 'adjust' | 'escalate' | 'abort';
  target: string;
  params: Record<string, any>;
}

class MultiLayerFeedbackController {
  private tokenFilter: TokenLevelFeedback;
  private sentenceEvaluator: SentenceLevelFeedback;
  private taskTracker: TaskLevelFeedback;
  private sessionMonitor: SessionLevelFeedback;

  constructor() {
    this.tokenFilter = new TokenLevelFeedback();
    this.sentenceEvaluator = new SentenceLevelFeedback();
    this.taskTracker = new TaskLevelFeedback();
    this.sessionMonitor = new SessionLevelFeedback();
  }

  async processOutput(
    output: string,
    context: AgentContext
  ): Promise<FeedbackResult[]> {
    const results: FeedbackResult[] = [];

    // 第一层：Token 级反馈（同步，低延迟）
    const tokenResult = this.tokenFilter.evaluate(output);
    results.push(tokenResult);

    // 如果 Token 级发现严重问题，立即返回
    if (tokenResult.correctiveActions.some(a => a.type === 'abort')) {
      return results;
    }

    // 第二层：语句级反馈（异步，中等延迟）
    const sentenceResult = await this.sentenceEvaluator.evaluate(
      output, context.task
    );
    results.push(sentenceResult);

    // 第三层：任务级反馈（在检查点触发）
    if (context.isCheckpoint) {
      const taskResult = await this.taskTracker.evaluate(
        context.taskProgress, context.taskGoal
      );
      results.push(taskResult);
    }

    // 第四层：会话级反馈（定期或结束时触发）
    if (context.isSessionEnd || context.shouldCheckSession) {
      const sessionResult = await this.sessionMonitor.evaluate(
        context.sessionHistory
      );
      results.push(sessionResult);
    }

    return results;
  }

  // 聚合所有层的纠正动作
  aggregateActions(results: FeedbackResult[]): CorrectiveAction[] {
    const actions: CorrectiveAction[] = [];
    // 优先级：abort > escalate > retry > adjust
    for (const result of results) {
      for (const action of result.correctiveActions) {
        if (action.type === 'abort') return [action];
        actions.push(action);
      }
    }
    return actions;
  }
}

// === Token 级反馈 ===
class TokenLevelFeedback {
  private blockedPatterns: RegExp[] = [
    /\b(hack|exploit|malware)\b/i,
    /DROP\s+TABLE/i,
    /rm\s+-rf\s+\//,
  ];

  evaluate(output: string): FeedbackResult {
    const violations: string[] = [];

    for (const pattern of this.blockedPatterns) {
      if (pattern.test(output)) {
        violations.push(`匹配到危险模式: ${pattern}`);
      }
    }

    return {
      level: FeedbackLevel.TOKEN,
      score: violations.length === 0 ? 1.0 : 0.0,
      details: violations.length === 0
        ? 'Token 级检查通过'
        : `发现 ${violations.length} 个违规`,
      correctiveActions: violations.length > 0
        ? [{ type: 'abort', target: 'output', params: { violations } }]
        : []
    };
  }
}

// === 语句级反馈 ===
class SentenceLevelFeedback {
  async evaluate(output: string, task: TaskDescription): Promise<FeedbackResult> {
    const checks = [
      this.checkRelevance(output, task),
      this.checkCompleteness(output, task),
      this.checkFormat(output, task),
      this.checkFactualConsistency(output)
    ];

    const scores = checks.map(c => c.score);
    const avgScore = scores.reduce((a, b) => a + b, 0) / scores.length;

    const actions: CorrectiveAction[] = [];
    if (avgScore < 0.6) {
      actions.push({
        type: 'retry',
        target: 'llm_call',
        params: {
          additionalInstructions: checks
            .filter(c => c.score < 0.7)
            .map(c => c.feedback)
        }
      });
    }

    return {
      level: FeedbackLevel.SENTENCE,
      score: avgScore,
      details: checks.map(c => `${c.aspect}: ${c.score}`).join(', '),
      correctiveActions: actions
    };
  }

  private checkRelevance(output: string, task: TaskDescription) {
    const keywords = task.keywords || [];
    const covered = keywords.filter(kw =>
      output.toLowerCase().includes(kw.toLowerCase())
    ).length;
    const score = keywords.length > 0 ? covered / keywords.length : 0.8;
    return {
      aspect: '相关性', score,
      feedback: score < 0.7 ? '请更紧密地围绕任务关键词展开' : ''
    };
  }

  private checkCompleteness(output: string, task: TaskDescription) {
    const requiredSections = task.requiredSections || [];
    const covered = requiredSections.filter(s =>
      output.includes(s)
    ).length;
    const score = requiredSections.length > 0
      ? covered / requiredSections.length : 0.8;
    return {
      aspect: '完整性', score,
      feedback: score < 0.7 ? '请确保覆盖所有要求的章节' : ''
    };
  }

  private checkFormat(output: string, task: TaskDescription) {
    let score = 0.8;
    if (task.requiredFormat === 'markdown' && !output.includes('#')) score -= 0.3;
    if (task.maxLength && output.length > task.maxLength) score -= 0.2;
    if (task.minLength && output.length < task.minLength) score -= 0.2;
    return {
      aspect: '格式', score: Math.max(score, 0),
      feedback: score < 0.7 ? '请遵循指定的格式要求' : ''
    };
  }

  private checkFactualConsistency(output: string) {
    // 简化版：检查是否包含明显的矛盾或错误标记
    const errorPatterns = [/\b(?:不确定|可能错误|待验证)\b/i];
    const hasIssues = errorPatterns.some(p => p.test(output));
    return {
      aspect: '事实一致性', score: hasIssues ? 0.5 : 0.9,
      feedback: hasIssues ? '请验证事实性陈述的准确性' : ''
    };
  }
}

// === 任务级反馈 ===
class TaskLevelFeedback {
  async evaluate(progress: TaskProgress, goal: TaskGoal): Promise<FeedbackResult> {
    const completionRate = progress.completedSteps / progress.totalSteps;
    const timeRemaining = goal.deadline - Date.now();
    const estimatedTimeNeeded = progress.remainingSteps * progress.avgStepDuration;

    const isOnTrack = completionRate >= progress.expectedCompletionRate;
    const hasEnoughTime = timeRemaining > estimatedTimeNeeded;

    const score = (isOnTrack ? 0.5 : 0.2) + (hasEnoughTime ? 0.5 : 0.2);

    const actions: CorrectiveAction[] = [];
    if (!isOnTrack) {
      actions.push({
        type: 'adjust',
        target: 'task_strategy',
        params: { suggestion: '考虑简化当前步骤或跳过非关键子任务' }
      });
    }
    if (!hasEnoughTime) {
      actions.push({
        type: 'escalate',
        target: 'human',
        params: { message: '任务可能无法在截止日期前完成，需要人工介入' }
      });
    }

    return {
      level: FeedbackLevel.TASK,
      score,
      details: `完成率: ${(completionRate * 100).toFixed(1)}%, 时间${hasEnoughTime ? '充足' : '紧张'}`,
      correctiveActions: actions
    };
  }
}

// === 会话级反馈 ===
class SessionLevelFeedback {
  async evaluate(history: SessionHistory): Promise<FeedbackResult> {
    const totalTurns = history.turns.length;
    const retryCount = history.turns.filter(t => t.wasRetried).length;
    const userCorrections = history.turns.filter(t => t.userCorrected).length;

    const retryRate = retryCount / Math.max(totalTurns, 1);
    const correctionRate = userCorrections / Math.max(totalTurns, 1);

    const score = 1.0 - (retryRate * 0.4) - (correctionRate * 0.6);

    return {
      level: FeedbackLevel.SESSION,
      score: Math.max(score, 0),
      details: `重试率: ${(retryRate * 100).toFixed(1)}%, 纠正率: ${(correctionRate * 100).toFixed(1)}%`,
      correctiveActions: score < 0.5
        ? [{ type: 'escalate', target: 'system', params: { message: '会话质量持续偏低，建议检查 Harness 配置' } }]
        : []
    };
  }
}

// 辅助类型
interface AgentContext {
  task: TaskDescription;
  isCheckpoint: boolean;
  taskProgress: TaskProgress;
  taskGoal: TaskGoal;
  isSessionEnd: boolean;
  shouldCheckSession: boolean;
  sessionHistory: SessionHistory;
}

interface TaskDescription {
  keywords?: string[];
  requiredSections?: string[];
  requiredFormat?: string;
  maxLength?: number;
  minLength?: number;
}

interface TaskProgress {
  completedSteps: number;
  totalSteps: number;
  remainingSteps: number;
  avgStepDuration: number;
  expectedCompletionRate: number;
}

interface TaskGoal {
  deadline: number;
}

interface SessionHistory {
  turns: Array<{
    wasRetried: boolean;
    userCorrected: boolean;
  }>;
}

3.2.3 正反馈与负反馈

在控制论中，反馈分为两类：

• 负反馈（Negative Feedback）：减少偏差，使系统趋向稳定。在 Agent 中，负反馈表现为"当输出偏离目标时进行纠正"。
• 正反馈（Positive Feedback）：放大偏差，使系统加速远离平衡态。在 Agent 中，正反馈可用于"当输出质量好时，鼓励更多类似行为"。

Agent 中的正反馈应用：

class PositiveFeedbackAmplifier {
  private successPatterns: Map<string, number> = new Map();

  // 当 Agent 输出质量高时，记录成功模式
  recordSuccess(output: string, strategy: string, quality: number): void {
    if (quality > 0.85) {
      const count = this.successPatterns.get(strategy) || 0;
      this.successPatterns.set(strategy, count + 1);
    }
  }

  // 在后续任务中，优先选择成功率高的策略
  selectStrategy(candidates: string[]): string {
    let bestStrategy = candidates[0];
    let bestScore = 0;

    for (const strategy of candidates) {
      const score = this.successPatterns.get(strategy) || 0;
      if (score > bestScore) {
        bestScore = score;
        bestStrategy = strategy;
      }
    }

    return bestStrategy;
  }
}

3.2.4 反馈延迟与稳定性

反馈延迟是控制系统稳定性的关键因素。过长的反馈延迟会导致系统振荡甚至发散。

Agent 系统中的反馈延迟来源：

反馈层	典型延迟	影响
Token 级	<10ms	几乎无影响
语句级	100ms-2s（LLM 评估）	可能导致超时
任务级	分钟级	需要异步处理
会话级	小时级	仅用于长期优化

延迟补偿策略：

class DelayCompensator {
  private predictions: Map<string, PredictedState> = new Map();

  // Smith 预估器：预测反馈到达时的系统状态
  predictState(currentState: AgentState, delay: number): AgentState {
    // 基于历史趋势预测延迟期间的状态变化
    const trend = this.computeTrend(currentState);
    return {
      ...currentState,
      estimatedOutput: currentState.output + trend * delay,
      isPredicted: true
    };
  }

  // 当实际反馈到达时，修正预测
  updatePrediction(stateId: string, actualFeedback: FeedbackSignal): void {
    const predicted = this.predictions.get(stateId);
    if (predicted) {
      const error = actualFeedback.quality - predicted.estimatedQuality;
      // 用预测误差更新趋势模型
      this.adjustTrend(stateId, error);
    }
  }

  private computeTrend(state: AgentState): number {
    // 基于最近 N 个状态点计算线性趋势
    return state.recentQualityTrend || 0;
  }

  private adjustTrend(stateId: string, error: number): void {
    // 自适应调整趋势预测
    const predicted = this.predictions.get(stateId);
    if (predicted) {
      predicted.trendAdjustment += error * 0.1; // 学习率
    }
  }
}

interface AgentState {
  output: string;
  estimatedOutput?: string;
  isPredicted?: boolean;
  recentQualityTrend?: number;
}

interface PredictedState {
  estimatedQuality: number;
  trendAdjustment: number;
}

---

3.3 前馈控制与预测机制

3.3.1 前馈控制的原理

与反馈控制（事后纠正）不同，前馈控制（Feedforward Control） 是一种"事前预防"的控制策略。它通过预测系统可能出现的偏差，在偏差发生之前就采取措施。

在 Agent 系统中，前馈控制的核心思想是：根据任务特征和环境状态，预先调整 Prompt 和约束参数，以预防常见错误。

干扰信号 d(k)
      │
      ▼
[前馈控制器] ──→ 控制输入 u(k) ──→ [执行器] ──→ 输出 y(k)
      │                                    │
      └────────────────────────────────────┘
           (基于干扰预测提前调整)

3.3.2 基于经验的前馈控制

class FeedforwardController {
  // 任务类型 → 常见问题 → 预防策略
  private preventionRules: Map<string, PreventionRule[]> = new Map();

  constructor() {
    this.initializeRules();
  }

  private initializeRules(): void {
    // 代码生成任务的前馈规则
    this.preventionRules.set('code_generation', [
      {
        condition: (ctx) => ctx.language === 'python',
        prevention: '注意：请确保所有导入语句完整，使用类型注解，遵循 PEP 8 规范',
        priority: 1
      },
      {
        condition: (ctx) => ctx.requiresTests,
        prevention: '请在编写代码后同时提供单元测试',
        priority: 2
      }
    ]);

    // 数据分析任务的前馈规则
    this.preventionRules.set('data_analysis', [
      {
        condition: (ctx) => ctx.datasetSize > 1000000,
        prevention: '数据集较大，请使用分批处理和流式计算，避免内存溢出',
        priority: 1
      },
      {
        condition: (ctx) => ctx.hasSensitiveData,
        prevention: '数据包含敏感信息，请确保脱敏处理，不要在输出中暴露个人数据',
        priority: 1
      }
    ]);

    // 文档写作任务的前馈规则
    this.preventionRules.set('documentation', [
      {
        condition: (ctx) => ctx.targetAudience === 'beginner',
        prevention: '目标读者是初学者，请避免使用专业术语，必要时提供解释',
        priority: 1
      },
      {
        condition: (ctx) => ctx.includesCodeExamples,
        prevention: '请确保所有代码示例可运行，包含完整的导入和依赖说明',
        priority: 2
      }
    ]);
  }

  generatePreventiveInstructions(
    taskType: string,
    context: TaskContext
  ): string[] {
    const rules = this.preventionRules.get(taskType) || [];
    const applicable = rules
      .filter(rule => rule.condition(context))
      .sort((a, b) => a.priority - b.priority);

    return applicable.map(rule => rule.prevention);
  }

  // 动态学习新的前馈规则
  learnNewRule(
    taskType: string,
    failurePattern: string,
    prevention: string
  ): void {
    const rules = this.preventionRules.get(taskType) || [];
    rules.push({
      condition: (ctx) => ctx.description?.includes(failurePattern) || false,
      prevention,
      priority: rules.length + 1
    });
    this.preventionRules.set(taskType, rules);
  }
}

interface PreventionRule {
  condition: (ctx: TaskContext) => boolean;
  prevention: string;
  priority: number;
}

interface TaskContext {
  language?: string;
  requiresTests?: boolean;
  datasetSize?: number;
  hasSensitiveData?: boolean;
  targetAudience?: string;
  includesCodeExamples?: boolean;
  description?: string;
}

Python 实现：前馈控制器

from typing import Dict, List, Callable, Any
from dataclasses import dataclass

@dataclass
class PreventionRule:
    condition: Callable[[Dict[str, Any]], bool]
    prevention: str
    priority: int

class FeedforwardController:
    """前馈控制器：基于任务特征预防常见问题"""

    def __init__(self):
        self.prevention_rules: Dict[str, List[PreventionRule]] = {}
        self._initialize_rules()

    def _initialize_rules(self):
        # 代码生成任务
        self.prevention_rules['code_generation'] = [
            PreventionRule(
                condition=lambda ctx: ctx.get('language') == 'python',
                prevention='注意：请确保所有导入语句完整，使用类型注解，遵循 PEP 8',
                priority=1
            ),
            PreventionRule(
                condition=lambda ctx: ctx.get('requires_tests', False),
                prevention='请在编写代码后同时提供单元测试',
                priority=2
            ),
        ]

        # 数据分析任务
        self.prevention_rules['data_analysis'] = [
            PreventionRule(
                condition=lambda ctx: ctx.get('dataset_size', 0) > 1_000_000,
                prevention='数据集较大，请使用分批处理，避免内存溢出',
                priority=1
            ),
            PreventionRule(
                condition=lambda ctx: ctx.get('has_sensitive_data', False),
                prevention='数据包含敏感信息，请确保脱敏处理',
                priority=1
            ),
        ]

    def generate_preventive_instructions(
        self, task_type: str, context: Dict[str, Any]
    ) -> List[str]:
        rules = self.prevention_rules.get(task_type, [])
        applicable = sorted(
            [r for r in rules if r.condition(context)],
            key=lambda r: r.priority
        )
        return [r.prevention for r in applicable]

    def learn_new_rule(self, task_type: str,
                       failure_pattern: str, prevention: str):
        rules = self.prevention_rules.get(task_type, [])
        rules.append(PreventionRule(
            condition=lambda ctx, fp=failure_pattern: fp in ctx.get('description', ''),
            prevention=prevention,
            priority=len(rules) + 1
        ))
        self.prevention_rules[task_type] = rules

3.3.3 反馈 + 前馈的复合控制

在实际的企业级 Harness 中，最有效的控制策略是反馈 + 前馈的复合控制：

class CompositeController {
  private feedforward: FeedforwardController;
  private feedback: MultiLayerFeedbackController;
  private adapter: ControlAdapter;

  constructor() {
    this.feedforward = new FeedforwardController();
    this.feedback = new MultiLayerFeedbackController();
    this.adapter = new ControlAdapter();
  }

  async executeTask(
    task: TaskDescription,
    context: TaskContext,
    llm: LLMClient
  ): Promise<TaskResult> {
    // 阶段1：前馈控制 — 生成预防性指令
    const preventiveInstructions = this.feedforward.generatePreventiveInstructions(
      task.type, context
    );

    // 将预防性指令注入 Prompt
    const enrichedPrompt = this.buildEnrichedPrompt(task, preventiveInstructions);

    // 阶段2：执行 + 反馈控制
    let output = await llm.generate(enrichedPrompt);
    let feedbackResults = await this.feedback.processOutput(output, {
      task,
      isCheckpoint: true,
      taskProgress: { completedSteps: 0, totalSteps: 1, remainingSteps: 0, avgStepDuration: 0, expectedCompletionRate: 1 },
      taskGoal: { deadline: Date.now() + 60000 },
      isSessionEnd: false,
      shouldCheckSession: false,
      sessionHistory: { turns: [] }
    });

    // 阶段3：根据反馈调整
    const actions = this.feedback.aggregateActions(feedbackResults);
    let iterations = 0;
    const maxIterations = 3;

    while (actions.length > 0 && iterations < maxIterations) {
      const retryAction = actions.find(a => a.type === 'retry');
      if (!retryAction) break;

      // 将反馈信息作为额外上下文注入
      const adjustedPrompt = this.buildAdjustedPrompt(
        task, preventiveInstructions, feedbackResults
      );
      output = await llm.generate(adjustedPrompt);
      feedbackResults = await this.feedback.processOutput(output, {
        task,
        isCheckpoint: true,
        taskProgress: { completedSteps: 0, totalSteps: 1, remainingSteps: 0, avgStepDuration: 0, expectedCompletionRate: 1 },
        taskGoal: { deadline: Date.now() + 60000 },
        isSessionEnd: false,
        shouldCheckSession: false,
        sessionHistory: { turns: [] }
      });

      iterations++;
    }

    // 阶段4：学习 — 将本次经验反馈给前馈控制器
    if (iterations > 0) {
      // 说明前馈不够充分，需要学习新的规则
      const failurePatterns = feedbackResults
        .flatMap(r => r.correctiveActions)
        .map(a => a.target);
      for (const pattern of failurePatterns) {
        this.feedforward.learnNewRule(
          task.type,
          pattern,
          `预防性检查: ${pattern}`
        );
      }
    }

    return {
      output,
      iterations: iterations + 1,
      preventiveInstructions,
      feedbackHistory: feedbackResults
    };
  }

  private buildEnrichedPrompt(
    task: TaskDescription,
    preventiveInstructions: string[]
  ): string {
    let prompt = `任务：${JSON.stringify(task)}\n\n`;
    if (preventiveInstructions.length > 0) {
      prompt += `重要注意事项（请严格遵守）：\n`;
      prompt += preventiveInstructions.map(i => `- ${i}`).join('\n');
      prompt += '\n\n';
    }
    return prompt;
  }

  private buildAdjustedPrompt(
    task: TaskDescription,
    preventiveInstructions: string[],
    feedbackResults: FeedbackResult[]
  ): string {
    let prompt = this.buildEnrichedPrompt(task, preventiveInstructions);
    prompt += `\n=== 反馈信息 ===\n`;
    for (const result of feedbackResults) {
      prompt += `[${result.level}] 评分: ${result.score.toFixed(2)} - ${result.details}\n`;
    }
    prompt += `\n请根据以上反馈改进输出。\n`;
    return prompt;
  }
}

interface TaskResult {
  output: string;
  iterations: number;
  preventiveInstructions: string[];
  feedbackHistory: FeedbackResult[];
}

class ControlAdapter {
  adaptParameters(baseParams: any, adjustments: any): any {
    return { ...baseParams, ...adjustments };
  }
}

---

3.4 PID 控制器在 Agent 中的应用

3.4.1 PID 控制器原理

PID 控制器（Proportional-Integral-Derivative Controller）是工业控制中最广泛使用的控制器类型。它通过三个分量来计算控制输出：

u(t) = Kp × e(t) + Ki × ∫e(τ)dτ + Kd × de(t)/dt

其中：
  e(t) = r(t) - y(t)   误差（目标值 - 实际值）
  Kp — 比例系数：响应当前误差
  Ki — 积分系数：消除累积误差（稳态误差）
  Kd — 微分系数：预测误差变化趋势

3.4.2 Agent PID 控制器设计

将 PID 控制器应用于 Agent 行为控制：

• P（比例）：当前输出质量与目标的差距 → 调整 Prompt 的约束强度
• I（积分）：历史累积偏差 → 调整策略方向
• D（微分）：质量变化趋势 → 预测并提前干预

class AgentPIDController {
  private kp: number;  // 比例系数
  private ki: number;  // 积分系数
  private kd: number;  // 微分系数

  private integral: number = 0;
  private previousError: number = 0;
  private history: Array<{ error: number; timestamp: number }> = [];

  // 控制输出限制（防积分饱和）
  private outputMin: number;
  private outputMax: number;
  private integralLimit: number;

  constructor(config: {
    kp: number; ki: number; kd: number;
    outputMin?: number; outputMax?: number;
    integralLimit?: number;
  }) {
    this.kp = config.kp;
    this.ki = config.ki;
    this.kd = config.kd;
    this.outputMin = config.outputMin ?? -1;
    this.outputMax = config.outputMax ?? 1;
    this.integralLimit = config.integralLimit ?? 10;
  }

  /**
   * 计算控制输出
   * @param targetQuality 目标质量（0-1）
   * @param currentQuality 当前质量（0-1）
   * @param dt 时间步长（秒）
   */
  compute(
    targetQuality: number,
    currentQuality: number,
    dt: number = 1
  ): PIDOutput {
    const error = targetQuality - currentQuality;

    // P: 比例分量
    const proportional = this.kp * error;

    // I: 积分分量（带抗积分饱和）
    this.integral += error * dt;
    this.integral = Math.max(
      -this.integralLimit,
      Math.min(this.integralLimit, this.integral)
    );
    const integralComponent = this.ki * this.integral;

    // D: 微分分量
    const derivative = (error - this.previousError) / dt;
    const derivativeComponent = this.kd * derivative;

    // 总控制输出（带限幅）
    const rawOutput = proportional + integralComponent + derivativeComponent;
    const output = Math.max(
      this.outputMin,
      Math.min(this.outputMax, rawOutput)
    );

    this.previousError = error;
    this.history.push({ error, timestamp: Date.now() });

    return {
      output,
      proportional,
      integral: integralComponent,
      derivative: derivativeComponent,
      error
    };
  }

  // 将 PID 输出转换为具体的 Prompt 调整策略
  translateToPromptAdjustment(pidOutput: PIDOutput): PromptAdjustment {
    const adjustment: PromptAdjustment = {
      constraintStrength: 0,
      additionalInstructions: [],
      temperatureDelta: 0
    };

    // 正输出 = 需要提高质量 → 增加约束
    if (pidOutput.output > 0) {
      adjustment.constraintStrength = Math.min(pidOutput.output, 1);
      if (pidOutput.proportional > 0.3) {
        adjustment.additionalInstructions.push(
          '请更加严格地遵循任务要求'
        );
      }
      if (pidOutput.integral > 0.5) {
        adjustment.additionalInstructions.push(
          '之前的回答持续偏离目标，请重新审视任务需求'
        );
      }
      adjustment.temperatureDelta = -pidOutput.output * 0.2;
    }

    // 负输出 = 过度约束 → 放松约束
    if (pidOutput.output < -0.3) {
      adjustment.constraintStrength = pidOutput.output;
      adjustment.temperatureDelta = Math.abs(pidOutput.output) * 0.1;
    }

    return adjustment;
  }

  // 自动调参（Ziegler-Nichols 方法简化版）
  autoTune(responses: Array<{ input: number; output: number }>): void {
    // 找到临界增益 Ku 和临界周期 Tu
    let ku = 0;
    let tu = 1;

    for (let i = 1; i < responses.length; i++) {
      const gain = responses[i].output / (responses[i].input || 0.001);
      ku = Math.max(ku, Math.abs(gain));
    }

    // Ziegler-Nichols PID 调参公式
    this.kp = 0.6 * ku;
    this.ki = 2 * this.kp / tu;
    this.kd = this.kp * tu / 8;
  }

  reset(): void {
    this.integral = 0;
    this.previousError = 0;
    this.history = [];
  }

  getHistory(): Array<{ error: number; timestamp: number }> {
    return [...this.history];
  }
}

interface PIDOutput {
  output: number;
  proportional: number;
  integral: number;
  derivative: number;
  error: number;
}

interface PromptAdjustment {
  constraintStrength: number;
  additionalInstructions: string[];
  temperatureDelta: number;
}

Python 实现：Agent PID 控制器

import time
from dataclasses import dataclass, field
from typing import List, Optional

@dataclass
class PIDOutput:
    output: float
    proportional: float
    integral: float
    derivative: float
    error: float

@dataclass
class PromptAdjustment:
    constraint_strength: float = 0.0
    additional_instructions: List[str] = field(default_factory=list)
    temperature_delta: float = 0.0

class AgentPIDController:
    """Agent 行为 PID 控制器"""

    def __init__(self, kp: float = 1.0, ki: float = 0.1, kd: float = 0.05,
                 output_min: float = -1.0, output_max: float = 1.0,
                 integral_limit: float = 10.0):
        self.kp = kp
        self.ki = ki
        self.kd = kd
        self.output_min = output_min
        self.output_max = output_max
        self.integral_limit = integral_limit

        self._integral = 0.0
        self._previous_error = 0.0
        self._history: list = []

    def compute(self, target_quality: float, current_quality: float,
                dt: float = 1.0) -> PIDOutput:
        error = target_quality - current_quality

        # P: 比例
        proportional = self.kp * error

        # I: 积分（抗饱和）
        self._integral += error * dt
        self._integral = max(-self.integral_limit,
                            min(self.integral_limit, self._integral))
        integral_component = self.ki * self._integral

        # D: 微分
        derivative = (error - self._previous_error) / max(dt, 0.001)
        derivative_component = self.kd * derivative

        # 总输出
        raw_output = proportional + integral_component + derivative_component
        output = max(self.output_min, min(self.output_max, raw_output))

        self._previous_error = error
        self._history.append({'error': error, 'timestamp': time.time()})

        return PIDOutput(
            output=output, proportional=proportional,
            integral=integral_component, derivative=derivative_component,
            error=error
        )

    def translate_to_prompt_adjustment(self, pid_output: PIDOutput) -> PromptAdjustment:
        adj = PromptAdjustment()

        if pid_output.output > 0:
            adj.constraint_strength = min(pid_output.output, 1.0)
            if pid_output.proportional > 0.3:
                adj.additional_instructions.append('请更加严格地遵循任务要求')
            if pid_output.integral > 0.5:
                adj.additional_instructions.append('之前的回答持续偏离目标，请重新审视任务需求')
            adj.temperature_delta = -pid_output.output * 0.2

        if pid_output.output < -0.3:
            adj.constraint_strength = pid_output.output
            adj.temperature_delta = abs(pid_output.output) * 0.1

        return adj

    def auto_tune(self, responses: list) -> None:
        """简化的 Ziegler-Nichols 自动调参"""
        ku = 0.0
        tu = 1.0
        for i in range(1, len(responses)):
            inp = responses[i].get('input', 0.001)
            gain = responses[i].get('output', 0) / max(abs(inp), 0.001)
            ku = max(ku, abs(gain))

        self.kp = 0.6 * ku
        self.ki = 2 * self.kp / tu
        self.kd = self.kp * tu / 8

    def reset(self):
        self._integral = 0.0
        self._previous_error = 0.0
        self._history.clear()

3.4.3 PID 控制器在 Token 预算管理中的应用

PID 控制器在 Harness 中一个非常实用的场景是 Token 预算动态管理：

class TokenBudgetPIDController {
  private pid: AgentPIDController;
  private tokenBudget: number;
  private tokensUsed: number = 0;
  private taskComplexity: number;

  constructor(totalBudget: number, estimatedComplexity: number) {
    this.tokenBudget = totalBudget;
    this.taskComplexity = estimatedComplexity;
    this.pid = new AgentPIDController({
      kp: 0.5,
      ki: 0.1,
      kd: 0.05,
      outputMin: -0.5,
      outputMax: 0.5
    });
  }

  /**
   * 每一步调用后更新预算策略
   */
  updateAfterStep(tokensUsedInStep: number, stepIndex: number, totalSteps: number): TokenBudgetAdvice {
    this.tokensUsed += tokensUsedInStep;

    // 计算理想消耗进度
    const expectedUsage = (this.tokenBudget * stepIndex) / totalSteps;
    const actualUsage = this.tokensUsed;

    // 误差：正数 = 消耗过多，需要节约
    const usageRatio = actualUsage / Math.max(expectedUsage, 1);
    const qualityEstimate = Math.min(1.0, 1.0 / usageRatio);

    const pidOutput = this.pid.compute(1.0, qualityEstimate);

    // 计算剩余预算的分配策略
    const remainingBudget = this.tokenBudget - this.tokensUsed;
    const remainingSteps = totalSteps - stepIndex;
    const suggestedPerStep = remainingBudget / Math.max(remainingSteps, 1);

    // PID 输出调整建议的每步预算
    const adjustedPerStep = suggestedPerStep * (1 - pidOutput.output * 0.3);

    return {
      remainingBudget,
      remainingSteps,
      suggestedTokensPerStep: Math.max(adjustedPerStep, 100),
      isOverBudget: this.tokensUsed > this.tokenBudget * 0.9,
      advice: pidOutput.output > 0.2
        ? 'Token 消耗过快，请精简输出'
        : pidOutput.output < -0.2
        ? 'Token 预算充足，可以适当增加细节'
        : 'Token 使用进度正常'
    };
  }
}

interface TokenBudgetAdvice {
  remainingBudget: number;
  remainingSteps: number;
  suggestedTokensPerStep: number;
  isOverBudget: boolean;
  advice: string;
}

---

3.5 状态机与工作流控制

3.5.1 有限状态机（FSM）在 Agent 中的应用

Agent 的任务执行通常不是简单的"输入→输出"，而是一个多阶段的状态转换过程。有限状态机（Finite State Machine, FSM）是建模这种过程最自然的工具。

// === Agent 任务状态机 ===
enum AgentTaskState {
  INIT = 'init',
  PLANNING = 'planning',
  EXECUTING = 'executing',
  VALIDATING = 'validating',
  RECOVERING = 'recovering',
  COMPLETED = 'completed',
  FAILED = 'failed'
}

interface StateTransition {
  from: AgentTaskState;
  to: AgentTaskState;
  condition: (context: AgentTaskContext) => boolean;
  action?: (context: AgentTaskContext) => Promise<void>;
}

class AgentTaskStateMachine {
  private currentState: AgentTaskState;
  private transitions: StateTransition[];
  private stateHistory: Array<{
    state: AgentTaskState;
    enteredAt: number;
    exitedAt?: number;
  }> = [];

  constructor(initialState: AgentTaskState = AgentTaskState.INIT) {
    this.currentState = initialState;
    this.transitions = this.defineTransitions();
    this.stateHistory.push({
      state: initialState,
      enteredAt: Date.now()
    });
  }

  private defineTransitions(): StateTransition[] {
    return [
      // INIT → PLANNING
      {
        from: AgentTaskState.INIT,
        to: AgentTaskState.PLANNING,
        condition: (ctx) => ctx.task !== null,
        action: async (ctx) => {
          ctx.plan = await this.generatePlan(ctx.task!);
        }
      },
      // PLANNING → EXECUTING
      {
        from: AgentTaskState.PLANNING,
        to: AgentTaskState.EXECUTING,
        condition: (ctx) => ctx.plan !== null && ctx.plan!.steps.length > 0,
        action: async (ctx) => {
          ctx.currentStep = 0;
        }
      },
      // EXECUTING → VALIDATING（当前步骤完成）
      {
        from: AgentTaskState.EXECUTING,
        to: AgentTaskState.VALIDATING,
        condition: (ctx) => ctx.stepResult !== null,
        action: async (ctx) => {
          ctx.validationResult = await this.validateStep(ctx.stepResult!);
        }
      },
      // VALIDATING → EXECUTING（验证通过，下一步）
      {
        from: AgentTaskState.VALIDATING,
        to: AgentTaskState.EXECUTING,
        condition: (ctx) =>
          ctx.validationResult?.passed === true &&
          ctx.currentStep! < ctx.plan!.steps.length - 1,
        action: async (ctx) => {
          ctx.currentStep!++;
          ctx.stepResult = null;
          ctx.validationResult = null;
        }
      },
      // VALIDATING → COMPLETED（最后一步验证通过）
      {
        from: AgentTaskState.VALIDATING,
        to: AgentTaskState.COMPLETED,
        condition: (ctx) =>
          ctx.validationResult?.passed === true &&
          ctx.currentStep! >= ctx.plan!.steps.length - 1,
        action: async (ctx) => {
          ctx.finalResult = this.aggregateResults(ctx);
        }
      },
      // VALIDATING → RECOVERING（验证失败）
      {
        from: AgentTaskState.VALIDATING,
        to: AgentTaskState.RECOVERING,
        condition: (ctx) => ctx.validationResult?.passed === false,
        action: async (ctx) => {
          ctx.recoveryAttempts = (ctx.recoveryAttempts || 0) + 1;
        }
      },
      // RECOVERING → EXECUTING（恢复成功）
      {
        from: AgentTaskState.RECOVERING,
        to: AgentTaskState.EXECUTING,
        condition: (ctx) => (ctx.recoveryAttempts || 0) <= 3,
        action: async (ctx) => {
          ctx.stepResult = null;
          // 调整策略后重试当前步骤
        }
      },
      // RECOVERING → FAILED（恢复失败）
      {
        from: AgentTaskState.RECOVERING,
        to: AgentTaskState.FAILED,
        condition: (ctx) => (ctx.recoveryAttempts || 0) > 3,
        action: async (ctx) => {
          ctx.failureReason = '恢复尝试次数超过上限';
        }
      },
      // EXECUTING → FAILED（执行异常）
      {
        from: AgentTaskState.EXECUTING,
        to: AgentTaskState.FAILED,
        condition: (ctx) => ctx.fatalError !== null,
        action: async (ctx) => {
          ctx.failureReason = ctx.fatalError?.message || '未知错误';
        }
      }
    ];
  }

  /**
   * 尝试推进状态机
   */
  async tick(context: AgentTaskContext): Promise<{
    stateChanged: boolean;
    newState: AgentTaskState;
    previousState: AgentTaskState;
  }> {
    const previousState = this.currentState;

    // 查找可触发的转换
    const applicableTransitions = this.transitions.filter(
      t => t.from === this.currentState && t.condition(context)
    );

    if (applicableTransitions.length === 0) {
      return { stateChanged: false, newState: this.currentState, previousState };
    }

    // 选择第一个匹配的转换
    const transition = applicableTransitions[0];

    // 执行转换动作
    if (transition.action) {
      await transition.action(context);
    }

    // 更新状态
    this.stateHistory[this.stateHistory.length - 1].exitedAt = Date.now();
    this.currentState = transition.to;
    this.stateHistory.push({
      state: transition.to,
      enteredAt: Date.now()
    });

    return {
      stateChanged: true,
      newState: transition.to,
      previousState
    };
  }

  getState(): AgentTaskState {
    return this.currentState;
  }

  getHistory(): Array<{ state: AgentTaskState; enteredAt: number; exitedAt?: number }> {
    return [...this.stateHistory];
  }

  isTerminal(): boolean {
    return this.currentState === AgentTaskState.COMPLETED ||
           this.currentState === AgentTaskState.FAILED;
  }

  private async generatePlan(task: any): Promise<any> {
    return { steps: [{ action: 'execute' }], estimatedSteps: 1 };
  }

  private async validateStep(result: any): Promise<{ passed: boolean }> {
    return { passed: result !== null && result !== undefined };
  }

  private aggregateResults(ctx: AgentTaskContext): any {
    return { success: true, steps: ctx.plan?.steps.length || 0 };
  }
}

interface AgentTaskContext {
  task?: any;
  plan?: { steps: any[]; estimatedSteps?: number } | null;
  currentStep?: number;
  stepResult?: any;
  validationResult?: { passed: boolean } | null;
  recoveryAttempts?: number;
  fatalError?: Error | null;
  failureReason?: string;
  finalResult?: any;
}

3.5.2 层次状态机（HSM）

对于复杂的多层任务，简单的 FSM 不够用，需要层次状态机（Hierarchical State Machine）：

class HierarchicalStateMachine {
  private states: Map<string, StateNode> = new Map();
  private activeStates: Set<string> = new Set();
  private context: Record<string, any> = {};

  addState(node: StateNode): void {
    this.states.set(node.id, node);
  }

  start(initialStateId: string): void {
    this.activeStates.clear();
    this.activateState(initialStateId);
  }

  private activateState(stateId: string): void {
    const state = this.states.get(stateId);
    if (!state) return;

    this.activeStates.add(stateId);

    // 执行进入动作
    if (state.onEnter) {
      state.onEnter(this.context);
    }

    // 如果有子状态机，激活初始子状态
    if (state.children && state.initialChild) {
      this.activateState(state.initialChild);
    }
  }

  private deactivateState(stateId: string): void {
    const state = this.states.get(stateId);
    if (!state) return;

    // 先停用子状态
    if (state.children) {
      for (const childId of state.children) {
        if (this.activeStates.has(childId)) {
          this.deactivateState(childId);
        }
      }
    }

    // 执行退出动作
    if (state.onExit) {
      state.onExit(this.context);
    }

    this.activeStates.delete(stateId);
  }

  tick(): void {
    // 从最深层的活跃状态开始检查转换
    const activeList = [...this.activeStates];
    for (const stateId of activeList.reverse()) {
      const state = this.states.get(stateId);
      if (!state || !state.transitions) continue;

      for (const transition of state.transitions) {
        if (transition.guard && !transition.guard(this.context)) continue;

        // 触发转换
        if (transition.action) {
          transition.action(this.context);
        }

        this.deactivateState(stateId);
        this.activateState(transition.target);
        return; // 每次 tick 只触发一个转换
      }
    }
  }

  getActiveStates(): string[] {
    return [...this.activeStates];
  }
}

interface StateNode {
  id: string;
  children?: string[];
  initialChild?: string;
  onEnter?: (ctx: Record<string, any>) => void;
  onExit?: (ctx: Record<string, any>) => void;
  transitions?: Array<{
    target: string;
    guard?: (ctx: Record<string, any>) => boolean;
    action?: (ctx: Record<string, any>) => void;
  }>;
}

3.5.3 工作流引擎

基于状态机，我们可以构建一个完整的工作流引擎，用于编排复杂的多步 Agent 任务：

// === Agent 工作流引擎 ===
interface WorkflowStep {
  id: string;
  name: string;
  type: 'llm_call' | 'tool_call' | 'human_approval' | 'condition' | 'parallel';
  config: Record<string, any>;
  onSuccess: string;    // 成功时的下一步
  onFailure: string;    // 失败时的下一步
  timeout?: number;     // 超时时间（毫秒）
  retryPolicy?: RetryPolicy;
}

interface RetryPolicy {
  maxRetries: number;
  backoffMs: number;
  backoffMultiplier: number;
}

class WorkflowEngine {
  private steps: Map<string, WorkflowStep> = new Map();
  private currentStepId: string;
  private context: WorkflowContext;
  private stepResults: Map<string, any> = new Map();
  private retryCount: Map<string, number> = new Map();

  constructor(
    steps: WorkflowStep[],
    initialStepId: string,
    initialContext: Record<string, any> = {}
  ) {
    for (const step of steps) {
      this.steps.set(step.id, step);
    }
    this.currentStepId = initialStepId;
    this.context = {
      variables: initialContext,
      startTime: Date.now(),
      stepHistory: []
    };
  }

  async execute(llm: LLMClient, tools: ToolRegistry): Promise<WorkflowResult> {
    const maxTotalSteps = 100;
    let stepCount = 0;

    while (stepCount < maxTotalSteps) {
      const step = this.steps.get(this.currentStepId);
      if (!step) {
        return {
          success: false,
          error: `步骤 ${this.currentStepId} 不存在`,
          stepCount,
          context: this.context
        };
      }

      stepCount++;
      const startTime = Date.now();

      try {
        // 执行步骤
        const result = await this.executeStep(step, llm, tools);
        this.stepResults.set(step.id, result);
        this.context.stepHistory.push({
          stepId: step.id,
          status: 'success',
          duration: Date.now() - startTime,
          result: typeof result === 'string' ? result.substring(0, 200) : JSON.stringify(result).substring(0, 200)
        });

        // 判断是否到达终止步骤
        if (step.onSuccess === 'END') {
          return {
            success: true,
            finalResult: result,
            stepCount,
            context: this.context
          };
        }

        this.currentStepId = step.onSuccess;
      } catch (error: any) {
        const retries = this.retryCount.get(step.id) || 0;
        const maxRetries = step.retryPolicy?.maxRetries || 0;

        if (retries < maxRetries) {
          // 重试
          this.retryCount.set(step.id, retries + 1);
          const backoff = (step.retryPolicy?.backoffMs || 1000) *
            Math.pow(step.retryPolicy?.backoffMultiplier || 2, retries);
          await this.sleep(backoff);
          continue;
        }

        this.context.stepHistory.push({
          stepId: step.id,
          status: 'failure',
          duration: Date.now() - startTime,
          error: error.message
        });

        if (step.onFailure === 'END') {
          return {
            success: false,
            error: error.message,
            stepCount,
            context: this.context
          };
        }

        this.currentStepId = step.onFailure;
      }
    }

    return {
      success: false,
      error: '超过最大步骤数限制',
      stepCount,
      context: this.context
    };
  }

  private async executeStep(
    step: WorkflowStep,
    llm: LLMClient,
    tools: ToolRegistry
  ): Promise<any> {
    switch (step.type) {
      case 'llm_call':
        return await llm.generate(step.config.prompt, {
          temperature: step.config.temperature || 0.7,
          maxTokens: step.config.maxTokens || 4096
        });

      case 'tool_call':
        const tool = tools.get(step.config.toolName);
        if (!tool) throw new Error(`工具 ${step.config.toolName} 不存在`);
        return await tool.execute(step.config.params);

      case 'condition':
        const conditionResult = this.evaluateCondition(step.config.expression);
        return conditionResult ? 'true' : 'false';

      case 'human_approval':
        // 等待人工审批（实际实现中需要异步等待）
        return { approved: true };

      case 'parallel':
        const parallelResults = await Promise.all(
          (step.config.parallelSteps || []).map(
            (subStep: any) => this.executeStep(subStep, llm, tools)
          )
        );
        return parallelResults;

      default:
        throw new Error(`未知步骤类型: ${step.type}`);
    }
  }

  private evaluateCondition(expression: string): boolean {
    // 简单的条件评估（实际应用中需要安全的表达式解析器）
    try {
      const vars = this.context.variables;
      return Boolean(new Function(...Object.keys(vars), `return ${expression}`)(
        ...Object.values(vars)
      ));
    } catch {
      return false;
    }
  }

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

interface WorkflowContext {
  variables: Record<string, any>;
  startTime: number;
  stepHistory: Array<{
    stepId: string;
    status: 'success' | 'failure';
    duration: number;
    result?: string;
    error?: string;
  }>;
}

interface WorkflowResult {
  success: boolean;
  finalResult?: any;
  error?: string;
  stepCount: number;
  context: WorkflowContext;
}

// 工具注册表
class ToolRegistry {
  private tools: Map<string, Tool> = new Map();

  register(name: string, tool: Tool): void {
    this.tools.set(name, tool);
  }

  get(name: string): Tool | undefined {
    return this.tools.get(name);
  }
}

interface Tool {
  name: string;
  description: string;
  execute(params: any): Promise<any>;
}

interface LLMClient {
  generate(prompt: string, options?: any): Promise<string>;
}

Python 实现：工作流引擎

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

@dataclass
class RetryPolicy:
    max_retries: int = 3
    backoff_ms: int = 1000
    backoff_multiplier: float = 2.0

@dataclass
class WorkflowStep:
    id: str
    name: str
    step_type: str  # 'llm_call', 'tool_call', 'condition', 'parallel'
    config: Dict[str, Any]
    on_success: str
    on_failure: str
    timeout_ms: Optional[int] = None
    retry_policy: Optional[RetryPolicy] = None

class WorkflowEngine:
    """Agent 工作流引擎"""

    def __init__(self, steps: List[WorkflowStep], initial_step_id: str,
                 initial_context: Optional[Dict] = None):
        self.steps: Dict[str, WorkflowStep] = {s.id: s for s in steps}
        self.current_step_id = initial_step_id
        self.context: Dict[str, Any] = {
            'variables': initial_context or {},
            'start_time': time.time(),
            'step_history': []
        }
        self.step_results: Dict[str, Any] = {}
        self.retry_count: Dict[str, int] = {}

    async def execute(self, llm_generate: Callable,
                      tools: Dict[str, Callable]) -> Dict[str, Any]:
        max_total_steps = 100
        step_count = 0

        while step_count < max_total_steps:
            step = self.steps.get(self.current_step_id)
            if not step:
                return {
                    'success': False,
                    'error': f'步骤 {self.current_step_id} 不存在',
                    'step_count': step_count,
                    'context': self.context
                }

            step_count += 1
            start_time = time.time()

            try:
                result = await self._execute_step(step, llm_generate, tools)
                self.step_results[step.id] = result
                duration = (time.time() - start_time) * 1000
                self.context['step_history'].append({
                    'step_id': step.id, 'status': 'success',
                    'duration_ms': duration
                })

                if step.on_success == 'END':
                    return {
                        'success': True, 'final_result': result,
                        'step_count': step_count, 'context': self.context
                    }
                self.current_step_id = step.on_success

            except Exception as e:
                retries = self.retry_count.get(step.id, 0)
                max_retries = step.retry_policy.max_retries if step.retry_policy else 0

                if retries < max_retries:
                    self.retry_count[step.id] = retries + 1
                    backoff = (step.retry_policy.backoff_ms if step.retry_policy else 1000) * \
                              (step.retry_policy.backoff_multiplier if step.retry_policy else 2) ** retries
                    await asyncio.sleep(backoff / 1000)
                    continue

                self.context['step_history'].append({
                    'step_id': step.id, 'status': 'failure',
                    'error': str(e)
                })

                if step.on_failure == 'END':
                    return {
                        'success': False, 'error': str(e),
                        'step_count': step_count, 'context': self.context
                    }
                self.current_step_id = step.on_failure

        return {
            'success': False, 'error': '超过最大步骤数限制',
            'step_count': step_count, 'context': self.context
        }

    async def _execute_step(self, step: WorkflowStep,
                            llm_generate: Callable,
                            tools: Dict[str, Callable]) -> Any:
        if step.step_type == 'llm_call':
            return await llm_generate(
                step.config.get('prompt', ''),
                temperature=step.config.get('temperature', 0.7)
            )
        elif step.step_type == 'tool_call':
            tool = tools.get(step.config.get('tool_name', ''))
            if not tool:
                raise ValueError(f"工具不存在: {step.config.get('tool_name')}")
            return await tool(**step.config.get('params', {}))
        elif step.step_type == 'condition':
            return self._evaluate_condition(step.config.get('expression', 'False'))
        elif step.step_type == 'parallel':
            tasks = [
                self._execute_step(s, llm_generate, tools)
                for s in step.config.get('parallel_steps', [])
            ]
            return await asyncio.gather(*tasks)
        else:
            raise ValueError(f"未知步骤类型: {step.step_type}")

    def _evaluate_condition(self, expression: str) -> bool:
        try:
            return bool(eval(expression, {}, self.context['variables']))
        except Exception:
            return False

---

3.6 分层控制架构

3.6.1 三层控制模型

在企业级 Harness 中，控制系统采用三层分层架构，这是从工业自动化领域借鉴的经典模式：

┌───────────────────────────────────────────────────────┐
│                  策略层 (Strategic Layer)               │
│  职责：全局策略制定、长期规划、资源分配                    │
│  频率：低频（分钟/小时级）                               │
│  组件：PolicyEngine, ResourceAllocator, GoalManager     │
├───────────────────────────────────────────────────────┤
│                  战术层 (Tactical Layer)                │
│  职责：任务编排、步骤协调、局部优化                       │
│  频率：中频（秒级）                                     │
│  组件：WorkflowEngine, StepCoordinator, PIDController   │
├───────────────────────────────────────────────────────┤
│                  执行层 (Execution Layer)               │
│  职责：具体执行、实时约束、即时反馈                       │
│  频率：高频（毫秒级）                                    │
│  组件：LLMCaller, ToolExecutor, OutputValidator         │
└───────────────────────────────────────────────────────┘

3.6.2 策略层实现

class StrategicController {
  private policyEngine: PolicyEngine;
  private goalManager: GoalManager;
  private resourceAllocator: ResourceAllocator;

  constructor(config: StrategicConfig) {
    this.policyEngine = new PolicyEngine(config.policies);
    this.goalManager = new GoalManager();
    this.resourceAllocator = new ResourceAllocator(config.resourceLimits);
  }

  /**
   * 接收用户意图，生成任务计划
   */
  async plan(intent: UserIntent): Promise<StrategicPlan> {
    // 1. 解析意图，分解为子目标
    const goals = this.goalManager.decomposeIntent(intent);

    // 2. 评估每个子目标的资源需求
    const resourceEstimates = await Promise.all(
      goals.map(g => this.estimateResources(g))
    );

    // 3. 根据资源限制调整计划
    const feasiblePlan = this.resourceAllocator.allocate(goals, resourceEstimates);

    // 4. 应用全局策略
    const policyConstraints = this.policyEngine.evaluate(feasiblePlan);

    return {
      goals: feasiblePlan,
      constraints: policyConstraints,
      estimatedCost: resourceEstimates.reduce((sum, r) => sum + r.tokens, 0),
      estimatedTime: resourceEstimates.reduce((sum, r) => sum + r.time, 0)
    };
  }

  private async estimateResources(goal: Goal): Promise<ResourceEstimate> {
    return {
      tokens: goal.complexity * 2000,
      time: goal.complexity * 30000,
      toolCalls: goal.requiresTools ? goal.complexity * 2 : 0
    };
  }
}

interface StrategicConfig {
  policies: Policy[];
  resourceLimits: ResourceLimits;
}

interface UserIntent {
  description: string;
  priority: 'low' | 'medium' | 'high' | 'critical';
  deadline?: number;
}

interface Goal {
  id: string;
  description: string;
  complexity: number;
  dependencies: string[];
  requiresTools: boolean;
}

interface StrategicPlan {
  goals: Goal[];
  constraints: PolicyConstraint[];
  estimatedCost: number;
  estimatedTime: number;
}

interface ResourceEstimate {
  tokens: number;
  time: number;
  toolCalls: number;
}

interface ResourceLimits {
  maxTokens: number;
  maxTime: number;
  maxToolCalls: number;
}

interface PolicyConstraint {
  type: string;
  constraint: string;
}

interface Policy {
  name: string;
  condition: (plan: any) => boolean;
  constraint: string;
}

class PolicyEngine {
  private policies: Policy[];

  constructor(policies: Policy[]) {
    this.policies = policies;
  }

  evaluate(plan: any): PolicyConstraint[] {
    return this.policies
      .filter(p => p.condition(plan))
      .map(p => ({ type: p.name, constraint: p.constraint }));
  }
}

class GoalManager {
  decomposeIntent(intent: UserIntent): Goal[] {
    // 简化的意图分解
    return [{
      id: 'main',
      description: intent.description,
      complexity: 3,
      dependencies: [],
      requiresTools: false
    }];
  }
}

class ResourceAllocator {
  private limits: ResourceLimits;

  constructor(limits: ResourceLimits) {
    this.limits = limits;
  }

  allocate(goals: Goal[], estimates: ResourceEstimate[]): Goal[] {
    const totalTokens = estimates.reduce((sum, e) => sum + e.tokens, 0);
    if (totalTokens > this.limits.maxTokens) {
      // 按优先级裁剪
      return goals.slice(0, Math.ceil(goals.length * this.limits.maxTokens / totalTokens));
    }
    return goals;
  }
}

3.6.3 战术层实现

class TacticalController {
  private workflowEngine: WorkflowEngine;
  private pidController: AgentPIDController;
  private stepCoordinator: StepCoordinator;

  constructor(config: TacticalConfig) {
    this.pidController = new AgentPIDController({
      kp: config.kp || 0.5,
      ki: config.ki || 0.1,
      kd: config.kd || 0.05
    });
    this.stepCoordinator = new StepCoordinator();
    this.workflowEngine = null as any; // 在 plan 阶段初始化
  }

  /**
   * 接收策略层的计划，编排为具体工作流
   */
  orchestrate(plan: StrategicPlan): WorkflowStep[] {
    const steps: WorkflowStep[] = [];

    for (const goal of plan.goals) {
      // 为每个目标生成工作流步骤
      const goalSteps = this.stepCoordinator.generateSteps(goal, plan.constraints);
      steps.push(...goalSteps);
    }

    // 添加验证和汇总步骤
    steps.push({
      id: 'validate_all',
      name: '验证所有结果',
      type: 'condition',
      config: { expression: 'all_results_valid' },
      onSuccess: 'summarize',
      onFailure: 'recover'
    });

    steps.push({
      id: 'summarize',
      name: '汇总结果',
      type: 'llm_call',
      config: { prompt: '请汇总以下结果...' },
      onSuccess: 'END',
      onFailure: 'END'
    });

    return steps;
  }

  /**
   * 在步骤执行间进行 PID 调节
   */
  adjustBetweenSteps(
    stepIndex: number,
    stepResult: any,
    targetQuality: number
  ): TacticalAdjustment {
    const currentQuality = this.estimateQuality(stepResult);
    const pidOutput = this.pidController.compute(targetQuality, currentQuality);
    const adjustment = this.pidController.translateToPromptAdjustment(pidOutput);

    return {
      pidOutput,
      promptAdjustment: adjustment,
      recommendation: this.generateRecommendation(pidOutput, stepIndex)
    };
  }

  private estimateQuality(result: any): number {
    if (!result) return 0;
    if (typeof result === 'string') {
      return Math.min(result.length / 1000, 1.0);
    }
    return 0.5;
  }

  private generateRecommendation(pidOutput: PIDOutput, stepIndex: number): string {
    if (pidOutput.output > 0.3) {
      return `步骤 ${stepIndex} 质量偏低，建议在下一步增加验证和细节`;
    }
    if (pidOutput.output < -0.3) {
      return `步骤 ${stepIndex} 质量良好，可以保持当前策略`;
    }
    return `步骤 ${stepIndex} 质量在可接受范围内`;
  }
}

interface TacticalConfig {
  kp?: number;
  ki?: number;
  kd?: number;
}

interface TacticalAdjustment {
  pidOutput: PIDOutput;
  promptAdjustment: PromptAdjustment;
  recommendation: string;
}

class StepCoordinator {
  generateSteps(goal: Goal, constraints: PolicyConstraint[]): WorkflowStep[] {
    return [{
      id: `execute_${goal.id}`,
      name: `执行: ${goal.description}`,
      type: 'llm_call',
      config: {
        prompt: goal.description,
        constraints: constraints.map(c => c.constraint)
      },
      onSuccess: `validate_${goal.id}`,
      onFailure: `recover_${goal.id}`,
      retryPolicy: { maxRetries: 2, backoffMs: 1000, backoffMultiplier: 2 }
    }, {
      id: `validate_${goal.id}`,
      name: `验证: ${goal.description}`,
      type: 'condition',
      config: { expression: `result_${goal.id}_valid` },
      onSuccess: goal.dependencies.length > 0 ? goal.dependencies[0] : 'validate_all',
      onFailure: `recover_${goal.id}`
    }];
  }
}

3.6.4 执行层实现

class ExecutionController {
  private outputValidator: RealtimeOutputValidator;
  private toolExecutor: SafeToolExecutor;
  private tokenTracker: TokenBudgetPIDController;

  constructor(config: ExecutionConfig) {
    this.outputValidator = new RealtimeOutputValidator(config.validationRules);
    this.toolExecutor = new SafeToolExecutor(config.toolSafetyRules);
    this.tokenTracker = new TokenBudgetPIDController(
      config.tokenBudget,
      config.estimatedComplexity
    );
  }

  /**
   * 执行单个 LLM 调用（带实时约束）
   */
  async executeLLMCall(
    prompt: string,
    llm: LLMClient,
    constraints: ExecutionConstraints
  ): Promise<ExecutionResult> {
    // 1. 输入验证
    const inputValidation = this.validateInput(prompt, constraints);
    if (!inputValidation.valid) {
      return { success: false, error: inputValidation.reason };
    }

    // 2. 执行 LLM 调用（带超时控制）
    const startTime = Date.now();
    let output: string;
    try {
      output = await Promise.race([
        llm.generate(prompt, constraints.llmOptions),
        this.timeout(constraints.timeoutMs || 30000)
      ]) as string;
    } catch (error: any) {
      return { success: false, error: `LLM 调用失败: ${error.message}` };
    }

    // 3. 输出验证
    const outputValidation = this.outputValidator.validate(output, constraints);
    if (!outputValidation.valid) {
      return {
        success: false,
        error: `输出验证失败: ${outputValidation.reason}`,
        partialOutput: output
      };
    }

    // 4. Token 预算更新
    const tokensUsed = this.estimateTokens(prompt + output);
    const budgetAdvice = this.tokenTracker.updateAfterStep(
      tokensUsed, constraints.stepIndex, constraints.totalSteps
    );

    return {
      success: true,
      output,
      tokensUsed,
      duration: Date.now() - startTime,
      budgetAdvice
    };
  }

  /**
   * 执行工具调用（带安全检查）
   */
  async executeToolCall(
    toolName: string,
    params: Record<string, any>,
    tools: ToolRegistry
  ): Promise<ExecutionResult> {
    // 安全检查
    const safetyCheck = this.toolExecutor.checkSafety(toolName, params);
    if (!safetyCheck.safe) {
      return { success: false, error: `工具调用不安全: ${safetyCheck.reason}` };
    }

    const tool = tools.get(toolName);
    if (!tool) {
      return { success: false, error: `工具 ${toolName} 不存在` };
    }

    try {
      const result = await tool.execute(params);
      return { success: true, output: result };
    } catch (error: any) {
      return { success: false, error: `工具执行失败: ${error.message}` };
    }
  }

  private validateInput(prompt: string, constraints: ExecutionConstraints): {
    valid: boolean; reason?: string;
  } {
    if (prompt.length > (constraints.maxPromptLength || 100000)) {
      return { valid: false, reason: 'Prompt 超过长度限制' };
    }
    if (constraints.blockedPatterns?.some(p => new RegExp(p).test(prompt))) {
      return { valid: false, reason: 'Prompt 包含被禁止的模式' };
    }
    return { valid: true };
  }

  private estimateTokens(text: string): number {
    return Math.ceil(text.length / 4); // 粗略估计
  }

  private timeout(ms: number): Promise<never> {
    return new Promise((_, reject) =>
      setTimeout(() => reject(new Error('超时')), ms)
    );
  }
}

interface ExecutionConfig {
  validationRules: ValidationRule[];
  toolSafetyRules: SafetyRule[];
  tokenBudget: number;
  estimatedComplexity: number;
}

interface ExecutionConstraints {
  timeoutMs?: number;
  maxPromptLength?: number;
  blockedPatterns?: string[];
  llmOptions?: any;
  stepIndex: number;
  totalSteps: number;
}

interface ExecutionResult {
  success: boolean;
  output?: any;
  error?: string;
  partialOutput?: string;
  tokensUsed?: number;
  duration?: number;
  budgetAdvice?: TokenBudgetAdvice;
}

interface ValidationRule {
  name: string;
  validate: (output: string) => boolean;
  message: string;
}

interface SafetyRule {
  toolPattern: string;
  blockedParams: string[];
  requiresApproval: boolean;
}

class RealtimeOutputValidator {
  private rules: ValidationRule[];

  constructor(rules: ValidationRule[]) {
    this.rules = rules;
  }

  validate(output: string, constraints: ExecutionConstraints): {
    valid: boolean; reason?: string;
  } {
    for (const rule of this.rules) {
      if (!rule.validate(output)) {
        return { valid: false, reason: rule.message };
      }
    }
    return { valid: true };
  }
}

class SafeToolExecutor {
  private safetyRules: SafetyRule[];

  constructor(safetyRules: SafetyRule[]) {
    this.safetyRules = safetyRules;
  }

  checkSafety(toolName: string, params: Record<string, any>): {
    safe: boolean; reason?: string;
  } {
    for (const rule of this.safetyRules) {
      if (new RegExp(rule.toolPattern).test(toolName)) {
        for (const blocked of rule.blockedParams) {
          if (JSON.stringify(params).includes(blocked)) {
            return { safe: false, reason: `参数包含被阻止的值: ${blocked}` };
          }
        }
      }
    }
    return { safe: true };
  }
}

---

3.7 自适应控制系统

3.7.1 模型参考自适应控制（MRAC）

在传统的控制系统中，控制器参数是固定的。但 Agent 系统的特性（LLM 行为）是高度非线性和时变的——同一个 Prompt 在不同时间可能产生截然不同的结果。模型参考自适应控制（Model Reference Adaptive Control, MRAC） 可以动态调整控制器参数以适应系统特性的变化。

class AdaptiveController {
  private referenceModel: ReferenceModel;
  private parameterEstimator: ParameterEstimator;
  private baseController: AgentPIDController;

  constructor() {
    this.referenceModel = new ReferenceModel();
    this.parameterEstimator = new ParameterEstimator();
    this.baseController = new AgentPIDController({
      kp: 1.0, ki: 0.1, kd: 0.05
    });
  }

  /**
   * 每次执行后更新控制器参数
   */
  adapt(input: string, output: string, quality: number): void {
    // 1. 计算参考模型的理想输出
    const idealResponse = this.referenceModel.predict(input);

    // 2. 计算实际系统与参考模型的偏差
    const modelError = quality - idealResponse.expectedQuality;

    // 3. 更新参数估计
    this.parameterEstimator.update(modelError, input, output);

    // 4. 调整控制器参数
    const newParams = this.parameterEstimator.getOptimalParams();
    this.baseController = new AgentPIDController(newParams);
  }

  async control(signal: ControlSignal, llm: LLMClient): Promise<string> {
    // 使用自适应调整后的 PID 控制器
    const pidOutput = this.baseController.compute(1.0, 0.5);
    const adjustment = this.baseController.translateToPromptAdjustment(pidOutput);

    let prompt = signal.input;
    if (adjustment.additionalInstructions.length > 0) {
      prompt += '\n' + adjustment.additionalInstructions.join('\n');
    }

    return await llm.generate(prompt);
  }
}

class ReferenceModel {
  // 基于历史数据建立的参考模型
  private successRates: Map<string, number> = new Map();

  predict(input: string): { expectedQuality: number } {
    // 基于输入特征预测理想质量
    const features = this.extractFeatures(input);
    const expectedQuality = this.computeExpectedQuality(features);
    return { expectedQuality };
  }

  private extractFeatures(input: string): Record<string, number> {
    return {
      length: input.length,
      complexity: (input.match(/\b\w{10,}\b/g) || []).length,
      hasCode: input.includes('```') ? 1 : 0,
      hasStructure: input.includes('#') || input.includes('-') ? 1 : 0
    };
  }

  private computeExpectedQuality(features: Record<string, number>): number {
    let quality = 0.7; // 基线质量
    if (features.hasCode) quality += 0.05;
    if (features.hasStructure) quality += 0.05;
    if (features.length > 500) quality -= 0.05;
    return Math.min(Math.max(quality, 0), 1);
  }
}

class ParameterEstimator {
  private samples: Array<{ error: number; input: string; output: string }> = [];
  private currentParams = { kp: 1.0, ki: 0.1, kd: 0.05 };

  update(error: number, input: string, output: string): void {
    this.samples.push({ error, input, output });

    // 保留最近 100 个样本
    if (this.samples.length > 100) {
      this.samples.shift();
    }

    this.recalibrate();
  }

  private recalibrate(): void {
    if (this.samples.length < 10) return;

    // 基于样本统计调整参数
    const avgError = this.samples.reduce((sum, s) => sum + s.error, 0) / this.samples.length;
    const errorVariance = this.samples.reduce(
      (sum, s) => sum + Math.pow(s.error - avgError, 2), 0
    ) / this.samples.length;

    // 高方差 → 增加微分（预测趋势）
    // 高均值误差 → 增加积分（消除稳态误差）
    this.currentParams = {
      kp: 1.0 + avgError * 0.5,
      ki: 0.1 + Math.abs(avgError) * 0.2,
      kd: 0.05 + errorVariance * 0.3
    };
  }

  getOptimalParams(): { kp: number; ki: number; kd: number } {
    return { ...this.currentParams };
  }
}

3.7.2 增益调度控制

另一种自适应策略是增益调度（Gain Scheduling）：根据不同的操作条件（任务类型、复杂度、阶段），使用不同的控制器参数。

class GainSchedulingController {
  private schedules: Map<string, PIDParams> = new Map();
  private currentSchedule: string = 'default';

  constructor() {
    // 预定义不同场景的 PID 参数
    this.schedules.set('default', { kp: 1.0, ki: 0.1, kd: 0.05 });
    this.schedules.set('code_generation', { kp: 0.8, ki: 0.2, kd: 0.1 });
    this.schedules.set('creative_writing', { kp: 0.5, ki: 0.05, kd: 0.02 });
    this.schedules.set('data_analysis', { kp: 1.2, ki: 0.15, kd: 0.08 });
    this.schedules.set('high_stakes', { kp: 1.5, ki: 0.3, kd: 0.15 });
    this.schedules.set('fast_response', { kp: 0.6, ki: 0.02, kd: 0.01 });
  }

  selectSchedule(context: SchedulingContext): string {
    // 基于上下文选择最佳调度表
    if (context.isHighStakes) return 'high_stakes';
    if (context.requiresFastResponse) return 'fast_response';

    switch (context.taskType) {
      case 'code': return 'code_generation';
      case 'writing': return 'creative_writing';
      case 'analysis': return 'data_analysis';
      default: return 'default';
    }
  }

  getController(context: SchedulingContext): AgentPIDController {
    const scheduleId = this.selectScheduleSchedule(context);
    this.currentSchedule = scheduleId;
    const params = this.schedules.get(scheduleId) || this.schedules.get('default')!;
    return new AgentPIDController(params);
  }

  // 运行时学习新的调度参数
  updateSchedule(scheduleId: string, newParams: PIDParams): void {
    const current = this.schedules.get(scheduleId);
    if (current) {
      // 指数加权移动平均（平滑更新）
      const alpha = 0.3;
      this.schedules.set(scheduleId, {
        kp: current.kp * (1 - alpha) + newParams.kp * alpha,
        ki: current.ki * (1 - alpha) + newParams.ki * alpha,
        kd: current.kd * (1 - alpha) + newParams.kd * alpha
      });
    }
  }
}

interface PIDParams {
  kp: number;
  ki: number;
  kd: number;
}

interface SchedulingContext {
  taskType: string;
  isHighStakes: boolean;
  requiresFastResponse: boolean;
  complexity: number;
}

---

3.8 控制稳定性分析

3.8.1 Agent 系统的稳定性问题

在控制论中，稳定性是系统最重要的属性之一。一个不稳定的控制系统会产生振荡、发散甚至崩溃的行为。

Agent 系统的稳定性问题表现为：

• 振荡：Agent 在多个策略间反复切换，无法收敛
• 发散：输出质量越来越差（例如不断添加不必要的内容）
• 死锁：Agent 在验证-重试循环中无限循环

3.8.2 Lyapunov 稳定性分析

我们可以用 Lyapunov 函数 来分析 Agent 控制系统的稳定性：

class StabilityAnalyzer {
  /**
   * 定义 Lyapunov 函数 V(e)
   * V(e) > 0 for e != 0
   * V(0) = 0
   * dV/dt < 0 (能量递减)
   */
  lyapunovFunction(error: number, errorHistory: number[]): {
    v: number;        // Lyapunov 函数值
    dv: number;       // 变化率
    isStable: boolean;
  } {
    // V(e) = e² (最简单的 Lyapunov 函数)
    const v = error * error;

    // dV/dt ≈ V(k) - V(k-1)
    const prevError = errorHistory.length > 0
      ? errorHistory[errorHistory.length - 1] : 0;
    const prevV = prevError * prevError;
    const dv = v - prevV;

    // 稳定性判据：dV/dt < 0 表示系统趋向稳定
    const isStable = dv <= 0 || Math.abs(error) < 0.1;

    return { v, dv, isStable };
  }

  /**
   * 检测振荡
   */
  detectOscillation(errorHistory: number[], windowSize: number = 10): {
    isOscillating: boolean;
    frequency: number;
    amplitude: number;
  } {
    if (errorHistory.length < windowSize) {
      return { isOscillating: false, frequency: 0, amplitude: 0 };
    }

    const recent = errorHistory.slice(-windowSize);
    const mean = recent.reduce((sum, e) => sum + e, 0) / recent.length;
    const centered = recent.map(e => e - mean);

    // 计算零交叉次数（振荡频率的近似）
    let zeroCrossings = 0;
    for (let i = 1; i < centered.length; i++) {
      if (centered[i] * centered[i - 1] < 0) zeroCrossings++;
    }

    const frequency = zeroCrossings / (2 * windowSize);
    const amplitude = Math.max(...recent.map(Math.abs)) - Math.min(...recent.map(Math.abs));
    const isOscillating = frequency > 0.3 && amplitude > 0.2;

    return { isOscillating, frequency, amplitude };
  }

  /**
   * 检测发散
   */
  detectDivergence(errorHistory: number[]): {
    isDiverging: boolean;
    trend: number;
  } {
    if (errorHistory.length < 5) {
      return { isDiverging: false, trend: 0 };
    }

    // 线性回归斜率
    const n = errorHistory.length;
    const xMean = (n - 1) / 2;
    const yMean = errorHistory.reduce((sum, e) => sum + e, 0) / n;

    let numerator = 0;
    let denominator = 0;
    for (let i = 0; i < n; i++) {
      numerator += (i - xMean) * (Math.abs(errorHistory[i]) - Math.abs(yMean));
      denominator += (i - xMean) * (i - xMean);
    }

    const trend = denominator !== 0 ? numerator / denominator : 0;
    const isDiverging = trend > 0.1; // 误差绝对值在增长

    return { isDiverging, trend };
  }
}

3.8.3 稳定性保证机制

class StabilityGuarantor {
  private analyzer: StabilityAnalyzer;
  private maxOscillations: number;
  private maxDivergenceSteps: number;
  private fallbackStrategy: FallbackStrategy;

  constructor(config: StabilityConfig) {
    this.analyzer = new StabilityAnalyzer();
    this.maxOscillations = config.maxOscillations || 3;
    this.maxDivergenceSteps = config.maxDivergenceSteps || 5;
    this.fallbackStrategy = config.fallbackStrategy || FallbackStrategy.SAFE_DEFAULT;
  }

  /**
   * 在每一步执行后检查稳定性
   */
  checkStability(errorHistory: number[]): StabilityReport {
    const recentErrors = errorHistory.slice(-20);

    // Lyapunov 稳定性
    const currentError = recentErrors[recentErrors.length - 1] || 0;
    const lyapunov = this.analyzer.lyapunovFunction(currentError, recentErrors);

    // 振荡检测
    const oscillation = this.analyzer.detectOscillation(recentErrors);

    // 发散检测
    const divergence = this.analyzer.detectDivergence(recentErrors);

    // 综合判断
    let status: StabilityStatus = 'stable';
    let action: StabilityAction = 'continue';

    if (oscillation.isOscillating) {
      status = 'oscillating';
      action = 'dampen';
    }
    if (divergence.isDiverging) {
      status = 'diverging';
      action = 'reset';
    }
    if (!lyapunov.isStable && status === 'stable') {
      status = 'marginal';
      action = 'monitor';
    }

    return {
      status,
      action,
      lyapunovValue: lyapunov.v,
      oscillationFrequency: oscillation.frequency,
      divergenceTrend: divergence.trend,
      recommendation: this.generateRecommendation(status, oscillation, divergence)
    };
  }

  private generateRecommendation(
    status: StabilityStatus,
    oscillation: { isOscillating: boolean; amplitude: number },
    divergence: { isDiverging: boolean; trend: number }
  ): string {
    switch (status) {
      case 'stable':
        return '系统稳定，继续当前控制策略';
      case 'marginal':
        return '系统处于临界稳定状态，建议增加约束或降低迭代步长';
      case 'oscillating':
        return `检测到振荡（振幅: ${oscillation.amplitude.toFixed(2)}），建议增加阻尼（增大 Kd 参数）`;
      case 'diverging':
        return `检测到发散（趋势: ${divergence.trend.toFixed(2)}），建议重置控制器并采用保守策略`;
      default:
        return '未知状态';
    }
  }
}

enum FallbackStrategy {
  SAFE_DEFAULT = 'safe_default',
  HUMAN_ESCALATION = 'human_escalation',
  GRACEFUL_DEGRADATION = 'graceful_degradation'
}

interface StabilityConfig {
  maxOscillations?: number;
  maxDivergenceSteps?: number;
  fallbackStrategy?: FallbackStrategy;
}

type StabilityStatus = 'stable' | 'marginal' | 'oscillating' | 'diverging';
type StabilityAction = 'continue' | 'dampen' | 'reset' | 'monitor';

interface StabilityReport {
  status: StabilityStatus;
  action: StabilityAction;
  lyapunovValue: number;
  oscillationFrequency: number;
  divergenceTrend: number;
  recommendation: string;
}

---

3.9 完整控制系统集成

3.9.1 统一的 Harness 控制引擎

将以上所有组件整合为一个统一的控制引擎：

class HarnessControlEngine {
  // 分层控制器
  private strategic: StrategicController;
  private tactical: TacticalController;
  private execution: ExecutionController;

  // 辅助系统
  private pid: AgentPIDController;
  private feedforward: FeedforwardController;
  private feedback: MultiLayerFeedbackController;
  private stateMachine: AgentTaskStateMachine;
  private stability: StabilityGuarantor;
  private adaptive: AdaptiveController;
  private gainScheduler: GainSchedulingController;

  private errorHistory: number[] = [];
  private config: ControlEngineConfig;

  constructor(config: ControlEngineConfig) {
    this.config = config;

    // 初始化各组件
    this.strategic = new StrategicController({
      policies: config.policies || [],
      resourceLimits: config.resourceLimits || { maxTokens: 100000, maxTime: 300000, maxToolCalls: 50 }
    });

    this.tactical = new TacticalController({
      kp: config.kp, ki: config.ki, kd: config.kd
    });

    this.execution = new ExecutionController({
      validationRules: config.validationRules || [],
      toolSafetyRules: config.toolSafetyRules || [],
      tokenBudget: config.tokenBudget || 100000,
      estimatedComplexity: config.estimatedComplexity || 5
    });

    this.pid = new AgentPIDController({
      kp: config.kp || 1.0,
      ki: config.ki || 0.1,
      kd: config.kd || 0.05
    });

    this.feedforward = new FeedforwardController();
    this.feedback = new MultiLayerFeedbackController();
    this.stateMachine = new AgentTaskStateMachine();
    this.stability = new StabilityGuarantor({
      maxOscillations: 3,
      maxDivergenceSteps: 5,
      fallbackStrategy: FallbackStrategy.GRACEFUL_DEGRADATION
    });
    this.adaptive = new AdaptiveController();
    this.gainScheduler = new GainSchedulingController();
  }

  /**
   * 完整的任务控制流程
   */
  async controlTask(
    intent: UserIntent,
    llm: LLMClient,
    tools: ToolRegistry
  ): Promise<ControlTaskResult> {
    const startTime = Date.now();
    const log: ControlLog[] = [];

    try {
      // ===== 阶段1：战略规划 =====
      log.push({ phase: 'strategic', action: 'plan', timestamp: Date.now() });
      const plan = await this.strategic.plan(intent);
      log.push({
        phase: 'strategic', action: 'plan_complete',
        details: `${plan.goals.length} goals, ${plan.estimatedCost} tokens`,
        timestamp: Date.now()
      });

      // ===== 阶段2：战术编排 =====
      log.push({ phase: 'tactical', action: 'orchestrate', timestamp: Date.now() });
      const workflowSteps = this.tactical.orchestrate(plan);

      // 前馈控制：注入预防性指令
      const preventiveInstructions = this.feedforward.generatePreventiveInstructions(
        intent.description.substring(0, 50), // 用描述作为 taskType
        { description: intent.description }
      );

      // ===== 阶段3：执行 =====
      const workflowEngine = new WorkflowEngine(
        workflowSteps,
        workflowSteps[0]?.id || 'execute_main',
        { instructions: preventiveInstructions }
      );

      log.push({ phase: 'execution', action: 'start_workflow', timestamp: Date.now() });
      const workflowResult = await workflowEngine.execute(llm, tools);

      // ===== 阶段4：反馈与自适应 =====
      const quality = workflowResult.success ? 0.8 : 0.3;
      this.errorHistory.push(1.0 - quality);

      // 稳定性检查
      const stabilityReport = this.stability.checkStability(this.errorHistory);
      log.push({
        phase: 'feedback', action: 'stability_check',
        details: stabilityReport.status,
        timestamp: Date.now()
      });

      // 自适应调整
      if (workflowResult.finalResult) {
        this.adaptive.adapt(
          intent.description,
          typeof workflowResult.finalResult === 'string'
            ? workflowResult.finalResult : JSON.stringify(workflowResult.finalResult),
          quality
        );
      }

      return {
        success: workflowResult.success,
        output: workflowResult.finalResult,
        error: workflowResult.error,
        controlLog: log,
        stabilityReport,
        totalDuration: Date.now() - startTime,
        totalTokens: workflowResult.context?.stepHistory?.length
          ? workflowResult.context.stepHistory.length * 2000
          : 0
      };

    } catch (error: any) {
      return {
        success: false,
        error: error.message,
        controlLog: log,
        stabilityReport: {
          status: 'diverging',
          action: 'reset',
          lyapunovValue: Infinity,
          oscillationFrequency: 0,
          divergenceTrend: 1,
          recommendation: '发生未捕获异常，需要重置系统'
        },
        totalDuration: Date.now() - startTime,
        totalTokens: 0
      };
    }
  }

  getDiagnostics(): ControlDiagnostics {
    return {
      errorHistory: [...this.errorHistory],
      pidHistory: this.pid.getHistory(),
      stabilityStatus: this.stability.checkStability(this.errorHistory),
      config: { ...this.config }
    };
  }
}

interface ControlEngineConfig {
  kp?: number;
  ki?: number;
  kd?: number;
  policies?: Policy[];
  resourceLimits?: ResourceLimits;
  validationRules?: ValidationRule[];
  toolSafetyRules?: SafetyRule[];
  tokenBudget?: number;
  estimatedComplexity?: number;
}

interface ControlLog {
  phase: string;
  action: string;
  details?: string;
  timestamp: number;
}

interface ControlTaskResult {
  success: boolean;
  output?: any;
  error?: string;
  controlLog: ControlLog[];
  stabilityReport: StabilityReport;
  totalDuration: number;
  totalTokens: number;
}

interface ControlDiagnostics {
  errorHistory: number[];
  pidHistory: Array<{ error: number; timestamp: number }>;
  stabilityStatus: StabilityReport;
  config: ControlEngineConfig;
}

Python 实现：统一控制引擎

import time
from dataclasses import dataclass, field
from typing import Dict, List, Any, Optional

@dataclass
class ControlLog:
    phase: str
    action: str
    details: str = ''
    timestamp: float = field(default_factory=time.time)

@dataclass
class ControlTaskResult:
    success: bool
    output: Any = None
    error: Optional[str] = None
    control_log: List[ControlLog] = field(default_factory=list)
    total_duration_ms: float = 0
    total_tokens: int = 0

class HarnessControlEngine:
    """统一 Harness 控制引擎（Python 版）"""

    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.pid = AgentPIDController(
            kp=config.get('kp', 1.0),
            ki=config.get('ki', 0.1),
            kd=config.get('kd', 0.05)
        )
        self.feedforward = FeedforwardController()
        self.error_history: List[float] = []

    async def control_task(self, intent: str,
                           llm_generate: Any,
                           tools: Dict[str, Any]) -> ControlTaskResult:
        start_time = time.time()
        log: List[ControlLog] = []

        try:
            # 阶段1：前馈控制
            log.append(ControlLog(phase='feedforward', action='generate_prevention'))
            preventive = self.feedforward.generate_preventive_instructions(
                'general', {'description': intent}
            )

            # 阶段2：构建增强 Prompt
            enhanced_prompt = intent
            if preventive:
                enhanced_prompt += '\n\n注意事项：\n' + '\n'.join(f'- {p}' for p in preventive)

            # 阶段3：执行
            log.append(ControlLog(phase='execution', action='llm_call'))
            output = await llm_generate(enhanced_prompt)

            # 阶段4：反馈
            quality = min(len(output) / 1000, 1.0) if output else 0.0
            self.error_history.append(1.0 - quality)

            # PID 调节
            pid_output = self.pid.compute(1.0, quality)
            adjustment = self.pid.translate_to_prompt_adjustment(pid_output)

            log.append(ControlLog(
                phase='feedback', action='pid_adjust',
                details=f'quality={quality:.2f}, pid_output={pid_output.output:.2f}'
            ))

            return ControlTaskResult(
                success=True, output=output,
                control_log=log,
                total_duration_ms=(time.time() - start_time) * 1000,
                total_tokens=len(enhanced_prompt + (output or '')) // 4
            )

        except Exception as e:
            return ControlTaskResult(
                success=False, error=str(e),
                control_log=log,
                total_duration_ms=(time.time() - start_time) * 1000
            )

3.9.2 控制引擎使用示例

// === 完整使用示例 ===
async function main() {
  // 1. 创建控制引擎
  const engine = new HarnessControlEngine({
    kp: 0.8,
    ki: 0.15,
    kd: 0.05,
    tokenBudget: 50000,
    estimatedComplexity: 5,
    policies: [
      {
        name: 'no_sensitive_data',
        condition: (plan) => true,
        constraint: '输出中不得包含个人身份信息'
      },
      {
        name: 'max_tool_calls',
        condition: (plan) => plan.estimatedToolCalls > 20,
        constraint: '工具调用次数不超过20次'
      }
    ],
    validationRules: [
      {
        name: 'no_error_markers',
        validate: (output) => !/ERROR|undefined/i.test(output),
        message: '输出包含错误标记'
      }
    ],
    toolSafetyRules: [
      {
        toolPattern: 'file_.*',
        blockedParams: ['/etc/', '/root/', '.ssh'],
        requiresApproval: true
      }
    ]
  });

  // 2. 模拟 LLM 客户端
  const mockLLM: LLMClient = {
    async generate(prompt: string, options?: any): Promise<string> {
      return `这是基于以下请求生成的回答：\n${prompt.substring(0, 100)}...\n\n` +
        `生成的内容包含了相关的分析和结论。`;
    }
  };

  // 3. 模拟工具注册表
  const tools = new ToolRegistry();
  tools.register('search', {
    name: 'search',
    description: '搜索工具',
    async execute(params: any) {
      return [{ title: '结果1', snippet: '...' }];
    }
  });

  // 4. 执行任务
  const result = await engine.controlTask(
    {
      description: '分析2024年AI市场趋势并撰写报告',
      priority: 'high'
    },
    mockLLM,
    tools
  );

  console.log('执行结果:', {
    success: result.success,
    duration: result.totalDuration,
    stability: result.stabilityReport.status,
    recommendation: result.stabilityReport.recommendation
  });

  // 5. 查看诊断信息
  const diagnostics = engine.getDiagnostics();
  console.log('系统诊断:', {
    errorTrend: diagnostics.errorHistory,
    stability: diagnostics.stabilityStatus.status
  });
}

---

3.10 本章小结与最佳实践

3.10.1 核心概念回顾

概念	核心思想	Agent 映射
反馈控制	基于输出偏差进行纠正	评估 Agent 输出，不达标则重试
前馈控制	预测偏差并提前预防	根据任务特征注入预防性指令
PID 控制	P(当前) + I(累积) + D(趋势)	动态调节 Prompt 约束强度
状态机	有限状态的有序转换	多阶段任务编排
分层控制	策略-战术-执行三层分离	企业级 Harness 架构基础
自适应控制	参数随系统特性动态调整	根据 LLM 行为调整控制策略
Lyapunov 稳定性	系统能量递减保证收敛	防止 Agent 行为振荡或发散

3.10.2 控制设计七原则

1. 闭环优于开环：始终引入反馈机制，不要让 Agent “盲飞”
2. 前馈补充反馈：用经验规则预防常见问题，减少反馈迭代的开销
3. 分层解耦：策略层做决策、战术层做编排、执行层做约束
4. 有限迭代：所有反馈循环都必须有最大迭代次数限制
5. 稳定性优先：宁可保守（欠拟合）也不要激进（振荡发散）
6. 自适应进化：控制系统本身也需要从经验中学习和优化
7. 可观测性：控制过程的每一步都应该被记录和可追溯

3.10.3 常见反模式

反模式	问题	正确做法
无反馈循环	输出不可控	引入多层评估 + 重试
无限重试	死循环风险	设置 maxIterations + 退避策略
过度约束	输出僵化	动态调整约束强度（PID）
单一评估器	评估偏差	多评估器集成
忽略延迟	振荡发散	延迟补偿 + 异步评估
固定参数	无法适应变化	自适应控制 + 增益调度

3.10.4 生产环境控制配置模板

# Harness 控制引擎配置模板
control_engine:
  pid:
    kp: 0.8
    ki: 0.15
    kd: 0.05
    output_min: -0.5
    output_max: 0.5
    integral_limit: 10

  feedback:
    max_iterations: 3
    quality_threshold: 0.8
    levels:
      - token     # Token 级（同步）
      - sentence  # 语句级（异步）
      - task      # 任务级（检查点）
      - session   # 会话级（定期）

  feedforward:
    enabled: true
    rules_file: ./prevention_rules.yaml
    learning_rate: 0.1

  stability:
    max_oscillations: 3
    max_divergence_steps: 5
    fallback_strategy: graceful_degradation
    lyapunov_check: true

  adaptive:
    enabled: true
    method: gain_scheduling  # or 'mrac'
    sample_window: 100
    update_frequency: 10

  resource_limits:
    max_tokens: 100000
    max_time_ms: 300000
    max_tool_calls: 50
    max_retry_per_step: 3

---

3.11 延伸阅读

1. Wiener, N. (1948). Cybernetics: Or Control and Communication in the Animal and the Machine
2. Åström, K. J., & Murray, R. M. (2021). Feedback Systems: An Introduction for Scientists and Engineers
3. He, C. et al. (2026). Harness Engineering for Language Agents — Section 3: Control Architecture
4. Franklin, G. F. et al. (2019). Feedback Control of Dynamic Systems

---

下一章预告

在理解了 Control（控制）维度之后，下一章我们将深入 Agency（能动性） 维度——探讨如何赋予 Agent 自主决策、目标导向和灵活应变的能力。我们将从决策空间设计、意图分解、异常自主处理等角度，构建一个具有真正"能动性"的 Agent 系统。