Pipecat — Testing & Evals¶

uv add --dev pytest pytest-asyncio aiohttp

# pytest.ini
[pytest]
asyncio_mode = auto

from pipecat.frames.frames import TextFrame, TTSSpeakFrame
from pipecat.processors.frame_processor import FrameProcessor
from pipecat.tests.utils import run_test


class Redactor(FrameProcessor):
    async def process_frame(self, frame, direction):
        await super().process_frame(frame, direction)
        if isinstance(frame, TextFrame):
            frame = TextFrame(text="".join("*" if c.isdigit() else c for c in frame.text))
        await self.push_frame(frame, direction)


async def test_redactor_masks_digits():
    down, up = await run_test(
        Redactor(),
        frames_to_send=[TextFrame("card 4111 1111"), TTSSpeakFrame("ok")],
        expected_down_frames=[TextFrame, TTSSpeakFrame],        # types, in order
    )
    assert down[0].text == "card **** ****"

received DOWN frames = [TextFrame, ]
expected DOWN frames = [TextFrame, TextFrame, ]
AssertionError

import json
import time

import pytest
from aiohttp import web


def sse(delta, finish=None) -> bytes:
    payload = {
        "id": "c", "object": "chat.completion.chunk", "created": int(time.time()), "model": "fake",
        "choices": [{"index": 0, "delta": delta, "finish_reason": finish}],
    }
    return f"data: {json.dumps(payload)}\n\n".encode()


@pytest.fixture
async def fake_llm():
    """A local OpenAI-compatible server: first request -> tool call, request after a tool result -> text."""
    requests: list[dict] = []

    async def completions(request: web.Request) -> web.StreamResponse:
        body = await request.json()
        requests.append(body)
        resp = web.StreamResponse(headers={"Content-Type": "text/event-stream"})
        await resp.prepare(request)
        if body["messages"][-1]["role"] == "tool":
            for part in ("Your order ", "ord-42 ", "has shipped."):
                await resp.write(sse({"role": "assistant", "content": part}))
            await resp.write(sse({}, "stop"))
        else:
            call = {"index": 0, "id": "call_1", "type": "function",
                    "function": {"name": "lookup_order", "arguments": json.dumps({"order_id": "ord-42"})}}
            await resp.write(sse({"role": "assistant", "tool_calls": [call]}))
            await resp.write(sse({}, "tool_calls"))
        await resp.write(b"data: [DONE]\n\n")
        return resp

    app = web.Application()
    app.add_routes([web.post("/v1/chat/completions", completions)])
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, "127.0.0.1", 0)
    await site.start()
    port = site._server.sockets[0].getsockname()[1]
    yield f"http://127.0.0.1:{port}/v1", requests
    await runner.cleanup()

from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.tests.utils import SleepFrame, run_test


async def test_tool_loop_end_to_end(fake_llm):
    base_url, requests = fake_llm
    calls = []

    async def lookup_order(params: FunctionCallParams, order_id: str):
        """Look up an order's status.

        Args:
            order_id: The order number.
        """
        calls.append(order_id)
        await params.result_callback({"order_id": order_id, "status": "shipped"})

    llm = OpenAILLMService(
        api_key="sk-test",
        base_url=base_url,
        settings=OpenAILLMService.Settings(model="fake", system_instruction="You are a support agent."),
    )
    context = LLMContext(messages=[{"role": "user", "content": "Where is ord-42?"}], tools=[lookup_order])
    pair = LLMContextAggregatorPair(context)

    await run_test(
        Pipeline([pair.user(), llm, pair.assistant()]),
        frames_to_send=[LLMRunFrame(), SleepFrame(sleep=1.0)],      # give the 2nd inference time to finish
    )

    assert calls == ["ord-42"]                                       # the tool ran with the model's arguments
    assert len(requests) == 2                                        # tool call, then the answer
    assert requests[0]["messages"][0] == {"role": "system", "content": "You are a support agent."}
    assert [m["role"] for m in requests[1]["messages"]] == ["system", "user", "assistant", "tool"]
    assert [m["role"] for m in context.messages] == ["user", "assistant", "tool", "assistant"]
    assert context.messages[-1]["content"] == "Your order ord-42 has shipped."

from pipecat.frames.frames import (
    BotStartedSpeakingFrame, BotStoppedSpeakingFrame, UserStoppedSpeakingFrame,
    VADUserStartedSpeakingFrame, VADUserStoppedSpeakingFrame,
)
from pipecat.observers.user_bot_latency_observer import UserBotLatencyObserver
from pipecat.pipeline.worker import PipelineParams

obs = UserBotLatencyObserver()
measured = []


@obs.event_handler("on_latency_measured")
async def on_measured(observer, latency):
    measured.append(latency)


await run_test(
    Passthrough(),                                   # any processor that pushes every frame
    frames_to_send=[
        VADUserStartedSpeakingFrame(), VADUserStoppedSpeakingFrame(), UserStoppedSpeakingFrame(),
        SleepFrame(sleep=0.4), BotStartedSpeakingFrame(), SleepFrame(sleep=0.2), BotStoppedSpeakingFrame(),
    ],
    observers=[obs],
    pipeline_params=PipelineParams(enable_metrics=True),
)
assert 0.3 < measured[0] < 0.6                       # 0.40 s in the run

from pipecat.evals.transport import EvalTransportParams
from pipecat.transports.base_transport import TransportParams

transport_params = {
    "eval": lambda: EvalTransportParams(audio_in_enabled=True, audio_out_enabled=True),
    "webrtc": lambda: TransportParams(audio_in_enabled=True, audio_out_enabled=True),
}

# capital.yaml
name: capital_question

scenarios:
  - name: berlin
    turns:
      - user: "What is the capital of Germany?"
        expect:
          - event: response
            text_contains: "Berlin"
            text_excludes: "Paris"

  - name: wrong_expectation
    turns:
      - user: "What is the capital of Germany?"
        expect:
          - event: response
            text_contains: "Munich"
            within_ms: 3000

python bot.py                                   # terminal 1
pipecat eval run capital.yaml -v --timeout 3    # terminal 2 (-t/--timeout: default per-expectation seconds)

      turn 0 → "What is the capital of Germany?"
        ✓ llm_response — "The capital of Germany is Berlin. "
  ✓ ws://localhost:7862/ws capital_question/berlin (164ms)
      turn 0 → "What is the capital of Germany?"
        ✗ llm_response — not satisfied within 3000ms: does not contain 'Munich'
  ✗ ws://localhost:7862/ws capital_question/wrong_expectation 0/1 turns (3047ms)

  Failures (1 of 2):

  ✗ ws://localhost:7862/ws capital_question/wrong_expectation 0/1 turns passed
      • turn 0 llm_response text_mismatch — not satisfied within 3000ms: does not contain 'Munich'

  1/2 passed, 1 failed  ·  3.2s

name: refund_request

scenarios:
  - name: approved_refund
    persona: |
      Sam, asking for a refund on a damaged blender. Has the order number
      ready and answers one question at a time.
    goal: "Get a refund for the damaged blender."
    success: "the agent started a refund for the blender"
    metrics:
      - name: empathy
        criterion: "the reply acknowledges the customer's frustration"
        min_score: 1

import pytest

from pipecat.evals.scenario import EvalScenarioFile
from pipecat.evals.session import EvalSession

BOT_URL = "ws://localhost:7862/ws"


async def test_berlin():
    scenario = EvalScenarioFile.load("capital.yaml")["capital_question/berlin"]
    result = await EvalSession.from_scenario(scenario, BOT_URL).run()
    assert result.passed, "\n".join(str(f) for f in result.failures)


async def test_wrong_expectation_is_reported():
    scenario = EvalScenarioFile.load("capital.yaml")["capital_question/wrong_expectation"]
    result = await EvalSession.from_scenario(scenario, BOT_URL).run()
    assert not result.passed