# -*- 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 tags. # The trailing assistant message (is_first=False) is wrapped in a # minimal 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 + "\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" "" ), }, 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, )