一、为什么离线评测比在线评测更重要？

在跨境电商AI Agent上线之前，你面临一个尴尬局面：没有真实用户流量，无法通过A/B测试或线上监控来验证Agent效果。而一旦带着未经验证的Agent直接上线，后果可能是灾难性的——编造价格、误调预算、用西班牙语回复日本客户……这些事故在业界都有真实案例。

离线评测就是在真实用户到来之前，用标准化的方式模拟用户请求，全面验证Agent的各项能力。 它不仅能帮你提前发现Bug，还能量化评估每次Prompt调整、工具优化、模型升级带来的效果变化，为上线决策提供数据支撑。

本文将从四个维度构建完整的离线评测体系：

1. 单步能力测试——验证工具调用、意图识别等原子能力
2. 端到端场景测试——验证完整业务链路
3. 效果评估体系——量化回答质量
4. Benchmark数据集建设——可持续迭代的测试资产

---

二、单步能力测试：Agent的"单元测试"

2.1 工具调用准确性测试

工具调用是Agent的"手脚"，调用错了，后面全错。我们需要验证：给定用户输入，Agent能否正确选择工具并填入正确的参数。

import json
import pytest
from typing import Dict, List, Any
from langchain_core.tools import BaseTool

class ToolCallValidator:
    """工具调用验证器"""
    
    def __init__(self, agent_executor, expected_tool_calls: List[Dict]):
        self.agent = agent_executor
        self.expected = expected_tool_calls  # 期望的工具调用序列
    
    def validate(self, user_input: str) -> Dict[str, Any]:
        """验证Agent的工具调用是否符合预期"""
        result = self.agent.invoke({"input": user_input})
        
        # 提取实际工具调用记录
        actual_calls = self._extract_tool_calls(result)
        
        # 逐条对比
        passed = []
        failed = []
        
        for i, (actual, expected) in enumerate(zip(actual_calls, self.expected)):
            tool_match = actual.get("tool") == expected.get("tool")
            args_match = self._compare_args(actual.get("args"), expected.get("args"))
            
            if tool_match and args_match:
                passed.append({"step": i, "status": "pass"})
            else:
                failed.append({
                    "step": i,
                    "expected": expected,
                    "actual": actual,
                    "reason": f"工具不匹配" if not tool_match else f"参数不匹配"
                })
        
        return {
            "input": user_input,
            "total_steps": len(self.expected),
            "passed": len(passed),
            "failed": len(failed),
            "details": {"passed": passed, "failed": failed}
        }
    
    def _extract_tool_calls(self, result: Dict) -> List[Dict]:
        """从Agent执行结果中提取工具调用记录"""
        # LangChain的中间步骤格式: [(action, observation), ...]
        calls = []
        for step in result.get("intermediate_steps", []):
            action, observation = step
            calls.append({
                "tool": action.tool,
                "args": action.tool_input,
                "observation": observation
            })
        return calls
    
    def _compare_args(self, actual: Dict, expected: Dict) -> bool:
        """比较参数（支持模糊匹配）"""
        if not actual or not expected:
            return actual == expected
        
        for key, expected_value in expected.items():
            if key not in actual:
                return False
            # 支持正则表达式匹配
            if isinstance(expected_value, str) and expected_value.startswith("regex:"):
                import re
                pattern = expected_value.replace("regex:", "")
                if not re.match(pattern, str(actual[key])):
                    return False
            elif actual[key] != expected_value:
                return False
        return True

# 使用示例
def test_search_tool_calling():
    """测试商品搜索工具调用"""
    expected_calls = [
        {
            "tool": "search_products",
            "args": {"query": "4k monitor", "max_price": 1000}
        }
    ]
    
    validator = ToolCallValidator(agent_executor, expected_calls)
    result = validator.validate("帮我找几款1000美元以内的4K显示器")
    
    assert result["failed"] == 0, f"工具调用测试失败: {result}"

2.2 意图识别准确率测试

意图识别是Agent的"大脑"第一关，错了后面的推理都会跑偏。

class IntentAccuracyTester:
    """意图识别准确率测试"""
    
    def __init__(self, agent):
        self.agent = agent
    
    def run_test(self, test_cases: List[Dict]) -> Dict:
        """
        test_cases格式:
        [
            {"input": "我想退货", "expected_intent": "REFUND"},
            {"input": "查一下库存", "expected_intent": "STOCK_CHECK"},
        ]
        """
        results = {
            "total": len(test_cases),
            "correct": 0,
            "wrong": [],
            "accuracy": 0.0
        }
        
        for case in test_cases:
            # 通过调用Agent获取意图（需提前在Prompt中让Agent输出意图标签）
            response = self.agent.invoke({"input": case["input"]})
            predicted = self._extract_intent(response)
            
            if predicted == case["expected_intent"]:
                results["correct"] += 1
            else:
                results["wrong"].append({
                    "input": case["input"],
                    "expected": case["expected_intent"],
                    "predicted": predicted
                })
        
        results["accuracy"] = results["correct"] / results["total"] * 100
        return results
    
    def _extract_intent(self, response: Dict) -> str:
        """从Agent回复中提取意图标签"""
        # 假设Agent在回复中包含了意图标签，或通过结构化输出获取
        output = response.get("output", "")
        # 解析意图的简单实现
        import re
        match = re.search(r'INTENT[:：]\s*(\w+)', output)
        return match.group(1) if match else "UNKNOWN"

2.3 集成Pytest实现自动化回归测试

将以上测试用例集成到CI/CD流水线，每次代码变更自动运行：

# test_agent.py
import pytest
from agent import agent_executor
from tool_call_validator import ToolCallValidator
from intent_tester import IntentAccuracyTester

class TestAgentToolCalls:
    """Agent工具调用测试套件"""
    
    @pytest.fixture
    def agent(self):
        return agent_executor
    
    def test_search_products(self, agent):
        """测试商品搜索场景"""
        expected = [{"tool": "search_products", "args": {"query": "wireless mouse"}}]
        validator = ToolCallValidator(agent, expected)
        result = validator.validate("帮我看看有没有无线鼠标")
        assert result["failed"] == 0
    
    def test_stock_check(self, agent):
        """测试库存查询场景"""
        expected = [{"tool": "check_stock", "args": {"product_id": "M-2024-001"}}]
        validator = ToolCallValidator(agent, expected)
        result = validator.validate("查一下M-2024-001还有多少库存")
        assert result["failed"] == 0
    
    def test_complex_multi_step(self, agent):
        """测试多步场景：搜索→查库存→推荐"""
        expected = [
            {"tool": "search_products", "args": {"query": "bluetooth speaker", "max_price": 200}},
            {"tool": "check_stock", "args": {"product_id": "regex:BTS-.+"}}
        ]
        validator = ToolCallValidator(agent, expected)
        result = validator.validate("帮我找200美金以内的蓝牙音箱，然后告诉我库存")
        assert result["failed"] == 0

# 运行: pytest test_agent.py -v

---

三、端到端场景测试：模拟真实业务流程

单步测试通过不代表端到端能跑通。端到端测试模拟真实用户从提问到获得最终答案的完整链路。

3.1 场景模板设计

from dataclasses import dataclass
from typing import List, Optional, Callable

@dataclass
class E2ETestCase:
    """端到端测试用例"""
    name: str                    # 场景名称
    user_input: str              # 用户输入
    expected_tool_sequence: List[str]  # 期望调用的工具序列
    expected_output_contains: Optional[List[str]] = None  # 期望输出包含的关键词
    expected_output_not_contains: Optional[List[str]] = None  # 不应包含的关键词
    custom_assertion: Optional[Callable] = None  # 自定义断言函数
    max_iterations: int = 5      # 最大迭代次数

class E2ETestRunner:
    """端到端测试运行器"""
    
    def __init__(self, agent_executor):
        self.agent = agent_executor
        self.results = []
    
    def run(self, test_cases: List[E2ETestCase]) -> Dict:
        """运行所有端到端测试"""
        summary = {"total": len(test_cases), "passed": 0, "failed": 0, "details": []}
        
        for case in test_cases:
            result = self._run_single(case)
            summary["details"].append(result)
            if result["status"] == "pass":
                summary["passed"] += 1
            else:
                summary["failed"] += 1
        
        summary["pass_rate"] = summary["passed"] / summary["total"] * 100
        return summary
    
    def _run_single(self, case: E2ETestCase) -> Dict:
        """执行单个端到端测试"""
        try:
            # 执行Agent
            response = self.agent.invoke(
                {"input": case.user_input},
                config={"max_iterations": case.max_iterations}
            )
            
            # 提取中间步骤
            intermediate_steps = response.get("intermediate_steps", [])
            tool_sequence = [step[0].tool for step in intermediate_steps]
            
            # 验证工具序列
            if tool_sequence != case.expected_tool_sequence:
                return {
                    "name": case.name,
                    "status": "fail",
                    "reason": f"工具序列不匹配: 期望 {case.expected_tool_sequence}, 实际 {tool_sequence}"
                }
            
            output = response.get("output", "")
            
            # 验证输出包含关键词
            if case.expected_output_contains:
                for keyword in case.expected_output_contains:
                    if keyword.lower() not in output.lower():
                        return {
                            "name": case.name,
                            "status": "fail",
                            "reason": f"输出缺少关键词: {keyword}"
                        }
            
            # 验证输出不包含关键词
            if case.expected_output_not_contains:
                for keyword in case.expected_output_not_contains:
                    if keyword.lower() in output.lower():
                        return {
                            "name": case.name,
                            "status": "fail",
                            "reason": f"输出包含禁止词: {keyword}"
                        }
            
            # 自定义断言
            if case.custom_assertion:
                if not case.custom_assertion(output, intermediate_steps):
                    return {
                        "name": case.name,
                        "status": "fail",
                        "reason": "自定义断言失败"
                    }
            
            return {"name": case.name, "status": "pass"}
            
        except Exception as e:
            return {"name": case.name, "status": "fail", "reason": str(e)}

# 定义测试用例
test_cases = [
    E2ETestCase(
        name="商品推荐场景",
        user_input="我想买一款200美金以内的蓝牙音箱，有推荐吗",
        expected_tool_sequence=["search_products"],
        expected_output_contains=["音箱", "$"],
        expected_output_not_contains=["null", "error"]
    ),
    E2ETestCase(
        name="库存查询场景",
        user_input="查一下库存，SKU是M-2024-001",
        expected_tool_sequence=["check_stock"],
        expected_output_contains=["库存", "件"],
        expected_output_not_contains=["0", "缺货"]  # 注意：这里仅作示例，实际库存可能确实为0
    ),
    E2ETestCase(
        name="复杂多步场景",
        user_input="找一款降噪耳机，然后告诉我有没有库存",
        expected_tool_sequence=["search_products", "check_stock"],
        expected_output_contains=["耳机", "库存"]
    )
]

# 运行测试
runner = E2ETestRunner(agent_executor)
results = runner.run(test_cases)
print(f"通过率: {results['pass_rate']:.1f}%")
print(f"通过: {results['passed']}, 失败: {results['failed']}")

---

四、效果评估体系：量化回答质量

工具调用对了，但回答质量差，用户照样不满意。效果评估关注的是"说得好不好"——准确性、完整性、友好度。

4.1 多维度打分机制

import re
from typing import Dict, List

class QualityScorer:
    """回答质量评分器"""
    
    def __init__(self, llm):
        self.llm = llm
    
    def score(self, user_input: str, agent_output: str, tool_results: List[Dict]) -> Dict:
        """
        从多个维度对Agent回答进行评分（0-100）
        """
        scores = {
            "accuracy": self._score_accuracy(agent_output, tool_results),
            "completeness": self._score_completeness(user_input, agent_output),
            "friendliness": self._score_friendliness(agent_output),
            "formatting": self._score_formatting(agent_output),
            "conciseness": self._score_conciseness(agent_output)
        }
        scores["overall"] = sum(scores.values()) / len(scores)
        return scores
    
    def _score_accuracy(self, output: str, tool_results: List[Dict]) -> float:
        """准确性：检查输出数据是否与工具结果一致"""
        # 提取工具结果中的所有数字和名称
        tool_data = []
        for result in tool_results:
            tool_data.extend(re.findall(r'\$\d+|\d+件|[A-Z]{2,}-\d+', str(result)))
        
        # 检查输出中是否包含这些数据
        if not tool_data:
            return 70.0  # 无数据可比时给中等分
        
        matched = 0
        for data in tool_data:
            if data in output:
                matched += 1
        
        return (matched / len(tool_data)) * 100
    
    def _score_completeness(self, user_input: str, output: str) -> float:
        """完整性：是否回答了用户所有问题"""
        # 用LLM判断（简化版用关键词）
        questions = re.findall(r'[？?]', user_input)
        answer_indicators = ["是", "有", "可以", "能够", "查询", "结果", "如下"]
        
        if not questions:
            return 90.0
        
        # 检查回答中是否包含多个指示词
        indicator_count = sum(1 for ind in answer_indicators if ind in output)
        return min(100, (indicator_count / len(answer_indicators)) * 100 + 20)
    
    def _score_friendliness(self, output: str) -> float:
        """友好度：语气和礼貌程度"""
        polite_words = ["您好", "请", "谢谢", "感谢", "欢迎", "祝"]
        friendly_score = sum(1 for word in polite_words if word in output)
        
        # 检查是否有生硬的命令式语句
        if "你必须" in output or "你要" in output:
            friendly_score -= 2
        
        return min(100, (friendly_score / len(polite_words)) * 100 + 30)
    
    def _score_formatting(self, output: str) -> float:
        """格式化：是否结构清晰"""
        # 检查是否有列表、分段、标题等
        score = 60.0  # 基础分
        
        if "\n" in output:
            score += 10
        if re.search(r'[1-9][.、]', output):  # 编号列表
            score += 10
        if re.search(r'[-*•]', output):  # 项目符号
            score += 10
        if re.search(r'【.*?】|\[.*?\]', output):  # 加粗标题
            score += 10
        
        return min(100, score)
    
    def _score_conciseness(self, output: str) -> float:
        """简洁度：是否冗长"""
        word_count = len(output)
        if word_count < 50:
            return 100.0
        elif word_count < 150:
            return 80.0
        elif word_count < 300:
            return 60.0
        elif word_count < 500:
            return 40.0
        else:
            return 20.0

# 使用示例
scorer = QualityScorer(llm)
scores = scorer.score(user_query, agent_output, tool_results)
print(f"质量评分: {scores}")

4.2 批量评测报告

class BenchmarkReporter:
    """批量评测报告生成器"""
    
    def __init__(self, scorer: QualityScorer):
        self.scorer = scorer
    
    def run_benchmark(self, test_dataset: List[Dict]) -> Dict:
        """
        在测试数据集上批量运行评分
        test_dataset: [{"input": "...", "expected": "..."}]
        """
        results = []
        
        for item in test_dataset:
            # 调用Agent
            response = self.agent.invoke({"input": item["input"]})
            output = response.get("output", "")
            tool_results = self._extract_tool_results(response)
            
            # 评分
            scores = self.scorer.score(item["input"], output, tool_results)
            
            results.append({
                "input": item["input"],
                "output": output,
                "scores": scores,
                "expected": item.get("expected", "")
            })
        
        # 汇总统计
        avg_scores = {}
        for key in results[0]["scores"].keys():
            avg_scores[key] = sum(r["scores"][key] for r in results) / len(results)
        
        return {
            "total": len(results),
            "average_scores": avg_scores,
            "details": results
        }

---

五、Benchmark数据集建设

离线评测的核心资产是测试数据集。没有标准化的数据集，评测就无从谈起。

5.1 数据集结构设计

# benchmark_dataset.json
{
    "version": "1.0",
    "created_at": "2026-07-17",
    "categories": {
        "product_search": {
            "description": "商品搜索类问题",
            "test_cases": [
                {
                    "id": "PS-001",
                    "input": "帮我找一款500元以内的机械键盘",
                    "expected_tool": "search_products",
                    "expected_params": {"query": "mechanical keyboard", "max_price": 500},
                    "expected_output_contains": ["键盘", "￥"]
                },
                {
                    "id": "PS-002",
                    "input": "有没有适合MacBook的USB-C扩展坞推荐",
                    "expected_tool": "search_products",
                    "expected_params": {"query": "USB-C hub MacBook"},
                    "expected_output_contains": ["扩展坞", "USB"]
                }
            ]
        },
        "stock_check": {
            "description": "库存查询类问题",
            "test_cases": [...]
        },
        "multi_step": {
            "description": "多步复杂任务",
            "test_cases": [...]
        }
    }
}

5.2 数据集的持续迭代

import json
import hashlib
from datetime import datetime

class BenchmarkManager:
    """Benchmark数据集管理器"""
    
    def __init__(self, dataset_path: str):
        self.path = dataset_path
        self.dataset = self._load()
    
    def _load(self) -> Dict:
        with open(self.path, 'r') as f:
            return json.load(f)
    
    def add_test_case(self, category: str, test_case: Dict):
        """新增测试用例"""
        # 生成唯一ID
        test_case["id"] = f"{category[:2]}-{hashlib.md5(test_case['input'].encode()).hexdigest()[:6]}"
        test_case["added_at"] = datetime.now().isoformat()
        
        self.dataset["categories"][category]["test_cases"].append(test_case)
        self._save()
    
    def _save(self):
        with open(self.path, 'w') as f:
            json.dump(self.dataset, f, ensure_ascii=False, indent=2)
    
    def get_stats(self) -> Dict:
        """获取数据集统计信息"""
        total = 0
        for cat, data in self.dataset["categories"].items():
            total += len(data["test_cases"])
        return {
            "total_cases": total,
            "categories": list(self.dataset["categories"].keys())
        }

---

六、完整评测流水线

将以上所有能力整合成一条完整的CI/CD流水线：

# .github/workflows/agent_benchmark.yml
name: Agent Benchmark Test

on:
  pull_request:
    paths:
      - 'agent/**'
      - 'prompts/**'

jobs:
  benchmark:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      
      - name: Setup Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'
      
      - name: Install dependencies
        run: pip install -r requirements.txt
      
      - name: Run unit tests
        run: pytest test_agent.py -v --junitxml=unit_test_report.xml
      
      - name: Run E2E tests
        run: python run_e2e_tests.py --output e2e_report.json
      
      - name: Run quality benchmark
        run: python run_benchmark.py --dataset benchmark_dataset.json --output quality_report.json
      
      - name: Check pass threshold
        run: |
          PASS_RATE=$(jq '.pass_rate' e2e_report.json)
          if (( $(echo "$PASS_RATE < 95.0" | bc -l) )); then
            echo "端到端测试通过率 $PASS_RATE% < 95%，阻断合并"
            exit 1
          fi
      
      - name: Upload reports
        uses: actions/upload-artifact@v3
        with:
          name: benchmark-reports
          path: |
            unit_test_report.xml
            e2e_report.json
            quality_report.json

---

七、总结

评测层级	评测内容	工具/方法	触发时机
单元测试	工具调用、意图识别	ToolCallValidator + Pytest	每次PR
E2E测试	完整业务流程	E2ETestRunner	每次PR
质量评测	回答准确性、友好度	QualityScorer	每周/模型升级
Benchmark	持续数据集积累	BenchmarkManager	持续迭代

核心原则：

1. 自动化优先——所有评测都应能无人工参与运行
2. 阈值卡点——在CI中设置通过率门槛（如≥95%），不达标则阻断合并
3. 数据集即资产——测试数据集要像代码一样版本化管理
4. 多维度评估——不能只看工具调用正确率，回答质量同样重要
5. 持续迭代——线上发现的问题要及时转化为测试用例

当你的评测流水线能稳定运行、通过率保持在95%以上时，Agent上线就不再是一场"开盲盒"式的赌博，而是一次有数据支撑的自信发布。