llm_response_universal

LLM response aggregators for handling conversation context and message aggregation.

This module provides aggregators that process and accumulate LLM responses, user inputs, and conversation context. These aggregators handle the flow between speech-to-text, LLM processing, and text-to-speech components in conversational AI pipelines.

class pipecat.processors.aggregators.llm_response_universal.LLMUserAggregatorParams(user_turn_strategies: UserTurnStrategies | None = None, user_mute_strategies: list[BaseUserMuteStrategy] = <factory>, user_turn_stop_timeout: float = 5.0, user_idle_timeout: float = 0, vad_analyzer: VADAnalyzer | None = None, audio_idle_timeout: float = 1.0, filter_incomplete_user_turns: bool = False, user_turn_completion_config: UserTurnCompletionConfig | None = None)[source]

Bases: object

Parameters for configuring LLM user aggregation behavior.

Parameters:

user_turn_strategies – User turn start and stop strategies.
user_mute_strategies – List of user mute strategies.
user_turn_stop_timeout – Time in seconds to wait before considering the user’s turn finished.
user_idle_timeout – Timeout in seconds for detecting user idle state. The aggregator will emit an on_user_turn_idle event when the user has been idle (not speaking) for this duration. Set to 0 to disable idle detection.
vad_analyzer – Voice Activity Detection analyzer instance.
audio_idle_timeout – Timeout in seconds to force speech stop when no audio frames are received while in SPEAKING state (e.g. user mutes mic mid-speech). Set to 0 to disable. Defaults to 1.0.
filter_incomplete_user_turns – Whether to filter out incomplete user turns. When enabled, the LLM outputs a turn completion marker at the start of each response: ✓ (complete), ○ (incomplete short), or ◐ (incomplete long). Incomplete responses are suppressed and timeouts trigger re-prompting.
user_turn_completion_config – Configuration for turn completion behavior including custom instructions, timeouts, and prompts. Only used when filter_incomplete_user_turns is True.

user_turn_strategies: UserTurnStrategies | None = None

user_mute_strategies: list[BaseUserMuteStrategy]

user_turn_stop_timeout: float = 5.0

user_idle_timeout: float = 0

vad_analyzer: VADAnalyzer | None = None

audio_idle_timeout: float = 1.0

filter_incomplete_user_turns: bool = False

user_turn_completion_config: UserTurnCompletionConfig | None = None

class pipecat.processors.aggregators.llm_response_universal.LLMAssistantAggregatorParams(enable_auto_context_summarization: bool = False, auto_context_summarization_config: LLMAutoContextSummarizationConfig | None = None, enable_context_summarization: bool | None = None, context_summarization_config: LLMContextSummarizationConfig | None = None)[source]

Bases: object

Parameters for configuring LLM assistant aggregation behavior.

Parameters:

enable_auto_context_summarization – Enable automatic context summarization when token or message-count limits are reached (disabled by default). When enabled, older conversation messages are automatically compressed into summaries to manage context size.
auto_context_summarization_config – Configuration for automatic context summarization. Controls trigger thresholds, message preservation, and summarization prompts. If None, uses default LLMAutoContextSummarizationConfig values.

enable_auto_context_summarization: bool = False

auto_context_summarization_config: LLMAutoContextSummarizationConfig | None = None

enable_context_summarization: bool | None = None

context_summarization_config: LLMContextSummarizationConfig | None = None

class pipecat.processors.aggregators.llm_response_universal.UserTurnStoppedMessage(content: str, timestamp: str, user_id: str | None = None)[source]

Bases: object

A user turn stopped message containing a user transcript update.

A message in a conversation transcript containing the user content. This is the aggregated transcript that is then used in the context.

Parameters:

content – The message content/text.
timestamp – When the user turn started.
user_id – Optional identifier for the user.

content: str

timestamp: str

user_id: str | None = None

class pipecat.processors.aggregators.llm_response_universal.AssistantTurnStoppedMessage(content: str, interrupted: bool, timestamp: str)[source]

Bases: object

An assistant turn stopped message containing an assistant transcript update.

A message in a conversation transcript containing the assistant content. This is the aggregated transcript that is then used in the context.

Parameters:

content – The message content/text. May be empty if the LLM returned zero tokens (e.g. turn was interrupted before any tokens were received or pushed)
interrupted – Whether the assistant turn was interrupted.
timestamp – When the assistant turn started.

content: str

interrupted: bool

timestamp: str

class pipecat.processors.aggregators.llm_response_universal.AssistantThoughtMessage(content: str, timestamp: str)[source]

Bases: object

An assistant thought message containing an assistant thought update.

A message in a conversation transcript containing the assistant thought content.

Parameters:

content – The message content/text.
timestamp – When the thought started.

content: str

timestamp: str

class pipecat.processors.aggregators.llm_response_universal.LLMContextAggregator(*, context: LLMContext, role: str, **kwargs)[source]

Bases: FrameProcessor

Base LLM aggregator that uses an LLMContext for conversation storage.

This aggregator maintains conversation state using an LLMContext and pushes LLMContextFrame objects as aggregation frames. It provides common functionality for context-based conversation management.

__init__(*, context: LLMContext, role: str, **kwargs)[source]

Initialize the context response aggregator.

Parameters:

context – The LLM context to use for conversation storage.
role – The role this aggregator represents (e.g. “user”, “assistant”).
**kwargs – Additional arguments passed to parent class.

Get messages from the LLM context.

Returns:: List of message dictionaries from the context.

property role: str

Get the role for this aggregator.

Returns:: The role string for this aggregator.

property context

Get the LLM context.

Returns:: The LLMContext instance used by this aggregator.

async push_context_frame(direction: FrameDirection = FrameDirection.DOWNSTREAM)[source]

Push a context frame in the specified direction.

Parameters:: direction – The direction to push the frame (upstream or downstream).

add_messages(messages)[source]

Add messages to the context.

Parameters:: messages – Messages to add to the conversation context.

set_messages(messages)[source]

Set the context messages.

Parameters:: messages – Messages to replace the current context messages.

Transform the context messages using a provided function.

Parameters:: transform – A function that takes the current list of messages and returns a modified list of messages to set in the context.

set_tools(tools: ToolsSchema | NotGiven)[source]

Set tools in the context.

Parameters:: tools – List of tool definitions to set in the context.

set_tool_choice(tool_choice: Literal['none', 'auto', 'required'] | dict)[source]

Set tool choice in the context.

Parameters:: tool_choice – Tool choice configuration for the context.

async reset()[source]: Reset the aggregation state.

abstractmethod async push_aggregation() → str[source]

Push the current aggregation downstream.

Returns:: The pushed aggregation.

aggregation_string() → str[source]

Get the current aggregation as a string.

Returns:: The concatenated aggregation string.

class pipecat.processors.aggregators.llm_response_universal.LLMUserAggregator(context: LLMContext, *, params: LLMUserAggregatorParams | None = None, **kwargs)[source]

Bases: LLMContextAggregator

User LLM aggregator that aggregates user input during active user turns.

This aggregator uses a turn controller and operates within turn boundaries defined by the controller’s configured user turn strategies. User turn start strategies indicate when a user turn begins, while user turn stop strategies signal when the user turn has ended.

The aggregator collects and aggregates speech-to-text transcriptions that occur while a user turn is active and pushes the final aggregation when the user turn is finished.

Event handlers available:

on_user_turn_started: Called when the user turn starts
on_user_turn_stopped: Called when the user turn ends
on_user_turn_stop_timeout: Called when no user turn stop strategy triggers
on_user_turn_idle: Called when the user has been idle for the configured timeout
on_user_mute_started: Called when the user becomes muted
on_user_mute_stopped: Called when the user becomes unmuted

Example:

@aggregator.event_handler("on_user_turn_started")
async def on_user_turn_started(aggregator, strategy: BaseUserTurnStartStrategy):
    ...

@aggregator.event_handler("on_user_turn_stopped")
async def on_user_turn_stopped(aggregator, strategy: BaseUserTurnStopStrategy, message: UserTurnStoppedMessage):
    ...

@aggregator.event_handler("on_user_turn_stop_timeout")
async def on_user_turn_stop_timeout(aggregator):
    ...

@aggregator.event_handler("on_user_turn_idle")
async def on_user_turn_idle(aggregator):
    ...

@aggregator.event_handler("on_user_mute_started")
async def on_user_mute_started(aggregator):
    ...

@aggregator.event_handler("on_user_mute_stopped")
async def on_user_mute_stopped(aggregator):
    ...

__init__(context: LLMContext, *, params: LLMUserAggregatorParams | None = None, **kwargs)[source]

Initialize the user context aggregator.

Parameters:

context – The LLM context for conversation storage.
params – Configuration parameters for aggregation behavior.
**kwargs – Additional arguments.

async cleanup()[source]: Clean up processor resources.

async process_frame(frame: Frame, direction: FrameDirection)[source]

Process frames for user speech aggregation and context management.

Parameters:

frame – The frame to process.
direction – The direction of frame flow in the pipeline.

async push_aggregation() → str[source]: Push the current aggregation.

class pipecat.processors.aggregators.llm_response_universal.LLMAssistantAggregator(context: LLMContext, *, params: LLMAssistantAggregatorParams | None = None, **kwargs)[source]

Bases: LLMContextAggregator

Assistant LLM aggregator that processes bot responses and function calls.

This aggregator handles the complex logic of processing assistant responses including:

Text frame aggregation between response start/end markers
Function call lifecycle management
Context updates with timestamps
Tool execution and result handling
Interruption handling during responses

The aggregator manages function calls in progress and coordinates between text generation and tool execution phases of LLM responses.

Event handlers available:

on_assistant_turn_started: Called when the assistant turn starts
on_assistant_turn_stopped: Called when the assistant turn ends
on_assistant_thought: Called when an assistant thought is available
on_summary_applied: Called when a context summarization is applied

Example:

@aggregator.event_handler("on_assistant_turn_started")
async def on_assistant_turn_started(aggregator):
    ...

@aggregator.event_handler("on_assistant_turn_stopped")
async def on_assistant_turn_stopped(aggregator, message: AssistantTurnStoppedMessage):
    ...

@aggregator.event_handler("on_assistant_thought")
async def on_assistant_thought(aggregator, message: AssistantThoughtMessage):
    ...

@aggregator.event_handler("on_summary_applied")
async def on_summary_applied(aggregator, summarizer, event: SummaryAppliedEvent):
    ...

__init__(context: LLMContext, *, params: LLMAssistantAggregatorParams | None = None, **kwargs)[source]

Initialize the assistant context aggregator.

Parameters:

context – The OpenAI LLM context for conversation storage.
params – Configuration parameters for aggregation behavior.
**kwargs – Additional arguments.

property has_function_calls_in_progress: bool

Check if there are any function calls currently in progress.

Returns:: True if function calls are in progress, False otherwise.

async reset()[source]: Reset the aggregation state.

async process_frame(frame: Frame, direction: FrameDirection)[source]

Process frames for assistant response aggregation and function call management.

Parameters:

frame – The frame to process.
direction – The direction of frame flow in the pipeline.

async push_aggregation() → str[source]: Push the current assistant aggregation with timestamp.

async push_context_frame(direction: FrameDirection = FrameDirection.DOWNSTREAM)[source]

Push a context frame in the specified direction.

Parameters:: direction – The direction to push the frame (upstream or downstream).

class pipecat.processors.aggregators.llm_response_universal.LLMContextAggregatorPair(context: LLMContext, *, user_params: LLMUserAggregatorParams | None = None, assistant_params: LLMAssistantAggregatorParams | None = None)[source]

Bases: object

Pair of LLM context aggregators for updating context with user and assistant messages.

__init__(context: LLMContext, *, user_params: LLMUserAggregatorParams | None = None, assistant_params: LLMAssistantAggregatorParams | None = None)[source]

Initialize the LLM context aggregator pair.

Parameters:

context – The context to be managed by the aggregators.
user_params – Parameters for the user context aggregator.
assistant_params – Parameters for the assistant context aggregator.

user() → LLMUserAggregator[source]

Get the user context aggregator.

Returns:: The user context aggregator instance.

assistant() → LLMAssistantAggregator[source]

Get the assistant context aggregator.

Returns:: The assistant context aggregator instance.