AI Agent 自我监督与自主行为改进：Claude Dreaming 揭示的 Agent 运行时治理新范式

import numpy as np
from collections import deque
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class BehaviorMetric:
    """Agent 行为指标"""
    timestamp: float
    response_length: int
    tool_calls: int
    confidence: float  # 0.0 - 1.0
    safety_flags: int  # 触发的安全规则数量

class SelfSupervisionMonitor:
    """Agent 自我监督监控器"""
    
    def __init__(self, window_size: int = 100, z_threshold: float = 2.0):
        self.window_size = window_size  # 滑动窗口大小
        self.z_threshold = z_threshold  # Z-Score 告警阈值
        self.metrics: deque = deque(maxlen=window_size)
        self.deviations: List[dict] = []
    
    def record(self, metric: BehaviorMetric):
        """记录一次 Agent 交互的行为指标"""
        self.metrics.append(metric)
        if len(self.metrics) >= 30:  # 至少需要 30 个样本
            self._detect_anomalies(metric)
    
    def _detect_anomalies(self, current: BehaviorMetric):
        """检测当前指标是否偏离历史分布"""
        historical = list(self.metrics)[:-1]  # 排除当前值
        
        for field in ['response_length', 'tool_calls', 'confidence']:
            values = [getattr(m, field) for m in historical]
            mean = np.mean(values)
            std = np.std(values)
            
            if std > 0:  # 避免除以零
                z_score = abs((getattr(current, field) - mean) / std)
                
                if z_score > self.z_threshold:
                    deviation = {
                        "field": field,
                        "value": getattr(current, field),
                        "mean": round(mean, 2),
                        "std": round(std, 2),
                        "z_score": round(z_score, 2),
                        "severity": "high" if z_score > 3.0 else "medium"
                    }
                    self.deviations.append(deviation)
                    print(f"⚠️ 偏差检测: {field} Z-Score={z_score:.2f}")
    
    def get_deviation_rate(self) -> float:
        """计算偏差率：偏差次数 / 总交互次数"""
        if not self.metrics:
            return 0.0
        return len(self.deviations) / len(self.metrics)
    
    def get_improvement_suggestions(self) -> List[str]:
        """基于偏差模式生成改进建议"""
        suggestions = []
        if not self.deviations:
            return ["✅ 当前行为模式稳定，无需调整"]
        
        # 统计偏差类型分布
        field_counts = {}
        for d in self.deviations:
            field_counts[d["field"]] = field_counts.get(d["field"], 0) + 1
        
        if field_counts.get("response_length", 0) > 3:
            suggestions.append("📏 响应长度波动较大 → 建议降低 Temperature 或添加长度约束")
        
        if field_counts.get("tool_calls", 0) > 3:
            suggestions.append("🔧 工具调用频率异常 → 检查工具选择逻辑，可能需要添加工具调用前验证")
        
        if field_counts.get("confidence", 0) > 3:
            suggestions.append("🎯 置信度波动异常 → Agent 可能在不确定时过度自信，建议校准置信度输出")
        
        return suggestions

# 使用示例
monitor = SelfSupervisionMonitor(window_size=100, z_threshold=2.0)

# 模拟 Agent 交互
for i in range(50):
    monitor.record(BehaviorMetric(
        timestamp=i * 60,
        response_length=np.random.normal(500, 50),
        tool_calls=np.random.poisson(3),
        confidence=np.random.normal(0.85, 0.05),
        safety_flags=0
    ))

# 模拟偏差出现
for i in range(5):
    monitor.record(BehaviorMetric(
        timestamp=(50 + i) * 60,
        response_length=np.random.normal(800, 100),  # 响应长度突然增加
        tool_calls=np.random.poisson(8),              # 工具调用突然增多
        confidence=np.random.normal(0.95, 0.02),      # 过度自信
        safety_flags=1
    ))

print(f"偏差率: {monitor.get_deviation_rate():.2%}")
print("改进建议:", monitor.get_improvement_suggestions())

from dataclasses import dataclass, field
from typing import List, Dict, Optional
from enum import Enum
import json
import time

class DeviationType(Enum):
    GOAL_DRIFT = "目标漂移"
    TOOL_MISUSE = "工具滥用"
    INCONSISTENCY = "行为不一致"
    SAFETY_VIOLATION = "安全违规"
    CONFIDENCE_MISMATCH = "置信度不匹配"

@dataclass
class InteractionRecord:
    """单次交互记录"""
    id: str
    input_text: str
    output_text: str
    tools_used: List[str]
    confidence: float
    user_feedback: Optional[str]  # None 表示无反馈
    timestamp: float

@dataclass
class DeviationReport:
    """偏差检测报告"""
    type: DeviationType
    severity: float  # 0.0 - 1.0
    description: str
    evidence: str  # 检测依据
    root_cause_hypothesis: str

@dataclass
class ImprovementPlan:
    """改进计划"""
    target_deviation: DeviationType
    strategy: str  # 改进策略描述
    actions: List[str]  # 具体行动项
    expected_outcome: str
    verification_method: str

class DreamingCycle:
    """Dreaming 风格的自主改进循环"""
    
    def __init__(self):
        self.interaction_history: List[InteractionRecord] = []
        self.improvement_plans: List[ImprovementPlan] = []
        self.system_prompt_additions: List[str] = []
        self.cycle_count: int = 0
    
    def add_interaction(self, record: InteractionRecord):
        """添加交互记录"""
        self.interaction_history.append(record)
    
    def run_cycle(self, max_history: int = 50) -> Optional[ImprovementPlan]:
        """执行一次完整的 Dreaming 循环"""
        self.cycle_count += 1
        print(f"\n🌙 Dreaming Cycle #{self.cycle_count} 开始...")
        
        # 阶段 1: 回顾
        recent = self.interaction_history[-max_history:]
        print(f"  📋 回顾 {len(recent)} 次交互")
        patterns = self._extract_patterns(recent)
        
        # 阶段 2: 检测
        deviations = self._detect_deviations(patterns, recent)
        if not deviations:
            print("  ✅ 未检测到显著偏差")
            return None
        
        print(f"  ⚠️ 检测到 {len(deviations)} 项偏差")
        for d in deviations:
            print(f"    - {d.type.value}: {d.description}")
        
        # 阶段 3: 改进
        plan = self._generate_improvement_plan(deviations[0])  # 优先处理最严重的偏差
        self.improvement_plans.append(plan)
        print(f"  📝 生成改进计划: {plan.strategy}")
        
        # 阶段 4: 应用改进
        self._apply_improvement(plan)
        print(f"  ✅ 改进已应用")
        
        return plan
    
    def _extract_patterns(self, interactions: List[InteractionRecord]) -> Dict:
        """从交互历史中提取行为模式"""
        patterns = {
            "avg_tools_per_call": sum(len(r.tools_used) for r in interactions) / max(len(interactions), 1),
            "avg_confidence": sum(r.confidence for r in interactions) / max(len(interactions), 1),
            "tool_frequency": {},
            "feedback_ratio": sum(1 for r in interactions if r.user_feedback) / max(len(interactions), 1)
        }
        
        # 统计工具使用频率
        for r in interactions:
            for tool in r.tools_used:
                patterns["tool_frequency"][tool] = patterns["tool_frequency"].get(tool, 0) + 1
        
        return patterns
    
    def _detect_deviations(self, patterns: Dict, interactions: List[InteractionRecord]) -> List[DeviationReport]:
        """检测行为偏差"""
        deviations = []
        
        # 检测 1: 置信度漂移
        recent_conf = [r.confidence for r in interactions[-10:]]
        older_conf = [r.confidence for r in interactions[-30:-10]]
        if older_conf and recent_conf:
            recent_avg = sum(recent_conf) / len(recent_conf)
            older_avg = sum(older_conf) / len(older_conf)
            if recent_avg - older_avg > 0.15:  # 置信度显著上升
                deviations.append(DeviationReport(
                    type=DeviationType.CONFIDENCE_MISMATCH,
                    severity=0.7,
                    description=f"置信度从 {older_avg:.2f} 升至 {recent_avg:.2f}，可能存在过度自信",
                    evidence="连续 10 次交互的置信度均值显著高于历史水平",
                    root_cause_hypothesis="可能是近期成功交互过多，导致 Agent 高估自身能力"
                ))
        
        # 检测 2: 工具使用异常
        for tool, count in patterns["tool_frequency"].items():
            ratio = count / max(len(interactions), 1)
            if ratio > 0.8:  # 80% 的交互都使用了同一工具
                deviations.append(DeviationReport(
                    type=DeviationType.TOOL_MISUSE,
                    severity=0.6,
                    description=f"工具 {tool} 使用频率过高（{ratio:.0%}），可能存在工具依赖",
                    evidence=f"在 {len(interactions)} 次交互中，{count} 次使用了 {tool}",
                    root_cause_hypothesis="Agent 可能对特定工具产生了过度依赖，忽略了其他合适的工具"
                ))
        
        # 按严重程度排序
        deviations.sort(key=lambda d: d.severity, reverse=True)
        return deviations
    
    def _generate_improvement_plan(self, deviation: DeviationReport) -> ImprovementPlan:
        """基于偏差生成改进计划"""
        if deviation.type == DeviationType.CONFIDENCE_MISMATCH:
            return ImprovementPlan(
                target_deviation=DeviationType.CONFIDENCE_MISMATCH,
                strategy="置信度校准",
                actions=[
                    "在系统提示中添加「在不确定时明确表达不确定性」的指令",
                    "启用自反思推理模式：先生成回答，再评估置信度",
                    "引入置信度上限：当置信度 > 0.95 时自动降为 0.85"
                ],
                expected_outcome="Agent 的置信度分布回归到历史正常范围（0.7-0.9）",
                verification_method="监控后续 20 次交互的置信度均值和方差"
            )
        elif deviation.type == DeviationType.TOOL_MISUSE:
            return ImprovementPlan(
                target_deviation=DeviationType.TOOL_MISUSE,
                strategy="工具使用多样化",
                actions=[
                    "在系统提示中列出所有可用工具及其适用场景",
                    "添加工具选择前检查：「当前场景最适合的工具是什么？」",
                    "记录每次工具选择的理由，便于后续审查"
                ],
                expected_outcome="工具使用分布更加均衡，单一工具使用频率降至 50% 以下",
                verification_method="统计后续 30 次交互的工具使用分布"
            )
        else:
            return ImprovementPlan(
                target_deviation=deviation.type,
                strategy=f"针对{deviation.type.value}的通用改进",
                actions=["回顾偏差案例，在系统提示中追加针对性约束"],
                expected_outcome="偏差率降低",
                verification_method="监控偏差率变化趋势"
    def _apply_improvement(self, plan: ImprovementPlan):
        """应用改进计划到系统提示"""
        addition = f"\n[改进 #{self.cycle_count}] {plan.strategy}：" + "；".join(plan.actions)
        self.system_prompt_additions.append(addition)
        print(f"    系统提示追加: {addition[:80]}...")

# 使用示例
dreaming = DreamingCycle()

# 模拟添加交互
for i in range(40):
    dreaming.add_interaction(InteractionRecord(
        id=f"int-{i}",
        input_text=f"用户查询 {i}",
        output_text=f"回答 {i}",
        tools_used=["search"] if i < 30 else ["search", "search", "calculator"],
        confidence=0.80 if i < 30 else 0.96,  # 后半段置信度飙升
        user_feedback=None,
        timestamp=time.time() + i * 60
    ))

# 运行 Dreaming 循环
plan = dreaming.run_cycle()
if plan:
    print(f"\n改进计划详情:")
    print(f"  策略: {plan.strategy}")
    print(f"  行动: {plan.actions}")
    print(f"  预期: {plan.expected_outcome}")
    print(f"  验证: {plan.verification_method}")