wire up system_instruction in OpenAI, Anthropic and AWS Bedrock

This commit is contained in:
Aleix Conchillo Flaqué
2026-03-04 13:25:07 -08:00
parent f76b8d2982
commit 01f0caf252
3 changed files with 33 additions and 7 deletions

View File

@@ -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

View File

@@ -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

View File

@@ -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(