Pipecat-integratie

In deze handleiding lees je hoe je Pipecat gebruikt als LLM-pipeline in een Speech Engine-brainserver. Speech Engine verwerkt de spraakcyclus — spraak-naar-tekst, beurtwisseling en tekst-naar-spraak — terwijl Pipecat tekst genereert via een samenstelbare pipeline van processors (LLM-aanroepen, RAG, functieaanroepen, guardrails, contentfilters).

Architectuur

De Speech Engine SDK vormt de buitenste laag — de callback on_transcript wordt uitgevoerd telkens wanneer de gebruiker is uitgesproken. In de callback bouw je een Pipecat-pipeline, voer je de conversatiegeschiedenis in als een LLMContextFrame en stream je de tekstoutput van de pipeline terug naar Speech Engine. ElevenLabs zet de tekst om in spraak en speelt die af voor de gebruiker.

De Pipecat-pipeline wordt alleen uitgevoerd voor de duur van één beurt. Wanneer een nieuw transcript binnenkomt, wordt de vorige pipeline geannuleerd voordat de volgende wordt uitgevoerd — zo wordt de onderbrekingsafhandeling van Speech Engine doorgegeven aan de pipeline.

Wanneer je dit patroon gebruikt

Pipecat is ideaal wanneer je brein meer nodig heeft dan één LLM-aanroep:

• Samenstelbare processors voor retrieval-augmented generation, functieaanroepen of guardrails
• Framegebaseerde middleware die verkeer bij elke stap kan inspecteren, transformeren of blokkeren
• Herbruikbare pipelinefragmenten die door meerdere agents worden gedeeld

Als je brein bestaat uit “transcript erin, LLM-aanroep eruit”, is de Snelstartgids voor Speech Engine eenvoudiger. Kies Pipecat wanneer de pipeline zelf het interessante onderdeel is.

Vereisten

• Een Speech Engine. Volg de Snelstartgids voor Speech Engine om er een te maken.
• Python 3.10+ (vereist door pipecat-ai).
• Publieke HTTPS-tunnel voor de brainserver (bijv. ngrok).

Afhankelijkheden installeren

pip install "pipecat-ai[openai]" "elevenlabs" "python-dotenv"

pipecat-ai[openai] installeert de OpenAI LLM-service. Vervang de extra door die van een andere provider (anthropic, google, enzovoort) als je dat liever wilt.

Het Pipecat-brein bouwen

Het brein bestaat uit twee onderdelen: een TextSink-processor die gestreamde tekst naar een asyncio.Queue doorstuurt, en een coroutine run_pipecat_brain die een pipeline voor één beurt bouwt en chunks als async iterator oplevert.

import asyncio
import os
from typing import AsyncIterator
from dotenv import load_dotenv
from pipecat.frames.frames import (
    Frame,
    LLMContextFrame,
    LLMFullResponseEndFrame,
    LLMTextFrame,
)
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.openai.llm import OpenAILLMService
load_dotenv()
SYSTEM_PROMPT = (
    "You are a helpful voice assistant. Keep responses concise and conversational."
)
class TextSink(FrameProcessor):
    """Drain LLMTextFrame text into an asyncio.Queue."""
    def __init__(self, queue: asyncio.Queue):
        super().__init__()
        self._queue = queue
    async def process_frame(self, frame: Frame, direction: FrameDirection):
        await super().process_frame(frame, direction)
        if isinstance(frame, LLMTextFrame):
            await self._queue.put(frame.text)
        elif isinstance(frame, LLMFullResponseEndFrame):
            await self._queue.put(None)  # sentinel
        await self.push_frame(frame, direction)
def build_messages(transcript: list[dict]) -> list[dict]:
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    for turn in transcript:
        role = "assistant" if turn["role"] == "agent" else turn["role"]
        messages.append({"role": role, "content": turn["content"]})
    return messages
async def run_pipecat_brain(transcript: list[dict]) -> AsyncIterator[str]:
    """Yield response text chunks from a one-turn Pipecat pipeline."""
    llm = OpenAILLMService(
        api_key=os.environ["OPENAI_API_KEY"],
        model="gpt-4o-mini",
    )
    queue: asyncio.Queue[str | None] = asyncio.Queue()
    sink = TextSink(queue)
    task = PipelineTask(Pipeline([llm, sink]))
    runner = PipelineRunner(handle_sigint=False)
    async def drive():
        context = LLMContext(build_messages(transcript))
        await task.queue_frame(LLMContextFrame(context))
        await task.stop_when_done()
    run_task = asyncio.create_task(runner.run(task))
    drive_task = asyncio.create_task(drive())
    try:
        while True:
            chunk = await queue.get()
            if chunk is None:
                break
            yield chunk
    finally:
        await task.cancel()
        await asyncio.gather(run_task, drive_task, return_exceptions=True)

De pipeline bevat alleen de LLM-service en de sink — geen STT- of TTS-processors, omdat Speech Engine die verwerkt. LLMContextFrame is de input; LLMTextFrame-chunks zijn de output.

run_pipecat_brain is een async generator. Elke opgeleverde chunk gaat rechtstreeks naar Speech Engine, zodat de agent begint te spreken voordat het volledige antwoord klaar is.

Koppel het aan de Speech Engine-server

De functie send_response van de Speech Engine SDK accepteert een string of een willekeurige async iterable van strings, dus je kunt run_pipecat_brain(transcript) rechtstreeks doorgeven. Zet de ConversationMessage-objecten die Speech Engine verstrekt om naar gewone dicts voordat je ze aan het brein doorgeeft.

import asyncio
import os
from dotenv import load_dotenv
from elevenlabs import AsyncElevenLabs
from brain import run_pipecat_brain
load_dotenv()
elevenlabs = AsyncElevenLabs(api_key=os.environ["ELEVENLABS_API_KEY"])
SPEECH_ENGINE_ID = os.environ["SPEECH_ENGINE_ID"]
async def on_transcript(transcript, session):
    history = [{"role": m.role, "content": m.content} for m in transcript]
    await session.send_response(run_pipecat_brain(history))
async def main():
    engine = await elevenlabs.speech_engine.get(SPEECH_ENGINE_ID)
    await engine.serve(
        port=3001,
        path="/ws",
        debug=True,
        on_transcript=on_transcript,
    )
if __name__ == "__main__":
    asyncio.run(main())

De Speech Engine SDK annuleert de taak van de vorige beurt wanneer een nieuw transcript binnenkomt. Daarmee worden de async generator en de onderliggende PipelineTask geannuleerd via het blok try/finally in run_pipecat_brain.

De server uitvoeren

ngrok http 3001
python server.py

Maak vanuit een browser verbinding met Speech Engine met hetzelfde tokenendpoint en dezelfde clientcode als in de snelstartgids. De Pipecat-pipeline draait aan de serverkant; de browser ziet een gewone Speech Engine-conversatie.

De pipeline uitbreiden

Een Pipecat-pipeline met alleen tekst kan elke frameprocessor bevatten die werkt met LLMTextFrame of LLMContextFrame. Enkele veelvoorkomende toevoegingen:

• Guardrails: een FrameProcessor vóór de LLM die LLMContextFrame inspecteert en onveilige context vervangt of blokkeert.
• Functieaanroepen: registreer tools bij de OpenAILLMService en Pipecat verwerkt tool-call-frames standaard. De uiteindelijke assistenttekst komt nog steeds binnen als LLMTextFrame.
• Redeneren in meerdere fasen: koppel twee OpenAILLMService-instanties, met daartussen een aangepaste processor die de context herschrijft voor de tweede doorgang.
• Outputfiltering: een FrameProcessor na de LLM die elke LLMTextFrame inspecteert en niet-toegestane content verwijdert of herschrijft voordat die TextSink bereikt.

De vorm van de pipeline blijft hetzelfde — Pipeline([processor_a, llm, processor_b, sink]) — en run_pipecat_brain verandert niet.

Overwegingen voor productie

• Veilig annuleren: PipelineTask.cancel() kan vastlopen als deze wordt aangeroepen voordat de pipeline volledig is gestart (pipecat-ai/pipecat#4276). Het bovenstaande patroon met try/finally is veilig, omdat cancel() pas wordt uitgevoerd nadat er minstens één frame in de wachtrij is geplaatst.
• Promptinjectie: spraak-naar-tekst-output is gebruikersinvoer. Valideer of normaliseer het transcript voordat je het aan de LLM doorgeeft, vooral als een downstreamprocessor de tekst gebruikt in toolaanroepen of databasequeries.
• Authenticatie voor de brainserver: stel een gedeeld geheim in voor Speech Engine en controleer dit in de brainserver om ongeautoriseerde verbindingen met je /ws-endpoint te voorkomen:

  await elevenlabs.speech_engine.update(
      speech_engine_id=SPEECH_ENGINE_ID,
      speech_engine={"request_headers": {"x-api-key": os.environ["SHARED_SECRET"]}},
  )
• LLM-provider: pipecat-ai[openai] bevat OpenAILLMService. Installeer voor Anthropic pipecat-ai[anthropic] en gebruik AnthropicLLMService; de rest van de pipeline blijft ongewijzigd.

Volgende stappen