encoding/encoding_dsv4.py
28.3 KB · 761 lines · python Raw
1 """
2 DeepSeek-V4 Encoding
3
4 A self-contained implementation for encoding/decoding DeepSeek-V4 chat messages
5 with tool calling, thinking mode, and quick instruction task support.
6 """
7
8 from typing import Any, Dict, List, Union, Optional, Tuple
9 import copy
10 import json
11 import re
12
13 # ============================================================
14 # Special Tokens
15 # ============================================================
16
17 bos_token: str = "<｜begin▁of▁sentence｜>"
18 eos_token: str = "<｜end▁of▁sentence｜>"
19 thinking_start_token: str = "<think>"
20 thinking_end_token: str = "</think>"
21 dsml_token: str = "｜DSML｜"
22
23 USER_SP_TOKEN = "<｜User｜>"
24 ASSISTANT_SP_TOKEN = "<｜Assistant｜>"
25 LATEST_REMINDER_SP_TOKEN = "<｜latest_reminder｜>"
26
27 # Task special tokens for internal classification tasks
28 DS_TASK_SP_TOKENS = {
29 "action": "<｜action｜>",
30 "query": "<｜query｜>",
31 "authority": "<｜authority｜>",
32 "domain": "<｜domain｜>",
33 "title": "<｜title｜>",
34 "read_url": "<｜read_url｜>",
35 }
36 VALID_TASKS = set(DS_TASK_SP_TOKENS.keys())
37
38 # ============================================================
39 # Templates
40 # ============================================================
41
42 system_msg_template: str = "{content}"
43 user_msg_template: str = "{content}"
44 latest_reminder_msg_template: str = "{content}"
45 assistant_msg_template: str = "{reasoning}{content}{tool_calls}" + eos_token
46 assistant_msg_wo_eos_template: str = "{reasoning}{content}{tool_calls}"
47 thinking_template: str = "{reasoning_content}"
48
49 response_format_template: str = (
50 "## Response Format:\n\nYou MUST strictly adhere to the following schema to reply:\n{schema}"
51 )
52 tool_call_template: str = (
53 "<{dsml_token}invoke name=\"{name}\">\n{arguments}\n</{dsml_token}invoke>"
54 )
55 tool_calls_template = (
56 "<{dsml_token}{tc_block_name}>\n{tool_calls}\n</{dsml_token}{tc_block_name}>"
57 )
58 tool_calls_block_name: str = "tool_calls"
59
60 tool_output_template: str = (
61 "<tool_result>{content}</tool_result>"
62 )
63
64 # Reasoning effort levels. In thinking mode, the prompt for the selected level is
65 # prepended at the very beginning of the conversation. `low` is the default and
66 # adds nothing.
67 REASONING_EFFORT_PROMPTS: Dict[str, str] = {
68 "low": "",
69 "high": (
70 "Reasoning Effort: Absolute maximum with no shortcuts permitted.\n"
71 "You MUST be very thorough in your thinking and comprehensively decompose the problem to resolve the root cause, rigorously stress-testing your logic against all potential paths, edge cases, and adversarial scenarios.\n"
72 "Explicitly write out your entire deliberation process, documenting every intermediate step, considered alternative, and rejected hypothesis to ensure absolutely no assumption is left unchecked.\n\n"
73 ),
74 "max": (
75 "Reasoning Effort: Beyond maximum — exhaustive, relentless, and uncompromising.\n"
76 "You MUST reason with the utmost depth and rigor, leaving absolutely nothing to chance: exhaustively decompose the problem into its most fundamental components, trace every causal chain to its root, and resolve the underlying cause rather than any surface symptom.\n"
77 "Do not stop reasoning until you have independently verified the solution from multiple angles and are certain that no assumption remains unchecked and no error remains undiscovered.\n\n"
78 ),
79 }
80 DEFAULT_REASONING_EFFORT = "low"
81
82 TOOLS_TEMPLATE = """## Tools
83
84 You have access to a set of tools to help answer the user's question. You can invoke tools by writing a "<{dsml_token}tool_calls>" block like the following:
85
86 <{dsml_token}tool_calls>
87 <{dsml_token}invoke name="$TOOL_NAME">
88 <{dsml_token}parameter name="$PARAMETER_NAME" string="true|false">$PARAMETER_VALUE</{dsml_token}parameter>
89 ...
90 </{dsml_token}invoke>
91 <{dsml_token}invoke name="$TOOL_NAME2">
92 ...
93 </{dsml_token}invoke>
94 </{dsml_token}tool_calls>
95
96 String parameters should be specified as is and set `string="true"`. For all other types (numbers, booleans, arrays, objects), pass the value in JSON format and set `string="false"`.
97
98 If thinking_mode is enabled (triggered by {thinking_start_token}), you MUST output your complete reasoning inside {thinking_start_token}...{thinking_end_token} BEFORE any tool calls or final response.
99
100 Otherwise, output directly after {thinking_end_token} with tool calls or final response.
101
102 ### Available Tool Schemas
103
104 {tool_schemas}
105
106 You MUST strictly follow the above defined tool name and parameter schemas to invoke tool calls.
107 """
108
109 # ============================================================
110 # Utility Functions
111 # ============================================================
112
113 def to_json(value: Any) -> str:
114 """Serialize a value to JSON string."""
115 try:
116 return json.dumps(value, ensure_ascii=False)
117 except:
118 return json.dumps(value, ensure_ascii=True)
119
120
121 def tools_from_openai_format(tools):
122 """Extract function definitions from OpenAI-format tool list."""
123 return [tool["function"] for tool in tools]
124
125
126 def tool_calls_from_openai_format(tool_calls):
127 """Convert OpenAI-format tool calls to internal format."""
128 return [
129 {
130 "name": tool_call["function"]["name"],
131 "arguments": tool_call["function"]["arguments"],
132 }
133 for tool_call in tool_calls
134 ]
135
136
137 def tool_calls_to_openai_format(tool_calls):
138 """Convert internal tool calls to OpenAI format."""
139 return [
140 {
141 "type": "function",
142 "function": {
143 "name": tool_call["name"],
144 "arguments": tool_call["arguments"],
145 }
146 }
147 for tool_call in tool_calls
148 ]
149
150
151 def encode_arguments_to_dsml(tool_call: Dict[str, str]) -> str:
152 """
153 Encode tool call arguments into DSML parameter format.
154
155 Args:
156 tool_call: Dict with "name" and "arguments" (JSON string) keys.
157
158 Returns:
159 DSML-formatted parameter string.
160 """
161 p_dsml_template = '<{dsml_token}parameter name="{key}" string="{is_str}">{value}</{dsml_token}parameter>'
162 P_dsml_strs = []
163
164 try:
165 arguments = json.loads(tool_call["arguments"])
166 except Exception as err:
167 arguments = {"arguments": tool_call["arguments"]}
168
169 for k, v in arguments.items():
170 p_dsml_str = p_dsml_template.format(
171 dsml_token=dsml_token,
172 key=k,
173 is_str="true" if isinstance(v, str) else "false",
174 value=v if isinstance(v, str) else to_json(v),
175 )
176 P_dsml_strs.append(p_dsml_str)
177
178 return "\n".join(P_dsml_strs)
179
180
181 def decode_dsml_to_arguments(tool_name: str, tool_args: Dict[str, Tuple[str, str]]) -> Dict[str, str]:
182 """
183 Decode DSML parameters back to a tool call dict.
184
185 Args:
186 tool_name: Name of the tool.
187 tool_args: Dict mapping param_name -> (value, is_string_flag).
188
189 Returns:
190 Dict with "name" and "arguments" (JSON string) keys.
191 """
192 def _decode_value(key: str, value: str, string: str):
193 if string == "true":
194 value = to_json(value)
195 return f"{to_json(key)}: {value}"
196
197 tool_args_json = "{" + ", ".join([_decode_value(k, v, string=is_str) for k, (v, is_str) in tool_args.items()]) + "}"
198 return dict(name=tool_name, arguments=tool_args_json)
199
200
201 def render_tools(tools: List[Dict[str, Union[str, Dict[str, Any]]]]) -> str:
202 """
203 Render tool schemas into the system prompt format.
204
205 Args:
206 tools: List of tool schema dicts (each with name, description, parameters).
207
208 Returns:
209 Formatted tools section string.
210 """
211 tools_json = [to_json(t) for t in tools]
212
213 return TOOLS_TEMPLATE.format(
214 tool_schemas="\n".join(tools_json),
215 dsml_token=dsml_token,
216 thinking_start_token=thinking_start_token,
217 thinking_end_token=thinking_end_token,
218 )
219
220
221 def find_last_user_index(messages: List[Dict[str, Any]]) -> int:
222 """Find the index of the last user/developer message."""
223 last_user_index = -1
224 for idx in range(len(messages) - 1, -1, -1):
225 if messages[idx].get("role") in ["user", "developer"]:
226 last_user_index = idx
227 break
228 return last_user_index
229
230
231 # ============================================================
232 # Message Rendering
233 # ============================================================
234
235 def render_message(index: int, messages: List[Dict[str, Any]], thinking_mode: str, drop_thinking: bool = True, reasoning_effort: Optional[str] = None) -> str:
236 """
237 Render a single message at the given index into its encoded string form.
238
239 This is the core function that converts each message in the conversation
240 into the DeepSeek-V4 format.
241
242 Args:
243 index: Index of the message to render.
244 messages: Full list of messages in the conversation.
245 thinking_mode: Either "chat" or "thinking".
246 drop_thinking: Whether to drop reasoning content from earlier turns.
247 reasoning_effort: Reasoning effort level, one of "low", "high", "max".
248 None is treated as "low".
249
250 Returns:
251 Encoded string for this message.
252 """
253 assert 0 <= index < len(messages)
254 assert thinking_mode in ["chat", "thinking"], f"Invalid thinking_mode `{thinking_mode}`"
255
256 prompt = ""
257 msg = messages[index]
258 last_user_idx = find_last_user_index(messages)
259
260 role = msg.get("role")
261 content = msg.get("content")
262 tools = msg.get("tools")
263 response_format = msg.get("response_format")
264 tool_calls = msg.get("tool_calls")
265 reasoning_content = msg.get("reasoning_content")
266 wo_eos = msg.get("wo_eos", False)
267
268 if tools:
269 tools = tools_from_openai_format(tools)
270 if tool_calls:
271 tool_calls = tool_calls_from_openai_format(tool_calls)
272
273 # Reasoning effort prefix (only at index 0 in thinking mode; "low" adds nothing)
274 reasoning_effort = reasoning_effort or DEFAULT_REASONING_EFFORT
275 assert reasoning_effort in REASONING_EFFORT_PROMPTS, \
276 f"Invalid reasoning effort: {reasoning_effort}, expected one of {list(REASONING_EFFORT_PROMPTS)}"
277 if index == 0 and thinking_mode == "thinking":
278 prompt += REASONING_EFFORT_PROMPTS[reasoning_effort]
279
280 if role == "system":
281 prompt += system_msg_template.format(content=content or "")
282 if tools:
283 prompt += "\n\n" + render_tools(tools)
284 if response_format:
285 prompt += "\n\n" + response_format_template.format(schema=to_json(response_format))
286
287 elif role == "developer":
288 assert content, f"Invalid message for role `{role}`: {msg}"
289
290 content_developer = USER_SP_TOKEN
291 content_developer += content
292
293 if tools:
294 content_developer += "\n\n" + render_tools(tools)
295 if response_format:
296 content_developer += "\n\n" + response_format_template.format(schema=to_json(response_format))
297
298 prompt += user_msg_template.format(content=content_developer)
299
300 elif role == "user":
301 prompt += USER_SP_TOKEN
302
303 # Handle content blocks (tool results mixed with text)
304 content_blocks = msg.get("content_blocks")
305 if content_blocks:
306 parts = []
307 for block in content_blocks:
308 block_type = block.get("type")
309 if block_type == "text":
310 parts.append(block.get("text", ""))
311 elif block_type == "tool_result":
312 tool_content = block.get("content", "")
313 if isinstance(tool_content, list):
314 text_parts = []
315 for b in tool_content:
316 if b.get("type") == "text":
317 text_parts.append(b.get("text", ""))
318 else:
319 text_parts.append(f"[Unsupported {b.get('type')}]")
320 tool_content = "\n\n".join(text_parts)
321 parts.append(tool_output_template.format(content=tool_content))
322 else:
323 parts.append(f"[Unsupported {block_type}]")
324 prompt += "\n\n".join(parts)
325 else:
326 prompt += content or ""
327
328 elif role == "latest_reminder":
329 prompt += LATEST_REMINDER_SP_TOKEN + latest_reminder_msg_template.format(content=content)
330
331 elif role == "tool":
332 raise NotImplementedError("deepseek_v4 merges tool messages into user; please preprocess with merge_tool_messages()")
333
334 elif role == "assistant":
335 thinking_part = ""
336 tc_content = ""
337
338 if tool_calls:
339 tc_list = [
340 tool_call_template.format(
341 dsml_token=dsml_token,
342 name=tc.get("name"),
343 arguments=encode_arguments_to_dsml(tc)
344 )
345 for tc in tool_calls
346 ]
347 tc_content += '\n\n' + tool_calls_template.format(
348 dsml_token=dsml_token,
349 tool_calls="\n".join(tc_list),
350 tc_block_name=tool_calls_block_name,
351 )
352
353 summary_content = content or ""
354 rc = reasoning_content or ""
355
356 # Check if previous message has a task - if so, this is a task output (no thinking)
357 prev_has_task = index - 1 >= 0 and messages[index - 1].get("task") is not None
358
359 if thinking_mode == "thinking" and not prev_has_task:
360 if not drop_thinking or index > last_user_idx:
361 thinking_part = thinking_template.format(reasoning_content=rc) + thinking_end_token
362 else:
363 thinking_part = ""
364
365 if wo_eos:
366 prompt += assistant_msg_wo_eos_template.format(
367 reasoning=thinking_part,
368 content=summary_content,
369 tool_calls=tc_content,
370 )
371 else:
372 prompt += assistant_msg_template.format(
373 reasoning=thinking_part,
374 content=summary_content,
375 tool_calls=tc_content,
376 )
377 else:
378 raise NotImplementedError(f"Unknown role: {role}")
379
380 # Append transition tokens based on what follows
381 if index + 1 < len(messages) and messages[index + 1].get("role") not in ["assistant", "latest_reminder"]:
382 return prompt
383
384 task = messages[index].get("task")
385 if task is not None:
386 # Task special token for internal classification tasks
387 assert task in VALID_TASKS, f"Invalid task: '{task}'. Valid tasks are: {list(VALID_TASKS)}"
388 task_sp_token = DS_TASK_SP_TOKENS[task]
389
390 if task != "action":
391 # Non-action tasks: append task sp token directly after the message
392 prompt += task_sp_token
393 else:
394 # Action task: append Assistant + thinking token + action sp token
395 prompt += ASSISTANT_SP_TOKEN
396 prompt += thinking_end_token if thinking_mode != "thinking" else thinking_start_token
397 prompt += task_sp_token
398
399 elif messages[index].get("role") in ["user", "developer"]:
400 # Normal generation: append Assistant + thinking token
401 prompt += ASSISTANT_SP_TOKEN
402 if not drop_thinking and thinking_mode == "thinking":
403 prompt += thinking_start_token
404 elif drop_thinking and thinking_mode == "thinking" and index >= last_user_idx:
405 prompt += thinking_start_token
406 else:
407 prompt += thinking_end_token
408
409 return prompt
410
411
412 # ============================================================
413 # Preprocessing
414 # ============================================================
415
416 def merge_tool_messages(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
417 """
418 Merge tool messages into the preceding user message using content_blocks format.
419
420 DeepSeek-V4 does not have a standalone "tool" role; instead, tool results
421 are encoded as <tool_result> blocks within user messages.
422
423 This function converts a standard OpenAI-format conversation (with separate
424 "tool" role messages) into V4 format where tool results are merged into
425 user messages.
426
427 Args:
428 messages: List of message dicts in OpenAI format.
429
430 Returns:
431 Processed message list with tool messages merged into user messages.
432 """
433 merged: List[Dict[str, Any]] = []
434
435 for msg in messages:
436 msg = copy.deepcopy(msg)
437 role = msg.get("role")
438
439 if role == "tool":
440 # Convert tool message to a user message with tool_result block
441 tool_block = {
442 "type": "tool_result",
443 "tool_use_id": msg.get("tool_call_id", ""),
444 "content": msg.get("content", ""),
445 }
446 # Merge into previous message if it's already a user (merged tool)
447 if merged and merged[-1].get("role") == "user" and "content_blocks" in merged[-1]:
448 merged[-1]["content_blocks"].append(tool_block)
449 else:
450 merged.append({
451 "role": "user",
452 "content_blocks": [tool_block],
453 })
454 elif role == "user":
455 text_block = {"type": "text", "text": msg.get("content", "")}
456 if merged and merged[-1].get("role") == "user" and "content_blocks" in merged[-1] and merged[-1].get("task") is None:
457 merged[-1]["content_blocks"].append(text_block)
458 else:
459 new_msg = {
460 "role": "user",
461 "content": msg.get("content", ""),
462 "content_blocks": [text_block],
463 }
464 # Preserve extra fields (task, wo_eos, mask, etc.)
465 for key in ("task", "wo_eos", "mask"):
466 if key in msg:
467 new_msg[key] = msg[key]
468 merged.append(new_msg)
469 else:
470 merged.append(msg)
471
472 return merged
473
474
475 def sort_tool_results_by_call_order(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
476 """
477 Sort tool_result blocks within user messages by the order of tool_calls
478 in the preceding assistant message.
479
480 Args:
481 messages: Preprocessed message list (after merge_tool_messages).
482
483 Returns:
484 Message list with sorted tool result blocks.
485 """
486 last_tool_call_order: Dict[str, int] = {}
487
488 for msg in messages:
489 role = msg.get("role")
490 if role == "assistant" and msg.get("tool_calls"):
491 last_tool_call_order = {}
492 for idx, tc in enumerate(msg["tool_calls"]):
493 tc_id = tc.get("id") or tc.get("function", {}).get("id", "")
494 if tc_id:
495 last_tool_call_order[tc_id] = idx
496
497 elif role == "user" and msg.get("content_blocks"):
498 tool_blocks = [b for b in msg["content_blocks"] if b.get("type") == "tool_result"]
499 if len(tool_blocks) > 1 and last_tool_call_order:
500 sorted_blocks = sorted(
501 tool_blocks,
502 key=lambda b: last_tool_call_order.get(b.get("tool_use_id", ""), 0)
503 )
504 sorted_idx = 0
505 new_blocks = []
506 for block in msg["content_blocks"]:
507 if block.get("type") == "tool_result":
508 new_blocks.append(sorted_blocks[sorted_idx])
509 sorted_idx += 1
510 else:
511 new_blocks.append(block)
512 msg["content_blocks"] = new_blocks
513
514 return messages
515
516
517 # ============================================================
518 # Main Encoding Function
519 # ============================================================
520
521 def encode_messages(
522 messages: List[Dict[str, Any]],
523 thinking_mode: str,
524 context: Optional[List[Dict[str, Any]]] = None,
525 drop_thinking: bool = True,
526 add_default_bos_token: bool = True,
527 reasoning_effort: Optional[str] = None,
528 ) -> str:
529 """
530 Encode a list of messages into the DeepSeek-V4 prompt format.
531
532 This is the main entry point for encoding conversations. It handles:
533 - BOS token insertion
534 - Thinking mode with optional reasoning content dropping
535 - Tool message merging into user messages
536 - Multi-turn conversation context
537
538 Args:
539 messages: List of message dicts to encode.
540 thinking_mode: Either "chat" or "thinking".
541 context: Optional preceding context messages (already encoded prefix).
542 drop_thinking: If True, drop reasoning_content from earlier assistant turns
543 (only keep reasoning for messages after the last user message).
544 add_default_bos_token: Whether to prepend BOS token at conversation start.
545 reasoning_effort: Reasoning effort level, one of "low", "high", "max".
546 Only takes effect in thinking mode. None is treated as "low".
547
548 Returns:
549 The encoded prompt string.
550 """
551 context = context if context else []
552
553 # Preprocess: merge tool messages and sort tool results
554 messages = merge_tool_messages(messages)
555 messages = sort_tool_results_by_call_order(context + messages)[len(context):]
556 if context:
557 context = merge_tool_messages(context)
558 context = sort_tool_results_by_call_order(context)
559
560 full_messages = context + messages
561
562 prompt = bos_token if add_default_bos_token and len(context) == 0 else ""
563
564 # Resolve drop_thinking: if any message has tools defined, don't drop thinking
565 effective_drop_thinking = drop_thinking
566 if any(m.get("tools") for m in full_messages):
567 effective_drop_thinking = False
568
569 if thinking_mode == "thinking" and effective_drop_thinking:
570 full_messages = _drop_thinking_messages(full_messages)
571 # After dropping, recalculate how many messages to render
572 # (context may have shrunk too)
573 num_to_render = len(full_messages) - len(_drop_thinking_messages(context))
574 context_len = len(full_messages) - num_to_render
575 else:
576 num_to_render = len(messages)
577 context_len = len(context)
578
579 for idx in range(num_to_render):
580 prompt += render_message(
581 idx + context_len,
582 full_messages,
583 thinking_mode=thinking_mode,
584 drop_thinking=effective_drop_thinking,
585 reasoning_effort=reasoning_effort,
586 )
587
588 return prompt
589
590
591 def _drop_thinking_messages(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
592 """
593 Drop reasoning_content and non-essential messages before the last user message.
594
595 Behavior:
596 - Messages with role in ["user", "system", "tool", "latest_reminder"] are always kept.
597 - Messages at or after the last user index are always kept.
598 - Assistant messages before the last user get reasoning_content removed.
599 - Developer messages before the last user are dropped entirely.
600 """
601 last_user_idx = find_last_user_index(messages)
602 result = []
603 keep_roles = {"user", "system", "tool", "latest_reminder", "direct_search_results"}
604
605 for idx, msg in enumerate(messages):
606 role = msg.get("role")
607 if role in keep_roles or idx >= last_user_idx:
608 result.append(msg)
609 elif role == "assistant":
610 msg = copy.copy(msg)
611 msg.pop("reasoning_content", None)
612 result.append(msg)
613 # developer and other roles before last_user_idx are dropped
614
615 return result
616
617
618 # ============================================================
619 # Parsing (Decoding model output)
620 # ============================================================
621
622 def _read_until_stop(index: int, text: str, stop: List[str]) -> Tuple[int, str, Optional[str]]:
623 """
624 Read text from index until one of the stop strings is found.
625
626 Returns:
627 Tuple of (new_index, content_before_stop, matched_stop_string_or_None).
628 """
629 min_pos = len(text)
630 matched_stop = None
631
632 for s in stop:
633 pos = text.find(s, index)
634 if pos != -1 and pos < min_pos:
635 min_pos = pos
636 matched_stop = s
637
638 if matched_stop:
639 content = text[index:min_pos]
640 return min_pos + len(matched_stop), content, matched_stop
641 else:
642 content = text[index:]
643 return len(text), content, None
644
645
646 def parse_tool_calls(index: int, text: str) -> Tuple[int, Optional[str], List[Dict[str, str]]]:
647 """
648 Parse DSML tool calls from text starting at the given index.
649
650 Args:
651 index: Starting position in text.
652 text: The full text to parse.
653
654 Returns:
655 Tuple of (new_index, last_stop_token, list_of_tool_call_dicts).
656 Each tool call dict has "name" and "arguments" keys.
657 """
658 tool_calls: List[Dict[str, Any]] = []
659 stop_token = None
660 tool_calls_end_token = f"</{dsml_token}{tool_calls_block_name}>"
661
662 while index < len(text):
663 index, _, stop_token = _read_until_stop(index, text, [f"<{dsml_token}invoke", tool_calls_end_token])
664 if _ != ">\n":
665 raise ValueError(f"Tool call format error: expected '>\\n' but got '{_}'")
666
667 if stop_token == tool_calls_end_token:
668 break
669
670 if stop_token is None:
671 raise ValueError("Missing special token in tool calls")
672
673 index, tool_name_content, stop_token = _read_until_stop(index, text, [f"<{dsml_token}parameter", f"</{dsml_token}invoke"])
674
675 p_tool_name = re.findall(r'^\s*name="(.*?)">\n$', tool_name_content, flags=re.DOTALL)
676 if len(p_tool_name) != 1:
677 raise ValueError(f"Tool name format error: '{tool_name_content}'")
678 tool_name = p_tool_name[0]
679
680 tool_args: Dict[str, Tuple[str, str]] = {}
681 while stop_token == f"<{dsml_token}parameter":
682 index, param_content, stop_token = _read_until_stop(index, text, [f"/{dsml_token}parameter"])
683
684 param_kv = re.findall(r'^ name="(.*?)" string="(true|false)">(.*?)<$', param_content, flags=re.DOTALL)
685 if len(param_kv) != 1:
686 raise ValueError(f"Parameter format error: '{param_content}'")
687 param_name, string, param_value = param_kv[0]
688
689 if param_name in tool_args:
690 raise ValueError(f"Duplicate parameter name: '{param_name}'")
691 tool_args[param_name] = (param_value, string)
692
693 index, content, stop_token = _read_until_stop(index, text, [f"<{dsml_token}parameter", f"</{dsml_token}invoke"])
694 if content != ">\n":
695 raise ValueError(f"Parameter format error: expected '>\\n' but got '{content}'")
696
697 tool_call = decode_dsml_to_arguments(tool_name=tool_name, tool_args=tool_args)
698 tool_calls.append(tool_call)
699
700 return index, stop_token, tool_calls
701
702
703 def parse_message_from_completion_text(text: str, thinking_mode: str) -> Dict[str, Any]:
704 """
705 Parse a model completion text into a structured assistant message.
706
707 This function takes the raw text output from the model (a single assistant turn)
708 and extracts:
709 - reasoning_content (thinking block)
710 - content (summary/response)
711 - tool_calls (if any)
712
713 NOTE: This function is designed to parse only correctly formatted strings and
714 will raise ValueError for malformed output.
715
716 Args:
717 text: The raw completion text (including EOS token).
718 thinking_mode: Either "chat" or "thinking".
719
720 Returns:
721 Dict with keys: "role", "content", "reasoning_content", "tool_calls".
722 tool_calls are in OpenAI format.
723 """
724 summary_content, reasoning_content, tool_calls = "", "", []
725 index, stop_token = 0, None
726 tool_calls_start_token = f"\n\n<{dsml_token}{tool_calls_block_name}"
727
728 is_thinking = thinking_mode == "thinking"
729 is_tool_calling = False
730
731 if is_thinking:
732 index, content_delta, stop_token = _read_until_stop(index, text, [thinking_end_token, tool_calls_start_token])
733 reasoning_content = content_delta
734 assert stop_token == thinking_end_token, "Invalid thinking format: missing </think>"
735
736 index, content_delta, stop_token = _read_until_stop(index, text, [eos_token, tool_calls_start_token])
737 summary_content = content_delta
738 if stop_token == tool_calls_start_token:
739 is_tool_calling = True
740 else:
741 assert stop_token == eos_token, "Invalid format: missing EOS token"
742
743 if is_tool_calling:
744 index, stop_token, tool_calls = parse_tool_calls(index, text)
745
746 index, tool_ends_text, stop_token = _read_until_stop(index, text, [eos_token])
747 assert not tool_ends_text, "Unexpected content after tool calls"
748
749 assert len(text) == index and stop_token in [eos_token, None], "Unexpected content at end"
750
751 for sp_token in [bos_token, eos_token, thinking_start_token, thinking_end_token, dsml_token]:
752 assert sp_token not in summary_content and sp_token not in reasoning_content, \
753 f"Unexpected special token '{sp_token}' in content"
754
755 return {
756 "role": "assistant",
757 "content": summary_content,
758 "reasoning_content": reasoning_content,
759 "tool_calls": tool_calls_to_openai_format(tool_calls)
760 }
761