接入 Langfuse 追踪

智能体一旦"会调工具、会多轮循环"，出了问题就特别难查：是提示词没写好？是工具返回了脏数据？还是模型压根没调对工具？Langfuse 是 LLM 可观测性平台（开源，可自托管），把一次 Agent 运行录成一棵"轨迹树"，能回放每一步的工具调用、token 消耗、耗时与成本，是调试与评测的基础设施。

两种使用方式

  • 云服务：cloud.langfuse.com 注册即可，零运维；
  • 自托管：官方仓库一条命令起服务，数据完全在内网：
git clone https://github.com/langfuse/langfuse.git
cd langfuse
docker compose up -d        # 拉取镜像并启动 web + 数据库
# 打开 http://localhost:3000 注册账号

两步接入：先配环境变量

进入项目页创建 Project 后，在「项目设置 → API Keys」拿一对密钥，配到环境变量（对应自托管地址填 LANGFUSE_BASE_URL）：

$env:LANGFUSE_PUBLIC_KEY = "pk-lf-..."
$env:LANGFUSE_SECRET_KEY = "sk-lf-..."
$env:LANGFUSE_BASE_URL   = "http://localhost:3000"   # 云服务可不填，默认云端
pip install langfuse

两步接入：代码里包一层 span

get_client() 读上面三个变量返回全局客户端；用 start_as_current_observation 上下文管理器包住一段逻辑，进入自动记开始时间、退出自动收尾。span 是"一次任务"，generation 是"一次模型调用"（可嵌套在 span 内）：

from langfuse import get_client

langfuse = get_client()

def ask_llm(prompt: str) -> str:
    # 示意：真实项目这里调用你的模型封装（第 02 章 chat()），返回文本
    return "示例回复"

with langfuse.start_as_current_observation(as_type="span", name="agent-run") as span:
    with langfuse.start_as_current_observation(
        as_type="generation", name="llm-call", model="deepseek-chat"
    ) as gen:
        reply = ask_llm("用一句话介绍向量数据库")
        gen.update(input="用一句话介绍向量数据库", output=reply)  # 有 usage 也可一并传
    span.update(output=reply)

langfuse.flush()   # 短命脚本（脚本结束/函数返回前）记得冲刷，否则事件来不及上报

跑完到 Langfuse 的 Traces 页刷新，能看到一条 agent-run 记录：点开是一棵 span 树，每条 generation 都标了模型、耗时；输入输出都可在树上逐层回放。

装饰器：@observe 自动包函数

不想手动包 with，可用 @observe() 装饰器（name/as_type 等参数以官方文档为准）：

from langfuse import observe

@observe(name="chat_with_agent")
def chat_with_agent(question: str) -> str:
    # 函数体内所有真实调用（含嵌套的 LLM 调用、工具）都会自动归入这条 span
    return "回复内容"

函数被装饰后，其运行期间内部产生的子 span 会自动挂在它下面，形成完整调用链。

与 LangGraph Agent 集成（推荐）

第 18 章起的 LangGraph 图是"循环执行节点"的，最省事的接法是塞一个 LangChain CallbackHandler，整张图一次运行自动生成一条完整 trace（每个节点、每次模型调用、每个工具都单独成 span）：

from langfuse.langchain import CallbackHandler

handler = CallbackHandler()          # 同样读 LANGFUSE_* 环境变量
config = {
    "configurable": {"thread_id": "t-001"},
    "callbacks": [handler],          # LangGraph 把回调透传给内部节点
}
result = graph.invoke({"messages": [HumanMessage("今天上海天气适合跑步吗？")]}, config)
# 异步流式同理：await graph.astream(..., {"callbacks": [handler]})

跑完到 Traces 页找最新一条：左侧是时间线，展开能看到 agent 节点与 tools 节点交替出现；点某次 generation 右侧显示完整输入/输出与 token 统计，工具的入参出参也一目了然——第 27 章评测前的"人工看失败样本"，就是靠这份回放。

看一次运行能看到什么

  • 结构与耗时：总时长、各 span 占比，定位"卡在工具还是卡在模型"；
  • Token 与成本：每条 generation 的 prompt/completion token，平台按模型单价折算成本；
  • 工具轨迹：每次调用的参数与返回原文，便于复现"模型为什么答错"。

实战：顺着一次回放定位问题

用户报"答错了"时，别靠猜——回放那条 trace 按顺序查：

  1. 打开 Traces 页按时间/用户筛出那条运行，先看总时长与各 span 占比：慢在模型调用还是慢在工具；
  2. 点开 model 类 generation 看输入输出：提示词是否把话说清、模型回复是不是答非所问；
  3. 找工具类 span 看入参出参：参数传错（如把城市名传成了日期）还是工具返回了空/脏数据；
  4. 对比 token 数：答案极短却烧了很多 token，多半是无效多轮空转（配合第 32 章"成本失控"排查）。 Langfuse 还提供 datasets/experiments：把第 27 章的回归集上传后，可对"提示词 A vs 提示词 B"批量打分对比，把"改了有没有变好"从感觉变成数据（也可接 CI 自动跑）。

常见问题

  • 控制台一条数据都没有？ ① 环境变量拼写核对：LANGFUSE_PUBLIC_KEY / LANGFUSE_SECRET_KEY / LANGFUSE_BASE_URL；② 短脚本结尾忘了 flush()；③ 自托管地址不通，先 curl http://localhost:3000/api/public/health 探活。
  • 只有 span、没有 token/成本？ 手动埋点只记你传入的内容，token 与费用要平台按模型单价折算：用官方集成（LangChain CallbackHandler 等）会自动采集 usage；deepseek 等第三方模型若平台没有内置单价，需在项目"模型定价"里手动登记后才有金额列。
  • 会不会拖慢线上响应？ 上报是异步批量（后台线程发送），不阻塞业务；仍担心就调低采样率，或把 LANGFUSE_BASE_URL 指到就近的探针服务。
  • 多服务难串起来排查？ 让网关生成 trace id 并回传/打日志，用户报障时带 trace id 来查；上报时给 span 打上 user/session/tags（propagate_attributes），UI 里按这些字段过滤即可定位单用户单会话的全链路。

敏感信息与成本建议

  • 不要把 API Key、密码、完整隐私文本放进 input/output；平台侧可开数据脱敏（masking），代码侧也可在 update 前自行截断或替换敏感字段；
  • 自托管时把 base_url 指内网地址，数据不出内网；云服务则只上传允许外发的内容；
  • 控制采样率：生产环境按比例采样（如 10%），既保可观测又省平台存储与成本；
  • 密钥务必走环境变量/密钥管理，别硬编码进代码或提交进仓库。 小结：Langfuse 把"一次 Agent 运行"变成可回放的轨迹树；接入就两步——配好 LANGFUSE_PUBLIC_KEY/SECRET_KEY/BASE_URL 环境变量，再用 get_client() 包 span/generation，或用 @observe / LangGraph CallbackHandler 自动插桩；回放时重点看模型 token、延迟与工具输入输出，敏感数据注意脱敏与采样。
笔记加载中…

标签：#Agent实战

返回知识库