wehub/agentscope-ai--agentscope
1
0
Fork 0
Code Issues Pull Requests Actions Deployments Models Agent Notes Packages Projects Releases Wiki Activity
Files
main
agentscope-ai--agentscope/tests/formatter_deepseek_test.py
T
wehub-resource-sync c3bf08ac8d
K8s Workspace Integration Tests / k8s-workspace-tests (push) Has been cancelled
Details
Pre-commit / run (ubuntu-latest) (push) Has been cancelled
Details
Python Unittest Coverage / test (macos-15, 3.11) (push) Has been cancelled
Details
Python Unittest Coverage / test (ubuntu-latest, 3.11) (push) Has been cancelled
Details
Python Unittest Coverage / test (windows-latest, 3.11) (push) Has been cancelled
Details
Web UI / check (push) Has been cancelled
Details
chore: import upstream snapshot with attribution
2026-07-13 12:39:27 +08:00

533 lines
18 KiB
Python
Raw Permalink Blame History

# -*- coding: utf-8 -*-
"""Comprehensive formatter unit tests for DeepSeekChatFormatter and
DeepSeekMultiAgentFormatter, with exact ground-truth comparisons.
"""
from unittest import IsolatedAsyncioTestCase
from agentscope.formatter import (
DeepSeekChatFormatter,
DeepSeekMultiAgentFormatter,
)
from agentscope.message import (
UserMsg,
AssistantMsg,
SystemMsg,
TextBlock,
DataBlock,
Base64Source,
ToolCallBlock,
ToolResultBlock,
ToolResultState,
ThinkingBlock,
HintBlock,
)
class TestDeepSeekFormatter(IsolatedAsyncioTestCase):
"""Comprehensive tests for DeepSeek Chat and MultiAgent formatters."""
async def asyncSetUp(self) -> None:
"""Set up shared fixtures and ground-truth dicts."""
_hist_prompt = (
DeepSeekMultiAgentFormatter().conversation_history_prompt
)
self.msgs_system = [
SystemMsg(
name="system",
content="You're a helpful assistant.",
),
]
self.msgs_conversation = [
UserMsg(
name="user",
content="What is the capital of France?",
),
AssistantMsg(
name="assistant",
content="The capital of France is Paris.",
),
UserMsg(
name="user",
content="What is the capital of Germany?",
),
AssistantMsg(
name="assistant",
content="The capital of Germany is Berlin.",
),
UserMsg(
name="user",
content="What is the capital of Japan?",
),
]
self.msgs_tools = [
AssistantMsg(
name="assistant",
content=[
ToolCallBlock(
id="call_1",
name="get_capital",
input='{"country": "Japan"}',
),
ToolResultBlock(
id="call_1",
name="get_capital",
output=[
TextBlock(text="The capital of Japan is Tokyo."),
],
state=ToolResultState.SUCCESS,
),
TextBlock(text="The capital of Japan is Tokyo."),
],
),
]
# --- Chat formatter ground truth ---
# DeepSeek content is a plain string (not a list of blocks).
# All assistant messages include `reasoning_content` (empty string if
# no ThinkingBlock).
self.gt_chat = [
{"role": "system", "content": "You're a helpful assistant."},
{"role": "user", "content": "What is the capital of France?"},
{
"role": "assistant",
"content": "The capital of France is Paris.",
"reasoning_content": "",
},
{"role": "user", "content": "What is the capital of Germany?"},
{
"role": "assistant",
"content": "The capital of Germany is Berlin.",
"reasoning_content": "",
},
{"role": "user", "content": "What is the capital of Japan?"},
{
"role": "assistant",
"content": None,
"reasoning_content": "",
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {
"name": "get_capital",
"arguments": '{"country": "Japan"}',
},
},
],
},
{
"role": "tool",
"tool_call_id": "call_1",
"content": "The capital of Japan is Tokyo.",
"name": "get_capital",
},
{
"role": "assistant",
"content": "The capital of Japan is Tokyo.",
"reasoning_content": "",
},
]
# --- MultiAgent formatter ground truth ---
# System content is a plain string.
# History is a plain string (not a list) with <history> tags.
# The trailing assistant message (is_first=False) is wrapped in a
# minimal <history> block without the full prompt prefix.
self._gt_trailing_asst = {
"role": "assistant",
"content": "The capital of Japan is Tokyo.",
"reasoning_content": "",
}
self._gt_tool_call = {
"role": "assistant",
"content": None,
"reasoning_content": "",
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {
"name": "get_capital",
"arguments": '{"country": "Japan"}',
},
},
],
}
self._gt_tool_result = {
"role": "tool",
"tool_call_id": "call_1",
"content": "The capital of Japan is Tokyo.",
"name": "get_capital",
}
self.gt_multiagent = [
{"role": "system", "content": "You're a helpful assistant."},
{
"role": "user",
"content": (
_hist_prompt + "<history>\n"
"user: What is the capital of France?\n"
"assistant: The capital of France is Paris.\n"
"user: What is the capital of Germany?\n"
"assistant: The capital of Germany is Berlin.\n"
"user: What is the capital of Japan?\n"
"</history>"
),
},
self._gt_tool_call,
self._gt_tool_result,
self._gt_trailing_asst,
]
# ------------------------------------------------------------------
# DeepSeekChatFormatter tests
# ------------------------------------------------------------------
async def test_chat_formatter(self) -> None:
"""Chat formatter produces exact output for various subsets."""
fmt = DeepSeekChatFormatter()
# Full history
res = await fmt.format(
[*self.msgs_system, *self.msgs_conversation, *self.msgs_tools],
)
self.assertListEqual(self.gt_chat, res)
# Without system
res = await fmt.format([*self.msgs_conversation, *self.msgs_tools])
self.assertListEqual(self.gt_chat[1:], res)
# Without conversation
n_tools_gt = len(self.gt_chat) - 1 - len(self.msgs_conversation)
res = await fmt.format([*self.msgs_system, *self.msgs_tools])
self.assertListEqual(
[self.gt_chat[0]] + self.gt_chat[-n_tools_gt:],
res,
)
# Without tools
res = await fmt.format([*self.msgs_system, *self.msgs_conversation])
self.assertListEqual(self.gt_chat[:-n_tools_gt], res)
# Empty
self.assertListEqual([], await fmt.format([]))
async def test_chat_formatter_reasoning_content_always_present(
self,
) -> None:
"""Every assistant message always has a reasoning_content field."""
fmt = DeepSeekChatFormatter()
msgs = [AssistantMsg(name="assistant", content="Answer")]
res = await fmt.format(msgs)
self.assertListEqual(
[
{
"role": "assistant",
"content": "Answer",
"reasoning_content": "",
},
],
res,
)
async def test_chat_formatter_thinking_block(self) -> None:
"""ThinkingBlock is placed into reasoning_content."""
fmt = DeepSeekChatFormatter()
msgs = [
AssistantMsg(
name="assistant",
content=[
ThinkingBlock(thinking="Let me think..."),
TextBlock(text="Answer"),
],
),
]
res = await fmt.format(msgs)
self.assertListEqual(
[
{
"role": "assistant",
"content": "Answer",
"reasoning_content": "Let me think...",
},
],
res,
)
# ------------------------------------------------------------------
# DeepSeekMultiAgentFormatter tests
# ------------------------------------------------------------------
async def test_multiagent_formatter(self) -> None:
"""MultiAgent formatter produces exact output for various subsets."""
fmt = DeepSeekMultiAgentFormatter()
# Full
res = await fmt.format(
[*self.msgs_system, *self.msgs_conversation, *self.msgs_tools],
)
self.assertListEqual(self.gt_multiagent, res)
# Without system
res = await fmt.format([*self.msgs_conversation, *self.msgs_tools])
self.assertListEqual(self.gt_multiagent[1:], res)
# Without tools
res = await fmt.format([*self.msgs_system, *self.msgs_conversation])
self.assertListEqual(self.gt_multiagent[:2], res)
# System only
res = await fmt.format(self.msgs_system)
self.assertListEqual([self.gt_multiagent[0]], res)
# Conversation only
res = await fmt.format(self.msgs_conversation)
self.assertListEqual([self.gt_multiagent[1]], res)
# Tools only
res = await fmt.format(self.msgs_tools)
self.assertListEqual(
[
self._gt_tool_call,
self._gt_tool_result,
self._gt_trailing_asst,
],
res,
)
# System + tools
res = await fmt.format([*self.msgs_system, *self.msgs_tools])
self.assertListEqual(
[
self.gt_multiagent[0],
self._gt_tool_call,
self._gt_tool_result,
self._gt_trailing_asst,
],
res,
)
# Empty
self.assertListEqual([], await fmt.format([]))
async def test_chat_formatter_complex_multi_step(self) -> None:
"""Complex multi-step sequence with interleaved thinking, text,
tool calls, and tool results."""
fmt = DeepSeekChatFormatter()
msgs = [
AssistantMsg(
name="assistant",
content=[
ThinkingBlock(thinking="thinking_1"),
TextBlock(text="text_1"),
ToolCallBlock(
id="call_1",
name="func_1",
input='{"arg": "value1"}',
),
ToolCallBlock(
id="call_2",
name="func_2",
input='{"arg": "value2"}',
),
ToolResultBlock(
id="call_1",
name="func_1",
output=[TextBlock(text="result_1")],
state=ToolResultState.SUCCESS,
),
ToolResultBlock(
id="call_2",
name="func_2",
output=[TextBlock(text="result_2")],
state=ToolResultState.SUCCESS,
),
ThinkingBlock(thinking="thinking_2"),
TextBlock(text="text_2"),
ToolCallBlock(
id="call_3",
name="func_3",
input='{"arg": "value3"}',
),
ToolResultBlock(
id="call_3",
name="func_3",
output=[TextBlock(text="result_3")],
state=ToolResultState.SUCCESS,
),
ToolCallBlock(
id="call_4",
name="func_4",
input='{"arg": "value4"}',
),
ToolResultBlock(
id="call_4",
name="func_4",
output=[TextBlock(text="result_4")],
state=ToolResultState.SUCCESS,
),
ThinkingBlock(thinking="thinking_3"),
TextBlock(text="text_3"),
],
),
]
res = await fmt.format(msgs)
self.assertListEqual(
[
{
"role": "assistant",
"content": "text_1",
"reasoning_content": "thinking_1",
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {
"name": "func_1",
"arguments": '{"arg": "value1"}',
},
},
{
"id": "call_2",
"type": "function",
"function": {
"name": "func_2",
"arguments": '{"arg": "value2"}',
},
},
],
},
{
"role": "tool",
"tool_call_id": "call_1",
"content": "result_1",
"name": "func_1",
},
{
"role": "tool",
"tool_call_id": "call_2",
"content": "result_2",
"name": "func_2",
},
{
"role": "assistant",
"content": "text_2",
"reasoning_content": "thinking_2",
"tool_calls": [
{
"id": "call_3",
"type": "function",
"function": {
"name": "func_3",
"arguments": '{"arg": "value3"}',
},
},
],
},
{
"role": "tool",
"tool_call_id": "call_3",
"content": "result_3",
"name": "func_3",
},
{
"role": "assistant",
"content": None,
"reasoning_content": "",
"tool_calls": [
{
"id": "call_4",
"type": "function",
"function": {
"name": "func_4",
"arguments": '{"arg": "value4"}',
},
},
],
},
{
"role": "tool",
"tool_call_id": "call_4",
"content": "result_4",
"name": "func_4",
},
{
"role": "assistant",
"content": "text_3",
"reasoning_content": "thinking_3",
},
],
res,
)
async def test_chat_formatter_hint_block(self) -> None:
"""HintBlock flushes preceding content and becomes a user message."""
fmt = DeepSeekChatFormatter()
msgs = [
AssistantMsg(
name="assistant",
content=[
TextBlock(text="Let me think about that."),
HintBlock(hint="Remember to be concise."),
TextBlock(text="Here is my answer."),
],
),
]
res = await fmt.format(msgs)
self.assertListEqual(
[
{
"role": "assistant",
"content": "Let me think about that.",
"reasoning_content": "",
},
{
"role": "user",
"content": "Remember to be concise.",
},
{
"role": "assistant",
"content": "Here is my answer.",
"reasoning_content": "",
},
],
res,
)
async def test_chat_formatter_hint_block_multimodal(self) -> None:
"""DeepSeek is text-only — DataBlock degrades to a placeholder
string."""
fmt = DeepSeekChatFormatter()
msgs = [
AssistantMsg(
name="assistant",
content=[
HintBlock(
hint=[
TextBlock(text="Inspect this screenshot:"),
DataBlock(
source=Base64Source(
data="ZmFrZSBpbWFnZSBkYXRh",
media_type="image/png",
),
),
],
),
],
),
]
res = await fmt.format(msgs)
self.assertListEqual(
[
{
"role": "user",
"content": (
"Inspect this screenshot:\n"
"[image/png attached, not supported by this provider]"
),
},
],
res,
)
Reference in New Issue View Git Blame Copy Permalink