测试与评估

测试与评估是构建可靠 AI 应用的关键环节。LangChain 提供了多种工具和方法来测试 Agent 的功能正确性、响应质量和性能表现。

概述

完整的测试评估体系包括：

功能测试        → Agent 能否正确调用工具？
正确性测试      → 回答是否准确？
质量评估        → 回答是否流畅、有查？
性能测试        → 响应速度如何？
回归测试        → 修改后是否破坏原有功能？

测试前的准备

# 被测 Agent
from langgraph.prebuilt import create_agent
from langchain_openai import ChatOpenAI
from langchain.tools import tool

@tool
def add(a: float, b: float) -> float:
    """计算两个数字的和"""
    return a + b

@tool
def multiply(a: float, b: float) -> float:
    """计算两个数字的积"""
    return a * b

@tool
def get_weather(city: str) -> str:
    """获取城市天气"""
    weather_data = {
        "北京": "晴朗，25°C",
        "上海": "多云，22°C",
        "广州": "阵雨，28°C"
    }
    return weather_data.get(city, f"没有 {city} 的天气数据")

tools = [add, multiply, get_weather]

def create_test_agent():
    return create_agent(
        model=ChatOpenAI(model="gpt-4o-mini", temperature=0),
        tools=tools,
        system_prompt="你是一个计算和天气助手。"
    )

功能测试

基本功能测试

def test_agent_basic():
    """测试 Agent 基本功能"""
    agent = create_test_agent()
    
    # 简单对话
    result = agent.invoke({
        "messages": [("human", "你好")]
    })
    response = result["messages"][-1].content
    assert len(response) > 0, "Agent 应该有回应"
    print("✅ 基本对话测试通过")

def test_tool_calling():
    """测试工具调用"""
    agent = create_test_agent()
    
    # 测试加法
    result = agent.invoke({
        "messages": [("human", "计算 123 + 456")]
    })
    response = result["messages"][-1].content
    assert "579" in response, f"加法结果应包含 579，实际: {response}"
    print("✅ 工具调用测试通过")

def test_multiple_tools():
    """测试多工具协作"""
    agent = create_test_agent()
    
    result = agent.invoke({
        "messages": [("human", "计算 (10 + 20) * 3")]
    })
    response = result["messages"][-1].content
    assert "90" in response, f"计算结果应包含 90，实际: {response}"
    print("✅ 多工具协作测试通过")

# 运行测试
test_agent_basic()
test_tool_calling()
test_multiple_tools()

边缘情况测试

def test_edge_cases():
    """测试边缘情况"""
    agent = create_test_agent()
    
    test_cases = [
        ("空输入测试", ""),
        ("特殊字符", "@#$%^&*"),
        ("超长输入", "你好 " * 1000),
        ("混合语言", "Hello 你好 Bonjour"),
    ]
    
    for name, input_text in test_cases:
        try:
            result = agent.invoke({
                "messages": [("human", input_text)]
            })
            print(f"✅ {name}: Agent 正常处理")
        except Exception as e:
            print(f"❌ {name}: 出错 - {e}")

test_edge_cases()

错误处理测试

def test_error_handling():
    """测试错误处理"""
    agent = create_test_agent()
    
    # 测试工具参数错误
    result = agent.invoke({
        "messages": [("human", "计算 a + b")]
    })
    response = result["messages"][-1].content
    # Agent 应优雅处理参数缺失
    print(f"错误处理响应: {response[:100]}")
    
    # 测试无此城市
    result = agent.invoke({
        "messages": [("human", "东京的天气怎么样？")]
    })
    response = result["messages"][-1].content
    assert "没有" in response or "东京" in response
    print("✅ 错误处理测试通过")

test_error_handling()

评估框架

基于参考答案的评估

def evaluate_answers(agent, test_cases: list):
    """
    评估 Agent 回答的正确性
    test_cases: [(问题, 预期答案关键词), ...]
    """
    results = []
    
    for question, expected_keywords in test_cases:
        result = agent.invoke({
            "messages": [("human", question)]
        })
        response = result["messages"][-1].content
        
        # 检查是否包含所有预期关键词
        matches = all(kw in response for kw in expected_keywords)
        results.append({
            "question": question,
            "response": response[:100],
            "pass": matches
        })
        
        status = "✅" if matches else "❌"
        print(f"{status} Q: {question[:30]}...")
    
    # 统计
    pass_count = sum(1 for r in results if r["pass"])
    print(f"\n总计: {pass_count}/{len(results)} 通过")
    
    return results

# 测试用例
test_cases = [
    ("100 + 200 等于多少？", ["300"]),
    ("3 乘以 5 是多少？", ["15"]),
    ("北京的天气怎么样？", ["北京", "晴朗"]),
]

agent = create_test_agent()
evaluate_answers(agent, test_cases)

基于 LLM 的自动评估

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

class LLMEvaluator:
    """使用 LLM 评估回答质量"""
    
    def __init__(self):
        self.evaluator_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    
    def evaluate(self, question: str, answer: str, criteria: str) -> dict:
        """评估单个回答"""
        prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一个 AI 助手回答质量评估员。
请根据以下标准评估回答质量（1-5分）：

评估标准：{criteria}

请在 JSON 中返回：
- score: 1-5 的分数
- reasoning: 评分的理由
- suggestions: 改进建议"""),
            ("human", "问题：{question}\n\n回答：{answer}")
        ])
        
        chain = prompt | self.evaluator_llm
        result = chain.invoke({
            "question": question,
            "answer": answer,
            "criteria": criteria
        })
        
        return {
            "question": question,
            "score": result.content,
            "criteria": criteria
        }
    
    def batch_evaluate(self, qa_pairs: list, criteria: str) -> list:
        """批量评估"""
        results = []
        for question, answer in qa_pairs:
            eval_result = self.evaluate(question, answer, criteria)
            results.append(eval_result)
        return results

# 使用示例
evaluator = LLMEvaluator()

qa_pairs = [
    ("100 + 200 = ?", "100 + 200 = 300"),
    ("Python 是什么？", "Python 是一种编程语言"),
]

criteria = "1. 答案是否准确 2. 是否完整 3. 是否清晰"
results = evaluator.batch_evaluate(qa_pairs, criteria)

回归测试

import json
from pathlib import Path

class RegressionTestSuite:
    """回归测试套件"""
    
    def __init__(self, agent):
        self.agent = agent
        self.test_history = []
    
    def add_test_case(self, name: str, input_msg: str, 
                      expected_keywords: list = None,
                      expected_tools: list = None):
        """添加测试用例"""
        self.test_history.append({
            "name": name,
            "input": input_msg,
            "expected_keywords": expected_keywords or [],
            "expected_tools": expected_tools or [],
        })
    
    def run_all(self) -> bool:
        """运行所有测试"""
        all_passed = True
        
        for test in self.test_history:
            try:
                result = self.agent.invoke({
                    "messages": [("human", test["input"])]
                })
                
                response = result["messages"][-1].content
                
                # 检查关键词
                keywords_ok = all(
                    kw in response 
                    for kw in test["expected_keywords"]
                )
                
                status = "✅" if keywords_ok else "❌"
                if not keywords_ok:
                    all_passed = False
                
                print(f"{status} {test['name']}")
                
            except Exception as e:
                print(f"❌ {test['name']}: {e}")
                all_passed = False
        
        return all_passed
    
    def save_test_suite(self, path: str):
        """保存测试套件"""
        Path(path).write_text(
            json.dumps(self.test_history, ensure_ascii=False, indent=2)
        )
    
    def load_test_suite(self, path: str):
        """加载测试套件"""
        self.test_history = json.loads(Path(path).read_text())

# 使用
suite = RegressionTestSuite(create_test_agent())
suite.add_test_case("加法测试", "123 + 456 = ?", expected_keywords=["579"])
suite.add_test_case("乘法测试", "12 * 34 = ?", expected_keywords=["408"])
suite.add_test_case("天气测试", "北京的天气", expected_keywords=["北京"])

suite.run_all()
suite.save_test_suite("test_suite.json")

性能测试

import time
import statistics

class PerformanceTester:
    """性能测试工具"""
    
    def __init__(self, agent, n_runs: int = 5):
        self.agent = agent
        self.n_runs = n_runs
    
    def measure_response_time(self, input_msg: str) -> dict:
        """测量响应时间"""
        times = []
        
        for i in range(self.n_runs):
            start = time.time()
            self.agent.invoke({
                "messages": [("human", input_msg)]
            })
            elapsed = time.time() - start
            times.append(elapsed)
        
        return {
            "min": min(times),
            "max": max(times),
            "avg": statistics.mean(times),
            "median": statistics.median(times),
            "p95": sorted(times)[int(len(times) * 0.95)],
        }
    
    def benchmark(self, scenarios: list) -> dict:
        """
        基准测试
        scenarios: [(名称, 输入消息), ...]
        """
        results = {}
        
        for name, input_msg in scenarios:
            timing = self.measure_response_time(input_msg)
            results[name] = timing
            print(f"\n📊 {name}:")
            print(f"   平均: {timing['avg']:.2f}s")
            print(f"   中位数: {timing['median']:.2f}s")
            print(f"   最小: {timing['min']:.2f}s")
            print(f"   最大: {timing['max']:.2f}s")
        
        return results

# 使用
tester = PerformanceTester(create_test_agent(), n_runs=3)

scenarios = [
    ("简单对话", "你好"),
    ("简单计算", "123 + 456 = ?"),
    ("多步计算", "计算 (10 + 20) * (5 + 3)"),
]

tester.benchmark(scenarios)

端到端测试

import asyncio

class E2ETester:
    """端到端测试"""
    
    def __init__(self):
        self.test_scenarios = []
    
    def add_scenario(self, name: str, messages: list, 
                     validation_fn=None):
        """
        添加测试场景
        messages: 消息列表 [user_msg1, user_msg2, ...]
        validation_fn: 验证函数，接收 messages 列表和 agent 返回
        """
        self.test_scenarios.append({
            "name": name,
            "messages": messages,
            "validation": validation_fn
        })
    
    async def test_conversation(self, agent, scenario: dict) -> bool:
        """测试完整对话流程"""
        name = scenario["name"]
        messages = scenario["messages"]
        validation = scenario["validation"]
        
        try:
            results = []
            for msg in messages:
                result = agent.invoke({
                    "messages": [("human", msg)]
                })
                results.append(result)
            
            # 验证
            if validation:
                passed = validation(results)
            else:
                passed = True
            
            status = "✅" if passed else "❌"
            print(f"{status} {name}")
            return passed
            
        except Exception as e:
            print(f"❌ {name}: 出错 - {e}")
            return False
    
    async def run_all(self, agent) -> dict:
        """运行所有场景"""
        results = {}
        for scenario in self.test_scenarios:
            passed = await self.test_conversation(agent, scenario)
            results[scenario["name"]] = passed
        return results

# 测试对话连贯性
def create_conversation_test():
    tester = E2ETester()
    
    # 场景：连续对话
    def check_conversation_flow(results):
        # 检查每次对话都有响应
        for i, result in enumerate(results):
            response = result["messages"][-1].content
            if not response:
                print(f"   步骤 {i+1}: 无响应")
                return False
        return True
    
    tester.add_scenario(
        name="连续对话测试",
        messages=[
            "你好，我叫小明",
            "你还记得我的名字吗？",
            "100 + 200 等于多少？",
            "我刚才问了什么计算题？"
        ],
        validation_fn=check_conversation_flow
    )
    
    return tester

# 运行端到端测试
async def main():
    from langgraph.checkpoint import InMemorySaver
    
    agent = create_agent(
        model=ChatOpenAI(model="gpt-4o-mini", temperature=0),
        tools=tools,
        checkpointer=InMemorySaver(),
        system_prompt="你是一个计算助手。"
    )
    
    tester = create_conversation_test()
    results = await tester.run_all(agent)
    
    passed = sum(1 for v in results.values() if v)
    total = len(results)
    print(f"\n端到端测试: {passed}/{total} 通过")

asyncio.run(main())

评估指标

指标	描述	测量方式
准确率	回答是否正确	与参考答案比对
召回率	是否覆盖所有关键信息	关键词匹配
响应时间	端到端延迟	时间测量
token 效率	完成任务所需的 token 数	统计 token 用量
工具调用成功率	工具调用的准确率	工具返回分析
用户满意度	回答质量和有用性	LLM 评估/人工评估

持续集成（CI）

# ci_test.py - CI/CD 测试脚本
import sys

def run_ci_tests():
    """CI 测试入口"""
    test_suite = RegressionTestSuite(create_test_agent())
    test_suite.add_test_case("加法", "1 + 1 = ?", ["2"])
    test_suite.add_test_case("乘法", "2 * 3 = ?", ["6"])
    
    passed = test_suite.run_all()
    
    if not passed:
        print("❌ CI 测试失败")
        sys.exit(1)
    else:
        print("✅ CI 测试通过")

if __name__ == "__main__":
    run_ci_tests()

测试报告

import json
from datetime import datetime

class TestReporter:
    """测试报告生成器"""
    
    def __init__(self):
        self.results = []
    
    def add_result(self, test_name: str, passed: bool, 
                   details: str = "", duration: float = 0):
        self.results.append({
            "test": test_name,
            "passed": passed,
            "details": details,
            "duration": duration,
            "timestamp": datetime.now().isoformat()
        })
    
    def generate_report(self) -> str:
        """生成 Markdown 报告"""
        total = len(self.results)
        passed = sum(1 for r in self.results if r["passed"])
        
        report = f"# 测试报告\n\n"
        report += f"**时间**: {datetime.now().strftime('%Y-%m-%d %H:%M')}\n\n"
        report += f"**通过率**: {passed}/{total} ({passed/total*100:.1f}%)\n\n"
        
        report += "## 测试详情\n\n"
        report += "| 测试 | 状态 | 耗时 |\n"
        report += "|------|------|------|\n"
        
        for r in self.results:
            status = "✅" if r["passed"] else "❌"
            report += f"| {r['test']} | {status} | {r['duration']:.2f}s |\n"
        
        return report
    
    def save_report(self, path: str):
        """保存报告"""
        with open(path, "w") as f:
            f.write(self.generate_report())
        print(f"报告已保存到: {path}")

# 使用
reporter = TestReporter()
# ... 运行测试并添加结果 ...
reporter.save_report("test_report.md")

最佳实践

1. 编写全面的测试套件：覆盖正常、异常和边界情况
2. 使用确定性设置：temperature=0 保证结果可复现
3. 自动化测试执行：集成到 CI/CD 流程
4. 跟踪回归：每次修改后运行回归测试
5. 监控性能基线：追踪响应时间的变化趋势
6. 评估真实场景：使用真实用户查询作为测试用例
7. 结合人工审核：自动化评估 + 人工抽样审核

下一步

• Callbacks 与追踪：使用 LangSmith 追踪和调试 Agent
• RAG 应用设计：测试和评估 RAG 系统
• Streaming 流式输出：测试流式输出场景
• Memory 记忆：测试记忆功能的正确性