一个能“想”、能“记”、能“做”的AI智能体，如果没有质量保障体系，就像一辆没有刹车和仪表盘的高性能跑车——跑得越快，风险越大。

在前三篇文章中，我们完成了AI智能体的架构设计、知识库构建和工具调用能力建设。一个能“想”、能“记”、能“做”的数字员工已经具备了上岗的基本条件。

但还有一个致命的问题没有解决：如何保证AI的输出质量？如何在大模型“胡说八道”时及时止损？

大模型的本质是概率模型，它永远存在产生错误输出的可能——这种错误在业界被称为“幻觉”（Hallucination）。研究显示，当前AI智能体单步任务成功率仅58%，多步任务成功率仅35%。在跨境电商场景中，一个错误的客服回复可能导致客户流失，一个错误的广告调价可能浪费数千美元预算，一个错误的库存同步可能引发断货或超卖。

质量保障不是锦上添花，而是生产级AI系统的生命线。

今天这篇文章，我们就来构建AI智能体的“品控体系”——包括输出质量评估、异常检测和人工兜底三大模块。

一、质量评估：给AI的“考试成绩单”

1.1 为什么需要自动化评估？

传统软件测试有明确的“对”与“错”——输入1+1，输出必须是2。但大模型的输出是自然语言，没有唯一正确的答案。同一个问题，可以有多种合理的回答方式。

这就带来了评估的挑战：如何用程序判断一段自然语言文本的质量？

答案是：用AI评估AI。这种范式被称为“LLM-as-a-Judge”（以大模型为裁判）。用一个能力更强的模型（如GPT-5.5）来评估目标模型的输出质量，在多个维度上打分。

在跨境电商场景中，我们需要从以下维度评估AI输出：

评估维度	说明	跨境电商示例
忠实度（Faithfulness）	回答是否基于检索到的知识，而非模型凭空编造	客服回答的退货政策是否与知识库一致
答案相关性（Answer Relevancy）	回答是否切题，有没有答非所问	用户问物流，AI有没有跑去聊产品参数
上下文召回（Contextual Recall）	是否遗漏了重要的上下文信息	回答是否覆盖了用户问题中的所有关键点
上下文精确度（Contextual Precision）	检索到的信息是否与问题相关	RAG检索到的文档片段是否真的有用
工具正确性（Tool Correctness）	Agent调用工具的参数和时机是否正确	调价时是否传对了SKU和价格
任务完成度（Task Completion）	多步任务是否全部完成	用户要“查订单+查库存”，AI是否都做了

1.2 基于Ragas的RAG质量评估

Ragas是目前最流行的RAG系统评估框架之一。它提供了丰富的评估指标，可以量化RAG系统的检索质量和生成质量。

from ragas import SingleTurnSample
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_recall,
    context_precision,
)
from ragas.llms import LangchainLLMWrapper
from langchain.chat_models import ChatOpenAI
import asyncio

class RAGQualityEvaluator:
    """
    基于Ragas的RAG输出质量评估器
    评估跨境电商AI客服的回答质量
    """
    
    def __init__(self, judge_model: str = "gpt-5.5"):
        # 使用GPT-5.5作为裁判模型
        self.evaluator_llm = LangchainLLMWrapper(
            ChatOpenAI(model=judge_model)
        )
        # 初始化评估指标
        self.metrics = [
            faithfulness,           # 忠实度：回答是否基于检索内容
            answer_relevancy,       # 答案相关性：是否切题
            context_recall,         # 上下文召回：是否覆盖关键信息
            context_precision,      # 上下文精确度：检索内容是否相关
        ]
    
    async def evaluate_single_response(
        self,
        user_input: str,
        response: str,
        retrieved_contexts: list,
        ground_truth: str = None
    ) -> dict:
        """
        评估单条AI回复的质量
        
        Args:
            user_input: 用户原始问题
            response: AI生成的回答
            retrieved_contexts: RAG检索到的知识片段列表
            ground_truth: 标准答案（如有）
        
        Returns:
            各维度的评分字典
        """
        # 构造Ragas评估样本
        sample = SingleTurnSample(
            user_input=user_input,
            response=response,
            retrieved_contexts=retrieved_contexts,
            reference=ground_truth,  # 可选
        )
        
        # 执行评估
        scores = {}
        for metric in self.metrics:
            score = await metric.single_turn_ascore(sample)
            scores[metric.name] = score
        
        # 计算综合质量分（加权平均）
        weights = {
            'faithfulness': 0.35,
            'answer_relevancy': 0.25,
            'context_recall': 0.20,
            'context_precision': 0.20,
        }
        overall_score = sum(
            scores.get(name, 0) * weights.get(name, 0) 
            for name in weights.keys()
        )
        scores['overall_quality'] = overall_score
        
        return scores

# 使用示例
async def main():
    evaluator = RAGQualityEvaluator()
    
    # 模拟一个跨境电商客服场景
    scores = await evaluator.evaluate_single_response(
        user_input="我的订单EB12345678显示已签收但我没收到货，怎么办？",
        response="很抱歉给您带来不便。根据系统记录，订单EB12345678已于7月15日签收，签收人为J. SMITH。建议您先与家人或邻居确认是否代收。如确认未收到，我们可以为您发起物流调查，调查周期约3-5个工作日。",
        retrieved_contexts=[
            "订单EB12345678于2026-07-15 14:23签收，签收人：J. SMITH",
            "未收到货处理流程：1.确认收货地址 2.联系快递公司 3.发起物流调查",
            "物流调查周期：3-5个工作日"
        ]
    )
    
    print(f"忠实度: {scores['faithfulness']:.2f}")
    print(f"答案相关性: {scores['answer_relevancy']:.2f}")
    print(f"综合质量分: {scores['overall_quality']:.2f}")
    
    # 如果综合质量分低于阈值，触发告警
    if scores['overall_quality'] < 0.7:
        print("⚠️ 警告：回答质量偏低，建议人工复核")

# asyncio.run(main())

1.3 基于DeepEval的Agent行为评估

DeepEval是另一个强大的开源LLM评估框架，特别适合评估AI Agent的行为。它支持Agent维度的评估指标，如任务完成度和工具正确性。

from deepeval import evaluate
from deepeval.metrics import (
    TaskCompletionMetric,
    ToolCorrectnessMetric,
    AnswerRelevancyMetric,
)
from deepeval.test_case import LLMTestCase

class AgentQualityEvaluator:
    """
    基于DeepEval的AI Agent行为评估器
    专门评估Agent的工具调用和任务完成情况
    """
    
    def __init__(self):
        # TaskCompletionMetric: 评估多步任务是否全部完成
        self.task_metric = TaskCompletionMetric(
            threshold=0.7,
            model="gpt-5.5"
        )
        # ToolCorrectnessMetric: 评估工具调用是否正确
        self.tool_metric = ToolCorrectnessMetric(
            threshold=0.8,
            model="gpt-5.5"
        )
        # AnswerRelevancyMetric: 评估回答相关性
        self.relevancy_metric = AnswerRelevancyMetric(
            threshold=0.7,
            model="gpt-5.5"
        )
    
    def evaluate_agent_action(
        self,
        user_input: str,
        agent_response: str,
        expected_tools: list,
        actual_tool_calls: list,
        expected_output: str = None
    ) -> dict:
        """
        评估Agent的一次行动
        
        Args:
            user_input: 用户输入
            agent_response: Agent的最终回复
            expected_tools: 期望调用的工具列表
            actual_tool_calls: 实际调用的工具列表
            expected_output: 期望的输出（可选）
        """
        test_case = LLMTestCase(
            input=user_input,
            actual_output=agent_response,
            expected_output=expected_output,
            # DeepEval支持通过additional_metadata传递工具调用信息
            additional_metadata={
                "expected_tools": expected_tools,
                "actual_tool_calls": actual_tool_calls,
            }
        )
        
        # 执行各维度评估
        self.task_metric.measure(test_case)
        self.tool_metric.measure(test_case)
        self.relevancy_metric.measure(test_case)
        
        return {
            "task_completion": self.task_metric.score,
            "tool_correctness": self.tool_metric.score,
            "answer_relevancy": self.relevancy_metric.score,
            "overall_pass": all([
                self.task_metric.is_successful(),
                self.tool_metric.is_successful(),
                self.relevancy_metric.is_successful(),
            ])
        }

# 使用示例
evaluator = AgentQualityEvaluator()
result = evaluator.evaluate_agent_action(
    user_input="把SKU-A100的价格调到25.99，然后查一下库存",
    agent_response="已完成价格调整，当前库存156件",
    expected_tools=["adjust_price", "query_inventory"],
    actual_tool_calls=["adjust_price", "query_inventory"],
)
print(f"任务完成度: {result['task_completion']}")
print(f"工具正确性: {result['tool_correctness']}")
print(f"整体通过: {result['overall_pass']}")

1.4 自定义评估指标：跨境电商专属

跨境电商场景中有许多业务特定的质量要求，通用指标无法覆盖。我们可以基于Rubric-based Evaluation（基于量规的评估）自定义指标。

from ragas.metrics import AspectCritic
from ragas.llms import LangchainLLMWrapper

class EcommerceCustomMetrics:
    """
    跨境电商专属评估指标
    """
    
    def __init__(self, evaluator_llm):
        self.llm = evaluator_llm
        
        # 指标1：合规性——回答是否遵守平台政策
        self.compliance_metric = AspectCritic(
            name="compliance",
            llm=self.llm,
            definition="""评估该客服回答是否完全符合亚马逊的客户服务政策。
            检查要点：
            1. 是否明确告知用户退货政策
            2. 是否承诺了无法兑现的时效
            3. 是否使用了禁止的措辞（如"保证退款"）
            评分标准：1-5分，5分表示完全合规"""
        )
        
        # 指标2：专业性——回答是否专业、有温度
        self.professionalism_metric = AspectCritic(
            name="professionalism",
            llm=self.llm,
            definition="""评估该客服回答的专业程度和用户体验。
            检查要点：
            1. 语气是否礼貌、 empathetic（共情）
            2. 信息是否清晰、有条理
            3. 是否提供了明确的下一步行动指引
            评分标准：1-5分"""
        )
        
        # 指标3：多语言准确性（针对非英语市场）
        self.multilingual_accuracy = AspectCritic(
            name="multilingual_accuracy",
            llm=self.llm,
            definition="""评估小语种（西班牙语/德语/法语等）回答的准确性。
            检查要点：
            1. 语法是否正确
            2. 术语翻译是否准确（尤其是跨境电商术语）
            3. 是否存在文化不适宜的表述
            评分标准：1-5分"""
        )
    
    async def evaluate(self, user_input: str, response: str) -> dict:
        """执行跨境电商专属评估"""
        sample = SingleTurnSample(
            user_input=user_input,
            response=response,
        )
        
        scores = {
            'compliance': await self.compliance_metric.single_turn_ascore(sample),
            'professionalism': await self.professionalism_metric.single_turn_ascore(sample),
            'multilingual_accuracy': await self.multilingual_accuracy.single_turn_ascore(sample),
        }
        
        # 任何一项低于3分（满分5分）都需要人工复核
        scores['need_human_review'] = any(score < 3.0 for score in scores.values())
        
        return scores

二、异常检测：在“坏事发生前”拉响警报

评估是事后分析，异常检测是实时监控。我们要在AI输出的过程中，实时判断是否存在异常，及时拉响警报。

2.1 幻觉检测（Hallucination Detection）

幻觉是AI Agent最危险的故障模式——模型“自信满满”地输出错误信息。在跨境电商中，一个关于退货政策的幻觉可能导致客服给出错误指引，引发客户投诉甚至平台处罚。

幻觉检测的核心思路是：将AI的回答拆解为原子陈述（Atomic Claims），然后逐一验证每个陈述是否可在检索到的知识中找到依据。

import re
from typing import List, Dict

class HallucinationDetector:
    """
    跨境电商AI回复的幻觉检测器
    基于"陈述分解+事实核查"的方法
    """
    
    def __init__(self, llm_client, knowledge_retriever):
        self.llm = llm_client
        self.retriever = knowledge_retriever
    
    def extract_claims(self, response: str) -> List[str]:
        """
        将AI的回答拆解为原子陈述
        每个陈述是一个独立的、可验证的事实断言
        """
        prompt = f"""
        请将以下文本拆解为独立的原子陈述（事实断言）。
        每个陈述应该是可以用"是/否"或具体数据验证的独立事实。
        只输出陈述列表，每行一个。
        
        文本：{response}
        
        陈述：
        """
        claims_text = self.llm.generate(prompt)
        claims = [c.strip() for c in claims_text.split('\n') if c.strip()]
        return claims
    
    def verify_claim(self, claim: str, contexts: List[str]) -> Dict:
        """
        验证单个陈述是否可在上下文中找到依据
        使用NLI（自然语言推理）或LLM判断
        """
        prompt = f"""
        判断以下陈述是否可以从提供的上下文中得到支持。
        上下文：{contexts}
        
        陈述：{claim}
        
        请回答：
        - SUPPORTED: 陈述完全被上下文支持
        - PARTIAL: 陈述部分被支持，但有关键信息缺失
        - CONTRADICTED: 陈述与上下文矛盾
        - UNVERIFIABLE: 上下文中没有相关信息可以验证
        
        只输出一个标签。
        """
        result = self.llm.generate(prompt).strip()
        return {
            'claim': claim,
            'verdict': result,
            'is_hallucination': result in ['CONTRADICTED', 'UNVERIFIABLE']
        }
    
    def detect(self, response: str, retrieved_contexts: List[str]) -> Dict:
        """
        检测回答中的幻觉
        
        Returns:
            {
                'total_claims': 10,
                'hallucinated_claims': 2,
                'hallucination_rate': 0.2,
                'is_hallucinating': True,  # 幻觉率超过阈值
                'details': [...]
            }
        """
        # Step 1: 拆解为原子陈述
        claims = self.extract_claims(response)
        
        # Step 2: 逐一验证
        results = []
        for claim in claims:
            result = self.verify_claim(claim, retrieved_contexts)
            results.append(result)
        
        # Step 3: 统计幻觉
        hallucinated = [r for r in results if r['is_hallucination']]
        hallucination_rate = len(hallucinated) / len(claims) if claims else 0
        
        # Step 4: 判断是否触发告警（幻觉率>15%或存在矛盾陈述）
        is_hallucinating = (
            hallucination_rate > 0.15 or 
            any(r['verdict'] == 'CONTRADICTED' for r in results)
        )
        
        return {
            'total_claims': len(claims),
            'hallucinated_claims': len(hallucinated),
            'hallucination_rate': hallucination_rate,
            'is_hallucinating': is_hallucinating,
            'details': results
        }

# 使用示例
detector = HallucinationDetector(llm_client, retriever)
result = detector.detect(
    response="您的订单EB12345678已发货，预计7月22日送达。根据我们的政策，所有退货都需在30天内申请。",
    retrieved_contexts=[
        "订单EB12345678状态：已发货，预计送达日期2026-07-22",
        "退货政策：30天内可申请退货，需保持商品原状"
    ]
)
if result['is_hallucinating']:
    print(f"⚠️ 检测到幻觉！幻觉率: {result['hallucination_rate']:.1%}")
    for detail in result['details']:
        if detail['is_hallucination']:
            print(f"  - 可疑陈述: {detail['claim']} ({detail['verdict']})")

2.2 置信度打分与动态阈值

不是所有幻觉都一样严重。一个关于“预计送达日期”的幻觉，和一个关于“退货政策”的幻觉，风险等级完全不同。

class ConfidenceScorer:
    """
    跨境电商AI回复的置信度打分器
    不同维度的异常有不同的风险权重
    """
    
    # 跨境电商场景的风险等级定义
    RISK_WEIGHTS = {
        'price': 1.0,           # 涉及价格的信息——最高风险
        'refund': 0.95,         # 涉及退款——极高风险
        'policy': 0.9,          # 涉及平台政策——高风险
        'inventory': 0.7,       # 涉及库存——中高风险
        'shipping_date': 0.6,   # 涉及物流时间——中风险
        'product_spec': 0.4,    # 涉及产品规格——中低风险
        'general': 0.2,         # 通用信息——低风险
    }
    
    def __init__(self, llm_client):
        self.llm = llm_client
    
    def classify_risk_topic(self, claim: str) -> str:
        """判断陈述涉及的话题类型"""
        # 实际可用分类模型，这里用关键词匹配简化
        keywords = {
            'price': ['价格', '降价', '涨价', '折扣', 'price', 'discount'],
            'refund': ['退款', '退货', '赔偿', 'refund', 'return'],
            'policy': ['政策', '规则', '规定', 'policy', 'rule'],
            'inventory': ['库存', '缺货', '补货', 'inventory', 'stock'],
            'shipping_date': ['送达', '发货', '物流', 'shipping', 'delivery'],
            'product_spec': ['规格', '参数', '尺寸', 'spec', 'size'],
        }
        for topic, words in keywords.items():
            if any(w in claim.lower() for w in words):
                return topic
        return 'general'
    
    def calculate_confidence(self, response: str, verification_results: List[Dict]) -> Dict:
        """
        计算综合置信度分数
        """
        total_weight = 0
        weighted_score = 0
        
        for result in verification_results:
            claim = result['claim']
            topic = self.classify_risk_topic(claim)
            weight = self.RISK_WEIGHTS.get(topic, 0.2)
            
            # 根据验证结果打分
            if result['verdict'] == 'SUPPORTED':
                claim_score = 1.0
            elif result['verdict'] == 'PARTIAL':
                claim_score = 0.5
            else:
                claim_score = 0.0
            
            total_weight += weight
            weighted_score += weight * claim_score
        
        overall_confidence = weighted_score / total_weight if total_weight > 0 else 0.5
        
        # 动态阈值：如果涉及高风险话题，阈值提高
        high_risk_topics = ['price', 'refund', 'policy']
        has_high_risk = any(
            self.classify_risk_topic(r['claim']) in high_risk_topics 
            for r in verification_results
        )
        threshold = 0.85 if has_high_risk else 0.7
        
        return {
            'confidence_score': overall_confidence,
            'threshold': threshold,
            'is_passing': overall_confidence >= threshold,
            'has_high_risk_content': has_high_risk,
            'action': 'pass' if overall_confidence >= threshold else 'block'
        }

2.3 日志追踪与可观测性

异常检测需要完整的执行轨迹作为分析依据。我们需要记录每一次AI决策的完整上下文。

import json
import uuid
from datetime import datetime
from typing import Dict, Any, List

class AgentTrace:
    """
    AI Agent执行轨迹记录器
    用于异常回溯和根因分析
    """
    
    def __init__(self, storage):
        self.storage = storage  # 可接入Elasticsearch、S3等
    
    def trace_agent_execution(
        self,
        session_id: str,
        user_input: str,
        agent_response: str,
        tool_calls: List[Dict],
        retrieved_contexts: List[str],
        quality_scores: Dict,
        anomaly_detection: Dict,
        final_action: str,  # 'pass', 'block', 'human_review'
    ) -> str:
        """
        记录一次完整的Agent执行轨迹
        """
        trace_id = str(uuid.uuid4())
        
        trace = {
            'trace_id': trace_id,
            'session_id': session_id,
            'timestamp': datetime.now().isoformat(),
            'user_input': user_input,
            'agent_response': agent_response,
            'tool_calls': tool_calls,
            'retrieved_contexts': retrieved_contexts,
            'quality_scores': quality_scores,
            'anomaly_detection': anomaly_detection,
            'final_action': final_action,
        }
        
        # 存储到持久化存储
        self.storage.save(f"traces/{trace_id}.json", trace)
        return trace_id
    
    def get_trace(self, trace_id: str) -> Dict:
        """查询执行轨迹"""
        return self.storage.load(f"traces/{trace_id}.json")
    
    def query_anomalies(self, time_range: tuple, min_severity: str = 'medium') -> List[Dict]:
        """查询指定时间范围内的异常记录"""
        # 实际实现中，使用Elasticsearch等搜索引擎
        pass

三、人工兜底：为AI系上“安全带”

评估和检测是“发现问题”，兜底是“解决问题”。当AI的输出质量不达标或检测到异常时，系统必须有能力将控制权交还给人类。

3.1 三级兜底策略

我们设计三级兜底策略，根据异常的严重程度采取不同的应对措施：

级别	触发条件	应对措施	适用场景
L1 - 自动修正	轻微异常（置信度0.6-0.7）	重新生成或补充检索	信息不完整、表述模糊
L2 - 人工审核	中等异常（置信度0.4-0.6）或高风险话题	推送到审核队列，人工确认	价格调整、退款申请
L3 - 完全拦截	严重异常（置信度<0.4）或检测到矛盾	拒绝输出，直接转人工	政策矛盾、严重幻觉

class HumanFallbackMechanism:
    """
    人工兜底机制
    三级兜底策略：自动修正 → 人工审核 → 完全拦截
    """
    
    def __init__(self, llm_client, review_queue, notifier):
        self.llm = llm_client
        self.review_queue = review_queue  # 审核任务队列
        self.notifier = notifier          # 通知服务（Slack/邮件）
    
    def handle_response(
        self,
        user_input: str,
        agent_response: str,
        quality_scores: Dict,
        anomaly_result: Dict,
        confidence_result: Dict,
        retrieved_contexts: List[str],
    ) -> Dict:
        """
        根据质量和异常检测结果，决定处理策略
        """
        confidence = confidence_result['confidence_score']
        is_hallucinating = anomaly_result.get('is_hallucinating', False)
        has_high_risk = confidence_result.get('has_high_risk_content', False)
        
        # === L3: 完全拦截 ===
        if confidence < 0.4 or is_hallucinating:
            return self._level3_block(user_input, agent_response, confidence_result)
        
        # === L2: 人工审核 ===
        if confidence < 0.6 or (has_high_risk and confidence < 0.7):
            return self._level2_human_review(
                user_input, agent_response, 
                quality_scores, confidence_result
            )
        
        # === L1: 自动修正 ===
        if confidence < 0.7:
            return self._level1_auto_correct(
                user_input, agent_response, retrieved_contexts
            )
        
        # === 正常通过 ===
        return {
            'action': 'pass',
            'final_response': agent_response,
            'confidence': confidence,
        }
    
    def _level1_auto_correct(self, user_input: str, response: str, contexts: List[str]) -> Dict:
        """
        L1: 自动修正——重新生成回答，增加更多上下文
        """
        prompt = f"""
        用户问题：{user_input}
        
        之前的回答不够完善，请基于以下更完整的上下文重新回答：
        {contexts}
        
        要求：回答要完整、准确、基于事实。
        """
        new_response = self.llm.generate(prompt)
        
        return {
            'action': 'auto_corrected',
            'final_response': new_response,
            'original_response': response,
            'confidence': 0.75,
        }
    
    def _level2_human_review(
        self, 
        user_input: str, 
        response: str, 
        quality_scores: Dict,
        confidence_result: Dict
    ) -> Dict:
        """
        L2: 人工审核——推送到审核队列，等待人工确认
        """
        review_id = str(uuid.uuid4())
        
        # 创建审核任务
        review_task = {
            'review_id': review_id,
            'user_input': user_input,
            'candidate_response': response,
            'quality_scores': quality_scores,
            'confidence_score': confidence_result['confidence_score'],
            'has_high_risk': confidence_result.get('has_high_risk_content', False),
            'created_at': datetime.now().isoformat(),
            'status': 'pending',
            'reviewer': None,
            'review_comment': None,
        }
        
        # 推送到审核队列
        self.review_queue.push(review_task)
        
        # 发送通知
        self.notifier.notify_slack(
            f"📋 需要人工审核\n"
            f"用户问题: {user_input[:100]}...\n"
            f"AI回答: {response[:100]}...\n"
            f"置信度: {confidence_result['confidence_score']:.2f}\n"
            f"审核链接: https://your-platform.com/review/{review_id}"
        )
        
        # 返回占位回复
        return {
            'action': 'human_review',
            'review_id': review_id,
            'final_response': "您的请求已收到，正在由客服专员处理，请稍候。",
            'confidence': confidence_result['confidence_score'],
            'status': 'pending_review'
        }
    
    def _level3_block(self, user_input: str, response: str, confidence_result: Dict) -> Dict:
        """
        L3: 完全拦截——拒绝输出，直接转人工
        """
        # 记录严重异常
        self._log_severe_anomaly(user_input, response, confidence_result)
        
        # 发送紧急通知
        self.notifier.notify_slack(
            f"🚨 【紧急】AI输出被完全拦截\n"
            f"用户问题: {user_input}\n"
            f"AI回答: {response}\n"
            f"置信度: {confidence_result['confidence_score']:.2f}\n"
            f"原因: 检测到严重幻觉或置信度过低"
        )
        
        return {
            'action': 'blocked',
            'final_response': "抱歉，系统暂时无法处理您的请求，已转接人工客服，请稍候。",
            'confidence': confidence_result['confidence_score'],
            'blocked_response': response,
        }
    
    def _log_severe_anomaly(self, user_input: str, response: str, confidence_result: Dict):
        """记录严重异常到审计日志"""
        # 实际实现中写入Elasticsearch或数据库
        pass

3.2 人工审核工作台

人工审核需要有一个友好的界面，让审核人快速判断AI的回答是否可接受。

class ReviewWorkflow:
    """
    人工审核工作流
    支持：批准、拒绝、修改后批准
    """
    
    def __init__(self, review_queue, agent_executor):
        self.review_queue = review_queue
        self.agent = agent_executor
    
    def get_pending_reviews(self, limit: int = 20) -> List[Dict]:
        """获取待审核任务列表"""
        return self.review_queue.pop(limit)
    
    def approve(self, review_id: str, modified_response: str = None) -> Dict:
        """
        审核通过
        Args:
            review_id: 审核任务ID
            modified_response: 审核人修改后的回答（可选）
        """
        task = self.review_queue.get(review_id)
        if not task:
            return {'error': '审核任务不存在'}
        
        final_response = modified_response or task['candidate_response']
        
        # 更新任务状态
        task['status'] = 'approved'
        task['reviewed_at'] = datetime.now().isoformat()
        task['final_response'] = final_response
        
        # 将最终回答返回给用户
        self._deliver_to_user(task['session_id'], final_response)
        
        # 将审核通过的回答加入反馈数据集，用于后续优化
        self._add_to_feedback_dataset(task)
        
        return {'status': 'approved', 'final_response': final_response}
    
    def reject(self, review_id: str, reason: str) -> Dict:
        """审核拒绝"""
        task = self.review_queue.get(review_id)
        if not task:
            return {'error': '审核任务不存在'}
        
        task['status'] = 'rejected'
        task['reviewed_at'] = datetime.now().isoformat()
        task['reject_reason'] = reason
        
        # 通知用户
        self._deliver_to_user(
            task['session_id'], 
            "您的请求已被客服处理，如有疑问请继续咨询。"
        )
        
        # 记录拒绝原因，用于后续优化
        self._log_rejection(task)
        
        return {'status': 'rejected', 'reason': reason}

四、完整的质量保障工作流

让我们把所有模块串联起来，形成一个完整的质量保障闭环：

class QualityAssurancePipeline:
    """
    AI智能体质量保障全管道
    串联：评估 → 检测 → 兜底
    """
    
    def __init__(
        self,
        evaluator: RAGQualityEvaluator,
        detector: HallucinationDetector,
        scorer: ConfidenceScorer,
        fallback: HumanFallbackMechanism,
        tracer: AgentTrace,
    ):
        self.evaluator = evaluator
        self.detector = detector
        self.scorer = scorer
        self.fallback = fallback
        self.tracer = tracer
    
    async def process(
        self,
        session_id: str,
        user_input: str,
        agent_response: str,
        tool_calls: List[Dict],
        retrieved_contexts: List[str],
    ) -> Dict:
        """
        完整质量保障流程
        """
        # 1. 质量评估
        quality_scores = await self.evaluator.evaluate_single_response(
            user_input=user_input,
            response=agent_response,
            retrieved_contexts=retrieved_contexts,
        )
        
        # 2. 幻觉检测
        anomaly_result = self.detector.detect(agent_response, retrieved_contexts)
        
        # 3. 置信度打分
        # 从幻觉检测结果中提取验证细节
        verification_results = anomaly_result.get('details', [])
        confidence_result = self.scorer.calculate_confidence(
            agent_response, verification_results
        )
        
        # 4. 兜底决策
        fallback_result = self.fallback.handle_response(
            user_input=user_input,
            agent_response=agent_response,
            quality_scores=quality_scores,
            anomaly_result=anomaly_result,
            confidence_result=confidence_result,
            retrieved_contexts=retrieved_contexts,
        )
        
        # 5. 记录执行轨迹
        trace_id = self.tracer.trace_agent_execution(
            session_id=session_id,
            user_input=user_input,
            agent_response=agent_response,
            tool_calls=tool_calls,
            retrieved_contexts=retrieved_contexts,
            quality_scores=quality_scores,
            anomaly_detection=anomaly_result,
            final_action=fallback_result['action'],
        )
        
        return {
            'trace_id': trace_id,
            'quality_scores': quality_scores,
            'anomaly_detection': anomaly_result,
            'confidence': confidence_result,
            'final_action': fallback_result['action'],
            'final_response': fallback_result['final_response'],
            'need_follow_up': fallback_result['action'] in ['human_review', 'blocked'],
        }

五、总结

本文我们为AI智能体构建了完整的“品控体系”：

1. 质量评估：使用Ragas和DeepEval框架，从忠实度、相关性、任务完成度等多维度量化AI输出质量
2. 异常检测：通过陈述分解+事实核查检测幻觉，通过置信度打分识别高风险内容
3. 人工兜底：三级兜底策略（自动修正→人工审核→完全拦截），确保高风险操作有“人”把关

这三层保障共同构成了AI智能体的“刹车系统”——让它在犯错时能被及时纠正，在不确定时能请求帮助，在危险时能被紧急制动。

至此，我们这个“能想、能记、能做、有品控”的跨境电商AI智能体，已经具备了完整的生产级能力。 下一篇，我们将进行系统上线的效果复盘，分享真实的业务数据和持续优化方向。