智能体从验证到上线的十大陷阱

// ❌ 反模式：仅靠 Prompt 约束
const systemPrompt = `你是销售助手。折扣不能超过10%。`;

// ✅ 正确模式：系统层面的硬性约束
interface DiscountPolicy {
  maxDiscountPercent: number;
  maxDiscountAmount: number;
  requiresApproval: boolean;
  approvalThreshold: number;
}

class SalesAgentGuardrail {
  private policy: DiscountPolicy = {
    maxDiscountPercent: 10,
    maxDiscountAmount: 500,
    requiresApproval: true,
    approvalThreshold: 200,
  };

  async applyDiscount(
    orderId: string,
    requestedDiscount: number
  ): Promise<DiscountResult> {
    // 硬性上限检查 —— Agent 无法绕过
    if (requestedDiscount > this.policy.maxDiscountPercent) {
      return {
        allowed: false,
        reason: `折扣 ${requestedDiscount}% 超过上限 ${this.policy.maxDiscountPercent}%`,
        action: 'BLOCKED',
      };
    }

    const discountAmount = this.calculateAmount(orderId, requestedDiscount);

    // 金额门控 —— 触发人工审批
    if (discountAmount > this.policy.approvalThreshold) {
      return {
        allowed: false,
        reason: '折扣金额超过阈值，需人工审批',
        action: 'PENDING_APPROVAL',
        approvalRequest: { orderId, discountAmount, requestedDiscount },
      };
    }

    return { allowed: true, action: 'APPLIED', discountAmount };
  }
}

from dataclasses import dataclass
from typing import List

@dataclass
class RetrievalConfig:
    max_chunks: int = 5          # 严格限制最大片段数
    min_relevance_score: float = 0.82  # 相关性阈值
    max_context_tokens: int = 2000     # 上下文 Token 上限
    rerank_model: str = "cross-encoder/ms-marco-MiniLM-L-12-v2"

class SmartRetriever:
    def __init__(self, config: RetrievalConfig):
        self.config = config
        self.reranker = CrossEncoder(config.rerank_model)
    
    def retrieve(self, query: str, documents: List[str]) -> List[str]:
        # 第一阶段：语义召回（宽松阈值，Top-50）
        candidates = self.vector_search(query, top_k=50)
        
        # 第二阶段：交叉编码器精排（严格阈值）
        scored = self.reranker.predict(
            [(query, doc.content) for doc in candidates]
        )
        ranked = sorted(
            zip(candidates, scored), key=lambda x: x[1], reverse=True
        )
        
        # 第三阶段：动态裁剪——按质量截断而非固定数量
        selected = []
        total_tokens = 0
        for doc, score in ranked:
            if score < self.config.min_relevance_score:
                break  # 分数低于阈值直接截断
            if total_tokens + doc.token_count > self.config.max_context_tokens:
                break  # 超出 Token 预算
            if len(selected) >= self.config.max_chunks:
                break
            selected.append(doc)
            total_tokens += doc.token_count
        
        return selected

# OpenTelemetry 配置示例：Agent 追踪
apiVersion: v1
kind: ConfigMap
metadata:
  name: agent-otel-config
data:
  collector-config.yaml: |
    receivers:
      otlp:
        protocols:
          grpc:
            endpoint: "0.0.0.0:4317"
    processors:
      batch:
        timeout: 5s
      attributes:
        actions:
          - key: agent.session_id
            action: insert
          - key: agent.step_index
            action: insert
          - key: llm.model
            action: insert
          - key: llm.token_usage.input
            action: insert
          - key: llm.token_usage.output
            action: insert
          - key: tool.name
            action: insert
          - key: tool.result_status
            action: insert
    exporters:
      otlp:
        endpoint: "tempo:4317"
      prometheus:
        endpoint: "0.0.0.0:8889"
    service:
      pipelines:
        traces:
          receivers: [otlp]
          processors: [batch, attributes]
          exporters: [otlp]
        metrics:
          receivers: [otlp]
          processors: [batch]
          exporters: [prometheus]

interface CheckpointConfig {
  stepIndex: number;
  validator: (output: any) => ValidationResult;
  onFailure: 'retry' | 'escalate' | 'abort';
  maxRetries: number;
}

class CheckpointedPipeline {
  private checkpoints: Map<number, CheckpointConfig> = new Map();

  addCheckpoint(config: CheckpointConfig) {
    this.checkpoints.set(config.stepIndex, config);
  }

  async execute(steps: AgentStep[]): Promise<PipelineResult> {
    const results: StepResult[] = [];

    for (let i = 0; i < steps.length; i++) {
      const output = await steps[i].run(results);
      results.push(output);

      // 到达 Checkpoint 时执行验证
      const checkpoint = this.checkpoints.get(i);
      if (checkpoint) {
        const validation = checkpoint.validator(output);
        if (!validation.passed) {
          switch (checkpoint.onFailure) {
            case 'retry':
              // 从当前步骤重试，最多 N 次
              return this.retryFrom(i, steps, results, checkpoint.maxRetries);
            case 'escalate':
              // 触发人工审核
              return this.escalateToHuman(i, output, validation);
            case 'abort':
              return { status: 'aborted', failedAt: i, reason: validation.error };
          }
        }
      }
    }
    return { status: 'completed', results };
  }
}

// 抽象 LLM 接口 —— 与具体提供商解耦
interface LLMProvider {
  complete(request: CompletionRequest): Promise<CompletionResponse>;
  streamComplete(request: CompletionRequest): AsyncIterable<StreamChunk>;
}

interface CompletionRequest {
  messages: Message[];
  tools?: ToolDefinition[];
  temperature?: number;
  maxTokens?: number;
}

// 具体实现可随时替换
class OpenAIProvider implements LLMProvider { /* ... */ }
class AnthropicProvider implements LLMProvider { /* ... */ }
class LocalModelProvider implements LLMProvider { /* ... */ }

// Agent 核心只依赖抽象接口
class Agent {
  constructor(
    private llm: LLMProvider,
    private tools: ToolRegistry,
    private memory: MemoryStore
  ) {}

  async run(input: string): Promise<AgentResponse> {
    // 业务逻辑与 LLM 提供商无关
    const plan = await this.plan(input);
    return this.execute(plan);
  }
}

from dataclasses import dataclass, field
from typing import List, Callable
import json

@dataclass
class EvalCase:
    input: str
    expected_behavior: str
    category: str  # "happy_path" | "edge_case" | "adversarial"
    validators: List[Callable] = field(default_factory=list)

@dataclass 
class EvalResult:
    case_id: str
    passed: bool
    score: float
    latency_ms: int
    token_usage: int
    failure_reason: str = ""

class AgentEvalPipeline:
    def __init__(self, agent, eval_cases: List[EvalCase]):
        self.agent = agent
        self.cases = eval_cases
        self.results: List[EvalResult] = []
    
    def run_suite(self) -> EvalSummary:
        for case in self.cases:
            result = self._evaluate_single(case)
            self.results.append(result)
        
        return EvalSummary(
            total=len(self.results),
            passed=sum(1 for r in self.results if r.passed),
            avg_score=sum(r.score for r in self.results) / len(self.results),
            avg_latency=sum(r.latency_ms for r in self.results) / len(self.results),
            failures_by_category=self._group_failures(),
            regression_detected=self._check_regression(),
        )
    
    def _check_regression(self) -> bool:
        """对比上次运行结果，检测回归"""
        previous = self._load_baseline()
        if not previous:
            return False
        current_score = sum(r.score for r in self.results) / len(self.results)
        return current_score < previous.avg_score - 0.05  # 5% 退化阈值