Usa ThunderPhone desde Pipecat

Ejecuta un agente de voz de ThunderPhone como el servicio de voz a voz dentro de una canalización de Pipecat, con tu propio transporte y telefonía.

Pipecat es un framework de código abierto para crear agentes de voz a partir de servicios componibles. ThunderPhone se integra como un servicio LLM de voz a voz: Pipecat envía el audio de quien llama, ThunderPhone devuelve la voz del agente, las transcripciones y las llamadas a funciones, y el transporte de Pipecat (Daily, LiveKit, Twilio, WebRTC o un micrófono local) transporta el audio. El reconocimiento de voz, el modelo de lenguaje, la voz, la gestión de turnos, 47 idiomas y las herramientas se ejecutan en ThunderPhone.

Las llamadas realizadas de esta forma aparecen en el historial de llamadas y se facturan según la tarifa por minuto de tu producto, como cualquier otra llamada en tiempo real. No hay suscripción ni se requiere un número de teléfono de ThunderPhone.

Instalar

pip install pipecat-thunderphone
export THUNDERPHONE_API_KEY=sk_live_...   # a secret API key

El paquete encapsula el servicio OpenAI Realtime de Pipecat, porque el WebSocket Realtime de ThunderPhone utiliza el mismo protocolo. Requiere pipecat-ai 1.8 o una versión posterior.

Ejecutar un agente guardado

Todo lo que hace el agente (prompt, voz, motor, idiomas, herramientas, saludo y verificaciones de silencio) se configura en ThunderPhone. El pipeline solo transporta audio.

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
from pipecat_thunderphone import ThunderPhoneRealtimeLLMService
 
llm = ThunderPhoneRealtimeLLMService(agent_id=12)
 
context = LLMContext()
aggregators = LLMContextAggregatorPair(context)
pipeline = Pipeline([
    transport.input(),
    aggregators.user(),
    llm,
    aggregators.assistant(),
    transport.output(),
])
worker = PipelineWorker(pipeline, params=PipelineParams(allow_interruptions=True))

Un agente guardado inicia la llamada según su propia programación, por lo que el servicio no solicita una respuesta inicial a Pipecat. Pasa greet_on_connect=True si aun así quieres una.

Configurar la sesión en línea

Sin agent_id, las instrucciones y las herramientas provienen del contexto de Pipecat, igual que con OpenAI. product selecciona el motor y voice selecciona la voz de ThunderPhone.

from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.services.llm_service import FunctionCallParams
 
llm = ThunderPhoneRealtimeLLMService(product="bolt", voice="olivia", language="es")
 
async def check_availability(params: FunctionCallParams):
    slots = await calendar.free_slots(params.arguments["date"])
    await params.result_callback({"slots": slots})
 
llm.register_function("check_availability", check_availability)
 
context = LLMContext(
    messages=[{"role": "system", "content": "You are Acme Dental's receptionist."}],
    tools=ToolsSchema(standard_tools=[
        FunctionSchema(
            name="check_availability",
            description="Free appointment slots on a date",
            properties={"date": {"type": "string"}},
            required=["date"],
        )
    ]),
)

Las sesiones en línea son controladas por el cliente: el agente habla primero porque Pipecat solicita una respuesta cuando llega el contexto y permanece en silencio durante los períodos de silencio, salvo que agregues un mensaje o solicites otra respuesta.

Opciones

Argumento	Significado
`api_key`	Clave secreta (`sk_live_...`). De forma predeterminada, `THUNDERPHONE_API_KEY`.
`agent_id`	Ejecuta un agente guardado. Es mutuamente excluyente con `product` y `voice`.
`product`	Motor para sesiones en línea: `spark`, `bolt` o `storm`.
`voice`	Nombre de voz de ThunderPhone para sesiones en línea.
`language`	Sugerencia de idioma principal para sesiones en línea; por ejemplo, `es`.
`from_number`, `to_number`	Números que se registrarán en la llamada cuando la plataforma se conecte a una línea telefónica.
`live_transcripts`	Transmite fragmentos de la transcripción de quien llama durante la enunciación (se cobra un cargo adicional).
`greet_on_connect`	Solicita una primera respuesta en cuanto la sesión esté lista.
`end_task_on_call_ended`	Envía `EndWorkerFrame` cuando ThunderPhone finaliza la llamada (activado de forma predeterminada).

Eventos

ThunderPhone agrega eventos de plataforma al protocolo en tiempo real. El servicio los entrega a los controladores:

@llm.event_handler("on_call_ended")
async def on_call_ended(service, event):
    print("call ended:", event["reason"], "call id:", service.call_id)

service.call_id se establece cuando la sesión está activa. Úsalo con GET /v1/calls/{call_id} para obtener la grabación, la transcripción y la calificación después de la llamada.

Límites

La detección de turnos se realiza del lado del servidor y siempre está activada; los turnos controlados por Pipecat (turn_detection=False) no son compatibles.
Las funciones registradas en el servicio se ejecutan solo para sesiones en línea. Un agente guardado ejecuta sus propias herramientas en ThunderPhone.
El audio es PCM mono de 16 bits a 24 kHz en ambas direcciones.
Los fotogramas de video se ignoran.

Usa ThunderPhone desde Pipecat

Instalar#

Ejecutar un agente guardado#

Configurar la sesión en línea#

Opciones#

Eventos#

Límites#