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(add_tool_change_messages: bool = False, audio_idle_timeout: float = 1.0, 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, filter_incomplete_user_turns: bool = False, user_turn_completion_config: UserTurnCompletionConfig | None = None)[source]

Bases: object

Parameters for configuring LLM user aggregation behavior.

Parameters:

add_tool_change_messages – When True, on each LLMSetToolsFrame the aggregator computes the diff against the currently advertised tools and appends a developer-role message to the context describing additions/removals. Helps the LLM stay coherent across mid-conversation tool changes, mitigating several flavors of tool-call-related hallucination: calling tools that have been removed, avoiding tools that have been re-added, and hallucinating output (made-up answers or tool-call-shaped non-tool-calls) when tools are unavailable. Only standard tools are diffed; custom (LLM-specific) tools are ignored. When using LLMContextAggregatorPair, prefer setting this via its add_tool_change_messages argument instead. Defaults to False.
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.
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.
filter_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.

Deprecated since version 1.2.0: Use user_turn_strategies=FilterIncompleteUserTurnStrategies() instead. Will be removed in 2.0.0.
user_turn_completion_config –
Configuration for turn completion behavior including custom instructions, timeouts, and prompts. Only used when filter_incomplete_user_turns is True (deprecated path) — for the new strategy-based API, pass the config directly to FilterIncompleteUserTurnStrategies(config=...).

Deprecated since version 1.2.0: Pass the config directly to FilterIncompleteUserTurnStrategies(config=...) instead. Will be removed in 2.0.0.

add_tool_change_messages: bool = False

audio_idle_timeout: float = 1.0

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

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, add_tool_change_messages: bool = False, 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.
add_tool_change_messages – When True, on each LLMSetToolsFrame the aggregator computes the diff against the currently advertised tools and appends a developer-role message to the context describing additions/removals. Helps the LLM stay coherent across mid-conversation tool changes, mitigating several flavors of tool-call-related hallucination: calling tools that have been removed, avoiding tools that have been re-added, and hallucinating output (made-up answers or tool-call-shaped non-tool-calls) when tools are unavailable. Only standard tools are diffed; custom (LLM-specific) tools are ignored. When using LLMContextAggregatorPair, prefer setting this via its add_tool_change_messages argument instead. Defaults to False.
enable_context_summarization –
Legacy field name.

Deprecated since version 1.2.0: Use enable_auto_context_summarization instead. Will be removed in 2.0.0.
context_summarization_config –
Legacy field name.

Deprecated since version 1.2.0: Use auto_context_summarization_config instead. Will be removed in 2.0.0.

enable_auto_context_summarization: bool = False

auto_context_summarization_config: LLMAutoContextSummarizationConfig | None = None

add_tool_change_messages: bool = False

enable_context_summarization: bool | None = None

context_summarization_config: LLMContextSummarizationConfig | None = None

class pipecat.processors.aggregators.llm_response_universal.UserTurnStoppedMessage(content: str | None, 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. None in realtime mode (realtime_service_mode=True) when fired from a user-turn-stop frame: in realtime mode the user message isn’t finalized until the assistant response start acts as the effective end-of-turn signal. Subscribers that need the finalized text should listen to on_user_turn_message_added instead.
timestamp – When the user turn started.
user_id – Optional identifier for the user.

content: str | None

timestamp: str

user_id: str | None = None

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

Bases: object

A message accompanying on_user_turn_message_added.

Fired when a user message is written to the LLM context. In cascade mode (realtime_service_mode=False) this coincides with on_user_turn_stopped. In realtime mode (realtime_service_mode=True) the write is triggered by the assistant response start (or session end), so this event fires decoupled from user-turn-end frames. content is always populated.

Parameters:

content – The aggregated user transcript.
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, add_tool_change_messages: bool = False, **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.

TOOL_ACTIVATION_MESSAGE_TEMPLATE = 'The following function(s) have just been added and may now be called: {function_names}. Any previously available functions remain available.'

TOOL_DEACTIVATION_MESSAGE_TEMPLATE = 'The following function(s) have just been removed and should not be called: {function_names}. Any previously available functions remain available. The removed function(s) may become available again later, in which case you will be informed.'

__init__(*, context: LLMContext, role: str, add_tool_change_messages: bool = False, **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”).
add_tool_change_messages – See the field of the same name on the aggregator-specific params dataclasses. Subclasses propagate this from their params.
**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, _realtime_service_mode: bool | 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_turn_message_added: Called when a user message is written to context. In realtime mode (realtime_service_mode=True) the write is triggered by the assistant response start rather than the user-turn-end frame, so this fires decoupled from on_user_turn_stopped — subscribe here for the finalized user text.
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):
    ...

# In realtime mode (realtime_service_mode=True) the user message
# is written when the assistant response starts, not at
# user-turn-end — subscribe here for the finalized text.
@aggregator.event_handler("on_user_turn_message_added")
async def on_user_turn_message_added(aggregator, message: UserTurnMessageAddedMessage):
    ...

@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, _realtime_service_mode: bool | None = None, **kwargs)[source]

Initialize the user context aggregator.

Parameters:

context – The LLM context for conversation storage.
params – Configuration parameters for aggregation behavior.
_realtime_service_mode – Pair-internal. Realtime-mode flag propagated from LLMContextAggregatorPair (None = auto-configure from service metadata). Not intended for direct use — construct the aggregators via the pair.
**kwargs – Additional arguments.

async cleanup()[source]: Release this aggregator’s resources at teardown.

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, _realtime_service_mode: bool | None = None, _paired_user_aggregator: LLMUserAggregator | 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, _realtime_service_mode: bool | None = None, _paired_user_aggregator: LLMUserAggregator | None = None, **kwargs)[source]

Initialize the assistant context aggregator.

Parameters:

context – The OpenAI LLM context for conversation storage.
params – Configuration parameters for aggregation behavior.
_realtime_service_mode – Pair-internal. Realtime-mode flag propagated from LLMContextAggregatorPair (None = auto-configure from service metadata). Not intended for direct use — construct the aggregators via the pair.
_paired_user_aggregator – Pair-internal. Back-reference to the paired LLMUserAggregator. The assistant flushes it on LLMFullResponseStartFrame so the user message lands in context before the assistant turn starts.
**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 cleanup()[source]: Release this aggregator’s resources at teardown.

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, add_tool_change_messages: bool | None = None, realtime_service_mode: bool | 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, add_tool_change_messages: bool | None = None, realtime_service_mode: bool | 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.
add_tool_change_messages – When provided, sets the field of the same name on both user_params and assistant_params, overriding any value already set on either. This is the preferred way to enable tool-change announcements: it ensures both aggregators participate, which makes the feature robust regardless of which aggregator handles a given LLMSetToolsFrame. The shared context guarantees the announcement is added exactly once (the second aggregator’s diff is empty by the time it sees the frame). Leave as None to respect per-params settings.
realtime_service_mode – Configures the pair for use with a realtime (speech-to-speech) LLM service. When enabled, context writes become trailing — driven by the content stream itself (transcripts, LLMFullResponseStartFrame) rather than turn frames — and turn-end strategies stop waiting for transcripts. None (the default) auto-configures: the mode turns on when a realtime service announces itself via service metadata, and stays off otherwise. True / False force it on or off (False keeps legacy pre-realtime context-write behavior). Both halves share this setting.

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.