wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #3855 from pipecat-ai/mb/context-summarization-improvements

Browse Source 
Improve context summarization with dedicated LLM, timeout, and observability
This commit is contained in:
Mark Backman
2026-02-27 15:24:38 -05:00
committed by GitHub
parent 9747e8da4a f74af9b9c7
commit 83e29eb478
14 changed files with 912 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/pipecat/frames/frames.py
View File

@@ -2019,6 +2019,8 @@ class LLMContextSummaryRequestFrame(ControlFrame):
the summary text.
summarization_prompt: System prompt instructing the LLM how to generate
the summary.
summarization_timeout: Maximum time in seconds for the LLM to generate a
summary. When None, a default timeout of 120s is applied.
"""
request_id: str
@@ -2026,6 +2028,7 @@ class LLMContextSummaryRequestFrame(ControlFrame):
min_messages_to_keep: int
target_context_tokens: int
summarization_prompt: str
summarization_timeout: Optional[float] = None
@dataclass

123
src/pipecat/processors/aggregators/llm_context_summarizer.py
View File

@@ -6,8 +6,10 @@
"""This module defines a summarizer for managing LLM context summarization."""
import asyncio
import uuid
from typing import Optional
from dataclasses import dataclass
from typing import TYPE_CHECKING, Optional
from loguru import logger
@@ -22,10 +24,33 @@ from pipecat.processors.aggregators.llm_context import LLMContext, LLMSpecificMe
from pipecat.utils.asyncio.task_manager import BaseTaskManager
from pipecat.utils.base_object import BaseObject
from pipecat.utils.context.llm_context_summarization import (
DEFAULT_SUMMARIZATION_TIMEOUT,
LLMContextSummarizationConfig,
LLMContextSummarizationUtil,
)
if TYPE_CHECKING:
from pipecat.services.llm_service import LLMService
@dataclass
class SummaryAppliedEvent:
"""Event data emitted when context summarization completes successfully.
Parameters:
original_message_count: Number of messages before summarization.
new_message_count: Number of messages after summarization.
summarized_message_count: Number of messages that were compressed
into the summary.
preserved_message_count: Number of recent messages preserved
uncompressed.
"""
original_message_count: int
new_message_count: int
summarized_message_count: int
preserved_message_count: int
class LLMContextSummarizer(BaseObject):
"""Summarizer for managing LLM context summarization.
@@ -39,6 +64,10 @@ class LLMContextSummarizer(BaseObject):
- on_request_summarization: Emitted when summarization should be triggered.
The aggregator should broadcast this frame to the LLM service.
- on_summary_applied: Emitted after a summary has been successfully applied
to the context. Receives a SummaryAppliedEvent with metrics about the
compression.
Example::
@summarizer.event_handler("on_request_summarization")
@@ -49,6 +78,10 @@ class LLMContextSummarizer(BaseObject):
context=frame.context,
...
)
@summarizer.event_handler("on_summary_applied")
async def on_summary_applied(summarizer, event: SummaryAppliedEvent):
logger.info(f"Compressed {event.original_message_count} -> {event.new_message_count} messages")
"""
def __init__(
@@ -74,6 +107,7 @@ class LLMContextSummarizer(BaseObject):
self._pending_summary_request_id: Optional[str] = None
self._register_event_handler("on_request_summarization", sync=True)
self._register_event_handler("on_summary_applied")
@property
def task_manager(self) -> BaseTaskManager:
@@ -198,8 +232,10 @@ class LLMContextSummarizer(BaseObject):
async def _request_summarization(self):
"""Request context summarization from LLM service.
Creates a summarization request frame and emits it via event handler.
Tracks the request ID to match async responses and prevent race conditions.
Creates a summarization request frame and either handles it directly
using a dedicated LLM (if configured) or emits it via event handler
for the pipeline's primary LLM. Tracks the request ID to match async
responses and prevent race conditions.
"""
# Generate unique request ID
request_id = str(uuid.uuid4())
@@ -218,10 +254,63 @@ class LLMContextSummarizer(BaseObject):
min_messages_to_keep=min_keep,
target_context_tokens=self._config.target_context_tokens,
summarization_prompt=self._config.summary_prompt,
summarization_timeout=self._config.summarization_timeout,
)
# Emit event for aggregator to broadcast
await self._call_event_handler("on_request_summarization", request_frame)
if self._config.llm:
# Use dedicated LLM directly — no need to involve the pipeline
self.task_manager.create_task(
self._generate_summary_with_dedicated_llm(self._config.llm, request_frame),
f"{self}-dedicated-llm-summary",
)
else:
# Emit event for aggregator to broadcast to the pipeline LLM
await self._call_event_handler("on_request_summarization", request_frame)
async def _generate_summary_with_dedicated_llm(
self, llm: "LLMService", frame: LLMContextSummaryRequestFrame
):
"""Generate summary using a dedicated LLM service.
Calls the dedicated LLM's _generate_summary directly and feeds the
result back through _handle_summary_result, bypassing the pipeline.
Args:
llm: The dedicated LLM service to use for summarization.
frame: The summarization request frame.
"""
timeout = frame.summarization_timeout or DEFAULT_SUMMARIZATION_TIMEOUT
try:
summary, last_index = await asyncio.wait_for(
llm._generate_summary(frame),
timeout=timeout,
)
result_frame = LLMContextSummaryResultFrame(
request_id=frame.request_id,
summary=summary,
last_summarized_index=last_index,
)
except asyncio.TimeoutError:
error = f"Context summarization timed out after {timeout}s"
logger.error(f"{self}: {error}")
result_frame = LLMContextSummaryResultFrame(
request_id=frame.request_id,
summary="",
last_summarized_index=-1,
error=error,
)
except Exception as e:
error = f"Error generating context summary: {e}"
logger.error(f"{self}: {error}")
result_frame = LLMContextSummaryResultFrame(
request_id=frame.request_id,
summary="",
last_summarized_index=-1,
error=error,
)
await self._handle_summary_result(result_frame)
async def _handle_summary_result(self, frame: LLMContextSummaryResultFrame):
"""Handle context summarization result from LLM service.
@@ -306,8 +395,10 @@ class LLMContextSummarizer(BaseObject):
# Get recent messages to keep
recent_messages = messages[last_summarized_index + 1 :]
# Create summary message as an assistant message
summary_message = {"role": "assistant", "content": f"Conversation summary: {summary}"}
# Create summary message as a user message (the summary is context
# provided *to* the assistant, not something the assistant said)
summary_content = self._config.summary_message_template.format(summary=summary)
summary_message = {"role": "user", "content": summary_content}
# Reconstruct context
new_messages = []
@@ -317,9 +408,23 @@ class LLMContextSummarizer(BaseObject):
new_messages.extend(recent_messages)
# Update context
original_message_count = len(messages)
num_system_preserved = 1 if first_system_msg else 0
self._context.set_messages(new_messages)
# Messages actually summarized = index range minus the preserved system message
summarized_count = last_summarized_index + 1 - num_system_preserved
logger.info(
f"{self}: Applied context summary, compressed {last_summarized_index + 1} messages "
f"into summary. Context now has {len(new_messages)} messages (was {len(messages)})"
f"{self}: Applied context summary, compressed {summarized_count} messages "
f"into summary. Context now has {len(new_messages)} messages (was {original_message_count})"
)
# Emit event for observability
event = SummaryAppliedEvent(
original_message_count=original_message_count,
new_message_count=len(new_messages),
summarized_message_count=summarized_count,
preserved_message_count=len(recent_messages) + num_system_preserved,
)
await self._call_event_handler("on_summary_applied", event)

10
src/pipecat/services/llm_service.py
View File

@@ -62,6 +62,7 @@ from pipecat.services.ai_service import AIService
from pipecat.services.settings import LLMSettings
from pipecat.turns.user_turn_completion_mixin import UserTurnCompletionLLMServiceMixin
from pipecat.utils.context.llm_context_summarization import (
DEFAULT_SUMMARIZATION_TIMEOUT,
LLMContextSummarizationUtil,
)
@@ -436,8 +437,15 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
last_index = -1
error = None
timeout = frame.summarization_timeout or DEFAULT_SUMMARIZATION_TIMEOUT
try:
summary, last_index = await self._generate_summary(frame)
summary, last_index = await asyncio.wait_for(
self._generate_summary(frame),
timeout=timeout,
)
except asyncio.TimeoutError:
await self.push_error(error_msg=f"Context summarization timed out after {timeout}s")
except Exception as e:
error = f"Error generating context summary: {e}"
await self.push_error(error, exception=e)

24
src/pipecat/utils/context/llm_context_summarization.py
View File

@@ -11,12 +11,18 @@ context when token limits are reached, enabling efficient long-running conversat
"""
from dataclasses import dataclass
from typing import List, Optional
from typing import TYPE_CHECKING, List, Optional
if TYPE_CHECKING:
from pipecat.services.llm_service import LLMService
from loguru import logger
from pipecat.processors.aggregators.llm_context import LLMContext, LLMSpecificMessage
# Fallback timeout (seconds) used when summarization_timeout is None.
DEFAULT_SUMMARIZATION_TIMEOUT = 120.0
# Token estimation constants
CHARS_PER_TOKEN = 4 # Industry-standard heuristic: 1 token ≈ 4 characters
TOKEN_OVERHEAD_PER_MESSAGE = 10 # Estimated structural overhead per message
@@ -73,6 +79,19 @@ class LLMContextSummarizationConfig:
immediate conversational context.
summarization_prompt: Custom prompt for the LLM to use when generating
summaries. If None, uses DEFAULT_SUMMARIZATION_PROMPT.
summary_message_template: Template for formatting the summary when
injected into context. Must contain ``{summary}`` as a placeholder
for the generated summary text. Allows applications to wrap the
summary in custom delimiters (e.g., XML tags) so that system
prompts can distinguish summaries from live conversation.
llm: Optional separate LLM service for generating summaries. When set,
summarization requests are sent to this service instead of the
pipeline's primary LLM. Useful for routing summarization to a
cheaper/faster model (e.g., Gemini Flash) while keeping an
expensive model for conversation. If None, uses the pipeline LLM.
summarization_timeout: Maximum time in seconds to wait for the LLM to
generate a summary. If the call exceeds this timeout, summarization
is aborted with an error and future summarizations are unblocked.
"""
max_context_tokens: int = 8000
@@ -80,6 +99,9 @@ class LLMContextSummarizationConfig:
max_unsummarized_messages: int = 20
min_messages_after_summary: int = 4
summarization_prompt: Optional[str] = None
summary_message_template: str = "Conversation summary: {summary}"
llm: Optional["LLMService"] = None
summarization_timeout: float = DEFAULT_SUMMARIZATION_TIMEOUT
def __post_init__(self):
"""Validate configuration parameters."""