Pipecat

import os

from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.worker import PipelineParams, PipelineWorker
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
    LLMContextAggregatorPair,
    LLMUserAggregatorParams,
)
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
from pipecat.workers.runner import WorkerRunner

transport_params = {
    "webrtc": lambda: TransportParams(audio_in_enabled=True, audio_out_enabled=True),
    "daily": lambda: DailyParams(audio_in_enabled=True, audio_out_enabled=True),
    "twilio": lambda: FastAPIWebsocketParams(audio_in_enabled=True, audio_out_enabled=True),
}


async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
    stt = DeepgramSTTService(api_key=os.environ["DEEPGRAM_API_KEY"])
    tts = CartesiaTTSService(
        api_key=os.environ["CARTESIA_API_KEY"],
        settings=CartesiaTTSService.Settings(voice="86e30c1d-714b-4074-a1f2-1cb6b552fb49"),
    )
    llm = OpenAILLMService(
        api_key=os.environ["OPENAI_API_KEY"],
        settings=OpenAILLMService.Settings(
            system_instruction="You are DocsAssistant on a phone call. Answer in short plain sentences."
        ),
    )

    context = LLMContext()
    user_agg, assistant_agg = LLMContextAggregatorPair(
        context, user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer())
    )

    pipeline = Pipeline([
        transport.input(), stt, user_agg, llm, tts, transport.output(), assistant_agg,
    ])
    worker = PipelineWorker(pipeline, params=PipelineParams(enable_metrics=True))
    runner = WorkerRunner(handle_sigint=runner_args.handle_sigint)
    await runner.add_workers(worker)

    @transport.event_handler("on_client_connected")
    async def on_connected(transport, client):
        context.add_message({"role": "developer", "content": "Greet the caller and say you are an AI assistant."})
        await worker.queue_frames([LLMRunFrame()])

    @transport.event_handler("on_client_disconnected")
    async def on_disconnected(transport, client):
        await runner.cancel()

    await runner.run()


async def bot(runner_args: RunnerArguments):
    transport = await create_transport(runner_args, transport_params)
    await run_bot(transport, runner_args)


if __name__ == "__main__":
    from pipecat.runner.run import main
    main()

import json
import os

from fastapi import FastAPI, WebSocket

from pipecat.serializers.twilio import TwilioFrameSerializer
from pipecat.transports.websocket.fastapi import (
    FastAPIWebsocketParams,
    FastAPIWebsocketTransport,
)

app = FastAPI()


@app.websocket("/twilio/media")
async def media(websocket: WebSocket):
    await websocket.accept()
    # Twilio sends a "connected" message and then a "start" message carrying the stream and call SIDs.
    await websocket.receive_text()
    start = json.loads(await websocket.receive_text())["start"]

    serializer = TwilioFrameSerializer(
        stream_sid=start["streamSid"],
        call_sid=start["callSid"],
        account_sid=os.environ["TWILIO_ACCOUNT_SID"],
        auth_token=os.environ["TWILIO_AUTH_TOKEN"],
    )
    transport = FastAPIWebsocketTransport(
        websocket=websocket,
        params=FastAPIWebsocketParams(
            audio_in_enabled=True,
            audio_out_enabled=True,
            add_wav_header=False,
            serializer=serializer,
        ),
    )
    await run_bot(transport)      # builds the same pipeline as above, with 8 kHz audio

from langchain.agents import create_agent
from langchain_core.messages import AIMessageChunk
from langgraph.checkpoint.memory import InMemorySaver

from pipecat.frames.frames import (
    Frame,
    LLMContextFrame,
    LLMFullResponseEndFrame,
    LLMFullResponseStartFrame,
    TextFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor

from docsassistant.tools import search_docs      # the shared retrieval tool
from docsassistant.models import chat_model      # a LangChain chat model

agent = create_agent(
    chat_model,
    tools=[search_docs],
    system_prompt="You are DocsAssistant, speaking on a call. Reply in short plain sentences with no lists or markdown.",
    checkpointer=InMemorySaver(),
)


class DocsAgentProcessor(FrameProcessor):
    def __init__(self, thread_id: str):
        super().__init__()
        self._config = {"configurable": {"thread_id": thread_id}}

    async def process_frame(self, frame: Frame, direction: FrameDirection):
        await super().process_frame(frame, direction)
        if not isinstance(frame, LLMContextFrame):
            await self.push_frame(frame, direction)
            return

        messages = frame.context.get_messages()
        last = messages[-1] if messages else {}
        text = last.get("content", "") if isinstance(last, dict) else ""
        if not isinstance(text, str) or not text.strip():
            return

        await self.push_frame(LLMFullResponseStartFrame())
        async for chunk, meta in agent.astream(
            {"messages": [{"role": "user", "content": text}]},
            self._config,
            stream_mode="messages",
        ):
            if isinstance(chunk, AIMessageChunk) and chunk.content and not chunk.tool_call_chunks:
                out = TextFrame(chunk.content if isinstance(chunk.content, str) else "")
                out.includes_inter_frame_spaces = True
                await self.push_frame(out)
        await self.push_frame(LLMFullResponseEndFrame())

from pipecat.services.kokoro.tts import KokoroTTSService
from pipecat.services.ollama.llm import OLLamaLLMService
from pipecat.services.whisper.stt import Model, WhisperSTTService

stt = WhisperSTTService(
    device="cuda",                 # or "cpu"
    compute_type="int8_float16",   # "int8" on CPU
    settings=WhisperSTTService.Settings(model=Model.SMALL.value),
)

llm = OLLamaLLMService(
    base_url="http://ollama:11434/v1",
    settings=OLLamaLLMService.Settings(
        model="qwen3:8b",
        system_instruction="You are DocsAssistant on a call. Reply in short plain sentences.",
    ),
)

tts = KokoroTTSService(
    settings=KokoroTTSService.Settings(voice="af_heart"),
)
# then: Pipeline([transport.input(), stt, user_agg, llm, tts, transport.output(), assistant_agg])

from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.processors.aggregators.llm_response_universal import (
    LLMContextAggregatorPair,
    LLMUserAggregatorParams,
)
from pipecat.turns.user_stop import TurnAnalyzerUserTurnStopStrategy
from pipecat.turns.user_turn_strategies import UserTurnStrategies, default_user_turn_start_strategies

strategies = UserTurnStrategies(
    start=default_user_turn_start_strategies(),
    stop=[TurnAnalyzerUserTurnStopStrategy(turn_analyzer=LocalSmartTurnAnalyzerV3(cpu_count=2))],
)

user_agg, assistant_agg = LLMContextAggregatorPair(
    context,
    user_params=LLMUserAggregatorParams(
        vad_analyzer=SileroVADAnalyzer(),
        user_turn_strategies=strategies,
    ),
)

from pipecat.flows import FlowManager, NodeConfig


def ask_node() -> NodeConfig:
    return NodeConfig(
        name="ask",
        role_message="You are DocsAssistant on a phone call. Use short plain sentences with no markdown.",
        task_messages=[{
            "role": "developer",
            "content": "Ask which product the caller needs help with, then call select_product with the answer.",
        }],
        functions=[select_product],
    )


async def select_product(flow_manager: FlowManager, product: str) -> tuple[str, NodeConfig]:
    """Record the product the caller is asking about.

    Args:
        product: The product name the caller mentioned.
    """
    flow_manager.state["product"] = product
    return product, answer_node(product)


def answer_node(product: str) -> NodeConfig:
    return NodeConfig(
        name="answer",
        task_messages=[{
            "role": "developer",
            "content": f"Answer questions about {product}. When the caller is done, say goodbye.",
        }],
        functions=[],
        post_actions=[{"type": "end_conversation"}],
    )

# in run_bot, after building the pipeline and worker:
flow_manager = FlowManager(worker=worker, llm=llm, context_aggregator=context_aggregator, transport=transport)
# in on_client_connected:   await flow_manager.initialize(ask_node())

FROM python:3.12-slim

RUN apt-get update && apt-get install -y --no-install-recommends espeak-ng libsndfile1 \
    && rm -rf /var/lib/apt/lists/*
WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt     # includes pipecat-ai[whisper,kokoro,silero,local-smart-turn,webrtc,runner]

COPY bot.py ./
COPY docsassistant ./docsassistant
ENV HF_HOME=/models
EXPOSE 7860
CMD ["python", "bot.py", "-t", "webrtc", "--host", "0.0.0.0", "--port", "7860"]