Tuner

https://docs.usetuner.ai/
Admin
Tuner is an observability and monitoring platform built for voice AI agents, providing full...
Tokens:94,681
Snippets:946
Trust Score:4.1
Update:3 months ago
Show doc for...
Tokens:
Raw
### Install tuner-stt-observer

Source: https://docs.usetuner.ai/docs/api-and-integrations/libraries/tuner-stt

Install the core tuner library and the STT observer. This is the base installation required for using the STT observer.

```shell
pip install tuner-core tuner-stt-observer
```

--------------------------------

### Install Tuner Libraries

Source: https://docs.usetuner.ai/docs/api-and-integrations/custom-integration/quickstart

Install the core Tuner library and optional provider SDKs using pip. Install provider SDKs with extras like '[deepgram]' or '[cartesia]' to include only the necessary dependencies.

```shell
pip install tuner-core "tuner-stt-observer[deepgram]" "tuner-tts-observer[cartesia]" tuner-langchain
```

--------------------------------

### Minimal Tuner Langchain Starter

Source: https://docs.usetuner.ai/docs/api-and-integrations/libraries/tuner-langchain

This snippet demonstrates a basic setup for using TunerAccumulator and TunerLangGraphHandler with Langchain. It includes functions to handle individual turns and build a complete transcript at the end of a call. Ensure you have the necessary Tuner and Langchain libraries installed.

```python
import time
from tuner_langchain import TunerAccumulator, CaptureConfig
from tuner_langchain.handlers import TunerLangGraphHandler
from tuner_langchain.segment_builder import segments_from_invocation

# Per call
call_start_ms = int(time.time() * 1000)
accumulator = TunerAccumulator(capture=CaptureConfig(tool_inputs=False))
handler = TunerLangGraphHandler(accumulator)

voice_turns = []   # your own STT/TTS turns

# Per turn
def handle_turn(user_text: str, state: dict) -> str:
    voice_turns.append({"role": "user", "text": user_text, "start_ms": ...})
    result = compiled_graph.invoke(state, config={"callbacks": [handler]})
    response = result["response"]
    voice_turns.append({"role": "agent", "text": response, "start_ms": ...})
    return response

# At hang-up
def build_transcript() -> list[dict]:
    call_start_ns = call_start_ms * 1_000_000
    segments = []
    for inv in accumulator.get_invocations():
        segments.extend(segments_from_invocation(inv, call_start_ns))
    return sorted(voice_turns + segments, key=lambda e: e.get("start_ms", 0))

# POST build_transcript() inside your Tuner Create Call payload
```

--------------------------------

### Install Tuner for LLM Tracing

Source: https://docs.usetuner.ai/docs/api-and-integrations/custom-integration/quickstart

Install Tuner with the Langchain integration for LLM tracing capabilities.

```bash
tuner-core tuner-langchain
```

--------------------------------

### Install Tuner Core and TTS Observer

Source: https://docs.usetuner.ai/docs/api-and-integrations/libraries/tuner-tts

Install the necessary Tuner libraries for core functionality and TTS observation.

```shell
pip install tuner-core tuner-tts-observer
```

--------------------------------

### Install Speechmatics Adapter

Source: https://docs.usetuner.ai/docs/api-and-integrations/libraries/tuner-stt

Install specific STT provider adapters as optional extras to minimize dependencies. Use this command to install the Speechmatics adapter.

```shell
pip install "tuner-stt-observer[speechmatics]"   # SpeechmaticsAdapter
```

--------------------------------

### Full Pipecat Bot Example

Source: https://docs.usetuner.ai/docs/api-and-integrations/connecting-to-pipecat

This Python code demonstrates a full example of setting up a Pipecat bot. It includes initializing observers, defining a cost calculator, and configuring a pipeline for call processing. Ensure you have the TUNER_API_KEY and TUNER_WORKSPACE_ID environment variables set.

```python
import os
import uuid
from tuner_pipecat_sdk import Observer
from tuner_pipecat_sdk.models import DisconnectReason
from pipecat.observers.turn_tracking_observer import TurnTrackingObserver
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.task import PipelineTask, PipelineParams


async def run_bot(call_data, context, transport, stt, llm, tts, context_aggregator):
    # call_data is the payload from your telephony WebSocket handler
    # e.g. _, call_data = await parse_telephony_websocket(websocket)

    # Read the SIP Call-ID from the provider payload — the field name varies
    # Twilio example: forward it as a customParameter (see SIP section above)
    sip_call_id = call_data.get("customParameters", {}).get("SipCallId")

    turn_tracker = TurnTrackingObserver()
    _reason = None

    # update the rates to match your models
    def calculate_cost(usage) -> float:
        # OpenAI gpt-4o-mini pricing (per token / per character / per second)
        llm_cost  = (usage.llm_prompt_tokens     or 0) * 0.000_000_150
        llm_cost += (usage.llm_completion_tokens or 0) * 0.000_000_600
        # OpenAI TTS-1: $15 per 1M characters
        tts_cost  = (usage.tts_characters        or 0) * 0.000_015
        # OpenAI gpt-4o-transcribe: $6 per 1M audio seconds
        stt_cost  = usage.stt_audio_seconds            * 0.000_006
        return (llm_cost + tts_cost + stt_cost) * 100

    observer = Observer(
        api_key=os.environ["TUNER_API_KEY"],
        workspace_id=int(os.environ["TUNER_WORKSPACE_ID"]),
        agent_id="fa4da74c-...",
        call_id=str(uuid.uuid4()),
        base_url="https://api.usetuner.ai",
        call_type="phone_call",
        asr_model="deepgram/nova-3",
        llm_model="gpt-4o-mini",
        tts_model="cartesia/sonic",
        sip_call_id=sip_call_id,
        disconnection_reason_resolver=lambda: _reason,
        cost_calculator=calculate_cost,
    )

    observer.attach_turn_tracking_observer(turn_tracker)

    pipeline = Pipeline([
        transport.input(),
        stt,
        context_aggregator.user(),
        llm,
        tts,
        transport.output(),
        context_aggregator.assistant(),
    ])

    task = PipelineTask(
        pipeline,
        params=PipelineParams(
            enable_metrics=True,
            enable_usage_metrics=True,
        ),
        observers=[observer, observer.latency_observer, turn_tracker],
    )

    _reason = DisconnectReason.USER_HANGUP

```

--------------------------------

### Full TunerPlugin Configuration Example

Source: https://docs.usetuner.ai/docs/api-and-integrations/connecting-to-livekit

A comprehensive example demonstrating how to configure the TunerPlugin with API keys, workspace ID, agent ID, call type, recording URL resolver, cost calculator, metadata, agent version, timeouts, retries, and enablement.

```typescript
import { TunerPlugin } from '@usetuner.ai/livekit-sdk';
import type { metrics } from '@livekit.ai/agents';

function calculateCost(usage: metrics.UsageSummary): number {
  const usd =
    usage.llmPromptTokens * 0.000_003 +
    usage.llmCompletionTokens * 0.000_015 +
    usage.ttsCharactersCount * 0.000_030;
  return usd * 100; // Tuner expects USD cents
}

async function getRecordingUrl(roomName: string, jobId: string): Promise<string> {
  return (await myStorage.getUrl(jobId)) ?? 'pending';
}

new TunerPlugin(session, ctx, {
  apiKey: process.env.TUNER_API_KEY,
  workspaceId: Number.parseInt(process.env.TUNER_WORKSPACE_ID!, 10),
  agentId: process.env.TUNER_AGENT_ID,
  callType: 'phone_call',
  recordingUrlResolver: getRecordingUrl,
  costCalculator: calculateCost,
  extraMetadata: { env: 'prod', region: 'us-east-1' },
  agentVersion: 42,
  timeoutSeconds: 20,
  maxRetries: 3,
  enabled: process.env.NODE_ENV === 'production',
});
```

--------------------------------

### Install Tuner LiveKit SDK for Python

Source: https://docs.usetuner.ai/docs/api-and-integrations/connecting-to-livekit

Install the Python SDK using pip. Ensure you have Python 3.10+, livekit-agents 1.4+, and aiohttp 3.9+.

```shell
pip install tuner-livekit-sdk
```

--------------------------------

### Install Tuner TTS Observer with WebSocket Support

Source: https://docs.usetuner.ai/docs/api-and-integrations/libraries/tuner-tts

Install the tuner-tts-observer library with the 'cartesia' extra for WebSocket support, which is recommended.

```shell
pip install "tuner-tts-observer[cartesia]"
```

--------------------------------

### Install LiveKit SDK for Node.js

Source: https://docs.usetuner.ai/docs/api-and-integrations/connecting-to-livekit

Install the Tuner AI SDK for Node.js using npm. This requires Node.js 18 or later, and specific versions of the LiveKit agents and RTC SDK.

```shell
npm install @usetuner.ai/livekit-sdk
```

--------------------------------

### Install tuner-langchain

Source: https://docs.usetuner.ai/docs/api-and-integrations/libraries/tuner-langchain

Install the tuner-langchain library using pip. Ensure you have Python 3.10+ and compatible versions of langchain-core and langgraph.

```bash
pip install tuner-langchain
```

--------------------------------

### Install STT Provider Adapter

Source: https://docs.usetuner.ai/docs/api-and-integrations/libraries/tuner-stt

Install specific STT provider adapters as optional extras to minimize dependencies. Use this command to install the Deepgram adapter.

```shell
pip install "tuner-stt-observer[deepgram]"       # DeepgramAdapter
```

--------------------------------

### Install LangChain Extra

Source: https://docs.usetuner.ai/docs/api-and-integrations/connecting-to-pipecat

Install the optional LangChain extra for the tuner-pipecat-sdk. This is required to use LangChain or LangGraph integration features.

```shell
pip install tuner-pipecat-sdk[langchain]
```

--------------------------------

### Initiate Tuner Agent Setup in Chat

Source: https://docs.usetuner.ai/docs/mcp/set-up-your-agent

Paste this command into your chat with the AI to begin configuring your agent's observability using the Tuner MCP server. Follow the 'tuner_setup_agent' prompt provided by the server.

```text
Set up Tuner observability for this agent using the Tuner MCP.
Fetch and follow the 'tuner_setup_agent' prompt from the Tuner MCP server.
```

--------------------------------

### Install Tuner for STT Observability with Deepgram

Source: https://docs.usetuner.ai/docs/api-and-integrations/custom-integration/quickstart

Install Tuner with the STT observer for Deepgram to enable speech-to-text observability.

```bash
tuner-core "tuner-stt-observer[deepgram]"
```

--------------------------------

### Install Tuner with Speechmatics and Cartesia

Source: https://docs.usetuner.ai/docs/api-and-integrations/custom-integration/quickstart

Install the Tuner packages required for a stack using Speechmatics for STT and Cartesia for TTS.

```bash
tuner-core "tuner-stt-observer[speechmatics]" "tuner-tts-observer[cartesia]"
```

--------------------------------

### Example E-commerce Support Bot System Prompt

Source: https://docs.usetuner.ai/user-guide/quick-start/how-to-review-your-first-call

This is an example system prompt for an e-commerce support bot. It defines the bot's persona and primary function, and sets a constraint on the types of questions it should answer.

```text
You are a friendly and helpful e-commerce support bot. Your primary goal is to answer customer questions about their orders and provide tracking information. Do not answer questions that are not related to e-commerce support.
```

--------------------------------

### Install Tuner Pipecat SDK

Source: https://docs.usetuner.ai/docs/api-and-integrations/connecting-to-pipecat

Install the Tuner Pipecat SDK using pip. Ensure you have Python 3.11-3.13 and pipecat-ai version 1.0.0 or higher.

```shell
pip install tuner-pipecat-sdk
```

--------------------------------

### Install Tuner with Deepgram, LangGraph, and Cartesia

Source: https://docs.usetuner.ai/docs/api-and-integrations/custom-integration/quickstart

Install the necessary Tuner packages for a stack including Deepgram for STT, LangGraph for orchestration, and Cartesia for TTS.

```bash
tuner-core "tuner-stt-observer[deepgram]" "tuner-tts-observer[cartesia]" tuner-langchain
```

--------------------------------

### Add Tuner Plugin to Python Agent Entrypoint

Source: https://docs.usetuner.ai/docs/api-and-integrations/connecting-to-livekit

Import TunerPlugin and instantiate it with the AgentSession and context before starting the session. This automatically enables call data capture.

```python
from tuner import TunerPlugin

async def entrypoint(ctx: JobContext):
    session = AgentSession(...)
    TunerPlugin(session, ctx)          # wires itself automatically
    await session.start(...)
```

--------------------------------

### Node.js Webhook Setup

Source: https://docs.usetuner.ai/docs/api-and-integrations/vapi-transient-assistants

Set up a Node.js Express webhook to receive VAPI end-of-call reports and send them to Tuner. This handler should be fire-and-forget.

```javascript
app.post("/vapi/webhook", (req, res) => {
  const message = req.body?.message;

  if (message?.type === "end-of-call-report") {
    // fire-and-forget, never blocks, never throws
    sendCallToTuner(message);
  }

  res.sendStatus(200);
});
```

--------------------------------

### Example SIP INVITE with Call-ID Header

Source: https://docs.usetuner.ai/docs/simulation/other-platforms

This is a standard SIP INVITE request containing the essential Call-ID header that Tuner uses for linking.

```sip
INVITE sip:your-agent@provider.example SIP/2.0
Call-ID: 2jf9wCCXfwgxKkTOfLcEsA5LD6f
...
```

--------------------------------

### Python FastAPI Webhook Setup

Source: https://docs.usetuner.ai/docs/api-and-integrations/vapi-transient-assistants

Set up a Python FastAPI webhook to receive VAPI end-of-call reports and send them to Tuner. This handler should never block or raise exceptions.

```python
from send_to_tuner import send_call_to_tuner

# your existing VAPI webhook endpoint (FastAPI shown)
@app.post("/vapi/webhook")
async def vapi_webhook(request: Request):
    message = (await request.json()).get("message") or {}

    if message.get("type") == "end-of-call-report":
        # never blocks your handler, never raises
        send_call_to_tuner(message)

    return {}
```

--------------------------------

### Python LiveKit Agent with Tuner Plugin

Source: https://docs.usetuner.ai/docs/api-and-integrations/connecting-to-livekit

Example of setting up a LiveKit agent in Python with the Tuner plugin. It includes extracting SIP correlation IDs, calculating call costs, and resolving recording URLs. Ensure environment variables for API key, workspace ID, and agent ID are set.

```python
import os
from livekit import rtc
from livekit.agents import JobContext, AgentSession
from tuner import TunerPlugin


def _extract_sip_correlation_id(ctx: JobContext) -> str | None:
    for participant in ctx.room.remote_participants.values():
        if participant.kind != rtc.ParticipantKind.PARTICIPANT_KIND_SIP:
            continue
        attributes = dict(getattr(participant, "attributes", {}) or {})
        sip_call_id_full = attributes.get("sip.callIDFull")
        if isinstance(sip_call_id_full, str) and sip_call_id_full:
            return sip_call_id_full
    return None


def calculate_cost(usage) -> float:
    usd = (
        usage.llm_prompt_tokens     * 0.000_003
        + usage.llm_completion_tokens * 0.000_015
        + usage.tts_characters_count  * 0.000_030
    )
    return usd * 100  # Tuner expects USD cents


async def get_recording_url(room_name: str, job_id: str) -> str:
    return await my_storage.get_url(job_id) or "pending"


async def entrypoint(ctx: JobContext):
    session = AgentSession(...)

    await ctx.connect()
    sip_correlation_id = _extract_sip_correlation_id(ctx)

    TunerPlugin(
        session,
        ctx,
        api_key=os.environ["TUNER_API_KEY"],
        workspace_id=int(os.environ["TUNER_WORKSPACE_ID"]),
        agent_id=os.environ["TUNER_AGENT_ID"],
        call_type="phone_call",
        recording_url_resolver=get_recording_url,
        cost_calculator=calculate_cost,
        sip_correlation_id=sip_correlation_id,
        extra_metadata={"env": "prod", "region": "us-east-1"},
        timeout_seconds=20.0,
        max_retries=3,
        enabled=True,
    )

    await session.start(...)

```

--------------------------------

### Calculate Segment Timing Information

Source: https://docs.usetuner.ai/docs/api-and-integrations/vapi-transient-assistants

Calculates the start, end, and duration in milliseconds for a given segment relative to the call's start time. Handles cases where call start time or segment timing data might be null.

```typescript
/** Work out a row's start/end/duration relative to the start of the call. */
function segmentTiming(
  m: Json,
  callStartMs: number | null,
): { startMs: number | null; endMs: number | null; durationMs: number | null } {
  let startMs: number | null = null;
  let endMs: number | null = null;
  let durationMs: number | null = null;

```

--------------------------------

### Initialize Tuner Observer and Turn Tracker

Source: https://docs.usetuner.ai/docs/api-and-integrations/connecting-to-pipecat

Create an Observer instance with your API key, workspace, agent, and call IDs. Attach a TurnTrackingObserver for call turn tracking.

```python
import uuid
from tuner_pipecat_sdk import Observer
from pipecat.observers.turn_tracking_observer import TurnTrackingObserver

turn_tracker = TurnTrackingObserver()

observer = Observer(
    api_key="YOUR_TUNER_API_KEY",
    workspace_id=42,
    agent_id="fa4da74c-...",
    call_id=str(uuid.uuid4()),
    base_url="https://api.usetuner.ai",
    asr_model="deepgram/nova-3",
    llm_model="gpt-4o-mini",
    tts_model="cartesia/sonic",
)

observer.attach_turn_tracking_observer(turn_tracker)
```

--------------------------------

### Wire SIP Correlation ID into Entrypoint

Source: https://docs.usetuner.ai/docs/api-and-integrations/connecting-to-livekit

This example shows how to integrate the SIP correlation ID extraction into your agent's entrypoint. Ensure you connect to the room first, then read the ID, and finally pass it to TunerPlugin.

```python
async def entrypoint(ctx: JobContext):
    await ctx.connect()                                     # 1. connect first
    sip_correlation_id = _extract_sip_correlation_id(ctx)   # 2. then read

    session = AgentSession(...)

    TunerPlugin(
        session, ctx,
        sip_correlation_id=sip_correlation_id,              # 3. pass it in
        # ... your other TunerPlugin options
    )

    await session.start(...)
```

--------------------------------

### Agent Initialization

Source: https://docs.usetuner.ai/docs/api-and-integrations/connecting-to-livekit

Initializes an Agent with a Langchain LLMAdapter. Configuration for thread ID is passed within the adapter.

```python
agent = Agent(
    instructions="",
    llm=langchain.LLMAdapter(
        wrapped_chain,
        stream_mode="messages",
        config={"configurable": {"thread_id": thread_id}},
    ),
)
    
```

--------------------------------

### Initialize Observer with Inline Credentials

Source: https://docs.usetuner.ai/docs/api-and-integrations/connecting-to-pipecat

Instantiate the Observer class by passing your Tuner API key, workspace ID, agent ID, and optionally the base URL directly in the code. A unique call ID is also generated.

```python
Observer(
    api_key="tr_api_...",
    workspace_id=123,
    agent_id="fa4da74c-...",
    call_id=str(uuid.uuid4()),
    base_url="https://api.usetuner.ai",
)
```

--------------------------------

### Calculate Segment Timing Relative to Call Start

Source: https://docs.usetuner.ai/docs/api-and-integrations/vapi-transient-assistants

This helper function calculates the start, end, and duration of a speech segment in milliseconds, relative to the beginning of the call. It supports different input formats for timing information.

```python
def _segment_timing(m: Json, call_start_ms: Optional[int]):
    """Work out a row's start/end/duration relative to the start of the call."""
    start_ms: Optional[int] = None
    end_ms: Optional[int] = None
    duration_ms: Optional[int] = None

    seconds = m.get("secondsFromStart")
    if seconds is not None:
        start_ms = math.ceil(seconds * 1000)
    if m.get("duration") is not None:
        duration_ms = math.ceil(m["duration"])

    if start_ms is not None and duration_ms is not None:
        end_ms = start_ms + duration_ms
    elif call_start_ms is not None:
        if m.get("time") is not None and start_ms is None:
            start_ms = math.ceil(m["time"] - call_start_ms)
        if m.get("endTime") is not None:
            end_ms = math.ceil(m["endTime"] - call_start_ms)
        if start_ms is not None and end_ms is not None:
            duration_ms = end_ms - start_ms
    return start_ms, end_ms, duration_ms
```

--------------------------------

### Configure Python SDK Credentials Inline

Source: https://docs.usetuner.ai/docs/api-and-integrations/connecting-to-livekit

Pass API key, workspace ID, and agent ID directly in the TunerPlugin constructor.

```python
TunerPlugin(
    session, ctx,
    api_key="tr_api_...",
    workspace_id=123,
    agent_id="fa4da74c-...",
)
```

--------------------------------

### End Call with Agent Hangup Reason (Tool Example)

Source: https://docs.usetuner.ai/docs/api-and-integrations/connecting-to-livekit

Example of using `context.session.shutdown()` and `get_job_context().shutdown(reason="agent_hangup")` within a function tool to properly end a call and record the reason.

```python
@function_tool
async def end_call(self, context: RunContext):
    """End the call once the user has been helped."""
    await context.session.shutdown(drain=True)
    get_job_context().shutdown(reason="agent_hangup")
```

--------------------------------

### Create Call with Tenant-Specific Metadata

Source: https://docs.usetuner.ai/docs/agent-configurations/evaluating-dynamic-agents

This Python snippet demonstrates how to log a call with dynamic tenant instructions embedded in the metadata. Ensure the `tuner` library is imported and necessary variables like `WORKSPACE_ID`, `AGENT_ID`, `session_id`, `start_ts`, `end_ts`, `recording_url`, and `transcript` are defined.

```python
tuner.create_call(
    workspace_id=WORKSPACE_ID,
    agent_remote_identifier=AGENT_ID,
    body={
        "call_id": f"call-{session_id}",
        "call_type": "web_call",
        "call_status": "call_ended",
        "start_timestamp": start_ts,
        "end_timestamp": end_ts,
        "recording_url": recording_url,
        "transcript_with_tool_calls": transcript,
        "metadata": {
            "instructions": tenant.system_prompt,
            "tenant": tenant.id,
        }
    }
)
```

--------------------------------

### Check Problem Timeline

Source: https://docs.usetuner.ai/user-guide/quick-start/diagnose-your-agent

Inquire about the historical occurrence of a problem to determine if it started with a specific agent version or has been present historically.

```text
Did this problem start at a specific agent version, or has it always been there?
```

--------------------------------

### Create a Tuner Session

Source: https://docs.usetuner.ai/docs/api-and-integrations/custom-integration/quickstart

Initialize a TunerSession at the beginning of each call. Configure the session with desired ASR, LLM, and TTS models, and provide a unique, retry-stable call_id for idempotent ingestion.

```python
from tuner_core import TunerConfig, TunerSession

session = TunerSession(
    config=TunerConfig.from_env(
        asr_model="nova-2",
        llm_model="gpt-4o-mini",
        tts_model="sonic-2",
    ),
    call_id=call_id,  # your unique, retry-stable identifier for this call
)
```

--------------------------------

### Get Agent

Source: https://docs.usetuner.ai/llms.txt

Retrieves details for a specific agent using its unique Tuner ID. This is useful for fetching agent configurations or status.

```APIDOC
## GET /agents/{agent_id}

### Description
Get a single agent by its Tuner ID.

### Method
GET

### Endpoint
/agents/{agent_id}

### Parameters
#### Path Parameters
- **agent_id** (string) - Required - The unique Tuner ID of the agent to retrieve.
```

--------------------------------

### Set Call Type Label

Source: https://docs.usetuner.ai/docs/api-and-integrations/connecting-to-pipecat

Override the default call type label for an Observer. Use this to categorize calls, for example, as 'web_call' or 'phone_call'.

```python
Observer(..., call_type="web_call")
Observer(..., call_type="phone_call")
```

--------------------------------

### Building Tuner Transcript Segments

Source: https://docs.usetuner.ai/docs/api-and-integrations/libraries/tuner-langchain

Convert recorded brain layer events into Tuner transcript rows. Timestamps are converted to milliseconds since call start.

```python
from tuner_langchain import segments_from_invocation

# Assuming 'invocations' is the output from handler.get_invocations()
segments = segments_from_invocation(invocations)

```
Tokens:
Raw