5. 导出管道（Export Pipeline）

---

1. 管道组成

Span 结束
    │
    ▼
_LangfuseBatchSpanProcessor.on_end()     ← batch_export=True（默认）
    │   或 _LangfuseSpanProcessor        ← batch_export=False
    │
    ├─ _should_skip_span()               过滤
    ├─ _transform_span_for_langfuse()     映射
    │
    ▼
BatchSpanProcessor 内部队列
    │  （定时 / 批量 flush）
    ▼
_LangfuseOTLPExporter.export(spans)
    │
    ▼
HTTP POST {host}/api/public/otel/v1/traces
    Headers:
      Authorization: Basic base64(pk:sk)
      Content-Type: application/x-protobuf

---

2. OTLP Exporter 配置

auth_string = base64.b64encode(f"{public_key}:{secret_key}".encode()).decode()
endpoint = f"{host.rstrip('/')}/api/public/otel/v1/traces"
headers = {"Authorization": f"Basic {auth_string}"}
exporter = _LangfuseOTLPExporter(endpoint=endpoint, headers=headers)

• 协议：OTLP over HTTP/protobuf
• 认证：HTTP Basic Auth（public_key 作 username，secret_key 作 password）
• 与 Langfuse SDK 的 REST API 不同，这是 Langfuse 的 OTel 接入端点

---

3. Batch vs Simple Processor

模式	类	行为
batch_export=True（默认）	_LangfuseBatchSpanProcessor	异步批量，降低 HTTP 开销，进程退出前可能需 flush
batch_export=False	_LangfuseSpanProcessor	每个 Span 结束立即导出，调试友好

生产环境建议保持默认 batch_export=True。

---

4. Span 过滤规则（_should_skip_span）

默认 enable_a2a_trace=False 时，以下 Span 不会 导出到 Langfuse：

按 instrumentation scope 过滤

scope.name	说明
a2a-python-sdk	A2A 协议 SDK 自动埋点
opentelemetry.instrumentation.httpx	HTTP 客户端
opentelemetry.instrumentation.urllib3	HTTP 客户端
opentelemetry.instrumentation.requests	HTTP 客户端
opentelemetry.instrumentation.asgi	ASGI 服务
opentelemetry.instrumentation.fastapi	FastAPI 路由

按 Span 名称过滤

以 HTTP 、GET 、POST 等 HTTP 方法前缀开头的 Span。

关闭过滤

LangfuseConfig(..., enable_a2a_trace=True)

A2A 服务部署时需要看到 a2a-sdk 和 HTTP 层 Trace 时可开启。

---

5. Debug 日志

_LangfuseOTLPExporter.export() 在发送前对每个 Span 打 debug 日志：

=== Exporting span to Langfuse ===
{span JSON}

_LangfuseBatchSpanProcessor.on_end() 也会 log 映射后的 Span JSON。

排查「有埋点无上报」时，先确认：

1. langfuse_setup() 是否成功
2. log level 是否能看到 exporter 输出
3. 网络是否可达 {host}/api/public/otel/v1/traces

---

6. 全局配置 _langfuse_config

setup() 将 LangfuseConfig 存入模块级 _langfuse_config，供 SpanProcessor 读取：

• compatibility_old_version — 选择新旧映射函数
• enable_a2a_trace — 控制过滤行为

只应调用一次 setup()，重复调用会覆盖全局 TracerProvider。

---

7. 与 Langfuse SDK flush 的对比

cc_to_langfuse	trpc-agent
显式 langfuse.flush() 后才推进 offset	BatchSpanProcessor 自动批量发送
flush 失败则不推进状态，保证 at-least-once	OTel SDK 内置重试/队列（取决于 SDK 配置）
进程退出需手动 flush	短生命周期脚本需注意 Batch 未及时 flush

对于一次性脚本，可在结束前调用：

from opentelemetry import trace
provider = trace.get_tracer_provider()
if hasattr(provider, 'force_flush'):
    provider.force_flush()

---

8. 故障排查清单

症状	检查项
完全无 Trace	setup() 是否在 Runner 前调用；credentials 是否正确
只有部分 Span	是否被 _should_skip_span 过滤
Generation 无 token	LLM response 是否含 usage_metadata
Trace 名被覆盖	AgentTool 嵌套场景是否误设 trace.name
旧版 Langfuse 显示异常	开启 compatibility_old_version=True
401 / 403	public_key:secret_key 是否正确

---

9. 多后端共存

tRPC-Agent 生态还支持 Galileo、ZhiyanLLM 等 OTel 后端。Langfuse 的 setup() 会 trace.set_tracer_provider() 设为全局 Provider。

若需同时上报多个后端，需在应用层合并 SpanProcessor 到同一 TracerProvider，或选用支持 multi-exporter 的配置方式——默认 langfuse_setup() 会替换全局 Provider，与其他 setup() 互斥，以最后调用者为准。