3.2 上下文漂移的检测与修复 | Loop Engineering Agent 行为偏移诊断与重新锚定指南

导读

在 Loop Engineering 的长时间自主循环中，Agent 最隐蔽也最危险的失效模式不是崩溃，而是"温柔的偏航"——上下文漂移。Agent 表面上仍在正常工作，但其行为已悄然偏离原始目标。本节深入分析漂移成因，提供可编程的检测与修复方案，帮你构建具备自我纠偏能力的稳健循环系统。

学习目标

• 解释上下文漂移的定义、三种类型及其在自主循环中的表现形式
• 实现基于目标一致性评分的漂移检测器
• 设计按严重程度分级的重新锚定协议（Re-anchoring Protocol）
• 将漂移检测集成到 discover → plan → execute → verify 的循环框架中

核心概念

什么是上下文漂移？

上下文漂移（Context Drift）是指在 Loop Engineering 的循环执行过程中，Agent 对原始目标、验证标准和执行策略的理解逐渐发生偏移，导致最终产出与预期不一致的现象。与突然崩溃不同，漂移是渐进式的、不易察觉的，因此更具危险性。

漂移的三种类型

漂移成因与检测修复闭环

环境准备

前置知识

• 已完成 3.1 节状态持久化与上下文管理的学习
• 理解 Loop Engineering 四阶段循环
• 具备 Python 基础编程能力

环境要求

• Python 3.10+
• 依赖包：pip install numpy

分步实战

步骤一：构建漂移检测器

漂移检测器的核心思想是：在循环的每个迭代点，将 Agent 当前的行为表现与初始目标进行比对，计算一个"漂移分数"。基于 Jaccard 相似度实现关键词级别的目标一致性检查。

"""context_drift_detector.py — 基于 Goal-Standard-Strategy 三维评分模型"""
import hashlib, re, time
from dataclasses import dataclass, field
from enum import Enum

class DriftSeverity(Enum):
    NONE = "none"; MINOR = "minor"; MODERATE = "moderate"
    SEVERE = "severe"; CRITICAL = "critical"

@dataclass
class GoalSnapshot:
    original_goal: str
    success_criteria: list[str]
    key_constraints: list[str]
    created_at: float = field(default_factory=time.time)
    goal_hash: str = ""
    def __post_init__(self):
        content = self.original_goal + "|".join(self.success_criteria)
        self.goal_hash = hashlib.sha256(content.encode()).hexdigest()[:16]

@dataclass
class DriftReport:
    iteration: int; goal_drift_score: float; standard_drift_score: float
    strategy_drift_score: float; composite_score: float
    severity: DriftSeverity; evidence: list[str] = field(default_factory=list)
    recommendation: str = ""; timestamp: float = field(default_factory=time.time)

STOPWORDS = {'的','了','在','是','有','和','就','不','都','一','上','也',
             '到','说','要','去','会','着','没有','the','a','an','is','are',
             'to','of','and','in','that','it','be','was','were'}

class ContextDriftDetector:
    def __init__(self, goal_snapshot: GoalSnapshot,
                 minor_threshold=0.3, moderate_threshold=0.6, severe_threshold=0.8):
        self.goal_snapshot = goal_snapshot
        self.minor_threshold = minor_threshold
        self.moderate_threshold = moderate_threshold
        self.severe_threshold = severe_threshold
        self.history: list[DriftReport] = []
        self._baseline_kw = self._extract_keywords(goal_snapshot.original_goal)

    def _extract_keywords(self, text: str) -> set[str]:
        words = set(re.findall(r'[\u4e00-\u9fff]+', text))
        words.update(re.findall(r'[a-zA-Z_]+', text.lower()))
        return words - STOPWORDS

    def _overlap_score(self, a: str, b: str) -> float:
        ka, kb = self._extract_keywords(a), self._extract_keywords(b)
        if not ka or not kb: return 0.0
        return len(ka & kb) / len(ka | kb)

    def _check_goal(self, action: str) -> tuple[float, list[str]]:
        evidence = []
        overlap = self._overlap_score(self.goal_snapshot.original_goal, action)
        if overlap < 0.3:
            evidence.append(f"行为与目标重叠度仅{overlap:.0%},可能偏离主目标")
        irrelevant = {'调试','优化性能','重构','美化格式','补充注释','添加日志'}
        if self._extract_keywords(action) & irrelevant:
            evidence.append("检测到非核心任务活动,可能存在范围偏移")
            overlap *= 0.7
        return max(0.0, min(1.0, 1.0 - overlap)), evidence

    def _check_standard(self, output: str, history: list[str]) -> tuple[float, list[str]]:
        evidence = []
        if len(history) >= 3:
            recent_avg = sum(len(o) for o in history[-3:]) / 3
            early_avg = sum(len(o) for o in history[:3]) / 3
            if recent_avg < early_avg * 0.5:
                evidence.append(f"近期产出长度({recent_avg:.0f})相比早期({early_avg:.0f})显著缩短")
        return min(1.0, len(evidence) * 0.25), evidence

    def detect(self, iteration: int, action: str, output: str = "",
               output_history: list[str] = None) -> DriftReport:
        output_history = output_history or []
        all_ev = []
        gd, ge = self._check_goal(action); all_ev.extend(ge)
        sd, se = self._check_standard(output, output_history); all_ev.extend(se)
        strat_drift = 0.0
        if iteration >= 15: strat_drift = min(0.5, 0.2 + (iteration-15)*0.02)
        elif iteration >= 5: strat_drift = 0.1 + (iteration-5)*0.01
        composite = gd*0.5 + sd*0.3 + strat_drift*0.2
        if composite < self.minor_threshold:
            sev, rec = DriftSeverity.NONE, "继续执行"
        elif composite < self.moderate_threshold:
            sev, rec = DriftSeverity.MINOR, "增加监控频率,准备重新锚定"
        elif composite < self.severe_threshold:
            sev, rec = DriftSeverity.MODERATE, "执行重新锚定协议"
        else:
            sev, rec = DriftSeverity.CRITICAL, "立即停止循环,请求人工介入"
        report = DriftReport(iteration=iteration, goal_drift_score=gd,
            standard_drift_score=sd, strategy_drift_score=strat_drift,
            composite_score=composite, severity=sev,
            evidence=all_ev, recommendation=rec)
        self.history.append(report)
        return report

# 演示
if __name__ == "__main__":
    snap = GoalSnapshot(
        original_goal="实现用户登录功能的完整测试套件,包含正常登录、密码错误、账号锁定三种场景",
        success_criteria=["覆盖三种测试场景","使用pytest框架","包含断言验证"],
        key_constraints=["不修改源代码","仅编写测试"])
    det = ContextDriftDetector(snap)
    actions = [
        ("编写用户登录正常流程测试用例","test_login_normal.py 已创建"),
        ("编写密码错误场景测试用例","test_login_wrong_password.py 已创建"),
        ("优化测试框架配置文件","pytest.ini 已优化,添加彩色输出"),
        ("重构现有测试代码命名规范","测试文件统一重命名为test_前缀"),
        ("补充日志输出功能到测试框架","添加了详细的日志中间件")]
    outs = []
    for i,(a,o) in enumerate(actions,1):
        r = det.detect(i, a, o, outs); outs.append(o)
        print(f"迭代{i}: 漂移={r.composite_score:.3f} {r.severity.value}")
        for e in r.evidence: print(f"  ⚠ {e}")
​

步骤二：实现重新锚定协议

根据漂移严重程度生成不同力度的修复计划——从简单的目标重载到完整的重置流程。

"""re_anchoring_protocol.py — 将漂移的 Agent 重新对齐到原始目标"""
import json, time
from dataclasses import dataclass, field
from enum import Enum

class AnchoringAction(Enum):
    RELOAD_GOAL = "reload_goal"; COMPRESS_CONTEXT = "compress_context"
    RELOAD_SKILLS = "reload_skills"; RESET_VERIFIER = "reset_verifier"
    ESCALATE_HUMAN = "escalate_human"

@dataclass
class AnchoringStep:
    action: AnchoringAction; description: str; priority: int
    executed: bool = False; result: str | None = None

class ReAnchoringProtocol:
    def __init__(self, goal_file_path: str, skill_dirs: list[str] = None):
        self.goal_file_path = goal_file_path
        self.skill_dirs = skill_dirs or []

    def create_plan(self, drift_score: float, evidence: list[str]) -> list[AnchoringStep]:
        plan = []
        if drift_score < 0.3:
            plan.append(AnchoringStep(AnchoringAction.RELOAD_GOAL,
                "重新加载目标文件,提醒Agent当前任务", 1))
        elif drift_score < 0.6:
            plan.append(AnchoringStep(AnchoringAction.RELOAD_GOAL,
                "重新加载目标文件", 1))
            plan.append(AnchoringStep(AnchoringAction.COMPRESS_CONTEXT,
                "压缩近期上下文,保留关键决策和产出", 2))
        elif drift_score < 0.8:
            plan.extend([
                AnchoringStep(AnchoringAction.RELOAD_GOAL,"重新加载目标和成功标准",1),
                AnchoringStep(AnchoringAction.COMPRESS_CONTEXT,"深度压缩上下文",2),
                AnchoringStep(AnchoringAction.RELOAD_SKILLS,"重新加载相关Skills",3),
                AnchoringStep(AnchoringAction.RESET_VERIFIER,"重置验证器标准",4)])
        else:
            plan.append(AnchoringStep(AnchoringAction.ESCALATE_HUMAN,
                "危急漂移,停止自动执行,升级人工处理", 1))
        return plan

    def execute_plan(self, plan: list[AnchoringStep]) -> dict:
        results = {"total": len(plan), "executed": 0, "failed": 0, "details": []}
        for step in sorted(plan, key=lambda s: s.priority):
            if step.action == AnchoringAction.RELOAD_GOAL:
                try:
                    with open(self.goal_file_path) as f: f.read()
                    step.result = "目标文件已加载"; step.executed = True
                except FileNotFoundError:
                    step.result = "错误:目标文件不存在"
            elif step.action == AnchoringAction.COMPRESS_CONTEXT:
                step.result = "上下文已压缩"; step.executed = True
            elif step.action == AnchoringAction.RELOAD_SKILLS:
                step.result = f"已重新加载{len(self.skill_dirs)}个技能"; step.executed = True
            elif step.action == AnchoringAction.RESET_VERIFIER:
                step.result = "验证器标准已重置"; step.executed = True
            elif step.action == AnchoringAction.ESCALATE_HUMAN:
                step.result = "已发送人工升级请求"; step.executed = True
            results["details"].append({"action": step.action.value, "result": step.result})
            results["executed" if step.executed else "failed"] += 1
        return results
​

步骤三：集成到循环引擎

将漂移检测与修复机制嵌入 discover-plan-execute-verify 主循环，实现漂移感知的自动化引擎。

"""drift_aware_loop.py — 集成漂移感知的循环引擎"""
import json
from context_drift_detector import ContextDriftDetector, GoalSnapshot, DriftSeverity
from re_anchoring_protocol import ReAnchoringProtocol

class DriftAwareLoop:
    def __init__(self, goal: str, criteria: list[str], constraints: list[str],
                 max_iter=20, check_interval=2):
        self.snap = GoalSnapshot(goal, criteria, constraints)
        self.detector = ContextDriftDetector(self.snap)
        self.anchoring = ReAnchoringProtocol("/tmp/current_goal.json")
        self.max_iter = max_iter; self.check_interval = check_interval
        self.iteration = 0; self.outputs = []; self.paused = False

    def run(self, step_fn):
        """step_fn(iteration) -> (action, output)"""
        print(f"🚀 循环启动 | 最大迭代: {self.max_iter}")
        while self.iteration < self.max_iter:
            self.iteration += 1
            action, output = step_fn(self.iteration)
            self.outputs.append(output)
            if self.iteration % self.check_interval == 0:
                report = self.detector.detect(
                    self.iteration, action, output, self.outputs)
                print(f"  迭代{self.iteration}: 漂移={report.composite_score:.3f} "
                      f"[{report.severity.value}]")
                for e in report.evidence: print(f"    ⚠ {e}")
                if report.severity == DriftSeverity.CRITICAL:
                    print("  🛑 触发人工升级"); self.paused = True; break
                if report.composite_score >= self.detector.minor_threshold:
                    plan = self.anchoring.create_plan(report.composite_score, report.evidence)
                    res = self.anchoring.execute_plan(plan)
                    print(f"  🔧 锚定: {res['executed']}/{res['total']}步完成")
        return {"iterations": self.iteration, "paused": self.paused}

# 使用示例: 传入你的 step 函数即可
if __name__ == "__main__":
    def my_step(i):
        actions = [
            ("编写登录正常流程测试","test_login.py已创建"),
            ("编写密码错误测试","test_wrong_pwd.py已创建"),
            ("优化测试框架配置","pytest.ini已优化"),
            ("重构测试命名","文件重命名完成"),
            ("添加日志中间件","日志已添加")]
        idx = min(i-1, len(actions)-1)
        return actions[idx]
    loop = DriftAwareLoop(
        goal="编写用户登录功能完整测试套件",
        criteria=["覆盖三种场景","使用pytest","包含断言"],
        constraints=["不修改源代码"], max_iter=6, check_interval=2)
    result = loop.run(my_step)
    print(json.dumps(result, ensure_ascii=False))
​

完整示例：Agent 从测试编写偏移到代码重构

以下模拟真实场景：Agent 在编写测试时逐渐"顺手"开始重构代码，漂移系统自动检测并纠正。

"""e2e_drift_demo.py — 模拟偏移-检测-修复的完整流程"""
from context_drift_detector import ContextDriftDetector, GoalSnapshot
from re_anchoring_protocol import ReAnchoringProtocol

snap = GoalSnapshot(
    original_goal="为用户注册模块编写单元测试,覆盖邮箱验证、密码强度、重复注册三种场景",
    success_criteria=["覆盖三种场景","使用pytest","断言覆盖边界条件"],
    key_constraints=["仅编写测试","不修改源代码","使用fixture"])
detector = ContextDriftDetector(snap)
anchoring = ReAnchoringProtocol("/tmp/goal_anchor.json")

# Agent行为轨迹: 前期正常,中期偏移到重构
trajectory = [
    ("编写邮箱验证正常流程测试","创建test_email.py,3个测试用例"),
    ("编写密码强度检查测试","创建test_password.py,测试弱密码和强密码"),
    ("发现密码验证逻辑可优化,顺手修改","重构了password_validator.py"),  # 偏移开始
    ("继续优化注册模块代码结构","重构register_service.py,提取公共方法"),  # 偏移加剧
    ("为重构代码添加类型注解","为所有函数添加type hints")]  # 完全偏离

outputs, anchored = [], False
for i,(action,output) in enumerate(trajectory,1):
    r = detector.detect(i, action, output, outputs); outputs.append(output)
    icon = "✅" if r.severity.value=="none" else "⚠️"
    print(f"[{icon}] 迭代{i}: 漂移={r.composite_score:.3f} [{r.severity.value}]")
    for e in r.evidence: print(f"    📋 {e}")
    if r.composite_score >= 0.3 and not anchored:
        anchored = True
        print("🔧 触发重新锚定!")
        plan = anchoring.create_plan(r.composite_score, r.evidence)
        res = anchoring.execute_plan(plan)
        for d in res['details']: print(f"  → {d['action']}: {d['result']}")
​

常见问题FAQ

Q1：上下文漂移和 Agent 幻觉有什么区别？如何区分？

幻觉（Hallucination）是 Agent 在单次推理中生成了不真实的信息，而漂移是多轮循环中逐渐偏离原始目标的渐进过程。幻觉是"内容错误"，漂移是"方向偏移"。产生幻觉的 Agent 可能仍在正确方向上努力，而漂移的 Agent 可能每步都"正确"但已不在通往目标的路径上。检测方法也不同：幻觉需事实核查，漂移需目标一致性检查。

Q2：为什么不能仅靠自动停止条件防止漂移？Agent 偏移但仍在正常工作时怎么办？

这正是漂移的危险所在。迭代上限和 Token 预算只能防无限循环，无法检测"有效的错误工作"。Agent 可能偏离到重构、美化格式等看似合理的任务上，每次验证都能通过。解决方案是增加独立的漂移检测层，用原始目标的锚点持续校准方向。Loop Engineering 的六大原语中，State 和 Skills 构成了锚点系统——漂移时通过重新加载状态文件和 Skills 定义拉回正轨。

Q3：如何评估检测器的误报率？过于敏感会不会频繁打断工作？

关键工程权衡建议：（1）采用渐进式阈值，初期设高触发阈值（0.5），积累数据后再调整；（2）记录每次锚定的实际效果来校准；（3）引入趋势分析——连续3次分数上升比单次突增更可靠。Addy Osmani 强调 /goal 模式中的目标锚定应是轻量级操作，不应给循环增加过多开销。

最佳实践与避坑

✅ 最佳实践

循环启动时立即创建目标快照。 不要等第一次检测才记录目标。第一步就将原始目标、成功标准、关键约束保存到状态文件，这些是后续漂移检测的基准。

使用多维漂移评分。 不要仅靠单一指标。采用目标、标准、策略三个维度加权评分，能更准确捕捉不同类型的偏移。

锚定协议分级执行。 轻微漂移只重新注入目标提示，中度漂移压缩上下文，严重漂移执行完整重置。避免一刀切。

记录漂移历史用于模式分析。 长期积累可以发现哪些任务易漂移、Agent 通常在哪个阶段偏移、哪些修复策略最有效。

⚠️ 避坑指南

不要将漂移检测与验证阶段混淆。 verify 检查"产出是否正确"，漂移检测检查"方向是否正确"，是两个独立维度。

不要忽视上下文窗口截断。 循环后期早期目标可能已被截断，Agent 物理上"看不见"原始目标。状态文件同步（3.1节）是关键解法。

不要过度依赖 Agent 自我报告。 Agent 说"我在按目标执行"和实际行为间可能有差距。检测应基于可观察行为（代码变更、文件操作），而非语言描述。

避免锚定过于频繁。 每次锚定都中断工作流、消耗 Token。建议锚定后设置冷却期。

本节小结

本节分析了 Loop Engineering 中上下文漂移这一隐蔽而危险的失效模式。我们学习了漂移的三种类型（目标漂移、标准漂移、策略漂移）及成因机制，实现了基于多维评分的检测器，设计了按严重程度分级的重新锚定协议，并集成到完整循环引擎中。

核心要点：漂移是渐进式方向偏移；检测应基于可观察行为而非自我描述；锚定协议分级执行；状态文件和 Skills 构成锚点系统。

下一节将学习多代理编排与子代理模式——将单个循环扩展为多代理协作系统，其中漂移检测尤为重要，因为多代理间的协调会引入更复杂的偏移风险。

延伸阅读

1. Peter Steinberger, "Loop Engineering: The New Paradigm for AI-Driven Development" (2026)
2. Addy Osmani, "Claude Code /goal Pattern: Keeping AI Agents on Track" — Google 工程实践
3. Boris Cherny, "Claude Code Sub-agents and Loop Architecture" — Claude Code 设计
4. OpenClaw Documentation, "State Persistence and Skills System"
5. "Concept Drift Detection in Autonomous Systems" (2025) — 学术研究

关键词：上下文漂移、Context Drift、Loop Engineering、Agent偏移检测、重新锚定协议、Re-anchoring Protocol、自主循环、状态持久化、目标一致性、漂移检测器
难度：⭐⭐⭐⭐ 中高级
预计阅读时间：25 分钟