| 1 | |
| 2 | from .dirty_json import DirtyJson |
| 3 | import regex, re |
| 4 | from helpers.modules import load_classes_from_file, load_classes_from_folder # keep here for backwards compatibility |
| 5 | from typing import Any |
| 6 | |
| 7 | def json_parse_dirty(json: str) -> dict[str, Any] | None: |
| 8 | if not json or not isinstance(json, str): |
| 9 | return None |
| 10 | |
| 11 | first_data: dict[str, Any] | None = None |
| 12 | for ext_json in extract_json_root_strings(json.strip()): |
| 13 | data = _parse_json_root_object(ext_json) |
| 14 | if data is None: |
| 15 | continue |
| 16 | if first_data is None: |
| 17 | first_data = data |
| 18 | if _is_tool_request(data): |
| 19 | return data |
| 20 | return first_data |
| 21 | |
| 22 | |
| 23 | def extract_tool_request(content: str) -> dict[str, Any] | None: |
| 24 | if not content or not isinstance(content, str): |
| 25 | return None |
| 26 | |
| 27 | content = content.strip() |
| 28 | if not (content.startswith("{") and content.endswith("}")): |
| 29 | return None |
| 30 | |
| 31 | root = extract_json_root_string(content) |
| 32 | if root != content: |
| 33 | return None |
| 34 | |
| 35 | request = _parse_json_root_object(root) |
| 36 | return request if request is not None and _is_tool_request(request) else None |
| 37 | |
| 38 | |
| 39 | def is_misformatted_tool_request(content: str) -> bool: |
| 40 | if not content or not isinstance(content, str): |
| 41 | return False |
| 42 | |
| 43 | content = content.strip() |
| 44 | roots = extract_json_root_strings(content) |
| 45 | if ( |
| 46 | len(roots) > 1 |
| 47 | and content.startswith("{") |
| 48 | and content.endswith("}") |
| 49 | and any(extract_tool_request(root) is not None for root in roots) |
| 50 | ): |
| 51 | return True |
| 52 | |
| 53 | for fenced_content in re.findall( |
| 54 | r"```(?:json)?\s*(.*?)```", content, flags=re.IGNORECASE | re.DOTALL |
| 55 | ): |
| 56 | request = json_parse_dirty(fenced_content) |
| 57 | if isinstance(request, dict) and _is_tool_request(request): |
| 58 | return True |
| 59 | |
| 60 | if ( |
| 61 | not content.endswith("}") |
| 62 | or re.match(r'^\{\s*"thoughts"\s*:', content) is None |
| 63 | ): |
| 64 | return False |
| 65 | |
| 66 | request = json_parse_dirty(content) |
| 67 | thoughts = request.get("thoughts") if isinstance(request, dict) else None |
| 68 | thoughts_text = ( |
| 69 | "\n".join(thought for thought in thoughts if isinstance(thought, str)) |
| 70 | if isinstance(thoughts, list) |
| 71 | else "" |
| 72 | ) |
| 73 | return ( |
| 74 | isinstance(thoughts, list) |
| 75 | and all( |
| 76 | f'{field}\":' in thoughts_text |
| 77 | for field in ("headline", "tool_name", "tool_args") |
| 78 | ) |
| 79 | ) |
| 80 | |
| 81 | |
| 82 | def normalize_tool_request(tool_request: Any) -> tuple[str, dict]: |
| 83 | if not isinstance(tool_request, dict): |
| 84 | raise ValueError("Tool request must be a dictionary") |
| 85 | if ( |
| 86 | not tool_request.get("tool_name") |
| 87 | and not tool_request.get("tool") |
| 88 | and "actions" in tool_request |
| 89 | ): |
| 90 | actions = tool_request["actions"] |
| 91 | # Text tool calls allow one request per turn; do not silently discard extras. |
| 92 | if ( |
| 93 | not isinstance(actions, list) |
| 94 | or len(actions) != 1 |
| 95 | or not isinstance(actions[0], dict) |
| 96 | ): |
| 97 | raise ValueError( |
| 98 | "Tool request actions wrapper must contain exactly one dictionary" |
| 99 | ) |
| 100 | tool_request = actions[0] |
| 101 | |
| 102 | tool_name = tool_request.get("tool_name") |
| 103 | if not tool_name or not isinstance(tool_name, str): |
| 104 | tool_name = tool_request.get("tool") |
| 105 | if ( |
| 106 | (not tool_name or not isinstance(tool_name, str)) |
| 107 | and tool_request.get("type") == "function" |
| 108 | ): |
| 109 | tool_name = tool_request.get("name") |
| 110 | if not tool_name or not isinstance(tool_name, str): |
| 111 | raise ValueError("Tool request must have a tool_name (type string) field") |
| 112 | tool_args = tool_request.get("tool_args") |
| 113 | if not isinstance(tool_args, dict): |
| 114 | tool_args = tool_request.get("args") |
| 115 | if not isinstance(tool_args, dict) and tool_request.get("type") == "function": |
| 116 | tool_args = tool_request.get("parameters") |
| 117 | if not isinstance(tool_args, dict): |
| 118 | raise ValueError("Tool request must have a tool_args (type dictionary) field") |
| 119 | tool_args = dict(tool_args) |
| 120 | if ":" in tool_name: |
| 121 | tool_name, action = tool_name.split(":", 1) |
| 122 | if not tool_name or not action: |
| 123 | raise ValueError("tool_name method suffix must include tool and action") |
| 124 | tool_args.setdefault("action", action) |
| 125 | method = tool_args.get("method") |
| 126 | if "action" not in tool_args and isinstance(method, str) and method: |
| 127 | tool_args["action"] = method |
| 128 | return tool_name, tool_args |
| 129 | |
| 130 | |
| 131 | def extract_json_root_string(content: str) -> str | None: |
| 132 | first_root: str | None = None |
| 133 | for root in extract_json_root_strings(content): |
| 134 | if first_root is None: |
| 135 | first_root = root |
| 136 | data = _parse_json_root_object(root) |
| 137 | if data is not None and _is_tool_request(data): |
| 138 | return root |
| 139 | return first_root |
| 140 | |
| 141 | |
| 142 | def extract_json_root_strings(content: str) -> list[str]: |
| 143 | if not content or not isinstance(content, str): |
| 144 | return [] |
| 145 | |
| 146 | if content.lstrip().startswith("["): |
| 147 | return [] |
| 148 | |
| 149 | roots: list[str] = [] |
| 150 | for start in _json_root_object_starts(content): |
| 151 | parser = DirtyJson() |
| 152 | try: |
| 153 | parser.parse(content[start:]) |
| 154 | except Exception: |
| 155 | continue |
| 156 | |
| 157 | if not parser.completed: |
| 158 | continue |
| 159 | |
| 160 | roots.append(content[start : start + parser.index]) |
| 161 | return roots |
| 162 | |
| 163 | |
| 164 | def _json_root_object_starts(content: str) -> list[int]: |
| 165 | starts: list[int] = [] |
| 166 | depth = 0 |
| 167 | quote: str | None = None |
| 168 | escaped = False |
| 169 | |
| 170 | for index, char in enumerate(content): |
| 171 | if quote: |
| 172 | if escaped: |
| 173 | escaped = False |
| 174 | elif char == "\\": |
| 175 | escaped = True |
| 176 | elif char == quote: |
| 177 | quote = None |
| 178 | continue |
| 179 | |
| 180 | if depth and char in ['"', "'", "`"]: |
| 181 | quote = char |
| 182 | elif char == "{": |
| 183 | if depth == 0: |
| 184 | starts.append(index) |
| 185 | depth += 1 |
| 186 | elif depth and char == "[": |
| 187 | depth += 1 |
| 188 | elif depth and char in ["}", "]"]: |
| 189 | depth -= 1 |
| 190 | |
| 191 | return starts |
| 192 | |
| 193 | |
| 194 | def _parse_json_root_object(root: str) -> dict[str, Any] | None: |
| 195 | try: |
| 196 | data = DirtyJson.parse_string(root) |
| 197 | except Exception: |
| 198 | return None |
| 199 | return data if isinstance(data, dict) else None |
| 200 | |
| 201 | |
| 202 | def _is_tool_request(data: dict[str, Any]) -> bool: |
| 203 | try: |
| 204 | normalize_tool_request(data) |
| 205 | except ValueError: |
| 206 | return False |
| 207 | return True |
| 208 | |
| 209 | |
| 210 | def extract_json_object_string(content): |
| 211 | start = content.find("{") |
| 212 | if start == -1: |
| 213 | return "" |
| 214 | |
| 215 | # Find the first '{' |
| 216 | end = content.rfind("}") |
| 217 | if end == -1: |
| 218 | # If there's no closing '}', return from start to the end |
| 219 | return content[start:] |
| 220 | else: |
| 221 | # If there's a closing '}', return the substring from start to end |
| 222 | return content[start : end + 1] |
| 223 | |
| 224 | |
| 225 | def extract_json_string(content): |
| 226 | # Regular expression pattern to match a JSON object |
| 227 | pattern = r'\{(?:[^{}]|(?R))*\}|\[(?:[^\[\]]|(?R))*\]|"(?:\\.|[^"\\])*"|true|false|null|-?\d+(?:\.\d+)?(?:[eE][+-]?\d+)?' |
| 228 | |
| 229 | # Search for the pattern in the content |
| 230 | match = regex.search(pattern, content) |
| 231 | |
| 232 | if match: |
| 233 | # Return the matched JSON string |
| 234 | return match.group(0) |
| 235 | else: |
| 236 | return "" |
| 237 | |
| 238 | |
| 239 | def fix_json_string(json_string): |
| 240 | # Function to replace unescaped line breaks within JSON string values |
| 241 | def replace_unescaped_newlines(match): |
| 242 | return match.group(0).replace("\n", "\\n") |
| 243 | |
| 244 | # Use regex to find string values and apply the replacement function |
| 245 | fixed_string = re.sub( |
| 246 | r'(?<=: ")(.*?)(?=")', replace_unescaped_newlines, json_string, flags=re.DOTALL |
| 247 | ) |
| 248 | return fixed_string |