diff --git a/src/pipecat/services/anthropic/llm.py b/src/pipecat/services/anthropic/llm.py index 8b25fd2d5..0601d52f5 100644 --- a/src/pipecat/services/anthropic/llm.py +++ b/src/pipecat/services/anthropic/llm.py @@ -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. diff --git a/src/pipecat/services/aws/llm.py b/src/pipecat/services/aws/llm.py index 87771b03b..59494950c 100644 --- a/src/pipecat/services/aws/llm.py +++ b/src/pipecat/services/aws/llm.py @@ -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. diff --git a/src/pipecat/services/google/llm.py b/src/pipecat/services/google/llm.py index bcaa79483..faf3f2f52 100644 --- a/src/pipecat/services/google/llm.py +++ b/src/pipecat/services/google/llm.py @@ -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. diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index b61bdbbbc..cbf295e9c 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -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, diff --git a/src/pipecat/services/openai/base_llm.py b/src/pipecat/services/openai/base_llm.py index 3903b6c98..e591b4b4e 100644 --- a/src/pipecat/services/openai/base_llm.py +++ b/src/pipecat/services/openai/base_llm.py @@ -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]: