Integração com Pipecat

Este guia mostra como usar o Pipecat como pipeline de LLM dentro de um servidor de cérebro do Speech Engine. O Speech Engine cuida do ciclo de voz — conversão de fala em texto, alternância de turnos e conversão de texto em fala — enquanto o Pipecat gera texto por meio de um pipeline combinável de processadores (chamadas de LLM, RAG, chamadas de função, guardrails, filtros de conteúdo).

Arquitetura

O SDK do Speech Engine funciona como a camada externa — o callback on_transcript é acionado sempre que o usuário termina de falar. Dentro do callback, você cria um pipeline do Pipecat, fornece o histórico da conversa como um LLMContextFrame e transmite a saída de texto do pipeline de volta para o Speech Engine. A ElevenLabs converte o texto em fala e o reproduz para o usuário.

O pipeline do Pipecat é executado apenas durante um turno. Quando uma nova transcrição chega, o pipeline anterior é cancelado antes que o próximo seja executado — é assim que o tratamento de interrupções do Speech Engine é propagado para o pipeline.

Quando usar este padrão

O Pipecat se destaca quando seu cérebro precisa de mais do que uma única chamada de LLM:

• Processadores combináveis para geração aumentada por recuperação, chamadas de função ou guardrails
• Middleware baseado em frames que pode inspecionar, transformar ou bloquear o tráfego em cada etapa
• Fragmentos de pipeline reutilizáveis e compartilhados entre vários agentes

Se o seu cérebro é “transcrição entra, chamada de LLM sai”, o guia de início rápido do Speech Engine é mais simples. Use o Pipecat quando o próprio pipeline for a parte mais interessante.

Pré-requisitos

• Um Speech Engine. Siga o guia de início rápido do Speech Engine para criar um.
• Python 3.10+ (exigido por pipecat-ai).
• Túnel HTTPS público para o servidor do cérebro (por exemplo, ngrok).

Instale as dependências

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

pipecat-ai[openai] inclui o serviço de LLM da OpenAI. Se preferir, troque o extra por outro provedor (anthropic, google etc.).

Crie o cérebro do Pipecat

O cérebro tem duas partes: um processador TextSink que envia o texto transmitido para uma asyncio.Queue e uma corrotina run_pipecat_brain que cria um pipeline de um turno e produz blocos como um iterador assíncrono.

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)

O pipeline contém apenas o serviço de LLM e o coletor — sem processadores de STT ou TTS, pois o Speech Engine cuida deles. LLMContextFrame é a entrada; os blocos de LLMTextFrame são a saída.

run_pipecat_brain é um gerador assíncrono. Cada bloco gerado vai direto para o Speech Engine, para que o agente comece a falar antes de a resposta completa ficar pronta.

Conecte-o ao servidor do Speech Engine

O send_response do SDK do Speech Engine aceita uma string ou qualquer iterável assíncrono de strings, então você pode passar run_pipecat_brain(transcript) diretamente. Converta os objetos ConversationMessage fornecidos pelo Speech Engine em dicionários simples antes de passá-los ao cérebro.

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

O SDK do Speech Engine cancela a tarefa do turno anterior quando uma nova transcrição chega, o que cancela o gerador assíncrono e a PipelineTask subjacente por meio do bloco try/finally em run_pipecat_brain.

Execute o servidor

ngrok http 3001
python server.py

Conecte-se ao Speech Engine a partir de um navegador usando o mesmo endpoint de token e código de cliente mostrados no guia de início rápido. O pipeline do Pipecat é executado no servidor; o navegador vê uma conversa normal do Speech Engine.

Estenda o pipeline

Um pipeline do Pipecat somente de texto pode incluir qualquer processador de frames que opere em LLMTextFrame ou LLMContextFrame. Algumas adições comuns:

• Guardrails: um FrameProcessor colocado antes do LLM, que inspeciona LLMContextFrame e substitui ou bloqueia contexto inseguro.
• Chamadas de função: registre ferramentas no OpenAILLMService, e o Pipecat processará nativamente os frames de chamada de ferramenta. O texto final do assistente ainda chega como LLMTextFrame.
• Raciocínio em vários estágios: encadeie duas instâncias de OpenAILLMService, com um processador personalizado entre elas que reescreve o contexto para a segunda passagem.
• Filtragem de saída: um FrameProcessor colocado depois do LLM, que inspeciona cada LLMTextFrame e descarta ou reescreve conteúdo não permitido antes que ele chegue ao TextSink.

A estrutura do pipeline permanece a mesma — Pipeline([processor_a, llm, processor_b, sink]) — e run_pipecat_brain não muda.

Considerações para produção

• Segurança no cancelamento: PipelineTask.cancel() pode causar deadlock se for chamado antes de o pipeline ser totalmente iniciado (pipecat-ai/pipecat#4276). O padrão try/finally acima é seguro porque cancel() é executado somente depois que pelo menos um frame foi enfileirado.
• Injeção de prompt: a saída da conversão de fala em texto é uma entrada do usuário. Valide ou normalize a transcrição antes de fornecê-la ao LLM, especialmente se algum processador posterior usar o texto em chamadas de ferramenta ou consultas ao banco de dados.
• Autenticação do servidor do cérebro: defina um segredo compartilhado no Speech Engine e verifique-o no servidor do cérebro para impedir conexões não autorizadas ao endpoint /ws:

  await elevenlabs.speech_engine.update(
      speech_engine_id=SPEECH_ENGINE_ID,
      speech_engine={"request_headers": {"x-api-key": os.environ["SHARED_SECRET"]}},
  )
• Provedor de LLM: pipecat-ai[openai] inclui OpenAILLMService. Para Anthropic, instale pipecat-ai[anthropic] e use AnthropicLLMService; o restante do pipeline não muda.

Próximas etapas