Agent意图漂移六栈：从隐性失焦到显性纠偏的全栈实测

Intent = {
  Objective,        // 目标：Agent要解决什么问题
  Outcomes,         // 期望结果：成功的样子
  Health_Metrics,   // 健康指标：如何衡量执行质量
  Constraints,      // 约束：不可逾越的边界
  Guardrails,       // 护栏：偏离时的自动保护
  Context,          // 上下文：决策所需的背景知识
  Feedback          // 反馈：人类纠正信号的处理方式
}

from pydantic import BaseModel, Field
from enum import Enum
from typing import Optional

class DriftSeverity(str, Enum):
    NONE = "none"
    LOW = "low"        # 偏差<10%，自动纠正
    MEDIUM = "medium"  # 偏差10-30%，需确认
    HIGH = "high"      # 偏差>30%，暂停执行
    CRITICAL = "critical"  # 违反约束，立即停止

class IntentSpec(BaseModel):
    objective: str = Field(..., description="一句话目标声明")
    success_criteria: list[str] = Field(
        ..., min_length=1, 
        description="可验证的完成条件"
    )
    health_metrics: dict[str, float] = Field(
        default_factory=dict,
        description="指标名 -> 阈值，如 {'relevance': 0.8}"
    )
    hard_constraints: list[str] = Field(
        default_factory=list,
        description="不可违反的约束"
    )
    max_steps: int = Field(default=50, description="最大执行步数")
    drift_threshold: DriftSeverity = Field(
        default=DriftSeverity.MEDIUM,
        description="触发人工介入的漂移等级"
    )
    intent_embedding: Optional[list[float]] = Field(
        default=None,
        description="原始意图的向量表示，用于相似度检测"
    )

# 使用示例
research_intent = IntentSpec(
    objective="调研2026年AI Agent框架市场份额",
    success_criteria=[
        "覆盖至少5个主流框架",
        "包含GitHub Star数和npm下载数",
        "每个框架至少3个用户评价来源"
    ],
    health_metrics={"relevance": 0.85, "freshness": 0.7},
    hard_constraints=["不引用2025年1月前的数据", "不使用未验证的社交媒体数据"],
    max_steps=30,
    drift_threshold=DriftSeverity.MEDIUM,
)

import numpy as np
from dataclasses import dataclass

@dataclass
class IDSConfig:
    semantic_weight: float = 0.5   # 语义相似度权重
    structural_weight: float = 0.3 # 结构偏差权重
    temporal_weight: float = 0.2   # 时间异常权重
    drift_threshold: float = 0.65  # 漂移触发阈值

class IntentDriftDetector:
    def __init__(self, original_intent_embedding, config: IDSConfig):
        self.intent_emb = original_intent_embedding
        self.cfg = config
        self.step_times = []
        self.plan_steps = []

    def compute_ids(self, current_state_embedding, 
                    step_index: int, step_duration: float) -> float:
        # 1. 语义信号：当前状态与原始意图的余弦距离
        semantic_score = 1 - np.dot(
            self.intent_emb, current_state_embedding
        ) / (np.linalg.norm(self.intent_emb) * 
             np.linalg.norm(current_state_embedding) + 1e-8)

        # 2. 结构信号：执行步骤是否偏离预设计划
        structural_score = self._structural_deviation(step_index)

        # 3. 时间信号：当前步骤耗时是否异常
        temporal_score = self._temporal_anomaly(step_duration)

        # 融合为前缀单调分数
        ids = (
            self.cfg.semantic_weight * semantic_score +
            self.cfg.structural_weight * structural_score +
            self.cfg.temporal_weight * temporal_score
        )
        return min(ids, 1.0)  # 截断到[0,1]

    def is_drifting(self, **kwargs) -> bool:
        return self.compute_ids(**kwargs) > self.cfg.drift_threshold

from enum import Enum
from collections import deque

class DriftPattern(str, Enum):
    GRADUAL = "gradual"      # 渐进偏移
    CASCADING = "cascading"  # 级联漂移
    ANCHORING = "anchoring"  # 锚定漂移
    FIRST_PERSON = "first_person"  # 自述漂移
    OSCILLATING = "oscillating"    # 振荡漂移

class DriftClassifier:
    def __init__(self, window_size: int = 10):
        self.ids_history = deque(maxlen=window_size)
        self.topic_history = deque(maxlen=window_size)

    def classify(self, current_ids: float, 
                 current_topic: str) -> DriftPattern:
        self.ids_history.append(current_ids)
        self.topic_history.append(current_topic)

        if len(self.ids_history) < 5:
            return DriftPattern.GRADUAL

        ids_slope = np.polyfit(
            range(len(self.ids_history)), 
            list(self.ids_history), 1
        )[0]
        topic_set = set(self.topic_history)

        # 振荡：IDS反复升降，主题反复切换
        if self._is_oscillating() and len(topic_set) > 3:
            return DriftPattern.OSCILLATING

        # 锚定：IDS突然跳高后维持高位
        if self._is_anchored():
            return DriftPattern.ANCHORING

        # 自述：检测输出中的第一人称叙事
        if self._has_first_person_commitment():
            return DriftPattern.FIRST_PERSON

        # 渐进：IDS缓慢上升
        if ids_slope > 0.01:
            return DriftPattern.GRADUAL

        return DriftPattern.GRADUAL

ASI = w1·TC + w2·RA + w3·BP + w4·IAC
其中：
  TC  = Task Consistency (任务一致性)
  RA  = Role Adherence (角色遵循度)  
  BP  = Behavioral Predictability (行为可预测性)
  IAC = Inter-Agent Coherence (Agent间一致性)
  wi  = 领域自适应权重

Plan（规划）→ Inspect（检查）→ Evolve（演化）→ Verify（验证）
     ↑                                         |
     └─────────────── 反馈修正 ←─────────────────┘

from typing import Protocol

class TrajectoryStep(Protocol):
    action: str
    observation: str
    intent_alignment: float  # 当前步骤与原始意图的对齐度

class PIVOT:
    def __init__(self, intent: IntentSpec, llm):
        self.intent = intent
        self.llm = llm
        self.trajectory: list[TrajectoryStep] = []

    async def plan(self, task: str) -> list[str]:
        """生成初始计划"""
        prompt = f"""
        Objective: {self.intent.objective}
        Constraints: {self.intent.hard_constraints}
        Max steps: {self.intent.max_steps}

        Generate a step-by-step plan. Each step must be 
        directly traceable to the objective.
        """
        return await self.llm.generate_plan(prompt)

    async def inspect(self, step_result: TrajectoryStep) -> dict:
        """检查执行结果是否偏离意图"""
        return {
            "alignment": step_result.intent_alignment,
            "drift_detected": step_result.intent_alignment < 0.7,
            "drift_severity": self._severity(step_result.intent_alignment),
            "correction_needed": step_result.intent_alignment < 0.5,
        }

    async def evolve(self, inspection: dict, remaining_plan: list[str]):
        """根据检查结果演化剩余计划"""
        if inspection["drift_detected"]:
            # 回溯到上一个对齐点，重新规划后续步骤
            aligned_idx = self._find_last_aligned_step()
            return await self.llm.replan(
                self.intent, 
                self.trajectory[:aligned_idx+1],
                len(remaining_plan)
            )
        return remaining_plan

    async def verify(self, final_result) -> dict:
        """验证最终结果是否满足意图规格"""
        checks = {
            criterion: await self._verify_criterion(
                final_result, criterion
            )
            for criterion in self.intent.success_criteria
        }
        return {
            "passed": all(checks.values()),
            "criteria_results": checks,
            "asi_score": self._compute_asi(),
        }

graph TD
    A[任务输入] --> B[意图规格化]
    B --> C[计划生成器]
    C --> D[验证约束注入]
    D --> E[多Agent任务分配]
    E --> F1[Agent 1: 执行+自检]
    E --> F2[Agent 2: 执行+自检]
    E --> F3[Agent 3: 执行+自检]
    F1 --> G[交叉验证层]
    F2 --> G
    F3 --> G
    G --> H{漂移检测}
    H -->|通过| I[结果聚合]
    H -->|检测到漂移| J[回溯重规划]
    J --> C
    I --> K[最终验证]
    K --> L[输出]

初始轨迹池 → 执行 → 评估 → 选择优质轨迹 → 变异 → 新轨迹池
                ↑                                    |
                └──────────── 迭代进化 ←─────────────┘

from dataclasses import dataclass, field

@dataclass
class Trajectory:
    steps: list[dict]
    success: bool
    intent_alignment_scores: list[float]
    token_cost: int

    @property
    def fitness(self) -> float:
        """轨迹适应度 = 成功率 × 对齐度 / Token成本"""
        if not self.success:
            return 0.0
        avg_alignment = sum(self.intent_alignment_scores) / len(
            self.intent_alignment_scores
        )
        return avg_alignment / (self.token_cost / 1000 + 1)

@dataclass
class TrajectoryPool:
    trajectories: list[Trajectory] = field(default_factory=list)
    max_size: int = 100

    def select_top_k(self, k: int = 5) -> list[Trajectory]:
        """进化选择：返回适应度最高的k条轨迹"""
        sorted_trajs = sorted(
            self.trajectories, 
            key=lambda t: t.fitness, 
            reverse=True
        )
        return sorted_trajs[:k]

    def crossover(self, t1: Trajectory, t2: Trajectory) -> Trajectory:
        """轨迹交叉：从两条成功轨迹中组合最优片段"""
        crossover_point = len(t1.steps) // 2
        new_steps = t1.steps[:crossover_point] + t2.steps[crossover_point:]
        return Trajectory(
            steps=new_steps,
            success=True,  # 待验证
            intent_alignment_scores=[],
            token_cost=0,
        )

class SubgoalDrivenExecutor:
    def __init__(self, master_intent: IntentSpec):
        self.master_intent = master_intent
        self.subgoals: list[IntentSpec] = []
        self.completed: list[dict] = []

    def decompose(self) -> list[IntentSpec]:
        """将主意图分解为可独立验证的子目标"""
        # 每个子目标继承主意图的约束和健康指标
        return [
            IntentSpec(
                objective=f"子目标{i+1}: {obj}",
                success_criteria=criteria,
                health_metrics=self.master_intent.health_metrics,
                hard_constraints=self.master_intent.hard_constraints,
                max_steps=self.master_intent.max_steps // len(self.subgoals),
            )
            for i, (obj, criteria) in enumerate(self._plan_subgoals())
        ]

    async def execute_with_correction(self):
        for i, subgoal in enumerate(self.subgoals):
            result = await self._execute_subgoal(subgoal)
            verification = await self._verify_subgoal(result, subgoal)

            if not verification["passed"]:
                # 子目标级重试，不影响已完成部分
                result = await self._retry_subgoal(
                    subgoal, 
                    max_retries=3,
                    context=verification["failure_reason"]
                )
            self.completed.append(result)

        return await self._aggregate_results()

import json
from datetime import datetime

class IntentPersistenceManager:
    def __init__(self, storage_backend):
        self.storage = storage_backend

    async def checkpoint(self, intent: IntentSpec, 
                         agent_state: dict) -> str:
        """在关键节点保存意图+状态快照"""
        checkpoint = {
            "intent": intent.model_dump(),
            "state": {
                "completed_steps": agent_state["completed_steps"],
                "pending_plan": agent_state["pending_plan"],
                "asi_history": agent_state["asi_history"],
                "drift_corrections": agent_state["drift_corrections"],
            },
            "timestamp": datetime.utcnow().isoformat(),
            "asi_at_checkpoint": self._current_asi(agent_state),
        }
        return await self.storage.save(checkpoint)

    async def restore(self, checkpoint_id: str) -> tuple[IntentSpec, dict]:
        """从检查点恢复意图+状态"""
        data = await self.storage.load(checkpoint_id)
        intent = IntentSpec(**data["intent"])

        # 验证恢复的意图是否与原始意图一致
        restoration_fidelity = await self._verify_restoration(intent)
        if restoration_fidelity < 0.8:
            # 降级到摘要级恢复
            intent = await self._summarized_restore(data)

        return intent, data["state"]

graph TB
    subgraph "Layer 1: 意图规格化"
        I1[IntentSpec定义] --> I2[七组件框架]
        I2 --> I3[意图向量编码]
    end

    subgraph "Layer 2: 漂移感知"
        D1[IDS实时计算] --> D2[语义·结构·时间三信号]
        D2 --> D3[漂移模式分类]
    end

    subgraph "Layer 3: 漂移量化"
        Q1[ASI复合度量] --> Q2[四维度评分]
        Q2 --> Q3[成本影响归因]
    end

    subgraph "Layer 4: 轨迹验证"
        V1[PIVOT四阶段循环] --> V2[VERIMAP多Agent验证]
        V2 --> V3[子目标独立校验]
    end

    subgraph "Layer 5: 轨迹校正"
        C1[SE-Agent进化选择] --> C2[子目标级重试]
        C2 --> C3[轨迹池优化]
    end

    subgraph "Layer 6: 意图持久化"
        P1[三层存储策略] --> P2[跨会话恢复]
        P2 --> P3[定期意图校验]
    end

    I3 --> D1
    D3 --> Q1
    Q3 --> V1
    V3 --> C1
    C3 --> P1
    P3 -.->|反馈信号| D1

    style I1 fill:#4a90d9,color:#fff
    style D1 fill:#e74c3c,color:#fff
    style Q1 fill:#f39c12,color:#fff
    style V1 fill:#27ae60,color:#fff
    style C1 fill:#8e44ad,color:#fff
    style P1 fill:#2c3e50,color:#fff