前几篇我们讲清了 Checkpoint（存档）和 Interrupts（刹车）。智能体跑起来之后，还有一个刚需：把「它到底怎么想的、调了哪些工具、花了多少 token」完整录下来，方便开发排障。这一篇以 Langfuse 为例，从概念、架构到一次 HTTP 请求触发的全链路采集工作流，尽量用流程图讲透——并结合 mystu 项目里刚落地的 Callback 集成，说明数据是在哪一层、以什么形态被采集和上报的。

---

1. 为什么需要 Langfuse（以及它和日志的区别）

应用日志能告诉你「某次请求开始了 / 失败了」，但很难回答下面这些问题：

问题	普通 logging	Langfuse
模型一共调了几次？每次 input/output 是什么？	需手工拼接	自动按 Generation 记录
工具调用顺序、参数、返回值？	往往只有一行 info	按 Span 树状展示
一次对话跨多轮 HTTP（含 HITL resume）怎么串起来？	靠 thread_id 自己 grep	Session 聚合多条 Trace
token 用量、延迟、成本？	需自己算	UI 直接统计
流式输出中途 token 怎么落盘？	几乎无法还原	Callback + flush 完整 span

Langfuse 的定位是 LLM 应用的可观测性平台（LLM Observability）：专门采集、存储、展示 Agent / RAG / Chat 链路的 Trace（链路追踪） 数据，而不是替代传统 APM 或业务日志。

一句话：Checkpoint 负责「状态能不能续上」，Langfuse 负责「过程能不能看清楚」。

---

2. 核心概念：Trace、Observation、Generation、Session

在 Langfuse 里，数据不是扁平日志，而是一棵有层级的观测树。先把名词对齐，后面看数据流才不会乱。

2.1 Trace（链路）

• 含义：一次「可观测的执行单元」，通常对应应用里的一次用户操作或一次 Agent 运行。
• 典型映射：mystu 里一次 POST /agent/api/chat、一次 /chat/stream、一次 /chat/resume 各产生 一条 Trace（若启用了 tracing 且命中采样）。
• 包含：开始/结束时间、metadata（user_id、thread_id、api 类型等）、其下所有 Observation。

2.2 Observation（观测节点 / Span）

Langfuse v3+ 底层与 OpenTelemetry 对齐，对外仍常用「Observation」一词。Observation 是 Trace 里的节点，分多种 type：

type	含义	典型场景
GENERATION	一次 LLM 调用	ChatDeepSeek 的 prompt + completion、token 数
SPAN	一段逻辑 / 工具 / 子流程	@tool 执行、检索、代码节点
EVENT	瞬时事件	较少在 Callback 路径手动使用

Observation 之间可以 parent / child 嵌套，形成调用树——这正是「链路追踪」名字的由来。

2.3 Generation（生成）

GENERATION 是 Observation 的一种，专指 大模型推理。Langfuse 会尽量记录：

• model 名称
• input（prompt / messages）
• output（completion）
• usage（prompt tokens、completion tokens、total）
• latency（首 token、总耗时等，视集成方式而定）

2.4 Session 与 User

概念	作用	mystu 映射
Session	把多条 Trace 归到同一次「对话会话」	langfuse_session_id = thread_id
User	按登录用户筛选、统计	langfuse_user_id = str(user_id)
Tags	轻量分类标签	chat / stream / resume

这样，HITL 场景下「第一次 chat 触发 interrupt」和「第二次 resume 批准执行」会在 Langfuse UI 的 同一 Session 下看到两条 Trace，便于还原完整故事线。

---

3. Langfuse 平台架构：数据最终存到哪

自托管 Langfuse 时，应用进程不会直接写数据库，而是走 Ingest API。整体架构可以概括为：

3.1 各层职责

组件	职责
Python SDK	在应用内挂载 Callback / @observe，把 LangChain/LangGraph 事件转成 OTEL/Langfuse 事件，内存队列 + 后台线程/异步批量发送
Ingest API	鉴权（public/secret key）、接收 batch、校验 schema
Worker	解析 batch、关联 trace/observation、写入存储
PostgreSQL	项目、用户、API Key、trace 元信息、评分等
ClickHouse	大规模 observation 事件、时序分析、聚合查询
Web UI	Trace 详情、Session 视图、成本仪表盘、人工标注等

对应用开发者来说：只需关心 SDK 怎么初始化、Callback 怎么挂、什么时候 flush；存储细节由 Langfuse 运维负责。

---

4. 集成方式概览：LangChain Callback 在干什么

mystu 采用的是 方案 A：LangGraph run config 注入 CallbackHandler（见 docs/brainstorms/2026-06-24-langfuse-tracing-requirements.md）。

LangChain / LangGraph 在执行链路上预留了 Callback 钩子。langfuse.langchain.CallbackHandler 实现了这些钩子，在以下时机被调用：

要点：

1. 业务代码几乎不改——tracing 是旁路；Agent 的 prompt、工具、HITL 逻辑与改造前一致。
2. 每个 HTTP 请求应使用新的 CallbackHandler() 实例——避免并发请求共用 handler 导致 trace 串线。
3. Singleton 的是 Langfuse() client——负责连接配置与发送队列；在 mystu 的 FastAPI lifespan 里初始化一次即可。

---

5. mystu 中的采集挂载点（从进程启动到一次 chat）

下面把 mystu 项目里的真实代码路径嵌进 Langfuse 工作流，便于对照仓库阅读。

5.1 进程启动：Client 初始化

对应代码：

• mystu/controller/__init__.py：lifespan 里调用 init_langfuse_client() / shutdown_langfuse_client()
• mystu/buildagent/agent/langfuse_tracing.py：读取 modelConfig.load_langfuse_config()，构造 Langfuse(public_key, secret_key, host=...)

降级原则：初始化失败只打 warning，不阻断 Web 启动——与 MCP 加载失败的处理哲学一致。

5.2 一次 /chat 请求的完整采集工作流

这是本文的核心全链路，按时间顺序展开：

逐步说明：

步骤	发生了什么	采集侧效果
1	API 鉴权通过，进入 achat	尚无 trace
2	_build_invoke_messages 组装 messages（含记忆注入的 SystemMessage）	这些 messages 会作为后续 LLM Generation 的 input 被 Callback 捕获
3	_build_run_config 检查 is_tracing_active() + should_sample()	未启用则完全不创建 handler，业务零侵入
4	创建 CallbackHandler()，写入 metadata	Trace 带上 user_id、thread_id、langfuse_session_id、langfuse_tags: [chat]
5	get_agent().ainvoke(..., config=config)	LangGraph 开始跑图；Callback 创建 Trace 根节点
6	模型节点执行	GENERATION：DeepSeek 的 messages in / AI message out / tokens
7	工具节点执行（如 place_order、记忆工具、MCP 天气工具）	SPAN：tool name、args、return value
8	若触发 HITL interrupt	图在工具前暂停；Trace 里通常已能看到 tool call 意图（pending 的调用结构）
9	safe_flush()	强制把 SDK 内存队列刷到 Ingest（非流式路径在 ainvoke 返回后立即执行）
10	Ingest → Worker → DB	UI 可查询

metadata 注入逻辑（节选）：

# mystu/buildagent/agent/langfuse_tracing.py（概念示意）
metadata = {
    "langfuse_user_id": str(user_id),
    "langfuse_session_id": thread_id,
    "langfuse_tags": [api],  # chat | stream | resume
    "thread_id": thread_id,
    "user_id": user_id,
    "api": api,
    "llm_model": llm_model,
}
config["callbacks"].append(CallbackHandler())

5.3 流式 /chat/stream：多出来的 flush 环节

流式场景下，LLM token 分多次到达 Callback（on_llm_new_token 等），Trace 在内存里逐步变长，若进程过早结束或队列未刷盘，会出现「UI 里 trace 为空或不完整」的问题。

mystu 的处理：

对应 deepagent.astream_chat 的 try / finally：finally 里调用 safe_flush()，保证 SSE [DONE] 前后数据落盘。

5.4 HITL /chat/resume：Session 串联第二条 Trace

中断恢复是「两次 HTTP，一次对话故事」的典型场景：

Langfuse UI 按 Session = thread_id 筛选，即可看到：

• Trace A：langfuse_tags: [chat]，停在审批前
• Trace B：langfuse_tags: [resume]，metadata.decision: approve，含工具执行与后续模型回复

这与 Checkpoint 用同一 thread_id 续跑状态是平行关系：Checkpoint 管状态，Langfuse 管观测。

---

6. SDK 内部：从 Callback 事件到 HTTP Batch

理解「为什么可以异步、为什么要 flush」，需要往下看 SDK 一层（与具体 mystu 代码无关，属于 Langfuse 通用机制）。

机制	说明
异步上报	默认不在 LLM 热路径上同步等 HTTP；事件入队，后台批量发送，降低对首 token 延迟的影响
flush()	告诉 SDK「这条 trace 的业务侧已结束，请尽快把队列发出去」；流式、短进程、测试里尤其重要
shutdown()	进程退出前 flush 剩余队列并关闭 sender；mystu 在 lifespan finally 调用
失败行为	上报失败一般只记 SDK 内部错误/日志，不应抛到业务；mystu 对 flush/init 包了 try/except + warning

Langfuse Python SDK v3/v4 基于 OpenTelemetry 语义构建 observation 树；LangChain Callback 是一层「把 LangChain 事件翻译成 OTEL span 的适配器」。因此同一套后端也能接 OTEL exporter，但 mystu v1 刻意只用 Callback，不做 FastAPI HTTP 根 trace，避免双轨复杂度。

---

7. 启用、采样与降级：控制「采不采、报不报」

mystu 在应用层实现了两道闸门（见 modelConfig.py + langfuse_tracing.py）：

环境变量语义（详见 AGENTS.md）：

变量	作用
LANGFUSE_ENABLED	应用总开关，默认 false
LANGFUSE_PUBLIC_KEY / LANGFUSE_SECRET_KEY	项目密钥，启用时必填
LANGFUSE_HOST	自托管 ingest 地址
LANGFUSE_SAMPLE_RATE	0.0–1.0，应用层采样；0 = 启用逻辑但永不挂 handler

注意：应用层采样与 Langfuse SDK 内置采样是两层概念；mystu v1 以应用层为准，行为更可预测。

---

8. 在 Langfuse UI 里你会看到什么

一次典型的 mystu chat（模型回答 + 调用了工具）在 UI 中大致呈现为：

Session: <thread_id>
└── Trace: <auto trace id>   tags: [chat]   user: <user_id>
    ├── Generation: ChatDeepSeek
    │     input:  [SystemMessage 记忆注入, HumanMessage 用户问题, ...]
    │     output: AIMessage（含 tool_calls 或纯文本）
    │     usage:  prompt / completion tokens
    ├── Span: place_order（或 memory / MCP 工具名）
    │     input:  { contract, side, quantity }
    │     output: "订单已提交..." 或 interrupt 前状态
    └── Generation: （若有第二轮模型总结）

开发排障时的常用路径：

1. 用 Session ID = thread_id 找到用户整段对话
2. 看 Generation 核对模型到底读了哪些记忆、回了什么
3. 看 Tool Span 核对参数是否与用户意图一致
4. 对比 chat 与 resume 两条 Trace，还原 HITL 审批前后差异
5. 看 latency / tokens 定位慢请求与成本异常

---

9. 与 Checkpoint、Interrupt 的关系（别混了）

维度	LangGraph Checkpoint	Langfuse Trace
目的	状态持久化，支持多轮与 resume	可观测性，支持排障与分析
存储	mystu：data/checkpoints.sqlite	Langfuse：PG + ClickHouse
键	thread_id	Session 也用 thread_id（mystu 映射）
失败影响	丢失则无法续聊	丢失则看不见 trace，业务仍可运行
HITL	必须，否则无法 resume	可选，但强烈建议开启以便审工具调用

两者独立：关掉 Langfuse 不影响对话记忆；删掉 checkpoint 不影响已上报的 trace（只是无法再 resume 到同一状态）。

---

10. 自托管部署时数据流还要经过哪一跳

若 Langfuse 与 mystu 不在同一台机器，网络路径如下：

运维需保证：

• mystu 进程 → Langfuse Ingest 网络可达
• 密钥在 Langfuse 项目设置中创建，写入 mystu .env
• Langfuse 侧配置 retention、RBAC——mystu 当前为全量记录消息正文，合规由 Langfuse 访问控制承担

---

11. 小结：全链路一张图

记住四句话：

1. Trace 管一次运行，Session 管一段对话，Generation/Span 管树里每个节点。
2. Callback 挂在 LangGraph config 上，业务代码旁路采集。
3. SDK 异步批量上报，流式与进程结束前要 flush。
4. tracing 失败必须降级，不能拖垮 Agent 主路径。

---

12. 用 Langfuse 官方 Skill 赋能 mystu

mystu 的 Langfuse 集成不是「凭记忆写 SDK」，而是遵循 Langfuse 官方 Cursor Skill（仓库内路径：.cursor/skills/langfuse/）的 Documentation First 原则：先查当前版文档与 reference，再改代码。Skill 让 AI 助手在接入、审计、排障 tracing 时有一致工作流，避免 SDK 版本漂移（本项目使用 langfuse Python SDK v4 + OTEL）。

12.1 Skill 是什么、放在哪

项	说明
来源	github.com/langfuse/skills 的 skills/langfuse
本仓库路径	.cursor/skills/langfuse/SKILL.md + references/
何时触发	在 Cursor 中讨论 Langfuse 集成、查文档、用 CLI 读 trace、审计现有 instrumentation 时，Agent 应自动读取该 Skill
更新方式	见 AGENTS.md「Langfuse Agent Skill」：git clone 到 .vendor/langfuse-skills 后覆盖复制（.vendor/ 已 gitignore）

Skill 核心原则（摘自 SKILL.md）：

1. Documentation First — 不凭记忆写集成代码，先拉取 langfuse.com 当前文档
2. CLI for Data Access — 用 langfuse-cli 查 trace / session / score
3. Best Practices by Use Case — 按场景读 references/ 下对应文件
4. Use Latest SDK — 默认跟随最新 Langfuse SDK（mystu 约束 langfuse>=4.9.0）

12.2 用 Skill 实现 mystu tracing 的推荐工作流

下面是从零到 mystu 当前集成的标准路径（本项目实际走过）：

在 Cursor 里可直接用的提示示例：

Install the Langfuse AI skill from github.com/langfuse/skills and use it to add tracing to this application with Langfuse following best practices.

或针对 mystu 已有集成的审计：

使用 .cursor/skills/langfuse，对照 references/instrumentation.md 审计 mystu/buildagent/agent/langfuse_tracing.py 是否符合 baseline。

12.3 Skill references 与 mystu 代码的对应关系

Agent 读 Skill 时应优先打开 references/instrumentation.md（接入与审计）。其余 reference 按场景选用：

Skill reference	用途	mystu 中的落点 / 下一步
instrumentation.md	接入 baseline、trace 名、session、metadata	langfuse_tracing.py：mystu-agent-{chat|stream|resume}、langfuse_session_id=thread_id
sdk-upgrade.md	SDK 大版本升级	升级 pyproject.toml 中 langfuse 后重跑 tests/test_langfuse_tracing.py
error-analysis.md	从 trace 建失败分类、决定改什么	配合 docs/langfuse-trace-practical-guide.md
user-feedback.md	对 trace 打 score（ thumbs / 评分）	v1 未接入；未来可在 API 层写 score
cli.md	CLI 查数据	见 §12.4
prompt-migration.md	prompt 迁到 Langfuse 托管	当前 prompt 仍在 iron_ore_forecast.md

instrumentation baseline 在 mystu 的满足情况：

Baseline 要求	mystu 实现
可检索 trace 名	langfuse_trace_name → mystu-agent-chat / stream / resume
Model / token	LangChain Callback 自动采集 ChatDeepSeek Generation
Session 聚合	langfuse_session_id = thread_id
User 归因	langfuse_user_id = str(user_id)
流式完整 span	astream_chat 的 finally: safe_flush()
失败不阻断业务	auth_check 失败 → is_tracing_export_ready() 为 false，不挂 Callback

12.4 Skill 推荐的文档检索方式（Agent 内部）

Skill 规定查 Langfuse 文档的优先级（实现 mystu 集成时 Agent 应遵循）：

1. https://langfuse.com/llms.txt — 全站文档索引
2. 单页 Markdown — 如 https://langfuse.com/integrations/frameworks/langchain.md
3. 搜索 API — https://langfuse.com/api/search-docs?query=...（含 GitHub Issues 索引，排障 401 时有用）

不要在未读文档的情况下硬编码 OTEL 端点或 SDK v3 写法；v4 以 base_url + CallbackHandler 为准。

12.5 用 Skill 配套的 CLI 验证 mystu 上报

配置与 mystu .env 相同的环境变量后，可用 CLI 核对 Cloud 是否收到 trace（Skill references/cli.md）：

# 发现可用 API 资源
npx langfuse-cli api __schema

# 列出最近 trace（具体 resource 名以 __schema 为准）
export LANGFUSE_PUBLIC_KEY=pk-lf-...
export LANGFUSE_SECRET_KEY=sk-lf-...
export LANGFUSE_BASE_URL=https://cloud.langfuse.com   # 美国区: us.cloud.langfuse.com

npx langfuse-cli api traces list --limit 5

CLI 适合：脚本化抽检、CI 夜间检查、不打开浏览器时确认 mystu-agent-stream 是否入库。

12.6 与 CE 工作流配合

阶段	建议
需求	/ce-brainstorm 定范围（Cloud vs 自托管、三接口、HITL）
计划	/ce-plan → docs/plans/2026-06-24-001-feat-langfuse-tracing-plan.md
实现	/ce-work + 附加 langfuse Skill，按 instrumentation.md 实现
排障 401	/ce-compound 沉淀 → docs/solutions/integration-issues/langfuse-cloud-401-tracing-auth.md
日常看 trace	docs/langfuse-trace-practical-guide.md（UI 向）；Skill error-analysis.md（质量向）

一句话：Skill 负责「按官方最新实践做对接与审计」，mystu 文档负责「讲清原理与本仓库代码路径」——两者互补，不宜只读一篇。

---

13. 延伸阅读

文档	内容
.cursor/skills/langfuse/SKILL.md	Langfuse 官方 Cursor Skill（Documentation First、CLI、references 索引）
.cursor/skills/langfuse/references/instrumentation.md	接入 baseline 与审计清单
docs/langfuse-trace-practical-guide.md	UI 查看 trace、日常开发工作流
docs/solutions/integration-issues/langfuse-cloud-401-tracing-auth.md	Cloud 401 鉴权失败解法
docs/checkpoint-explained.md	thread_id 与存档机制
docs/interrupts-explained.md	HITL 中断与 resume
docs/brainstorms/2026-06-24-langfuse-tracing-requirements.md	mystu 需求与范围
docs/plans/2026-06-24-001-feat-langfuse-tracing-plan.md	实现计划与 U1–U6
Langfuse LangChain 集成官方文档	Callback 与 metadata 约定
Langfuse Python SDK 概览	flush / shutdown / OTEL

---