mirror of
https://github.com/agent0ai/agent-zero.git
synced 2026-05-23 12:44:31 +00:00
100 lines
3.1 KiB
Python
100 lines
3.1 KiB
Python
|
|
from .dirty_json import DirtyJson
|
|
import regex, re
|
|
from helpers.modules import load_classes_from_file, load_classes_from_folder # keep here for backwards compatibility
|
|
from typing import Any
|
|
|
|
def json_parse_dirty(json: str) -> dict[str, Any] | None:
|
|
if not json or not isinstance(json, str):
|
|
return None
|
|
|
|
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
|
|
return None
|
|
|
|
|
|
def normalize_tool_request(tool_request: Any) -> tuple[str, dict]:
|
|
if not isinstance(tool_request, dict):
|
|
raise ValueError("Tool request must be a dictionary")
|
|
tool_name = tool_request.get("tool_name")
|
|
if not tool_name or not isinstance(tool_name, str):
|
|
tool_name = tool_request.get("tool")
|
|
if not tool_name or not isinstance(tool_name, str):
|
|
raise ValueError("Tool request must have a tool_name (type string) field")
|
|
tool_args = tool_request.get("tool_args")
|
|
if not isinstance(tool_args, dict):
|
|
tool_args = tool_request.get("args")
|
|
if not isinstance(tool_args, dict):
|
|
raise ValueError("Tool request must have a tool_args (type dictionary) field")
|
|
return tool_name, tool_args
|
|
|
|
|
|
def extract_json_root_string(content: str) -> str | None:
|
|
if not content or not isinstance(content, str):
|
|
return None
|
|
|
|
start = content.find("{")
|
|
if start == -1:
|
|
return None
|
|
first_array = content.find("[")
|
|
if first_array != -1 and first_array < start:
|
|
return None
|
|
|
|
parser = DirtyJson()
|
|
try:
|
|
parser.parse(content[start:])
|
|
except Exception:
|
|
return None
|
|
|
|
if not parser.completed:
|
|
return None
|
|
|
|
return content[start : start + parser.index]
|
|
|
|
|
|
def extract_json_object_string(content):
|
|
start = content.find("{")
|
|
if start == -1:
|
|
return ""
|
|
|
|
# Find the first '{'
|
|
end = content.rfind("}")
|
|
if end == -1:
|
|
# If there's no closing '}', return from start to the end
|
|
return content[start:]
|
|
else:
|
|
# If there's a closing '}', return the substring from start to end
|
|
return content[start : end + 1]
|
|
|
|
|
|
def extract_json_string(content):
|
|
# Regular expression pattern to match a JSON object
|
|
pattern = r'\{(?:[^{}]|(?R))*\}|\[(?:[^\[\]]|(?R))*\]|"(?:\\.|[^"\\])*"|true|false|null|-?\d+(?:\.\d+)?(?:[eE][+-]?\d+)?'
|
|
|
|
# Search for the pattern in the content
|
|
match = regex.search(pattern, content)
|
|
|
|
if match:
|
|
# Return the matched JSON string
|
|
return match.group(0)
|
|
else:
|
|
return ""
|
|
|
|
|
|
def fix_json_string(json_string):
|
|
# Function to replace unescaped line breaks within JSON string values
|
|
def replace_unescaped_newlines(match):
|
|
return match.group(0).replace("\n", "\\n")
|
|
|
|
# Use regex to find string values and apply the replacement function
|
|
fixed_string = re.sub(
|
|
r'(?<=: ")(.*?)(?=")', replace_unescaped_newlines, json_string, flags=re.DOTALL
|
|
)
|
|
return fixed_string
|