Appearance
测试
前置阅读:Agent 实战指南 · 中间件概览
本节解决什么问题
AI Agent 的行为具有非确定性--相同的输入可能产生不同的输出。但没有测试的 Agent 在生产环境中就像没有安全绳的高空作业者。本节解决的核心问题是:如何在不调用真实 LLM API 的前提下,系统性地测试 create_agent 产生的 Agent 的工具逻辑、节点行为、图集成和中间件。
本节所有示例均未进行真实模型调用。 使用
FakeMessagesListChatModel返回预设消息序列,测试速度快、零成本、完全确定。
核心概念
必须深刻理解,不能跳过:
create_agent的输出结构是{"messages": [...]}。
create_agent返回CompiledStateGraph。调用agent.invoke({"messages": [...]})后,结果是一个字典,其中result["messages"]是完整的消息列表(包含 HumanMessage、AIMessage、ToolMessage 等)。最终答案在result["messages"][-1]。这与 0.x 时代的
AgentExecutor完全不同:旧 API 返回{"output": "...", "intermediate_steps": [...]},1.0 中 不存在 这两个键。
前端类比
测试 Agent 的思路与测试 React 组件非常相似:工具单测验证单个 @tool 函数(类似测试纯工具函数);节点单测验证图中的节点逻辑(类似测试 custom hook);图集成测试验证完整 Agent 行为(类似 Testing Library 的用户行为测试);LangSmith 评估则类似 E2E 测试(Cypress/Playwright),在真实环境中检验端到端行为。
原生语义:Agent 测试的关键在于用 Fake 模型替换真实 LLM。FakeMessagesListChatModel 是 langchain-core 提供的测试工具,它按顺序返回预设的 AIMessage 列表。当 Agent 的 model node 调用它时,它会依次返回预设消息--包括 tool_calls 和最终回复。这样你就能精确控制 Agent 的每一步行为,验证工具执行和中间件逻辑。
测试层次
准备:被测代码
以下是贯穿本节的被测代码。一个 EnviroNexus 环保知识库 Agent,包含工具和自定义节点。
python
# app.py
import os
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
@tool
def lookup_factor(keyword: str) -> str:
"""根据关键词查询环保因子信息。
Args:
keyword: 因子名称或别名(如 COD、化学需氧量)
"""
alias_map = {"化学需氧量": "COD", "SO2": "二氧化硫"}
normalized = alias_map.get(keyword, keyword)
factors = {
"COD": {"unit": "mg/L", "standard": "GB 11914-89"},
"二氧化硫": {"unit": "mg/m³", "standard": "HJ 482-2009"},
}
result = factors.get(normalized)
if not result:
return f"未找到因子: {keyword}"
return str(result)
@tool
def validate_evidence(factor_name: str) -> str:
"""校验因子的证据来源是否可信。
Args:
factor_name: 标准化的因子名称(如 COD、二氧化硫)
"""
evidence_db = {
"COD": "已校验: GB 11914-89 钾铬酸钾法",
"二氧化硫": "已校验: HJ 482-2009 甲醛吸收-副玫瑰苯胺分光光度法",
}
return evidence_db.get(factor_name, f"证据未校验: {factor_name}")
# 生产环境使用真实模型;测试环境传入 Fake 模型(见下文测试代码)
def build_agent(model=None):
"""构建 Agent,允许注入测试模型。"""
if model is None:
model = init_chat_model(os.environ["LLM_MODEL"])
return create_agent(
model=model,
tools=[lookup_factor, validate_evidence],
system_prompt=(
"你是 EnviroNexus 环保知识库助手。"
"用户询问因子时先用 lookup_factor 查询,"
"再用 validate_evidence 校验证据来源。"
"无证据校验的结论不得输出标准号。"
),
)1. 工具单元测试
工具是 Agent 中最容易测试的部分--它们是普通的 Python 函数。直接调用 .invoke() 传入参数字典即可。
python
# tests/unit/test_tools.py
import pytest
from app import lookup_factor, validate_evidence
class TestLookupFactor:
"""测试环保因子查询工具。"""
def test_lookup_cod_by_abbreviation(self):
result = lookup_factor.invoke({"keyword": "COD"})
assert "GB 11914-89" in result
assert "mg/L" in result
def test_lookup_cod_by_full_name(self):
"""中文全称应通过别名映射命中。"""
result = lookup_factor.invoke({"keyword": "化学需氧量"})
assert "GB 11914-89" in result
def test_lookup_unknown_factor(self):
result = lookup_factor.invoke({"keyword": "不存在的因子"})
assert "未找到" in result
@pytest.mark.parametrize("alias,expected_standard", [
("COD", "GB 11914-89"),
("化学需氧量", "GB 11914-89"),
("二氧化硫", "HJ 482-2009"),
("SO2", "HJ 482-2009"),
])
def test_alias_mapping(self, alias, expected_standard):
result = lookup_factor.invoke({"keyword": alias})
assert expected_standard in result
class TestValidateEvidence:
"""测试证据校验工具。"""
def test_validated_cod(self):
result = validate_evidence.invoke({"factor_name": "COD"})
assert "已校验" in result
assert "GB 11914-89" in result
def test_unvalidated_factor(self):
result = validate_evidence.invoke({"factor_name": "未知因子"})
assert "未校验" in result要点:工具单测不涉及任何模型调用,速度极快。EnviroNexus 的铁律是"无 evidence_refs 不得出标准结论",所以 validate_evidence 的测试必须覆盖校验通过和不通过两种路径。
2. 节点单元测试
create_agent 内部有 model node 和 tools node。当你构建自定义 LangGraph 图时,还会有自定义节点函数。节点单测验证这些函数在给定输入 state 下的行为。
python
# tests/unit/test_nodes.py
import pytest
from langchain.messages import AIMessage, ToolMessage
from app import lookup_factor
class TestToolsNode:
"""测试 create_agent 内部的 tools 节点行为。
create_agent 的 tools 节点接收含 tool_calls 的 AIMessage,
执行工具并返回 ToolMessage。
"""
def test_tool_execution_produces_tool_message(self):
"""工具执行后应生成 ToolMessage。"""
# 模拟模型产生的 tool_call
ai_message = AIMessage(
content="",
tool_calls=[{
"id": "call_001",
"name": "lookup_factor",
"args": {"keyword": "COD"},
}],
)
# 直接调用工具(等价于 tools node 对该 tool_call 的处理)
tool_result = lookup_factor.invoke({"keyword": "COD"})
# 构造预期的 ToolMessage
tool_message = ToolMessage(
content=tool_result,
tool_call_id="call_001",
)
assert "GB 11914-89" in tool_message.content
assert tool_message.tool_call_id == "call_001"
def test_tool_error_produces_error_message(self):
"""工具执行异常时应返回错误信息而非抛出异常。"""
ai_message = AIMessage(
content="",
tool_calls=[{
"id": "call_002",
"name": "lookup_factor",
"args": {"keyword": "不存在的因子"},
}],
)
tool_result = lookup_factor.invoke({"keyword": "不存在的因子"})
assert "未找到" in tool_result测试自定义 LangGraph 节点
如果你在 LangGraph 图中定义了自定义节点函数,可以直接测试它们:
python
# tests/unit/test_custom_nodes.py
from langgraph.graph import StateGraph, MessagesState, START, END
def route_by_intent(state: MessagesState) -> str:
"""根据最后一条消息内容路由到不同节点。"""
last_message = state["messages"][-1]
content = last_message.content if hasattr(last_message, "content") else str(last_message)
if "因子" in content or "标准" in content:
return "factor_agent"
return "general_chat"
class TestRouteByIntent:
"""测试意图路由节点。"""
def test_routes_to_factor_agent(self):
from langchain.messages import HumanMessage
state = {"messages": [HumanMessage(content="查询 COD 因子标准")]}
assert route_by_intent(state) == "factor_agent"
def test_routes_to_general_chat(self):
from langchain.messages import HumanMessage
state = {"messages": [HumanMessage(content="你好")]}
assert route_by_intent(state) == "general_chat"3. 图集成测试(Fake 模型)
图集成测试使用 FakeMessagesListChatModel 替换真实 LLM,验证完整的 Agent 执行流程。不产生任何真实模型调用费用。
FakeMessagesListChatModel 用法
python
from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
from langchain.messages import AIMessage
# 按顺序返回预设消息
fake_model = FakeMessagesListChatModel(responses=[
# 第 1 次模型调用:请求调用 lookup_factor 工具
AIMessage(
content="",
tool_calls=[{
"id": "call_1",
"name": "lookup_factor",
"args": {"keyword": "COD"},
}],
),
# 第 2 次模型调用:请求调用 validate_evidence 工具
AIMessage(
content="",
tool_calls=[{
"id": "call_2",
"name": "validate_evidence",
"args": {"factor_name": "COD"},
}],
),
# 第 3 次模型调用:生成最终回复
AIMessage(content="COD 的检测标准是 GB 11914-89,单位为 mg/L。证据来源已校验。"),
])完整集成测试
python
# tests/integration/test_agent_graph.py
import pytest
from langchain.messages import AIMessage, HumanMessage, ToolMessage
from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
from app import build_agent
@pytest.fixture
def fake_agent():
"""构建使用 Fake 模型的 Agent。未进行真实模型调用。"""
fake_model = FakeMessagesListChatModel(responses=[
AIMessage(
content="",
tool_calls=[{
"id": "call_1",
"name": "lookup_factor",
"args": {"keyword": "COD"},
}],
),
AIMessage(
content="",
tool_calls=[{
"id": "call_2",
"name": "validate_evidence",
"args": {"factor_name": "COD"},
}],
),
AIMessage(
content="COD 的检测标准是 GB 11914-89,单位为 mg/L。证据来源已校验。",
),
])
return build_agent(model=fake_model)
class TestAgentIntegration:
"""Agent 图集成测试 -- 使用 Fake 模型,未进行真实模型调用。"""
def test_agent_returns_messages_list(self, fake_agent):
"""create_agent 的结果应包含 messages 列表。"""
result = fake_agent.invoke({
"messages": [{"role": "user", "content": "查询 COD 的标准号"}],
})
# 核心断言:输出结构是 {"messages": [...]}
assert "messages" in result
assert isinstance(result["messages"], list)
def test_final_message_is_ai_message(self, fake_agent):
"""最后一条消息应是 AIMessage(最终回复)。"""
result = fake_agent.invoke({
"messages": [{"role": "user", "content": "查询 COD 的标准号"}],
})
final_message = result["messages"][-1]
assert isinstance(final_message, AIMessage)
assert "GB 11914-89" in final_message.content
def test_agent_calls_correct_tools_in_order(self, fake_agent):
"""Agent 应按顺序调用 lookup_factor 和 validate_evidence。"""
result = fake_agent.invoke({
"messages": [{"role": "user", "content": "查询 COD 的标准号并校验证据"}],
})
# 提取所有 tool_calls
tool_calls = []
for msg in result["messages"]:
if isinstance(msg, AIMessage) and msg.tool_calls:
for tc in msg.tool_calls:
tool_calls.append(tc["name"])
assert tool_calls == ["lookup_factor", "validate_evidence"]
def test_tool_messages_present(self, fake_agent):
"""工具调用后应生成对应的 ToolMessage。"""
result = fake_agent.invoke({
"messages": [{"role": "user", "content": "查询 COD 的标准号"}],
})
tool_messages = [
msg for msg in result["messages"]
if isinstance(msg, ToolMessage)
]
assert len(tool_messages) >= 1
# 第一个 ToolMessage 应包含 COD 因子信息
assert "GB 11914-89" in tool_messages[0].content
def test_full_message_history(self, fake_agent):
"""验证完整的消息历史。"""
result = fake_agent.invoke({
"messages": [{"role": "user", "content": "查询 COD"}],
})
messages = result["messages"]
# 1. HumanMessage(用户输入)
assert isinstance(messages[0], HumanMessage)
# 2. AIMessage with tool_calls(请求调用 lookup_factor)
assert isinstance(messages[1], AIMessage)
assert messages[1].tool_calls[0]["name"] == "lookup_factor"
# 3. ToolMessage(lookup_factor 的返回)
assert isinstance(messages[2], ToolMessage)
# 4. AIMessage with tool_calls(请求调用 validate_evidence)
assert isinstance(messages[3], AIMessage)
assert messages[3].tool_calls[0]["name"] == "validate_evidence"
# 5. ToolMessage(validate_evidence 的返回)
assert isinstance(messages[4], ToolMessage)
# 6. AIMessage(最终回复)
assert isinstance(messages[5], AIMessage)
assert "GB 11914-89" in messages[5].content测试错误处理
python
class TestAgentErrorHandling:
"""测试 Agent 对工具错误的处理 -- 未进行真实模型调用。"""
def test_agent_handles_unknown_factor(self):
"""Agent 在工具返回'未找到'时应给出合理回复。"""
fake_model = FakeMessagesListChatModel(responses=[
AIMessage(
content="",
tool_calls=[{
"id": "call_1",
"name": "lookup_factor",
"args": {"keyword": "未知污染物XYZ"},
}],
),
AIMessage(content="抱歉,知识库中未找到'未知污染物XYZ'的因子信息。"),
])
agent = build_agent(model=fake_model)
result = agent.invoke({
"messages": [{"role": "user", "content": "查询未知污染物XYZ的标准"}],
})
final = result["messages"][-1]
assert "未找到" in final.content or "未知" in final.content4. 中间件测试
中间件测试验证 middleware 在模型调用前后对 state 的修改是否正确。同样使用 Fake 模型。
python
# tests/unit/test_middleware.py
import pytest
from langchain.agents.middleware import PIIMiddleware
from langchain.messages import AIMessage, HumanMessage
from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
from app import build_agent
class TestPIIMiddleware:
"""测试 PII 脱敏中间件 -- 未进行真实模型调用。"""
def test_email_is_redacted_before_model(self):
"""PIIMiddleware 应在模型调用前脱敏邮箱。"""
fake_model = FakeMessagesListChatModel(responses=[
AIMessage(content="已收到您的咨询,请告诉我具体问题。"),
])
agent = build_agent(model=fake_model)
# 注意:这里需要替换 middleware,所以直接构建
from langchain.agents import create_agent
from app import lookup_factor, validate_evidence
agent = create_agent(
model=fake_model,
tools=[lookup_factor, validate_evidence],
system_prompt="你是客服助手。",
middleware=[
PIIMiddleware("email", strategy="redact"),
],
)
result = agent.invoke({
"messages": [{
"role": "user",
"content": "我的邮箱是 alice@example.com,帮我查询 COD 标准。",
}],
})
# 检查消息历史中是否进行了脱敏
# PIIMiddleware 会在 before_model 阶段修改输入消息
messages = result["messages"]
# 最终回复不应包含原始邮箱
final_content = messages[-1].content
assert "alice@example.com" not in final_content
def test_phone_is_masked(self):
"""PIIMiddleware 应对电话号码进行掩码处理。"""
fake_model = FakeMessagesListChatModel(responses=[
AIMessage(content="已处理您的请求。"),
])
from langchain.agents import create_agent
from app import lookup_factor
agent = create_agent(
model=fake_model,
tools=[lookup_factor],
system_prompt="你是客服助手。",
middleware=[
PIIMiddleware("phone", strategy="mask"),
],
)
result = agent.invoke({
"messages": [{
"role": "user",
"content": "我的电话是 13800138000,请帮我查询 COD。",
}],
})
final_content = result["messages"][-1].content
# 原始电话号码不应出现在最终输出中
assert "13800138000" not in final_content测试 SummarizationMiddleware
python
class TestSummarizationMiddleware:
"""测试摘要中间件 -- 未进行真实模型调用。"""
def test_summarization_triggers_on_threshold(self):
"""消息数超过阈值时应触发摘要。"""
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from app import lookup_factor
fake_model = FakeMessagesListChatModel(responses=[
# 模型会依次返回这些消息
AIMessage(content="回复 1"),
AIMessage(content="回复 2"),
AIMessage(content="回复 3"),
AIMessage(content="这是摘要:用户询问了环保因子。"),
AIMessage(content="最终回复"),
])
# 摘要也用 fake_model
summarizer = FakeMessagesListChatModel(responses=[
AIMessage(content="历史摘要:用户询问了环保因子信息。"),
])
agent = create_agent(
model=fake_model,
tools=[lookup_factor],
system_prompt="你是环保助手。",
middleware=[
SummarizationMiddleware(
model=summarizer,
trigger=("messages", 4), # 消息数超过 4 条时触发
keep=("messages", 2), # 保留最近 2 条
),
],
)
# 模拟多轮对话
messages = [
{"role": "user", "content": "查询 COD"},
{"role": "assistant", "content": "COD 是化学需氧量"},
{"role": "user", "content": "查询二氧化硫"},
{"role": "assistant", "content": "二氧化硫是 SO2"},
{"role": "user", "content": "再查一下 COD 的标准号"},
]
result = agent.invoke({"messages": messages})
# 验证结果包含 messages
assert "messages" in result
# 最终应有回复
assert len(result["messages"]) > 05. LangSmith 评估
LangSmith 评估使用数据集驱动的方式系统性测试 Agent。与上面的 Fake 模型测试不同,LangSmith 评估通常使用真实模型(但也可以用 Fake 模型)。
创建评估数据集
python
from langsmith import Client
client = Client()
dataset = client.create_dataset(
dataset_name="enviro_agent_test_v1",
description="EnviroNexus 环保知识库 Agent 测试用例",
)
examples = [
{
"input": {"messages": [{"role": "user", "content": "查询 COD 的标准号"}]},
"output": {"expected": "GB 11914-89"},
},
{
"input": {"messages": [{"role": "user", "content": "二氧化硫用什么方法检测"}]},
"output": {"expected": "HJ 482-2009"},
},
{
"input": {"messages": [{"role": "user", "content": "化学需氧量的单位是什么"}]},
"output": {"expected": "mg/L"},
},
{
"input": {"messages": [{"role": "user", "content": "你好"}]},
"output": {"expected_behavior": "打招呼,不调用工具"},
},
]
for example in examples:
client.create_example(
inputs=example["input"],
outputs=example["output"],
dataset_id=dataset.id,
)定义评估器
关键修正:评估器必须从 run.outputs["messages"] 读取结果,而不是 run.outputs["output"] 或 run.outputs["intermediate_steps"]。
python
from langchain.messages import AIMessage, ToolMessage
def correctness_evaluator(run, example) -> dict:
"""检查 Agent 最终回复是否包含预期答案。
create_agent 的输出结构是 {"messages": [...]},
最终回复在 run.outputs["messages"][-1]。
"""
messages = run.outputs.get("messages", [])
if not messages:
return {
"key": "correctness",
"score": 0.0,
"comment": "输出中没有 messages",
}
# 最终回复是最后一条 AIMessage 的 content
final_message = messages[-1]
prediction = (
final_message.content
if hasattr(final_message, "content")
else str(final_message)
)
expected = example.outputs.get("expected", "")
if expected:
score = 1.0 if expected in prediction else 0.0
return {
"key": "correctness",
"score": score,
"comment": f"预期: {expected}, 实际: {prediction[:100]}",
}
# 行为检查类用例
expected_behavior = example.outputs.get("expected_behavior", "")
return {
"key": "correctness",
"score": 1.0 if prediction else 0.0,
"comment": f"期望行为: {expected_behavior}",
}
def tool_usage_evaluator(run, example) -> dict:
"""检查工具使用是否合理。
从 messages 中提取 AIMessage 的 tool_calls 和 ToolMessage,
而非旧版的 intermediate_steps。
"""
messages = run.outputs.get("messages", [])
# 统计工具调用次数
tool_calls_count = sum(
1 for msg in messages
if isinstance(msg, AIMessage) and msg.tool_calls
)
tool_messages = [
msg for msg in messages if isinstance(msg, ToolMessage)
]
expected = example.outputs.get("expected", "")
if expected:
# 因子查询类问题应该调用工具
used_tool = tool_calls_count > 0
return {
"key": "tool_usage",
"score": 1.0 if used_tool else 0.0,
"comment": f"工具调用次数: {tool_calls_count}",
}
# 非因子查询问题不应调用工具
expected_behavior = example.outputs.get("expected_behavior", "")
if "不调用工具" in expected_behavior:
return {
"key": "tool_usage",
"score": 1.0 if tool_calls_count == 0 else 0.0,
"comment": f"预期不调用工具,实际调用 {tool_calls_count} 次",
}
return {
"key": "tool_usage",
"score": 1.0,
"comment": "无明确工具使用预期",
}
def evidence_validation_evaluator(run, example) -> dict:
"""EnviroNexus 专用:检查是否执行了证据校验。
铁律:无 evidence_refs 不得出标准结论。
"""
messages = run.outputs.get("messages", [])
tool_names_used = []
for msg in messages:
if isinstance(msg, AIMessage) and msg.tool_calls:
for tc in msg.tool_calls:
tool_names_used.append(tc["name"])
final_content = ""
if messages:
final_content = getattr(messages[-1], "content", "")
expected = example.outputs.get("expected", "")
if expected and any(
std in final_content for std in ["GB", "HJ", "DB"]
):
# 输出了标准号,检查是否校验了证据
validated = "validate_evidence" in tool_names_used
return {
"key": "evidence_validation",
"score": 1.0 if validated else 0.0,
"comment": (
"已校验证据" if validated
else "输出标准号但未校验证据来源"
),
}
return {
"key": "evidence_validation",
"score": 1.0,
"comment": "未输出标准号,无需证据校验",
}运行评估
python
from langsmith.evaluation import evaluate
def target(inputs: dict) -> dict:
"""执行 Agent 并返回结果。
返回 {"messages": [...]} 格式,评估器从此结构读取。
"""
# 生产评估使用真实模型;CI 中可用 Fake 模型
agent = build_agent() # 或 build_agent(model=fake_model)
result = agent.invoke(inputs)
return {"messages": result["messages"]}
results = evaluate(
target,
data="enviro_agent_test_v1",
evaluators=[
correctness_evaluator,
tool_usage_evaluator,
evidence_validation_evaluator,
],
experiment_prefix="enviro-agent-v1",
max_concurrency=2,
)
print(f"实验名称: {results.experiment_name}")
print(f"总用例数: {len(results.results)}")测试配置
pytest 配置
ini
# pytest.ini
[pytest]
markers =
integration: 图集成测试(使用 Fake 模型,无外部依赖)
slow: 慢速测试
langsmith: LangSmith 评估测试(需要 API Key 和网络)
testpaths = tests
python_files = test_*.py
python_functions = test_*
# 默认不运行 LangSmith 评估(需要真实 API Key)
addopts = -m "not langsmith"conftest.py
python
# tests/conftest.py
import os
import pytest
def pytest_configure(config):
config.addinivalue_line("markers", "integration: 图集成测试")
config.addinivalue_line("markers", "langsmith: LangSmith 评估测试")
@pytest.fixture(scope="session")
def llm_model():
"""从环境变量读取模型配置。"""
model = os.environ.get("LLM_MODEL")
if not model:
pytest.skip("LLM_MODEL 未设置,跳过需要真实模型的测试")
return model
@pytest.fixture(scope="session")
def langsmith_available():
"""检查 LangSmith 是否可用。"""
key = os.environ.get("LANGSMITH_API_KEY")
if not key:
pytest.skip("LANGSMITH_API_KEY 未设置,跳过评估测试")
return key项目结构
tests/
├── conftest.py # 共享 fixture
├── unit/ # 单元测试(极快、无外部依赖)
│ ├── test_tools.py # 工具函数测试
│ ├── test_nodes.py # 节点逻辑测试
│ └── test_middleware.py # 中间件测试
├── integration/ # 图集成测试(Fake 模型)
│ └── test_agent_graph.py # 完整 Agent 图测试
└── evaluation/ # LangSmith 评估
├── datasets/ # 测试数据集定义
├── evaluators/ # 自定义评估器
└── test_eval.py # 评估运行脚本CI/CD 集成
bash
# 本地开发:运行所有单元测试和集成测试(无真实模型调用)
pytest -v
# CI 中仅运行单元测试 + Fake 模型集成测试
pytest -m "not langsmith" -v
# 手动运行 LangSmith 评估(需要真实模型和 API Key)
pytest -m langsmith -v --tb=longyaml
# .github/workflows/test.yml 示例
# 单元测试 + Fake 模型集成测试: 每次 push 都运行(零成本)
# LangSmith 评估: 每周定时运行或手动触发常见错误与旧版 API 对照
旧版(仅用于读旧项目,新项目不得采用)
LangChain 0.x 时代使用 AgentExecutor,其输出结构与 create_agent 完全不同。以下代码在 LangChain 1.0 中 无法运行(ImportError):
python
# ❌ 0.x 写法 -- 1.0 已删除,仅用于理解旧项目
from langchain.agents import create_tool_calling_agent, AgentExecutor # ImportError
agent = create_tool_calling_agent(llm, [add_numbers], prompt) # 已删除
executor = AgentExecutor(agent=agent, tools=[add_numbers]) # 已删除
result = executor.invoke({"input": "计算 42 + 58"})
# 0.x 输出结构(1.0 中不存在)
print(result["output"]) # ❌ 1.0 没有 "output" 键
print(result["intermediate_steps"]) # ❌ 1.0 没有 "intermediate_steps" 键迁移方式:改用 create_agent,输出结构变为 {"messages": [...]}。最终回复在 result["messages"][-1].content。工具调用历史从 result["messages"] 中的 AIMessage.tool_calls 和 ToolMessage 提取。
输出结构对照
| 旧版 (AgentExecutor) | 当前 (create_agent) | 说明 |
|---|---|---|
result["output"] | result["messages"][-1].content | 最终回复文本 |
result["intermediate_steps"] | 遍历 result["messages"] 中的 AIMessage.tool_calls + ToolMessage | 工具调用历史 |
result["input"] | result["messages"][0] (首条 HumanMessage) | 用户输入 |
| 无直接对应 | result["structured_response"] | 结构化输出(使用 response_format= 时) |
评估器对照
python
# ❌ 旧版评估器(基于 AgentExecutor 输出)
def old_evaluator(run, example):
prediction = run.outputs.get("output", "") # 不存在
steps = run.outputs.get("intermediate_steps", []) # 不存在
...
# ✅ 当前评估器(基于 create_agent 输出)
def new_evaluator(run, example):
messages = run.outputs.get("messages", []) # 正确
prediction = messages[-1].content if messages else "" # 最终回复
tool_calls = [
tc for msg in messages
if isinstance(msg, AIMessage) and msg.tool_calls
for tc in msg.tool_calls
] # 工具调用历史
...适用与不适用场景
适用
- 工具函数的纯逻辑测试(参数校验、边界值、异常处理)
- Agent 图的流程测试(工具调用顺序、消息历史完整性)
- 中间件行为测试(PII 脱敏、摘要触发、HITL 中断)
- CI/CD 中的自动化回归测试(使用 Fake 模型,零成本)
- LangSmith 数据集驱动的评估(需要真实模型时)
不适用
- 验证模型推理质量(Fake 模型返回的是预设消息,不能验证模型的实际推理能力)
- 性能基准测试(Fake 模型无网络延迟和 token 生成开销)
- Prompt 工程的 A/B 测试(需要真实模型响应)
- 用户体验评估(需要真实模型的多样性和创造性)
下一步
- 使用 LangSmith Studio 可视化调试测试中发现的问题
- 了解 部署 将经过充分测试的 Agent 推向生产
- 学习 可观测性 在生产环境中持续监控 Agent
- 掌握 HITL 测试需要审批的 Agent 流程