Pipecat-integrering

Den här guiden visar hur du använder Pipecat som LLM-pipeline i en Speech Engine-hjärnserver. Speech Engine hanterar röstflödet — tal-till-text, turordning och text-till-tal — medan Pipecat hanterar textgenerering via en sammansättningsbar pipeline med processorer (LLM-anrop, RAG, funktionsanrop, skyddsräcken och innehållsfilter).

Arkitektur

Speech Engine SDK fungerar som det yttre lagret — dess on_transcript-callback aktiveras varje gång användaren har talat klart. I callbacken bygger du en Pipecat-pipeline, matar in konversationshistoriken som en LLMContextFrame och streamar pipelinens textutdata tillbaka till Speech Engine. ElevenLabs omvandlar texten till tal och spelar upp den för användaren.

Pipecat-pipelinen körs endast under en tur. När ett nytt transkript kommer avbryts den föregående pipelinen innan nästa körs — så sprids Speech Engines hantering av avbrott till pipelinen.

När du bör använda det här mönstret

Pipecat är utmärkt när din hjärna behöver mer än ett enda LLM-anrop:

• Sammansättningsbara processorer för retrieval-augmented generation, funktionsanrop eller skyddsräcken
• Rambaserad mellanprogramvara som kan granska, omvandla eller blockera trafik i varje steg
• Återanvändbara pipeline-fragment som delas mellan flera agenter

Om din hjärna är ”transkript in, LLM-anrop ut” är Speech Engine-snabbstarten enklare. Välj Pipecat när själva pipelinen är den intressanta delen.

Förutsättningar

• En Speech Engine. Följ Speech Engine-snabbstarten för att skapa en.
• Python 3.10+ (krävs av pipecat-ai).
• Offentlig HTTPS-tunnel för hjärnservern (t.ex. ngrok).

Installera beroenden

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

pipecat-ai[openai] installerar OpenAI LLM-tjänsten. Byt ut tillägget mot en annan leverantör (anthropic, google osv.) om du föredrar det.

Bygg Pipecat-hjärnan

Hjärnan har två delar: en TextSink-processor som tömmer streamad text till en asyncio.Queue, och korutinen run_pipecat_brain som bygger en pipeline för en tur och ger delar som en asynkron iterator.

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)

Pipelinen innehåller endast LLM-tjänsten och sinken — inga STT- eller TTS-processorer, eftersom Speech Engine hanterar dem. LLMContextFrame är indata och LLMTextFrame-delar är utdata.

run_pipecat_brain är en asynkron generator. Varje del som skickas vidare går direkt till Speech Engine, så agenten börjar tala innan hela svaret är klart.

Koppla den till Speech Engine-servern

Speech Engine SDK:s send_response accepterar en sträng eller valfri asynkron itererbar samling av strängar, så du kan skicka run_pipecat_brain(transcript) direkt. Konvertera ConversationMessage-objekten som Speech Engine tillhandahåller till vanliga dict-objekt innan du skickar dem till hjärnan.

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())

Speech Engine SDK avbryter den föregående turens uppgift när ett nytt transkript kommer, vilket avbryter den asynkrona generatorn och den underliggande PipelineTask via try/finally-blocket i run_pipecat_brain.

Kör servern

ngrok http 3001
python server.py

Anslut till Speech Engine från en webbläsare med samma tokenändpunkt och klientkod som visas i snabbstarten. Pipecat-pipelinen körs på serversidan; webbläsaren ser en vanlig Speech Engine-konversation.

Utöka pipelinen

En Pipecat-pipeline med enbart text kan innehålla valfri ramprocessor som arbetar med LLMTextFrame eller LLMContextFrame. Några vanliga tillägg:

• Skyddsräcken: en FrameProcessor före LLM:en som granskar LLMContextFrame och ersätter eller blockerar osäkert sammanhang.
• Funktionsanrop: registrera verktyg på OpenAILLMService, så hanterar Pipecat verktygsanropsramar direkt. Den slutliga assistenttexten kommer fortfarande som LLMTextFrame.
• Resonemang i flera steg: kedja två OpenAILLMService-instanser, med en anpassad processor emellan som skriver om sammanhanget för den andra körningen.
• Utdatafiltrering: en FrameProcessor efter LLM:en som granskar varje LLMTextFrame och tar bort eller skriver om otillåtet innehåll innan det når TextSink.

Pipelineformen förblir densamma — Pipeline([processor_a, llm, processor_b, sink]) — och run_pipecat_brain ändras inte.

Att tänka på för produktion

• Säker avbrytning: PipelineTask.cancel() kan låsa sig om den anropas innan pipelinen har startat helt (pipecat-ai/pipecat#4276). Mönstret try/finally ovan är säkert eftersom cancel() körs först efter att minst en ram har köats.
• Promptinjektion: tal-till-text-utdata är användarinmatning. Validera eller normalisera transkriptet innan du matar det till LLM:en, särskilt om någon efterföljande processor använder texten i verktygsanrop eller databasfrågor.
• Autentisering för hjärnservern: ange en delad hemlighet i Speech Engine och kontrollera den i hjärnservern för att förhindra obehöriga anslutningar till din /ws-ändpunkt:

  await elevenlabs.speech_engine.update(
      speech_engine_id=SPEECH_ENGINE_ID,
      speech_engine={"request_headers": {"x-api-key": os.environ["SHARED_SECRET"]}},
  )
• LLM-leverantör: pipecat-ai[openai] innehåller OpenAILLMService. För Anthropic installerar du pipecat-ai[anthropic] och använder AnthropicLLMService; resten av pipelinen är oförändrad.

Nästa steg