From 5c574eaad9e3ae7428385bedf29b2fd9af643a02 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 3 Sep 2025 10:58:41 -0400 Subject: [PATCH] Add support for universal `LLMContext` to Anthropic LLM service --- .../14b-function-calling-anthropic-video.py | 2 +- ...ion-calling-anthropic-universal-context.py | 100 +++++-- src/pipecat/adapters/base_llm_adapter.py | 3 +- .../adapters/services/anthropic_adapter.py | 261 +++++++++++++++++- .../adapters/services/gemini_adapter.py | 6 +- src/pipecat/services/anthropic/llm.py | 80 ++++-- src/pipecat/services/google/llm.py | 12 +- src/pipecat/services/openai/base_llm.py | 2 +- 8 files changed, 394 insertions(+), 72 deletions(-) diff --git a/examples/foundational/14b-function-calling-anthropic-video.py b/examples/foundational/14b-function-calling-anthropic-video.py index fee1a9574..f2364cbee 100644 --- a/examples/foundational/14b-function-calling-anthropic-video.py +++ b/examples/foundational/14b-function-calling-anthropic-video.py @@ -97,7 +97,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): llm = AnthropicLLMService( api_key=os.getenv("ANTHROPIC_API_KEY"), model="claude-3-7-sonnet-latest", - enable_prompt_caching_beta=True, + params=AnthropicLLMService.InputParams(enable_prompt_caching_beta=True), ) llm.register_function("get_weather", get_weather) llm.register_function("get_image", get_image) diff --git a/examples/foundational/14z-function-calling-anthropic-universal-context.py b/examples/foundational/14z-function-calling-anthropic-universal-context.py index b11d9e268..9129e52e3 100644 --- a/examples/foundational/14z-function-calling-anthropic-universal-context.py +++ b/examples/foundational/14z-function-calling-anthropic-universal-context.py @@ -5,6 +5,7 @@ # +import asyncio import os from dotenv import load_dotenv @@ -20,25 +21,49 @@ from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.runner.types import RunnerArguments -from pipecat.runner.utils import create_transport +from pipecat.runner.utils import ( + create_transport, + get_transport_client_id, + maybe_capture_participant_camera, +) from pipecat.services.anthropic.llm import AnthropicLLMService from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.services.llm_service import FunctionCallParams from pipecat.transports.base_transport import BaseTransport, TransportParams -from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams from pipecat.transports.services.daily import DailyParams load_dotenv(override=True) +# Global variable to store the client ID +client_id = "" + + async def get_weather(params: FunctionCallParams): location = params.arguments["location"] await params.result_callback(f"The weather in {location} is currently 72 degrees and sunny.") -async def fetch_restaurant_recommendation(params: FunctionCallParams): - await params.result_callback({"name": "The Golden Dragon"}) +async def get_image(params: FunctionCallParams): + question = params.arguments["question"] + logger.debug(f"Requesting image with user_id={client_id}, question={question}") + + # Request the image frame + await params.llm.request_image_frame( + user_id=client_id, + function_name=params.function_name, + tool_call_id=params.tool_call_id, + text_content=question, + ) + + # Wait a short time for the frame to be processed + await asyncio.sleep(0.5) + + # Return a result to complete the function call + await params.result_callback( + f"I've captured an image from your camera and I'm analyzing what you asked about: {question}" + ) # We store functions so objects (e.g. SileroVADAnalyzer) don't get @@ -48,16 +73,13 @@ transport_params = { "daily": lambda: DailyParams( audio_in_enabled=True, audio_out_enabled=True, - vad_analyzer=SileroVADAnalyzer(), - ), - "twilio": lambda: FastAPIWebsocketParams( - audio_in_enabled=True, - audio_out_enabled=True, + video_in_enabled=True, vad_analyzer=SileroVADAnalyzer(), ), "webrtc": lambda: TransportParams( audio_in_enabled=True, audio_out_enabled=True, + video_in_enabled=True, vad_analyzer=SileroVADAnalyzer(), ), } @@ -76,9 +98,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): llm = AnthropicLLMService( api_key=os.getenv("ANTHROPIC_API_KEY"), model="claude-3-7-sonnet-latest", + params=AnthropicLLMService.InputParams(enable_prompt_caching_beta=True), ) llm.register_function("get_weather", get_weather) - llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) + llm.register_function("get_image", get_image) weather_function = FunctionSchema( name="get_weather", @@ -91,27 +114,44 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): }, required=["location"], ) - restaurant_function = FunctionSchema( - name="get_restaurant_recommendation", - description="Get a restaurant recommendation", + get_image_function = FunctionSchema( + name="get_image", + description="Get an image from the video stream.", properties={ - "location": { + "question": { "type": "string", - "description": "The city and state, e.g. San Francisco, CA", - }, + "description": "The question that the user is asking about the image.", + } }, - required=["location"], + required=["question"], ) - tools = ToolsSchema(standard_tools=[weather_function, restaurant_function]) + tools = ToolsSchema(standard_tools=[weather_function, get_image_function]) - # todo: test with very short initial user message + system_prompt = """\ +You are a helpful assistant who converses with a user and answers questions. Respond concisely to general questions. - # messages = [{"role": "system", - # "content": "You are a helpful assistant who can report the weather in any location in the universe. Respond concisely. Your response will be turned into speech so use only simple words and punctuation."}, - # {"role": "user", - # "content": " Start the conversation by introducing yourself."}] +Your response will be turned into speech so use only simple words and punctuation. - messages = [{"role": "user", "content": "Say 'hello' to start the conversation."}] +You have access to two tools: get_weather and get_image. + +You can respond to questions about the weather using the get_weather tool. + +You can answer questions about the user's video stream using the get_image tool. Some examples of phrases that \ +indicate you should use the get_image tool are: +- What do you see? +- What's in the video? +- Can you describe the video? +- Tell me about what you see. +- Tell me something interesting about what you see. +- What's happening in the video? + +If you need to use a tool, simply use the tool. Do not tell the user the tool you are using. Be brief and concise. + """ + + messages = [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": "Start the conversation by introducing yourself."}, + ] context = LLMContext(messages, tools) context_aggregator = LLMContextAggregatorPair(context) @@ -119,8 +159,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): pipeline = Pipeline( [ transport.input(), # Transport user input - stt, - context_aggregator.user(), # User spoken responses + stt, # STT + context_aggregator.user(), # User speech to text llm, # LLM tts, # TTS transport.output(), # Transport bot output @@ -139,7 +179,13 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): @transport.event_handler("on_client_connected") async def on_client_connected(transport, client): - logger.info(f"Client connected") + logger.info(f"Client connected: {client}") + + await maybe_capture_participant_camera(transport, client) + + global client_id + client_id = get_transport_client_id(transport, client) + # Kick off the conversation. await task.queue_frames([LLMRunFrame()]) diff --git a/src/pipecat/adapters/base_llm_adapter.py b/src/pipecat/adapters/base_llm_adapter.py index 2996c811c..2aae514e1 100644 --- a/src/pipecat/adapters/base_llm_adapter.py +++ b/src/pipecat/adapters/base_llm_adapter.py @@ -39,11 +39,12 @@ class BaseLLMAdapter(ABC, Generic[TLLMInvocationParams]): """ @abstractmethod - def get_llm_invocation_params(self, context: LLMContext) -> TLLMInvocationParams: + def get_llm_invocation_params(self, context: LLMContext, **kwargs) -> TLLMInvocationParams: """Get provider-specific LLM invocation parameters from a universal LLM context. Args: context: The LLM context containing messages, tools, etc. + **kwargs: Additional provider-specific arguments that subclasses can use. Returns: Provider-specific parameters for invoking the LLM. diff --git a/src/pipecat/adapters/services/anthropic_adapter.py b/src/pipecat/adapters/services/anthropic_adapter.py index 45c8bced2..75cbc8b9c 100644 --- a/src/pipecat/adapters/services/anthropic_adapter.py +++ b/src/pipecat/adapters/services/anthropic_adapter.py @@ -6,12 +6,25 @@ """Anthropic LLM adapter for Pipecat.""" -from typing import Any, Dict, List, TypedDict +import copy +import json +from dataclasses import dataclass +from typing import Any, Dict, List, Optional, TypedDict + +from anthropic import NOT_GIVEN, NotGiven +from anthropic.types.message_param import MessageParam +from anthropic.types.tool_union_param import ToolUnionParam +from loguru import logger from pipecat.adapters.base_llm_adapter import BaseLLMAdapter from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.adapters.schemas.tools_schema import ToolsSchema -from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_context import ( + LLMContext, + LLMContextMessage, + LLMSpecificMessage, + LLMStandardMessage, +) class AnthropicLLMInvocationParams(TypedDict): @@ -20,7 +33,9 @@ class AnthropicLLMInvocationParams(TypedDict): This is a placeholder until support for universal LLMContext machinery is added for Anthropic. """ - pass + system: str | NotGiven + messages: List[MessageParam] + tools: List[ToolUnionParam] | NotGiven class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]): @@ -30,20 +45,31 @@ class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]): to the specific format required by Anthropic's Claude models for function calling. """ - def get_llm_invocation_params(self, context: LLMContext) -> AnthropicLLMInvocationParams: + def get_llm_invocation_params( + self, context: LLMContext, enable_prompt_caching: bool + ) -> AnthropicLLMInvocationParams: """Get Anthropic-specific LLM invocation parameters from a universal LLM context. This is a placeholder until support for universal LLMContext machinery is added for Anthropic. Args: context: The LLM context containing messages, tools, etc. + enable_prompt_caching: Whether prompt caching should be enabled. Returns: Dictionary of parameters for invoking Anthropic's LLM API. """ - raise NotImplementedError("Universal LLMContext is not yet supported for Anthropic.") + messages = self._from_universal_context_messages(self._get_messages(context)) + return { + "system": messages.system, + "messages": self._with_cache_control_markers(messages.messages) + if enable_prompt_caching + else messages.messages, + # NOTE: LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN) + "tools": self.from_standard_tools(context.tools), + } - def get_messages_for_logging(self, context) -> List[Dict[str, Any]]: + def get_messages_for_logging(self, context: LLMContext) -> List[Dict[str, Any]]: """Get messages from a universal LLM context in a format ready for logging about Anthropic. Removes or truncates sensitive data like image content for safe logging. @@ -56,7 +82,228 @@ class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]): Returns: List of messages in a format ready for logging about Anthropic. """ - raise NotImplementedError("Universal LLMContext is not yet supported for Anthropic.") + # Get messages in Anthropic's format + messages = self._from_universal_context_messages(self._get_messages(context)).messages + + # Sanitize messages for logging + messages_for_logging = [] + for message in messages: + msg = copy.deepcopy(message) + if "content" in msg: + if isinstance(msg["content"], list): + for item in msg["content"]: + if item["type"] == "image": + item["source"]["data"] = "..." + messages_for_logging.append(msg) + return messages_for_logging + + def _get_messages(self, context: LLMContext) -> List[LLMContextMessage]: + return context.get_messages("anthropic") + + @dataclass + class ConvertedMessages: + """Container for Anthropic-formatted messages converted from universal context.""" + + messages: List[MessageParam] + system: str | NotGiven + + def _from_universal_context_messages( + self, universal_context_messages: List[LLMContextMessage] + ) -> ConvertedMessages: + system = NOT_GIVEN + messages = [] + + # first, map messages using self._from_universal_context_message(m) + try: + messages = [self._from_universal_context_message(m) for m in universal_context_messages] + except Exception as e: + logger.error(f"Error mapping messages: {e}") + + # See if we should pull the system message out of our messages list. + if messages and messages[0]["role"] == "system": + if len(messages) == 1: + # If we have only have a system message in the list, all we can really do + # without introducing too much magic is change the role to "user". + messages[0]["role"] = "user" + else: + # If we have more than one message, we'll pull the system message out of the + # list. + system = messages[0]["content"] + messages.pop(0) + + # Convert any subsequent "system"-role messages to "user"-role + # messages, as Anthropic doesn't support system input messages. + for message in messages: + if message["role"] == "system": + message["role"] = "user" + + # Merge consecutive messages with the same role. + i = 0 + while i < len(messages) - 1: + current_message = messages[i] + next_message = messages[i + 1] + if current_message["role"] == next_message["role"]: + # Convert content to list of dictionaries if it's a string + if isinstance(current_message["content"], str): + current_message["content"] = [ + {"type": "text", "text": current_message["content"]} + ] + if isinstance(next_message["content"], str): + next_message["content"] = [{"type": "text", "text": next_message["content"]}] + # Concatenate the content + current_message["content"].extend(next_message["content"]) + # Remove the next message from the list + messages.pop(i + 1) + else: + i += 1 + + # Avoid empty content in messages + for message in messages: + if isinstance(message["content"], str) and message["content"] == "": + message["content"] = "(empty)" + elif isinstance(message["content"], list) and len(message["content"]) == 0: + message["content"] = [{"type": "text", "text": "(empty)"}] + + return self.ConvertedMessages(messages=messages, system=system) + + def _from_universal_context_message(self, message: LLMContextMessage) -> MessageParam: + if isinstance(message, LLMSpecificMessage): + return message.message + return self._from_standard_message(message) + + def _from_standard_message(self, message: LLMStandardMessage) -> MessageParam: + """Convert standard universal context message to Anthropic format. + + Handles conversion of text content, tool calls, and tool results. + Empty text content is converted to "(empty)". + + Args: + message: Message in standard universal context format. + + Returns: + Message in Anthropic format. + + Examples: + Input standard format:: + + { + "role": "assistant", + "tool_calls": [ + { + "id": "123", + "function": {"name": "search", "arguments": '{"q": "test"}'} + } + ] + } + + Output Anthropic format:: + + { + "role": "assistant", + "content": [ + { + "type": "tool_use", + "id": "123", + "name": "search", + "input": {"q": "test"} + } + ] + } + """ + if message["role"] == "tool": + return { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": message["tool_call_id"], + "content": message["content"], + }, + ], + } + if message.get("tool_calls"): + tc = message["tool_calls"] + ret = {"role": "assistant", "content": []} + for tool_call in tc: + function = tool_call["function"] + arguments = json.loads(function["arguments"]) + new_tool_use = { + "type": "tool_use", + "id": tool_call["id"], + "name": function["name"], + "input": arguments, + } + ret["content"].append(new_tool_use) + return ret + content = message.get("content") + if isinstance(content, str): + # fix empty text + if content == "": + content = "(empty)" + elif isinstance(content, list): + for item in content: + # fix empty text + if item["type"] == "text" and item["text"] == "": + item["text"] = "(empty)" + # handle image_url -> image conversion + if item["type"] == "image_url": + item["type"] = "image" + item["source"] = { + "type": "base64", + "media_type": "image/jpeg", + "data": item["image_url"]["url"].split(",")[1], + } + del item["image_url"] + # In the case where there's a single image in the list (like what + # would result from a UserImageRawFrame), ensure that the image + # comes before text, as recommended by Anthropic docs + # (https://docs.anthropic.com/en/docs/build-with-claude/vision#example-one-image) + image_indices = [i for i, item in enumerate(content) if item["type"] == "image"] + text_indices = [i for i, item in enumerate(content) if item["type"] == "text"] + if len(image_indices) == 1 and text_indices: + img_idx = image_indices[0] + first_txt_idx = text_indices[0] + if img_idx > first_txt_idx: + # Move image before the first text + image_item = content.pop(img_idx) + content.insert(first_txt_idx, image_item) + + return message + + def _with_cache_control_markers(self, messages: List[MessageParam]) -> List[MessageParam]: + """Add cache control markers to messages for prompt caching. + + Args: + messages: List of messages in Anthropic format. + + Returns: + List of messages with cache control markers added. + """ + + def add_cache_control_marker(messages: List[MessageParam], negative_index: int): + if len(messages) > -(negative_index + 1) and messages[negative_index]["role"] == "user": + if isinstance(messages[negative_index]["content"], str): + messages[negative_index]["content"] = [ + {"type": "text", "text": messages[negative_index]["content"]} + ] + messages[negative_index]["content"][-1]["cache_control"] = {"type": "ephemeral"} + + try: + messages_with_markers = copy.deepcopy(messages) + # Add cache control markers to the *last two* user messages. Why? + # - The marker at the last recent user message tells Anthropic to + # cache the prompt up to that point. + # - The marker at the second-to-last user message tells Anthropic + # to look up the cached prompt that goes up to that point (the + # point that *was* the last user message the previous turn). + # If we only added the marker to the last user message, we'd only + # ever be adding to the cache, never looking up from it. + add_cache_control_marker(messages_with_markers, -1) + add_cache_control_marker(messages_with_markers, -3) + return messages_with_markers + except Exception as e: + logger.error(f"Error adding cache control marker: {e}") + return messages_with_markers @staticmethod def _to_anthropic_function_format(function: FunctionSchema) -> Dict[str, Any]: diff --git a/src/pipecat/adapters/services/gemini_adapter.py b/src/pipecat/adapters/services/gemini_adapter.py index ca17791a3..0b649f452 100644 --- a/src/pipecat/adapters/services/gemini_adapter.py +++ b/src/pipecat/adapters/services/gemini_adapter.py @@ -67,7 +67,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): return { "system_instruction": messages.system_instruction, "messages": messages.messages, - # NOTE; LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN) + # NOTE: LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN) "tools": self.from_standard_tools(context.tools), } @@ -192,14 +192,14 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): def _from_standard_message( self, message: LLMStandardMessage, already_have_system_instruction: bool ) -> Content | str: - """Convert universal context message to Google Content object. + """Convert standard universal context message to Google Content object. Handles conversion of text, images, and function calls to Google's format. System instructions are returned as a plain string. Args: - message: Message in universal context format. + message: Message in standard universal context format. already_have_system_instruction: Whether we already have a system instruction Returns: diff --git a/src/pipecat/services/anthropic/llm.py b/src/pipecat/services/anthropic/llm.py index dcf21cf12..e059842ef 100644 --- a/src/pipecat/services/anthropic/llm.py +++ b/src/pipecat/services/anthropic/llm.py @@ -24,7 +24,10 @@ from loguru import logger from PIL import Image from pydantic import BaseModel, Field -from pipecat.adapters.services.anthropic_adapter import AnthropicLLMAdapter +from pipecat.adapters.services.anthropic_adapter import ( + AnthropicLLMAdapter, + AnthropicLLMInvocationParams, +) from pipecat.frames.frames import ( ErrorFrame, Frame, @@ -215,18 +218,18 @@ class AnthropicLLMService(LLMService): The LLM's response as a string, or None if no response is generated. """ messages = [] - system = [] + system = NOT_GIVEN if isinstance(context, LLMContext): - # Future code will be something like this: - # adapter = self.get_llm_adapter() - # params: AnthropicLLMInvocationParams = adapter.get_llm_invocation_params(context) - # messages = params["messages"] - # system = params["system_instruction"] - raise NotImplementedError("Universal LLMContext is not yet supported for Anthropic.") + adapter: AnthropicLLMAdapter = self.get_llm_adapter() + params = adapter.get_llm_invocation_params( + context, enable_prompt_caching=self._settings["enable_prompt_caching_beta"] + ) + messages = params["messages"] + system = params["system"] else: context = AnthropicLLMContext.upgrade_to_anthropic(context) messages = context.messages - system = getattr(context, "system", None) or system_instruction + system = getattr(context, "system", None) or system_instruction or NOT_GIVEN # LLM completion response = await self._client.messages.create( @@ -277,8 +280,31 @@ class AnthropicLLMService(LLMService): assistant = AnthropicAssistantContextAggregator(context, params=assistant_params) return AnthropicContextAggregatorPair(_user=user, _assistant=assistant) + def _get_llm_invocation_params( + self, context: OpenAILLMContext | LLMContext + ) -> AnthropicLLMInvocationParams: + # Universal LLMContext + if isinstance(context, LLMContext): + adapter: AnthropicLLMAdapter = self.get_llm_adapter() + params = adapter.get_llm_invocation_params( + context, enable_prompt_caching=self._settings["enable_prompt_caching_beta"] + ) + return params + + # Anthropic-specific context + messages = ( + context.get_messages_with_cache_control_markers() + if self._settings["enable_prompt_caching_beta"] + else context.messages + ) + return AnthropicLLMInvocationParams( + system=context.system, + messages=messages, + tools=context.tools, + ) + @traced_llm - async def _process_context(self, context: OpenAILLMContext): + async def _process_context(self, context: OpenAILLMContext | LLMContext): # Usage tracking. We track the usage reported by Anthropic in prompt_tokens and # completion_tokens. We also estimate the completion tokens from output text # and use that estimate if we are interrupted, because we almost certainly won't @@ -294,24 +320,22 @@ class AnthropicLLMService(LLMService): await self.push_frame(LLMFullResponseStartFrame()) await self.start_processing_metrics() + params_from_context = self._get_llm_invocation_params(context) + + if isinstance(context, LLMContext): + adapter = self.get_llm_adapter() + context_type_for_logging = "universal" + messages_for_logging = adapter.get_messages_for_logging(context) + else: + context_type_for_logging = "LLM-specific" + messages_for_logging = context.get_messages_for_logging() logger.debug( - f"{self}: Generating chat [{context.system}] | {context.get_messages_for_logging()}" + f"{self}: Generating chat from {context_type_for_logging} context [{params_from_context['system']}] | {messages_for_logging}" ) - messages = context.messages - if self._settings["enable_prompt_caching_beta"]: - messages = context.get_messages_with_cache_control_markers() - - api_call = self._client.messages.create - if self._settings["enable_prompt_caching_beta"]: - api_call = self._client.beta.prompt_caching.messages.create - await self.start_ttfb_metrics() params = { - "tools": context.tools or [], - "system": context.system, - "messages": messages, "model": self.model_name, "max_tokens": self._settings["max_tokens"], "stream": True, @@ -320,9 +344,12 @@ class AnthropicLLMService(LLMService): "top_p": self._settings["top_p"], } + # Messages, system, tools + params.update(params_from_context) + params.update(self._settings["extra"]) - response = await self._create_message_stream(api_call, params) + response = await self._create_message_stream(self._client.messages.create, params) await self.stop_ttfb_metrics() @@ -405,7 +432,10 @@ class AnthropicLLMService(LLMService): prompt_tokens + cache_creation_input_tokens + cache_read_input_tokens ) if total_input_tokens >= 1024: - context.turns_above_cache_threshold += 1 + if hasattr( + context, "turns_above_cache_threshold" + ): # LLMContext doesn't have this attribute + context.turns_above_cache_threshold += 1 await self.run_function_calls(function_calls) @@ -451,7 +481,7 @@ class AnthropicLLMService(LLMService): if isinstance(frame, OpenAILLMContextFrame): context: "AnthropicLLMContext" = AnthropicLLMContext.upgrade_to_anthropic(frame.context) elif isinstance(frame, LLMContextFrame): - raise NotImplementedError("Universal LLMContext is not yet supported for Anthropic.") + context = frame.context elif isinstance(frame, LLMMessagesFrame): context = AnthropicLLMContext.from_messages(frame.messages) elif isinstance(frame, VisionImageRawFrame): diff --git a/src/pipecat/services/google/llm.py b/src/pipecat/services/google/llm.py index dd4dce18c..2dfdf0fbf 100644 --- a/src/pipecat/services/google/llm.py +++ b/src/pipecat/services/google/llm.py @@ -858,8 +858,7 @@ class GoogleLLMService(LLMService): self, context: OpenAILLMContext ) -> AsyncIterator[GenerateContentResponse]: logger.debug( - # f"{self}: Generating chat [{self._system_instruction}] | {context.get_messages_for_logging()}" - f"{self}: Generating chat from OpenAI context {context.get_messages_for_logging()}" + f"{self}: Generating chat from LLM-specific context [{context.system_message}] | {context.get_messages_for_logging()}" ) params = GeminiLLMInvocationParams( @@ -874,13 +873,12 @@ class GoogleLLMService(LLMService): self, context: LLMContext ) -> AsyncIterator[GenerateContentResponse]: adapter = self.get_llm_adapter() - logger.debug( - # f"{self}: Generating chat [{self._system_instruction}] | {context.get_messages_for_logging()}" - f"{self}: Generating chat from universal context {adapter.get_messages_for_logging(context)}" - ) - params: GeminiLLMInvocationParams = adapter.get_llm_invocation_params(context) + logger.debug( + f"{self}: Generating chat from universal context [{params['system_instruction']}] | {adapter.get_messages_for_logging(context)}" + ) + return await self._stream_content(params) @traced_llm diff --git a/src/pipecat/services/openai/base_llm.py b/src/pipecat/services/openai/base_llm.py index 682c0f227..fa90b7cf4 100644 --- a/src/pipecat/services/openai/base_llm.py +++ b/src/pipecat/services/openai/base_llm.py @@ -279,7 +279,7 @@ class BaseOpenAILLMService(LLMService): self, context: OpenAILLMContext ) -> AsyncStream[ChatCompletionChunk]: logger.debug( - f"{self}: Generating chat from OpenAI context {context.get_messages_for_logging()}" + f"{self}: Generating chat from LLM-specific context {context.get_messages_for_logging()}" ) messages: List[ChatCompletionMessageParam] = context.get_messages()