智能体应用让大语言模型自行决定解决问题的后续步骤。这种灵活性非常强大,但模型的黑盒特性使得难以预测对智能体某部分的调整会如何影响其他部分。要构建生产就绪的智能体,全面的测试至关重要。
测试智能体有几种方法:
单元测试 使用内存模拟对象独立测试智能体的小型、确定性部分,以便快速、确定地断言确切行为。
集成测试 使用真实网络调用测试智能体,以确认组件协同工作、凭证和模式匹配,以及延迟可接受。
智能体应用往往更依赖集成测试,因为它们将多个组件链接在一起,并且必须处理由于LLM的非确定性特性带来的不稳定性。
单元测试
模拟聊天模型
对于不需要API调用的逻辑,您可以使用内存存根来模拟响应。
LangChain 提供 GenericFakeChatModel 用于模拟文本响应。它接受一个响应迭代器(AIMessages 或字符串),每次调用返回一个。它支持常规和流式用法。
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
model = GenericFakeChatModel( messages = iter ([
AIMessage( content = "" , tool_calls = [ToolCall( name = "foo" , args = { "bar" : "baz" }, id = "call_1" )]),
"bar"
]))
model.invoke( "hello" )
# AIMessage(content='', ..., tool_calls=[{'name': 'foo', 'args': {'bar': 'baz'}, 'id': 'call_1', 'type': 'tool_call'}])
如果我们再次调用模型,它将返回迭代器中的下一个项目:
model.invoke( "hello, again!" )
# AIMessage(content='bar', ...)
InMemorySaver 检查点
为了在测试期间启用持久化,您可以使用 InMemorySaver 检查点。这允许您模拟多个轮次来测试依赖于状态的行为:
from langgraph.checkpoint.memory import InMemorySaver
agent = create_agent(
model,
tools = [],
checkpointer = InMemorySaver()
)
# 第一次调用
agent.invoke(HumanMessage( content = "I live in Sydney, Australia." ))
# 第二次调用:第一条消息被持久化(悉尼位置),因此模型返回 GMT+10 时间
agent.invoke(HumanMessage( content = "What's my local time?" ))
集成测试
许多智能体行为只有在使用真实LLM时才会出现,例如智能体决定调用哪个工具、如何格式化响应,或者提示修改是否影响整个执行轨迹。LangChain的 agentevals 包提供了专门设计用于使用实时模型测试智能体轨迹的评估器。
AgentEvals 允许您通过执行轨迹匹配 或使用LLM法官 来轻松评估智能体的轨迹(消息的确切序列,包括工具调用):
轨迹匹配 为给定输入硬编码参考轨迹,并通过逐步比较验证运行。 适用于测试明确定义的工作流,其中您了解预期行为。当您对应该调用哪些工具以及调用顺序有特定期望时使用此方法。这种方法具有确定性、快速且成本效益高,因为它不需要额外的LLM调用。
LLM作为法官 使用LLM定性验证智能体的执行轨迹。“法官”LLM根据提示标准(可以包括参考轨迹)审查智能体的决策。 更灵活,可以评估效率和适当性等细微方面,但需要LLM调用且确定性较低。当您希望评估智能体轨迹的整体质量和合理性,而没有严格的工具调用或顺序要求时使用。
安装 AgentEvals
或者,直接克隆 AgentEvals 仓库 。
轨迹匹配评估器
AgentEvals 提供 create_trajectory_match_evaluator 函数来将智能体的轨迹与参考轨迹进行匹配。有四种模式可供选择:
模式 描述 使用场景
strict相同顺序的消息和工具调用的精确匹配 测试特定序列(例如,策略查找在授权之前) unordered相同的工具调用可以以任何顺序出现 验证顺序无关紧要时的信息检索 subset智能体仅调用参考中的工具(无额外工具) 确保智能体不超过预期范围 superset智能体至少调用参考工具(允许额外工具) 验证至少执行了所需的最小操作
strict 模式确保轨迹包含相同顺序的相同消息和工具调用,尽管允许消息内容存在差异。当您需要强制执行特定的操作序列时,这非常有用,例如要求在授权操作之前进行策略查找。from langchain.agents import create_agent
from langchain.tools import tool
from langchain.messages import HumanMessage, AIMessage, ToolMessage
from agentevals.trajectory.match import create_trajectory_match_evaluator
@tool
def get_weather ( city : str ):
"""Get weather information for a city."""
return f "It's 75 degrees and sunny in { city } ."
agent = create_agent( "openai:gpt-4o" , tools = [get_weather])
evaluator = create_trajectory_match_evaluator(
trajectory_match_mode = "strict" ,
)
def test_weather_tool_called_strict ():
result = agent.invoke({
"messages" : [HumanMessage( content = "What's the weather in San Francisco?" )]
})
reference_trajectory = [
HumanMessage( content = "What's the weather in San Francisco?" ),
AIMessage( content = "" , tool_calls = [
{ "id" : "call_1" , "name" : "get_weather" , "args" : { "city" : "San Francisco" }}
]),
ToolMessage( content = "It's 75 degrees and sunny in San Francisco." , tool_call_id = "call_1" ),
AIMessage( content = "The weather in San Francisco is 75 degrees and sunny." ),
]
evaluation = evaluator(
outputs = result[ "messages" ],
reference_outputs = reference_trajectory
)
# {
# 'key': 'trajectory_strict_match',
# 'score': True,
# 'comment': None,
# }
assert evaluation[ "score" ] is True
unordered 模式允许以任何顺序进行相同的工具调用,这在您希望验证是否检索了特定信息但不关心顺序时很有帮助。例如,智能体可能需要检查城市的天气和事件,但顺序无关紧要。from langchain.agents import create_agent
from langchain.tools import tool
from langchain.messages import HumanMessage, AIMessage, ToolMessage
from agentevals.trajectory.match import create_trajectory_match_evaluator
@tool
def get_weather ( city : str ):
"""Get weather information for a city."""
return f "It's 75 degrees and sunny in { city } ."
@tool
def get_events ( city : str ):
"""Get events happening in a city."""
return f "Concert at the park in { city } tonight."
agent = create_agent( "openai:gpt-4o" , tools = [get_weather, get_events])
evaluator = create_trajectory_match_evaluator(
trajectory_match_mode = "unordered" ,
)
def test_multiple_tools_any_order ():
result = agent.invoke({
"messages" : [HumanMessage( content = "What's happening in SF today?" )]
})
# 参考显示了与实际执行顺序不同的工具调用顺序
reference_trajectory = [
HumanMessage( content = "What's happening in SF today?" ),
AIMessage( content = "" , tool_calls = [
{ "id" : "call_1" , "name" : "get_events" , "args" : { "city" : "SF" }},
{ "id" : "call_2" , "name" : "get_weather" , "args" : { "city" : "SF" }},
]),
ToolMessage( content = "Concert at the park in SF tonight." , tool_call_id = "call_1" ),
ToolMessage( content = "It's 75 degrees and sunny in SF." , tool_call_id = "call_2" ),
AIMessage( content = "Today in SF: 75 degrees and sunny with a concert at the park tonight." ),
]
evaluation = evaluator(
outputs = result[ "messages" ],
reference_outputs = reference_trajectory,
)
# {
# 'key': 'trajectory_unordered_match',
# 'score': True,
# }
assert evaluation[ "score" ] is True
superset 和 subset 模式匹配部分轨迹。superset 模式验证智能体至少调用了参考轨迹中的工具,允许额外的工具调用。subset 模式确保智能体没有调用超出参考范围的任何工具。from langchain.agents import create_agent
from langchain.tools import tool
from langchain.messages import HumanMessage, AIMessage, ToolMessage
from agentevals.trajectory.match import create_trajectory_match_evaluator
@tool
def get_weather ( city : str ):
"""Get weather information for a city."""
return f "It's 75 degrees and sunny in { city } ."
@tool
def get_detailed_forecast ( city : str ):
"""Get detailed weather forecast for a city."""
return f "Detailed forecast for { city } : sunny all week."
agent = create_agent( "openai:gpt-4o" , tools = [get_weather, get_detailed_forecast])
evaluator = create_trajectory_match_evaluator(
trajectory_match_mode = "superset" ,
)
def test_agent_calls_required_tools_plus_extra ():
result = agent.invoke({
"messages" : [HumanMessage( content = "What's the weather in Boston?" )]
})
# 参考仅需要 get_weather,但智能体可能调用额外工具
reference_trajectory = [
HumanMessage( content = "What's the weather in Boston?" ),
AIMessage( content = "" , tool_calls = [
{ "id" : "call_1" , "name" : "get_weather" , "args" : { "city" : "Boston" }},
]),
ToolMessage( content = "It's 75 degrees and sunny in Boston." , tool_call_id = "call_1" ),
AIMessage( content = "The weather in Boston is 75 degrees and sunny." ),
]
evaluation = evaluator(
outputs = result[ "messages" ],
reference_outputs = reference_trajectory,
)
# {
# 'key': 'trajectory_superset_match',
# 'score': True,
# 'comment': None,
# }
assert evaluation[ "score" ] is True
您还可以设置 tool_args_match_mode 属性和/或 tool_args_match_overrides 来自定义评估器如何考虑实际轨迹与参考轨迹中工具调用之间的相等性。默认情况下,只有具有相同参数且调用相同工具的工具调用才被视为相等。访问仓库 了解更多详情。
LLM作为法官评估器
您还可以使用LLM通过 create_trajectory_llm_as_judge 函数来评估智能体的执行路径。与轨迹匹配评估器不同,它不需要参考轨迹,但如果可用,也可以提供参考轨迹。
from langchain.agents import create_agent
from langchain.tools import tool
from langchain.messages import HumanMessage, AIMessage, ToolMessage
from agentevals.trajectory.llm import create_trajectory_llm_as_judge, TRAJECTORY_ACCURACY_PROMPT
@tool
def get_weather ( city : str ):
"""Get weather information for a city."""
return f "It's 75 degrees and sunny in { city } ."
agent = create_agent( "openai:gpt-4o" , tools = [get_weather])
evaluator = create_trajectory_llm_as_judge(
model = "openai:o3-mini" ,
prompt = TRAJECTORY_ACCURACY_PROMPT ,
)
def test_trajectory_quality ():
result = agent.invoke({
"messages" : [HumanMessage( content = "What's the weather in Seattle?" )]
})
evaluation = evaluator(
outputs = result[ "messages" ],
)
# {
# 'key': 'trajectory_accuracy',
# 'score': True,
# 'comment': 'The provided agent trajectory is reasonable...'
# }
assert evaluation[ "score" ] is True
如果您有参考轨迹,可以向提示中添加额外变量并传入参考轨迹。下面,我们使用预构建的 TRAJECTORY_ACCURACY_PROMPT_WITH_REFERENCE 提示并配置 reference_outputs 变量: evaluator = create_trajectory_llm_as_judge(
model = "openai:o3-mini" ,
prompt = TRAJECTORY_ACCURACY_PROMPT_WITH_REFERENCE ,
)
evaluation = judge_with_reference(
outputs = result[ "messages" ],
reference_outputs = reference_trajectory,
)
有关LLM如何评估轨迹的更多可配置性,请访问仓库 。
异步支持
所有 agentevals 评估器都支持 Python asyncio。对于使用工厂函数的评估器,可以通过在函数名中的 create_ 后添加 async 来获得异步版本。
from agentevals.trajectory.llm import create_async_trajectory_llm_as_judge, TRAJECTORY_ACCURACY_PROMPT
from agentevals.trajectory.match import create_async_trajectory_match_evaluator
async_judge = create_async_trajectory_llm_as_judge(
model = "openai:o3-mini" ,
prompt = TRAJECTORY_ACCURACY_PROMPT ,
)
async_evaluator = create_async_trajectory_match_evaluator(
trajectory_match_mode = "strict" ,
)
async def test_async_evaluation ():
result = await agent.ainvoke({
"messages" : [HumanMessage( content = "What's the weather?" )]
})
evaluation = await async_judge( outputs = result[ "messages" ])
assert evaluation[ "score" ] is True
LangSmith 集成
为了随时间跟踪实验,您可以将评估器结果记录到 LangSmith ,这是一个用于构建生产级LLM应用程序的平台,包括追踪、评估和实验工具。
首先,通过设置所需的环境变量来设置 LangSmith:
export LANGSMITH_API_KEY = "your_langsmith_api_key"
export LANGSMITH_TRACING = "true"
LangSmith 提供了两种主要的运行评估方法:pytest 集成和 evaluate 函数。
import pytest
from langsmith import testing as t
from agentevals.trajectory.llm import create_trajectory_llm_as_judge, TRAJECTORY_ACCURACY_PROMPT
trajectory_evaluator = create_trajectory_llm_as_judge(
model = "openai:o3-mini" ,
prompt = TRAJECTORY_ACCURACY_PROMPT ,
)
@pytest.mark.langsmith
def test_trajectory_accuracy ():
result = agent.invoke({
"messages" : [HumanMessage( content = "What's the weather in SF?" )]
})
reference_trajectory = [
HumanMessage( content = "What's the weather in SF?" ),
AIMessage( content = "" , tool_calls = [
{ "id" : "call_1" , "name" : "get_weather" , "args" : { "city" : "SF" }},
]),
ToolMessage( content = "It's 75 degrees and sunny in SF." , tool_call_id = "call_1" ),
AIMessage( content = "The weather in SF is 75 degrees and sunny." ),
]
# 将输入、输出和参考输出记录到 LangSmith
t.log_inputs({})
t.log_outputs({ "messages" : result[ "messages" ]})
t.log_reference_outputs({ "messages" : reference_trajectory})
trajectory_evaluator(
outputs = result[ "messages" ],
reference_outputs = reference_trajectory
)
使用 pytest 运行评估: pytest test_trajectory.py --langsmith-output
结果将自动记录到 LangSmith。
或者,您可以在 LangSmith 中创建一个数据集并使用 evaluate 函数: from langsmith import Client
from agentevals.trajectory.llm import create_trajectory_llm_as_judge, TRAJECTORY_ACCURACY_PROMPT
client = Client()
trajectory_evaluator = create_trajectory_llm_as_judge(
model = "openai:o3-mini" ,
prompt = TRAJECTORY_ACCURACY_PROMPT ,
)
def run_agent ( inputs ):
"""您的智能体函数,返回轨迹消息。"""
return agent.invoke(inputs)[ "messages" ]
experiment_results = client.evaluate(
run_agent,
data = "your_dataset_name" ,
evaluators = [trajectory_evaluator]
)
结果将自动记录到 LangSmith。
记录和重放 HTTP 调用
调用真实LLM API的集成测试可能很慢且昂贵,尤其是在CI/CD管道中频繁运行时。我们建议使用库来记录HTTP请求和响应,然后在后续运行中重放它们,而无需进行实际的网络调用。
您可以使用 vcrpy 来实现这一点。如果您使用 pytest,pytest-recording 插件 提供了一种简单的方法来以最少的配置启用此功能。请求/响应记录在磁带中,然后在后续运行中使用这些磁带来模拟真实的网络调用。
设置您的 conftest.py 文件以从磁带中过滤掉敏感信息:
import pytest
@pytest.fixture ( scope = "session" )
def vcr_config ():
return {
"filter_headers" : [
( "authorization" , "XXXX" ),
( "x-api-key" , "XXXX" ),
# ... 您想要屏蔽的其他头部
],
"filter_query_parameters" : [
( "api_key" , "XXXX" ),
( "key" , "XXXX" ),
],
}
然后配置您的项目以识别 vcr 标记:
[pytest]
markers =
vcr: record/replay HTTP via VCR
addopts = -- record-mode =once
--record-mode=once 选项在第一次运行时记录HTTP交互,并在后续运行中重放它们。
现在,只需用 vcr 标记装饰您的测试:
@pytest.mark.vcr ()
def test_agent_trajectory ():
# ...
第一次运行此测试时,您的智能体将进行真实的网络调用,pytest 将在 tests/cassettes 目录中生成一个磁带文件 test_agent_trajectory.yaml。后续运行将使用该磁带来模拟真实的网络调用,前提是智能体的请求与上一次运行相比没有变化。如果发生变化,测试将失败,您需要删除磁带并重新运行测试以记录新的交互。
当您修改提示、添加新工具或更改预期轨迹时,您保存的磁带将过时,您现有的测试将失败 。您应删除相应的磁带文件并重新运行测试以记录新的交互。