> This is a page from the ElevenLabs documentation. For a complete page index, fetch https://elevenlabs.io/docs/llms.txt. For the full documentation in a single file, fetch https://elevenlabs.io/docs/llms-full.txt. # Pipecat-integrering Den här guiden visar hur du använder [Pipecat](https://www.pipecat.ai/) 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). > **Note** > > Den här guiden gäller endast Python eftersom Pipecat är ett Python-ramverk på serversidan. Det finns ingen > Node-motsvarighet för pipeline-processorerna. Paketet `pipecat-client-js` finns, men är en > webbläsarklient som kommunicerar med en Pipecat-server, inte ett sätt att bygga pipelines i TypeScript. ## 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. ```mermaid sequenceDiagram participant Browser participant EL as ElevenLabs participant Brain as Brain Server (engine.serve) participant PC as Pipecat Pipeline Browser->>EL: User speaks (audio) EL->>Brain: on_transcript(history) Brain->>PC: LLMContextFrame(history) PC-->>Brain: LLMTextFrame chunks Brain-->>EL: send_response(async iterator) EL->>Browser: Agent speaks (audio) ``` 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](/docs/sv/eleven-api/guides/cookbooks/speech-engine) enklare. Välj Pipecat när själva pipelinen är den intressanta delen. ## Förutsättningar * En Speech Engine. Följ [Speech Engine-snabbstarten](/docs/sv/eleven-api/guides/cookbooks/speech-engine) för att skapa en. * Python 3.10+ (krävs av `pipecat-ai`). * Offentlig HTTPS-tunnel för hjärnservern (t.ex. [ngrok](https://ngrok.com)). ## Installera beroenden ```bash 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. **`brain.py`** ```python title="brain.py" maxLines=0 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. **`server.py`** ```python title="server.py" maxLines=0 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 ```bash ngrok http 3001 python server.py ``` Anslut till Speech Engine från en webbläsare med samma [tokenändpunkt och klientkod](/docs/sv/eleven-api/guides/cookbooks/speech-engine#client-setup) 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](https://github.com/pipecat-ai/pipecat/issues/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: ```python 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 #### [Speech Engine-snabbstart](/docs/sv/eleven-api/guides/cookbooks/speech-engine) Bygg hjärnservern och webbläsarklienten från början till slut. #### [LiveKit-integrering](/docs/sv/eleven-api/guides/how-to/speech-engine/livekit-integration) Använd Speech Engine som röstlager för ett LiveKit-rum. #### [Python SDK-referens](/docs/sv/eleven-api/resources/libraries/speech-engine/python-sdk-reference) Klasser, metoder och händelser för Speech Engine Python SDK. #### [JavaScript SDK-referens](/docs/sv/eleven-api/resources/libraries/speech-engine/javascript-sdk-reference) Klasser, metoder och händelser för Speech Engine JavaScript SDK.