Roark

​

Overview

• Run automated simulations that call your agent over the Daily transport using lifelike personas and branching flows
• Score every call — simulated or live — against built-in metrics plus your own LLM-graded custom metrics
• Trace each turn (STT → LLM → TTS, tool calls, latency) via OpenTelemetry and inspect it on the call
• Monitor production traffic continuously and turn any real call into a repeatable regression test

​

How the integration works

1. Sync and capture with the observer. The pipecat-roark package adds a RoarkObserver to your pipeline. It registers (“syncs”) your agent to Roark the first time it runs, then streams every call’s transcript, tool calls, and a stereo recording to Roark. See the Roark analytics observer reference for full setup.
2. Simulate over the Daily transport. Once your Pipecat Cloud credentials are connected, Roark starts your agent on Pipecat Cloud, joins the Daily room as a simulated caller, drives your flow with a chosen persona, and collects the transcript, recording, traces, and metrics automatically — no code changes beyond the observer.

​

Getting started

pip install pipecat-roark

import os

from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat_roark import RoarkObserver

roark = RoarkObserver(
    api_key=os.environ["ROARK_API_KEY"],
    agent_id="support-bot-v3",
    agent_name="Support Bot v3",
    agent_prompt=SYSTEM_PROMPT,
)

pipeline = Pipeline([
    transport.input(), stt, context_aggregator.user(), llm, tts,
    transport.output(),
    roark.audio_processor,  # after transport.output() — L=user, R=bot
    context_aggregator.assistant(),
])

task = PipelineTask(pipeline, params=PipelineParams(observers=[roark]))

​

Simulations

​

Personas

• Voice: language and accent (US, British, Indian, and more), gender, and optional background noise (e.g. office)
• Speech: pace, clarity (clear, vague, rambling), and natural disfluencies like “um” and “uh”
• Behavior: base emotion (neutral, cheerful, confused, frustrated, skeptical, rushed), intent clarity, confirmation style, and memory reliability
• Context: an optional backstory and custom key-value properties (account number, membership status, and so on)

​

Flows

• Scripted — a visual graph of nodes and edges that lays out an exact path. Nodes cover customer turns, agent turns, silence, DTMF keypad input, and voicemail; edges let the conversation branch or loop back, so one flow can cover many outcomes.
• Improv — a plain-language brief the simulated caller improvises from, for open-ended conversations you don’t want to script turn by turn.

• Generate from your agent prompt — Roark drafts flows from a description of what your agent does
• Generate from a real call — turn a production conversation into a repeatable flow
• Build it yourself — lay out the graph (scripted) or write the brief (improv) in the visual editor

​

Metrics

• Built-in metrics are collected automatically with no configuration, spanning latency and response time, interruptions and overlap, sentiment and emotion, speech quality (DNSMOS), and compliance checks like PII handling and prompt-injection resistance.
• Custom metrics are defined in natural language and graded by Roark Prism, Roark’s model built for scoring voice conversations — for example identity verification (boolean), empathy (scale), call reason (classification), or upsell attempts (count).
• Thresholds turn any metric into a pass/fail outcome (e.g. Response Time < 1000ms), applied per call or aggregated across a run.

​

Tracing

import os
import uuid

from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat_roark import RoarkObserver


def setup_roark_tracing() -> None:
    """Must run BEFORE constructing PipelineTask."""
    resource = Resource.create({
        "deployment.environment": os.getenv("ENVIRONMENT", "production"),
        # Optional: tag spans for the trace explorer.
        "roark.project.tag.env": os.getenv("ENVIRONMENT", "production"),
    })
    provider = TracerProvider(resource=resource)
    provider.add_span_processor(
        BatchSpanProcessor(
            OTLPSpanExporter(
                endpoint="https://api.roark.ai/v1/traces",
                headers={"Authorization": f"Bearer {os.environ['ROARK_API_KEY']}"},
                timeout=30,
            )
        )
    )
    trace.set_tracer_provider(provider)


setup_roark_tracing()

# Generate ONE id and pass it to both sides so Roark can correlate.
call_id = str(uuid.uuid4())

roark = RoarkObserver(
    api_key=os.environ["ROARK_API_KEY"],
    agent_id="support-bot-v3",
    pipecat_call_id=call_id,  # appears on the Roark record as pipecatCallId
)

task = PipelineTask(
    pipeline,
    params=PipelineParams(observers=[roark]),
    enable_tracing=True,
    conversation_id=call_id,  # set as the conversation.id span attribute
)

​

Observability

​

Next steps