Langfuse v4 数据链路参考

0 核对基准

1 架构与数据流

两个自有进程：Web（Next.js，界面与所有接口，含上报）和 Worker（Express，处理上报、评估、导出）。v4 没有新增服务。

2 客户端对接的六种情况

写入路径由 resolveOtelWritePath 决定，优先级：direct_header > direct_env > direct_scope > direct_org_cutoff（仅 Cloud）> dual。worker/src/queues/otelIngestionQueue.ts

3 上报协议与属性

入口

Python SDK 默认请求头

Authorization: Basic …
x-langfuse-sdk-name: python
x-langfuse-sdk-version: 5.2.0
x-langfuse-public-key: pk-lf-…
x-langfuse-ingestion-version: 4

langfuse/_client/span_processor.py

属性名（Langfuse 约定）

langfuse/_client/attributes.py · packages/shared/src/server/otel/OtelIngestionProcessor.ts

user_id / session_id 的查找顺序

1. span 属性：langfuse.user.id、user.id、langfuse.observation.metadata.langfuse_user_id、langfuse.trace.metadata.langfuse_user_id、ai.telemetry.metadata.userId
2. 资源属性：langfuse.user.id、user.id
3. 最后：user.name（span，再资源）

session 还识别 gen_ai.conversation.id。服务端不会在批内从根 span 继承；资源属性对整个服务生效，可用于“一个服务 = 一个租户”。

SDK 传播

from langfuse import get_client, propagate_attributes
with propagate_attributes(user_id="u-42", session_id="s-7", tags=["prod"], metadata={"tenant": "acme"}):
    ...   # 这里创建的每个 span 都带上这些属性
# as_baggage=True：通过 OTel Baggage（HTTP 头）传给下游服务 —— 会发给所有下游，不要放敏感值

langfuse/_client/propagation.py

4 转存与队列

OTel 队列任务只带指针：fileKey、authCheck.scope（projectId、orgId）、sdkName、sdkVersion、ingestionVersion、propagatedHeaders。分片键 {projectId}-{fileKey}。OtelIngestionProcessor.ts publishToOtelIngestionQueue · processEventBatch.ts · eventBucketPath.ts

5 后台处理与写入路径

写入缓冲：LANGFUSE_INGESTION_CLICKHOUSE_WRITE_BATCH_SIZE=1000、…_WRITE_INTERVAL_MS=1000、…_MAX_ATTEMPTS=3。worker/src/services/IngestionService/index.ts · worker/src/env.ts

6 表结构

events_full（迁移 0039）

ENGINE = ReplacingMergeTree(event_ts, is_deleted)
PARTITION BY toYYYYMM(start_time)
PRIMARY KEY (project_id, toStartOfMinute(start_time), xxHash32(trace_id))
ORDER BY (project_id, toStartOfMinute(start_time), xxHash32(trace_id), span_id, start_time)
SAMPLE BY xxHash32(trace_id)
SETTINGS index_granularity_bytes = '64Mi', merge_max_block_size_bytes = '64Mi', enable_full_text_index = 1, …

索引：bloom_filter（span_id、trace_id、user_id、session_id）；text 索引 splitByNonAlpha（lower(input)、lower(output)、metadata_names、metadata_values）；ngrambf（metadata 拼接值、lower(name)、lower(trace_name)、lower(user_id)、lower(session_id)）。canonical/0039, 0043, 0049

events_core（0040）与 events_core_mv（0041）

结构相同；物化视图写入时 leftUTF8(input, 200)、leftUTF8(output, 200)、每个 metadata 值截到 200。查询写逻辑表 events，编译期路由：读到被截断列 → events_full，否则 events_core；短值等值/前缀的 metadata 过滤在 core 上也正确；events_truncated 强制 core。packages/shared/src/server/query-ast/eventsRouting.ts

元数据展平规则

读取还原：mapFromArrays(arrayReverse(metadata_names), arrayReverse(metadata_values))。键名本身含点时与路径无法区分。packages/shared/src/server/otel/utils.ts · query-ast/compile.ts

7 格式转换总表

8 大体积数据：七道关口

9 脱敏

回调契约

POST {LANGFUSE_INGESTION_MASKING_CALLBACK_URL}
Content-Type: application/json
X-Langfuse-Org-Id: <orgId>
X-Langfuse-Project-Id: <projectId>
<LANGFUSE_INGESTION_MASKING_PROPAGATED_HEADERS 中列出的上报请求头>

[ { "resource": {...}, "scopeSpans": [ { "scope": {...}, "spans": [ ... ] } ] } ]   ← 原始 resourceSpans

→ 200，响应体：同结构的 JSON（改写后的 resourceSpans）

packages/shared/src/server/ee/ingestionMasking · worker/src/queues/otelIngestionQueue.ts · worker/src/features/otel-ingestion/prepareOtelBatch.ts（开启 early media 时顺序：脱敏 → 校验 → 抽取媒体）

建议的三层方案

1. 客户端：SDK mask 处理确定字段（手机号、证件号）；媒体在此抽走。
2. 网关：OTel Collector 作为唯一入口；统一加 x-langfuse-ingestion-version: 4；补租户资源属性；删除不允许出域的属性；限制批大小；采样保留根 span。
3. 服务端：脱敏回调 + FAIL_CLOSED=true，做模型识别等复杂处理；事件桶与媒体桶 SSE-KMS + 最小权限 + 短生命周期；打开字段溢出。
4. 导出侧：导到自有桶，先读清单，按 id 去重，用 metadata 数组还原结构；回查原件用 blob_storage_file_path。

10 双写与迁移

写入模式

另：events_only 要求 LANGFUSE_MIGRATION_V4_ALLOW_PREVIEW_OPT_IN=true。worker/src/env.ts validateV4Flags

双写传播作业

• 暂存表 observations_batch_staging：按 s3_first_seen_timestamp 3 分钟分区，TTL 48 h，ttl_only_drop_parts。
• Redis 游标 langfuse:event-propagation:last-processed-partition；只处理早于 now() − LANGFUSE_EXPERIMENT_EVENT_PROPAGATION_PARTITION_DELAY_MINUTES（默认 10）的分区。
• SQL：observations_batch_staging FINAL LEFT JOIN traces（LIMIT 1 BY project_id, id 取最新，回看 ≤ 7 天）→ INSERT events_full；mapConcat(obs.metadata, trace.metadata)，observation 优先；缺父节点 → t-<trace_id>；排除 24 h 内的实验 trace；parallel_view_processing=1；超时 30 min。
• 健康检查：GET :3030/api/health?failIfEventPropagationStuck=true，阈值 LANGFUSE_EVENT_PROPAGATION_STUCK_THRESHOLD_MINUTES（代码默认 35）。

worker/src/features/eventPropagation/handleEventPropagationJob.ts · canonical/0038

历史回填（后台迁移链）

1. createRootSpansFromTraces：span_id = t-||trace_id，type SPAN，按 yyyymm 分区分块
2. rewriteObservationsToPidTidSorting
3. backfillEventsFullFromObservations（JOIN traces 补属性，trace 元数据不复制）
4. backfillEventsFullFromDatasetRunItems
5. dropPidTidSortingTables（LANGFUSE_BACKGROUND_MIGRATION_V4_DROP_PID_TID_SORTING_TABLES）

11 导出

字段组：core（必选）· basic · time · io · metadata · model · usage · prompt · metrics · trace_context · tools。packages/shared/src/domain/observation-field-groups.ts · worker/src/features/blobstorage

12 环境变量速查

13 文档与代码的差异

14 改造风险清单

• 同一 span 重复上报 → 两条记录（读取不保证去重）。
• trace 属性未复制到子 span → 子 span 按用户/会话查不到。
• 未识别属性进 metadata.attributes / metadata.resourceAttributes，嵌套键不能直接过滤。
• events_only 下 /api/public/ingestion 只收评分。
• 脱敏回调放行模式下，回调故障期间未脱敏数据入库；S3 原始事件始终未脱敏。
• 大值放在 metadata 时不受写入兜底截断，可能导致写入失败；打开字段溢出。
• 导出：latency 单位随创建日期不同；Parquet 缺单价列；导出未选 io/metadata 时内容是截断的。
• 采集器采样丢掉根 span → 层级断裂、根上的整体输入输出缺失。

15 代码位置索引