人工智能 ·

Langfuse 实战:把 AI 的慢、贵和答错,变成可定位的问题

用 Python SDK v4 接入 Langfuse,串起 Trace、提示词版本与评估。讲清最小 RAG 埋点、失败样本、自托管组件和数据脱敏,避免只看 Token 看板。

AI 应用上线后,最让人头疼的往往不是“模型不够聪明”,而是用户说了一句“刚才又答错了”,你却不知道它错在哪一步。

检索没找到资料,还是提示词漏了条件?模型慢,还是工具超时后重试了三次?换了一个便宜模型,单次调用省钱了,整条任务却更贵?

Langfuse 的价值,是把这些问题接回一次具体任务的执行过程。 你可以看到调用链、输入输出、耗时、用量和评估结果,再决定改哪里。看板是入口,修得动问题才是收益。

本文按 2026 年 10 月 11 日核对的官方文档与 Python SDK 4.17.0 编写。示例演示埋点结构,不调用真实模型;这里没有虚构线上压测数据或部署成绩。

先分清 Trace、Observation 和 Session

概念 适合记录什么 常见误用
Trace 一次完整任务,如一轮问答、一次 Agent 运行 把整天请求塞进一个 Trace
Observation 任务内的一步,如检索、模型调用、工具执行 只记最终答案,看不到中间过程
Session 关联多轮对话或连续任务 用 Session 代替单次任务边界

把一次问答设为一个 Trace,把检索与生成放进它的子节点。多轮聊天再用 Session 串起来。当前 SDK 基于 OpenTelemetry;正确传播上下文,比给每个函数起一个漂亮名字更重要。

Agent 循环尤其不能只包一层 generation。每次模型调用应单独记录,工具调用跟着对应步骤走,否则你只能看到总账,不知道是哪次结果把上下文撑大了。

先接一条链路,别急着接全公司

用一个 RAG 问答入口练手就够了。先把这三个节点接清楚:

answer-question
├── retrieve-documents
└── draft-answer

根节点记录用户问题和最终结果,检索节点记录文档 ID、版本和实际传给模型的片段,generation 记录模型输入输出及真实用量。

节点名保持稳定,例如 retrieve-documents。订单号、用户问题和请求 ID 放在属性里,不要拼进名称,否则同一类任务很难聚合。再补上发布版本、环境和提示词版本,才有条件比较更新前后的变化。

Python 最小示例:先验证能看见过程

本例固定 Python SDK 4.17.0,需要 Python 3.10 或以上。SDK 大版本、服务端版本和功能兼容范围请分别核对,别把旧教程里的方法直接混进来。

python3 -m venv .venv
source .venv/bin/activate
python -m pip install "langfuse==4.17.0"

export LANGFUSE_PUBLIC_KEY="pk-lf-REPLACE_ME"
export LANGFUSE_SECRET_KEY="sk-lf-REPLACE_ME"
export LANGFUSE_BASE_URL="https://cloud.langfuse.com"

Key 来自 Langfuse 项目设置。上面的 BASE_URL 是欧洲 Cloud 入口,其他地区或自托管要换成自己的地址。密钥放服务端环境变量,不能放网页前端。只有写一个 .env 文件不会让 Python 自动加载它;这里直接用 shell 导出变量。

把下面保存为 trace_demo.py,执行 python trace_demo.py:

from langfuse import get_client

langfuse = get_client()
question = "How long are repeat article visits deduplicated?"

with langfuse.start_as_current_observation(
    as_type="span",
    name="answer-question",
    input={"question": question},
    metadata={"release": "demo-v1", "environment": "development"},
) as root:
    with langfuse.start_as_current_observation(
        as_type="span", name="retrieve-documents", input=question
    ) as retrieval:
        documents = [{"id": "privacy-v1", "text": "Repeat visits count once within 30 minutes."}]
        retrieval.update(output=documents)

    with langfuse.start_as_current_observation(
        as_type="generation",
        name="draft-answer",
        input={"question": question, "documents": documents},
        metadata={"simulation": True},
    ) as generation:
        answer = "Repeat visits count once within 30 minutes. [privacy-v1]"
        generation.update(output=answer)

    root.update(output={"answer": answer})

langfuse.flush()

注意:文档和答案都是写死的演示数据,simulation=True 明确标记模拟。这个 generation 并未发起模型请求,也没有上报虚构的 Token 或费用。接真实应用时,把这两段替换成你的检索函数和模型调用;记录实际模型标识,并从提供商响应写入真实 usage。

短命脚本退出前调用 flush(),避免队列里的事件尚未发送。常驻服务按生命周期统一处理,别在每个 token 或每个小步骤后强制 flush。异步任务、线程和队列的上下文传播也要单独验证,不能因为用了 context manager 就默认跨进程仍然完整。

看到 Trace 后,该查什么

我会先找一条用户确认有问题的请求,然后按顺序看:

  1. 输入条件:缺了产品型号、日期或适用版本吗?
  2. 检索结果:文档是否相关,是否过期,关键条件有没有被切分丢掉?
  3. 模型调用:实际发出的上下文是否完整,是否偷偷多了重试?
  4. 工具结果:接口成功返回,是否真的完成了业务动作?

慢要看各步骤耗时和任务总时长;流式响应还要区分首个 token 与最终完成。贵要算整条任务里的所有调用与重试。平台的费用估算依赖模型价格配置和 usage 数据,最终账单仍要跟提供商核对。

Langfuse 不会自动证明答案正确,也不会替你实施业务权限。Trace 能显示一次越权尝试,不等于它能够阻止越权。鉴权、参数验证、幂等和写入确认仍应由应用落实。

提示词管理:记住“当时到底用了哪一版”

提示词不只是放到后台让人编辑。真正有用的是把版本与调用关联起来,出问题时能复现,改动后能比较。

先在后台创建名为 support-answer 的 text prompt,内容里使用 {{question}} 和 {{context}},再把验收过的版本标成 production:

prompt = langfuse.get_prompt("support-answer", label="production")
compiled = prompt.compile(question=question, context=documents[0]["text"])

compile 只替换模板变量,不会调用模型。运行时取 production 是为了拿到明确选择的版本;新建版本不应直接等于上线。将返回的 prompt 对象按当前集成方式关联到 generation,才能追溯版本。

SDK 有提示词缓存,但不能据此承诺“服务完全没依赖”。首次获取、缓存刷新和故障仍要考虑。重要业务可以在启动时预取,并准备经过验证的回退提示词;同时保留日志,避免降级后仍以为跑的是新版本。

评估:从错误现场,走到回归样本

我更建议先收集二三十个真实、脱敏的案例:正常问题、模糊问题、资料缺失、工具超时,以及系统不该回答的问题。给每个样本写清预期行为,再比较模型、提示词或检索策略的变化。

检查方法 适合什么 需要防什么
确定性代码检查 JSON 格式、必填字段、业务条件 格式正确但内容错误
人工标注 专业判断、边界案例、评判标准校准 标准不统一、审核成本高
LLM-as-a-Judge 按明确量表批量评估 偏好、评分不稳定、评审费用

线上评估监测真实流量,离线实验在固定 Dataset 上比较候选版本。两者互补,不能拿一个平均分掩盖重要失败。模型当评委之前,先用人工标注校准量表;涉及权限和重要业务规则,保留明确的代码检查。

评估任务同样有成本。抽样比例、评审模型和运行频率要一起算,不要把每条请求都评一遍后,再惊讶于“监控比业务还贵”。

Cloud 还是自托管:别漏算运维

想验证效果,可以先用 Cloud 配测试项目和脱敏样本;能否上传生产数据,要先按实际数据要求判断。自托管适合希望控制部署位置与数据流向的团队,但它不等于装一个容器就结束。

当前官方自托管架构包含 Web、Worker、PostgreSQL、ClickHouse、Redis/Valkey 与 S3/Blob 存储。PostgreSQL 负责事务数据,ClickHouse 负责可观测数据,缓存服务承担队列,对象存储保存原始事件与多模态附件。

官方将 Docker Compose 定位为本地使用和测试方案,默认不提供高可用、扩缩容或备份。正式使用还要处理 TLS、鉴权、数据库与对象存储备份、队列积压、容量、保留周期以及升级恢复。

开源核心与 Enterprise 附加功能也要分清,部分能力需要许可证。Cloud 和自托管的套餐、限制与运维费用不在同一口径下,本文不拿一个未经核对的价格数字来替你算总成本。

上报前脱敏,接完后验证

Trace 可能比普通日志更敏感:用户提问、检索正文、工具参数和模型输出,可能一次凑齐。不要把真实邮箱、手机号、访问令牌、内部文件全文都当作“方便排查”的默认材料。

在数据离开应用之前完成脱敏。当前 Python SDK 的文档推荐新配置使用 mask_otel_spans,覆盖 SDK 及第三方埋点导出的 OpenTelemetry 属性;旧 mask 的覆盖范围不同。若使用自动集成,要检查它实际记录了什么,不能只检查自己手写的字段。

先接一条链路,确认正常请求、失败请求和脚本退出都能得到预期记录;检查数据内容和项目权限,再逐步扩大范围。

用 Langfuse 的第一阶段目标很朴素:用户说“这次又错了”,你能找到那次请求,解释错在哪里,并把它变成下次上线前的检查。 做到这一步,它就不只是又一块 AI 看板。

官方资料

以下资料核对日期为 2026 年 10 月 11 日;示例结构参照当前文档,完整业务接入仍需自己的项目凭证与验收。

微信扫码阅读与分享

本文链接二维码

扫码打开,再从微信右上角分享。