你在 LangSmith 里查看一个 CrewAI 任务的 trace，发现 Agent 声称执行了「数据库查询」，但 trace 里只有 LLM 调用记录，找不到任何 database_query tool call。或者在 Langfuse 里，某个 Temporal activity 的 span 出现了，但它里面调用的子 tool call 全部缺失——你需要知道查询了哪些参数、返回了什么，但 trace 里是空白的。这类问题在需要审计、合规或安全复查的场景里尤其棘手，因为你无法证明 Agent 做了（或没做）某件事。
常见原因
1. 工具是直接调用 Python 函数，没有通过框架的 trace 层
开发者把工具定义为普通 Python 函数，Agent 通过函数调用而不是框架的 tool_executor 执行，绕过了 trace 中间件。LangChain 的 @tool 装饰器或 Tool 类包装的工具会被自动 trace，但直接调用的函数不会。
怎么判断：检查工具的定义方式。如果是 def my_tool(x): ... 而不是 @tool def my_tool(x): ... 或 Tool(name="my_tool", func=my_tool)，就不会出现在 trace 里。
2. 异步工具在不同的 async context 里运行，span 没有正确关联
工具在一个新的 asyncio.Task 或线程里执行，trace 上下文（通过 contextvars.Context 传递）没有被复制到新的执行上下文里，导致工具的 span 没有被关联到父级 trace。
怎么判断：查看 Langfuse/LangSmith 里是否有「孤儿」span（没有父级 trace 的 span），这些孤儿 span 就是上下文丢失的工具调用。
3. Trace SDK 的 flush 没有在进程退出前完成
Langfuse.flush() 或 LangSmith 的后台写入线程在进程退出时还没有完成，导致最后一批 span 数据没有发送到服务器。在短生命周期的函数（如 AWS Lambda、Cloud Run）里尤其常见。
怎么判断：比较应用侧记录的「工具调用次数」与 trace 平台显示的次数，如果后者总是少几条，就是 flush 问题。
4. Trace SDK 的采样率低于 100%
为了控制 trace 数据量，SDK 被配置为只记录 50% 的调用。某些重要的工具调用正好被采样丢弃了。
怎么判断：检查 Langfuse 或 OpenTelemetry SDK 的 sample_rate 配置，如果小于 1.0，就会有调用被采样丢弃。
5. 工具调用发生在 LLM response 的 tool_use block 里，但 trace wrapper 只记录 text block
自定义的 trace wrapper 在处理 Anthropic response 时，只把 text 类型的 content block 记录到 span，忽略了 tool_use 类型的 block，导致所有工具调用都不在 trace 里。
怎么判断：检查 trace wrapper 的代码，找 response.content 的处理逻辑，确认是否对 tool_use block 也做了记录。
6. 并发工具调用的 span 时间戳重叠，被 trace 视图折叠
多个工具并发执行时，它们的 span 时间戳完全相同，trace 视图把它们合并显示为一条，看起来像只有一次调用。
怎么判断：在 trace 视图里查看 span 的详情（JSON 格式），检查是否有多条 span 有完全相同的 start_time。
最短修复路径
Step 1：用 @tool 装饰器或 Tool 类包装所有工具
from langchain_core.tools import tool
# 之前（不会被 trace）
def search_database(query: str) -> list[dict]:
return db.execute(query)
# 之后（自动 trace）
@tool
def search_database(query: str) -> list[dict]:
"""在数据库里搜索记录。query: SQL 查询语句"""
return db.execute(query)
# 或者用 StructuredTool
from langchain.tools import StructuredTool
search_tool = StructuredTool.from_function(
func=search_database,
name="search_database",
description="在数据库里搜索记录"
)Step 2：在异步工具里显式传播 trace 上下文
import contextvars
from langfuse.decorators import langfuse_context
async def spawn_tool_task(tool_fn, args: dict):
"""在新的 asyncio.Task 里保留 trace 上下文。"""
# 复制当前 contextvars 上下文
ctx = contextvars.copy_context()
# 在复制的上下文里运行工具
return await asyncio.get_event_loop().run_in_executor(
None,
lambda: ctx.run(tool_fn, **args)
)
# Langfuse 的 @observe 装饰器自动处理上下文传播
from langfuse.decorators import observe
@observe(name="search_database")
async def search_database(query: str) -> list[dict]:
return await db.async_execute(query)Step 3：在进程退出前强制 flush
import atexit, signal
from langfuse import Langfuse
langfuse = Langfuse()
# 注册退出钩子，确保 trace 数据被发送
atexit.register(langfuse.flush)
# 处理 SIGTERM（容器停止信号）
def handle_sigterm(signum, frame):
langfuse.flush()
sys.exit(0)
signal.signal(signal.SIGTERM, handle_sigterm)
# AWS Lambda / Cloud Run 等短生命周期函数：在 handler 返回前显式 flush
def lambda_handler(event, context):
try:
result = run_agent(event)
return result
finally:
langfuse.flush() # 必须在函数返回前 flushStep 4：自定义 Anthropic trace wrapper，记录 tool_use block
from anthropic.types import Message
def trace_anthropic_response(response: Message, span_id: str):
"""记录 Anthropic 响应里的所有 content block，包括 tool_use。"""
for i, block in enumerate(response.content):
if block.type == "text":
langfuse.span(
parent_id=span_id,
name=f"llm_text_block_{i}",
output=block.text
)
elif block.type == "tool_use":
langfuse.span(
parent_id=span_id,
name=f"tool_call_{block.name}",
input=block.input,
metadata={"tool_id": block.id, "tool_name": block.name}
)Step 5：把 trace 采样率设为 100%，对高频低值调用单独降采样
# langfuse SDK 设置
langfuse = Langfuse(
public_key="...",
secret_key="...",
# 不要设置 sample_rate < 1.0，而是在应用层控制哪些调用要 trace
)
# 对高频低价值调用（如心跳检查），在调用前判断是否 trace
def should_trace_tool_call(tool_name: str, args: dict) -> bool:
LOW_VALUE_TOOLS = {"health_check", "ping", "get_timestamp"}
return tool_name not in LOW_VALUE_TOOLS预防建议
- 所有 Agent 工具必须通过框架的
@tool装饰器或Tool类注册，不允许直接调用原始函数 - 异步工具用
@observe或@traceable装饰器包装，确保 trace 上下文自动传播 - 短生命周期进程（Lambda、Cloud Run）在退出前显式调用
langfuse.flush() - Trace 采样率保持 100%，降低数据量应该通过过滤低价值工具，而不是随机采样
- 建立「trace 完整性监控」：统计每次 Agent 执行的预期工具调用数与 trace 里记录的数量，不匹配时告警
- 在 staging 环境定期运行一个「trace 检查器」：执行已知的 Agent 任务，验证 trace 里包含所有预期的 tool call span
- 对安全、合规相关的工具调用（如数据库写入、外部 API 调用），在工具函数里加独立的日志记录，不依赖 trace SDK
常见问答 (FAQ)
Q: LangSmith 和 Langfuse 对工具调用的 trace 支持有什么差异？
A: LangSmith 与 LangChain 原生集成，对 @tool 装饰的函数自动 trace，对 Runnable 接口的工具也有完整支持。Langfuse 是与框架无关的，需要手动用 @observe 装饰或显式创建 span。两者都需要正确的异步上下文传播，否则都会出现孤儿 span。
Q: 如果工具是第三方库（无法加装饰器），如何 trace？
A: 用上下文管理器手动创建 span 包裹调用：with langfuse.start_as_current_span("third_party_tool") as span: result = third_party_lib.call(); span.set_output(result)。这样能捕获到调用，但工具的内部细节不可见。
Q: trace 里的工具调用顺序是否反映实际执行顺序？ A: 如果工具是串行执行的，span 的 start_time 顺序就是执行顺序。并发执行的工具 span 可能在 trace 视图里乱序显示，需要按 start_time 排序才能还原真实执行顺序。在做安全审计时，务必按 start_time 而不是记录插入顺序来重建执行序列。
Q: Trace 数据里包含工具的输入参数，这会不会有隐私问题？ A: 是的，如果工具接受用户数据或敏感参数，trace 就会包含这些信息。解决方案：在记录 span 前对输入参数做脱敏处理（如掩码 PII 字段），或者在 Langfuse 的数据保留策略里设置自动删除周期。参见[相关阅读]里关于 secret 泄漏的文章。