规避 Demo 陷阱，构建可投产的 Agent 系统

Level 0: 无 Agent（纯规则引擎）
  └── 完全确定，无 AI 介入

Level 1: 单轮 LLM 调用（ChatBot）
  └── 有 AI 回复，无工具调用，无状态

Level 2: 固定流程 Agent（Workflow + LLM）  ← 大部分生产系统在此
  └── 流程由代码控制，某些节点用 LLM 填充

Level 3: 动态决策 Agent（ReAct / Toolformer）
  └── Agent 自主选择工具和决策路径  ← 本文重点

Level 4: 自演进 Agent（持续学习）
  └── 能从历史交互中自动优化行为

# PEV 模式核心实现
class PlanExecuteVerifyAgent:
    def __init__(self, planner, executor, verifier, llm):
        self.planner = planner
        self.executor = executor
        self.verifier = verifier
        self.llm = llm
        
    def process(self, user_request: str) -> AgentResponse:
        # ===== PLAN 阶段 =====
        plan = self.planner.generate_plan(user_request)
        # plan 是结构化的 JSON，例如：
        # {
        #   "steps": [
        #     {"id": 1, "tool": "search_docs", "params": {...}},
        #     {"id": 2, "tool": "query_db", "params": {...}}
        #   ],
        #   "estimated_steps": 3,
        #   "estimated_tokens": 2000
        # }
        
        # 计划校验：是否在允许范围内？
        if not self.is_plan_valid(plan):
            return self.fallback_response("请求过于复杂，请简化或联系人工")
        
        # 计划审批（高风险操作）
        if self.requires_approval(plan):
            approval = self.request_approval(plan)
            if not approval.granted:
                return self.fallback_response("操作需要审批，请等待或联系管理员")
        
        # ===== EXECUTE 阶段 =====
        execution_state = ExecutionState()
        for step in plan['steps']:
            # 执行前检查：是否超预算？
            if self.budget_exceeded(execution_state):
                return self.fallback_response("执行预算超限，已停止")
            
            # 执行步骤
            result = self.executor.execute(step)
            execution_state.add_result(step['id'], result)
            
            # 检查是否需要提前终止（例如已找到答案）
            if self.should_early_stop(execution_state):
                break
        
        # ===== VERIFY 阶段 =====
        verification = self.verifier.verify(execution_state, user_request)
        
        if verification.passed:
            # 生成最终响应
            return self.generate_response(execution_state, verification)
        else:
            # 验证失败：尝试重试或降级
            return self.handle_verification_failure(
                execution_state, 
                verification, 
                user_request
            )

class Planner:
    def __init__(self, llm, tool_registry, plan_schema):
        self.llm = llm
        self.tool_registry = tool_registry
        self.plan_schema = plan_schema  # JSON Schema
        
    def generate_plan(self, user_request: str) -> dict:
        # 1. 获取可用工具列表（动态过滤，只给相关的）
        available_tools = self.tool_registry.get_tools_for_scenario(user_request)
        
        # 2. 构建结构化的 Prompt
        prompt = f"""
        你是一个任务规划专家。请根据用户请求生成执行计划。
        
        用户请求：{user_request}
        
        可用工具：
        {self.format_tools(available_tools)}
        
        输出格式必须符合以下 JSON Schema：
        {json.dumps(self.plan_schema, indent=2)}
        
        注意：
        1. 每一步必须使用列出的工具，不要发明新工具
        2. 步骤数不超过 5 步
        3. 如果请求需要的信息不明确，在第一步设置为 "ask_clarification"
        """
        
        # 3. 调用 LLM 生成计划
        response = self.llm.invoke(
            prompt,
            response_format="json",
            temperature=0.1  # 低温度保证确定性
        )
        
        # 4. 用 JSON Schema 严格校验
        plan = json.loads(response)
        jsonschema.validate(plan, self.plan_schema)
        
        return plan

// 工具定义（带 Schema）
@Data
@Builder
public class ToolDefinition {
    private String name;                    // 唯一标识
    private String description;             // 人类可读描述
    private Map<String, ParameterSchema> parameters;  // 参数 Schema
    private Set<String> requiredPermissions; // 所需权限
    private boolean isDestructive;          // 是否具有破坏性
    private int maxRetries;                 // 最大重试次数
    private long timeoutMs;                 // 超时时间
    private String fallbackTool;            // 降级时使用的备选工具
}

// 工具注册中心
@Component
public class ToolRegistry {
    private final Map<String, ToolDefinition> definitions = new ConcurrentHashMap<>();
    private final Map<String, ToolExecutor> executors = new ConcurrentHashMap<>();
    
    public void register(ToolDefinition def, ToolExecutor executor) {
        // 校验：工具名称不能重复
        if (definitions.containsKey(def.getName())) {
            throw new DuplicateToolException("工具已存在: " + def.getName());
        }
        
        // 校验：参数 Schema 必须有效
        validateSchema(def.getParameters());
        
        definitions.put(def.getName(), def);
        executors.put(def.getName(), executor);
        
        log.info("工具注册成功: {} (destructive: {})", 
            def.getName(), def.isDestructive());
    }
    
    public List<ToolDefinition> getAvailableTools(Set<String> permissions) {
        return definitions.values().stream()
            .filter(def -> permissions.containsAll(def.getRequiredPermissions()))
            .collect(Collectors.toList());
    }
    
    public ToolDefinition getDefinition(String name) {
        return definitions.get(name);
    }
}

class ToolExecutor:
    def __init__(self, registry, circuit_breaker):
        self.registry = registry
        self.cb = circuit_breaker
        
    def execute(self, tool_name: str, params: dict) -> ToolResult:
        tool_def = self.registry.get_definition(tool_name)
        if not tool_def:
            return ToolResult.error(f"未知工具: {tool_name}")
        
        # 1. 参数校验
        try:
            validate_params(params, tool_def.parameters)
        except ValidationError as e:
            return ToolResult.error(f"参数错误: {e}")
        
        # 2. 使用熔断器包裹执行
        @self.cb(tool_name)
        def _execute():
            executor = self.registry.get_executor(tool_name)
            return executor(params)
        
        # 3. 重试逻辑（带退避）
        retries = 0
        while retries <= tool_def.maxRetries:
            try:
                result = _execute()
                return self.wrap_result(result)
            except (TimeoutError, CircuitBreakerOpenError) as e:
                # 不重试：超时和熔断说明系统有问题
                return self.handle_tool_error(tool_def, e)
            except Exception as e:
                # 可重试的错误
                retries += 1
                if retries > tool_def.maxRetries:
                    # 尝试降级
                    return self.fallback(tool_def, params, e)
                time.sleep(2 ** retries * 0.1)  # 指数退避
        
        return ToolResult.error("所有重试均失败")
    
    def fallback(self, tool_def, params, error):
        if tool_def.fallbackTool:
            # 调用备选工具
            return self.execute(tool_def.fallbackTool, params)
        else:
            # 返回错误信息，让上层决定怎么处理
            return ToolResult.fallback(f"工具 {tool_def.name} 不可用: {error}")

# 工具沙箱配置
sandbox:
  # 读操作：允许直接访问（有缓存）
  read_operations:
    - search_docs
    - get_user_info
    - query_metrics
  
  # 写操作：必须经过审批队列
  write_operations:
    - update_ticket
    - send_notification
    - create_incident
  
  # 危险操作：只能在测试环境执行
  dangerous_operations:
    - delete_data
    - restart_service
    - modify_config
  
  # 隔离策略
  isolation:
    network: "restricted"  # 只能访问特定内网
    data: "readonly"       # 写操作需审批
    resources: "limited"   # CPU/内存限制

class MemorySystem:
    def __init__(self, redis_client, db_session, vector_store):
        self.redis = redis_client
        self.db = db_session
        self.vector_store = vector_store
        
    def get_context(self, session_id: str, user_id: str) -> MemoryContext:
        # L1：会话记忆（最近20轮）
        session_key = f"session:{session_id}"
        recent_history = self.redis.lrange(session_key, -20, -1)
        
        # L2：用户记忆
        user_profile = self.db.query(UserProfile).filter_by(user_id=user_id).first()
        
        # L3：场景记忆（只检索与当前问题相关的）
        current_query = recent_history[-1]['content'] if recent_history else ""
        similar_experiences = self.vector_store.search(
            query=current_query,
            top_k=3,
            filter={"user_id": user_id}
        )
        
        # 压缩：确保总 Token 数不超标
        context = self.compress([
            *recent_history[-5:],  # 最近5轮必须保留
            self.summarize(user_profile.history) if user_profile else None,
            *similar_experiences
        ], max_tokens=4000)
        
        return MemoryContext(context)
    
    def summarize(self, long_history: list) -> str:
        """用 LLM 压缩长历史"""
        if len(long_history) < 30:
            return ""
        
        # 调用 LLM 生成摘要
        summary = self.llm.invoke(f"请总结以下对话的关键信息：{long_history}")
        return f"[历史摘要] {summary}"

class MemoryManager:
    def forget_strategy(self, memory_type: str, content: dict) -> bool:
        """决定是否遗忘某条记忆"""
        if memory_type == "session":
            # 会话记忆：30分钟无活动后过期（由Redis TTL保证）
            return True  # Redis 自动处理
        
        if memory_type == "user_preference":
            # 用户偏好：超过30天未使用则遗忘
            last_used = content.get('last_used')
            if last_used and (datetime.now() - last_used).days > 30:
                return True
        
        if memory_type == "task_completion":
            # 任务完成记录：保留3个月用于分析
            completed_at = content.get('completed_at')
            if completed_at and (datetime.now() - completed_at).days > 90:
                return True
        
        return False

@Slf4j
@Component
public class AgentTracer {
    
    private static final String TRACE_ID_HEADER = "X-Trace-Id";
    
    public TracingContext startTrace(String sessionId, String userInput) {
        String traceId = UUID.randomUUID().toString();
        TracingContext ctx = TracingContext.builder()
            .traceId(traceId)
            .sessionId(sessionId)
            .startTime(Instant.now())
            .userInput(userInput)
            .build();
        
        // 结构化日志
        log.info("AGENT_TRACE_START | traceId={} | sessionId={} | input={}",
            traceId, sessionId, truncate(userInput, 100));
        
        return ctx;
    }
    
    public void logDecision(TracingContext ctx, String step, String decision, 
                            Map<String, Object> metadata) {
        log.info("AGENT_TRACE_STEP | traceId={} | step={} | decision={} | meta={}",
            ctx.getTraceId(), step, decision, toJson(metadata));
        
        // 同时写入专门的追踪存储（如Elasticsearch）
        traceRepository.save(TraceEntry.builder()
            .traceId(ctx.getTraceId())
            .timestamp(Instant.now())
            .step(step)
            .decision(decision)
            .metadata(metadata)
            .build()
        );
    }
    
    public void logToolCall(TracingContext ctx, String toolName, 
                            Map<String, Object> params, ToolResult result) {
        log.info("AGENT_TOOL_CALL | traceId={} | tool={} | params={} | result={} | latency={}ms",
            ctx.getTraceId(), toolName, toJson(params), 
            result.isSuccess() ? "success" : "fail",
            result.getLatency());
        
        // 审计日志（合规要求，长期存储）
        auditLog.toolCall(ctx.getTraceId(), toolName, params, result);
    }
}

# Prometheus 指标定义
metrics:
  - name: agent_requests_total
    labels: [scenario, model, outcome]
    type: counter
    
  - name: agent_latency_seconds
    labels: [scenario, phase]  # plan/execute/verify
    type: histogram
    buckets: [0.1, 0.5, 1.0, 2.0, 5.0, 10.0]
    
  - name: agent_tool_calls_total
    labels: [tool_name, success]
    type: counter
    
  - name: agent_plan_steps_count
    labels: [scenario]
    type: histogram
    buckets: [1, 2, 3, 5, 8]
    
  - name: agent_budget_exceeded_total
    labels: [reason]  # token/time/step
    type: counter
    
  - name: agent_human_escalation_total
    labels: [reason]  # confidence/verification/user_request
    type: counter

class AgentReplay:
    def __init__(self, trace_repository):
        self.trace_repo = trace_repository
        
    def replay_trace(self, trace_id: str, stop_at_step: int = None):
        """
        重放某次完整执行过程，用于调试
        """
        trace = self.trace_repo.get(trace_id)
        
        # 使用 Mock 工具替换真实工具（避免副作用）
        with self.use_mock_tools():
            # 从相同的输入开始
            agent = Agent()
            for step in trace.steps[:stop_at_step]:
                # 在每个决策点打印当前状态
                print(f"Step {step.index}: {step.decision}")
                print(f"  State: {step.state}")
                print(f"  LLM Input: {step.prompt}")
                print(f"  LLM Output: {step.response}")
                print("-" * 50)
            
            # 从断点继续执行
            if stop_at_step:
                remaining = trace.steps[stop_at_step:]
                agent.continue_from(remaining)

class TokenBudgetController:
    def __init__(self, limits: dict):
        self.max_per_request = limits.get('per_request', 8000)  # 单次请求
        self.max_per_session = limits.get('per_session', 50000) # 单次会话
        self.warning_threshold = 0.8
        
    def check_budget(self, session_id: str, current_tokens: int):
        # 1. 获取会话已消耗
        session_usage = redis.get(f"token_usage:{session_id}") or 0
        
        # 2. 检查是否超标
        if current_tokens > self.max_per_request:
            raise TokenBudgetExceeded(f"单次请求Token超限: {current_tokens}")
        
        if session_usage + current_tokens > self.max_per_session:
            raise TokenBudgetExceeded(f"会话Token超限，已用 {session_usage}")
        
        # 3. 预警
        if session_usage / self.max_per_session > self.warning_threshold:
            logger.warning(f"会话 {session_id} Token使用已达 {session_usage/self.max_per_session:.0%}")