面向业务场景的 AGI 大模型应用开发实践：避开 Demo 陷阱

□ 该场景当前有人在做，且花费时间>2小时/天
□ 输出质量有“好”与“差”的明显区分，不存在唯一正确答案
□ 有历史数据（至少1000条以上）可作为效果评估基线
□ 场景边界清晰（不是“让AI管公司”，而是“让AI整理会议纪要”）
□ 错误结果的影响可控（不会导致客户流失或法律纠纷）
□ 用户愿意为效率提升付费或付出行为改变成本
□ 有明确的成功衡量指标（如：处理时长从30分钟降到5分钟）

数据准备三层结构：
├── 层1：原始数据（直接从生产环境拉取）
│   ├── 客服对话记录（含拼写错误、口语化表达）
│   ├── 用户上传的PDF/Word（格式混乱、有水印）
│   └── 系统日志（含大量噪音）
│
├── 层2：清洗与标注数据（投入最大人力的环节）
│   ├── 脱敏处理（去除PII）
│   ├── 统一格式（HTML转Markdown、PDF转文本）
│   ├── 质量筛选（去除无效样本）
│   └── 人工标注（建立Gold Standard）
│
└── 层3：增强数据（用于RAG检索或微调）
    ├── 知识图谱构建（实体关系抽取）
    ├── Q&A对生成（用大模型辅助生成训练数据）
    └── 负样本构造（用于边界测试）

# 主动学习标注策略（用模型辅助筛选难例）
def active_learning_pipeline(unlabeled_data, model, budget=500):
    """
    仅标注模型最不确定的样本，用最少的人工标注实现最大效果提升
    """
    # 1. 用当前模型对未标注数据做预测
    predictions = model.predict(unlabeled_data)
    
    # 2. 按置信度排序，取最低的N条（模型最不确定的）
    uncertainties = [abs(p - 0.5) for p in predictions]  # 二分类举例
    uncertain_indices = np.argsort(uncertainties)[:budget]
    
    # 3. 推送给人标注
    samples_to_label = [unlabeled_data[i] for i in uncertain_indices]
    labeled = human_annotation(samples_to_label)
    
    # 4. 加入训练集，重新训练模型
    return labeled

# Golden Set 构建流程
class GoldenSetBuilder:
    def __init__(self):
        self.human_answers = []  # 人类专家答案
        self.ai_answers = []     # 待评估的AI答案
    
    def build(self, test_queries: List[str], num_experts=3):
        """
        每个问题由3位专家独立回答，取一致性最高的作为Golden Answer
        """
        for query in test_queries:
            # 1. 收集专家答案
            expert_answers = [get_expert_answer(query) for _ in range(num_experts)]
            
            # 2. 计算答案间的一致性
            similarities = []
            for i in range(num_experts):
                for j in range(i+1, num_experts):
                    sim = compute_similarity(expert_answers[i], expert_answers[j])
                    similarities.append(sim)
            
            # 3. 一致性>0.8的采纳，否则该问题留待讨论
            if np.mean(similarities) > 0.8:
                # 取投票最多的答案
                golden = vote(expert_answers)
                self.human_answers.append({
                    'query': query,
                    'golden_answer': golden,
                    'confidence': np.mean(similarities)
                })
            else:
                # 需要专家讨论达成共识
                golden = expert_consensus_discussion(expert_answers)
                self.human_answers.append({
                    'query': query,
                    'golden_answer': golden,
                    'confidence': 1.0  # 讨论后确信
                })
        
        return self.human_answers

# 评估监控配置
evaluation_pipeline:
  frequency: daily
  sample_size: 100  # 每天随机抽样100条生产请求
  
  metrics:
    - llm_as_judge:  # 用更强的大模型做评估
      model: "gpt-4-turbo"
      prompt_template: "请评价以下回答的质量，打分1-10..."
      
    - user_feedback:  # 收集用户反馈
      source: "thumbs_up/down_buttons"
      threshold: 0.8  # 满意度低于80%触发告警
      
    - hallucination_detection:
      method: "fact_checking_with_retrieval"
      threshold: 0.05  # 幻觉率超过5%触发告警
      
  alerting:
    - slack_channel: "#ai-alerts"
    - email: "ai-team@company.com"

# 错误：无限工具集
class UniversalAgent:
    tools = ALL_TOOLS  # 50+个工具
    
# 正确：场景限定的Agent
class CustomerSupportAgent:
    tools = [
        query_order_status,  # 查订单
        search_known_issues, # 查已知问题
        create_ticket        # 建工单
    ]  # 只允许3个工具
    
    def should_escalate(self, user_input):
        # 明确的人机交接策略
        if "投诉" in user_input or "退款" in user_input:
            return True
        return False

// 输出校验与修正器
@Component
public class OutputGuardian {
    
    private static final Pattern JSON_PATTERN = Pattern.compile(
        "\\{[^{}]*(?:\\{[^{}]*\\}[^{}]*)*\\}"
    );
    
    public StructuredOutput validateAndFix(String rawOutput, Class<?> expectedType) {
        // 1. 尝试提取JSON（模型经常加额外文字）
        Matcher m = JSON_PATTERN.matcher(rawOutput);
        if (!m.find()) {
            // 如果完全找不到JSON，触发降级
            return fallbackResponse("模型输出格式异常");
        }
        
        String jsonStr = m.group();
        
        try {
            // 2. 尝试反序列化
            Object result = objectMapper.readValue(jsonStr, expectedType);
            
            // 3. 业务规则校验
            validateBusinessRules(result);
            
            return new StructuredOutput(result, true);
            
        } catch (JsonProcessingException e) {
            // 4. JSON格式错误，尝试修复常见问题
            String fixed = fixCommonJsonErrors(jsonStr);
            try {
                Object result = objectMapper.readValue(fixed, expectedType);
                return new StructuredOutput(result, true, "已自动修复JSON格式");
            } catch (Exception ex) {
                // 5. 修复失败，降级
                return fallbackResponse("无法解析模型输出，请重试");
            }
        }
    }
}

# 智能路由示例
class ModelRouter:
    def route(self, query, context):
        # 1. 计算查询复杂度
        complexity = self.estimate_complexity(query)
        
        # 2. 检查缓存
        cached = self.cache.get(query)
        if cached:
            return cached, "cache_hit"
        
        # 3. 根据复杂度选择模型
        if complexity < 0.3:
            # 简单事实性问题 -> 小模型
            model = "deepseek-lite"
        elif complexity < 0.7:
            # 中等推理 -> 中模型
            model = "deepseek-chat"
        else:
            # 复杂推理 -> 大模型
            model = "gpt-4-turbo"
        
        # 4. 记录路由决策，用于后续优化
        self.log_decision(query, complexity, model)
        
        return model, "model_routed"

# 人机交接协议
class HandoffProtocol:
    def decide_handoff(self, ai_output, confidence, context):
        """
        决定是否需要人工介入
        """
        # 规则1：置信度低于阈值 -> 升级人工
        if confidence < 0.7:
            return HandoffDecision.ESCALATE
        
        # 规则2：涉及敏感操作 -> 强制人工
        if context.get('sensitive_operation'):
            return HandoffDecision.ESCALATE
        
        # 规则3：用户明确要求 -> 升级人工
        if context.get('user_requested_human'):
            return HandoffDecision.ESCALATE
        
        # 规则4：连续3次AI建议被用户拒绝 -> 升级人工
        if context.get('consecutive_rejections', 0) >= 3:
            return HandoffDecision.ESCALATE
        
        return HandoffDecision.AUTO_COMPLETE
    
    def prepare_handoff_context(self, ai_output):
        """为人工准备完整的上下文"""
        return {
            'ai_response': ai_output.text,
            'confidence': ai_output.confidence,
            'sources': ai_output.sources,
            'alternative_answers': ai_output.alternatives,
            'user_history': get_user_history(),
            'system_state': get_current_system_state()
        }

-- 反馈数据表设计
CREATE TABLE ai_feedback (
    id BIGSERIAL PRIMARY KEY,
    session_id VARCHAR(64) NOT NULL,
    query_text TEXT NOT NULL,
    ai_response TEXT NOT NULL,
    user_rating INT CHECK (user_rating BETWEEN 1 AND 5),
    user_correction TEXT,              -- 用户修改后的内容
    user_comment TEXT,
    was_accepted BOOLEAN,              -- 用户是否直接采纳
    response_time_ms INT,
    model_used VARCHAR(32),
    created_at TIMESTAMP DEFAULT NOW(),
    INDEX idx_created_at (created_at),
    INDEX idx_rating (user_rating)
);

def generate_improvement_report():
    with db_connection() as conn:
        # 1. 低满意度问题识别
        low_rated = conn.query("""
            SELECT query_text, user_comment, model_used
            FROM ai_feedback
            WHERE user_rating <= 2
            AND created_at > NOW() - INTERVAL '30 days'
            GROUP BY query_text, user_comment, model_used
            ORDER BY COUNT(*) DESC
            LIMIT 20
        """)
        
        # 2. 调用大模型分析根因
        analysis = llm.invoke(f"""
        以下是最低评分的用户反馈，请分析共性模式：
        {low_rated}
        
        请输出：
        1. 主要问题分类（理解错误/信息不全/格式问题/其他）
        2. 改进建议
        3. 优先级排序
        """)
        
        # 3. 自动生成改进任务
        create_jira_issues(analysis)
        
        return analysis