6.2 Test
本节介绍 LangChain 代理的测试方法，包括单元测试和集成测试两种策略。
概述
测试代理应用与测试传统软件有本质区别。由于代理行为的不确定性和对外部 LLM 的依赖，我们需要采用专门的测试策略：
┌─────────────────────────────────────────────────────────────┐
│ Agent 测试策略 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────┐ ┌─────────────────┐ │
│ │ 单元测试 │ │ 集成测试 │ │
│ │ (Unit Tests) │ │ (Integration) │ │
│ ├─────────────────┤ ├─────────────────┤ │
│ │ - Mock 模型响应 │ │ - 真实 LLM 调用 │ │
│ │ - 内存检查点 │ │ - 轨迹评估 │ │
│ │ - 快速反馈 │ │ - LLM 作为裁判 │ │
│ │ - CI/CD 友好 │ │ - 端到端验证 │ │
│ └─────────────────┘ └─────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘| 测试类型 | 目的 | 速度 | 成本 |
|---|---|---|---|
| 单元测试 | 验证逻辑正确性 | 快 | 低 |
| 集成测试 | 验证真实行为 | 慢 | 高 |
单元测试
Mock 聊天模型
LangChain 提供 GenericFakeChatModel 用于模拟 LLM 响应，无需网络调用：
python
from langchain_core.messages import AIMessage
from langchain_core.language_models import GenericFakeChatModel
# 创建模拟模型
fake_model = GenericFakeChatModel(
messages=iter([
AIMessage(content="我来帮你搜索天气信息"),
AIMessage(content="北京今天晴，25度"),
])
)
# 模型按顺序返回预设响应
response1 = fake_model.invoke([{"role": "user", "content": "查天气"}])
print(response1.content) # "我来帮你搜索天气信息"
response2 = fake_model.invoke([{"role": "user", "content": "结果呢"}])
print(response2.content) # "北京今天晴，25度"模拟工具调用
对于带工具的代理，需要模拟工具调用响应：
python
from langchain_core.messages import AIMessage, ToolCall
# 创建带工具调用的响应
tool_response = AIMessage(
content="",
tool_calls=[
ToolCall(
id="call_123",
name="weather_search",
args={"city": "北京"}
)
]
)
fake_model = GenericFakeChatModel(
messages=iter([
tool_response,
AIMessage(content="北京今天晴，温度25度"),
])
)状态持久化测试
使用 InMemorySaver 测试多轮对话的状态保持：
python
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END
# 创建带检查点的图
checkpointer = InMemorySaver()
graph = StateGraph(MessagesState)
# ... 添加节点和边 ...
agent = graph.compile(checkpointer=checkpointer)
# 测试多轮对话
config = {"configurable": {"thread_id": "test-thread-1"}}
# 第一轮
result1 = agent.invoke(
{"messages": [{"role": "user", "content": "记住我叫小明"}]},
config=config
)
# 第二轮 - 验证状态保持
result2 = agent.invoke(
{"messages": [{"role": "user", "content": "我叫什么名字?"}]},
config=config
)
# 断言状态被正确保持
assert "小明" in result2["messages"][-1].content完整单元测试示例
python
# tests/test_agent.py
import pytest
from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.language_models import GenericFakeChatModel
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END
class TestWeatherAgent:
"""天气代理单元测试"""
def setup_method(self):
"""每个测试前设置 Mock"""
self.fake_responses = [
AIMessage(content="正在查询天气..."),
AIMessage(content="北京：晴，25°C"),
]
self.fake_model = GenericFakeChatModel(
messages=iter(self.fake_responses)
)
def test_basic_query(self):
"""测试基本查询"""
response = self.fake_model.invoke([
HumanMessage(content="查询北京天气")
])
assert response.content == "正在查询天气..."
def test_multi_turn_conversation(self):
"""测试多轮对话"""
checkpointer = InMemorySaver()
# 创建简单图
def agent_node(state: MessagesState):
return {"messages": [self.fake_model.invoke(state["messages"])]}
graph = StateGraph(MessagesState)
graph.add_node("agent", agent_node)
graph.add_edge(START, "agent")
graph.add_edge("agent", END)
agent = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "test-1"}}
# 第一轮
result1 = agent.invoke(
{"messages": [HumanMessage(content="你好")]},
config=config
)
# 验证响应
assert len(result1["messages"]) == 2
assert result1["messages"][-1].content == "正在查询天气..."
def test_state_persistence(self):
"""测试状态持久化"""
checkpointer = InMemorySaver()
# 构建图并测试状态
# ... 具体实现 ...
pass
# 运行测试
# pytest tests/test_agent.py -v集成测试
集成测试使用真实 LLM，验证代理的实际行为。
安装评估包
bash
pip install agentevals轨迹匹配评估
agentevals 提供四种轨迹匹配模式：
┌─────────────────────────────────────────────────────────────┐
│ 轨迹匹配模式 │
├─────────────────────────────────────────────────────────────┤
│ │
│ Strict (严格) │
│ ├── 消息顺序必须完全一致 │
│ └── 工具调用顺序必须完全一致 │
│ │
│ Unordered (无序) │
│ ├── 相同的工具调用 │
│ └── 顺序可以不同 │
│ │
│ Subset (子集) │
│ ├── 代理只调用参考轨迹中的工具 │
│ └── 不允许额外调用 │
│ │
│ Superset (超集) │
│ ├── 代理至少调用参考轨迹中的工具 │
│ └── 允许额外调用 │
│ │
└─────────────────────────────────────────────────────────────┘轨迹匹配示例
python
from agentevals.trajectory import create_trajectory_match_evaluator
# 创建评估器
evaluator = create_trajectory_match_evaluator(
trajectory_match_mode="unordered" # 无序匹配
)
# 定义参考轨迹
reference_trajectory = [
{"role": "user", "content": "搜索北京天气并计算平均温度"},
{"role": "assistant", "tool_calls": [
{"name": "weather_search", "args": {"city": "北京"}}
]},
{"role": "tool", "content": "最高30度，最低20度"},
{"role": "assistant", "tool_calls": [
{"name": "calculator", "args": {"expression": "(30+20)/2"}}
]},
{"role": "tool", "content": "25"},
{"role": "assistant", "content": "北京平均温度25度"},
]
# 实际执行轨迹
actual_trajectory = agent.invoke(
{"messages": [{"role": "user", "content": "搜索北京天气并计算平均温度"}]}
)
# 评估
result = evaluator(
outputs=actual_trajectory["messages"],
reference_outputs=reference_trajectory
)
print(f"匹配分数: {result['score']}")
print(f"详细信息: {result['reasoning']}")LLM 作为裁判
当轨迹难以精确定义时，使用 LLM 进行定性评估：
python
from agentevals.trajectory import create_trajectory_llm_as_judge
# 创建 LLM 评估器
judge = create_trajectory_llm_as_judge(
model="gpt-4o",
rubric="""
评估代理是否正确完成了用户请求：
1. 是否理解了用户意图
2. 是否使用了正确的工具
3. 最终答案是否准确
4. 过程是否高效（没有不必要的步骤）
评分标准：
- 5分：完美完成
- 4分：基本完成，有小瑕疵
- 3分：部分完成
- 2分：有明显错误
- 1分：完全失败
"""
)
# 评估
result = judge(
inputs={"messages": [{"role": "user", "content": "帮我预订明天的机票"}]},
outputs=actual_trajectory["messages"]
)
print(f"评分: {result['score']}/5")
print(f"评语: {result['reasoning']}")异步评估
对于大规模测试，使用异步版本：
python
from agentevals.trajectory import create_async_trajectory_llm_as_judge
import asyncio
async def run_async_evaluation():
judge = create_async_trajectory_llm_as_judge(
model="gpt-4o",
rubric="评估代理行为是否正确..."
)
# 并行评估多个轨迹
tasks = [
judge(inputs=inp, outputs=out)
for inp, out in test_cases
]
results = await asyncio.gather(*tasks)
return results
# 运行
results = asyncio.run(run_async_evaluation())LangSmith 集成
Pytest 标记
使用 @pytest.mark.langsmith 自动记录测试结果：
python
import pytest
from langsmith import testing as ls_testing
@pytest.mark.langsmith
def test_agent_behavior():
"""测试代理行为 - 结果会自动上传到 LangSmith"""
# 执行代理
result = agent.invoke({
"messages": [{"role": "user", "content": "你好"}]
})
# 断言
assert "你好" in result["messages"][-1].content or \
"您好" in result["messages"][-1].content
# 返回输出供 LangSmith 记录
return result使用 Evaluate 函数
创建数据集并系统化测试：
python
from langsmith import Client
client = Client()
# 创建数据集
dataset = client.create_dataset("agent-test-cases")
# 添加测试用例
client.create_examples(
dataset_id=dataset.id,
inputs=[
{"messages": [{"role": "user", "content": "查询天气"}]},
{"messages": [{"role": "user", "content": "计算 1+1"}]},
{"messages": [{"role": "user", "content": "搜索新闻"}]},
],
outputs=[
{"expected": "天气信息"},
{"expected": "2"},
{"expected": "新闻列表"},
]
)
# 定义评估函数
def evaluate_agent(inputs):
return agent.invoke(inputs)
# 运行评估
results = client.evaluate(
evaluate_agent,
data=dataset.name,
evaluators=[
# 添加评估器
]
)
print(f"通过率: {results.summary['pass_rate']}")HTTP 调用管理
使用 vcrpy 录制回放
减少测试成本和延迟：
python
# conftest.py
import pytest
@pytest.fixture(scope="module")
def vcr_config():
return {
"filter_headers": [
"authorization",
"x-api-key",
],
"filter_query_parameters": [
"api_key",
],
"record_mode": "once", # 只录制一次
}python
# test_with_vcr.py
import pytest
@pytest.mark.vcr()
def test_agent_with_vcr():
"""
首次运行：录制 API 调用到 cassette 文件
后续运行：回放录制的响应
"""
result = agent.invoke({
"messages": [{"role": "user", "content": "测试"}]
})
assert result is not NoneCassette 文件结构
tests/
├── cassettes/
│ ├── test_agent_with_vcr.yaml
│ └── test_multi_turn.yaml
├── conftest.py
└── test_agent.py测试策略矩阵
| 场景 | 推荐方法 | 说明 |
|---|---|---|
| 本地开发 | 单元测试 + Mock | 快速反馈，无成本 |
| CI/CD | 单元测试 + VCR | 录制一次，重复使用 |
| 发布前 | 集成测试 | 真实 LLM 验证 |
| 回归测试 | 轨迹匹配 | 确保行为一致 |
| 质量评估 | LLM 裁判 | 定性分析 |
完整测试套件示例
python
# tests/conftest.py
import pytest
from langchain_core.language_models import GenericFakeChatModel
from langchain_core.messages import AIMessage
@pytest.fixture
def mock_model():
"""提供 Mock 模型"""
return GenericFakeChatModel(
messages=iter([
AIMessage(content="Mock 响应 1"),
AIMessage(content="Mock 响应 2"),
])
)
@pytest.fixture
def vcr_config():
"""VCR 配置"""
return {
"filter_headers": ["authorization"],
"record_mode": "once",
}python
# tests/test_agent.py
import pytest
from my_agent import create_agent
class TestAgentUnit:
"""单元测试"""
def test_basic_response(self, mock_model):
agent = create_agent(model=mock_model)
result = agent.invoke({"messages": [{"role": "user", "content": "hi"}]})
assert len(result["messages"]) > 0
def test_tool_selection(self, mock_model):
# 测试工具选择逻辑
pass
class TestAgentIntegration:
"""集成测试"""
@pytest.mark.vcr()
def test_real_conversation(self):
agent = create_agent()
result = agent.invoke({
"messages": [{"role": "user", "content": "你好"}]
})
assert result["messages"][-1].content
@pytest.mark.langsmith
def test_with_langsmith_tracking(self):
agent = create_agent()
result = agent.invoke({
"messages": [{"role": "user", "content": "测试"}]
})
return result最佳实践
测试金字塔
┌───────────┐
│ E2E 测试 │ ← 少量，验证关键流程
┌┴───────────┴┐
│ 集成测试 │ ← 适量，验证真实行为
┌┴─────────────┴┐
│ 单元测试 │ ← 大量，覆盖边界情况
└───────────────┘建议
- 优先单元测试：覆盖核心逻辑，快速反馈
- 使用 VCR：减少集成测试成本
- 定期真实测试：确保与 LLM 的兼容性
- 跟踪结果：利用 LangSmith 分析趋势
- 测试边界情况：空输入、超长输入、异常输入