> ## Documentation Index
> Fetch the complete documentation index at: https://platform.eldros.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Pipecat

> Hook the turn-stopped events, finalize at pipeline end — transcripts and Pipecat's native traces in one pipeline.

Observe a Pipecat pipeline through its `LLMContextAggregatorPair` turn events:

```python theme={null}
import eldros_sdk
from eldros_sdk.integrations.pipecat import observe_aggregators

eldros_sdk.init(traffic_type="prod")            # once, at startup

user_agg, assistant_agg = LLMContextAggregatorPair(context)
observer = observe_aggregators(user_agg, assistant_agg, session_id=conversation_id)

# ... build pipeline, run ...
await runner.run(task)
observer.finalize()                             # closes the final turn — required
```

<Warning>
  **Always call `observer.finalize()` when the pipeline ends.** A turn span closes when
  the *next* user turn starts; the conversation's final turn has no successor, and an
  unclosed span is never exported — skipping finalize silently drops the last exchange
  of every call.
</Warning>

## What gets captured

Each user ↔ assistant exchange (from `on_user_turn_stopped` / `on_assistant_turn_stopped`)
becomes a **turn span** — `input.value`/`output.value` plus structured `eldros.message`
events, grouped by your `session_id`. Assistant turns with empty content (realtime
mode) are skipped rather than recorded blank.

## Traces: use Pipecat's built-in OpenTelemetry

Pipecat ships native tracing with a `conversation → turn → stt/llm/tts` span
hierarchy. Point it at Eldros and stamp your session id on it — both layers land in
one pipeline:

```python theme={null}
from eldros_sdk.integrations.pipecat import tracing_exporter

setup_tracing(service_name="my-agent", exporter=tracing_exporter())

task = PipelineWorker(                          # PipelineTask on older Pipecat versions
    pipeline,
    enable_tracing=True,
    enable_turn_tracking=True,
    additional_span_attributes={"session.id": conversation_id},
)
```

`tracing_exporter()` returns an OTLP exporter wired to your resolved Eldros config —
it raises at startup if `init()` hasn't resolved credentials, so misconfiguration
fails loudly rather than dropping every trace.

<Note>
  Pipecat's API surface moves quickly (`PipelineTask` → `PipelineWorker`,
  `setup_tracing` import paths). The Eldros hooks are written defensively against
  version drift, but cross-check the Pipecat-side names against the Pipecat docs for
  your installed version.
</Note>