前阵子我调试一个基于LangGraph的AI Agent系统时遇到一个让人头疼的问题。用户提了个相当简单的需求Agent却先后调了两次工具、三次模型才完成回答。日志里能看到的只有一行行零散的模型调用记录谁在什么时候调了哪个模型、为什么调、中间到底发生了什么完全是一团黑。我把Langfuse接进系统之后才第一次真正体会到“全链路可观测”和“模型调用日志”之间的差距。这篇文章就把这次Langfuse在AI Agent系统里的工程实践完整记录下来先讲Agent场景下传统日志为什么不够用再拆解Langfuse的追踪模型和接入方式然后覆盖流式输出、并发、本地模型这些实战场景的埋点细节最后分享怎么把它当评测平台用。如果你正在搭Agent或者系统已经跑起来但出了问题只能靠猜这篇应该帮得上忙。1. AI Agent系统的可观测性困境为什么“模型调用日志”不够用了1.1 从单次问答到多步协作Agent处理链路的复杂度跃迁传统LLM应用比如一个简单的聊天机器人一次请求就是“用户问题进、模型回答出”的闭环。这种场景下可观测要做的无非是记录prompt、completion、token数和耗时一个接口日志就差不多了。但Agent系统直接把复杂度拉高了一个数量级。它不是单次模型调用而是把模型调用变成编排过程——模型自己决定下一步干什么、调哪个工具、看完工具结果再决定还要不要再调一次。每一次用户请求背后可能是好几轮模型决策每轮决策里又嵌套工具调用、知识库检索、代码执行。我用实际场景说明。假设你搭了一个基于FastAPILangChainLangGraph的Agent收到的问题是“帮我把上周周报里提到的三个项目按风险从高到低排序”。内部实际跑起来大概是先调工具读取周报文件模型分析内容、提取项目名再调一次工具去数据库查询风险数据模型对比排序最后格式化输出。这一趟下来模型至少被调了四次、工具被调了两次。如果中间某一步返回了脏数据模型还要重新规划、重新调用次数直接翻倍。这时候“记录每一次模型调用”只能告诉你“模型调了5次用了5000个token”。真正排查问题需要的是这5次调用分别发生在哪个环节哪一次是为了补救上一步的异常第二次调用为什么要花40秒是模型推理慢还是在等工具返回两次工具调用里哪次返回了错误结果导致模型重试重试烧掉多少成本用户的问题最终到底有没有被解决模型在哪个环节做了错误决策这些就是可观测性要回答的问题。可观测不是“有日志”而是让你在事后能完整重建一次请求内部的决策路径。1.2 传统日志方案的三宗罪上下文丢失、成本无法归因、评估靠感觉在接入Langfuse之前我系统里也不是没有日志。普通应用日志、OpenAI调用记录我都有但真到了排查问题的时候每一样都差点意思。第一宗罪是上下文丢失。应用日志按行记、按进程记一条模型调用日志里不会自带“这是Agent决策链的第几步”。你只知道某个时刻有一条请求进了模型服务但这条请求是哪个用户触发的、Agent当时处于什么状态、上一步工具返回了什么全都对不上。想靠日志反推一次完整请求只能靠时间戳硬猜一旦有并发请求直接就乱了。第二宗罪是成本无法归因。单次模型调用日志有token数和花费但一次Agent请求可能涉及主模型、工具调用模型、总结模型多次调用。总花费出来了具体哪一步最烧钱、是不是某个环节在反复空转重试完全看不出来。我排过一个线上案例某天Agent服务成本比平时高了3倍从聚合日志里只能看出“调用量涨了”看不出是因为某个工具开始频繁报错、导致模型反复重试。后来按trace把一次请求的成本拆开才看清一次正常请求里重试消耗的token占总消耗60%以上。第三宗罪是评估靠感觉。没有结构化追踪数据就回答不了“这次改动到底让系统变好还是变差了”。改了prompt、换了工具描述、升级了模型效果怎么样只能拉几个人试几个case凭直觉判断。这也是我后来把Langfuse同时当评测平台用的原因——没有数据做回归对比Agent系统的迭代就是盲人摸象。1.3 引入Langfuse前先想清楚的几个问题正式接入Langfuse之前我逼自己先回答了三个问题免得做到一半才发现方向不对。第一观测粒度要到什么程度我只关心模型输入输出还是要把工具调用、知识库检索、Agent状态流转全纳进来我的结论是至少要到trace级别。Agent场景里单条generation记录没有独立价值只有挂到一次完整请求的trace上才有意义。第二现有技术栈能不能兼容我的系统是FastAPILangChainLangGraphLangfuse对这两者有现成集成。如果团队用Spring AILangfuse也有Java SDK。如果Agent代码全是手写、只在最后调OpenAI那也可以用SDK手动埋点。兼容性在调研阶段可以放心。第三数据安全怎么处理生产环境真实用户数据直接送观测平台是危险的必须在埋点层做脱敏和字段过滤。这个问题我会在第六章详细讲。想清楚这三件事我才开始动手接。2. Langfuse的核心追踪模型认识Trace、Span、Observation与Generation2.1 四个基础概念的关系与适用场景Langfuse组织数据的方式很清晰用Trace追踪、Span跨度、Observation观测、Generation生成这四个核心概念。刚开始我对这四个概念的边界也是模糊的实际用下来我的理解是Trace是一次完整请求的根。从用户问题进来到最终回答返回整个过程是一条Trace。在Langfuse界面上一条Trace就是一条可展开的时间线。Span是Trace内部的一个阶段或子操作。比如“调用搜索工具”“读取数据库”“更新Agent状态”。Span可以嵌套子Span挂在父Span下面。Observation是个抽象概念。Span和Generation都属于Observation表示一个可以被观测、可以分配token和耗时的操作单元。Generation是Observation的特殊子类专门表示一次模型调用LLM生成或Embedding。它额外记录model name、prompt、completion、token usage、cost这些模型相关字段。用生活类比来记把一次Agent请求想象成外卖订单。Trace是整个订单Span是“商家出餐”“骑手配送”“用户签收”这些阶段Generation则是“商家出餐”这个阶段里“炒了一盘宫保鸡丁”——它是唯一需要关心食材成本、火候参数的具体动作。新手最常犯的错是纠结“这个该用Span还是Generation”。判断标准一句话只要是调用模型、嵌入、重排序这些会产生token消耗的操作用Generation只是流程阶段、工具调用、内部函数用Span。嵌套关系上用LangChain的链式或图式结构会自动生成层级手动埋点时控制好start和end就行。2.2 追踪数据是怎么组织的一条完整请求的生命周期拿LangGraph的Agent为例一次用户请求进来后Langfuse的回调处理器会自动创建一条Trace把Graph节点的执行过程映射成Span。真正调用模型的那一步会自动创建一条Generation关联到model、prompt、completion、tokens、latency这些字段。数据上送是异步的不会阻塞业务请求。Langfuse SDK在后台按批次POST数据到服务端如果你的业务是高频低延迟接口这个异步上送基本不影响响应时间。但异步有个副作用——崩溃即丢失。如果进程在数据上送之前崩了这一段追踪数据就没了。大部分场景可以接受如果有强审计需求建议在关键节点同步flush或定期强制发送。数据组织好之后Langfuse UI里就能看到一条完整时间线每一层都能展开看prompt、模型输出、token用量、耗时。到了这一步排查问题就从容多了。2.3 统一抽象从OpenAI SDK到本地模型的差异怎么抹平Langfuse做了一件很关键的事对不同模型来源做了统一抽象。无论模型是OpenAI、Anthropic、千问、DeepSeek还是本地部署的Ollama、LM Studio只要接入Langfuse调用记录都会被标准化成同一种Generation格式。这个抽象在Agent系统里价值特别明显。一个Agent可能同时用云模型做规划和总结、本地模型做信息提取或分类。没统一抽象的话每个数据源格式都不一样没法横向对比。统一之后我能在同一个界面里直接对比“同一个环节用千问和用GPT-4谁的延迟低、谁的token多”。这也是帮我下决心接入Langfuse的另一个原因——不用自己写一套多模型日志标准化管道了。3. 从监听回调到全链路追踪LangChain/LangGraph接入Langfuse的实操记录3.1 最简接入回调处理器一行搞定LangChain如果系统用的是LangChain接入Langfuse成本最低因为官方提供了现成的CallbackHandler。from langfuse.callback import CallbackHandler langfuse_handler CallbackHandler( trace_nameagent_request, user_iduser_123 ) # 传入callbacks即可 result chain.invoke( {question: question}, config{callbacks: [langfuse_handler]} )环境变量也要先配好LANGFUSE_PUBLIC_KEYyour_public_key LANGFUSE_SECRET_KEYyour_secret_key LANGFUSE_HOSThttps://cloud.langfuse.com如果用的LangGraphhandler要传进graph的invoke配置里result graph.invoke( {question: question}, config{ callbacks: [langfuse_handler], metadata: {session_id: session_id} } )我在实际项目里遇到一个坑不要把同一个handler实例在不同线程间共享。LangGraph如果配置了并行节点每个并行分支的上下文是独立的共享handler可能造成trace串线。我后来在每次invoke入口处都新建handler成本极低数据却干净了。3.2 LangGraph场景的差异State流转时如何保持Trace连贯LangChain跑链和LangGraph跑图最明显区别是State流转——Graph节点之间共享一个全局State对象节点可以读写不同字段。所以Langfuse接入LangGraph时Trace的粒度应该是整张图执行而不是单个节点。这样才能在一条Trace里看到Agent从规划、调度、到工具调用、最终生成的完整决策路径。实测中发现默认的LangGraph instrumentation能覆盖节点级别的Span和节点内模型Generation。但如果你在节点里直接调用了自定义函数比如一个自己写的HTTP工具这个调用默认不会进Trace需要手动补Spanfrom langfuse import Langfuse langfuse Langfuse() # 在自定义工具函数内部 with langfuse.span( namecustom_http_tool, input{url: url} ) as span: result http_client.get(url) span.update(output{status_code: result.status_code})补完之后Trace时间线才能完整还原“模型决定调工具 → 工具实际执行 → 模型读取结果 → 生成答案”的全过程。缺少这一步中间工具执行这段在界面上就是断的排查问题又只能猜。3.3 手动埋点的兜底方案不依赖ORM框架时的控制方式回调机制方便但覆盖不了所有场景。Agent系统如果是纯手写、没走LangChain/LangGraph就需要用SDK手动埋点。我自己的习惯是写一个统一追踪工具类把埋点逻辑收敛起来from langfuse import Langfuse import uuid langfuse Langfuse() class AgentTracer: def __init__(self, user_id: str, session_id: str None): self.trace langfuse.trace( idstr(uuid.uuid4()), nameagent_trace, user_iduser_id, session_idsession_id, metadata{env: prod, version: 2025.06.01} ) def create_span(self, name: str, input_data: dict): return self.trace.span(namename, inputinput_data) def generate(self, model: str, prompt: str, completion: str, usage: dict): return self.trace.generation( namellm_call, modelmodel, inputprompt, outputcompletion, usageusage ) # 使用 tracer AgentTracer(user_iduser_123) with tracer.create_span(retrieve_docs, {query: 风险排序}) as span: docs retriever.search(风险排序) span.update(output{doc_count: len(docs)})手动埋点的优点是灵活缺点是容易漏。某个分支忘了埋Trace链就断了。我的建议是把Agent的两个关键决策点当固定埋点模板工具调用前后、模型调用前后无论如何都要埋哪怕没有数据也要用占位符标记。这样Timeline至少是连贯的排查时不会被断点误导。4. 流式输出与并发场景下的埋点细节实测千问、本地模型与批量任务4.1 流式请求的Token计数陷阱为什么结束后看到0 tokensAgent系统里流式输出基本是标配可流式场景在Langfuse里有个特别容易踩的坑如果你只把LLM调用包在Generation里不做流式回调请求结束后Langfuse里看到的token usage可能是0模型输出也可能是空的。原因很简单——流式响应是分块到达的模型调用那一步的completion在结束时还没有完整内容。要让Langfuse记全必须对流式事件做累加。用LangChain的时候CallbackHandler会自动处理on_llm_new_token把增量token拼起来所以这个坑主要坑的是手动埋点或不用回调的调用。手动埋点的解法是先创建不带usage的generation占位然后在流式过程中增量更新gen tracer.trace.generation( namestream_llm, modelqwen-plus, input{messages: messages} ) buffer [] usage {input: 0, output: 0, total: 0, unit: TOKENS} for chunk in stream_response: delta chunk.choices[0].delta.content buffer.append(delta) output_text .join(buffer) # 增量更新 gen.update(outputoutput_text) # 流结束后补全usage gen.update(usageusage, endtime.time())这个方案我在千问流式接口、OpenAI流式接口上都验证过。增量更新比只更新一次多一些请求开销但换来的能力是排查问题时能看到“回答是逐步生成到一半卡住还是一开始就没内容”。这个价值远大于开销。4.2 条件采样与成本开关并发扛不住时的优雅降级Agent系统跑起来之后最现实的问题是并发一高观测平台本身可能成为瓶颈观测数据多了成本也直接翻倍。我的处理方式是条件采样不用全量。import random def should_trace(user_id: str, trace_name: str) - bool: # 内部用户全量采样 if is_internal_user(user_id): return True # 付费用户全量采样 if is_paying_user(user_id): return True # 高价值流程按比例采样 if trace_name high_value_flow: return random.random() 0.3 # 默认低采样率 return random.random() 0.05采样逻辑统一在入口处执行。如果决定不采样就不创建trace业务照常跑。这样把观测成本控制到可接受范围。Langfuse自己也支持服务端采样率配置但我更习惯在代码侧控制因为可以针对不同用户、不同流程做差异化策略。再加一个“成本开关”的思路Agent服务本身高负载时动态降低采样率。我实现过一个自适应逻辑——最近一分钟P95延迟超过阈值时默认采样率从0.1降到0.02。效果很直接观测写入压力降了业务延迟也稳了。4.3 本地模型接入LM Studio/Ollama没有OpenAI兼容层时的办法很多团队搭Agent会同时用本地模型比如LM Studio、Ollama跑的Qwen、Llama系列。这时候接Langfuse有两条路。最省事的是走OpenAI兼容接口。LM Studio和Ollama都提供OpenAI兼容的API地址把代码里的base_url指过去模型调用在Langfuse看来就跟OpenAI调用没区别。但要注意兼容层返回的usage信息不一定完整。有些本地模型服务的token usage是估算的或者干脆不返回。这边处理的办法是在调用侧根据输入长度做一次估算并回填usage。另一条路是直接用Langfuse SDK把本地模型调用包装成generation完全不依赖模型吐的元数据。这个方法适合没有OpenAI兼容层的场景核心是把模型输入输出规范化拿不到准确token数就按字符数和词表做一个粗略估算。经验法则token估算误差在10%以内不影响Agent系统的趋势判断但绝不能用估算值去算成本账单——那是真金白银必须回归真实计数。顺带说一句会有人问“claude code调用LM Studio本地模型能不能追踪”。这个场景本质上是IDE工具内部调用模型不在你自己的Agent代码里Langfuse直接追踪不到。常见的做法是在网关层做统一代理在代理处上报调用记录然后手动关联到外部会话。5. 从追踪到评测把Langfuse当成Agent系统的“验收实验室”5.1 用Dataset固化回归用例避免改prompt后悄悄变差追踪解决的是“系统当时发生了什么”评测解决的是“这次改动到底好还是坏”。Langfuse有个重要能力把一批输入固化成一个Dataset然后对不同的prompt版本、模型版本跑同一批数据对比结果。我在Agent系统上的实践方式从生产环境Trace里挑出覆盖高频场景的案例包括工具调用成功、工具调用失败、多轮纠错、边界输入收集进Dataset。每个case记录输入、期望输出、期望的工具调用顺序。每次要改prompt或调整Agent策略时对同一个Dataset跑新版在Langfuse里对比当前答案是否更接近期望输出、有没有减少多余的工具调用。这个流程本质上是给Agent系统建了一个回归测试集。以前改prompt要人工逐个case试现在直接批量跑、批量对比。改prompt最怕“修好一个case弄坏三个case”有了Dataset和评测对比这个顾虑基本就放下了。5.2 自动评分与人工评分的协同实践Langfuse支持两种评分方式人工评分评审员或用户反馈和自动评分模型评估或规则判断。自动评分我建议用在跟业务强相关的地方。比如分类任务用规则判断分类结果是否正确摘要质量这类主观项目用评测模型打分。这里有个坑评测模型选不好打分不稳定反而把判断搞乱。经验是自动评分只用于“有明确对错”或“有明确打分标准”的环节主观性强的地方保留人工评分。人工评分可以做成评审队列。把生产环境采样采样出来的Trace按用户反馈分类——点赞、点踩、自动规则标记的异常组装成待评审列表。每周固定时间集中过一次给异常Trace打分、打问题标签。这些带标签的Trace反过来成为Dataset新用例形成正循环。5.3 一次基于评测结果的Agent策略调整实例分享一个实际案例。我们有个场景把长文本按主题拆成结构化片段。最初的prompt在边界处老是漏掉部分内容评测分数一直在70分上下。通过Langfuse的Trace定位原因漏内容主要发生在两个地方文本长度超过模型上下文窗口导致前段被截断工具返回多段文本时模型只读了最后一段。针对Trace结果做了两个调整一是在工具返回文本时把分段数量、总长度等元信息明确提到prompt最前面二是在正式生成前增加一步“确认”——让模型先输出“我已读取全部N段文本准备开始拆分”再做正式拆分。调整后评测分数从70涨到88原因也很明确不是靠感觉改好的。这个例子的价值在于没有全链路Trace就很难定位漏内容是发生在工具调用阶段、上下文截断阶段还是模型输出阶段。有了Trace加评测定位和验证形成闭环这就是从“模型调用”到“全链路可观测”最实际的价值。6. 半年工程化实践下来的几条实在经验6.1 数据脱敏别把用户隐私文本送进观测平台这可能是生产环境接入时最容易被忽视的事。Langfuse的Trace会记录prompt和completion如果业务含用户隐私直接上送等于把敏感数据复制给第三方。用Langfuse Cloud要更谨慎自托管也只是一定程度上减少传输风险不等于可以裸奔。我的方案分三层第一层字段级过滤。接入前对prompt和completion里的手机号、邮箱、身份证号、银行卡号做正则替换。第二层环境隔离。生产环境单独建一个观测项目敏感业务只记元信息不记prompt正文。第三层访问控制。给Langfuse项目配好token权限只给真正需要排查问题的人开读权限其他人一律不开放。6.2 成本控制采样率与保存周期双管齐下Langfuse不是免费的时间和存储都有成本尤其Trace里存了prompt和completion全文存储增长比想象中快得多。几个控制手段采样率。非核心链路可以降到5%甚至更低核心链路按实际需要调整。不需要为了“完整性”全量采集。数据保留期。定期清理超过30天的Trace只保留关键case对应Trace并打tag长期保存。没打tag的批量清掉绝不手软。只保留关键字段。如果不需要评估和回放埋点时不传系统提示词和长文档正文只留用户输入和模型输出。实测这些手段能让观测成本降一个数量级同时不损失排查能力。对多数团队来说“全量全字段”是浪费不是严谨。6.3 自托管与云服务怎么选最后说部署方式。Langfuse支持云服务和自托管官方Docker compose直接拉起一套。我的取舍逻辑比较务实团队小、刚起步直接云服务。省去运维精力专注埋点和数据消费。有数据管控要求、希望数据完全在内网自托管也不复杂。测试环境一台2核4G机器就够生产环境建议挂外部Postgres和对象存储。如果选了云服务注意数据所在地合规要求别踩红线。我选择的是自托管加按需采样。自托管部署在内网访问快、数据不出内网心里踏实。代价是升级维护自己扛Langfuse版本更新节奏不慢建议固定版本部署别无脑跟latest。最后再分享一点个人体会。Langfuse在我这里最大的价值不是“看日志”而是让团队从“猜Agent为什么这么干”进化到“能完整回溯Agent为什么这么干”。Agent系统的黑盒是模型不是工程链路——链路部分完全可以用工具做到透明。如果你也在搭Agent强烈建议从一开始就把全链路追踪埋上不要等出了问题再补补的代价永远比埋的代价大。另外一个小心得埋点时给Trace打业务标签比如用户地域、Agent版本、场景类型。当时觉得是顺手现在做成本归因和效果分析全靠这些标签越早打越省事。