Voice Concierge (Pipecat + ElevenLabs)

pip install maximem-synap maximem-synap-pipecat pipecat-ai \
  pipecat-ai[deepgram,openai,elevenlabs,silero]

# .env
SYNAP_API_KEY=...
SYNAP_SERVER_URL=<maximem-server>
OPENAI_API_KEY=...
DEEPGRAM_API_KEY=...
ELEVENLABS_API_KEY=...
ELEVENLABS_VOICE_ID=...

import os
import asyncio
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.services.openai import OpenAILLMService
from pipecat.services.deepgram import DeepgramSTTService
from pipecat.services.elevenlabs import ElevenLabsTTSService
from pipecat.transports.services.daily import DailyTransport
from pipecat.vad.silero import SileroVADAnalyzer

from maximem_synap import MaximemSynapSDK
from synap_pipecat import SynapContextHook, SynapMemoryHook

CUSTOMER_ID = "your-business"

SYSTEM = """You are a phone concierge for <Business>.

- Use what you remember about the caller from prior calls — recent issues, preferences, ongoing situations.
- Voice rule: keep replies to 1–2 short sentences. Long replies feel wrong on the phone.
- If you need clarification, ask one focused question, not three.
- If the caller asks for a human, transfer immediately (transfer_to_agent tool)."""

async def run_call(caller_phone: str, room_url: str):
    sdk = MaximemSynapSDK()
    await sdk.initialize()

    transport = DailyTransport(
        room_url, None, "Concierge",
        params={"audio_in_enabled": True, "audio_out_enabled": True,
                "vad_analyzer": SileroVADAnalyzer()},
    )
    stt = DeepgramSTTService(api_key=os.environ["DEEPGRAM_API_KEY"])
    llm = OpenAILLMService(api_key=os.environ["OPENAI_API_KEY"], model="gpt-4o")
    tts = ElevenLabsTTSService(
        api_key=os.environ["ELEVENLABS_API_KEY"],
        voice_id=os.environ["ELEVENLABS_VOICE_ID"],
        model="eleven_turbo_v2_5",
    )

    context_hook = SynapContextHook(
        sdk=sdk,
        user_id=caller_phone,
        customer_id=CUSTOMER_ID,
        mode="fast",        # latency-critical
        max_results=5,
        system_template=(SYSTEM + "\n\nWhat we know about this caller:\n{context}"),
    )

    memory_hook = SynapMemoryHook(
        sdk=sdk,
        user_id=caller_phone,
        customer_id=CUSTOMER_ID,
        conversation_id=caller_phone,
        document_type="ai-chat-conversation",
        metadata={"channel": "voice"},
    )

    pipeline = Pipeline([
        transport.input(),
        stt,
        context_hook,    # injects caller memory into the LLM context
        llm,
        memory_hook,     # ingests the turn after the LLM responds
        tts,
        transport.output(),
    ])

    task = PipelineTask(pipeline)
    runner = PipelineRunner()
    await runner.run(task)

async def transfer_to_agent(call_id: str, queue: str) -> dict:
    # Use Twilio / your telephony provider to bridge the call to a queue.
    return await telephony.transfer(call_id, queue)

async def send_sms_followup(phone: str, body: str) -> dict:
    return await sms.send(phone, body)

# Register with the OpenAILLMService...