wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
0d1b834770f747b977e8b87b7b7b9cf84910987c
pipecat/tests/test_get_llm_invocation_params.py
Paul Kompfner 0d1b834770 Add developer message support to realtime adapters 
OpenAI Realtime, Grok Realtime, and AWS Nova Sonic adapters now convert
"developer" role messages to "user" (consistent with all other non-OpenAI
adapters). Previously these messages were silently dropped. Adds starter
unit tests for all three realtime adapters.
2026-03-24 16:05:15 -04:00

2234 lines
99 KiB
Python
Raw Blame History

#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""
Unit tests for LLM adapters' get_llm_invocation_params() method.
These tests focus specifically on the "messages" field generation for different adapters, ensuring:
For OpenAI adapter:
1. LLMStandardMessage objects are passed through unchanged
2. LLMSpecificMessage objects with llm='openai' are included and others are filtered out
3. Complex message structures (like multi-part content) are preserved
4. System instructions are preserved throughout messages at any position
5. system_instruction is prepended as a system message, with conflict warnings
6. Developer messages pass through without triggering warnings
For Gemini adapter:
1. LLMStandardMessage objects are converted to Gemini Content format
2. LLMSpecificMessage objects with llm='google' are included and others are filtered out
3. Complex message structures (image, audio, multi-text) are converted to appropriate Gemini format
4. System messages are extracted as system_instruction (without duplication)
5. Single system instruction is converted to user message when no other messages exist
6. Multiple system instructions: first extracted, later ones converted to user messages
7. system_instruction overrides context system message, with conflict warnings
8. Developer messages are converted to user
For Anthropic adapter:
1. LLMStandardMessage objects are converted to Anthropic MessageParam format
2. LLMSpecificMessage objects with llm='anthropic' are included and others are filtered out
3. Complex message structures (image, multi-text) are converted to appropriate Anthropic format
4. System messages: first extracted as system parameter, later ones converted to user messages
5. Consecutive messages with same role are merged into multi-content-block messages
6. Empty text content is converted to "(empty)"
7. system_instruction overrides context system message, with conflict warnings
8. Developer messages are converted to user
For AWS Bedrock adapter:
1. LLMStandardMessage objects are converted to AWS Bedrock format
2. LLMSpecificMessage objects with llm='aws' are included and others are filtered out
3. Complex message structures (image, multi-text) are converted to appropriate AWS Bedrock format
4. System messages: first extracted as system parameter, later ones converted to user messages
5. Consecutive messages with same role are merged into multi-content-block messages
6. Empty text content is converted to "(empty)"
7. system_instruction overrides context system message, with conflict warnings
8. Developer messages are converted to user
For OpenAI Responses adapter:
1. LLMContext messages are converted to Responses API input items
2. System and developer role messages are converted to developer role
3. Assistant tool_calls produce function_call input items
4. Tool messages produce function_call_output input items
5. Multimodal content conversion (text -> input_text, image_url -> input_image)
6. Tools schema flattening (nested function dict -> flat format)
7. system_instruction sets instructions (or becomes developer message if input is empty)
8. Developer messages pass through as developer role without triggering warnings
For BaseLLMAdapter helpers:
1. _extract_initial_system: system extraction and conversion logic
2. _resolve_system_instruction: conflict resolution between context and settings
"""
import unittest
from unittest.mock import patch
from google.genai.types import Content, Part
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.adapters.services.anthropic_adapter import AnthropicLLMAdapter
from pipecat.adapters.services.aws_nova_sonic_adapter import AWSNovaSonicLLMAdapter
from pipecat.adapters.services.bedrock_adapter import AWSBedrockLLMAdapter
from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter
from pipecat.adapters.services.grok_realtime_adapter import GrokRealtimeLLMAdapter
from pipecat.adapters.services.open_ai_adapter import OpenAILLMAdapter
from pipecat.adapters.services.open_ai_realtime_adapter import OpenAIRealtimeLLMAdapter
from pipecat.adapters.services.open_ai_responses_adapter import OpenAIResponsesLLMAdapter
from pipecat.adapters.services.perplexity_adapter import PerplexityLLMAdapter
from pipecat.processors.aggregators.llm_context import (
LLMContext,
LLMStandardMessage,
)
class TestOpenAIGetLLMInvocationParams(unittest.TestCase):
def setUp(self) -> None:
"""Sets up a common adapter instance for all tests."""
self.adapter = OpenAILLMAdapter()
def test_standard_messages_passed_through_unchanged(self):
"""Test that LLMStandardMessage objects are passed through unchanged to OpenAI params."""
# Create standard messages (OpenAI format)
standard_messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Hello, how are you?"},
{"role": "assistant", "content": "I'm doing well, thank you for asking!"},
]
# Create context with these messages
context = LLMContext(messages=standard_messages)
# Get invocation params
params = self.adapter.get_llm_invocation_params(context)
# Verify messages are passed through unchanged
self.assertEqual(params["messages"], standard_messages)
self.assertEqual(len(params["messages"]), 3)
# Verify content matches exactly
self.assertEqual(params["messages"][0]["content"], "You are a helpful assistant.")
self.assertEqual(params["messages"][1]["content"], "Hello, how are you?")
self.assertEqual(params["messages"][2]["content"], "I'm doing well, thank you for asking!")
def test_llm_specific_message_filtering(self):
"""Test that OpenAI-specific messages are included and others are filtered out."""
# Create messages with different LLM-specific ones
messages = [
{"role": "system", "content": "You are a helpful assistant."},
AnthropicLLMAdapter().create_llm_specific_message(
{"role": "user", "content": "Anthropic specific message"}
),
GeminiLLMAdapter().create_llm_specific_message(
{"role": "user", "content": "Gemini specific message"}
),
{"role": "user", "content": "Standard user message"},
self.adapter.create_llm_specific_message(
{"role": "assistant", "content": "OpenAI specific response"}
),
]
# Create context with these messages
context = LLMContext(messages=messages)
# Get invocation params
params = self.adapter.get_llm_invocation_params(context)
# Should only include standard messages and OpenAI-specific ones
# (3 total: system, standard user, openai assistant)
self.assertEqual(len(params["messages"]), 3)
# Verify the correct messages are included
self.assertEqual(params["messages"][0]["content"], "You are a helpful assistant.")
self.assertEqual(params["messages"][1]["content"], "Standard user message")
self.assertEqual(
params["messages"][2], {"role": "assistant", "content": "OpenAI specific response"}
)
def test_complex_message_content_preserved(self):
"""Test that complex message content (like multi-part messages) is preserved."""
# Create a message with complex content structure (text + image)
complex_image_message = {
"role": "user",
"content": [
{"type": "text", "text": "What's in this image?"},
{
"type": "image_url",
"image_url": {"url": "data:image/jpeg;base64,/9j/4AAQSkZJRgABAQAAAQABAAD..."},
},
],
}
# Create a message with multiple text blocks
multi_text_message = {
"role": "assistant",
"content": [
{"type": "text", "text": "Let me analyze this step by step:"},
{"type": "text", "text": "1. First, I'll examine the visual elements"},
{"type": "text", "text": "2. Then I'll provide my conclusions"},
],
}
messages = [
{"role": "system", "content": "You are a helpful assistant that can analyze images."},
complex_image_message,
multi_text_message,
]
# Create context with these messages
context = LLMContext(messages=messages)
# Get invocation params
params = self.adapter.get_llm_invocation_params(context)
# Verify complex content is preserved
self.assertEqual(len(params["messages"]), 3)
self.assertEqual(params["messages"][1], complex_image_message)
self.assertEqual(params["messages"][2], multi_text_message)
# Verify the image message structure is maintained
image_content = params["messages"][1]["content"]
self.assertIsInstance(image_content, list)
self.assertEqual(len(image_content), 2)
self.assertEqual(image_content[0]["type"], "text")
self.assertEqual(image_content[1]["type"], "image_url")
# Verify the multi-text message structure is maintained
text_content = params["messages"][2]["content"]
self.assertIsInstance(text_content, list)
self.assertEqual(len(text_content), 3)
for i, text_block in enumerate(text_content):
self.assertEqual(text_block["type"], "text")
self.assertEqual(text_content[0]["text"], "Let me analyze this step by step:")
self.assertEqual(text_content[1]["text"], "1. First, I'll examine the visual elements")
self.assertEqual(text_content[2]["text"], "2. Then I'll provide my conclusions")
def test_system_instructions_preserved_throughout_messages(self):
"""Test that OpenAI adapter preserves system instructions sprinkled throughout messages."""
# Create messages with system instructions at different positions
messages = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Hello!"},
{"role": "assistant", "content": "Hi there!"},
{"role": "system", "content": "Remember to be concise."},
{"role": "user", "content": "Tell me about Python."},
{"role": "system", "content": "Use simple language."},
{"role": "assistant", "content": "Python is a programming language."},
]
# Create context with these messages
context = LLMContext(messages=messages)
# Get invocation params
params = self.adapter.get_llm_invocation_params(context)
# OpenAI should preserve all messages unchanged, including multiple system messages
self.assertEqual(len(params["messages"]), 7)
# Verify system messages are preserved at their original positions
self.assertEqual(params["messages"][0]["role"], "system")
self.assertEqual(params["messages"][0]["content"], "You are a helpful assistant.")
self.assertEqual(params["messages"][3]["role"], "system")
self.assertEqual(params["messages"][3]["content"], "Remember to be concise.")
self.assertEqual(params["messages"][5]["role"], "system")
self.assertEqual(params["messages"][5]["content"], "Use simple language.")
# Verify other messages remain unchanged
self.assertEqual(params["messages"][1]["role"], "user")
self.assertEqual(params["messages"][2]["role"], "assistant")
self.assertEqual(params["messages"][4]["role"], "user")
self.assertEqual(params["messages"][6]["role"], "assistant")
def test_system_instruction_only(self):
"""system_instruction alone is prepended as a system message."""
messages: list[LLMStandardMessage] = [
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context, system_instruction="Be helpful.")
self.assertEqual(params["messages"][0]["role"], "system")
self.assertEqual(params["messages"][0]["content"], "Be helpful.")
self.assertEqual(params["messages"][1]["role"], "user")
def test_initial_system_message_only(self):
"""Initial system message without system_instruction passes through."""
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(len(params["messages"]), 2)
self.assertEqual(params["messages"][0]["role"], "system")
self.assertEqual(params["messages"][0]["content"], "You are helpful.")
def test_both_system_instruction_and_system_message_warns(self):
"""system_instruction + initial system message warns but allows both."""
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger:
params = self.adapter.get_llm_invocation_params(
context, system_instruction="Be concise."
)
mock_logger.warning.assert_called_once()
warning_msg = mock_logger.warning.call_args[0][0]
self.assertIn("may be unintended", warning_msg)
# Both are present: prepended system_instruction + original system message
self.assertEqual(params["messages"][0]["content"], "Be concise.")
self.assertEqual(params["messages"][1]["content"], "You are helpful.")
def test_both_system_instruction_and_developer_message_no_warning(self):
"""system_instruction + initial developer message does NOT warn."""
messages: list[LLMStandardMessage] = [
{"role": "developer", "content": "Extra context."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger:
params = self.adapter.get_llm_invocation_params(
context, system_instruction="Be concise."
)
mock_logger.warning.assert_not_called()
# system_instruction prepended, developer message stays in messages
self.assertEqual(params["messages"][0]["content"], "Be concise.")
self.assertEqual(params["messages"][1]["role"], "developer")
def test_warning_fires_only_once(self):
"""Conflict warning fires only once per adapter instance."""
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger:
self.adapter.get_llm_invocation_params(context, system_instruction="Be concise.")
self.adapter.get_llm_invocation_params(context, system_instruction="Be concise.")
mock_logger.warning.assert_called_once()
class TestGeminiGetLLMInvocationParams(unittest.TestCase):
def setUp(self) -> None:
"""Sets up a common adapter instance for all tests."""
self.adapter = GeminiLLMAdapter()
def test_standard_messages_converted_to_gemini_format(self):
"""Test that LLMStandardMessage objects are converted to Gemini Content format."""
# Create standard messages (OpenAI format)
standard_messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Hello, how are you?"},
{"role": "assistant", "content": "I'm doing well, thank you for asking!"},
]
# Create context with these messages
context = LLMContext(messages=standard_messages)
# Get invocation params
params = self.adapter.get_llm_invocation_params(context)
# Verify system instruction is extracted
self.assertEqual(params["system_instruction"], "You are a helpful assistant.")
# Verify messages are converted to Gemini format (2 messages: user + model)
self.assertEqual(len(params["messages"]), 2)
# Check first message (user)
user_msg = params["messages"][0]
self.assertIsInstance(user_msg, Content)
self.assertEqual(user_msg.role, "user")
self.assertEqual(len(user_msg.parts), 1)
self.assertEqual(user_msg.parts[0].text, "Hello, how are you?")
# Check second message (assistant -> model)
model_msg = params["messages"][1]
self.assertIsInstance(model_msg, Content)
self.assertEqual(model_msg.role, "model")
self.assertEqual(len(model_msg.parts), 1)
self.assertEqual(model_msg.parts[0].text, "I'm doing well, thank you for asking!")
def test_llm_specific_message_filtering(self):
"""Test that Gemini-specific messages are included and others are filtered out."""
# Create messages with different LLM-specific ones
messages = [
{"role": "system", "content": "You are a helpful assistant."},
OpenAILLMAdapter().create_llm_specific_message(
{"role": "user", "content": "OpenAI specific message"}
),
AnthropicLLMAdapter().create_llm_specific_message(
{"role": "user", "content": "Anthropic specific message"}
),
{"role": "user", "content": "Standard user message"},
self.adapter.create_llm_specific_message(
Content(role="model", parts=[Part(text="Gemini specific response")]),
),
]
# Create context with these messages
context = LLMContext(messages=messages)
# Get invocation params
params = self.adapter.get_llm_invocation_params(context)
# Should only include standard messages and Gemini-specific ones
# (2 total: converted standard user + gemini model)
self.assertEqual(len(params["messages"]), 2)
# Verify system instruction
self.assertEqual(params["system_instruction"], "You are a helpful assistant.")
# Verify the correct messages are included
self.assertEqual(params["messages"][0].role, "user")
self.assertEqual(params["messages"][0].parts[0].text, "Standard user message")
self.assertEqual(params["messages"][1].role, "model")
self.assertEqual(params["messages"][1].parts[0].text, "Gemini specific response")
def test_complex_message_content_preserved(self):
"""Test that complex message content (like multi-part messages) is preserved and converted.
This test covers image, audio, and multi-text content conversion to Gemini format.
"""
# Create a message with complex content structure (text + image)
# Using a minimal valid base64 image data
complex_image_message = {
"role": "user",
"content": [
{"type": "text", "text": "What's in this image?"},
{
"type": "image_url",
"image_url": {
"url": "data:image/jpeg;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg=="
},
},
],
}
# Create a message with multiple text blocks
multi_text_message = {
"role": "assistant",
"content": [
{"type": "text", "text": "Let me analyze this step by step:"},
{"type": "text", "text": "1. First, I'll examine the visual elements"},
{"type": "text", "text": "2. Then I'll provide my conclusions"},
],
}
# Create a message with audio input (text + audio)
# Using a minimal valid base64 audio data (16 bytes of WAV header)
audio_message = {
"role": "user",
"content": [
{"type": "text", "text": "Can you transcribe this audio?"},
{
"type": "input_audio",
"input_audio": {
"data": "UklGRiQAAABXQVZFZm10IBAAAAABAAEARKwAAIhYAQACABAAZGF0YQAAAAA=",
"format": "wav",
},
},
],
}
messages = [
{
"role": "system",
"content": "You are a helpful assistant that can analyze images and audio.",
},
complex_image_message,
multi_text_message,
audio_message,
]
# Create context with these messages
context = LLMContext(messages=messages)
# Get invocation params
params = self.adapter.get_llm_invocation_params(context)
# Verify system instruction
self.assertEqual(
params["system_instruction"],
"You are a helpful assistant that can analyze images and audio.",
)
# Verify complex content is converted to Gemini format
# Note: Gemini adapter may add system instruction back as user message in some cases
self.assertGreaterEqual(len(params["messages"]), 3)
# Find the different message types
user_with_image = None
model_with_text = None
user_with_audio = None
for msg in params["messages"]:
if msg.role == "user" and len(msg.parts) == 2:
# Check if it's image or audio based on the text content
if hasattr(msg.parts[0], "text") and "image" in msg.parts[0].text:
user_with_image = msg
elif hasattr(msg.parts[0], "text") and "audio" in msg.parts[0].text:
user_with_audio = msg
elif msg.role == "model" and len(msg.parts) == 3:
model_with_text = msg
# Verify the image message structure is converted properly
self.assertIsNotNone(user_with_image, "Should have user message with image")
self.assertEqual(len(user_with_image.parts), 2)
# First part should be text
self.assertEqual(user_with_image.parts[0].text, "What's in this image?")
# Second part should be image data (converted to Blob)
self.assertIsNotNone(user_with_image.parts[1].inline_data)
self.assertEqual(user_with_image.parts[1].inline_data.mime_type, "image/jpeg")
# Verify the audio message structure is converted properly
self.assertIsNotNone(user_with_audio, "Should have user message with audio")
self.assertEqual(len(user_with_audio.parts), 2)
# First part should be text
self.assertEqual(user_with_audio.parts[0].text, "Can you transcribe this audio?")
# Second part should be audio data (converted to Blob)
self.assertIsNotNone(user_with_audio.parts[1].inline_data)
self.assertEqual(user_with_audio.parts[1].inline_data.mime_type, "audio/wav")
# Verify the multi-text message structure is converted properly
self.assertIsNotNone(model_with_text, "Should have model message with multi-text")
self.assertEqual(len(model_with_text.parts), 3)
# All parts should be text
expected_texts = [
"Let me analyze this step by step:",
"1. First, I'll examine the visual elements",
"2. Then I'll provide my conclusions",
]
for i, expected_text in enumerate(expected_texts):
self.assertEqual(model_with_text.parts[i].text, expected_text)
def test_single_system_instruction_converted_to_user(self):
"""Test that when there's only a system instruction, it gets converted to user message."""
# Create context with only a system message
messages = [
{"role": "system", "content": "You are a helpful assistant."},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
# When there's only one message, it's converted to user in-place (not extracted)
# so system_instruction is None
self.assertIsNone(params["system_instruction"])
# The system message should be converted to a user message
self.assertEqual(len(params["messages"]), 1)
self.assertEqual(params["messages"][0].role, "user")
self.assertEqual(params["messages"][0].parts[0].text, "You are a helpful assistant.")
def test_multiple_system_instructions_handling(self):
"""Test that first system instruction is extracted, later ones converted to user messages."""
# Create messages with multiple system instructions
messages = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Hello!"},
{"role": "assistant", "content": "Hi there!"},
{"role": "system", "content": "Remember to be concise."},
{"role": "user", "content": "Tell me about Python."},
{"role": "system", "content": "Use simple language."},
{"role": "assistant", "content": "Python is a programming language."},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
# First system instruction should be extracted
self.assertEqual(params["system_instruction"], "You are a helpful assistant.")
# Should have 6 messages (original 7 minus 1 system instruction that was extracted)
self.assertEqual(len(params["messages"]), 6)
# Find the converted system messages (should be user role now)
converted_system_messages = []
for msg in params["messages"]:
if msg.role == "user" and (
msg.parts[0].text == "Remember to be concise."
or msg.parts[0].text == "Use simple language."
):
converted_system_messages.append(msg.parts[0].text)
# Should have 2 converted system messages
self.assertEqual(len(converted_system_messages), 2)
self.assertIn("Remember to be concise.", converted_system_messages)
self.assertIn("Use simple language.", converted_system_messages)
# Verify that regular user and assistant messages are preserved
user_messages = [msg for msg in params["messages"] if msg.role == "user"]
model_messages = [msg for msg in params["messages"] if msg.role == "model"]
# Should have 4 user messages: 2 original + 2 converted from system
self.assertEqual(len(user_messages), 4)
# Should have 2 model messages (converted from assistant)
self.assertEqual(len(model_messages), 2)
def test_system_instruction_only(self):
"""system_instruction alone becomes the system_instruction parameter."""
messages: list[LLMStandardMessage] = [
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context, system_instruction="Be helpful.")
self.assertEqual(params["system_instruction"], "Be helpful.")
def test_initial_system_message_only(self):
"""Initial system message is extracted as system_instruction."""
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(params["system_instruction"], "You are helpful.")
self.assertEqual(len(params["messages"]), 1)
def test_initial_developer_message_becomes_user(self):
"""Initial developer message without system_instruction becomes user, not system_instruction."""
messages: list[LLMStandardMessage] = [
{"role": "developer", "content": "Extra context."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertIsNone(params["system_instruction"])
self.assertEqual(len(params["messages"]), 2)
self.assertEqual(params["messages"][0].role, "user")
self.assertEqual(params["messages"][0].parts[0].text, "Extra context.")
def test_both_system_instruction_and_system_message_warns(self):
"""system_instruction + initial system message warns and uses system_instruction."""
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger:
params = self.adapter.get_llm_invocation_params(
context, system_instruction="Be concise."
)
mock_logger.warning.assert_called_once()
self.assertEqual(params["system_instruction"], "Be concise.")
def test_both_system_instruction_and_developer_message_no_warning(self):
"""system_instruction + initial developer message: no warning, developer becomes user."""
messages: list[LLMStandardMessage] = [
{"role": "developer", "content": "Extra context."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger:
params = self.adapter.get_llm_invocation_params(
context, system_instruction="Be concise."
)
mock_logger.warning.assert_not_called()
self.assertEqual(params["system_instruction"], "Be concise.")
def test_non_initial_system_message_not_extracted(self):
"""Non-initial system message is converted to user, not extracted as system instruction."""
messages: list[LLMStandardMessage] = [
{"role": "user", "content": "Hello"},
{"role": "system", "content": "Late system message"},
{"role": "user", "content": "How are you?"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
# No system instruction should be extracted from non-initial position
self.assertIsNone(params["system_instruction"])
# The system message should have been converted to user role in the Gemini Content
# (we check that 3 messages are present, meaning no extraction happened)
self.assertEqual(len(params["messages"]), 3)
def test_subsequent_developer_messages_converted_to_user(self):
"""Subsequent developer messages are converted to user role."""
messages: list[LLMStandardMessage] = [
{"role": "user", "content": "Hello"},
{"role": "developer", "content": "More instructions"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(len(params["messages"]), 2)
# Second message (developer) should be converted to user in Google format
self.assertEqual(params["messages"][1].role, "user")
class TestAnthropicGetLLMInvocationParams(unittest.TestCase):
def setUp(self) -> None:
"""Sets up a common adapter instance for all tests."""
self.adapter = AnthropicLLMAdapter()
def test_standard_messages_converted_to_anthropic_format(self):
"""Test that LLMStandardMessage objects are converted to Anthropic MessageParam format."""
# Create standard messages
standard_messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Hello, how are you?"},
{"role": "assistant", "content": "I'm doing well, thank you!"},
]
# Create context
context = LLMContext(messages=standard_messages)
# Get invocation params
params = self.adapter.get_llm_invocation_params(context, enable_prompt_caching=False)
# Verify system instruction is extracted
self.assertEqual(params["system"], "You are a helpful assistant.")
# Verify messages are in the params (2 messages after system extraction)
self.assertIn("messages", params)
self.assertEqual(len(params["messages"]), 2)
# Check first message (user)
user_msg = params["messages"][0]
self.assertEqual(user_msg["role"], "user")
self.assertEqual(user_msg["content"], "Hello, how are you?")
# Check second message (assistant)
assistant_msg = params["messages"][1]
self.assertEqual(assistant_msg["role"], "assistant")
self.assertEqual(assistant_msg["content"], "I'm doing well, thank you!")
def test_llm_specific_message_filtering(self):
"""Test that Anthropic-specific messages are included and others are filtered out."""
# Create anthropic-specific message content
anthropic_message_content = {
"role": "user",
"content": [
{"type": "text", "text": "Hello"},
{
"type": "image",
"source": {"type": "base64", "media_type": "image/jpeg", "data": "fake_data"},
},
],
}
messages = [
{"role": "user", "content": "Standard message"},
OpenAILLMAdapter().create_llm_specific_message(
{"role": "user", "content": "OpenAI specific"}
),
GeminiLLMAdapter().create_llm_specific_message(
{"role": "user", "content": "Google specific"}
),
self.adapter.create_llm_specific_message(anthropic_message_content),
{"role": "assistant", "content": "Response"},
]
# Create context
context = LLMContext(messages=messages)
# Get invocation params
params = self.adapter.get_llm_invocation_params(context, enable_prompt_caching=False)
# Should only have 2 messages after merging consecutive user messages: merged user + standard response
# (openai and google specific filtered out, standard + anthropic-specific merged)
self.assertEqual(len(params["messages"]), 2)
# First message: merged user message (standard + anthropic-specific)
user_msg = params["messages"][0]
self.assertEqual(user_msg["role"], "user")
self.assertIsInstance(user_msg["content"], list)
# Should have 3 content blocks: standard text + anthropic text + anthropic image
self.assertEqual(len(user_msg["content"]), 3)
self.assertEqual(user_msg["content"][0]["type"], "text")
self.assertEqual(user_msg["content"][0]["text"], "Standard message")
self.assertEqual(user_msg["content"][1]["type"], "text")
self.assertEqual(user_msg["content"][1]["text"], "Hello")
self.assertEqual(user_msg["content"][2]["type"], "image")
# Second message: standard response
self.assertEqual(params["messages"][1]["content"], "Response")
def test_consecutive_same_role_messages_merged(self):
"""Test that consecutive messages with the same role are merged into multi-content blocks."""
messages = [
{"role": "user", "content": "First user message"},
{"role": "user", "content": "Second user message"},
{"role": "user", "content": "Third user message"},
{"role": "assistant", "content": "First assistant message"},
{"role": "assistant", "content": "Second assistant message"},
]
# Create context
context = LLMContext(messages=messages)
# Get invocation params
params = self.adapter.get_llm_invocation_params(context, enable_prompt_caching=False)
# Should have 2 messages after merging (1 user, 1 assistant)
self.assertEqual(len(params["messages"]), 2)
# Check merged user message
user_msg = params["messages"][0]
self.assertEqual(user_msg["role"], "user")
self.assertIsInstance(user_msg["content"], list)
self.assertEqual(len(user_msg["content"]), 3)
self.assertEqual(user_msg["content"][0]["type"], "text")
self.assertEqual(user_msg["content"][0]["text"], "First user message")
self.assertEqual(user_msg["content"][1]["type"], "text")
self.assertEqual(user_msg["content"][1]["text"], "Second user message")
self.assertEqual(user_msg["content"][2]["type"], "text")
self.assertEqual(user_msg["content"][2]["text"], "Third user message")
# Check merged assistant message
assistant_msg = params["messages"][1]
self.assertEqual(assistant_msg["role"], "assistant")
self.assertIsInstance(assistant_msg["content"], list)
self.assertEqual(len(assistant_msg["content"]), 2)
self.assertEqual(assistant_msg["content"][0]["type"], "text")
self.assertEqual(assistant_msg["content"][0]["text"], "First assistant message")
self.assertEqual(assistant_msg["content"][1]["type"], "text")
self.assertEqual(assistant_msg["content"][1]["text"], "Second assistant message")
def test_empty_text_converted_to_empty_placeholder(self):
"""Test that empty text content is converted to "(empty)" string."""
messages = [
{"role": "user", "content": ""}, # Empty string
{
"role": "assistant",
"content": [
{"type": "text", "text": ""}, # Empty text in list content
{"type": "text", "text": "Valid text"},
],
},
]
# Create context
context = LLMContext(messages=messages)
# Get invocation params
params = self.adapter.get_llm_invocation_params(context, enable_prompt_caching=False)
# Check that empty string content was converted
user_msg = params["messages"][0]
self.assertEqual(user_msg["content"], "(empty)")
# Check that empty text in list content was converted
assistant_msg = params["messages"][1]
self.assertIsInstance(assistant_msg["content"], list)
self.assertEqual(assistant_msg["content"][0]["text"], "(empty)")
self.assertEqual(assistant_msg["content"][1]["text"], "Valid text")
def test_complex_message_content_preserved(self):
"""Test that complex message structures (text + image) are properly converted to Anthropic format."""
# Create a complex message with both text and image content
complex_message = {
"role": "user",
"content": [
{"type": "text", "text": "What do you see in this image?"},
{
"type": "image_url",
"image_url": {"url": "data:image/jpeg;base64,fake_image_data"},
},
{"type": "text", "text": "Please describe it in detail."},
],
}
messages = [
complex_message,
{"role": "assistant", "content": "I can see the image clearly."},
]
# Create context
context = LLMContext(messages=messages)
# Get invocation params
params = self.adapter.get_llm_invocation_params(context, enable_prompt_caching=False)
# Verify complex message structure is preserved and converted
self.assertEqual(len(params["messages"]), 2)
user_msg = params["messages"][0]
self.assertEqual(user_msg["role"], "user")
self.assertIsInstance(user_msg["content"], list)
self.assertEqual(len(user_msg["content"]), 3)
# Note: Anthropic adapter reorders single images to come before text, as per Anthropic docs
# Check image part (should be moved to first position and converted from image_url to image)
self.assertEqual(user_msg["content"][0]["type"], "image")
self.assertIn("source", user_msg["content"][0])
self.assertEqual(user_msg["content"][0]["source"]["type"], "base64")
self.assertEqual(user_msg["content"][0]["source"]["media_type"], "image/jpeg")
self.assertEqual(user_msg["content"][0]["source"]["data"], "fake_image_data")
# Check first text part (moved to second position)
self.assertEqual(user_msg["content"][1]["type"], "text")
self.assertEqual(user_msg["content"][1]["text"], "What do you see in this image?")
# Check second text part (moved to third position)
self.assertEqual(user_msg["content"][2]["type"], "text")
self.assertEqual(user_msg["content"][2]["text"], "Please describe it in detail.")
def test_multiple_system_instructions_handling(self):
"""Test that first system instruction is extracted, later ones converted to user messages."""
messages = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi there!"},
{"role": "system", "content": "Remember to be concise."}, # Later system message
]
# Create context
context = LLMContext(messages=messages)
# Get invocation params
params = self.adapter.get_llm_invocation_params(context, enable_prompt_caching=False)
# System instruction should be extracted from first message
self.assertEqual(params["system"], "You are a helpful assistant.")
# Should have 3 messages remaining (system message was removed, later system converted to user)
self.assertEqual(len(params["messages"]), 3)
self.assertEqual(params["messages"][0]["role"], "user")
self.assertEqual(params["messages"][0]["content"], "Hello")
self.assertEqual(params["messages"][1]["role"], "assistant")
self.assertEqual(params["messages"][1]["content"], "Hi there!")
# Later system message should be converted to user role
self.assertEqual(params["messages"][2]["role"], "user")
self.assertEqual(params["messages"][2]["content"], "Remember to be concise.")
def test_single_system_message_converted_to_user(self):
"""Test that a single system message is converted to user role when no other messages exist."""
messages = [
{"role": "system", "content": "You are a helpful assistant."},
]
# Create context
context = LLMContext(messages=messages)
# Get invocation params
params = self.adapter.get_llm_invocation_params(context, enable_prompt_caching=False)
# System should be NOT_GIVEN since we only have one message
from anthropic import NOT_GIVEN
self.assertEqual(params["system"], NOT_GIVEN)
# Single system message should be converted to user role
self.assertEqual(len(params["messages"]), 1)
self.assertEqual(params["messages"][0]["role"], "user")
self.assertEqual(params["messages"][0]["content"], "You are a helpful assistant.")
def test_system_instruction_only(self):
"""system_instruction alone becomes the system parameter."""
messages: list[LLMStandardMessage] = [
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(
context, enable_prompt_caching=False, system_instruction="Be helpful."
)
self.assertEqual(params["system"], "Be helpful.")
self.assertEqual(len(params["messages"]), 1)
self.assertEqual(params["messages"][0]["role"], "user")
def test_initial_developer_message_becomes_user(self):
"""Initial developer message without system_instruction becomes user, not system."""
from anthropic import NOT_GIVEN
messages: list[LLMStandardMessage] = [
{"role": "developer", "content": "Extra context."},
{"role": "assistant", "content": "OK"},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context, enable_prompt_caching=False)
self.assertEqual(params["system"], NOT_GIVEN)
self.assertEqual(len(params["messages"]), 3)
self.assertEqual(params["messages"][0]["role"], "user")
self.assertEqual(params["messages"][0]["content"], "Extra context.")
def test_both_system_instruction_and_system_message_warns(self):
"""system_instruction + initial system message warns and uses system_instruction."""
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger:
params = self.adapter.get_llm_invocation_params(
context,
enable_prompt_caching=False,
system_instruction="Be concise.",
)
mock_logger.warning.assert_called_once()
warning_msg = mock_logger.warning.call_args[0][0]
self.assertIn("Using system_instruction", warning_msg)
self.assertEqual(params["system"], "Be concise.")
def test_both_system_instruction_and_developer_message_no_warning(self):
"""system_instruction + initial developer message: no warning, developer becomes user."""
messages: list[LLMStandardMessage] = [
{"role": "developer", "content": "Extra context."},
{"role": "assistant", "content": "Hi"},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger:
params = self.adapter.get_llm_invocation_params(
context,
enable_prompt_caching=False,
system_instruction="Be concise.",
)
mock_logger.warning.assert_not_called()
self.assertEqual(params["system"], "Be concise.")
# Developer message should have been converted to "user"
self.assertEqual(params["messages"][0]["role"], "user")
self.assertEqual(params["messages"][0]["content"], "Extra context.")
def test_subsequent_developer_messages_converted_to_user(self):
"""Subsequent developer messages are converted to user role."""
messages: list[LLMStandardMessage] = [
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi"},
{"role": "developer", "content": "More instructions"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context, enable_prompt_caching=False)
# Developer message was converted to "user"
self.assertEqual(params["messages"][2]["role"], "user")
self.assertEqual(params["messages"][2]["content"], "More instructions")
def test_initial_system_discarded_when_system_instruction_provided(self):
"""Initial system message is discarded when system_instruction is provided."""
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "Old instruction."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
with patch("pipecat.adapters.base_llm_adapter.logger"):
params = self.adapter.get_llm_invocation_params(
context,
enable_prompt_caching=False,
system_instruction="New instruction.",
)
self.assertEqual(params["system"], "New instruction.")
# Only the user message should remain
self.assertEqual(len(params["messages"]), 1)
self.assertEqual(params["messages"][0]["role"], "user")
class TestAWSBedrockGetLLMInvocationParams(unittest.TestCase):
def setUp(self) -> None:
"""Sets up a common adapter instance for all tests."""
self.adapter = AWSBedrockLLMAdapter()
def test_standard_messages_converted_to_aws_bedrock_format(self):
"""Test that LLMStandardMessage objects are converted to AWS Bedrock format."""
# Create standard messages
standard_messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Hello, how are you?"},
{"role": "assistant", "content": "I'm doing well, thank you!"},
]
# Create context
context = LLMContext(messages=standard_messages)
# Get invocation params
params = self.adapter.get_llm_invocation_params(context)
# Verify system instruction is extracted (in AWS Bedrock format)
self.assertIsInstance(params["system"], list)
self.assertEqual(len(params["system"]), 1)
self.assertEqual(params["system"][0]["text"], "You are a helpful assistant.")
# Verify messages are in the params (2 messages after system extraction)
self.assertIn("messages", params)
self.assertEqual(len(params["messages"]), 2)
# Check first message (user) - should be converted to AWS Bedrock format
user_msg = params["messages"][0]
self.assertEqual(user_msg["role"], "user")
self.assertIsInstance(user_msg["content"], list)
self.assertEqual(len(user_msg["content"]), 1)
self.assertEqual(user_msg["content"][0]["text"], "Hello, how are you?")
# Check second message (assistant) - should be converted to AWS Bedrock format
assistant_msg = params["messages"][1]
self.assertEqual(assistant_msg["role"], "assistant")
self.assertIsInstance(assistant_msg["content"], list)
self.assertEqual(len(assistant_msg["content"]), 1)
self.assertEqual(assistant_msg["content"][0]["text"], "I'm doing well, thank you!")
def test_llm_specific_message_filtering(self):
"""Test that AWS-specific messages are included and others are filtered out."""
# Create aws-specific message content (which is what AWS Bedrock uses)
aws_message_content = {
"role": "user",
"content": [
{"text": "Hello"},
{"image": {"format": "jpeg", "source": {"bytes": b"fake_image_data"}}},
],
}
messages = [
{"role": "user", "content": "Standard message"},
OpenAILLMAdapter().create_llm_specific_message(
{"role": "user", "content": "OpenAI specific"}
),
GeminiLLMAdapter().create_llm_specific_message(
{"role": "user", "content": "Google specific"}
),
self.adapter.create_llm_specific_message(message=aws_message_content),
{"role": "assistant", "content": "Response"},
]
# Create context
context = LLMContext(messages=messages)
# Get invocation params
params = self.adapter.get_llm_invocation_params(context)
# Should only have 2 messages after merging consecutive user messages: merged user + standard response
# (openai and google specific filtered out, standard + aws-specific merged)
self.assertEqual(len(params["messages"]), 2)
# First message: merged user message (standard + aws-specific)
user_msg = params["messages"][0]
self.assertEqual(user_msg["role"], "user")
self.assertIsInstance(user_msg["content"], list)
# Should have 3 content blocks: standard text + aws text + aws image
self.assertEqual(len(user_msg["content"]), 3)
self.assertEqual(user_msg["content"][0]["text"], "Standard message")
self.assertEqual(user_msg["content"][1]["text"], "Hello")
self.assertIn("image", user_msg["content"][2])
# Second message: standard response
self.assertEqual(params["messages"][1]["content"][0]["text"], "Response")
def test_consecutive_same_role_messages_merged(self):
"""Test that consecutive messages with the same role are merged into multi-content blocks."""
messages = [
{"role": "user", "content": "First user message"},
{"role": "user", "content": "Second user message"},
{"role": "user", "content": "Third user message"},
{"role": "assistant", "content": "First assistant message"},
{"role": "assistant", "content": "Second assistant message"},
]
# Create context
context = LLMContext(messages=messages)
# Get invocation params
params = self.adapter.get_llm_invocation_params(context)
# Should have 2 messages after merging (1 user, 1 assistant)
self.assertEqual(len(params["messages"]), 2)
# Check merged user message
user_msg = params["messages"][0]
self.assertEqual(user_msg["role"], "user")
self.assertIsInstance(user_msg["content"], list)
self.assertEqual(len(user_msg["content"]), 3)
self.assertEqual(user_msg["content"][0]["text"], "First user message")
self.assertEqual(user_msg["content"][1]["text"], "Second user message")
self.assertEqual(user_msg["content"][2]["text"], "Third user message")
# Check merged assistant message
assistant_msg = params["messages"][1]
self.assertEqual(assistant_msg["role"], "assistant")
self.assertIsInstance(assistant_msg["content"], list)
self.assertEqual(len(assistant_msg["content"]), 2)
self.assertEqual(assistant_msg["content"][0]["text"], "First assistant message")
self.assertEqual(assistant_msg["content"][1]["text"], "Second assistant message")
def test_empty_text_converted_to_empty_placeholder(self):
"""Test that empty text content is converted to "(empty)" string."""
messages = [
{"role": "user", "content": ""}, # Empty string
{
"role": "assistant",
"content": [
{"type": "text", "text": ""}, # Empty text in list content
{"type": "text", "text": "Valid text"},
],
},
]
# Create context
context = LLMContext(messages=messages)
# Get invocation params
params = self.adapter.get_llm_invocation_params(context)
# Check that empty string content was converted
user_msg = params["messages"][0]
self.assertIsInstance(user_msg["content"], list)
self.assertEqual(user_msg["content"][0]["text"], "(empty)")
# Check that empty text in list content was converted
assistant_msg = params["messages"][1]
self.assertIsInstance(assistant_msg["content"], list)
self.assertEqual(assistant_msg["content"][0]["text"], "(empty)")
self.assertEqual(assistant_msg["content"][1]["text"], "Valid text")
def test_complex_message_content_preserved(self):
"""Test that complex message structures (text + image) are properly converted to AWS Bedrock format."""
# Create a complex message with both text and image content
# Use a valid base64 string for the image
complex_message = {
"role": "user",
"content": [
{"type": "text", "text": "What do you see in this image?"},
{
"type": "image_url",
"image_url": {
"url": "data:image/jpeg;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg=="
},
},
{"type": "text", "text": "Please describe it in detail."},
],
}
messages = [
complex_message,
{"role": "assistant", "content": "I can see the image clearly."},
]
# Create context
context = LLMContext(messages=messages)
# Get invocation params
params = self.adapter.get_llm_invocation_params(context)
# Verify complex message structure is preserved and converted
self.assertEqual(len(params["messages"]), 2)
user_msg = params["messages"][0]
self.assertEqual(user_msg["role"], "user")
self.assertIsInstance(user_msg["content"], list)
self.assertEqual(len(user_msg["content"]), 3)
# Note: AWS Bedrock adapter reorders single images to come before text, like Anthropic
# Check image part (should be moved to first position and converted from image_url to image)
self.assertIn("image", user_msg["content"][0])
self.assertEqual(user_msg["content"][0]["image"]["format"], "jpeg")
self.assertIn("source", user_msg["content"][0]["image"])
self.assertIn("bytes", user_msg["content"][0]["image"]["source"])
# Check first text part (moved to second position)
self.assertEqual(user_msg["content"][1]["text"], "What do you see in this image?")
# Check second text part (moved to third position)
self.assertEqual(user_msg["content"][2]["text"], "Please describe it in detail.")
def test_multiple_system_instructions_handling(self):
"""Test that first system instruction is extracted, later ones converted to user messages."""
messages = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi there!"},
{"role": "system", "content": "Remember to be concise."}, # Later system message
]
# Create context
context = LLMContext(messages=messages)
# Get invocation params
params = self.adapter.get_llm_invocation_params(context)
# System instruction should be extracted from first message (in AWS Bedrock format)
self.assertIsInstance(params["system"], list)
self.assertEqual(len(params["system"]), 1)
self.assertEqual(params["system"][0]["text"], "You are a helpful assistant.")
# Should have 3 messages remaining (system message was removed, later system converted to user)
self.assertEqual(len(params["messages"]), 3)
self.assertEqual(params["messages"][0]["role"], "user")
self.assertEqual(params["messages"][0]["content"][0]["text"], "Hello")
self.assertEqual(params["messages"][1]["role"], "assistant")
self.assertEqual(params["messages"][1]["content"][0]["text"], "Hi there!")
# Later system message should be converted to user role
self.assertEqual(params["messages"][2]["role"], "user")
self.assertEqual(params["messages"][2]["content"][0]["text"], "Remember to be concise.")
def test_single_system_message_handling(self):
"""Test that a single system message is converted to user role when no other messages exist."""
messages = [
{"role": "system", "content": "You are a helpful assistant."},
]
# Create context
context = LLMContext(messages=messages)
# Get invocation params
params = self.adapter.get_llm_invocation_params(context)
# When there's only one message, it's converted to user in-place (not extracted)
# so system is None
self.assertIsNone(params["system"])
# Single system message should be converted to user role
self.assertEqual(len(params["messages"]), 1)
self.assertEqual(params["messages"][0]["role"], "user")
self.assertEqual(
params["messages"][0]["content"][0]["text"], "You are a helpful assistant."
)
def test_system_instruction_only(self):
"""system_instruction alone becomes the system parameter."""
messages: list[LLMStandardMessage] = [
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context, system_instruction="Be helpful.")
self.assertEqual(params["system"], [{"text": "Be helpful."}])
def test_initial_developer_message_becomes_user(self):
"""Initial developer message without system_instruction becomes user, not system."""
messages: list[LLMStandardMessage] = [
{"role": "developer", "content": "Extra context."},
{"role": "assistant", "content": "OK"},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertIsNone(params["system"])
self.assertEqual(len(params["messages"]), 3)
self.assertEqual(params["messages"][0]["role"], "user")
self.assertEqual(params["messages"][0]["content"][0]["text"], "Extra context.")
def test_both_system_instruction_and_system_message_warns(self):
"""system_instruction + initial system message warns and uses system_instruction."""
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger:
params = self.adapter.get_llm_invocation_params(
context, system_instruction="Be concise."
)
mock_logger.warning.assert_called_once()
self.assertEqual(params["system"], [{"text": "Be concise."}])
def test_both_system_instruction_and_developer_message_no_warning(self):
"""system_instruction + initial developer message: no warning, developer becomes user."""
messages: list[LLMStandardMessage] = [
{"role": "developer", "content": "Extra context."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger:
params = self.adapter.get_llm_invocation_params(
context, system_instruction="Be concise."
)
mock_logger.warning.assert_not_called()
self.assertEqual(params["system"], [{"text": "Be concise."}])
self.assertEqual(params["messages"][0]["role"], "user")
def test_subsequent_developer_messages_converted_to_user(self):
"""Subsequent developer messages are converted to user role."""
messages: list[LLMStandardMessage] = [
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi"},
{"role": "developer", "content": "More instructions"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(params["messages"][2]["role"], "user")
class TestPerplexityGetLLMInvocationParams(unittest.TestCase):
def setUp(self) -> None:
"""Sets up a common adapter instance for all tests."""
self.adapter = PerplexityLLMAdapter()
def test_standard_messages_pass_through(self):
"""Test that a valid [user, assistant, user] sequence passes through unchanged."""
messages: list[LLMStandardMessage] = [
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi there!"},
{"role": "user", "content": "How are you?"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(len(params["messages"]), 3)
self.assertEqual(params["messages"][0]["role"], "user")
self.assertEqual(params["messages"][0]["content"], "Hello")
self.assertEqual(params["messages"][1]["role"], "assistant")
self.assertEqual(params["messages"][1]["content"], "Hi there!")
self.assertEqual(params["messages"][2]["role"], "user")
self.assertEqual(params["messages"][2]["content"], "How are you?")
def test_initial_system_message_preserved(self):
"""Test that a valid [system, user, assistant, user] sequence passes through unchanged."""
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi!"},
{"role": "user", "content": "Bye"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(len(params["messages"]), 4)
self.assertEqual(params["messages"][0]["role"], "system")
self.assertEqual(params["messages"][0]["content"], "You are a helpful assistant.")
self.assertEqual(params["messages"][1]["role"], "user")
self.assertEqual(params["messages"][2]["role"], "assistant")
self.assertEqual(params["messages"][3]["role"], "user")
def test_consecutive_same_role_messages_merged(self):
"""Test that consecutive user messages are merged into list-of-dicts content."""
messages: list[LLMStandardMessage] = [
{"role": "user", "content": "First message"},
{"role": "user", "content": "Second message"},
{"role": "assistant", "content": "Response"},
{"role": "user", "content": "Third message"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(len(params["messages"]), 3)
# First message should be merged users
merged = params["messages"][0]
self.assertEqual(merged["role"], "user")
self.assertIsInstance(merged["content"], list)
self.assertEqual(len(merged["content"]), 2)
self.assertEqual(merged["content"][0]["type"], "text")
self.assertEqual(merged["content"][0]["text"], "First message")
self.assertEqual(merged["content"][1]["type"], "text")
self.assertEqual(merged["content"][1]["text"], "Second message")
self.assertEqual(params["messages"][1]["role"], "assistant")
self.assertEqual(params["messages"][2]["role"], "user")
def test_non_initial_system_converted_to_user(self):
"""Test that non-initial system messages are converted to user and merged with adjacent user."""
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi!"},
{"role": "system", "content": "Be concise."},
{"role": "user", "content": "Tell me about Python."},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
# system(initial), user, assistant, merged(system→user + user)
self.assertEqual(len(params["messages"]), 4)
self.assertEqual(params["messages"][0]["role"], "system")
self.assertEqual(params["messages"][1]["role"], "user")
self.assertEqual(params["messages"][2]["role"], "assistant")
# The converted system→user and the following user should be merged
merged = params["messages"][3]
self.assertEqual(merged["role"], "user")
self.assertIsInstance(merged["content"], list)
self.assertEqual(len(merged["content"]), 2)
self.assertEqual(merged["content"][0]["text"], "Be concise.")
self.assertEqual(merged["content"][1]["text"], "Tell me about Python.")
def test_multiple_system_messages_at_start_preserved(self):
"""Test that multiple consecutive system messages at start pass through unchanged."""
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "system", "content": "Always be polite."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(len(params["messages"]), 3)
self.assertEqual(params["messages"][0]["role"], "system")
self.assertEqual(params["messages"][0]["content"], "You are a helpful assistant.")
self.assertEqual(params["messages"][1]["role"], "system")
self.assertEqual(params["messages"][1]["content"], "Always be polite.")
self.assertEqual(params["messages"][2]["role"], "user")
self.assertEqual(params["messages"][2]["content"], "Hello")
def test_trailing_assistant_removed(self):
"""Test that a trailing assistant message is removed."""
messages: list[LLMStandardMessage] = [
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi there!"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(len(params["messages"]), 1)
self.assertEqual(params["messages"][0]["role"], "user")
self.assertEqual(params["messages"][0]["content"], "Hello")
def test_only_system_messages_preserved(self):
"""Test that system-only contexts are left unchanged (no system→user conversion).
We intentionally do not convert trailing system messages to "user"
because that would make the transformation unstable across calls —
Perplexity has statefulness within a conversation, so a message that
was "user" in one call but becomes "system" in the next causes errors.
"""
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are a helpful assistant."},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(len(params["messages"]), 1)
self.assertEqual(params["messages"][0]["role"], "system")
def test_system_exposed_after_trailing_assistant_removed(self):
"""Test that a system message exposed by trailing assistant removal stays system.
It's important that initial system messages are never converted to
"user", because Perplexity has statefulness within a conversation — if
a message was sent as "system" in one call and then becomes "user" in a
later call (after more messages are appended), the API rejects it.
"""
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are helpful."},
{"role": "assistant", "content": "Sure thing."},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
# Trailing assistant removed → [system], system stays as-is
self.assertEqual(len(params["messages"]), 1)
self.assertEqual(params["messages"][0]["role"], "system")
self.assertEqual(params["messages"][0]["content"], "You are helpful.")
def test_consecutive_assistants_merged_then_trailing_removed(self):
"""Test that consecutive assistant messages are merged, then trailing assistant is removed."""
messages: list[LLMStandardMessage] = [
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "First response"},
{"role": "assistant", "content": "Second response"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
# After merging assistants we get [user, assistant(merged)], then trailing
# assistant is removed, leaving just [user]
self.assertEqual(len(params["messages"]), 1)
self.assertEqual(params["messages"][0]["role"], "user")
self.assertEqual(params["messages"][0]["content"], "Hello")
def test_tool_messages_preserved(self):
"""Test that tool messages pass through without modification."""
messages: list[LLMStandardMessage] = [
{"role": "user", "content": "What's the weather?"},
{
"role": "assistant",
"content": "Let me check.",
"tool_calls": [{"id": "1", "function": {"name": "get_weather", "arguments": "{}"}}],
},
{"role": "tool", "content": "Sunny, 72F", "tool_call_id": "1"},
{"role": "user", "content": "Thanks!"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(len(params["messages"]), 4)
self.assertEqual(params["messages"][0]["role"], "user")
self.assertEqual(params["messages"][1]["role"], "assistant")
self.assertEqual(params["messages"][2]["role"], "tool")
self.assertEqual(params["messages"][2]["content"], "Sunny, 72F")
self.assertEqual(params["messages"][3]["role"], "user")
def test_empty_messages(self):
"""Test that empty messages list returns empty."""
context = LLMContext(messages=[])
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(params["messages"], [])
class TestOpenAIResponsesGetLLMInvocationParams(unittest.TestCase):
def setUp(self) -> None:
"""Sets up a common adapter instance for all tests."""
self.adapter = OpenAIResponsesLLMAdapter()
def test_simple_user_assistant_messages(self):
"""Simple user/assistant text messages are converted correctly."""
messages: list[LLMStandardMessage] = [
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi there!"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(len(params["input"]), 2)
self.assertEqual(params["input"][0], {"role": "user", "content": "Hello"})
self.assertEqual(params["input"][1], {"role": "assistant", "content": "Hi there!"})
def test_system_role_converted_to_developer(self):
"""System role messages are converted to developer role."""
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(params["input"][0]["role"], "developer")
self.assertEqual(params["input"][0]["content"], "You are helpful.")
def test_developer_role_kept_as_developer(self):
"""Developer role messages are kept as developer role."""
messages: list[LLMStandardMessage] = [
{"role": "developer", "content": "Extra context."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(params["input"][0]["role"], "developer")
self.assertEqual(params["input"][0]["content"], "Extra context.")
def test_system_message_without_system_instruction_no_warning(self):
"""System message without system_instruction does not trigger a warning."""
adapter = OpenAIResponsesLLMAdapter()
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger:
adapter.get_llm_invocation_params(context)
mock_logger.warning.assert_not_called()
def test_system_message_with_system_instruction_triggers_warning(self):
"""System message + system_instruction triggers a conflict warning."""
adapter = OpenAIResponsesLLMAdapter()
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger:
adapter.get_llm_invocation_params(context, system_instruction="Be concise.")
mock_logger.warning.assert_called_once()
warning_msg = mock_logger.warning.call_args[0][0]
self.assertIn("system_instruction", warning_msg)
def test_developer_message_with_system_instruction_no_warning(self):
"""Developer message + system_instruction does NOT trigger a warning."""
adapter = OpenAIResponsesLLMAdapter()
messages: list[LLMStandardMessage] = [
{"role": "developer", "content": "Extra context."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger:
params = adapter.get_llm_invocation_params(context, system_instruction="Be concise.")
mock_logger.warning.assert_not_called()
# Developer message stays as developer, system_instruction becomes instructions
self.assertEqual(params["input"][0]["role"], "developer")
self.assertEqual(params["instructions"], "Be concise.")
def test_non_initial_system_message_no_warning(self):
"""Non-initial system messages are converted without a warning."""
messages: list[LLMStandardMessage] = [
{"role": "user", "content": "Hello"},
{"role": "system", "content": "New instruction"},
]
context = LLMContext(messages=messages)
adapter = OpenAIResponsesLLMAdapter()
with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger:
params = adapter.get_llm_invocation_params(context, system_instruction="Be helpful.")
mock_logger.warning.assert_not_called()
self.assertEqual(params["input"][1]["role"], "developer")
self.assertEqual(params["input"][1]["content"], "New instruction")
def test_conflict_warning_fires_only_once(self):
"""The conflict warning fires only once per adapter instance."""
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
adapter = OpenAIResponsesLLMAdapter()
with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger:
adapter.get_llm_invocation_params(context, system_instruction="Be concise.")
adapter.get_llm_invocation_params(context, system_instruction="Be concise.")
mock_logger.warning.assert_called_once()
def test_assistant_tool_calls_to_function_call(self):
"""Assistant messages with tool_calls produce function_call input items."""
messages = [
{
"role": "assistant",
"tool_calls": [
{
"id": "call_123",
"function": {
"name": "get_weather",
"arguments": '{"location": "SF"}',
},
"type": "function",
}
],
}
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(len(params["input"]), 1)
fc = params["input"][0]
self.assertEqual(fc["type"], "function_call")
self.assertEqual(fc["call_id"], "call_123")
self.assertEqual(fc["name"], "get_weather")
self.assertEqual(fc["arguments"], '{"location": "SF"}')
def test_multiple_tool_calls(self):
"""Multiple tool calls in one assistant message produce multiple function_call items."""
messages = [
{
"role": "assistant",
"tool_calls": [
{
"id": "call_1",
"function": {"name": "get_weather", "arguments": '{"location": "SF"}'},
"type": "function",
},
{
"id": "call_2",
"function": {
"name": "get_restaurant",
"arguments": '{"location": "SF"}',
},
"type": "function",
},
],
}
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(len(params["input"]), 2)
self.assertEqual(params["input"][0]["name"], "get_weather")
self.assertEqual(params["input"][1]["name"], "get_restaurant")
def test_tool_message_to_function_call_output(self):
"""Tool role messages produce function_call_output input items."""
messages = [
{
"role": "tool",
"content": '{"temperature": "72"}',
"tool_call_id": "call_123",
}
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(len(params["input"]), 1)
fco = params["input"][0]
self.assertEqual(fco["type"], "function_call_output")
self.assertEqual(fco["call_id"], "call_123")
self.assertEqual(fco["output"], '{"temperature": "72"}')
def test_mixed_conversation(self):
"""Mixed conversation with text + function calls converts correctly."""
messages = [
{"role": "user", "content": "What's the weather in SF?"},
{
"role": "assistant",
"tool_calls": [
{
"id": "call_abc",
"function": {"name": "get_weather", "arguments": '{"location": "SF"}'},
"type": "function",
}
],
},
{
"role": "tool",
"content": '{"temp": "72"}',
"tool_call_id": "call_abc",
},
{"role": "assistant", "content": "It's 72 degrees in SF."},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(len(params["input"]), 4)
self.assertEqual(params["input"][0]["role"], "user")
self.assertEqual(params["input"][1]["type"], "function_call")
self.assertEqual(params["input"][2]["type"], "function_call_output")
self.assertEqual(params["input"][3]["role"], "assistant")
def test_multimodal_text_conversion(self):
"""Multimodal text content parts are converted to input_text."""
messages = [
{
"role": "user",
"content": [
{"type": "text", "text": "What's in this image?"},
],
}
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
content = params["input"][0]["content"]
self.assertEqual(len(content), 1)
self.assertEqual(content[0]["type"], "input_text")
self.assertEqual(content[0]["text"], "What's in this image?")
def test_multimodal_image_conversion(self):
"""Multimodal image_url content parts are converted to input_image."""
messages = [
{
"role": "user",
"content": [
{"type": "text", "text": "Describe this:"},
{
"type": "image_url",
"image_url": {"url": "data:image/jpeg;base64,abc123"},
},
],
}
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
content = params["input"][0]["content"]
self.assertEqual(len(content), 2)
self.assertEqual(content[0]["type"], "input_text")
self.assertEqual(content[1]["type"], "input_image")
self.assertEqual(content[1]["image_url"], "data:image/jpeg;base64,abc123")
self.assertEqual(content[1]["detail"], "auto")
def test_multimodal_image_with_detail(self):
"""Image content parts preserve the detail setting when provided."""
messages = [
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {"url": "https://example.com/img.png", "detail": "high"},
},
],
}
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
content = params["input"][0]["content"]
self.assertEqual(content[0]["detail"], "high")
def test_tools_schema_flattening(self):
"""Tools schema with nested function dict is flattened to Responses API format."""
weather_fn = FunctionSchema(
name="get_weather",
description="Get the current weather",
properties={
"location": {"type": "string", "description": "The city"},
},
required=["location"],
)
tools = ToolsSchema(standard_tools=[weather_fn])
context = LLMContext(tools=tools)
params = self.adapter.get_llm_invocation_params(context)
tool_list = params["tools"]
self.assertEqual(len(tool_list), 1)
tool = tool_list[0]
self.assertEqual(tool["type"], "function")
self.assertEqual(tool["name"], "get_weather")
self.assertEqual(tool["description"], "Get the current weather")
self.assertIn("properties", tool["parameters"])
def test_empty_messages(self):
"""Empty messages list produces empty input list."""
context = LLMContext(messages=[])
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(params["input"], [])
def test_llm_specific_message_passthrough(self):
"""LLMSpecificMessage with llm='openai_responses' passes through."""
specific_msg = self.adapter.create_llm_specific_message(
{"type": "function_call", "call_id": "x", "name": "foo", "arguments": "{}"}
)
messages = [
{"role": "user", "content": "Hello"},
specific_msg,
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(len(params["input"]), 2)
self.assertEqual(params["input"][0]["role"], "user")
self.assertEqual(params["input"][1]["type"], "function_call")
def test_id_for_llm_specific_messages(self):
"""Adapter identifier is 'openai_responses'."""
self.assertEqual(self.adapter.id_for_llm_specific_messages, "openai_responses")
def test_system_instruction_with_messages_sets_instructions(self):
"""When system_instruction is provided and input is non-empty, sets instructions."""
messages: list[LLMStandardMessage] = [
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context, system_instruction="Be helpful.")
self.assertEqual(params["instructions"], "Be helpful.")
self.assertEqual(len(params["input"]), 1)
self.assertEqual(params["input"][0]["role"], "user")
def test_system_instruction_with_empty_input_becomes_developer_message(self):
"""When system_instruction is provided but input is empty, it becomes a developer message."""
context = LLMContext(messages=[])
params = self.adapter.get_llm_invocation_params(context, system_instruction="Be helpful.")
self.assertNotIn("instructions", params)
self.assertEqual(len(params["input"]), 1)
self.assertEqual(params["input"][0]["role"], "developer")
self.assertEqual(params["input"][0]["content"], "Be helpful.")
def test_no_system_instruction_omits_instructions(self):
"""When no system_instruction is provided, instructions key is absent."""
context = LLMContext(messages=[{"role": "user", "content": "Hi"}])
params = self.adapter.get_llm_invocation_params(context)
self.assertNotIn("instructions", params)
class TestOpenAIRealtimeGetLLMInvocationParams(unittest.TestCase):
def setUp(self) -> None:
self.adapter = OpenAIRealtimeLLMAdapter()
def test_system_message_extracted_as_instruction(self):
"""Initial system message is extracted as system_instruction."""
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(params["system_instruction"], "You are helpful.")
self.assertEqual(len(params["messages"]), 1)
def test_developer_message_becomes_user(self):
"""Developer message is converted to user, not extracted as system instruction."""
messages: list[LLMStandardMessage] = [
{"role": "developer", "content": "Extra context."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertIsNone(params["system_instruction"])
# Developer converted to user, then packed with the other user message
self.assertEqual(len(params["messages"]), 1)
def test_subsequent_developer_message_becomes_user(self):
"""Non-initial developer message is converted to user."""
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are helpful."},
{"role": "developer", "content": "Extra context."},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(params["system_instruction"], "You are helpful.")
# Developer message converted to user
self.assertEqual(len(params["messages"]), 1)
def test_empty_messages(self):
"""Empty messages list returns empty."""
context = LLMContext(messages=[])
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(params["messages"], [])
self.assertIsNone(params["system_instruction"])
class TestGrokRealtimeGetLLMInvocationParams(unittest.TestCase):
def setUp(self) -> None:
self.adapter = GrokRealtimeLLMAdapter()
def test_system_message_extracted_as_instruction(self):
"""Initial system message is extracted as system_instruction."""
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(params["system_instruction"], "You are helpful.")
self.assertEqual(len(params["messages"]), 1)
def test_developer_message_becomes_user(self):
"""Developer message is converted to user, not extracted as system instruction."""
messages: list[LLMStandardMessage] = [
{"role": "developer", "content": "Extra context."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertIsNone(params["system_instruction"])
# Developer converted to user, then packed with the other user message
self.assertEqual(len(params["messages"]), 1)
def test_subsequent_developer_message_becomes_user(self):
"""Non-initial developer message is converted to user."""
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are helpful."},
{"role": "developer", "content": "Extra context."},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(params["system_instruction"], "You are helpful.")
self.assertEqual(len(params["messages"]), 1)
def test_empty_messages(self):
"""Empty messages list returns empty."""
context = LLMContext(messages=[])
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(params["messages"], [])
self.assertIsNone(params["system_instruction"])
class TestAWSNovaSonicGetLLMInvocationParams(unittest.TestCase):
def setUp(self) -> None:
self.adapter = AWSNovaSonicLLMAdapter()
def test_system_message_extracted_as_instruction(self):
"""Initial system message is extracted as system_instruction."""
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(params["system_instruction"], "You are helpful.")
self.assertEqual(len(params["messages"]), 1)
def test_developer_message_becomes_user(self):
"""Developer message is converted to user, not extracted as system instruction."""
messages: list[LLMStandardMessage] = [
{"role": "developer", "content": "Extra context."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertIsNone(params["system_instruction"])
# Both messages should be present (developer as user, plus the real user)
self.assertEqual(len(params["messages"]), 2)
def test_subsequent_developer_message_becomes_user(self):
"""Non-initial developer message is converted to user."""
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are helpful."},
{"role": "developer", "content": "Extra context."},
{"role": "assistant", "content": "Hi"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(params["system_instruction"], "You are helpful.")
# Developer becomes user, plus assistant
self.assertEqual(len(params["messages"]), 2)
class TestBaseLLMAdapterHelpers(unittest.TestCase):
"""Tests for the shared helper methods on BaseLLMAdapter."""
def setUp(self):
# Use OpenAILLMAdapter as a concrete implementation for testing the base helpers
self.adapter = OpenAILLMAdapter()
def test_extract_system_message(self):
"""System message is extracted from messages[0]."""
messages = [
{"role": "system", "content": "Be helpful."},
{"role": "user", "content": "Hello"},
]
content = self.adapter._extract_initial_system(messages, system_instruction=None)
self.assertEqual(content, "Be helpful.")
self.assertEqual(len(messages), 1) # popped
def test_extract_developer_not_extracted(self):
"""Developer message is not extracted by _extract_initial_system."""
messages = [
{"role": "developer", "content": "Context."},
{"role": "user", "content": "Hello"},
]
content = self.adapter._extract_initial_system(messages, system_instruction=None)
self.assertIsNone(content)
self.assertEqual(len(messages), 2) # not popped
self.assertEqual(messages[0]["role"], "developer") # unchanged
def test_developer_with_system_instruction_not_extracted(self):
"""Developer message with system_instruction is not handled by _extract_initial_system."""
messages = [
{"role": "developer", "content": "Context."},
{"role": "user", "content": "Hello"},
]
content = self.adapter._extract_initial_system(messages, system_instruction="Be helpful.")
self.assertIsNone(content)
self.assertEqual(len(messages), 2) # not popped
self.assertEqual(messages[0]["role"], "developer") # unchanged by helper
def test_single_system_message_becomes_user(self):
"""Single system message is converted to user instead of extracting (empty prevention)."""
messages = [
{"role": "system", "content": "Be helpful."},
]
content = self.adapter._extract_initial_system(messages, system_instruction=None)
self.assertIsNone(content)
self.assertEqual(len(messages), 1) # not popped
self.assertEqual(messages[0]["role"], "user")
def test_single_system_message_with_system_instruction_warns(self):
"""Single system message + system_instruction still warns even though content isn't extracted."""
messages = [
{"role": "system", "content": "Be helpful."},
]
adapter = OpenAILLMAdapter()
with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger:
content = adapter._extract_initial_system(messages, system_instruction="Be concise.")
mock_logger.warning.assert_called_once()
self.assertIsNone(content)
self.assertEqual(messages[0]["role"], "user")
def test_non_system_message_ignored(self):
"""Non-system/developer first message is ignored."""
messages = [
{"role": "user", "content": "Hello"},
]
content = self.adapter._extract_initial_system(messages, system_instruction=None)
self.assertIsNone(content)
self.assertEqual(len(messages), 1)
def test_empty_messages(self):
"""Empty messages list returns None."""
messages = []
content = self.adapter._extract_initial_system(messages, system_instruction=None)
self.assertIsNone(content)
def test_resolve_both_system_discard(self):
"""Resolve with discard=True: system_instruction wins, warns."""
with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger:
result = self.adapter._resolve_system_instruction(
"from context", "from settings", discard_context_system=True
)
mock_logger.warning.assert_called_once()
self.assertEqual(result, "from settings")
def test_resolve_both_system_keep(self):
"""Resolve with discard=False: warns but returns system_instruction."""
with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger:
result = self.adapter._resolve_system_instruction(
"from context", "from settings", discard_context_system=False
)
mock_logger.warning.assert_called_once()
self.assertEqual(result, "from settings")
def test_resolve_only_system_instruction(self):
"""Only system_instruction: returns it, no warning."""
with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger:
result = self.adapter._resolve_system_instruction(
None, "from settings", discard_context_system=True
)
mock_logger.warning.assert_not_called()
self.assertEqual(result, "from settings")
def test_resolve_only_context_system_discard(self):
"""Only context system (discard=True): returns it."""
result = self.adapter._resolve_system_instruction(
"from context", None, discard_context_system=True
)
self.assertEqual(result, "from context")
def test_resolve_only_context_system_keep(self):
"""Only context system (discard=False): returns None (already in messages)."""
result = self.adapter._resolve_system_instruction(
"from context", None, discard_context_system=False
)
self.assertIsNone(result)
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink