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

Introduce an affordance to LLMService for generating a summary of a conversation directly (i.e. without going through the pipeline).

Browse Source 
This abstraction will allow us to update Pipecat Flows to avoid reaching into LLM service internals to generate summaries.

In addition to being a helpful refactor to remove a fragile part of Pipecat Flows, this change helps set the stage for supporting the upcoming `LLMSwitcher`, where the “active” LLM will only be determined at runtime—today, Pipecat Flows needs to know ahead of time what type of LLM it’s working with, to load an LLM-specific “adapter” that does the work of generating summaries, among other things.
This commit is contained in:
Paul Kompfner
2025-08-20 15:01:59 -04:00
parent 195146adb2
commit fe6063fdbe
5 changed files with 239 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
src/pipecat/services/anthropic/llm.py
View File

@@ -42,6 +42,7 @@ from pipecat.frames.frames import (
VisionImageRawFrame,
)
from pipecat.metrics.metrics import LLMTokenUsage
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response import (
LLMAssistantAggregatorParams,
LLMAssistantContextAggregator,
@@ -198,6 +199,56 @@ class AnthropicLLMService(LLMService):
response = await api_call(**params)
return response
async def generate_summary(
self, summary_prompt: str, context: LLMContext | OpenAILLMContext
) -> Optional[str]:
"""Generate a conversation summary from the given LLM context.
Args:
summary_prompt: The prompt to use to guide generating the summary.
context: The LLM context containing conversation history.
Returns:
The generated summary, or None if generation failed.
"""
try:
if isinstance(context, LLMContext):
# Not sure if it's strictly necessary to adapt messages here
# since they'll just be a string in the prompt, but erring on
# the side of putting them in the format the LLM would expect
# if consuming them directly (i.e. assuming greater LLM
# familiarity with its own format).
# adapter = self.get_llm_adapter()
# params: AnthropicLLMInvocationParams = adapter.get_llm_invocation_params(context)
# messages = params["messages"]
raise NotImplementedError(
"Universal LLMContext is not yet supported for Anthropic."
)
else:
messages = context.messages
prompt_messages = [
{
"role": "user",
"content": f"Conversation history: {messages}",
},
]
# LLM completion
response = await self._client.messages.create(
model=self.model_name,
messages=prompt_messages,
system=summary_prompt,
max_tokens=8192,
stream=False,
)
return response.content[0].text
except Exception as e:
logger.error(f"Anthropic summary generation failed: {e}", exc_info=True)
return None
@property
def enable_prompt_caching_beta(self) -> bool:
"""Check if prompt caching beta feature is enabled.

73
src/pipecat/services/aws/llm.py
View File

@@ -41,6 +41,7 @@ from pipecat.frames.frames import (
VisionImageRawFrame,
)
from pipecat.metrics.metrics import LLMTokenUsage
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response import (
LLMAssistantAggregatorParams,
LLMAssistantContextAggregator,
@@ -790,6 +791,78 @@ class AWSBedrockLLMService(LLMService):
"""
return True
async def generate_summary(
self, summary_prompt: str, context: LLMContext | OpenAILLMContext
) -> Optional[str]:
"""Generate a conversation summary from the given LLM context.
Args:
summary_prompt: The prompt to use to guide generating the summary.
context: The LLM context containing conversation history.
Returns:
The generated summary, or None if generation failed.
"""
try:
if isinstance(context, LLMContext):
# Not sure if it's strictly necessary to adapt messages here
# since they'll just be a string in the prompt, but erring on
# the side of putting them in the format the LLM would expect
# if consuming them directly (i.e. assuming greater LLM
# familiarity with its own format).
# adapter = self.get_llm_adapter()
# params: AWSBedrockLLMInvocationParams = adapter.get_llm_invocation_params(context)
# messages = params["messages"]
raise NotImplementedError(
"Universal LLMContext is not yet supported for AWS Bedrock."
)
else:
messages = context.messages
# Determine if we're using Claude or Nova based on model ID
model_id = self.model_name
# Prepare request parameters
request_params = {
"modelId": model_id,
"messages": [
{
"role": "user",
"content": [{"text": f"Conversation history: {messages}"}],
},
],
"inferenceConfig": {
"maxTokens": 8192,
"temperature": 0.7,
"topP": 0.9,
},
}
request_params["system"] = [{"text": summary_prompt}]
# Call Bedrock without streaming
response = self._client.converse(**request_params)
# Extract the response text
if (
"output" in response
and "message" in response["output"]
and "content" in response["output"]["message"]
):
content = response["output"]["message"]["content"]
if isinstance(content, list):
for item in content:
if item.get("text"):
return item["text"]
elif isinstance(content, str):
return content
return None
except Exception as e:
logger.error(f"Bedrock summary generation failed: {e}", exc_info=True)
return None
async def _create_converse_stream(self, client, request_params):
"""Create converse stream with optional timeout and retry.

52
src/pipecat/services/google/llm.py
View File

@@ -733,6 +733,58 @@ class GoogleLLMService(LLMService):
def _create_client(self, api_key: str, http_options: Optional[HttpOptions] = None):
self._client = genai.Client(api_key=api_key, http_options=http_options)
async def generate_summary(
self, summary_prompt: str, context: LLMContext | OpenAILLMContext
) -> Optional[str]:
"""Generate a conversation summary from the given LLM context.
Args:
summary_prompt: The prompt to use to guide generating the summary.
context: The LLM context containing conversation history.
Returns:
The generated summary, or None if generation failed.
"""
try:
if isinstance(context, LLMContext):
# Not sure if it's strictly necessary to adapt messages here
# since they'll just be a string in the prompt, but erring on
# the side of putting them in the format the LLM would expect
# if consuming them directly (i.e. assuming greater LLM
# familiarity with its own format).
adapter = self.get_llm_adapter()
params: GeminiLLMInvocationParams = adapter.get_llm_invocation_params(context)
messages = params["messages"]
else:
messages = context.messages
# Format conversation history as user message
contents = [
Content(role="user", parts=[Part(text=f"Conversation history: {messages}")])
]
# Use summary_prompt as system instruction
generation_config = GenerateContentConfig(system_instruction=summary_prompt)
# Use the new google-genai client's async method
response = await self._client.aio.models.generate_content(
model=self._model_name,
contents=contents,
config=generation_config,
)
# Extract text from response
if response.candidates and response.candidates[0].content:
for part in response.candidates[0].content.parts:
if part.text:
return part.text
return None
except Exception as e:
logger.error(f"Google summary generation failed: {e}", exc_info=True)
return None
def needs_mcp_alternate_schema(self) -> bool:
"""Check if this LLM service requires alternate MCP schema.

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

@@ -14,6 +14,7 @@ from typing import (
Awaitable,
Callable,
Dict,
List,
Mapping,
Optional,
Protocol,
@@ -190,6 +191,20 @@ class LLMService(AIService):
"""
return self._adapter
async def generate_summary(
self, summary_prompt: str, context: LLMContext | OpenAILLMContext
) -> Optional[str]:
"""Generate a conversation summary from the given LLM context.
Args:
summary_prompt: The prompt to use to guide generating the summary.
context: The LLM context containing conversation history.
Returns:
The generated summary, or None if generation failed.
"""
raise NotImplementedError(f"generate_summary() not supported by {self.__class__.__name__}")
def create_context_aggregator(
self,
context: OpenAILLMContext,

48
src/pipecat/services/openai/base_llm.py
View File

@@ -245,6 +245,54 @@ class BaseOpenAILLMService(LLMService):
params.update(self._settings["extra"])
return params
async def generate_summary(
self, summary_prompt: str, context: LLMContext | OpenAILLMContext
) -> Optional[str]:
"""Generate a conversation summary from the given LLM context.
Args:
summary_prompt: The prompt to use to guide generating the summary.
context: The LLM context containing conversation history.
Returns:
The generated summary, or None if generation failed.
"""
try:
if isinstance(context, LLMContext):
# Not sure if it's strictly necessary to adapt messages here
# since they'll just be a string in the prompt, but erring on
# the side of putting them in the format the LLM would expect
# if consuming them directly (i.e. assuming greater LLM
# familiarity with its own format).
adapter = self.get_llm_adapter()
params: OpenAILLMInvocationParams = adapter.get_llm_invocation_params(context)
messages = params["messages"]
else:
messages = context.messages
prompt_messages = [
{
"role": "system",
"content": summary_prompt,
},
{
"role": "user",
"content": f"Conversation history: {messages}",
},
]
# LLM completion
response = await self._client.chat.completions.create(
model=self.model_name,
messages=prompt_messages,
stream=False,
)
return response.choices[0].message.content
except Exception as e:
logger.error(f"OpenAI summary generation failed: {e}", exc_info=True)
return None
async def _stream_chat_completions_specific_context(
self, context: OpenAILLMContext
) -> AsyncStream[ChatCompletionChunk]: