mirror of
https://github.com/NCBM/plyngent.git
synced 2026-11-07 15:37:55 +08:00
NCBM
f89c8dec31
`provider_blocks` means something only to the surface that produced them (Anthropic `thinking` / server-tool blocks). A session that switched surface would otherwise send those dicts to a chat-completions provider as an assistant-message key it does not know; the request normalization drops the field, on a copy, so the history keeps it.
122 lines
4.5 KiB
Python
122 lines
4.5 KiB
Python
"""Compatibility normalization for chat-completions wire requests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Sequence
|
|
|
|
import msgspec
|
|
from msgspec import UNSET
|
|
|
|
from plyngent.lmproto.deepseek.openai_compat.model import ChatCompletionsParam as DeepseekChatCompletionsParam
|
|
from plyngent.lmproto.openai_compatible.compat import (
|
|
coerce_chat_completions_param,
|
|
coerce_chat_completions_param_any,
|
|
coerce_chat_messages,
|
|
coerce_developer_messages_to_system,
|
|
developer_to_system_message,
|
|
)
|
|
from plyngent.lmproto.openai_compatible.model import (
|
|
AnyChatMessage,
|
|
AssistantChatMessage,
|
|
ChatCompletionsParam,
|
|
DeveloperChatMessage,
|
|
SystemChatMessage,
|
|
UserChatMessage,
|
|
)
|
|
|
|
|
|
def _roles(messages: Sequence[object]) -> list[str]:
|
|
raw = msgspec.json.decode(msgspec.json.encode(messages))
|
|
return [str(item["role"]) for item in raw]
|
|
|
|
|
|
def test_developer_to_system_message_preserves_content_and_name() -> None:
|
|
msg = DeveloperChatMessage(content="follow these rules", name="policy")
|
|
converted = developer_to_system_message(msg)
|
|
assert isinstance(converted, SystemChatMessage)
|
|
assert converted.content == "follow these rules"
|
|
assert converted.name == "policy"
|
|
|
|
|
|
def test_coerce_developer_messages_to_system() -> None:
|
|
messages: list[AnyChatMessage] = [
|
|
SystemChatMessage(content="system"),
|
|
DeveloperChatMessage(content="dev"),
|
|
UserChatMessage(content="hi"),
|
|
AssistantChatMessage(content="hello"),
|
|
]
|
|
coerced = coerce_developer_messages_to_system(messages)
|
|
assert coerced is not messages
|
|
assert _roles(coerced) == ["system", "system", "user", "assistant"]
|
|
assert coerced[1].content == "dev"
|
|
|
|
|
|
def test_coerce_returns_equal_list_when_no_developer() -> None:
|
|
messages: list[AnyChatMessage] = [SystemChatMessage(content="system"), UserChatMessage(content="hi")]
|
|
assert coerce_developer_messages_to_system(messages) == messages
|
|
|
|
|
|
def test_coerce_chat_param_encodes_without_developer_role() -> None:
|
|
param = ChatCompletionsParam(
|
|
model="model",
|
|
messages=[
|
|
DeveloperChatMessage(content="dev"),
|
|
UserChatMessage(content="hi"),
|
|
],
|
|
)
|
|
coerced = coerce_chat_completions_param(param)
|
|
raw = msgspec.json.decode(msgspec.json.encode(coerced))
|
|
assert raw["messages"][0]["role"] == "system"
|
|
assert raw["messages"][1]["role"] == "user"
|
|
|
|
|
|
def test_deepseek_param_accepts_coerced_messages() -> None:
|
|
base = ChatCompletionsParam(
|
|
model="deepseek",
|
|
messages=[DeveloperChatMessage(content="dev"), UserChatMessage(content="hi")],
|
|
)
|
|
coerced = coerce_chat_completions_param(base)
|
|
raw = msgspec.json.decode(msgspec.json.encode(coerced))
|
|
deepseek_param = msgspec.convert(raw, DeepseekChatCompletionsParam)
|
|
assert _roles(deepseek_param.messages) == ["system", "user"]
|
|
|
|
|
|
def test_generic_coercion_handles_deepseek_runtime_param() -> None:
|
|
base = ChatCompletionsParam(
|
|
model="deepseek",
|
|
messages=[DeveloperChatMessage(content="dev"), UserChatMessage(content="hi")],
|
|
)
|
|
# Runtime calls may pass a generic param through a DeepSeek-typed method.
|
|
coerced = coerce_chat_completions_param_any(base)
|
|
raw = msgspec.json.decode(msgspec.json.encode(coerced))
|
|
assert raw["messages"][0]["role"] == "system"
|
|
|
|
|
|
def test_coerce_chat_messages_drops_provider_blocks() -> None:
|
|
"""Another surface's opaque blocks are not a chat-completions key."""
|
|
messages: list[AnyChatMessage] = [
|
|
UserChatMessage(content="hi"),
|
|
AssistantChatMessage(
|
|
content="done",
|
|
reasoning_content="hmm",
|
|
provider_blocks=[{"type": "thinking", "thinking": "hmm", "signature": "sig-1"}],
|
|
),
|
|
]
|
|
coerced = coerce_chat_messages(messages)
|
|
assistant = coerced[1]
|
|
assert isinstance(assistant, AssistantChatMessage)
|
|
assert assistant.provider_blocks is UNSET
|
|
# The reasoning slot is the provider-neutral one and stays.
|
|
assert assistant.reasoning_content == "hmm"
|
|
raw = msgspec.json.decode(msgspec.json.encode(ChatCompletionsParam(model="m", messages=coerced)))
|
|
assert "provider_blocks" not in raw["messages"][1]
|
|
# The history the copy came from keeps them for the surface that made them.
|
|
original = messages[1]
|
|
assert isinstance(original, AssistantChatMessage)
|
|
assert original.provider_blocks is not UNSET
|
|
|
|
|
|
def test_coerce_chat_messages_leaves_plain_history_alone() -> None:
|
|
messages: list[AnyChatMessage] = [UserChatMessage(content="hi"), AssistantChatMessage(content="done")]
|
|
assert coerce_chat_messages(messages) == messages
|