diff --git a/src/pipecat/services/anthropic/llm.py b/src/pipecat/services/anthropic/llm.py index 03190ef99..755ca6400 100644 --- a/src/pipecat/services/anthropic/llm.py +++ b/src/pipecat/services/anthropic/llm.py @@ -17,7 +17,7 @@ import io import json import re from dataclasses import dataclass, field -from typing import Any, ClassVar, Dict, List, Literal, Optional, Union +from typing import Any, Dict, List, Literal, Optional, Union import httpx from loguru import logger @@ -38,7 +38,6 @@ from pipecat.frames.frames import ( LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMMessagesFrame, - LLMTextFrame, LLMThoughtEndFrame, LLMThoughtStartFrame, LLMThoughtTextFrame, @@ -219,6 +218,7 @@ class AnthropicLLMService(LLMService): client=None, retry_timeout_secs: Optional[float] = 5.0, retry_on_timeout: Optional[bool] = False, + system_instruction: Optional[str] = None, **kwargs, ): """Initialize the Anthropic LLM service. @@ -230,6 +230,7 @@ class AnthropicLLMService(LLMService): client: Optional custom Anthropic client instance. retry_timeout_secs: Request timeout in seconds for retry logic. retry_on_timeout: Whether to retry the request once if it times out. + system_instruction: Optional system instruction to use as the system prompt. **kwargs: Additional arguments passed to parent LLMService. """ params = params or AnthropicLLMService.InputParams() @@ -265,6 +266,9 @@ class AnthropicLLMService(LLMService): ) # if the client is provided, use it and remove it, otherwise create a new one self._retry_timeout_secs = retry_timeout_secs self._retry_on_timeout = retry_on_timeout + self._system_instruction = system_instruction + if self._system_instruction: + logger.debug(f"{self}: Using system instruction: {self._system_instruction}") def can_generate_metrics(self) -> bool: """Check if this service can generate usage metrics. @@ -395,9 +399,11 @@ class AnthropicLLMService(LLMService): # Universal LLMContext if isinstance(context, LLMContext): adapter: AnthropicLLMAdapter = self.get_llm_adapter() - params = adapter.get_llm_invocation_params( + params: AnthropicLLMInvocationParams = adapter.get_llm_invocation_params( context, enable_prompt_caching=self._settings.enable_prompt_caching ) + if self._system_instruction: + params["system"] = self._system_instruction return params # Anthropic-specific context diff --git a/src/pipecat/services/aws/llm.py b/src/pipecat/services/aws/llm.py index 540ac4a8e..31f8085bc 100644 --- a/src/pipecat/services/aws/llm.py +++ b/src/pipecat/services/aws/llm.py @@ -19,7 +19,7 @@ import json import os import re from dataclasses import dataclass, field -from typing import Any, ClassVar, Dict, List, Optional +from typing import Any, Dict, List, Optional from loguru import logger from PIL import Image @@ -39,7 +39,6 @@ from pipecat.frames.frames import ( LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMMessagesFrame, - LLMTextFrame, UserImageRawFrame, ) from pipecat.metrics.metrics import LLMTokenUsage @@ -781,6 +780,7 @@ class AWSBedrockLLMService(LLMService): client_config: Optional[Config] = None, retry_timeout_secs: Optional[float] = 5.0, retry_on_timeout: Optional[bool] = False, + system_instruction: Optional[str] = None, **kwargs, ): """Initialize the AWS Bedrock LLM service. @@ -795,6 +795,7 @@ class AWSBedrockLLMService(LLMService): client_config: Custom boto3 client configuration. retry_timeout_secs: Request timeout in seconds for retry logic. retry_on_timeout: Whether to retry the request once if it times out. + system_instruction: Optional system instruction to use as the system prompt. **kwargs: Additional arguments passed to parent LLMService. """ params = params or AWSBedrockLLMService.InputParams() @@ -840,8 +841,11 @@ class AWSBedrockLLMService(LLMService): self._retry_timeout_secs = retry_timeout_secs self._retry_on_timeout = retry_on_timeout + self._system_instruction = system_instruction logger.info(f"Using AWS Bedrock model: {model}") + if self._system_instruction: + logger.debug(f"{self}: Using system instruction: {self._system_instruction}") def can_generate_metrics(self) -> bool: """Check if the service can generate usage metrics. @@ -1019,7 +1023,9 @@ class AWSBedrockLLMService(LLMService): # Universal LLMContext if isinstance(context, LLMContext): adapter: AWSBedrockLLMAdapter = self.get_llm_adapter() - params = adapter.get_llm_invocation_params(context) + params: AWSBedrockLLMInvocationParams = adapter.get_llm_invocation_params(context) + if self._system_instruction: + params["system"] = [{"text": self._system_instruction}] return params # AWS Bedrock-specific context diff --git a/src/pipecat/services/openai/base_llm.py b/src/pipecat/services/openai/base_llm.py index 40a2672f8..ed012d0df 100644 --- a/src/pipecat/services/openai/base_llm.py +++ b/src/pipecat/services/openai/base_llm.py @@ -11,7 +11,7 @@ import base64 import json from contextlib import asynccontextmanager from dataclasses import dataclass, field -from typing import Any, ClassVar, Dict, List, Mapping, Optional +from typing import Any, Dict, List, Mapping, Optional import httpx from loguru import logger @@ -117,6 +117,7 @@ class BaseOpenAILLMService(LLMService): params: Optional[InputParams] = None, retry_timeout_secs: Optional[float] = 5.0, retry_on_timeout: Optional[bool] = False, + system_instruction: Optional[str] = None, **kwargs, ): """Initialize the BaseOpenAILLMService. @@ -131,6 +132,7 @@ class BaseOpenAILLMService(LLMService): params: Input parameters for model configuration and behavior. retry_timeout_secs: Request timeout in seconds. Defaults to 5.0 seconds. retry_on_timeout: Whether to retry the request once if it times out. + system_instruction: Optional system instruction to prepend to messages. **kwargs: Additional arguments passed to the parent LLMService. """ params = params or BaseOpenAILLMService.InputParams() @@ -155,6 +157,7 @@ class BaseOpenAILLMService(LLMService): ) self._retry_timeout_secs = retry_timeout_secs self._retry_on_timeout = retry_on_timeout + self._system_instruction = system_instruction self._full_model_name: str = "" self._client = self.create_client( api_key=api_key, @@ -165,6 +168,9 @@ class BaseOpenAILLMService(LLMService): **kwargs, ) + if self._system_instruction: + logger.debug(f"{self}: Using system instruction: {self._system_instruction}") + def create_client( self, api_key=None, @@ -285,6 +291,14 @@ class BaseOpenAILLMService(LLMService): params.update(params_from_context) params.update(self._settings.extra) + + # Prepend system instruction if set + if self._system_instruction: + messages = params.get("messages", []) + params["messages"] = [ + {"role": "system", "content": self._system_instruction} + ] + messages + return params async def run_inference(