File size: 3,805 Bytes
150aa2a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | import ast
import json
import re
from typing import Any
tool_call_start = "<|tool_call:start|>"
tool_call_end = "<|tool_call:end|>"
_ARG_RE = re.compile(
r"<\|tool_arg:start\|>(.*?)<\|tool_arg:value\|>(.*?)<\|tool_arg:end\|>",
re.DOTALL,
)
_STRING_TYPES = {"string", "str", "text", "varchar", "char", "enum"}
_BOOL_TYPES = {"boolean", "bool", "binary"}
_INT_PREFIXES = ("int", "uint", "long", "short", "unsigned")
_FLOAT_PREFIXES = ("num", "float", "double", "decimal")
_JSON_TYPES = {"object", "array", "arr", "dict", "list"}
def _tool_properties(name: str, tools: Any | None) -> dict[str, Any]:
if not tools:
return {}
for tool in tools:
function = tool.get("function") if isinstance(tool, dict) else None
if function and function.get("name") == name:
params = function.get("parameters") or {}
return params.get("properties") or {}
return {}
def _schema_type(properties: dict[str, Any], key: str) -> str:
schema = properties.get(key) or {}
typ = schema.get("type", "string")
if isinstance(typ, list):
typ = next((item for item in typ if item != "null"), typ[0] if typ else "string")
return str(typ).lower()
def _coerce(value: str, key: str, properties: dict[str, Any]) -> Any:
value = value.strip()
typ = _schema_type(properties, key)
if value.lower() == "null":
return None
if typ in _STRING_TYPES:
return value
if typ in _BOOL_TYPES:
normalized = value.lower()
if normalized in {"1", "true", "yes"}:
return True
if normalized in {"0", "false", "no"}:
return False
return value
if typ.startswith(_INT_PREFIXES):
try:
return int(value)
except (TypeError, ValueError):
return value
if typ.startswith(_FLOAT_PREFIXES):
try:
number = float(value)
return int(number) if number.is_integer() else number
except (TypeError, ValueError):
return value
if typ in _JSON_TYPES or typ.startswith(("dict", "list")):
try:
return json.loads(value)
except (TypeError, ValueError, json.JSONDecodeError):
try:
return ast.literal_eval(value)
except (SyntaxError, ValueError):
return value
try:
return json.loads(value)
except (TypeError, json.JSONDecodeError):
return value
def parse_tool_call(text: str, tools: Any | None = None) -> dict[str, Any]:
body = text.strip()
if body.startswith(tool_call_start):
body = body[len(tool_call_start) :].strip()
if body.endswith(tool_call_end):
body = body[: -len(tool_call_end)].strip()
first_arg = body.find("<|tool_arg:start|>")
if first_arg == -1:
try:
payload = json.loads(body)
except json.JSONDecodeError as exc:
raise ValueError(f"No Solar tool arguments found in: {text!r}") from exc
name = payload.get("name") or payload.get("tool") or payload.get("function")
args = payload.get("arguments") or payload.get("args") or {}
if not name:
raise ValueError(f"No Solar tool name found in: {text!r}")
if isinstance(args, str):
args = json.loads(args)
return {"name": name, "arguments": args}
name = body[:first_arg].strip().splitlines()[0].strip()
if not name:
raise ValueError(f"No Solar tool name found in: {text!r}")
properties = _tool_properties(name, tools)
arguments: dict[str, Any] = {}
for key, value in _ARG_RE.findall(body[first_arg:]):
key = key.strip()
if key:
arguments[key] = _coerce(value, key, properties)
return {"name": name, "arguments": arguments}
|