Spaces:
Running on Zero
Running on Zero
| """Translate common Qwen/OpenClaude textual tool calls to OpenAI payloads.""" | |
| from __future__ import annotations | |
| import ast | |
| import html | |
| import json | |
| import re | |
| import shlex | |
| import uuid | |
| from collections.abc import Mapping | |
| from typing import Any | |
| JSON_TOOL_CALL_RE = re.compile( | |
| r"<(?P<tag>tool_call|function_call)>\s*(?P<payload>\{.*?\})\s*</(?P=tag)>", | |
| re.DOTALL | re.IGNORECASE, | |
| ) | |
| XML_JSON_TOOL_CALL_RE = re.compile( | |
| r"<xml>\s*(?P<payload>\{.*?\})\s*</xml>", | |
| re.DOTALL | re.IGNORECASE, | |
| ) | |
| STANDARD_XML_TOOL_CALL_RE = re.compile( | |
| r"<(?P<tag>tool_call|function_call)>\s*(?P<body>.*?)\s*</(?P=tag)>", | |
| re.DOTALL | re.IGNORECASE, | |
| ) | |
| FUNCTION_NAME_RE = re.compile( | |
| r"<function\s*=\s*(?P<name>[A-Za-z_][\w.-]*)\s*>", | |
| re.IGNORECASE, | |
| ) | |
| PARAMETER_RE = re.compile( | |
| r"<parameter\s*=\s*(?P<key>[A-Za-z_][\w.-]*)\s*>" | |
| r"(?P<value>.*?)</parameter\s*>", | |
| re.DOTALL | re.IGNORECASE, | |
| ) | |
| DASHED_XML_TOOL_CALL_RE = re.compile( | |
| r"<tool-call>\s*<name>\s*(?P<name>[A-Za-z_][\w.-]*)\s*</name>\s*" | |
| r"<arguments>\s*(?P<arguments>.*?)\s*</arguments>\s*</tool-call>", | |
| re.DOTALL | re.IGNORECASE, | |
| ) | |
| NAMED_ARGUMENT_RE = re.compile( | |
| r"<argument\s+name\s*=\s*(?P<quote>[\"'])" | |
| r"(?P<key>[A-Za-z_][\w.-]*)(?P=quote)\s*>" | |
| r"(?P<value>.*?)</argument\s*>", | |
| re.DOTALL | re.IGNORECASE, | |
| ) | |
| ELEMENT_ARGUMENT_RE = re.compile( | |
| r"<(?P<key>[A-Za-z_][\w.-]*)\s*>(?P<value>.*?)</(?P=key)\s*>", | |
| re.DOTALL | re.IGNORECASE, | |
| ) | |
| SELF_CLOSING_TOOL_RE = re.compile( | |
| r"<(?P<name>[A-Za-z_][\w.-]*)\b(?P<attributes>[^<>]*?)/\s*>", | |
| re.DOTALL, | |
| ) | |
| ATTRIBUTE_RE = re.compile( | |
| r'''(?P<key>[A-Za-z_][\w.-]*)\s*=\s*(?: | |
| "(?P<double>(?:\\.|[^"\\])*)" | |
| |'(?P<single>(?:\\.|[^'\\])*)' | |
| )''', | |
| re.DOTALL | re.VERBOSE, | |
| ) | |
| HTML_ENTITY_RE = re.compile( | |
| r"&(?:#[0-9]+|#[xX][0-9A-Fa-f]+|[A-Za-z][A-Za-z0-9]+);" | |
| ) | |
| ASSISTANT_CALLED_TOOL_RE = re.compile( | |
| r"^\s*\[Assistant called tool (?P<name>[A-Za-z_][\w.-]*) " | |
| r"with arguments (?P<arguments>\{.*\})\]\s*$", | |
| re.DOTALL, | |
| ) | |
| TEXTUAL_TOOL_CALL_RE = re.compile( | |
| r"^[ \t]*(?P<name>[A-Za-z_][\w.-]*)[ \t]+" | |
| r"(?:with|using)[ \t]+(?P<arguments>.+?)[ \t]*$", | |
| re.MULTILINE | re.IGNORECASE, | |
| ) | |
| FENCED_JSON_RE = re.compile( | |
| r"^\s*```(?:json)?\s*(?P<payload>\{.*\})\s*```\s*$", | |
| re.DOTALL | re.IGNORECASE, | |
| ) | |
| TOOL_CALL_CLOSE_RE = re.compile( | |
| r"</(?:tool_call|function_call)>\s*$", re.IGNORECASE | |
| ) | |
| XML_JSON_TOOL_CALL_CLOSE_RE = re.compile( | |
| r"<xml>\s*\{.*?\}\s*</xml>\s*$", | |
| re.DOTALL | re.IGNORECASE, | |
| ) | |
| SELF_CLOSING_TOOL_AT_END_RE = re.compile( | |
| r"<(?:tool\b|[A-Z][A-Za-z0-9_.-]*)\b[^<>]*/\s*>\s*(?:```)?\s*$", | |
| re.DOTALL, | |
| ) | |
| KNOWN_TEXTUAL_TOOL_NAMES = frozenset( | |
| { | |
| "agent", | |
| "askuserquestion", | |
| "bash", | |
| "edit", | |
| "enterplanmode", | |
| "glob", | |
| "grep", | |
| "lsp", | |
| "notebookedit", | |
| "read", | |
| "skill", | |
| "task", | |
| "todowrite", | |
| "webfetch", | |
| "websearch", | |
| "write", | |
| } | |
| ) | |
| def _looks_like_tool_payload(payload: object) -> bool: | |
| """Return whether a mapping has the OpenAI/Qwen tool-call shape.""" | |
| if not isinstance(payload, Mapping): | |
| return False | |
| function = payload.get("function") | |
| if isinstance(function, Mapping): | |
| return isinstance(function.get("name"), str) and bool(function.get("name")) | |
| return isinstance(payload.get("name"), str) and bool(payload.get("name")) | |
| def _terminal_json_tool_payload(text: str) -> tuple[int, int, Mapping[str, Any]] | None: | |
| """Recover a complete bare JSON tool call at the end of model output. | |
| Small coder models sometimes obey the JSON schema but omit the surrounding | |
| ``<tool_call>`` tags, occasionally after a short explanatory prefix. The | |
| normal extractor can parse a *pure* JSON response, but the generation | |
| stopping criterion previously failed to stop there, allowing the model to | |
| continue with prose and additional simulated calls. Scan JSON-object starts | |
| and accept only a terminal mapping with a tool-call shape. | |
| """ | |
| candidate_text = text.rstrip() | |
| decoder = json.JSONDecoder() | |
| for start, char in enumerate(candidate_text): | |
| if char != "{": | |
| continue | |
| try: | |
| payload, consumed = decoder.raw_decode(candidate_text[start:]) | |
| except json.JSONDecodeError: | |
| continue | |
| end = start + consumed | |
| if candidate_text[end:].strip(): | |
| continue | |
| if _looks_like_tool_payload(payload): | |
| return start, len(candidate_text), payload | |
| return None | |
| def has_complete_tool_call( | |
| text: str, | |
| allowed_names: set[str] | None = None, | |
| ) -> bool: | |
| """Return true once generation ended a valid supported tool-call form. | |
| When ``allowed_names`` is supplied, a syntactically complete hallucinated | |
| call to an unadvertised function must *not* stop generation. This matters | |
| for OpenClaude because the final parser rejects unadvertised names. | |
| """ | |
| if allowed_names is not None: | |
| calls, _ = extract_tool_calls(text, allowed_names) | |
| return bool(calls) | |
| fenced_json = FENCED_JSON_RE.fullmatch(text) | |
| return bool( | |
| TOOL_CALL_CLOSE_RE.search(text) | |
| or XML_JSON_TOOL_CALL_CLOSE_RE.search(text) | |
| or SELF_CLOSING_TOOL_AT_END_RE.search(text) | |
| or ( | |
| fenced_json | |
| and _looks_like_tool_payload(_mapping_literal(fenced_json.group("payload"))) | |
| ) | |
| or _terminal_json_tool_payload(text) | |
| or any( | |
| match.group("name").casefold() in KNOWN_TEXTUAL_TOOL_NAMES | |
| for match in TEXTUAL_TOOL_CALL_RE.finditer(text) | |
| ) | |
| ) | |
| def _canonical_name(name: Any, allowed_names: set[str] | None) -> str | None: | |
| if not isinstance(name, str) or not name: | |
| return None | |
| if not allowed_names: | |
| return name | |
| by_casefold = {candidate.casefold(): candidate for candidate in allowed_names} | |
| normalized = name.casefold() | |
| canonical = by_casefold.get(normalized) | |
| if canonical is not None: | |
| return canonical | |
| # OpenClaude exposes the legacy Agent executor as Task. Accept both names | |
| # in textual generations while returning the advertised catalog name. | |
| alias = {"agent": "task", "task": "agent"}.get(normalized) | |
| return by_casefold.get(alias) if alias else None | |
| def _coerce_value(value: str) -> Any: | |
| value = _unescape_entities(value.strip()) | |
| try: | |
| return json.loads(value) | |
| except json.JSONDecodeError: | |
| return value | |
| def _decode_attribute(value: str) -> str: | |
| try: | |
| value = json.loads(f'"{value}"') | |
| except json.JSONDecodeError: | |
| pass | |
| return _unescape_entities(value) | |
| def _unescape_entities(value: str) -> str: | |
| """Decode explicit entities without treating a URL's bare ``&`` as HTML. | |
| ``html.unescape`` accepts legacy semicolon-less names such as ``¤``. | |
| That turns a query key like ``¤t_weather`` into ``¤t_weather``. | |
| XML entities are terminated with a semicolon, so only decode that form. | |
| """ | |
| return HTML_ENTITY_RE.sub(lambda match: html.unescape(match.group(0)), value) | |
| def _attributes(raw: str) -> dict[str, str]: | |
| values: dict[str, str] = {} | |
| for match in ATTRIBUTE_RE.finditer(raw): | |
| value = match.group("double") | |
| if value is None: | |
| value = match.group("single") | |
| if value is not None: | |
| values[match.group("key")] = _decode_attribute(value) | |
| return values | |
| def _arguments(value: Any) -> dict[str, Any] | None: | |
| if isinstance(value, Mapping): | |
| return dict(value) | |
| if not isinstance(value, str): | |
| return None | |
| parsed = _mapping_literal(_unescape_entities(value)) | |
| return dict(parsed) if isinstance(parsed, Mapping) else None | |
| def _mapping_literal(value: str) -> Mapping[str, Any] | None: | |
| """Parse JSON or a Python-style mapping without evaluating expressions.""" | |
| try: | |
| parsed = json.loads(value) | |
| except json.JSONDecodeError: | |
| try: | |
| parsed = ast.literal_eval(value) | |
| except (SyntaxError, ValueError): | |
| return None | |
| return parsed if isinstance(parsed, Mapping) else None | |
| def normalize_openai_tool_arguments(value: Any) -> dict[str, Any]: | |
| """Return the mapping required by Qwen3's chat-template ``items`` filter. | |
| OpenAI serializes function arguments as a JSON string, while Qwen2.5's | |
| official template iterates them as a mapping when replaying tool history. | |
| Accept both representations so a completed tool call can be followed by a | |
| tool result without raising a template ``TypeError``. | |
| """ | |
| parsed = _arguments(value) | |
| return parsed if parsed is not None else {} | |
| def _openai_call( | |
| name: Any, | |
| arguments: Any, | |
| allowed_names: set[str] | None, | |
| ) -> dict[str, Any] | None: | |
| canonical_name = _canonical_name(name, allowed_names) | |
| if canonical_name is None: | |
| return None | |
| if isinstance(arguments, str): | |
| parsed = _arguments(arguments) | |
| arguments = parsed if parsed is not None else {} | |
| if not isinstance(arguments, Mapping): | |
| arguments = {} | |
| return { | |
| "id": f"call_{uuid.uuid4().hex[:24]}", | |
| "type": "function", | |
| "function": { | |
| "name": canonical_name, | |
| "arguments": json.dumps( | |
| dict(arguments), ensure_ascii=False, separators=(",", ":") | |
| ), | |
| }, | |
| } | |
| def _payload_call( | |
| payload: Any, | |
| allowed_names: set[str] | None, | |
| ) -> dict[str, Any] | None: | |
| if not isinstance(payload, Mapping): | |
| return None | |
| function = payload.get("function") | |
| if isinstance(function, Mapping): | |
| return _openai_call( | |
| function.get("name"), | |
| function.get("arguments", {}), | |
| allowed_names, | |
| ) | |
| return _openai_call( | |
| payload.get("name"), payload.get("arguments", {}), allowed_names | |
| ) | |
| def _xml_arguments(arguments: str) -> dict[str, Any]: | |
| named = { | |
| match.group("key"): _coerce_value(match.group("value")) | |
| for match in NAMED_ARGUMENT_RE.finditer(arguments) | |
| } | |
| if named: | |
| return named | |
| return { | |
| match.group("key"): _coerce_value(match.group("value")) | |
| for match in ELEMENT_ARGUMENT_RE.finditer(arguments) | |
| } | |
| def _textual_arguments(value: str) -> dict[str, Any] | None: | |
| """Parse Qwen's compact ``tool with key=value`` representation.""" | |
| raw = value.strip().rstrip(";").strip() | |
| mapping = _mapping_literal(raw) | |
| if isinstance(mapping, Mapping): | |
| return dict(mapping) | |
| # Normalize optional whitespace around '=' before shlex handles quoted | |
| # values containing spaces. No expressions are evaluated here. | |
| raw = re.sub( | |
| r"(?P<key>[A-Za-z_][\w.-]*)\s*=\s*", | |
| r"\g<key>=", | |
| raw, | |
| ) | |
| try: | |
| tokens = shlex.split(raw, posix=True) | |
| except ValueError: | |
| return None | |
| arguments: dict[str, Any] = {} | |
| for token in tokens: | |
| if "=" not in token: | |
| continue | |
| key, item = token.split("=", 1) | |
| if not re.fullmatch(r"[A-Za-z_][\w.-]*", key): | |
| continue | |
| arguments[key] = _coerce_value(item) | |
| return arguments or None | |
| def recover_forced_tool_call(text: str, tool_name: str) -> dict[str, Any] | None: | |
| """Recover argument-only JSON when exactly one tool is mandated. | |
| Some OpenAI-compatible coder models occasionally emit only the function | |
| argument object when the caller has already forced a single tool. The | |
| normal parser correctly refuses to guess a function name from that object. | |
| In the *single forced-tool* case, however, the name is unambiguous and the | |
| structured OpenAI call can be reconstructed safely without executing or | |
| evaluating arbitrary text. | |
| """ | |
| raw = text.strip() | |
| fenced = FENCED_JSON_RE.fullmatch(raw) | |
| if fenced: | |
| raw = fenced.group("payload") | |
| payload = _mapping_literal(raw) | |
| if not isinstance(payload, Mapping): | |
| return None | |
| if _looks_like_tool_payload(payload): | |
| return None | |
| arguments: Mapping[str, Any] = payload | |
| nested_arguments = payload.get("arguments") | |
| if len(payload) == 1 and isinstance(nested_arguments, Mapping): | |
| arguments = nested_arguments | |
| return _openai_call(tool_name, arguments, {tool_name}) | |
| def extract_tool_call( | |
| text: str, | |
| allowed_names: set[str] | None = None, | |
| ) -> tuple[dict[str, Any] | None, str]: | |
| """Extract the first supported tool call for backward compatibility.""" | |
| calls, visible = extract_tool_calls(text, allowed_names) | |
| return (calls[0] if calls else None), visible | |
| def extract_tool_calls( | |
| text: str, | |
| allowed_names: set[str] | None = None, | |
| ) -> tuple[list[dict[str, Any]], str]: | |
| """Extract all tool calls while accepting Qwen's common XML variations. | |
| Matching calls deliberately clear visible content. Agent clients should | |
| receive structured OpenAI calls rather than Markdown/XML renditions of the | |
| same calls before they execute the tools. | |
| """ | |
| candidates: list[tuple[int, int, dict[str, Any]]] = [] | |
| for match in JSON_TOOL_CALL_RE.finditer(text): | |
| call = _payload_call( | |
| _mapping_literal(match.group("payload")), | |
| allowed_names, | |
| ) | |
| if call: | |
| candidates.append((match.start(), match.end(), call)) | |
| for match in XML_JSON_TOOL_CALL_RE.finditer(text): | |
| call = _payload_call( | |
| _mapping_literal(match.group("payload")), | |
| allowed_names, | |
| ) | |
| if call: | |
| candidates.append((match.start(), match.end(), call)) | |
| for match in STANDARD_XML_TOOL_CALL_RE.finditer(text): | |
| body = match.group("body") | |
| function = FUNCTION_NAME_RE.search(body) | |
| if function: | |
| call = _openai_call( | |
| function.group("name"), | |
| { | |
| parameter.group("key"): _coerce_value(parameter.group("value")) | |
| for parameter in PARAMETER_RE.finditer(body) | |
| }, | |
| allowed_names, | |
| ) | |
| if call: | |
| candidates.append((match.start(), match.end(), call)) | |
| for match in DASHED_XML_TOOL_CALL_RE.finditer(text): | |
| call = _openai_call( | |
| match.group("name"), | |
| _xml_arguments(match.group("arguments")), | |
| allowed_names, | |
| ) | |
| if call: | |
| candidates.append((match.start(), match.end(), call)) | |
| for match in SELF_CLOSING_TOOL_RE.finditer(text): | |
| tag_name = match.group("name") | |
| attributes = _attributes(match.group("attributes")) | |
| if tag_name.casefold() == "tool": | |
| tool_name = attributes.pop("name", None) | |
| arguments = _arguments( | |
| attributes.pop("arguments", attributes.pop("args", "")) | |
| ) | |
| if arguments is None: | |
| arguments = attributes | |
| else: | |
| tool_name = tag_name | |
| arguments = attributes | |
| call = _openai_call(tool_name, arguments, allowed_names) | |
| if call: | |
| candidates.append((match.start(), match.end(), call)) | |
| for match in TEXTUAL_TOOL_CALL_RE.finditer(text): | |
| arguments = _textual_arguments(match.group("arguments")) | |
| if arguments is None: | |
| continue | |
| call = _openai_call(match.group("name"), arguments, allowed_names) | |
| if call: | |
| candidates.append((match.start(), match.end(), call)) | |
| assistant_called = ASSISTANT_CALLED_TOOL_RE.fullmatch(text) | |
| if assistant_called: | |
| call = _openai_call( | |
| assistant_called.group("name"), | |
| _arguments(assistant_called.group("arguments")), | |
| allowed_names, | |
| ) | |
| if call: | |
| candidates.append((assistant_called.start(), assistant_called.end(), call)) | |
| if not candidates: | |
| fenced_json = FENCED_JSON_RE.fullmatch(text) | |
| raw_json = fenced_json.group("payload") if fenced_json else text.strip() | |
| call = _payload_call(_mapping_literal(raw_json), allowed_names) | |
| if call: | |
| candidates.append((0, len(text), call)) | |
| if not candidates: | |
| terminal_json = _terminal_json_tool_payload(text) | |
| if terminal_json is not None: | |
| start, end, payload = terminal_json | |
| call = _payload_call(payload, allowed_names) | |
| if call: | |
| candidates.append((start, end, call)) | |
| if not candidates: | |
| return [], text | |
| # Different parsers can recognize the same outer wrapper. Keep one result | |
| # per source span while preserving the order produced by the model. | |
| unique: list[dict[str, Any]] = [] | |
| seen_spans: set[tuple[int, int]] = set() | |
| for start, end, call in sorted(candidates, key=lambda item: (item[0], item[1])): | |
| span = (start, end) | |
| if span in seen_spans: | |
| continue | |
| seen_spans.add(span) | |
| unique.append(call) | |
| return unique, "" | |