代码仓库ChainReaction

给 agent 写测试,最先撞上的是它没有稳定的返回值。同一句输入,模型这次老老实实调 lookup_rate,下次直接凭记忆把汇率编出来。温度调到 0 也不保证复现,推理后端本身不承诺确定性。

第二个麻烦是依赖链太长。一次 agent 运行会经过模型 API、工具背后的数据库或第三方接口、还有你自己的业务代码。出错的时候栈顶是 ValueError,真正的原因却可能是工具描述写得含糊,模型压根没打算调它。

所以 agent 的测试要拆开写。纯函数用普通单测盖住,模型决策用假模型钉死,前两层都对了再花钱跑真实模型。观测是另一半:测试回答「这次对不对」,trace 回答「它为什么这么走」。

下面所有终端输出都是本机实跑的原文。环境是 langchain 1.4.3、langchain-openai 1.6.7、langgraph 1.2.12、langsmith 0.14.4、pytest 9.1.1、agentevals 0.0.9,模型统一换成 DeepSeek。

为什么 agent 难测

三个原因叠在一起,任何一个单独拿出来都有成熟解法,凑一起就没有现成答案。

输出不确定。普通函数的测试是「输入 X,断言输出等于 Y」。agent 的输出是一段自然语言,措辞每次都在变。就算你只想验证「有没有查汇率」,也得先决定在哪一层断言:模型吐出的 tool_calls、工具收到的参数、还是最终文案里的数字。前两个稳定,第三个不稳定。

依赖外部服务。模型 API 会限流、会超时,工具背后的接口会 500。这些失败混在同一个调用栈里,看不出是谁的锅。把真实模型换掉之后,这类噪音就消失了,剩下的失败一定是你的代码写错了。

多步执行。agent 是「模型 → 工具 → 模型 → 工具」的循环,最终的回复只是最后一步的产物。中间任何一步走偏,最后那句话照样通顺。你没法靠读最终回复判断过程对不对,只能去看轨迹。

官方把测试分成三类:单元测试(假模型,快、免费、可复现)、集成测试(真模型,验证组件能不能对接上)、评测(按轨迹打分,用来抓回归)。它们不是替代关系,是三个不同的问题。

三个层次,别混着写

层次 用不用真模型 断言什么 单次耗时 什么时候跑
工具单测 不用 返回值、异常、参数 Schema 毫秒 每次保存
agent 行为测试 假模型 调了哪个工具、参数是什么、消息结构 一两秒 每次保存
集成测试 真模型 结构性质,不比对文案 几秒,花钱 CI、发版前
评测 真模型 轨迹得分 分钟,花钱 改 prompt、换模型之后

分层的意义在于失败时能立刻定位。假模型测试挂了,是你的 harness 或工具接线有问题;假模型测试全绿而集成测试挂了,是模型或 provider 的问题;两者都绿而评测掉分,是 prompt 或工具描述退化了。

工具函数先单独测

工具是一段普通 Python,只要不碰模型,测试方式和你写过的其他函数一样。把被测代码单独放一个模块,测试文件只 import 它:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
# lc_testing_app.py
from langchain.agents import create_agent
from langchain.tools import tool

RATES = {"USD": 7.12, "EUR": 7.68, "JPY": 0.047}


@tool
def lookup_rate(currency: str) -> str:
"""Look up the CNY exchange rate for a currency code."""
code = currency.upper()
if code not in RATES:
return f"no rate for {code}"
return f"{code}/CNY = {RATES[code]}"


@tool
def convert_currency(amount: float, rate: float) -> str:
"""Convert an amount to CNY using the given exchange rate."""
if rate <= 0:
raise ValueError("rate must be positive")
return f"{amount * rate:.2f}"


def build_agent(model, **kwargs):
return create_agent(model, tools=[lookup_rate, convert_currency], **kwargs)

测试直接用 .invoke(...) 调工具,不走模型:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
# test_lc_testing_tools.py
import pytest

from lc_testing_app import convert_currency, lookup_rate


def test_lookup_rate_uppercases_input():
assert lookup_rate.invoke({"currency": "usd"}) == "USD/CNY = 7.12"


def test_lookup_rate_unknown_code_is_not_an_exception():
# 查不到汇率是业务上可预期的结果,返回文本而不是抛错,
# 这样模型有机会自己纠正参数重试。
assert lookup_rate.invoke({"currency": "XYZ"}) == "no rate for XYZ"


def test_convert_currency_rounds_to_two_decimals():
assert convert_currency.invoke({"amount": 100, "rate": 7.123456}) == "712.35"


def test_convert_currency_rejects_non_positive_rate():
with pytest.raises(ValueError, match="rate must be positive"):
convert_currency.invoke({"amount": 100, "rate": 0})


def test_tool_schema_is_what_the_model_sees():
# 工具名、描述、参数名都是模型选择工具的依据,改动它们等于改行为。
assert lookup_rate.name == "lookup_rate"
assert "exchange rate" in lookup_rate.description
assert list(lookup_rate.args_schema.model_json_schema()["properties"]) == ["currency"]

最后一个测试值得留意。工具名、docstring、参数名会被翻译成模型看到的 JSON Schema,它们是行为的一部分。改一个错别字都算改行为,值得用测试钉住。

用假模型把模型调用换掉

官方单元测试页给的是 GenericFakeChatModel:传一个 AIMessage 或字符串的迭代器进去,每调一次吐一条。写死响应之后,agent 的循环就变成确定的了。

1
2
3
4
5
6
7
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain.messages import AIMessage, ToolCall

model = GenericFakeChatModel(messages=iter([
AIMessage(content="", tool_calls=[ToolCall(name="lookup_rate", args={"currency": "USD"}, id="call_1")]),
"USD/CNY = 7.12",
]))

第一条让 agent 去调工具,第二条给出最终回复。这里有一步文档没写:create_agent 在模型节点里会调 model.bind_tools(...),而 GenericFakeChatModel 没实现它,直接抛 NotImplementedError。文档里的例子只传了 tools=[],所以没暴露这个问题。

补法很简单,继承一层,让 bind_tools 返回自己。响应序列本来就是写死的,再做真正的工具绑定没有意义:

1
2
3
4
5
6
7
8
9
# lc_testing_fakes.py
from typing import Any, Sequence

from langchain_core.language_models.fake_chat_models import GenericFakeChatModel


class FakeToolCallingModel(GenericFakeChatModel):
def bind_tools(self, tools: Sequence[Any], **kwargs: Any):
return self

不写这层包装的话,报错发生在 create_agent 内部,栈里只能看到 langchain/agents/factory.py 的 _get_bound_model,第一次遇到容易找不到北。

行为测试的断言对象是工具调用,不是文案:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
# test_lc_testing_agent_fake.py
from langchain.messages import AIMessage, HumanMessage, ToolCall, ToolMessage

from lc_testing_app import build_agent
from lc_testing_fakes import FakeToolCallingModel


def make_model(*responses):
return FakeToolCallingModel(messages=iter(responses))


def collect_tool_calls(messages):
return [tc for msg in messages for tc in (getattr(msg, "tool_calls", None) or [])]


def test_agent_calls_lookup_rate_with_the_right_argument():
model = make_model(
AIMessage(content="", tool_calls=[ToolCall(name="lookup_rate", args={"currency": "USD"}, id="call_1")]),
"USD/CNY is 7.12.",
)
agent = build_agent(model)

result = agent.invoke({"messages": [HumanMessage(content="美元汇率是多少?")]})
messages = result["messages"]

calls = collect_tool_calls(messages)
assert [c["name"] for c in calls] == ["lookup_rate"]
assert calls[0]["args"] == {"currency": "USD"}

tool_messages = [m for m in messages if isinstance(m, ToolMessage)]
assert tool_messages[0].content == "USD/CNY = 7.12"
assert isinstance(messages[-1], AIMessage)
assert messages[-1].content == "USD/CNY is 7.12."

assert calls[0]["args"] == {"currency": "USD"} 是这一层最值钱的一行。工具被调了但参数传错,是 agent 最常见的故障,而最终回复往往看不出来。

再补一个反向用例,确认不需要工具时它不会乱调:

1
2
3
4
5
6
7
8
9
10
def test_agent_can_answer_without_calling_any_tool():
model = make_model("你好,需要什么帮助?")
agent = build_agent(model)

result = agent.invoke({"messages": [HumanMessage(content="你好")]})
messages = result["messages"]

assert collect_tool_calls(messages) == []
assert not [m for m in messages if isinstance(m, ToolMessage)]
assert messages[-1].content == "你好,需要什么帮助?"

工具抛异常时会发生什么

默认没有兜底。工具抛异常会打断整次 invoke,异常原样传到你手上:

1
2
3
4
5
6
7
8
def test_tool_exception_propagates_by_default():
model = make_model(
AIMessage(content="", tool_calls=[ToolCall(name="convert_currency", args={"amount": 100, "rate": 0}, id="call_1")]),
)
agent = build_agent(model)

with pytest.raises(ValueError, match="rate must be positive"):
agent.invoke({"messages": [HumanMessage(content="按 0 的汇率换算 100")]})

想让模型看到错误并自己纠正,挂上 ToolErrorMiddleware(需要 langchain>=1.3.14)。它把异常转成 status="error" 的 ToolMessage:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from langchain.agents.middleware import ToolErrorMiddleware


def test_tool_error_middleware_turns_exception_into_a_message():
def on_error(exc: Exception, request) -> str:
return f"{request.tool_call['name']} failed: {type(exc).__name__}"

model = make_model(
AIMessage(content="", tool_calls=[ToolCall(name="convert_currency", args={"amount": 100, "rate": 0}, id="call_1")]),
"汇率不合法,我没法换算。",
)
agent = build_agent(model, middleware=[ToolErrorMiddleware(on_error)])

result = agent.invoke({"messages": [HumanMessage(content="按 0 的汇率换算 100")]})
tool_messages = [m for m in result["messages"] if isinstance(m, ToolMessage)]

assert tool_messages[0].status == "error"
assert tool_messages[0].content == "convert_currency failed: ValueError"

官方建议 on_error 返回异常类型而不是原始异常文本,原始信息可能带内部细节。

多轮对话用 InMemorySaver

需要验证跨轮次的状态时,给 agent 挂 InMemorySaver,同一个 thread_id 复用历史:

1
2
3
4
5
6
7
8
9
10
11
12
13
from langgraph.checkpoint.memory import InMemorySaver


def test_history_is_reused_inside_one_thread():
model = make_model("记住了,你住上海。", "上海现在 24 度。")
agent = build_agent(model, checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "t-1"}}

agent.invoke({"messages": [HumanMessage(content="我住在上海")]}, config=config)
second = agent.invoke({"messages": [HumanMessage(content="我这儿天气如何?")]}, config=config)

assert len(second["messages"]) == 4
assert second["messages"][0].content == "我住在上海"

我第一版忘了传 checkpointer,第二次 invoke 只返回 2 条消息,断言挂了。这个失败很有代表性:假模型测试失败时,先怀疑接线,再怀疑模型。

pytest 实际长什么样

目录里就是这样几个文件:

1
2
3
4
5
6
7
8
9
Testing/
├── pytest.ini
├── conftest.py
├── app.py # 被测代码
├── fakes.py # 假模型
├── test_tools.py # 第一层
├── test_agent_fake.py # 第二层
├── test_agent_live.py # 第三层,默认不跑
└── test_langsmith_integration.py # LangSmith 集成

pytest.ini 把集成测试挡在默认运行之外:

1
2
3
4
[pytest]
markers =
integration: tests that call real LLM APIs
addopts = -m "not integration" -q

conftest.py 只负责把密钥从环境或 .env 里读进来:

1
2
3
from dotenv import load_dotenv

load_dotenv()

跑默认那批,1.5 秒结束:

1
2
3
4
5
6
7
8
9
10
11
12
============================= test session starts =============================
platform win32 -- Python 3.12.14, pytest-9.1.1, pluggy-1.6.0
rootdir: D:\Code\Python_Code\LangCraft\Testing
configfile: pytest.ini
plugins: anyio-4.15.1, langsmith-0.14.4, platformdirs-4.12.3
collected 12 items / 1 deselected / 11 selected

test_agent_fake.py ..... [ 45%]
test_langsmith_integration.py . [ 54%]
test_tools.py ..... [100%]

====================== 11 passed, 1 deselected in 4.90s =======================

11 个用例里没有一次网络请求,也不消耗 token。这套东西可以直接挂到 pre-commit 上。

真实模型的集成测试

集成测试只验证一件事:接上真实 provider 之后,schema、密钥、工具绑定这些环节能不能对上。所以断言的是结构,不是文案。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
# test_lc_testing_agent_live.py
import os

import pytest
from langchain.messages import AIMessage, HumanMessage
from langchain_openai import ChatOpenAI

from lc_testing_app import build_agent


@pytest.fixture(autouse=True)
def check_api_keys():
if not os.environ.get("DEEPSEEK_API_KEY"):
pytest.skip("DEEPSEEK_API_KEY not set")


def build_deepseek():
return ChatOpenAI(
api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com/v1",
model="deepseek-chat",
temperature=0.1,
max_tokens=400,
)


@pytest.mark.integration
def test_agent_calls_lookup_rate_against_real_model():
agent = build_agent(build_deepseek())

result = agent.invoke({"messages": [HumanMessage(content="欧元兑人民币今天多少?")]})
messages = result["messages"]

calls = [tc for msg in messages for tc in (getattr(msg, "tool_calls", None) or [])]
assert any(c["name"] == "lookup_rate" for c in calls)
assert isinstance(messages[-1], AIMessage)
assert len(messages[-1].content) > 0

显式跑一次:

1
2
3
4
5
6
7
8
9
10
11
$ pytest -m integration -v
============================= test session starts =============================
platform win32 -- Python 3.12.14, pytest-9.1.1, pluggy-1.6.0
rootdir: D:\Code\Python_Code\LangCraft\Testing
configfile: pytest.ini
plugins: anyio-4.15.1, langsmith-0.14.4, platformdirs-4.12.3
collected 12 items / 11 deselected / 1 selected

test_agent_live.py . [100%]

====================== 1 passed, 11 deselected in 3.14s =======================

3 秒,一次模型调用。check_api_keys 那个 fixture 没密钥就 skip,不会让 CI 红成一片。

成本上有几个能立刻做的动作:用便宜的小模型跑结构测试,把 max_tokens 压到 256 左右,一个用例只验一个行为。官方还提到用 vcrpy 加 pytest-recording 把 HTTP 交互录成 cassette 文件,首次录制之后重放,不再花钱。代价是改了 prompt 或加了工具之后 cassette 会过期,测试会红,得删掉重录。我没有实际启用这套,录制文件一旦进仓库就多了一份要维护的资产。

评测:数据集加评估器

测试回答「有没有坏」,评测回答「好了多少」。评估器是一个函数,拿实际输出和参考输出,返回一个分数:

1
2
3
def evaluator(*, outputs: dict, reference_outputs: dict):
score = compare_messages(outputs["messages"], reference_outputs["messages"])
return {"key": "evaluator_score", "score": score}

agentevals 提供了现成的轨迹比对器 create_trajectory_match_evaluator,四种模式:

模式 含义 什么时候用
strict 消息结构与工具调用顺序完全一致 必须按固定顺序执行,比如先查政策再授权
unordered 工具调用集合一致,顺序随意 只关心信息有没有查全
subset 只调了参考里的工具,没有多余的 防止 agent 越权
superset 至少调了参考里的工具,允许多调 验证最低要求的动作都做了

下面这段不需要网络,假模型加内存里写死的数据集就能跑完:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
# eval_trajectory.py
from agentevals.trajectory.match import create_trajectory_match_evaluator
from langchain.messages import AIMessage, HumanMessage, ToolCall, ToolMessage

from lc_testing_app import build_agent
from lc_testing_fakes import FakeToolCallingModel

DATASET = [
{
"inputs": "美元汇率是多少?",
"reference": [
HumanMessage(content="美元汇率是多少?"),
AIMessage(content="", tool_calls=[ToolCall(name="lookup_rate", args={"currency": "USD"}, id="c1")]),
ToolMessage(content="USD/CNY = 7.12", tool_call_id="c1"),
AIMessage(content="USD/CNY = 7.12"),
],
"script": [
AIMessage(content="", tool_calls=[ToolCall(name="lookup_rate", args={"currency": "USD"}, id="c1")]),
"USD/CNY = 7.12",
],
},
{
"inputs": "100 美元换人民币",
"reference": [
HumanMessage(content="100 美元换人民币"),
AIMessage(content="", tool_calls=[ToolCall(name="convert_currency", args={"amount": 100, "rate": 7.12}, id="c1")]),
ToolMessage(content="712.00", tool_call_id="c1"),
AIMessage(content="712.00"),
],
# 故意少调一次工具:只回答不换算,评测应该给 0 分
"script": ["100 美元大约是 712 元。"],
},
]


def run_case(case):
model = FakeToolCallingModel(messages=iter(case["script"]))
agent = build_agent(model)
result = agent.invoke({"messages": [HumanMessage(content=case["inputs"])]})
return result["messages"]


def main():
evaluator = create_trajectory_match_evaluator(trajectory_match_mode="superset")
rows = []
for case in DATASET:
outputs = run_case(case)
score = evaluator(outputs=outputs, reference_outputs=case["reference"])
rows.append((case["inputs"], score["key"], score["score"]))

width = max(len(r[0]) for r in rows) + 2
print("inputs".ljust(width) + "evaluator".ljust(34) + "score")
for inputs, key, score in rows:
print(inputs.ljust(width) + key.ljust(34) + str(score))
passed = sum(1 for r in rows if r[2] is True)
print(f"\n通过 {passed}/{len(rows)}")

跑出来的结果:

1
2
3
4
5
inputs      evaluator                         score
美元汇率是多少? trajectory_superset_match True
100 美元换人民币 trajectory_superset_match False

通过 1/2

第二条故意让假模型跳过工具、直接报答案。数值恰好是对的,文案读起来毫无破绽,但轨迹比对照样给了 False。这就是评测和「看一眼输出」的区别。

评估流程本身很短:

要跨时间对比,就得把结果落到 LangSmith。官方给了两条路:pytest 集成(pytest.mark.langsmith 配合 langsmith.testing 的 log_inputs / log_outputs / log_reference_outputs,用 pytest --langsmith-output 跑),或者用 Client().evaluate()。后者的参数是目标函数、data(数据集名字或 UUID)、evaluators 列表,可选 experiment_prefix 和 max_concurrency。数据集里每条 example 的 inputs 会被喂给目标函数,outputs 作为参考值传给评估器。

两条路我都跑了一遍。pytest 那条把输入、输出、参考输出一起落到平台:

1
2
3
4
5
6
7
8
9
10
11
Test Suite: blog.test_ls_integration
LangSmith URL:
https://smith.langchain.com/o/16749fb8-.../datasets/8e2315c4-.../compare?selectedSessions=72be27f2-...

┌───────────┬───────────┬───────────┬──────────┬────────┬──────────┬──────────┐
│ Test │ Inputs │ Ref │ Outputs │ Status │ Feedback │ Duration │
│ │ │ outputs │ │ │ │ │
├───────────┼───────────┼───────────┼──────────┼────────┼──────────┼──────────┤
│ ...ool_c… │ {"questi… │ {"answer… │ {"answe… │ passed │ │ 2.24s │
└───────────┴───────────┴───────────┴──────────┴────────┴──────────┴──────────┘
Averages 100% 2.24s

跑法是 pytest Testing/test_langsmith_integration.py --langsmith-output,测试函数上打 @pytest.mark.langsmith,再用 log_inputs / log_outputs / log_reference_outputs 把三份数据显式写进去。这三个函数收的都是字典,我第一次传了字符串,报 ValueError: dictionary update sequence element #0 has length 1; 2 is required,改成 log_inputs({"question": question}) 才正常。

Client().evaluate() 那条更直接(脚本 Testing/LiveEvaluation.py)。数据集两条 example,目标函数就是前面那个天气/汇率 agent,两个评估器分别检查「该调的工具有没有调」和「答案里有没有期望的数值」:

1
2
3
4
5
新建数据集: langchain-blog-eval
experiment: langchain-blog-eval-cb55c748

上海天气怎么样? -> tool_selected=1.0, answer_contains=1.0
美元汇率是多少? -> tool_selected=1.0, answer_contains=1.0

每条 example 一行分数,experiment_prefix 决定实验名。改一版 prompt 再跑,两个实验能在同一个数据集页面上并排比。这次两条满分只是因为问题太简单,评测的价值不在这里,而在于把「这次改动有没有让某条退化」变成一条能查的记录。

观测:把追踪打开

LangChain 的 agent 默认就支持 LangSmith 追踪,只要两个环境变量:

1
2
export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY=<your-api-key>

代码一行都不用改。默认落到 default 项目,用 LANGSMITH_PROJECT 换项目名;账号不在美国区还要设 LANGSMITH_ENDPOINT,比如 https://eu.api.smith.langchain.com,末尾不能带斜杠。

不想全局开,可以用 tracing_context 只追踪指定片段:

1
2
3
4
5
6
7
8
import langsmith as ls

with ls.tracing_context(enabled=True, project_name="langchain-testing-blog"):
agent.invoke({"messages": [{"role": "user", "content": "查一下美元汇率"}]})

# LANGSMITH_TRACING=true 时,这段仍然会被跳过
with ls.tracing_context(enabled=False):
agent.invoke({"messages": [{"role": "user", "content": "内部压测请求"}]})

给 trace 打标签和元数据,从 invoke 的 config 里传:

1
2
3
4
5
6
7
result = agent.invoke(
{"messages": [{"role": "user", "content": "查一下美元汇率"}]},
config={
"tags": ["production", "rate-agent", "v1.0"],
"metadata": {"user_id": "user_123", "environment": "production"},
},
)

metadata 和 tags 会被子 run 继承。按 environment=staging 过滤就能把测试流量从生产里摘出去。

trace 里到底有什么

LangSmith 的数据结构是分层的。一个 run 是最小工作单元,可以想成 OpenTelemetry 里的 span;一次操作里的所有 run 组成一条 trace,靠 trace_id 绑在一起;多轮会话的 trace 再串成 thread,靠 thread_id metadata 关联。一条 trace 最多 25000 个 run,超了会被拒收。

每个 run 都能展开看到完整输入输出、耗时、token 数。排查工具类问题时,你会一直盯着 run:工具节点 这一层往下看。

我这边实际测到的东西

追踪开关打开之后,SDK 会把每次运行回传。我用同一个 agent(查上海天气、查美元汇率)跑了一遍,再从 API 把 trace 树读回来(脚本 Testing/LiveTrace.py):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
LANGSMITH_TRACING = true
LANGSMITH_PROJECT = LangChain
agent 最终回复: 以下是查询结果:上海天气 24°C,晴朗;1 美元 = 7.12 人民币
本地端到端耗时: 2.16s

从 LangSmith 读回的 trace: 01a107ab-9ce1-7340-8e48-44f7087dbf0c
trace URL: https://smith.langchain.com/o/-/projects/p/ef54eff2-485e-4c20-952e-d5fa35bd6694/r/01a107ab-9ce1-7340-8e48-44f7087dbf0c

最终 trace 树(9 个 run):
- [chain] LangGraph tokens=762+125 status=success
- [chain] model tokens=326+74 status=success
- [llm] ChatOpenAI tokens=326+74 status=success
- [chain] tools status=success
- [tool] get_weather status=success
- [chain] tools status=success
- [tool] get_exchange_rate status=success
- [chain] model tokens=436+51 status=success
- [llm] ChatOpenAI tokens=436+51 status=success

结构和前面那张图对得上:一次 LangGraph 运行、两次模型调用、两个工具节点。两个工具落在两个串行的 tools 节点里,不是并行,这一轮 DeepSeek 没把两个调用放进同一次回复。token 也直接读得到,762+125 是整条 trace 的合计,两次模型调用分别是 326+74 和 436+51,成本按这个算。

顺带一个坑:run 刚跑完就立刻去查,根节点的 status 可能还是 pending,隔一会儿再查才是 success。上面这棵树是等它稳定之后读的。

把 LANGSMITH_API_KEY 拿掉再跑一次,是另一幅样子:

1
2
3
4
5
6
7
8
9
10
LANGSMITH_TRACING = true
LANGSMITH_API_KEY = <unset>
LANGSMITH_PROJECT = langchain-testing-blog
LANGSMITH_ENDPOINT = https://api.smith.langchain.com
agent 正常返回: traced reply
wait_for_all_tracers() 返回

...\langsmith\client.py:757: LangSmithMissingAPIKeyWarning: API key must be provided when using hosted LangSmith API
warnings.warn(
Failed to multipart ingest runs: langsmith.utils.LangSmithAuthError: Authentication failed for https://api.smith.langchain.com/runs/multipart. HTTPError('401 Client Error: Unauthorized for url: https://api.smith.langchain.com/runs/multipart', '{"error":"Unauthorized"}\n')trace=01a107ad-9abf-7b83-81eb-f7281a45396a,id=01a107ad-9abf-7b83-81eb-f7281a45396a; trace=01a107ad-9ba3-7413-be61-8df9e31dfe1b,id=01a107ad-9ba3-7413-be61-8df9e31dfe1b

两件事值得记下来。追踪是后台线程发的,agent.invoke 照常返回,回传失败不会打断业务。失败信息里带着 trace=... id=...,说明 run 在本地确实建出来了,只是被 401 挡在门外。线上 trace 丢了先查 key 和 endpoint,不用怀疑 agent 本身。

不想开账号、或者要在离线环境里看 trace 骨架,可以用回调自己收一棵树。下面这段跑的是真实 DeepSeek,输出是本机原文:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# trace_tree.py(节选)
class RunTreeRecorder(BaseCallbackHandler):
"""把 on_*_start / on_*_end 拼成一棵树,等价于 LangSmith 里一条 trace 的骨架。"""

def on_chain_start(self, serialized, inputs, *, run_id, parent_run_id=None, **kw):
self._start(run_id, parent_run_id, (serialized or {}).get("name") or "chain", "chain")

def on_chat_model_start(self, serialized, messages, *, run_id, parent_run_id=None, **kw):
self._start(run_id, parent_run_id, "ChatOpenAI", "llm")

def on_tool_start(self, serialized, input_str, *, run_id, parent_run_id=None, **kw):
self._start(run_id, parent_run_id, (serialized or {}).get("name") or "tool", "tool")

def on_llm_end(self, response, *, run_id, parent_run_id=None, **kw):
self._end(run_id)
msg = response.generations[0][0].message
self.runs[run_id]["usage"] = getattr(msg, "usage_metadata", None)
1
2
3
4
5
6
7
8
9
10
11
本地 run 树(回调收集,非 LangSmith)
- [chain] chain 1563ms
- [chain] chain 1033ms
- [llm] ChatOpenAI 1029ms tokens=326+74
- [chain] chain 2ms
- [tool] get_weather 0ms
- [chain] chain 1ms
- [tool] get_exchange_rate 0ms
- [chain] chain 519ms
- [llm] ChatOpenAI 518ms tokens=436+51
整条 trace 端到端: 1.57s, run 数: 9

形状和 LangSmith 里那条 trace 是一回事:agent 主循环包着两次模型调用和两个工具 run,工具 run 各自带着自己的入参出参。差别只是这里打在终端上,LangSmith 打在网页里。

用 trace 排查三类问题

拿到 trace 之后,大部分故障落在三个模式里。

工具没被调用。看第一次模型调用的输出消息,tool_calls 是空的就说明模型根本没打算调。原因通常在工具描述:lookup_rate 的 docstring 是给模型看的唯一说明书,写得太泛或者参数名有歧义,模型宁可自己回答。把 trace 里模型看到的完整工具 Schema 和对话历史放在一起看,问题基本自己跳出来。

参数传错。工具 run 的入参是模型生成的原样 JSON。类型对不上会在工具入口抛校验错误,字段名写错会变成默认值或者 KeyError。这类问题用假模型测试最容易提前拦住,写死一次错误参数就能复现。

循环次数过多。模型陷进「调工具 → 看不懂结果 → 再调一次」的循环,token 会飞。给 agent 挂上限:

1
2
3
4
5
6
from langchain.agents.middleware import ModelCallLimitMiddleware

agent = build_agent(
looping_model(),
middleware=[ModelCallLimitMiddleware(run_limit=4, exit_behavior="end")],
)

用一个永远只调同一个工具的假模型跑,加上限和不上限的差别是:

1
2
3
模型调用次数: 5  工具执行次数: 4
最后一条消息: AIMessage 'Model call limits exceeded: run limit (4/4)'
不设上限时: GraphRecursionError Recursion limit of 30 reached without hitting a stop condition.

exit_behavior="end" 会优雅收尾,留一条说明原因的消息;改成 "error" 则直接抛异常,适合在生产里快速失败。ModelCallLimitMiddleware 数的是模型调用,ToolCallLimitMiddleware 数的是工具调用,还能按工具名分别限制(tool_name="search", run_limit=3)。两个都支持 thread_limit,跨轮次累计,需要 checkpointer 才能记住。

不上限那行也值得看:默认递归上限是 9999,模型只要一直吐工具调用,agent 能转到天荒地老。上面这行是我把 recursion_limit 压到 30 跑出来的,默认值下要等更久。

成本和延迟

LangSmith 会自动统计 token 和花费。它需要三样东西:token 数、模型与 provider、以及价格表。用 LangChain 调模型时前两样是自动带上的,价格表在 workspace 的 model pricing 里配,内置了大部分 OpenAI、Anthropic、Gemini 的价目。DeepSeek 走 langchain-openai 的 OpenAI 兼容协议,ls_provider 会是 openai 而不是 deepseek,所以要自己加一条价格记录,Match Pattern 填 deepseek-chat。UI 里 token 分输入、输出、其他三类,其他那一类装的是工具调用和检索步骤的成本。

延迟不用等 LangSmith,回调里就有。上面那棵树里,两次模型调用占了 1029ms 和 518ms,两个工具加起来 3ms。工具慢还是模型慢,一眼就能分出来。如果 agent 跑在 serverless 里,进程可能在后台线程回传完成之前就被杀掉,官方建议把 LANGCHAIN_CALLBACKS_BACKGROUND 设成 "false",或者在退出前显式调一次 wait_for_all_tracers():

1
2
3
4
5
6
from langchain_core.tracers.langchain import wait_for_all_tracers

try:
agent.invoke({"messages": [{"role": "user", "content": "查一下美元汇率"}]})
finally:
wait_for_all_tracers()

LangSmith SaaS 的 trace 保留 180 天。想长期留着做回归基线,得把它存成 dataset,dataset 不受这个期限影响。

小结

  • 测试分三层,假模型那层性价比最高。它跑在秒级、不花钱、结果确定,能拦住参数传错和工具接线错误。
  • 断言工具调用和参数,不要断言文案。calls[0]["args"] == {"currency": "USD"} 比任何字符串匹配都稳。
  • GenericFakeChatModel 不支持 bind_tools,agent 会因此抛 NotImplementedError。继承一层让 bind_tools 返回自己就行。
  • 集成测试用 pytest marker 和默认 addopts 隔离,没密钥就 skip,别让它在每次保存时都跑。
  • 观测靠环境变量打开,trace 是 run 的树。排查时先看第一次模型调用的 tool_calls 是否为空,再看工具 run 的入参,最后看循环次数。
  • 评测用数据集加评估器。轨迹比对能抓到「答案对了但过程不对」,这是读最终回复永远发现不了的。