Pipecat Tracing 集成

本指南演示如何将 Litefuse 与 Pipecat 集成，对实时语音 Agent 进行可观测性与 tracing。按这些步骤操作，你将能在 Litefuse 控制台中监控、调试并评估你的 Pipecat Agent。

Pipecat (仓库) 是一个用于构建实时语音和多模态对话式 AI Agent 的开源 Python 框架。它由 Daily 开发，支持完全可编程的 AI 语音 Agent 以及多模态交互，是希望构建对话式 AI 系统的开发者的灵活方案。

Pipecat Agent 对话在 Litefuse 中的示例。

特性

分层 Tracing：跟踪完整对话、轮次以及服务调用
服务 Tracing：为 TTS、STT 和 LLM 服务提供带有丰富上下文的详细 span
TTFB 指标：捕获 Time To First Byte 指标用于延迟分析
使用统计：跟踪 TTS 字符数与 LLM token 使用情况

Trace 结构

Trace 按层级组织：

Conversation (conversation-uuid)
├── turn-1
│   ├── stt_deepgramsttservice
│   ├── llm_openaillmservice
│   └── tts_cartesiattsservice
└── turn-2
    ├── stt_deepgramsttservice
    ├── llm_openaillmservice
    └── tts_cartesiattsservice
    turn-N
    └── ...

这种组织方式让你能逐对话、逐轮次进行跟踪。

关于此 trace 结构的重要说明：

一条 trace = 一次完整对话：由于 Pipecat 组织其 OpenTelemetry span 的方式，一条 trace 代表一次完整的对话。这与典型的 Litefuse 模式不同 —— 通常一条 trace 对应一次交互，完整对话会被归入会话之下。在 Pipecat 中无需将多条 trace 归并到会话中 —— 你的整个对话已经包含在一条 trace 内。
默认 trace 名：默认情况下，所有 trace 名为 conversation 或对话 uuid。请参考下方重命名 Trace 中关于如何 patch span 来自定义 trace 名的示例。

关于 Pipecat span 属性的更多信息，请查看 Pipecat 文档。

端到端示例

视频指南端到端示例

快速开始

Pipecat 支持 OpenTelemetry tracing，Litefuse 提供了 OpenTelemetry endpoint。按以下步骤操作，你就可以为 Pipecat 应用启用 Litefuse tracing。

获取 Litefuse API Key

在 Litefuse Cloud 中创建项目，或自托管 Litefuse，并复制你的 API Key。

环境配置

对你的 Litefuse public 与 secret Key 进行 base64 编码：

terminal

echo -n "pk-lf-1234567890:sk-lf-1234567890" | base64

创建一个 .env 文件，写入 API Key 以启用 tracing：

.env

ENABLE_TRACING=true
# OTLP endpoint (defaults to localhost:4317 if not set)
OTEL_EXPORTER_OTLP_ENDPOINT="https://litefuse.cloud/api/public/otel"
OTEL_EXPORTER_OTLP_HEADERS=Authorization=Basic%20<base64_encoded_api_key>
# Set to any value to enable console output for debugging
# OTEL_CONSOLE_EXPORT=true

更多细节请参考 Litefuse OpenTelemetry 文档。

为你的 Pipeline Task 添加 OpenTelemetry

在你的 Pipecat 应用中启用 tracing：

main.py

# Initialize OpenTelemetry with the http exporter
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
 
# Configured automatically from .env
exporter = OTLPSpanExporter()
 
setup_tracing(
    service_name="pipecat-demo",
    exporter=exporter,
)
 
# Enable tracing in your PipelineTask
task = PipelineTask(
    pipeline,
    params=PipelineParams(
        allow_interruptions=True,
        enable_metrics=True,  # Required for some service metrics
    ),
    enable_tracing=True,  # Enables both turn and conversation tracing
    conversation_id="customer-123",  # Optional - will auto-generate if not provided
)

更多细节请参考 OpenTelemetry Python 文档。

添加 Trace 输入与输出

默认情况下，Pipecat trace 不会在 trace 层级包含对话的输入与输出。你可以 patch Pipecat 的服务装饰器，设置 langfuse.trace.input 与 langfuse.trace.output 属性，将首次 LLM 调用的消息作为 trace 输入，最后一次 LLM 响应作为 trace 输出：

main.py

def patch_trace_input_output():
    from pipecat.utils.tracing import service_decorators
    original = service_decorators.add_llm_span_attributes
    first_call = [True]
 
    def patched(span, *args, **kwargs):
        original(span, *args, **kwargs)
 
        # Set the input of the first LLM call as the trace input
        if first_call[0] and kwargs.get("messages"):
            span.set_attribute("langfuse.trace.input", kwargs["messages"])
            first_call[0] = False
 
        # Set the output of each LLM call as the trace output (last one wins)
        orig_set = span.set_attribute
        def new_set(key, value):
            orig_set(key, value)
            if key == "output":
                orig_set("langfuse.trace.output", value)
        span.set_attribute = new_set
 
    service_decorators.add_llm_span_attributes = patched
 
# Apply patch before tracing setup
patch_trace_input_output()

重命名 Trace（可选）

默认情况下，所有 Pipecat trace 都名为 conversation 或使用对话 UUID。要自定义 trace 名，可以扩展该 patch 以设置 langfuse.trace.name 属性：

main.py

def patch_trace_name():
    from pipecat.utils.tracing import service_decorators
    original = service_decorators.add_llm_span_attributes
    first_call = [True]
 
    def patched(span, *args, **kwargs):
        original(span, *args, **kwargs)
 
        # Set a custom trace name on the first LLM call
        if first_call[0]:
            span.set_attribute("langfuse.trace.name", "pipecat-chatbot")
            first_call[0] = False
 
    service_decorators.add_llm_span_attributes = patched
 
# Apply patch before tracing setup
patch_trace_name()

如果你想同时设置 trace 名、输入和输出，可以把这两个 patch 合并到一个函数中。

理解 Trace

Conversation Span：顶层 span，代表一次完整的对话
Turn Span：对话下的子 span，代表对话中的每一轮
Service Span：嵌套在 turn 下的详细服务操作
Service Attributes：每个服务都包含丰富的运行上下文：
- TTS：Voice ID、字符数、服务类型
- STT：转写文本、语言、模型
- LLM：消息、所用 token、模型、服务配置
Metrics：性能数据，例如 metrics.ttfb_ms 与处理耗时

工作原理

Tracing 系统由以下部分组成：

TurnTrackingObserver：检测对话轮次
TurnTraceObserver：为轮次与对话创建 span
Service Decorators：@traced_tts、@traced_stt、@traced_llm，用于服务特定的 tracing
Context Providers：在流水线不同部分之间共享上下文

排错

Litefuse 中没有 Trace：确认你的凭证正确，并参考此排错指南
缺少指标：检查 PipelineParams 中是否设置了 enable_metrics=True
连接错误：验证到 Litefuse 的网络连通性
Exporter 问题：尝试使用 Console exporter（OTEL_CONSOLE_EXPORT=true）验证 tracing 是否正常工作

参考

OpenAI Agents Pydantic AI

这个页面对你有帮助吗？

支持