Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 46 additions & 2 deletions helpers/extract_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,22 +3,66 @@
import regex, re
from helpers.modules import load_classes_from_file, load_classes_from_folder # keep here for backwards compatibility
from typing import Any
import logging

LOG = logging.getLogger(__name__)

def json_parse_dirty(json: str) -> dict[str, Any] | None:
if not json or not isinstance(json, str):
return None

# Attempt 1: Standard parser
ext_json = extract_json_object_string(json.strip())
if ext_json:
try:
data = DirtyJson.parse_string(ext_json)
if isinstance(data, dict):
return data
except Exception:
# If parsing fails, return None instead of crashing
return None
pass

# Attempt 2: Regex fallback - extract tool_name and tool_args from partial output
try:
result = _regex_fallback(json)
if result:
LOG.debug("json_parse_dirty: recovered via regex fallback")
return result
except Exception as e:
LOG.debug(f"json_parse_dirty: regex fallback failed: {e}")

return None


def _regex_fallback(text: str) -> dict[str, Any] | None:
"""Try to extract tool_name and tool_args from malformed JSON using regex."""
# Find tool_name value
tn_match = re.search(
r'"tool_name"\s*:\s*"([^"]+)"', text
)
if not tn_match:
return None
tool_name = tn_match.group(1)

# Find tool_args object
ta_match = re.search(
r'"tool_args"\s*:\s*(\{[^}]*\})', text, re.DOTALL
)
if not ta_match:
# Return with empty args if we at least have a tool_name
return {"tool_name": tool_name, "tool_args": {}}

args_str = ta_match.group(1)
try:
args = DirtyJson.parse_string(args_str)
if isinstance(args, dict):
return {"tool_name": tool_name, "tool_args": args}
except Exception:
pass

# Last resort: return tool_name with raw args string
return {"tool_name": tool_name, "tool_args": {"_raw": args_str}}


def extract_json_root_string(content: str) -> str | None:
if not content or not isinstance(content, str):
return None
Expand Down