1. 为什么“能跑通”的 Agent 项目上线三天就变成了黑盒我最早接触 AI Agent 是在一个内部知识库问答项目上。当时用 LangChain 把检索、工具调用、生成串起来本地跑得挺顺回答质量也还行。上线第一周用户反馈开始变多“同一个问题昨天答得对今天答错了”“有时候转圈十几秒才回”“它到底查了哪些文档能不能给我看看”。我打开日志看到的只有一行AgentExecutor finished中间发生了什么完全靠猜。这就是大多数 Agent 项目从 Demo 走向生产时撞上的第一堵墙调用链是黑盒。一次用户提问背后可能经历了意图识别、多轮工具调用、向量检索、重排、大模型生成、结果校验等七八个环节任何一个环节的参数变化、超时、返回异常都会让最终结果面目全非。而传统日志只能记录“入口”和“出口”中间过程要么没打要么打了一堆没法关联的散点。Langfuse 解决的正是这个问题。它把自己定位成 LLM 应用的可观测性平台核心能力可以概括成三件事Trace追踪把一次完整请求的所有环节串成一条时间线Span跨度记录每个子步骤的输入输出、耗时、模型参数Score评分把人工反馈或自动评估的结果挂到具体的 Trace 上。这三件事组合起来Agent 的每一次“思考”都变得可回放、可对比、可量化。这篇文章适合两类人看一类是已经把 Agent 跑起来、但被线上问题搞得焦头烂额的工程师另一类是正准备搭 Agent 系统、想从一开始就把可观测性设计进去的开发者。我会从接入方式、数据模型、评测体系、并发与成本、踩坑经验几个角度把 Langfuse 在真实 Agent 项目里的用法讲透。文中涉及的具体参数和配置一部分来自官方文档一部分是我在实际项目中反复调试后总结的实践值你可以直接参考但建议结合自己的业务量级做调整。2. Langfuse 的数据模型Trace、Span、Generation 到底怎么摆很多人第一次看 Langfuse 的界面会觉得信息很多不知道从哪看起。其实它的数据模型非常清晰理解了这个模型后面接入和排查都会顺很多。2.1 一次请求就是一条 TraceTrace 是 Langfuse 里最大的容器单位对应“一次完整的用户请求”。比如用户在对话框里问了一句“帮我查一下上个月的销售数据”从这句话进入系统到最终答案返回整个过程就是一条 Trace。Trace 有一个唯一的trace_id你可以自己生成也可以让 SDK 自动生成。我习惯在请求入口处手动生成并透传这样即使跨服务调用也能把同一条 Trace 的各个部分关联起来。Trace 上可以挂很多元数据用户 ID、会话 ID、标签、环境production/staging、版本号。这些字段看起来不起眼但在排查问题时极其有用。比如你可以按user_id过滤看某个用户最近的所有请求也可以按version对比新旧版本的表现差异。2.2 Span 是 Trace 里的一个步骤Span 代表 Trace 内部的一个操作单元。在 Agent 场景里一次工具调用、一次向量检索、一次重排都可以是一个 Span。Span 可以嵌套形成树状结构。比如“工具调用”这个 Span 下面可以再挂“参数构造”和“HTTP 请求”两个子 Span。Span 最核心的价值是耗时归因。当用户抱怨“怎么这么慢”时你打开 Trace 一看如果检索 Span 花了 3 秒生成 Span 花了 8 秒那优化重点就很明确了。我见过一个项目排查了半天以为是模型慢结果发现是重排服务在高峰期排队Span 一拉出来问题一目了然。2.3 Generation 是专门给大模型调用用的 SpanGeneration 是 Span 的一种特殊类型专门用来记录大模型调用。它比普通 Span 多了几个关键字段model模型名称、model_parameters温度、top_p 等、prompt输入提示词、completion模型输出、usagetoken 消耗。这些字段是后续做成本分析和质量评估的基础。我特别想强调usage字段。很多团队做 Agent 时只关心“能不能答对”不关心“花了多少钱”。等到月底账单出来才发现某个工具调用因为提示词写得太啰嗦每次都要消耗几千 token。Langfuse 会把每次 Generation 的 token 数记录下来你可以在仪表盘上按模型、按天、按用户维度看消耗趋势。这个数据对于控制成本是刚需。2.4 三者关系用一张表说清楚层级对应概念关键字段典型用途Trace一次完整请求trace_id, user_id, session_id, tags全链路回放、用户行为分析Span请求内的一个步骤name, start_time, end_time, input, output耗时归因、步骤排查Generation一次模型调用model, prompt, completion, usage成本分析、提示词优化、质量评估理解这个模型之后接入 Langfuse 就变成了“在合适的位置埋点”的问题。埋点位置的选择直接决定了你后续能看到什么。3. 接入实战在 Agent 的关键路径上埋点Langfuse 提供了 Python 和 JavaScript 的 SDK也支持通过 API 直接写入。对于大多数 Agent 项目我推荐用 SDK 的装饰器或上下文管理器方式接入侵入性小维护成本低。3.1 环境准备与初始化先装 SDKpip install langfuse然后在项目启动时初始化客户端。这里有个细节不要在每次请求里都 new 一个客户端那样会反复建立连接既慢又浪费资源。正确做法是在应用启动时初始化一个全局客户端通过依赖注入或模块级变量共享。from langfuse import Langfuse langfuse Langfuse( public_keypk-lf-..., secret_keysk-lf-..., hosthttps://your-langfuse-host, releaseagent-v1.2.0, environmentproduction )release和environment这两个参数建议一定要填。release可以填 Git commit 短哈希或版本号environment区分生产和测试。后面做版本对比时这两个字段就是筛选依据。3.2 用装饰器给函数自动埋点Langfuse 提供了observe()装饰器加在函数上就能自动生成 Span。这是最省事的接入方式。from langfuse.decorators import observe observe() def retrieve_documents(query: str): # 向量检索逻辑 return docs observe() def call_llm(prompt: str): # 模型调用逻辑 return response装饰器会自动记录函数的入参、返回值、执行耗时。如果函数内部抛异常异常信息也会被记录到 Span 上。这一点在排查线上问题时特别有用你能直接看到是哪一步炸了、炸的时候输入是什么。不过装饰器有个局限它记录的是函数的输入输出但如果你想把模型调用的 token 数、模型名称这些信息也记下来需要配合update_current_generation或update_current_span手动补充。3.3 手动控制 Trace 的粒度对于 Agent 这种多步骤场景我建议在入口处手动创建 Trace然后在关键步骤手动创建 Span。这样粒度更可控。from langfuse import Langfuse langfuse Langfuse(...) def handle_user_query(user_id: str, query: str): trace langfuse.trace( nameagent-query, user_iduser_id, input{query: query}, metadata{channel: web} ) # 检索步骤 retrieval_span trace.span(nameretrieval, input{query: query}) docs retrieve_documents(query) retrieval_span.end(output{doc_count: len(docs)}) # 生成步骤 generation trace.generation( nameanswer-generation, modelgpt-4o, inputbuild_prompt(query, docs) ) answer call_llm(...) generation.end( outputanswer, usage{input: 1200, output: 350} ) trace.update(output{answer: answer}) return answer这种写法的好处是你能精确控制每个 Span 的边界和记录内容。比如检索 Span 里你可以把召回的文档 ID 列表记进去后面排查“为什么答错了”时就能直接看到当时检索到了什么。3.4 在 LangChain 和 LangGraph 里的接入方式如果你的 Agent 是基于 LangChain 或 LangGraph 搭的Langfuse 有现成的 CallbackHandler接入成本很低。from langfuse.callback import CallbackHandler handler CallbackHandler( public_keypk-lf-..., secret_keysk-lf-..., hosthttps://your-langfuse-host, session_iduser-session-123, user_iduser-456 ) chain.invoke({input: query}, config{callbacks: [handler]})LangGraph 的接入类似把 handler 传进config即可。它会自动把每个节点执行、每次模型调用都记录成 Span 和 Generation。我实测下来LangGraph 的节点名称会自动成为 Span 名称所以给节点起个好名字很重要别用node_1、node_2这种后面看 Trace 时会很痛苦。提示CallbackHandler 在高并发场景下会频繁创建对象建议复用或使用连接池。如果 QPS 很高可以考虑异步写入模式避免阻塞主流程。4. 评测体系让 Agent 的回答质量从“感觉还行”变成“有数可查”可观测性解决的是“看得见”的问题但看得见不等于管得住。Agent 的回答质量到底怎么样需要一套评测体系来量化。Langfuse 的 Score 功能就是干这个的。4.1 人工评分最直接但最贵的方式最简单的做法是在 Trace 上挂人工评分。Langfuse 支持在界面上直接给某条 Trace 打分也支持通过 API 写入。langfuse.score( trace_idtrace.id, nameuser-feedback, value1, # 1 表示好评0 表示差评 comment回答准确引用了正确的文档 )人工评分的优点是准确缺点是贵且慢。我的经验是不要对所有请求都做人工评分而是抽样。按用户反馈点赞/点踩自动挂分再定期人工复核一部分。这样成本可控数据也有代表性。4.2 自动评估用模型给模型打分对于需要规模化评估的场景可以用另一个模型来做自动评分。Langfuse 本身不提供评估模型但你可以自己写评估逻辑然后把结果写回 Score。常见的自动评估维度有几个相关性回答是否切题有没有答非所问忠实度回答是否基于检索到的文档有没有编造完整性是否覆盖了问题的所有方面格式合规是否满足输出格式要求比如 JSON 结构def evaluate_answer(query, answer, docs): eval_prompt f 请评估以下回答的质量从相关性、忠实度、完整性三个维度打分1-5分。 问题{query} 检索文档{docs} 回答{answer} 请以 JSON 格式输出评分和理由。 result call_llm(eval_prompt) scores parse_json(result) for dim, score in scores.items(): langfuse.score( trace_idtrace.id, namefauto-{dim}, valuescore[value], commentscore[reason] )这里有个坑要注意评估模型和被评估模型不要用同一个。用同一个模型评估自己容易产生“自我偏好”分数虚高。我一般用能力更强的模型做评估或者用不同厂商的模型交叉评估。4.3 用数据集做回归测试Agent 系统最怕的是“改了一个提示词修好了 A 问题弄坏了 B 问题”。Langfuse 的 Dataset 功能可以帮你做回归测试。做法是把一批有代表性的问题包括边界 case整理成数据集每次发版前跑一遍对比新旧版本的评分。如果某个维度的分数明显下降就说明这次改动有问题。dataset langfuse.get_dataset(agent-regression-v1) for item in dataset.items: # 用新版本跑一遍 output run_agent(item.input) # 挂到数据集条目上 item.link( trace_or_observationtrace, run_namev1.3.0-release )跑完之后在 Langfuse 界面上可以按run_name对比不同版本的评分。这个流程跑顺了之后发版心里就有底了。4.4 评测频率与成本控制自动评估是要花钱的每次评估都是一次模型调用。我的建议是场景评估频率评估方式日常线上抽样 5%-10%自动评估 用户反馈发版前全量数据集自动评估 人工抽检重大改动全量数据集 线上灰度自动评估 人工全检这样既能控制成本又能在关键节点拿到足够的质量信号。5. 高并发下的 Langfuse别让可观测性拖垮主流程Agent 系统本身就比普通接口重一次请求可能涉及多次模型调用和工具调用延迟本来就高。如果可观测性的埋点再同步阻塞主流程用户体验会更差。这一节讲几个高并发场景下的实践要点。5.1 异步写入与批量上报Langfuse SDK 默认是同步发送数据的也就是说每次trace.span()、generation.end()都会触发一次网络请求。在低 QPS 下没问题但 QPS 一高这些请求会堆积拖慢主流程。解决办法是开启异步模式。Langfuse 的 Python SDK 支持通过环境变量或初始化参数配置异步langfuse Langfuse( public_key..., secret_key..., host..., flush_at100, # 攒够 100 条再发 flush_interval1.0, # 或者每 1 秒发一次 threads4 # 后台发送线程数 )flush_at和flush_interval是两个关键参数。flush_at控制批量大小flush_interval控制最大等待时间。两者是“或”的关系满足任一条件就发送。我一般设flush_at50、flush_interval2.0在延迟和数据实时性之间取平衡。注意异步模式下如果应用崩溃缓冲区里没发出去的数据会丢失。对于关键业务建议在请求结束时手动调用langfuse.flush()确保数据落库。5.2 采样不是所有请求都值得全量记录高并发场景下全量记录 Trace 的成本很高包括存储成本、网络成本、以及 Langfuse 服务端的处理成本。这时候需要采样。采样的策略有几种固定比例采样比如只记录 20% 的请求。实现简单但可能漏掉关键 case。基于错误采样正常请求少记出错请求全记。这个策略最实用因为排查问题主要看出错的。基于用户采样对 VIP 用户全量记录普通用户抽样。基于延迟采样慢请求全记快请求抽样。我通常组合使用错误请求 100% 记录慢请求超过 P95100% 记录正常请求按 10% 采样。这样既控制了数据量又保证了问题排查时有足够的信息。import random def should_sample(trace_metadata): if trace_metadata.get(has_error): return True if trace_metadata.get(latency_ms, 0) 5000: return True return random.random() 0.15.3 敏感信息脱敏Agent 系统处理的往往是用户真实数据直接把这些数据写到 Langfuse 上是有合规风险的。Langfuse 支持在 SDK 层面做脱敏。from langfuse import Langfuse def mask_sensitive(data): # 对手机号、邮箱、身份证号做脱敏 if isinstance(data, str): data re.sub(r\d{11}, ***********, data) data re.sub(r[\w\.-][\w\.-], ******, data) return data langfuse Langfuse( ..., maskmask_sensitive )mask函数会在数据发送前被调用你可以在这里做各种脱敏处理。这一步千万别省尤其是涉及用户隐私的业务。5.4 并发压测下的表现我在一个 QPS 200 左右的 Agent 服务上做过压测对比开启和关闭 Langfuse 的延迟差异。结论是开启异步写入后P99 延迟增加在 5% 以内如果不开异步P99 延迟会增加 30% 以上因为同步网络请求在高峰期会排队。所以结论很明确生产环境一定要开异步并且做好采样和脱敏。可观测性是为了让系统更可控不能反过来成为系统的负担。6. 踩坑记录那些文档里不会写的细节这一节记录几个我在实际项目中踩过的坑都是文档里不会写、但实际会遇到的。6.1 Trace 丢失请求还没结束进程就退出了Agent 服务如果用 Serverless 或短生命周期容器部署请求处理完进程可能立刻退出导致 Langfuse 缓冲区里的数据还没发出去就丢了。表现就是 Langfuse 上只能看到部分 Trace或者干脆看不到。解决办法是在请求处理结束时显式 flushtry: result handle_user_query(user_id, query) finally: langfuse.flush()flush()会阻塞直到缓冲区清空。虽然会增加一点延迟但能保证数据不丢。如果对延迟极其敏感可以把这个 flush 放到后台线程里做。6.2 Span 嵌套层级过深界面加载慢Agent 如果步骤很多Span 嵌套层级可能达到十几层。Langfuse 界面在渲染深层嵌套时会变慢排查问题时展开也很费劲。我的做法是控制嵌套深度一般不超过 5 层。对于特别复杂的流程把一些细节步骤合并成一个 Span只在需要时展开。比如“工具调用”这个 Span 下面不需要把参数构造、序列化、HTTP 请求都拆开合并成一个 Span 记录关键信息就够了。6.3 时间戳不一致导致 Trace 顺序错乱Langfuse 依赖时间戳来排序 Span。如果 Agent 服务部署在多台机器上机器时间不同步Span 的顺序就会乱看起来像是“后面的步骤先执行了”。解决办法是确保所有机器开启 NTP 时间同步。另外Langfuse SDK 支持传入自定义时间戳如果确实有时序问题可以在埋点时手动指定。6.4 评分数据写入失败但没报错Langfuse 的score()方法在写入失败时默认不抛异常只是静默失败。这导致我以为评分写进去了实际上没有。排查方法是看 SDK 的日志。把日志级别调到 DEBUG能看到每次写入的结果。如果发现失败检查网络、认证信息、以及 trace_id 是否正确。6.5 模型名称不规范导致统计混乱generation.end()里的model字段如果写法不统一比如有时写gpt-4o有时写gpt-4o-2024-08-06统计时会被当成两个模型成本分析就不准了。建议在项目里维护一个模型名称常量表所有埋点统一引用。这样统计维度才干净。7. 从可观测性到持续优化把数据用起来埋点、评测、排查都做完之后最后一步是把这些数据转化成实际的优化动作。否则可观测性就只是“看个热闹”。7.1 用 Trace 对比找出提示词退化每次改提示词后用同一批测试问题跑一遍然后在 Langfuse 上按release筛选对比新旧版本的评分和输出。我遇到过好几次“改了一个词整体分数没变但某类问题的准确率掉了 20%”的情况全靠这个对比发现。7.2 用耗时分布定位性能瓶颈在 Langfuse 仪表盘上看 Span 的耗时分布找出 P95 最高的几个 Span。这些就是优化重点。常见的瓶颈包括向量检索没有加缓存、重排服务并发度不够、模型调用没有做流式输出等。7.3 用用户反馈驱动数据集迭代用户点踩的 Trace 是最宝贵的数据。定期把这些 Trace 整理出来补充到回归测试数据集里。这样数据集会越来越贴近真实场景回归测试的有效性也会越来越高。7.4 成本优化的几个切入点通过 Langfuse 的 token 统计我总结出几个成本优化的方向提示词精简很多提示词里有大量冗余说明精简后 token 数能降 30% 以上缓存复用相同或相似的查询结果缓存减少重复模型调用模型分级简单任务用小模型复杂任务用大模型通过路由分发输出长度控制设置合理的max_tokens避免模型“话痨”这些优化做完成本能降一半以上而质量通过评测体系监控不会明显下降。8. 写在最后可观测性是 Agent 工程化的入场券我从最早用print调试 Agent到后来用日志系统再到接入 Langfuse最大的感受是Agent 系统的复杂度不在于“能不能跑”而在于“跑得好不好、为什么好、为什么不好”。没有可观测性这些问题全靠猜有了可观测性每个问题都能定位到具体的 Trace、具体的 Span、具体的参数。Langfuse 不是唯一的选择但它在这个方向上的数据模型设计得比较合理接入成本也低。如果你正在搭 Agent 系统我的建议是从第一天就把可观测性设计进去别等到线上出问题了再补。补的时候你会发现很多关键信息当时没记事后根本查不到。最后分享一个我自己的习惯每次发版前除了跑回归测试我还会随机抽 20 条线上 Trace 人工看一遍。这个动作花不了多少时间但经常能发现一些自动化指标覆盖不到的问题比如回答语气不对、格式虽然合规但可读性差、某些边界 case 处理得生硬。这些细节才是 Agent 从“能用”到“好用”的关键。 
长河君 · 2026/10/5 18:49:37 · 阅读 1.2k Langfuse 可观测性实战：让 AI Agent 从黑盒到透明
1. 为什么“能跑通”的 Agent 项目上线三天就变成了黑盒我最早接触 AI Agent 是在一个内部知识库问答项目上。当时用 LangChain 把检索、工具调用、生成串起来本地跑得挺顺回答质量也还行。上线第一周用户反馈开始变多“同一个问题昨天答得对今天答错了”“有时候转圈十几秒才回”“它到底查了哪些文档能不能给我看看”。我打开日志看到的只有一行AgentExecutor finished中间发生了什么完全靠猜。这就是大多数 Agent 项目从 Demo 走向生产时撞上的第一堵墙调用链是黑盒。一次用户提问背后可能经历了意图识别、多轮工具调用、向量检索、重排、大模型生成、结果校验等七八个环节任何一个环节的参数变化、超时、返回异常都会让最终结果面目全非。而传统日志只能记录“入口”和“出口”中间过程要么没打要么打了一堆没法关联的散点。Langfuse 解决的正是这个问题。它把自己定位成 LLM 应用的可观测性平台核心能力可以概括成三件事Trace追踪把一次完整请求的所有环节串成一条时间线Span跨度记录每个子步骤的输入输出、耗时、模型参数Score评分把人工反馈或自动评估的结果挂到具体的 Trace 上。这三件事组合起来Agent 的每一次“思考”都变得可回放、可对比、可量化。这篇文章适合两类人看一类是已经把 Agent 跑起来、但被线上问题搞得焦头烂额的工程师另一类是正准备搭 Agent 系统、想从一开始就把可观测性设计进去的开发者。我会从接入方式、数据模型、评测体系、并发与成本、踩坑经验几个角度把 Langfuse 在真实 Agent 项目里的用法讲透。文中涉及的具体参数和配置一部分来自官方文档一部分是我在实际项目中反复调试后总结的实践值你可以直接参考但建议结合自己的业务量级做调整。2. Langfuse 的数据模型Trace、Span、Generation 到底怎么摆很多人第一次看 Langfuse 的界面会觉得信息很多不知道从哪看起。其实它的数据模型非常清晰理解了这个模型后面接入和排查都会顺很多。2.1 一次请求就是一条 TraceTrace 是 Langfuse 里最大的容器单位对应“一次完整的用户请求”。比如用户在对话框里问了一句“帮我查一下上个月的销售数据”从这句话进入系统到最终答案返回整个过程就是一条 Trace。Trace 有一个唯一的trace_id你可以自己生成也可以让 SDK 自动生成。我习惯在请求入口处手动生成并透传这样即使跨服务调用也能把同一条 Trace 的各个部分关联起来。Trace 上可以挂很多元数据用户 ID、会话 ID、标签、环境production/staging、版本号。这些字段看起来不起眼但在排查问题时极其有用。比如你可以按user_id过滤看某个用户最近的所有请求也可以按version对比新旧版本的表现差异。2.2 Span 是 Trace 里的一个步骤Span 代表 Trace 内部的一个操作单元。在 Agent 场景里一次工具调用、一次向量检索、一次重排都可以是一个 Span。Span 可以嵌套形成树状结构。比如“工具调用”这个 Span 下面可以再挂“参数构造”和“HTTP 请求”两个子 Span。Span 最核心的价值是耗时归因。当用户抱怨“怎么这么慢”时你打开 Trace 一看如果检索 Span 花了 3 秒生成 Span 花了 8 秒那优化重点就很明确了。我见过一个项目排查了半天以为是模型慢结果发现是重排服务在高峰期排队Span 一拉出来问题一目了然。2.3 Generation 是专门给大模型调用用的 SpanGeneration 是 Span 的一种特殊类型专门用来记录大模型调用。它比普通 Span 多了几个关键字段model模型名称、model_parameters温度、top_p 等、prompt输入提示词、completion模型输出、usagetoken 消耗。这些字段是后续做成本分析和质量评估的基础。我特别想强调usage字段。很多团队做 Agent 时只关心“能不能答对”不关心“花了多少钱”。等到月底账单出来才发现某个工具调用因为提示词写得太啰嗦每次都要消耗几千 token。Langfuse 会把每次 Generation 的 token 数记录下来你可以在仪表盘上按模型、按天、按用户维度看消耗趋势。这个数据对于控制成本是刚需。2.4 三者关系用一张表说清楚层级对应概念关键字段典型用途Trace一次完整请求trace_id, user_id, session_id, tags全链路回放、用户行为分析Span请求内的一个步骤name, start_time, end_time, input, output耗时归因、步骤排查Generation一次模型调用model, prompt, completion, usage成本分析、提示词优化、质量评估理解这个模型之后接入 Langfuse 就变成了“在合适的位置埋点”的问题。埋点位置的选择直接决定了你后续能看到什么。3. 接入实战在 Agent 的关键路径上埋点Langfuse 提供了 Python 和 JavaScript 的 SDK也支持通过 API 直接写入。对于大多数 Agent 项目我推荐用 SDK 的装饰器或上下文管理器方式接入侵入性小维护成本低。3.1 环境准备与初始化先装 SDKpip install langfuse然后在项目启动时初始化客户端。这里有个细节不要在每次请求里都 new 一个客户端那样会反复建立连接既慢又浪费资源。正确做法是在应用启动时初始化一个全局客户端通过依赖注入或模块级变量共享。from langfuse import Langfuse langfuse Langfuse( public_keypk-lf-..., secret_keysk-lf-..., hosthttps://your-langfuse-host, releaseagent-v1.2.0, environmentproduction )release和environment这两个参数建议一定要填。release可以填 Git commit 短哈希或版本号environment区分生产和测试。后面做版本对比时这两个字段就是筛选依据。3.2 用装饰器给函数自动埋点Langfuse 提供了observe()装饰器加在函数上就能自动生成 Span。这是最省事的接入方式。from langfuse.decorators import observe observe() def retrieve_documents(query: str): # 向量检索逻辑 return docs observe() def call_llm(prompt: str): # 模型调用逻辑 return response装饰器会自动记录函数的入参、返回值、执行耗时。如果函数内部抛异常异常信息也会被记录到 Span 上。这一点在排查线上问题时特别有用你能直接看到是哪一步炸了、炸的时候输入是什么。不过装饰器有个局限它记录的是函数的输入输出但如果你想把模型调用的 token 数、模型名称这些信息也记下来需要配合update_current_generation或update_current_span手动补充。3.3 手动控制 Trace 的粒度对于 Agent 这种多步骤场景我建议在入口处手动创建 Trace然后在关键步骤手动创建 Span。这样粒度更可控。from langfuse import Langfuse langfuse Langfuse(...) def handle_user_query(user_id: str, query: str): trace langfuse.trace( nameagent-query, user_iduser_id, input{query: query}, metadata{channel: web} ) # 检索步骤 retrieval_span trace.span(nameretrieval, input{query: query}) docs retrieve_documents(query) retrieval_span.end(output{doc_count: len(docs)}) # 生成步骤 generation trace.generation( nameanswer-generation, modelgpt-4o, inputbuild_prompt(query, docs) ) answer call_llm(...) generation.end( outputanswer, usage{input: 1200, output: 350} ) trace.update(output{answer: answer}) return answer这种写法的好处是你能精确控制每个 Span 的边界和记录内容。比如检索 Span 里你可以把召回的文档 ID 列表记进去后面排查“为什么答错了”时就能直接看到当时检索到了什么。3.4 在 LangChain 和 LangGraph 里的接入方式如果你的 Agent 是基于 LangChain 或 LangGraph 搭的Langfuse 有现成的 CallbackHandler接入成本很低。from langfuse.callback import CallbackHandler handler CallbackHandler( public_keypk-lf-..., secret_keysk-lf-..., hosthttps://your-langfuse-host, session_iduser-session-123, user_iduser-456 ) chain.invoke({input: query}, config{callbacks: [handler]})LangGraph 的接入类似把 handler 传进config即可。它会自动把每个节点执行、每次模型调用都记录成 Span 和 Generation。我实测下来LangGraph 的节点名称会自动成为 Span 名称所以给节点起个好名字很重要别用node_1、node_2这种后面看 Trace 时会很痛苦。提示CallbackHandler 在高并发场景下会频繁创建对象建议复用或使用连接池。如果 QPS 很高可以考虑异步写入模式避免阻塞主流程。4. 评测体系让 Agent 的回答质量从“感觉还行”变成“有数可查”可观测性解决的是“看得见”的问题但看得见不等于管得住。Agent 的回答质量到底怎么样需要一套评测体系来量化。Langfuse 的 Score 功能就是干这个的。4.1 人工评分最直接但最贵的方式最简单的做法是在 Trace 上挂人工评分。Langfuse 支持在界面上直接给某条 Trace 打分也支持通过 API 写入。langfuse.score( trace_idtrace.id, nameuser-feedback, value1, # 1 表示好评0 表示差评 comment回答准确引用了正确的文档 )人工评分的优点是准确缺点是贵且慢。我的经验是不要对所有请求都做人工评分而是抽样。按用户反馈点赞/点踩自动挂分再定期人工复核一部分。这样成本可控数据也有代表性。4.2 自动评估用模型给模型打分对于需要规模化评估的场景可以用另一个模型来做自动评分。Langfuse 本身不提供评估模型但你可以自己写评估逻辑然后把结果写回 Score。常见的自动评估维度有几个相关性回答是否切题有没有答非所问忠实度回答是否基于检索到的文档有没有编造完整性是否覆盖了问题的所有方面格式合规是否满足输出格式要求比如 JSON 结构def evaluate_answer(query, answer, docs): eval_prompt f 请评估以下回答的质量从相关性、忠实度、完整性三个维度打分1-5分。 问题{query} 检索文档{docs} 回答{answer} 请以 JSON 格式输出评分和理由。 result call_llm(eval_prompt) scores parse_json(result) for dim, score in scores.items(): langfuse.score( trace_idtrace.id, namefauto-{dim}, valuescore[value], commentscore[reason] )这里有个坑要注意评估模型和被评估模型不要用同一个。用同一个模型评估自己容易产生“自我偏好”分数虚高。我一般用能力更强的模型做评估或者用不同厂商的模型交叉评估。4.3 用数据集做回归测试Agent 系统最怕的是“改了一个提示词修好了 A 问题弄坏了 B 问题”。Langfuse 的 Dataset 功能可以帮你做回归测试。做法是把一批有代表性的问题包括边界 case整理成数据集每次发版前跑一遍对比新旧版本的评分。如果某个维度的分数明显下降就说明这次改动有问题。dataset langfuse.get_dataset(agent-regression-v1) for item in dataset.items: # 用新版本跑一遍 output run_agent(item.input) # 挂到数据集条目上 item.link( trace_or_observationtrace, run_namev1.3.0-release )跑完之后在 Langfuse 界面上可以按run_name对比不同版本的评分。这个流程跑顺了之后发版心里就有底了。4.4 评测频率与成本控制自动评估是要花钱的每次评估都是一次模型调用。我的建议是场景评估频率评估方式日常线上抽样 5%-10%自动评估 用户反馈发版前全量数据集自动评估 人工抽检重大改动全量数据集 线上灰度自动评估 人工全检这样既能控制成本又能在关键节点拿到足够的质量信号。5. 高并发下的 Langfuse别让可观测性拖垮主流程Agent 系统本身就比普通接口重一次请求可能涉及多次模型调用和工具调用延迟本来就高。如果可观测性的埋点再同步阻塞主流程用户体验会更差。这一节讲几个高并发场景下的实践要点。5.1 异步写入与批量上报Langfuse SDK 默认是同步发送数据的也就是说每次trace.span()、generation.end()都会触发一次网络请求。在低 QPS 下没问题但 QPS 一高这些请求会堆积拖慢主流程。解决办法是开启异步模式。Langfuse 的 Python SDK 支持通过环境变量或初始化参数配置异步langfuse Langfuse( public_key..., secret_key..., host..., flush_at100, # 攒够 100 条再发 flush_interval1.0, # 或者每 1 秒发一次 threads4 # 后台发送线程数 )flush_at和flush_interval是两个关键参数。flush_at控制批量大小flush_interval控制最大等待时间。两者是“或”的关系满足任一条件就发送。我一般设flush_at50、flush_interval2.0在延迟和数据实时性之间取平衡。注意异步模式下如果应用崩溃缓冲区里没发出去的数据会丢失。对于关键业务建议在请求结束时手动调用langfuse.flush()确保数据落库。5.2 采样不是所有请求都值得全量记录高并发场景下全量记录 Trace 的成本很高包括存储成本、网络成本、以及 Langfuse 服务端的处理成本。这时候需要采样。采样的策略有几种固定比例采样比如只记录 20% 的请求。实现简单但可能漏掉关键 case。基于错误采样正常请求少记出错请求全记。这个策略最实用因为排查问题主要看出错的。基于用户采样对 VIP 用户全量记录普通用户抽样。基于延迟采样慢请求全记快请求抽样。我通常组合使用错误请求 100% 记录慢请求超过 P95100% 记录正常请求按 10% 采样。这样既控制了数据量又保证了问题排查时有足够的信息。import random def should_sample(trace_metadata): if trace_metadata.get(has_error): return True if trace_metadata.get(latency_ms, 0) 5000: return True return random.random() 0.15.3 敏感信息脱敏Agent 系统处理的往往是用户真实数据直接把这些数据写到 Langfuse 上是有合规风险的。Langfuse 支持在 SDK 层面做脱敏。from langfuse import Langfuse def mask_sensitive(data): # 对手机号、邮箱、身份证号做脱敏 if isinstance(data, str): data re.sub(r\d{11}, ***********, data) data re.sub(r[\w\.-][\w\.-], ******, data) return data langfuse Langfuse( ..., maskmask_sensitive )mask函数会在数据发送前被调用你可以在这里做各种脱敏处理。这一步千万别省尤其是涉及用户隐私的业务。5.4 并发压测下的表现我在一个 QPS 200 左右的 Agent 服务上做过压测对比开启和关闭 Langfuse 的延迟差异。结论是开启异步写入后P99 延迟增加在 5% 以内如果不开异步P99 延迟会增加 30% 以上因为同步网络请求在高峰期会排队。所以结论很明确生产环境一定要开异步并且做好采样和脱敏。可观测性是为了让系统更可控不能反过来成为系统的负担。6. 踩坑记录那些文档里不会写的细节这一节记录几个我在实际项目中踩过的坑都是文档里不会写、但实际会遇到的。6.1 Trace 丢失请求还没结束进程就退出了Agent 服务如果用 Serverless 或短生命周期容器部署请求处理完进程可能立刻退出导致 Langfuse 缓冲区里的数据还没发出去就丢了。表现就是 Langfuse 上只能看到部分 Trace或者干脆看不到。解决办法是在请求处理结束时显式 flushtry: result handle_user_query(user_id, query) finally: langfuse.flush()flush()会阻塞直到缓冲区清空。虽然会增加一点延迟但能保证数据不丢。如果对延迟极其敏感可以把这个 flush 放到后台线程里做。6.2 Span 嵌套层级过深界面加载慢Agent 如果步骤很多Span 嵌套层级可能达到十几层。Langfuse 界面在渲染深层嵌套时会变慢排查问题时展开也很费劲。我的做法是控制嵌套深度一般不超过 5 层。对于特别复杂的流程把一些细节步骤合并成一个 Span只在需要时展开。比如“工具调用”这个 Span 下面不需要把参数构造、序列化、HTTP 请求都拆开合并成一个 Span 记录关键信息就够了。6.3 时间戳不一致导致 Trace 顺序错乱Langfuse 依赖时间戳来排序 Span。如果 Agent 服务部署在多台机器上机器时间不同步Span 的顺序就会乱看起来像是“后面的步骤先执行了”。解决办法是确保所有机器开启 NTP 时间同步。另外Langfuse SDK 支持传入自定义时间戳如果确实有时序问题可以在埋点时手动指定。6.4 评分数据写入失败但没报错Langfuse 的score()方法在写入失败时默认不抛异常只是静默失败。这导致我以为评分写进去了实际上没有。排查方法是看 SDK 的日志。把日志级别调到 DEBUG能看到每次写入的结果。如果发现失败检查网络、认证信息、以及 trace_id 是否正确。6.5 模型名称不规范导致统计混乱generation.end()里的model字段如果写法不统一比如有时写gpt-4o有时写gpt-4o-2024-08-06统计时会被当成两个模型成本分析就不准了。建议在项目里维护一个模型名称常量表所有埋点统一引用。这样统计维度才干净。7. 从可观测性到持续优化把数据用起来埋点、评测、排查都做完之后最后一步是把这些数据转化成实际的优化动作。否则可观测性就只是“看个热闹”。7.1 用 Trace 对比找出提示词退化每次改提示词后用同一批测试问题跑一遍然后在 Langfuse 上按release筛选对比新旧版本的评分和输出。我遇到过好几次“改了一个词整体分数没变但某类问题的准确率掉了 20%”的情况全靠这个对比发现。7.2 用耗时分布定位性能瓶颈在 Langfuse 仪表盘上看 Span 的耗时分布找出 P95 最高的几个 Span。这些就是优化重点。常见的瓶颈包括向量检索没有加缓存、重排服务并发度不够、模型调用没有做流式输出等。7.3 用用户反馈驱动数据集迭代用户点踩的 Trace 是最宝贵的数据。定期把这些 Trace 整理出来补充到回归测试数据集里。这样数据集会越来越贴近真实场景回归测试的有效性也会越来越高。7.4 成本优化的几个切入点通过 Langfuse 的 token 统计我总结出几个成本优化的方向提示词精简很多提示词里有大量冗余说明精简后 token 数能降 30% 以上缓存复用相同或相似的查询结果缓存减少重复模型调用模型分级简单任务用小模型复杂任务用大模型通过路由分发输出长度控制设置合理的max_tokens避免模型“话痨”这些优化做完成本能降一半以上而质量通过评测体系监控不会明显下降。8. 写在最后可观测性是 Agent 工程化的入场券我从最早用print调试 Agent到后来用日志系统再到接入 Langfuse最大的感受是Agent 系统的复杂度不在于“能不能跑”而在于“跑得好不好、为什么好、为什么不好”。没有可观测性这些问题全靠猜有了可观测性每个问题都能定位到具体的 Trace、具体的 Span、具体的参数。Langfuse 不是唯一的选择但它在这个方向上的数据模型设计得比较合理接入成本也低。如果你正在搭 Agent 系统我的建议是从第一天就把可观测性设计进去别等到线上出问题了再补。补的时候你会发现很多关键信息当时没记事后根本查不到。最后分享一个我自己的习惯每次发版前除了跑回归测试我还会随机抽 20 条线上 Trace 人工看一遍。这个动作花不了多少时间但经常能发现一些自动化指标覆盖不到的问题比如回答语气不对、格式虽然合规但可读性差、某些边界 case 处理得生硬。这些细节才是 Agent 从“能用”到“好用”的关键。