mirror of
https://github.com/NCBM/plyngent.git
synced 2026-11-07 15:37:55 +08:00
267 lines
9.1 KiB
Python
267 lines
9.1 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import msgspec
|
|
|
|
from plyngent.agent.responses_bridge import (
|
|
chat_messages_to_responses_input,
|
|
chat_param_to_responses_kwargs,
|
|
reasoning_summary_text,
|
|
response_to_assistant_message,
|
|
response_to_chat_completion,
|
|
responses_status_to_finish_reason,
|
|
tool_items_to_response_tools,
|
|
)
|
|
from plyngent.lmproto.openai.model import Response, ResponseFunctionToolCallOutput
|
|
from plyngent.lmproto.openai_compatible.model import (
|
|
AssistantChatMessage,
|
|
AssistantFunctionTool,
|
|
AssistantFunctionToolCall,
|
|
ChatCompletionsParam,
|
|
SystemChatMessage,
|
|
ToolChatMessage,
|
|
ToolFunction,
|
|
ToolFunctionItem,
|
|
UserChatMessage,
|
|
)
|
|
|
|
|
|
def test_tool_items_to_response_tools() -> None:
|
|
items = [
|
|
ToolFunctionItem(
|
|
function=ToolFunction(
|
|
name="read_file",
|
|
description="Read a file",
|
|
parameters={"type": "object", "properties": {"path": {"type": "string"}}},
|
|
)
|
|
)
|
|
]
|
|
tools = tool_items_to_response_tools(items)
|
|
assert len(tools) == 1
|
|
assert tools[0].name == "read_file"
|
|
|
|
|
|
def test_chat_messages_to_input_and_instructions() -> None:
|
|
messages = [
|
|
SystemChatMessage(content="You are helpful."),
|
|
UserChatMessage(content="hi"),
|
|
AssistantChatMessage(
|
|
content=None,
|
|
tool_calls=[
|
|
AssistantFunctionToolCall(
|
|
id="call_1",
|
|
function=AssistantFunctionTool(name="read_file", arguments='{"path":"a"}'),
|
|
)
|
|
],
|
|
),
|
|
ToolChatMessage(content="file body", tool_call_id="call_1"),
|
|
]
|
|
instructions, items = chat_messages_to_responses_input(messages)
|
|
assert instructions == "You are helpful."
|
|
assert len(items) == 3 # user, function_call, function_call_output
|
|
|
|
|
|
def test_parallel_tool_outputs_interleaved_after_calls() -> None:
|
|
"""Regression: DeepSeek Responses rejects call, call, output, output batches.
|
|
|
|
Each ``function_call_output`` must directly follow its ``function_call`` so
|
|
strict/adjacency-matching servers (DeepSeek) accept parallel tool rounds.
|
|
"""
|
|
messages = [
|
|
UserChatMessage(content="run both"),
|
|
AssistantChatMessage(
|
|
content=None,
|
|
tool_calls=[
|
|
AssistantFunctionToolCall(
|
|
id="call_01_LPBuUG3KTaMDIwUoLwTE9127",
|
|
function=AssistantFunctionTool(name="read_file", arguments='{"path":"a"}'),
|
|
),
|
|
AssistantFunctionToolCall(
|
|
id="call_02_ABCD",
|
|
function=AssistantFunctionTool(name="regex_files", arguments='{"pattern":"todo"}'),
|
|
),
|
|
],
|
|
),
|
|
ToolChatMessage(content="file body", tool_call_id="call_01_LPBuUG3KTaMDIwUoLwTE9127"),
|
|
ToolChatMessage(content="grep hit", tool_call_id="call_02_ABCD"),
|
|
]
|
|
_, items = chat_messages_to_responses_input(messages)
|
|
types = [item.get("type") if isinstance(item, dict) else item.__struct_config__.tag for item in items]
|
|
assert types == [
|
|
"message",
|
|
"function_call",
|
|
"function_call_output",
|
|
"function_call",
|
|
"function_call_output",
|
|
]
|
|
# Each output is paired with its own call, in stream order.
|
|
call_a = items[1]
|
|
out_a = items[2]
|
|
call_b = items[3]
|
|
out_b = items[4]
|
|
assert isinstance(call_a, dict) and isinstance(call_b, dict)
|
|
assert isinstance(out_a, ResponseFunctionToolCallOutput)
|
|
assert isinstance(out_b, ResponseFunctionToolCallOutput)
|
|
assert out_a.call_id == call_a["call_id"]
|
|
assert out_b.call_id == call_b["call_id"]
|
|
|
|
|
|
def test_response_to_assistant_with_tools() -> None:
|
|
raw = {
|
|
"id": "resp_1",
|
|
"object": "response",
|
|
"created_at": 1,
|
|
"model": "gpt-test",
|
|
"status": "completed",
|
|
"output": [
|
|
{
|
|
"id": "msg_1",
|
|
"type": "message",
|
|
"role": "assistant",
|
|
"status": "completed",
|
|
"content": [{"type": "output_text", "text": "done", "annotations": []}],
|
|
},
|
|
{
|
|
"type": "function_call",
|
|
"call_id": "call_9",
|
|
"name": "add",
|
|
"arguments": '{"a":1}',
|
|
"status": "completed",
|
|
},
|
|
],
|
|
"usage": {"input_tokens": 10, "output_tokens": 5, "total_tokens": 15},
|
|
}
|
|
response = msgspec.convert(raw, Response)
|
|
from msgspec import UNSET
|
|
|
|
assistant = response_to_assistant_message(response)
|
|
assert assistant.content == "done"
|
|
assert assistant.tool_calls is not UNSET
|
|
assert isinstance(assistant.tool_calls, list)
|
|
call0 = assistant.tool_calls[0]
|
|
assert isinstance(call0, AssistantFunctionToolCall)
|
|
assert call0.id == "call_9"
|
|
assert call0.function.name == "add"
|
|
completion = response_to_chat_completion(response)
|
|
assert completion.choices[0].message.content == "done"
|
|
assert isinstance(completion.usage, dict)
|
|
assert completion.usage["input_tokens"] == 10
|
|
assert completion.choices[0].finish_reason == "tool_calls"
|
|
|
|
|
|
def _reasoning_response(
|
|
*,
|
|
output_reasoning: bool = False,
|
|
top_level: dict[str, Any] | None = None,
|
|
) -> Response:
|
|
output: list[dict[str, Any]] = []
|
|
if output_reasoning:
|
|
output.append(
|
|
{
|
|
"id": "reason_1",
|
|
"type": "reasoning",
|
|
"summary": [{"type": "summary_text", "text": "openai-style reasoning"}],
|
|
}
|
|
)
|
|
return Response(
|
|
id="resp_r",
|
|
created_at=1,
|
|
model="deepseek-v4-flash",
|
|
output=output,
|
|
reasoning=top_level,
|
|
)
|
|
|
|
|
|
def test_reasoning_summary_text_deepseek_top_level() -> None:
|
|
"""DeepSeek Responses returns reasoning in the top-level ``reasoning`` field."""
|
|
response = _reasoning_response(top_level={"content": [{"type": "reasoning_content", "text": "think step by step"}]})
|
|
assert reasoning_summary_text(response) == "think step by step"
|
|
assistant = response_to_assistant_message(response)
|
|
assert assistant.reasoning_content == "think step by step"
|
|
|
|
|
|
def test_reasoning_summary_text_prefers_top_level() -> None:
|
|
response = _reasoning_response(
|
|
output_reasoning=True,
|
|
top_level={"summary": [{"type": "summary_text", "text": "top-level reasoning"}]},
|
|
)
|
|
assert reasoning_summary_text(response) == "top-level reasoning"
|
|
|
|
|
|
def test_reasoning_summary_text_falls_back_to_output_items() -> None:
|
|
response = _reasoning_response(output_reasoning=True)
|
|
assert reasoning_summary_text(response) == "openai-style reasoning"
|
|
|
|
|
|
def test_reasoning_summary_text_empty_when_absent() -> None:
|
|
assert reasoning_summary_text(_reasoning_response()) == ""
|
|
|
|
|
|
def test_responses_status_to_finish_reason_incomplete() -> None:
|
|
body = {
|
|
"id": "resp_i",
|
|
"object": "response",
|
|
"created_at": 1,
|
|
"model": "gpt-test",
|
|
"status": "incomplete",
|
|
"incomplete_details": {"reason": "max_output_tokens"},
|
|
"output": [
|
|
{
|
|
"id": "msg_1",
|
|
"type": "message",
|
|
"role": "assistant",
|
|
"status": "incomplete",
|
|
"content": [{"type": "output_text", "text": "partial", "annotations": []}],
|
|
}
|
|
],
|
|
}
|
|
resp = msgspec.convert(body, Response)
|
|
assert responses_status_to_finish_reason(resp, has_tool_calls=False) == "length"
|
|
completion = response_to_chat_completion(resp)
|
|
assert completion.choices[0].finish_reason == "length"
|
|
|
|
|
|
def test_chat_param_to_responses_kwargs() -> None:
|
|
param = ChatCompletionsParam(
|
|
model="gpt-test",
|
|
messages=[SystemChatMessage(content="sys"), UserChatMessage(content="hi")],
|
|
tools=[ToolFunctionItem(function=ToolFunction(name="t", parameters={"type": "object"}))],
|
|
temperature=0.2,
|
|
)
|
|
kwargs = chat_param_to_responses_kwargs(param)
|
|
assert kwargs["model"] == "gpt-test"
|
|
assert kwargs["instructions"] == "sys"
|
|
assert kwargs["store"] is False
|
|
assert kwargs["temperature"] == 0.2
|
|
assert len(kwargs["tools"]) == 1
|
|
# No effort configured → no reasoning block.
|
|
assert "reasoning" not in kwargs
|
|
|
|
|
|
def test_chat_param_to_responses_kwargs_reasoning_effort() -> None:
|
|
param = ChatCompletionsParam(
|
|
model="gpt-test",
|
|
messages=[UserChatMessage(content="hi")],
|
|
reasoning_effort="xhigh",
|
|
)
|
|
kwargs = chat_param_to_responses_kwargs(param)
|
|
assert kwargs["reasoning"].effort == "xhigh"
|
|
|
|
|
|
def test_provider_tools_merged_after_local_functions() -> None:
|
|
param = ChatCompletionsParam(
|
|
model="gpt-test",
|
|
messages=[UserChatMessage(content="hi")],
|
|
tools=[ToolFunctionItem(function=ToolFunction(name="local_tool", parameters={"type": "object"}))],
|
|
)
|
|
kwargs = chat_param_to_responses_kwargs(
|
|
param,
|
|
provider_tools=[{"type": "web_search"}, {"type": "file_search", "vector_store_ids": ["vs_1"]}],
|
|
)
|
|
tools = kwargs["tools"]
|
|
assert len(tools) == 3
|
|
assert tools[0].name == "local_tool"
|
|
assert tools[1]["type"] == "web_search"
|
|
assert tools[2]["vector_store_ids"] == ["vs_1"]
|