mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-08 22:21:35 +00:00
feat(anthropic): placement policy for mid-conversation system messages
Pure functions over the OpenAI-format message list: split off the leading system run, keep later system messages as role=system at a placement Anthropic accepts on models flagged supports_mid_conversation_system (after a user turn, before an assistant turn or the end, never adjacent), and convert them to user turns in place elsewhere, keeping tool_result first in a merged user turn.
This commit is contained in:
Shifat Islam Santo
parent
f57e4b812c
commit
dedd37460c
2 changed files with 437 additions and 0 deletions
290
litellm/llms/anthropic/mid_conversation_system.py
Normal file
290
litellm/llms/anthropic/mid_conversation_system.py
Normal file
|
|
@ -0,0 +1,290 @@
|
|||
"""Placement policy for ``role: "system"`` messages that appear after the first turn
|
||||
of an Anthropic-shaped chat completions request.
|
||||
|
||||
Only the leading run of system messages belongs in the top-level ``system``
|
||||
parameter. Hoisting a later one there rewrites the cached prefix, so the provider
|
||||
re-bills the whole conversation at cache-write pricing on every reminder (#36559).
|
||||
|
||||
Models flagged ``supports_mid_conversation_system`` in the cost map accept the role
|
||||
inside ``messages`` under Anthropic's placement rules: the message must directly
|
||||
follow a user turn, must be the last entry or be followed by an assistant turn, and
|
||||
must not sit next to another system message. OpenAI-shaped clients put system
|
||||
messages anywhere, so this module moves each one to the nearest valid slot and
|
||||
merges runs that land together.
|
||||
|
||||
Models without the flag reject the role inside ``messages``. Their system messages
|
||||
become user turns in place, prefixed with an operator note so the model can tell
|
||||
the instruction apart from the user's own words. A run caught between a tool call
|
||||
and its result moves to just after the result so the ``tool_result`` block stays
|
||||
first in the merged user turn.
|
||||
|
||||
Every transformation here is a pure function of the message sequence: turn N's
|
||||
output stays a prefix of turn N+1's output, which is what keeps the provider-side
|
||||
prompt cache readable across turns. Messages are handled in OpenAI format; the
|
||||
Anthropic wire shape is built later by ``anthropic_messages_pt``.
|
||||
"""
|
||||
|
||||
from collections.abc import Iterator, Mapping, Sequence
|
||||
from itertools import groupby
|
||||
from typing import Final, Literal
|
||||
|
||||
from litellm.types.llms.anthropic import AnthropicMessagesSystemMessageParam, AnthropicSystemMessageContent
|
||||
from litellm.types.llms.openai import (
|
||||
AllMessageValues,
|
||||
ChatCompletionCachedContent,
|
||||
ChatCompletionSystemMessage,
|
||||
ChatCompletionTextObject,
|
||||
ChatCompletionUserMessage,
|
||||
)
|
||||
|
||||
CONVERTED_SYSTEM_NOTE: Final = (
|
||||
"Operator note (not from the user): the following was originally a mid-conversation system-role reminder."
|
||||
)
|
||||
|
||||
_USER_TYPE_ROLES: Final = frozenset({"user", "tool", "function"})
|
||||
_TOOL_ROLES: Final = frozenset({"tool", "function"})
|
||||
|
||||
_MessageKind = Literal["system", "tool", "user", "other"]
|
||||
_TextPart = tuple[str, ChatCompletionCachedContent | None]
|
||||
|
||||
|
||||
def _as_mapping(value: object) -> Mapping[str, object] | None:
|
||||
return value if isinstance(value, Mapping) else None
|
||||
|
||||
|
||||
def _as_items(value: object) -> tuple[object, ...]:
|
||||
return tuple(value) if isinstance(value, Sequence) and not isinstance(value, str) else ()
|
||||
|
||||
|
||||
def _field(message: object, key: str) -> object:
|
||||
"""A message field, whether the message is a dict or a pydantic ``Message``."""
|
||||
mapping: Final = _as_mapping(message)
|
||||
return mapping.get(key) if mapping is not None else getattr(message, key, None)
|
||||
|
||||
|
||||
def is_system_message(message: object) -> bool:
|
||||
return _field(message, "role") == "system"
|
||||
|
||||
|
||||
def _is_user_type(message: object) -> bool:
|
||||
return _field(message, "role") in _USER_TYPE_ROLES
|
||||
|
||||
|
||||
def _kind(message: object) -> _MessageKind:
|
||||
role: Final = _field(message, "role")
|
||||
if role == "system":
|
||||
return "system"
|
||||
if role in _TOOL_ROLES:
|
||||
return "tool"
|
||||
if role == "user":
|
||||
return "user"
|
||||
return "other"
|
||||
|
||||
|
||||
def split_leading_system_run(
|
||||
messages: Sequence[AllMessageValues],
|
||||
) -> tuple[tuple[AllMessageValues, ...], tuple[AllMessageValues, ...]]:
|
||||
"""Split ``messages`` into the leading run of system messages and everything after it."""
|
||||
leading_count: Final = next(
|
||||
(index for index, message in enumerate(messages) if not is_system_message(message)),
|
||||
len(messages),
|
||||
)
|
||||
return tuple(messages[:leading_count]), tuple(messages[leading_count:])
|
||||
|
||||
|
||||
def _cache_control(holder: object) -> ChatCompletionCachedContent | None:
|
||||
"""The client's ``cache_control`` rebuilt in the only shape Anthropic accepts."""
|
||||
value: Final = _as_mapping(_field(holder, "cache_control"))
|
||||
if value is None or value.get("type") != "ephemeral":
|
||||
return None
|
||||
ttl: Final = value.get("ttl")
|
||||
if ttl == "1h":
|
||||
one_hour: Final[ChatCompletionCachedContent] = {"type": "ephemeral", "ttl": "1h"}
|
||||
return one_hour
|
||||
if ttl == "5m":
|
||||
five_minutes: Final[ChatCompletionCachedContent] = {"type": "ephemeral", "ttl": "5m"}
|
||||
return five_minutes
|
||||
ephemeral: Final[ChatCompletionCachedContent] = {"type": "ephemeral"}
|
||||
return ephemeral
|
||||
|
||||
|
||||
def _text_parts(message: object) -> tuple[_TextPart, ...]:
|
||||
"""``(text, cache_control)`` for each non-empty text part of a system message.
|
||||
|
||||
Anthropic rejects empty text blocks and only accepts text in system content. A
|
||||
``cache_control`` on the message itself belongs to the block built from string
|
||||
content; block-level ``cache_control`` stays with its block.
|
||||
"""
|
||||
content: Final = _field(message, "content")
|
||||
if isinstance(content, str):
|
||||
return ((content, _cache_control(message)),) if content else ()
|
||||
return tuple(
|
||||
(text, _cache_control(part))
|
||||
for part in _as_items(content)
|
||||
if _field(part, "type") == "text"
|
||||
for text in (_field(part, "text"),)
|
||||
if isinstance(text, str) and text
|
||||
)
|
||||
|
||||
|
||||
def _openai_text_block(part: _TextPart) -> ChatCompletionTextObject:
|
||||
text, cache_control = part
|
||||
if cache_control is None:
|
||||
plain: Final[ChatCompletionTextObject] = {"type": "text", "text": text}
|
||||
return plain
|
||||
cached: Final[ChatCompletionTextObject] = {"type": "text", "text": text, "cache_control": cache_control}
|
||||
return cached
|
||||
|
||||
|
||||
def _anthropic_text_block(part: _TextPart) -> AnthropicSystemMessageContent:
|
||||
text, cache_control = part
|
||||
if cache_control is None:
|
||||
plain: Final[AnthropicSystemMessageContent] = {"type": "text", "text": text}
|
||||
return plain
|
||||
cached: Final[AnthropicSystemMessageContent] = {"type": "text", "text": text, "cache_control": cache_control}
|
||||
return cached
|
||||
|
||||
|
||||
def anthropic_system_messages(message: object) -> tuple[AnthropicMessagesSystemMessageParam, ...]:
|
||||
"""The Anthropic wire message for a system message, or nothing when it carries no text."""
|
||||
blocks: Final = tuple(_anthropic_text_block(part) for part in _text_parts(message))
|
||||
if not blocks:
|
||||
return ()
|
||||
wire: Final[AnthropicMessagesSystemMessageParam] = {
|
||||
"role": "system",
|
||||
"content": list(blocks), # mutable-ok: wire payload; cache_control hooks edit content blocks in place
|
||||
}
|
||||
return (wire,)
|
||||
|
||||
|
||||
def system_message_as_user(message: object) -> ChatCompletionUserMessage:
|
||||
"""A system message re-rolled as a user turn, prefixed with the operator note."""
|
||||
note: Final[ChatCompletionTextObject] = {"type": "text", "text": CONVERTED_SYSTEM_NOTE}
|
||||
content: Final[list[ChatCompletionTextObject]] = [ # mutable-ok: anthropic_messages_pt only recognises list content
|
||||
note,
|
||||
*(_openai_text_block(part) for part in _text_parts(message)),
|
||||
]
|
||||
turn: Final[ChatCompletionUserMessage] = {"role": "user", "content": content}
|
||||
return turn
|
||||
|
||||
|
||||
def _merged_system_message(run: Sequence[object]) -> tuple[ChatCompletionSystemMessage, ...]:
|
||||
parts: Final = tuple(part for message in run for part in _text_parts(message))
|
||||
if not parts:
|
||||
return ()
|
||||
content: Final[list[ChatCompletionTextObject]] = [ # mutable-ok: anthropic_messages_pt only recognises list content
|
||||
_openai_text_block(part) for part in parts
|
||||
]
|
||||
merged: Final[ChatCompletionSystemMessage] = {"role": "system", "content": content}
|
||||
return (merged,)
|
||||
|
||||
|
||||
def _converted_user_turns(run: Sequence[object]) -> tuple[ChatCompletionUserMessage, ...]:
|
||||
return tuple(system_message_as_user(message) for message in run if _text_parts(message))
|
||||
|
||||
|
||||
def _runs(messages: Sequence[AllMessageValues]) -> tuple[tuple[_MessageKind, tuple[AllMessageValues, ...]], ...]:
|
||||
return tuple((kind, tuple(group)) for kind, group in groupby(messages, key=_kind))
|
||||
|
||||
|
||||
def _converted_for_unflagged_model(messages: Sequence[AllMessageValues]) -> tuple[AllMessageValues, ...]:
|
||||
"""Convert every system message to a user turn in place.
|
||||
|
||||
A system run whose follower is a tool message is emitted after that tool run:
|
||||
``tool_result`` blocks have to open the merged user turn.
|
||||
"""
|
||||
runs: Final = _runs(messages)
|
||||
|
||||
def emit(index: int) -> tuple[AllMessageValues, ...]:
|
||||
kind, run = runs[index]
|
||||
follower: Final = runs[index + 1][0] if index + 1 < len(runs) else None
|
||||
if kind == "system":
|
||||
return () if follower == "tool" else _converted_user_turns(run)
|
||||
if kind == "tool" and index > 0 and runs[index - 1][0] == "system":
|
||||
return (*run, *_converted_user_turns(runs[index - 1][1]))
|
||||
return run
|
||||
|
||||
return tuple(message for index in range(len(runs)) for message in emit(index))
|
||||
|
||||
|
||||
def _user_type_blocks(messages: Sequence[AllMessageValues]) -> tuple[tuple[bool, tuple[int, ...]], ...]:
|
||||
"""Maximal groups of consecutive non-system messages, keyed by whether they are user-type.
|
||||
|
||||
Consecutive user-type messages become one user turn on the wire, so a group is
|
||||
the unit a system message can validly follow.
|
||||
"""
|
||||
indexed: Final = tuple((index, message) for index, message in enumerate(messages) if not is_system_message(message))
|
||||
return tuple(
|
||||
(is_user, tuple(index for index, _ in group))
|
||||
for is_user, group in groupby(indexed, key=lambda pair: _is_user_type(pair[1]))
|
||||
)
|
||||
|
||||
|
||||
def _system_runs(messages: Sequence[AllMessageValues]) -> tuple[tuple[int, ...], ...]:
|
||||
"""Index runs of consecutive system messages."""
|
||||
system_indices: Final = tuple(index for index, message in enumerate(messages) if is_system_message(message))
|
||||
return tuple(
|
||||
tuple(index for _, index in group)
|
||||
for _, group in groupby(enumerate(system_indices), key=lambda pair: pair[1] - pair[0])
|
||||
)
|
||||
|
||||
|
||||
def _anchor_block(
|
||||
run_start: int,
|
||||
messages: Sequence[AllMessageValues],
|
||||
blocks: Sequence[tuple[bool, tuple[int, ...]]],
|
||||
) -> int | None:
|
||||
"""The user-type block a system run must follow, or ``None`` when no user turn can host it.
|
||||
|
||||
``run_start`` is never 0 here: the leading system run was split off before this
|
||||
policy runs, so the message before a run is always a non-system message.
|
||||
"""
|
||||
if _is_user_type(messages[run_start - 1]):
|
||||
return next(index for index, (is_user, indices) in enumerate(blocks) if is_user and run_start - 1 in indices)
|
||||
return next((index for index, (is_user, indices) in enumerate(blocks) if is_user and indices[0] > run_start), None)
|
||||
|
||||
|
||||
def _placed_for_flagged_model(messages: Sequence[AllMessageValues]) -> tuple[AllMessageValues, ...]:
|
||||
"""Keep system messages as ``role: "system"`` at a placement Anthropic accepts.
|
||||
|
||||
A run already sitting after a user-type message stays with that user turn. A
|
||||
run after an assistant turn moves to just after the next user turn. Runs that
|
||||
share a user turn merge into one system message. A run with no user turn left
|
||||
to host it becomes user turns in place.
|
||||
"""
|
||||
blocks: Final = _user_type_blocks(messages)
|
||||
anchors: Final = tuple((run, _anchor_block(run[0], messages, blocks)) for run in _system_runs(messages))
|
||||
|
||||
def anchored_to(block_index: int) -> tuple[AllMessageValues, ...]:
|
||||
return tuple(messages[index] for run, anchor in anchors if anchor == block_index for index in run)
|
||||
|
||||
def converted_after(message_index: int) -> tuple[ChatCompletionUserMessage, ...]:
|
||||
return tuple(
|
||||
turn
|
||||
for run, anchor in anchors
|
||||
if anchor is None and run[0] == message_index + 1
|
||||
for turn in _converted_user_turns(tuple(messages[index] for index in run))
|
||||
)
|
||||
|
||||
def emit(block_index: int) -> Iterator[AllMessageValues]:
|
||||
is_user, indices = blocks[block_index]
|
||||
for index in indices:
|
||||
yield messages[index]
|
||||
yield from converted_after(index)
|
||||
if is_user:
|
||||
yield from _merged_system_message(anchored_to(block_index))
|
||||
|
||||
return tuple(message for block_index in range(len(blocks)) for message in emit(block_index))
|
||||
|
||||
|
||||
def place_mid_conversation_system(
|
||||
messages: Sequence[AllMessageValues],
|
||||
*,
|
||||
supports_mid_conversation_system: bool,
|
||||
) -> tuple[AllMessageValues, ...]:
|
||||
"""Apply the placement policy to the messages after the leading system run."""
|
||||
if not any(is_system_message(message) for message in messages):
|
||||
return tuple(messages)
|
||||
if supports_mid_conversation_system:
|
||||
return _placed_for_flagged_model(messages)
|
||||
return _converted_for_unflagged_model(messages)
|
||||
|
|
@ -0,0 +1,147 @@
|
|||
"""Placement policy for mid-conversation ``role: "system"`` messages on the chat path.
|
||||
|
||||
The provider-facing behaviour is covered through ``transform_request`` in the
|
||||
Anthropic, Vertex, Azure AI and Bedrock Invoke transformation tests; these pin
|
||||
the pure placement rules on the OpenAI-format message list.
|
||||
"""
|
||||
|
||||
import litellm
|
||||
from litellm.llms.anthropic.mid_conversation_system import (
|
||||
CONVERTED_SYSTEM_NOTE,
|
||||
place_mid_conversation_system,
|
||||
split_leading_system_run,
|
||||
)
|
||||
|
||||
|
||||
def _roles(messages):
|
||||
return [m["role"] if isinstance(m, dict) else m.role for m in messages]
|
||||
|
||||
|
||||
def _texts(message):
|
||||
return [block["text"] for block in message["content"]]
|
||||
|
||||
|
||||
def test_split_leading_system_run_keeps_later_system_messages_in_the_conversation():
|
||||
messages = [
|
||||
{"role": "system", "content": "one"},
|
||||
{"role": "system", "content": "two"},
|
||||
{"role": "user", "content": "q"},
|
||||
{"role": "system", "content": "reminder"},
|
||||
]
|
||||
|
||||
leading, later = split_leading_system_run(messages)
|
||||
|
||||
assert [m["content"] for m in leading] == ["one", "two"]
|
||||
assert _roles(later) == ["user", "system"]
|
||||
|
||||
|
||||
def test_flagged_placement_moves_a_system_run_after_the_user_turn_that_follows_it():
|
||||
messages = [
|
||||
{"role": "user", "content": "q1"},
|
||||
{"role": "assistant", "content": "a1"},
|
||||
{"role": "system", "content": "reminder"},
|
||||
{"role": "user", "content": "q2"},
|
||||
{"role": "assistant", "content": "a2"},
|
||||
]
|
||||
|
||||
placed = place_mid_conversation_system(messages, supports_mid_conversation_system=True)
|
||||
|
||||
assert _roles(placed) == ["user", "assistant", "user", "system", "assistant"]
|
||||
|
||||
|
||||
def test_flagged_placement_pushes_a_system_between_two_user_turns_after_both():
|
||||
"""Two user turns collapse into one on the wire, and a system message must
|
||||
be followed by an assistant turn or nothing."""
|
||||
messages = [
|
||||
{"role": "user", "content": "q1"},
|
||||
{"role": "system", "content": "reminder"},
|
||||
{"role": "user", "content": "q2"},
|
||||
]
|
||||
|
||||
placed = place_mid_conversation_system(messages, supports_mid_conversation_system=True)
|
||||
|
||||
assert _roles(placed) == ["user", "user", "system"]
|
||||
|
||||
|
||||
def test_flagged_placement_keeps_a_system_after_tool_results():
|
||||
messages = [
|
||||
{"role": "user", "content": "q1"},
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": None,
|
||||
"tool_calls": [{"id": "c1", "type": "function", "function": {"name": "f", "arguments": "{}"}}],
|
||||
},
|
||||
{"role": "tool", "tool_call_id": "c1", "content": "r"},
|
||||
{"role": "system", "content": "reminder"},
|
||||
{"role": "assistant", "content": "a2"},
|
||||
]
|
||||
|
||||
placed = place_mid_conversation_system(messages, supports_mid_conversation_system=True)
|
||||
|
||||
assert _roles(placed) == ["user", "assistant", "tool", "system", "assistant"]
|
||||
|
||||
|
||||
def test_flagged_placement_drops_a_system_message_with_no_text():
|
||||
messages = [
|
||||
{"role": "user", "content": "q1"},
|
||||
{"role": "system", "content": ""},
|
||||
{"role": "assistant", "content": "a1"},
|
||||
]
|
||||
|
||||
placed = place_mid_conversation_system(messages, supports_mid_conversation_system=True)
|
||||
|
||||
assert _roles(placed) == ["user", "assistant"]
|
||||
|
||||
|
||||
def test_placement_reads_roles_off_pydantic_messages_in_the_history():
|
||||
"""Callers routinely append the previous ``litellm.Message`` object straight
|
||||
into the history; placement must read its role without assuming a dict and
|
||||
hand the object through untouched."""
|
||||
assistant = litellm.Message(role="assistant", content="a1")
|
||||
messages = [
|
||||
{"role": "user", "content": "q1"},
|
||||
assistant,
|
||||
{"role": "system", "content": "reminder"},
|
||||
{"role": "user", "content": "q2"},
|
||||
]
|
||||
|
||||
placed = place_mid_conversation_system(messages, supports_mid_conversation_system=True)
|
||||
|
||||
assert _roles(placed) == ["user", "assistant", "user", "system"]
|
||||
assert placed[1] is assistant
|
||||
|
||||
|
||||
def test_unflagged_conversion_keeps_the_client_order_when_no_tool_result_follows():
|
||||
messages = [
|
||||
{"role": "user", "content": "q1"},
|
||||
{"role": "assistant", "content": "a1"},
|
||||
{"role": "system", "content": "reminder"},
|
||||
{"role": "user", "content": "q2"},
|
||||
]
|
||||
|
||||
placed = place_mid_conversation_system(messages, supports_mid_conversation_system=False)
|
||||
|
||||
assert _roles(placed) == ["user", "assistant", "user", "user"]
|
||||
assert _texts(placed[2]) == [CONVERTED_SYSTEM_NOTE, "reminder"]
|
||||
|
||||
|
||||
def test_unflagged_conversion_rebuilds_cache_control_on_the_converted_block():
|
||||
messages = [
|
||||
{"role": "user", "content": "q1"},
|
||||
{"role": "system", "content": "reminder", "cache_control": {"type": "ephemeral", "ttl": "1h"}},
|
||||
]
|
||||
|
||||
placed = place_mid_conversation_system(messages, supports_mid_conversation_system=False)
|
||||
|
||||
assert placed[1]["content"][1] == {
|
||||
"type": "text",
|
||||
"text": "reminder",
|
||||
"cache_control": {"type": "ephemeral", "ttl": "1h"},
|
||||
}
|
||||
|
||||
|
||||
def test_placement_is_a_no_op_without_later_system_messages():
|
||||
messages = [{"role": "user", "content": "q1"}, {"role": "assistant", "content": "a1"}]
|
||||
|
||||
assert place_mid_conversation_system(messages, supports_mid_conversation_system=False) == tuple(messages)
|
||||
assert place_mid_conversation_system(messages, supports_mid_conversation_system=True) == tuple(messages)
|
||||
Loading…
Add table
Reference in a new issue