pipecat/examples/foundational/54b-context-summarization-manual-openai.py

#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#

"""Example demonstrating manual context summarization via a function call.

This example shows how to trigger context summarization on demand rather than
automatically. The user can ask the bot to "summarize the conversation" and the
bot will call a function that pushes an LLMSummarizeContextFrame into the
pipeline, causing the LLM service to compress the conversation history.

Unlike example 54, automatic summarization is NOT enabled here. Summarization
only happens when the user explicitly requests it through the function call.
"""

import os

from dotenv import load_dotenv
from loguru import logger

from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import LLMRunFrame, LLMSummarizeContextFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
    LLMContextAggregatorPair,
    LLMUserAggregatorParams,
)
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams

load_dotenv(override=True)

# We use lambdas to defer transport parameter creation until the transport
# type is selected at runtime.
transport_params = {
    "daily": lambda: DailyParams(
        audio_in_enabled=True,
        audio_out_enabled=True,
    ),
    "twilio": lambda: FastAPIWebsocketParams(
        audio_in_enabled=True,
        audio_out_enabled=True,
    ),
    "webrtc": lambda: TransportParams(
        audio_in_enabled=True,
        audio_out_enabled=True,
    ),
}


async def summarize_conversation(params: FunctionCallParams):
    """Trigger manual context summarization via a pipeline frame."""
    logger.info("Tool called: summarize_conversation")
    await params.result_callback({"status": "summarization_requested"})
    await params.llm.queue_frame(LLMSummarizeContextFrame())


async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
    logger.info("Starting bot")

    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))

    tts = CartesiaTTSService(
        api_key=os.getenv("CARTESIA_API_KEY"),
        settings=CartesiaTTSService.Settings(
            voice="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
        ),
    )

    system_prompt = """You are a helpful LLM in a voice call. Your goal is to demonstrate your
                    capabilities in a succinct way. Your output will be spoken aloud, so avoid
                    special characters that can't easily be spoken, such as emojis or bullet points.
                    Respond to what the user said in a creative and helpful way.
                    If the user asks you to summarize the conversation, call the
                    summarize_conversation function. After summarization, briefly acknowledge
                    that the conversation history has been compressed.
                    """

    llm = OpenAILLMService(
        api_key=os.getenv("OPENAI_API_KEY"),
        settings=OpenAILLMService.Settings(
            system_instruction=system_prompt,
        ),
    )

    llm.register_function("summarize_conversation", summarize_conversation)

    summarize_function = FunctionSchema(
        name="summarize_conversation",
        description=(
            "Summarize and compress the conversation history. "
            "Call this when the user asks you to summarize the conversation "
            "or when you want to free up context space."
        ),
        properties={},
        required=[],
    )
    tools = ToolsSchema(standard_tools=[summarize_function])

    context = LLMContext(tools=tools)

    # Automatic summarization is NOT enabled here (enable_auto_context_summarization
    # defaults to False). The summarizer is still created internally so that
    # LLMSummarizeContextFrame frames pushed via the function call are handled.
    user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
        context,
        user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
    )

    pipeline = Pipeline(
        [
            transport.input(),  # Transport user input
            stt,
            user_aggregator,  # User responses
            llm,  # LLM
            tts,  # TTS
            transport.output(),  # Transport bot output
            assistant_aggregator,  # Assistant spoken responses
        ]
    )

    task = PipelineTask(
        pipeline,
        params=PipelineParams(
            enable_metrics=True,
            enable_usage_metrics=True,
        ),
        idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
    )

    @transport.event_handler("on_client_connected")
    async def on_client_connected(transport, client):
        logger.info("Client connected")
        # Kick off the conversation.
        context.add_message(
            {"role": "developer", "content": "Please introduce yourself to the user."}
        )
        await task.queue_frames([LLMRunFrame()])

    @transport.event_handler("on_client_disconnected")
    async def on_client_disconnected(transport, client):
        logger.info("Client disconnected")
        await task.cancel()

    runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)

    await runner.run(task)


async def bot(runner_args: RunnerArguments):
    """Main bot entry point compatible with Pipecat Cloud."""
    transport = await create_transport(runner_args, transport_params)
    await run_bot(transport, runner_args)


if __name__ == "__main__":
    from pipecat.runner.run import main

    main()