AI Coding Agent 质量与可靠性保障：从 Anthropic 事故复盘到生产级监控体系构建

from agents import Agent, Runner, function_tool, GuardrailFunctionOutput
from typing import Optional
import logging
import asyncio
import time

# 1. 输入护栏：验证用户输入的安全性
def input_guardrail(input: str) -> GuardrailFunctionOutput:
    """输入安全检查：防止 prompt injection 和恶意输入"""
    suspicious_patterns = [
        "ignore previous instructions",
        "you are now",
        "system prompt",
    ]
    for pattern in suspicious_patterns:
        if pattern.lower() in input.lower():
            return GuardrailFunctionOutput(
                output_info={"type": "injection_attempt", "pattern": pattern},
                tripwire_triggered=True
            )
    return GuardrailFunctionOutput(
        output_info={"type": "clean"},
        tripwire_triggered=False
    )

# 2. 输出护栏：验证 Agent 输出的安全性
def output_guardrail(output: str) -> GuardrailFunctionOutput:
    """输出安全检查：防止敏感信息泄露"""
    sensitive_patterns = [
        "api_key", "password", "secret", "token"
    ]
    for pattern in sensitive_patterns:
        if pattern.lower() in output.lower():
            return GuardrailFunctionOutput(
                output_info={"type": "data_leak", "pattern": pattern},
                tripwire_triggered=True
            )
    return GuardrailFunctionOutput(output_info={"type": "clean"}, tripwire_triggered=False)

# 3. 带有完整防护的 Agent 定义
research_agent = Agent(
    name="research-agent",
    instructions="你是一个研究助手，帮助用户查找和分析信息。",
    input_guardrails=[input_guardrail],
    output_guardrails=[output_guardrail],
    model="gpt-5.5",
    # 4. 超时和重试配置
    tool_config={
        "max_tool_calls": 20,
        "tool_timeout_seconds": 30,
        "retry_on_failure": True,
        "max_retries": 3,
    }
)

# 4. 会话管理：带有心跳和状态恢复
class ResilientSession:
    """弹性会话管理器：处理超时、状态恢复和上下文压缩"""
    
    def __init__(self, agent: Agent, heartbeat_interval: int = 3600):
        self.agent = agent
        self.heartbeat_interval = heartbeat_interval  # 1 小时心跳
        self.session_state = {}
        self.last_heartbeat = None
        self.context_budget = 128000  # 最大 token 数
        self.logger = logging.getLogger(__name__)
    
    async def run_with_recovery(self, user_input: str) -> dict:
        """带恢复机制的运行方法"""
        try:
            # 检查是否需要上下文压缩
            if self._estimate_tokens() > self.context_budget * 0.8:
                self._compress_context()
                self.logger.info("上下文已压缩")
            
            # 执行 Agent 运行，带超时
            result = await asyncio.wait_for(
                Runner.run(self.agent, user_input),
                timeout=120  # 2 分钟超时
            )
            
            # 更新心跳
            self.last_heartbeat = time.time()
            
            return {
                "status": "success",
                "result": result.output,
                "tokens_used": result.usage.total_tokens,
            }
        except asyncio.TimeoutError:
            self.logger.error("Agent 运行超时，执行恢复")
            return self._handle_timeout(user_input)
        except Exception as e:
            self.logger.error(f"Agent 运行异常: {e}")
            return self._handle_error(user_input, e)
    
    def _compress_context(self):
        """上下文压缩：保留关键信息，丢弃低优先级内容"""
        # 优先级：最近的对话 > 工具调用结果 > 早期对话
        # 实现取决于具体的记忆系统
        pass
    
    def _estimate_tokens(self) -> int:
        """估算当前会话的 token 使用量"""
        # 可以使用 tiktoken 进行估算
        import tiktoken
        enc = tiktoken.encoding_for_model("gpt-4")
        total = 0
        for key, value in self.session_state.items():
            total += len(enc.encode(str(value)))
        return total

from openai import OpenAI
import json
from dataclasses import dataclass
from typing import List
from difflib import SequenceMatcher

@dataclass
class EvalResult:
    test_name: str
    score: float  # 0-1
    reasoning: str
    passed: bool

class AgentQualityEvaluator:
    """AI Agent 质量评估器"""
    
    def __init__(self, judge_model: str = "gpt-5.5"):
        self.client = OpenAI()
        self.judge_model = judge_model
    
    def evaluate_task_completion(
        self, 
        task: str, 
        expected: str, 
        actual: str
    ) -> EvalResult:
        """评估任务完成度"""
        prompt = f"""
你是一个专业的 AI Agent 质量评估员。请评估以下 Agent 的执行结果。

## 任务
{task}

## 期望结果
{expected}

## 实际结果
{actual}

请从以下维度打分（0-1）：
1. 完整性：是否覆盖了任务的所有要求
2. 准确性：结果是否正确，有无幻觉
3. 实用性：结果是否可以直接使用

以 JSON 格式返回：
{{"score": 0.85, "reasoning": "...", "passed": true}}
"""
        response = self.client.chat.completions.create(
            model=self.judge_model,
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"},
            temperature=0.0,  # 评估需要确定性
        )
        
        result = json.loads(response.choices[0].message.content)
        return EvalResult(
            test_name="task_completion",
            score=result["score"],
            reasoning=result["reasoning"],
            passed=result.get("passed", result["score"] >= 0.7)
        )
    
    def evaluate_session_quality(
        self,
        conversation_history: List[dict],
        session_id: str,
    ) -> dict:
        """评估会话质量：检测遗忘、重复、幻觉"""
        issues = []
        
        # 1. 检测重复回答
        responses = [msg["content"] for msg in conversation_history if msg["role"] == "assistant"]
        for i in range(1, len(responses)):
            if self._similarity(responses[i-1], responses[i]) > 0.85:
                issues.append({
                    "type": "repetitive_response",
                    "description": "连续回复高度相似，可能出现了思考清除 Bug"
                })
        
        # 2. 检测上下文丢失
        references = self._extract_references(conversation_history)
        missing_refs = self._check_reference_usage(conversation_history, references)
        if missing_refs:
            issues.append({
                "type": "context_loss",
                "description": f"Agent 忘记提及的信息: {missing_refs}"
            })
        
        return {
            "session_id": session_id,
            "issues": issues,
            "quality_score": max(0, 1 - len(issues) * 0.25),
        }
    
    def _similarity(self, text1: str, text2: str) -> float:
        """计算文本相似度（简化版，实际应使用 embedding）"""
        return SequenceMatcher(None, text1, text2).ratio()
    
    def _extract_references(self, history: List[dict]) -> List[str]:
        """提取对话中的关键信息点"""
        # 简化实现：提取命名实体、代码片段等
        pass
    
    def _check_reference_usage(self, history: List[dict], references: List[str]) -> List[str]:
        """检查 Agent 是否正确引用了之前的信息"""
        missing = []
        for ref in references:
            found = False
            for msg in history:
                if msg["role"] == "assistant" and ref in msg["content"]:
                    found = True
                    break
            if not found:
                missing.append(ref)
        return missing


# 使用示例
evaluator = AgentQualityEvaluator()

# 测试任务完成度
result = evaluator.evaluate_task_completion(
    task="编写一个 Python 函数，实现快速排序算法",
    expected="一个正确的快速排序实现，包含边界条件处理",
    actual="def quicksort(arr): ... # 实际生成的代码"
)
print(f"任务完成度: {result.score:.2f}, 通过: {result.passed}")

from opentelemetry import trace, metrics
from opentelemetry.trace import Status, StatusCode
from opentelemetry.metrics import Counter, Histogram
import functools
import time

# 初始化 Tracer
tracer = trace.get_tracer("ai-agent")
meter = metrics.get_meter("ai-agent")

# 定义指标
task_success_counter = meter.create_counter(
    "agent.task.success", description="任务成功次数"
)
task_failure_counter = meter.create_counter(
    "agent.task.failure", description="任务失败次数"
)
tool_call_duration = meter.create_histogram(
    "agent.tool.duration", description="工具调用延迟", unit="ms"
)
token_usage_counter = meter.create_counter(
    "agent.token.usage", description="Token 使用量"
)
hallucination_counter = meter.create_counter(
    "agent.hallucination", description="幻觉检测次数"
)

class MonitoredAgent:
    """带有完整可观测性的 Agent 封装"""
    
    def __init__(self, agent, evaluator: AgentQualityEvaluator):
        self.agent = agent
        self.evaluator = evaluator
        self.session_traces = []
    
    async def run(self, user_input: str) -> dict:
        with tracer.start_as_current_span("agent.run") as span:
            span.set_attribute("input_length", len(user_input))
            start_time = time.time()
            
            try:
                result = await self.agent.run(user_input)
                
                # 记录结果属性
                span.set_attribute("output_length", len(result.output))
                span.set_attribute("tool_calls", len(result.tool_calls))
                span.set_status(Status(StatusCode.OK))
                
                # 质量评估
                quality = await self._evaluate_quality(user_input, result)
                span.set_attribute("quality_score", quality["score"])
                
                if quality.get("hallucination_detected"):
                    hallucination_counter.add(1)
                    span.add_event("hallucination_detected", {
                        "severity": quality.get("severity", "unknown")
                    })
                
                # 指标上报
                task_success_counter.add(1)
                duration_ms = (time.time() - start_time) * 1000
                span.set_attribute("duration_ms", duration_ms)
                
                return {
                    "output": result.output,
                    "quality": quality,
                    "trace_id": span.get_span_context().trace_id,
                }
                
            except Exception as e:
                span.set_status(Status(StatusCode.ERROR, str(e)))
                span.record_exception(e)
                task_failure_counter.add(1)
                raise
    
    async def _evaluate_quality(self, input_text: str, result) -> dict:
        """调用评估器进行实时质量检查"""
        return await self.evaluator.quick_eval(
            input_text, result.output, result.tool_calls
        )


# 使用 Trace 数据进行事后分析
def analyze_degradation(traces: list) -> dict:
    """分析 Agent 质量退化趋势"""
    hourly_metrics = {}
    
    for t in traces:
        hour = t["timestamp"][:13]  # YYYY-MM-DDTHH
        if hour not in hourly_metrics:
            hourly_metrics[hour] = {
                "total_tasks": 0,
                "success_tasks": 0,
                "hallucinations": 0,
                "avg_quality": [],
                "avg_latency": [],
            }
        
        hourly_metrics[hour]["total_tasks"] += 1
        if t["status"] == "success":
            hourly_metrics[hour]["success_tasks"] += 1
        if t.get("hallucination"):
            hourly_metrics[hour]["hallucinations"] += 1
        hourly_metrics[hour]["avg_quality"].append(t.get("quality_score", 0))
        hourly_metrics[hour]["avg_latency"].append(t.get("duration_ms", 0))
    
    # 计算趋势
    degradation_signals = []
    sorted_hours = sorted(hourly_metrics.keys())
    for i in range(1, len(sorted_hours)):
        prev_hour = sorted_hours[i-1]
        curr_hour = sorted_hours[i]
        prev_data = hourly_metrics[prev_hour]
        curr_data = hourly_metrics[curr_hour]
        
        # 检测质量突降
        prev_success_rate = prev_data["success_tasks"] / max(prev_data["total_tasks"], 1)
        curr_success_rate = curr_data["success_tasks"] / max(curr_data["total_tasks"], 1)
        
        if prev_success_rate - curr_success_rate > 0.15:  # 下降超过 15%
            degradation_signals.append({
                "hour": curr_hour,
                "signal": "success_rate_drop",
                "drop": round(prev_success_rate - curr_success_rate, 2)
            })
    
    return {
        "hourly_metrics": hourly_metrics,
        "degradation_signals": degradation_signals,
    }