Move OpenAI Responses adapter tests into test_get_llm_invocation_params.py

Consolidates all adapter get_llm_invocation_params tests in one file.
Adds new tests for developer message handling in the Responses adapter.
This commit is contained in:
Paul Kompfner
2026-03-20 15:04:30 -04:00
parent 7377d88cf5
commit 64ba013b68
2 changed files with 373 additions and 361 deletions

View File

@@ -47,6 +47,16 @@ For AWS Bedrock adapter:
7. system_instruction overrides context system message, with conflict warnings 7. system_instruction overrides context system message, with conflict warnings
8. Developer messages are promoted to system instruction or converted to user 8. Developer messages are promoted to system instruction or 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: For BaseLLMAdapter helpers:
1. _extract_initial_system_or_developer: system/developer extraction and conversion logic 1. _extract_initial_system_or_developer: system/developer extraction and conversion logic
2. _resolve_system_instruction: conflict resolution between context and settings 2. _resolve_system_instruction: conflict resolution between context and settings
@@ -57,10 +67,13 @@ from unittest.mock import patch
from google.genai.types import Content, Part 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.anthropic_adapter import AnthropicLLMAdapter
from pipecat.adapters.services.bedrock_adapter import AWSBedrockLLMAdapter from pipecat.adapters.services.bedrock_adapter import AWSBedrockLLMAdapter
from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter
from pipecat.adapters.services.open_ai_adapter import OpenAILLMAdapter from pipecat.adapters.services.open_ai_adapter import OpenAILLMAdapter
from pipecat.adapters.services.open_ai_responses_adapter import OpenAIResponsesLLMAdapter
from pipecat.adapters.services.perplexity_adapter import PerplexityLLMAdapter from pipecat.adapters.services.perplexity_adapter import PerplexityLLMAdapter
from pipecat.processors.aggregators.llm_context import ( from pipecat.processors.aggregators.llm_context import (
LLMContext, LLMContext,
@@ -1565,6 +1578,366 @@ class TestPerplexityGetLLMInvocationParams(unittest.TestCase):
self.assertEqual(params["messages"], []) 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 TestBaseLLMAdapterHelpers(unittest.TestCase): class TestBaseLLMAdapterHelpers(unittest.TestCase):
"""Tests for the shared helper methods on BaseLLMAdapter.""" """Tests for the shared helper methods on BaseLLMAdapter."""

View File

@@ -1,361 +0,0 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Unit tests for the OpenAI Responses API adapter.
Tests the conversion from LLMContext messages to Responses API input items, including:
1. Simple user/assistant text messages pass through (with correct role)
2. System role converted to developer role
3. First-message system role triggers a warning
4. Assistant messages with tool_calls produce function_call input items
5. Tool messages produce function_call_output input items
6. Mixed conversations with text + function calls convert correctly
7. Multimodal content conversion (text -> input_text, image_url -> input_image)
8. Tools schema flattening (nested function dict -> flat format)
9. Empty messages list
10. LLMSpecificMessage with llm="openai_responses" passes through
"""
import unittest
from unittest.mock import patch
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.adapters.services.open_ai_responses_adapter import OpenAIResponsesLLMAdapter
from pipecat.processors.aggregators.llm_context import LLMContext, LLMStandardMessage
class TestOpenAIResponsesAdapter(unittest.TestCase):
def setUp(self):
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_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_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.")
# Warning should have been emitted exactly once, not twice
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)
if __name__ == "__main__":
unittest.main()