fix(responses): normalize non-leading system messages in chat completion transformation

Pulls system messages to the beginning of the message list and merges
multiple system messages into a single leading block to prevent HTTP 400
errors from backends requiring strict chat template ordering.

Fixes #40693
This commit is contained in:
Johnny Wilson Dougherty 2026-09-11 18:53:12 +05:30
parent 9a715df212
commit 7268856285
2 changed files with 250 additions and 1 deletions

View file

@ -447,7 +447,86 @@ class LiteLLMCompletionResponsesConfig:
)
)
return messages
return LiteLLMCompletionResponsesConfig._normalize_system_messages(messages)
@staticmethod
def _normalize_system_messages(
messages: list[
AllMessageValues
| GenericChatCompletionMessage
| ChatCompletionMessageToolCall
| ChatCompletionResponseMessage
| Message
],
) -> list[
AllMessageValues
| GenericChatCompletionMessage
| ChatCompletionMessageToolCall
| ChatCompletionResponseMessage
| Message
]:
"""
Normalize system messages so all system content appears at the beginning.
If multiple system messages exist, merge their contents into a single leading system message
to comply with backend chat templates that require at most one leading system message.
"""
def _is_system(msg: object) -> bool:
if isinstance(msg, dict):
return msg.get("role") == "system"
elif hasattr(msg, "role"):
return msg.role == "system"
return False
system_messages: list[
AllMessageValues
| GenericChatCompletionMessage
| ChatCompletionMessageToolCall
| ChatCompletionResponseMessage
| Message
] = [m for m in messages if _is_system(m)]
if not system_messages:
return messages
non_system_messages: list[
AllMessageValues
| GenericChatCompletionMessage
| ChatCompletionMessageToolCall
| ChatCompletionResponseMessage
| Message
] = [m for m in messages if not _is_system(m)]
if len(system_messages) == 1:
if messages and _is_system(messages[0]):
return messages
return [system_messages[0]] + non_system_messages
merged_content_parts: list[str] = []
for sm in system_messages:
raw_content: object = None
if isinstance(sm, dict):
raw_content = sm.get("content")
elif hasattr(sm, "content"):
raw_content = sm.content
if isinstance(raw_content, str):
if raw_content:
merged_content_parts.append(raw_content)
elif isinstance(raw_content, list):
for block in raw_content:
if isinstance(block, str) and block:
merged_content_parts.append(block)
elif isinstance(block, dict):
text = block.get("text")
if isinstance(text, str) and text:
merged_content_parts.append(text)
merged_system_message = ChatCompletionSystemMessage(
role="system",
content="\n\n".join(merged_content_parts),
)
return [merged_system_message] + non_system_messages
@staticmethod
async def async_responses_api_session_handler(

View file

@ -0,0 +1,170 @@
"""
Tests for system message normalization in Responses API -> Chat Completion transformation.
Regression tests for issue #40693: Anthropic /v1/messages -> Responses -> Chat Completions can emit non-leading system messages.
"""
from typing import Any
import pytest
from litellm.responses.litellm_completion_transformation.transformation import (
LiteLLMCompletionResponsesConfig,
)
def test_reproduce_issue_40693_non_leading_system_message() -> None:
"""
Reproduces issue #40693:
When instructions are provided and the Responses input contains a system message
(e.g., Claude Code harness injecting skills/agent metadata after user prompt),
the resulting message sequence must NOT emit non-leading system messages.
All system messages must be normalized into a single leading system message.
"""
responses_api_request = {
"instructions": "You are Claude Code, an AI assistant.",
}
input_items = [
{"role": "user", "content": "Hello, please help with this repo."},
{
"role": "system",
"content": "Available skills: [git, bash, edit]\nAvailable tools: [search]",
},
]
messages = LiteLLMCompletionResponsesConfig.transform_responses_api_input_to_messages(
input=input_items,
responses_api_request=responses_api_request,
)
# 1. Exactly one leading system message at index 0
assert len(messages) == 2
assert messages[0]["role"] == "system"
assert messages[1]["role"] == "user"
# 2. No non-leading system messages
assert all(
(m.get("role") if isinstance(m, dict) else getattr(m, "role", None)) != "system"
for m in messages[1:]
)
# 3. Content from both instructions and subsequent system message are preserved
system_content = messages[0]["content"]
assert "You are Claude Code, an AI assistant." in system_content
assert "Available skills: [git, bash, edit]" in system_content
def test_single_non_leading_system_message_moved_to_start() -> None:
"""
When a single system message appears after a user message without instructions,
it should be moved to the beginning of the message list.
"""
responses_api_request: dict[str, Any] = {}
input_items = [
{"role": "user", "content": "What is the weather?"},
{"role": "system", "content": "Respond only in metric units."},
]
messages = LiteLLMCompletionResponsesConfig.transform_responses_api_input_to_messages(
input=input_items,
responses_api_request=responses_api_request,
)
assert len(messages) == 2
assert messages[0]["role"] == "system"
assert messages[0]["content"] == "Respond only in metric units."
assert messages[1]["role"] == "user"
assert messages[1]["content"] == "What is the weather?"
def test_already_leading_system_message_unchanged() -> None:
"""
When a single system message is already at the beginning, it should remain untouched.
"""
responses_api_request: dict[str, Any] = {}
input_items = [
{"role": "system", "content": "System prompt."},
{"role": "user", "content": "User prompt."},
]
messages = LiteLLMCompletionResponsesConfig.transform_responses_api_input_to_messages(
input=input_items,
responses_api_request=responses_api_request,
)
assert len(messages) == 2
assert messages[0]["role"] == "system"
assert messages[0]["content"] == "System prompt."
assert messages[1]["role"] == "user"
assert messages[1]["content"] == "User prompt."
def test_no_system_message() -> None:
"""
When no system message is provided, messages should remain unchanged.
"""
responses_api_request: dict[str, Any] = {}
input_items = [
{"role": "user", "content": "Hello!"},
]
messages = LiteLLMCompletionResponsesConfig.transform_responses_api_input_to_messages(
input=input_items,
responses_api_request=responses_api_request,
)
assert len(messages) == 1
assert messages[0]["role"] == "user"
assert messages[0]["content"] == "Hello!"
def test_multiple_system_messages_with_structured_blocks() -> None:
"""
Handles system messages with list content blocks (e.g. text/input_text blocks).
"""
responses_api_request = {
"instructions": "Instruction text.",
}
input_items = [
{
"role": "system",
"content": [
{"type": "text", "text": "Structured system block 1."},
{"type": "text", "text": "Structured system block 2."},
],
},
{"role": "user", "content": "Run tests."},
]
messages = LiteLLMCompletionResponsesConfig.transform_responses_api_input_to_messages(
input=input_items,
responses_api_request=responses_api_request,
)
assert len(messages) == 2
assert messages[0]["role"] == "system"
assert messages[1]["role"] == "user"
system_content = messages[0]["content"]
assert "Instruction text." in system_content
assert "Structured system block 1." in system_content
assert "Structured system block 2." in system_content
def test_transform_responses_api_request_to_chat_completion_request_normalizes_system() -> None:
"""
Verifies end-to-end transformation via transform_responses_api_request_to_chat_completion_request.
"""
request = LiteLLMCompletionResponsesConfig.transform_responses_api_request_to_chat_completion_request(
model="openai/qwen3.8-flash-next",
input=[
{"role": "user", "content": "Hello"},
{"role": "system", "content": "Follow instructions"},
],
responses_api_request={"instructions": "Be helpful"},
)
messages = request["messages"]
assert len(messages) == 2
assert messages[0]["role"] == "system"
assert "Be helpful" in messages[0]["content"]
assert "Follow instructions" in messages[0]["content"]
assert messages[1]["role"] == "user"
assert messages[1]["content"] == "Hello"