Pipecat Plugin

pip install deepslate-pipecat

import asyncio
import os

import aiohttp
from dotenv import load_dotenv

from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.transports.daily.transport import DailyParams, DailyTransport

from deepslate.pipecat import DeepslateOptions, DeepslateRealtimeLLMService, ElevenLabsTtsConfig

load_dotenv()

async def main():
    async with aiohttp.ClientSession() as session:
        room_name = os.environ["DAILY_ROOM_URL"].split("/")[-1]
        async with session.post(
            "https://api.daily.co/v1/meeting-tokens",
            headers={"Authorization": f"Bearer {os.environ['DAILY_API_KEY']}"},
            json={"properties": {"room_name": room_name}},
        ) as r:
            token = (await r.json())["token"]

    transport = DailyTransport(
        room_url=os.environ["DAILY_ROOM_URL"],
        token=token,
        bot_name="Deepslate Bot",
        params=DailyParams(
            audio_in_enabled=True,
            audio_out_enabled=True,
            vad_enabled=False,  # VAD is handled server-side by Deepslate
        ),
    )

    llm = DeepslateRealtimeLLMService(
        options=DeepslateOptions.from_env(
            system_prompt="You are a friendly and helpful AI assistant."
        ),
        tts_config=ElevenLabsTtsConfig.from_env(),
    )

    pipeline = Pipeline([transport.input(), llm, transport.output()])
    task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True))

    @transport.event_handler("on_participant_left")
    async def on_participant_left(transport, participant, reason):
        await task.cancel()

    await PipelineRunner().run(task)

if __name__ == "__main__":
    asyncio.run(main())

from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from deepslate.pipecat.frames import DeepslateSessionInitializedFrame, DeepslateDirectSpeechFrame

class WelcomeProcessor(FrameProcessor):
    async def process_frame(self, frame, direction):
        await self.push_frame(frame, direction)
        if isinstance(frame, DeepslateSessionInitializedFrame):
            await self.push_frame(
                DeepslateDirectSpeechFrame(text="Hello! How can I help you today?"),
                FrameDirection.DOWNSTREAM,
            )

pipeline = Pipeline([transport.input(), llm, WelcomeProcessor(), transport.output()])

import random
from pipecat.frames.frames import LLMSetToolsFrame
from pipecat.services.llm_service import FunctionCallParams

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "lookup_weather",
            "description": "Get the current weather for a given location.",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string", "description": "The city to look up."}
                },
                "required": ["location"],
            },
        },
    },
]

async def lookup_weather(params: FunctionCallParams):
    result = {
        "location": params.arguments.get("location", "unknown"),
        "temperature_celsius": random.randint(10, 35),
    }
    await params.result_callback(result)

# Register the handler on the service
llm.register_function("lookup_weather", lookup_weather)

# Queue tool definitions — synced to Deepslate after the pipeline starts
await task.queue_frame(LLMSetToolsFrame(tools=TOOLS))

from pipecat.frames.frames import LLMMessagesAppendFrame
from pipecat.processors.llm.base import OpenAILLMContextFrame

# Inject a user message mid-conversation
await task.queue_frame(
    LLMMessagesAppendFrame(
        messages=[{"role": "user", "content": "My name is Alice and I prefer short answers."}]
    )
)

from deepslate.pipecat.frames import DeepslateDirectSpeechFrame

await task.queue_frame(
    DeepslateDirectSpeechFrame(
        text="Welcome back! How can I help you today?",
        include_in_history=True,  # Records as an assistant turn (default: True)
    )
)

from deepslate.pipecat.frames import (
    DeepslateConversationQueryFrame,
    DeepslateConversationQueryResultFrame,
)

# Send the query
await task.queue_frame(
    DeepslateConversationQueryFrame(
        prompt="Summarize the conversation so far in one sentence.",
        instructions="Respond in plain text only, no formatting.",
    )
)

# Receive the result downstream in your pipeline
# DeepslateConversationQueryResultFrame.text contains the model's reply

from deepslate.pipecat.frames import (
    DeepslateExportChatHistoryFrame,
    DeepslateChatHistoryFrame,
)

# Request the export
await task.queue_frame(
    DeepslateExportChatHistoryFrame(
        await_pending=False,  # Set True to wait for any in-flight operations first
    )
)

# Handle the result in a downstream processor
# DeepslateChatHistoryFrame.messages is a list[ChatMessageDict]