企业官网建设流程全解析

最近在做一个基于 FastAPI + LangChain + LangGraph 的客服类 AI Agent，模型调用本身不贵，真正让我头疼的是“出了事没法查”。生产环境里用户问了一句稍微绕弯的话，Agent 就开始一本正经地胡说八道，工具调用链走到一半直接断掉，token 账单哗哗涨，但我连是哪一步烧的钱都说不清楚。日志明明都打了，可 Agent 的中间过程就像个黑盒：它到底理解了用户什么意图、调了哪个工具、上下文拼接成什么样、模型为什么走这条路，传统日志根本还原不出来。

后来我把 Langfuse 整套接进了项目，才真正体会到什么叫“从模型调用到全链路可观测”。这篇文章把我这段时间的工程实践完整复盘一遍，包含核心概念拆解、接入 LangChain/LangGraph 的最小操作路径、流式和并发场景的埋点处理、线上测评和成本追踪做法，以及我在自托管过程里踩过的一堆坑。想给 Agent 项目上可观测性的朋友，可以直接照着抄。

最近在做一个基于 FastAPI + LangChain + LangGraph 的客服类 AI Agent，模型调用本身不贵，真正让我头疼的是“出了事没法查”。生产环境里用户问了一句稍微绕弯的话，Agent 就开始一本正经地胡说八道，工具调用链走到一半直接断掉，token 账单哗哗涨，但我连是哪一步烧的钱都说不清楚。日志明明都打了，可 Agent 的中间过程就像个黑盒：它到底理解了用户什么意图、调了哪个工具、上下文拼接成什么样、模型为什么走这条路，传统日志根本还原不出来。

后来我把 Langfuse 整套接进了项目，才真正体会到什么叫“从模型调用到全链路可观测”。这篇文章把我这段时间的工程实践完整复盘一遍，包含核心概念拆解、接入 LangChain/LangGraph 的最小操作路径、流式和并发场景的埋点处理、线上测评和成本追踪做法，以及我在自托管过程里踩过的一堆坑。想给 Agent 项目上可观测性的朋友，可以直接照着抄。

1. 为什么 Agent 项目必须先解决可观测性

1.1 传统日志方案为什么不够

先说说我一开始的做法。最朴素的埋点方案就是打日志：请求进来打一条，每次 LLM 调用打一条，工具返回打一条，最后拼一个 summary。这套东西在普通 CRUD 接口上完全够用，但在 Agent 场景里会遇到几个非常尴尬的问题。

第一个问题是找不到“过程”。Agent 的执行是动态分支的，同一个问题今天走 A 工具，明天可能走 B 工具。普通日志只能记录发生了什么，却记不住“为什么发生”。比如某一次回答出错，我需要知道是意图识别错了、工具入参格式错了、还是模型最终生成的时候被上下文带偏了。这三个环节在日志里是三个孤立记录，没有一条共同的轨迹把它们串起来。

第二个问题是上下文不可见。Agent 的核心资产是它的上下文窗口，我打过很多日志去看 messages 数组，但每轮对话里系统提示词、历史消息、工具返回结果是怎么拼接的，日志里完全看不出来。可真正导致模型胡说的，往往就是这个拼接过程中的细微问题。

第三个问题是成本归因难。多轮对话里一个大模型调用可能携带了超长的历史上下文，账单上只显示“某天消耗了多少 token”，却说不清是哪个用户、哪次会话、哪一步调用烧掉的。用户投诉体验差，我连“这次对话为什么这么贵”都答不上来。

说到底，Agent 是一个多步骤、有状态、带概率性的系统。概率性意味着同样的输入可能有完全不同的执行路径，没有 trace 级别的记录，任何“复盘”都是盲人摸象。

1.2 链路追踪的三个层次

现在很多团队会提“可观测性”，但可观测性不是简单地把日志接进 ELK 就算完。在我理解里，一个合格的 Agent 可观测体系至少要覆盖三个层次：

第一层是指标层，回答 TPS、平均延迟、token 消耗量、错误率，这些数据用来回答“系统整体健康吗”。第二层是日志层，每一条调用的入参出参、模型 response、工具报错，这些用来回答“这一步执行得对不对”。第三层才是链路层，一次用户请求对应的完整执行轨迹：从意图理解到工具选择、工具调用、结果回填、最终生成，一条 trace 把所有环节按执行顺序和时间线串起来。

关键点在于这三层必须互相打通。指标异常可以下钻到具体某条 trace，trace 里的某个 span 又能展开成完整的日志明细。Langfuse 的设计恰好就是围绕这个逻辑来的，我后面会细讲它的数据模型。没打通之前，三层数据各自为政，出了线上故障还是靠人肉翻日志，效率低得让人绝望。

2. Langfuse 核心能力拆解

2.1 Trace、Span、Generation 的最小数据结构

Langfuse 的所有功能都建立在一套非常清晰的数据结构上。我第一次用的时候就觉得这个模型设计得聪明，它没有发明太多复杂概念，核心就四样东西：Trace、Span、Generation、Score。

Trace 对应一次完整的 Agent 任务，比如用户发起一次对话、执行一次数据分析任务。每个 Trace 有唯一 ID，我习惯把业务侧的 request_id 直接映射到 Trace ID 上，这样从用户投诉到链路定位只需要一步。

Span 是 Trace 内部的子单元，代表一次具名执行的阶段，比如“调用搜索工具”“读取数据库”“构建上下文”。Span 可以无限嵌套，形成一棵执行树。我们在 LangGraph 里每个节点就可以设计成一个 Span。

Generation 是 Span 内部更细的一层，专门用来记录模型调用。它保存模型名称、Prompt、Completion、token 用量、延迟、采样参数这些关键信息。为什么要单独拆出 Generation？因为普通 span 只记录“做了什么”，generation 还要回答“给模型喂了什么、模型吐了什么、花了多少钱”。

这套结构还原一个 Agent 任务非常够用。模型调用是 Agent 的最核心动作，也是成本黑洞，单列一层就保证了后续能做精细的成本统计和评测分析。

2.2 不是唯一选择：Langfuse 与其他可观测方案对比

我在选型时对比过几条路线。自建方案最原始，用数据库记录 trace，再拿 Grafana 画 dashboard，问题是链路关联、token 统计、数据集管理这些都需要从零开发，工程量远超想象。接 OpenTelemetry 也是个方向，语义标准好、通用性强，但 OTel 对 LLM 特有的 prompt、token 用量、模型名这些字段没有现成的语义约定，Agent 场景下用起来仍然别扭。LangSmith 做得很成熟，但它主打闭源托管，很多企业对数据出口有要求，这一条就劝退了。

Langfuse 最吸引我的点是开源、可自托管、数据完全在自己手里。它的 Cloud 版很省心，但国内生产环境我还是选了自己部署，数据不出内网，心理踏实。另一个加分项是它对 LangChain、LlamaIndex、OpenAI SDK、LangGraph 都有现成的回调集成，接入成本很低，不像自建方案那样什么都要自己造轮子。

3. 实操：把 Langfuse 接入 FastAPI + LangChain + LangGraph 的 Agent

3.1 五分钟先跑通一个最小 Trace

先别管复杂的 Agent，第一步先把 SDK 跑通。我假设你已经有一个 Langfuse 实例（或者 Cloud 项目），拿到了三个关键值：公钥、私钥、项目地址。自托管的话，公钥私钥在项目设置里创建；Cloud 版在账号设置里。

安装依赖，注意 Langfuse 的 SDK 和 LangChain 集成是分离的：

pip install langfuse langchain langchain-openai

初始化方式我踩过一个小坑：不要在每个函数里反复 new client，最好在模块加载时初始化一次：

from langfuse import Langfuse langfuse = Langfuse( public_key="pk-...", secret_key="sk-...", host="https://你的langfuse域名", # 自托管时建议加 timeout 和 debug，排查网络问题很有用 timeout=10000, debug=False )

跑通一个最小 trace 可以直接用 SDK 的方式，不接任何框架：

from langfuse import Langfuse trace = langfuse.trace(name="hello-trace", input={"msg": "你好"}) generation = trace.generation( name="greeting", model="gpt-4o-mini", input=[{"role": "user", "content": "你好"}] ) # 模拟一次模型调用 completion = "你好，我是客服助手" generation.end(output=completion, usage={"input": 12, "output": 9}) trace.update(output=completion)

运行完这段代码，去 Langfuse 界面的 Traces 页面刷新，就能看到一条完整的 trace 记录。这一步验证了 SDK 环境和网络都正常，后面接 LangChain 就有了基础。

3.2 接入 FastAPI 接口层，让每次请求都有完整 Trace

客服 Agent 肯定是一个 HTTP 服务。我用 FastAPI 做接口层，LangGraph 做 Agent 编排。Langfuse 对 LangChain 生态有官方回调处理器，直接把 callback 传给链或图就能自动埋点。

有一个细节值得提前说：LangGraph 的 invoke 可以传 config，而 Langfuse 回调会从 config 里自动取 trace 的上下文。最省事的做法是在请求入口新建 trace，然后通过 config 传入回调，让所有子节点自动挂到同一个 trace 下面。

from fastapi import FastAPI, Request from langfuse.callback import CallbackHandler from langfuse import Langfuse app = FastAPI() langfuse_client = Langfuse() def get_handler(request_id: str): return CallbackHandler( trace_id=request_id, public_key="pk-...", secret_key="sk-...", host="https://你的langfuse域名" ) @app.post("/api/chat") async def chat(request: Request): body = await request.json() user_msg = body["message"] session_id = body.get("session_id", "default") request_id = f"req_{uuid4().hex}" handler = get_handler(request_id) config = { "callbacks": [handler], "configurable": {"session_id": session_id}, "metadata": { "user_id": body.get("user_id"), "request_path": "/api/chat" } } result = await graph.ainvoke( {"messages": [{"role": "user", "content": user_msg}]}, config=config ) return {"reply": result["messages"][-1].content, "trace_id": request_id}

trace_id用业务侧 request_id，好处是以后如果有业务系统对接，日志和 trace 能直接对得上号。configurable里的session_id会映射到 Langfuse 的 Session，后续可以把一个用户的多次请求归到同一个会话里看全貌。

这里插一句：务必在返回给前端的数据里带上 trace_id。这样用户反馈问题的时候，直接在那条消息旁边就能找到 trace_id，进系统里定位就是几秒钟的事，省去两边来回沟通成本。

3.3 流式调用与并发场景怎么处理

只用ainvoke做一次性返回的场景比较简单，但真实客服机器人几乎都要做流式输出。LangGraph 支持astream，Langfuse 的回调处理器也实现了on_llm_new_token，理论上 token 用量会自动统计。不过我在实践里发现两个容易踩的坑。

第一个坑是异步环境下必须使用同一个回调实例贯穿整个流式过程。如果把 CallbackHandler 创建在某个子节点内部，每次节点执行都会新建一个回调，trace 的关联就会散掉。正确做法是在请求入口创建一次，通过 config 传给整个图。

第二个坑是流式响应结束之后要记得调用一次handler.flush()。Langfuse 的 SDK 是异步批量上报，不 flush 的话，请求结束了数据可能还滞留在内存里，尤其是快速连续多次流式调用，看到仪表盘数字迟迟不涨，多半就是这个原因。

关于并发，我的实践经验是：Langfuse SDK 本身是线程安全的，可以直接在多线程/多协程环境里共享同一个 client，不需要为每个请求创建新 client。真正要注意的是上报缓冲区和网络 I/O 对业务请求的干扰。我上线初期为了追求零误差，所有请求全量上报，结果 Langfuse 服务端的写入压力直接把业务进程的延迟拉高了。后来改成采样上报加错误全量上报，才把延迟恢复到正常水平。

流式场景还有一个实践心得：为了准确统计首字延迟和总时长，可以在 LangGraph 的节点里手动包一层 span，把流式迭代的逻辑包进去，这样 Langfuse 时间线上能看到每个节点耗时，而不仅限于 LLM 本身的耗时。

4. 全链路可观测之后：线上测评、成本分析与并发优化

4.1 线上评测怎么做

可观测性不只是拿来看链路、追问题，它对评测也很重要。AI Agent 上线后最大的难题是“这次改动到底是变好了还是变坏了”。Langfuse 的 Score 功能可以很好地承接这个需求。

Score 本质上就是给一条 trace 或一个 generation 打分数或打标签。最简单的用法是客服系统里让运营对回答质量打分，分数写回 Langfuse，后续能按分数过滤 trace，也能做回归分析。Post 一个 score 的代码非常轻量：

from langfuse import Langfuse langfuse = Langfuse() langfuse.score( trace_id="req_xxx", name="answer_quality", value=4.5, comment="回答完整但不够具体", user_id="运营账号123" )

更进阶的用法是搭自动化评测。把模型的回答和人工标注的标准答案放入数据集，离线跑一批实验，Langfuse 会在 Experiments 页面给出对比视图，能直接看到不同 prompt、不同模型版本在同一个数据集上的效果差异。我刚接入时是人工一条条看 trace，效率很低；后来整理了 100 条高频客服问题做成 dataset，每次改 prompt 就批量跑一次，效率完全不一样。

4.2 成本与延迟追踪

Langfuse 的成本洞察能力是基于 usage 字段的。OpenAI 系列的 SDK 会自动返回 prompt_tokens、completion_tokens，Langfuse 能根据模型单价自动算成本。如果你用的是自部署模型或者第三方模型，SDK 不一定会返回 token 用量，这时候要手动在 generation 结束的地方补充 usage。

实践中我发现，Langfuse 自带的成本统计是“按模型名 × token 数”来算的，所以模型名一定要写规范。我之前图省事，在两个节点里分别写成了 “gpt-4o-mini” 和 “gpt-4o-mini-2024-07-18”，结果成本明细里同一种模型被拆成两行，对账的时候非常痛苦。建议全项目统一一个模型名管理函数，避免手写漂移。

延迟追踪也是同样的道理。Langfuse 的 trace 详情页自带时间线瀑布图，能看到每个 span 的耗时。这对我定位“用户觉得机器人卡”非常有用：有一次用户反馈回复太慢，我打开 trace 一看，70% 的时间花在一个网页检索工具上，模型生成只占很小比例。顺着这个线索去优化检索超时和缓存策略，问题很快解决。

4.3 高并发下的采样策略

“AI Agent 怎么扛并发”是最近社区里讨论很多的话题。Langfuse 本身不解决并发问题，但它的采样策略直接决定了你能不能在高并发下持续观测而不拖垮业务。

一开始我全量上报，Langfuse 自托管实例的 CPU 和磁盘 I/O 明显升高，业务接口的 P99 延迟上涨了大约 30 毫秒。后来我改成三层采样策略：

第一层是无条件全量上报所有错误和慢请求。第二层是核心用户或付费用户全量上报。第三层是普通流量按 10% 采样。Langfuse 的 sample_rate 参数天然支持这个能力，但更灵活的做法是在创建回调的时候根据业务上下文手动决定是否上报。

这里再给一个我实践出来的重要技巧：采样率不要写死在代码里，做成环境变量。因为线上出问题的时候，你大概率想临时把采样率拉高到 100%，如果写死在代码里就要重新发版，非常蠢。我现在是把采样逻辑收敛在一个工厂函数里，按环境变量读取，线上临时调整只需要改配置、重启进程。

import os SAMPLE_RATE = float(os.getenv("LANGFUSE_SAMPLE_RATE", "0.1")) def should_trace(user_id: str, has_error: bool) -> bool: if has_error: return True if user_id in VIP_USER_SET: return True return random.random() < SAMPLE_RATE

5. 常见问题与排查技巧实录

5.1 数据没上报的六个原因

接入 Langfuse 最常见的问题就是“接口调了、界面啥也没有”。我梳理了六大高频原因，按出现频率排序：

第一，公钥私钥填错或者填反了，看日志里有没有 401 错误。第二，host 拼写多了末尾斜杠或少了路径前缀，导致网络请求 404。第三，项目里建了多个环境，SDK 数据被发到了另一个环境。第四，回调没有通过 config 传给 LangChain/LangGraph，链内部的调用没有挂到 trace 上。第五，进程提前退出，异步批量上报没来得及发送。第六，采样率设成了 0，数据被主动丢弃。

排查顺序我建议是：先看业务日志有没有报错，再抓包确认请求是否发出，然后确认 SDK 初始化的三个参数，最后检查 trace 的 session 条件和采样配置。别上来就怀疑 SDK 有问题，绝大多数时候是配置问题。

5.2 回调重复触发导致数据翻倍

我遇到过 trace 数据翻倍的情况，一个 LLM 调用在界面上出现两次。最后定位发现是回调被同时传到了 LangGraph 图的 invocate 参数和节点的 invoke 参数里，模型调用被两层配置各挂了一次回调，自然记录了两遍。

解决办法是统一回调的传递路径：要么只在图的顶层传 callbacks，要么只在节点内部传，不要两个地方同时传。Langfuse 的回调处理器会做一定的去重，但重复挂载时依然可能产生双份数据。这个坑排查起来特别费劲，我一度以为是自己代码里重复调用了模型。

5.3 数据量大导致自托管实例卡顿

自托管 Langfuse 在数据量上来之后，Traces 列表页会明显变慢。我踩过最严重的坑是磁盘空间被写满。Langfuse 默认会保存原始 input/output 数据，Agent 的 prompt 动不动就是几千 token，会话一多，Postgres 里的 JSON 字段体积增长非常快。

解决思路有三个：一是给 input/output 做脱敏，敏感信息不要放进 trace；二是开启 Langfuse 的存储裁剪策略，只保留摘要不保留全量内容；三是定期清理无价值的 trace 数据，比如测试流量。生产实践里我更推荐第一种脱敏加必要的裁剪组合，既保证了可观测性，又不至于把磁盘撑爆。

我还建议在自托管时把 Langfuse 的 Postgres 和 ClickHouse 数据目录挂到独立磁盘，定时监控磁盘水位。ClickHouse 是用来做分析查询的，高频写入对磁盘 IO 要求挺高，别和业务库混用。

5.4 调试技巧：local 模式与 UI 排错

最后给一个调试小技巧：开发环境里遇到 trace 不显示，可以临时打开 debug 模式：

trace = langfuse.trace(name="debug-trace", debug=True)

开了 debug 之后，SDK 会输出非常详细的上报日志，包括请求内容、状态码、错误信息。我很多次“死活找不到问题”的排查，都是靠这个日志定位到是网络代理还是鉴权失败。还有一个非常有用的能力是 UI 上的 “Single Trace View”，直接打开一条 trace 的 JSON 视图，能检查每一个 event 的原始字段是否符合预期。

结尾：几个值得记住的实践体会

最后分享一点个人体会。可观测系统的建设不要太贪心，一开始接入 Langfuse 时我把所有能埋的地方全部埋了一遍，结果数据量大到根本看不过来，操作界面也变得很卡。后来想明白一个道理：可观测性的目的是在需要的时候能定位问题，而不是把系统里发生的每一件事都永久存档。控制采样率、控制保留时长、控制字段冗余度，看起来是“减法”，实际是在为真正的排查效率做“加法”。

还有一点关于成本：接入 Langfuse 之后不要只顾着看 trace 图，要多看它的成本聚合视图。有一次我在 dashboard 上发现某个模型连续几天 token 消耗异常，追下去发现是工具调用循环把同一段长文档反复拼接进上下文，这个问题单靠代码 review 很难发现，但全链路观测给了一个非常具体的提示。

如果你正准备给自己负责的 AI Agent 项目上可观测性，我的建议是：先跑通最小链路，再扩场景，不要一上来就铺很大覆盖面。项目边界清晰、trace 结构设计合理，比埋点多但数据混乱要好得多。可观测这件事，做得多不如做得准。