File size: 11,649 Bytes
816198f | 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 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 | import ast
import json
import os
from typing import List, Dict, Any
import re
from utils.common import extract_candidate_object, _prefix_files
def extract_nlp_tool_calls(text: str, file_base_dirs: List | None = None, file_prefix = None, prefix_mode = "inference") -> List[Dict[str, Any]]:
TOOL_CALL_BLOCK_RE = re.compile(
r"<tool_call>\s*(.*?)\s*</tool_call>", # 匹配整个 tool_call 块，包括其中的 code 标签，而不是只匹配 JSON 对象
re.DOTALL | re.IGNORECASE,
)
calls = []
for m in TOOL_CALL_BLOCK_RE.finditer(text or ""):
block = m.group(1).strip()
print(f"<tool_call> block: {block}")
try:
# 代码
if "pythoninterpreter" in block.lower():
try:
# 找到 "pythoninterpreter" 这一行，从这一行之后作为 code_block
lines = block.splitlines()
# 找到含有 'pythoninterpreter' 的行号
start = None
for i, line in enumerate(lines):
if "pythoninterpreter" in line.lower():
start = i
break
if start is not None:
code_block = "\n".join(lines[start+1:])
else:
code_block = ""
except Exception as e:
code_block = ""
# 删除以 "```python ," 或 "```python," 或 "```python" 或 "```" 开头的行
code_lines = code_block.splitlines()
clean_lines = [""]
for line in code_lines:
stripped = line.strip()
if not (
stripped.startswith('```python ,') or
stripped.startswith('```python,') or
stripped.startswith('```python') or
stripped.startswith('```') or
stripped.startswith('<code>') or
stripped.startswith('</code>')
):
clean_lines.append(line)
code_raw = "\n".join(clean_lines).strip()
calls.append({"name": "execute_code", "arguments": {"code": code_raw}})
elif "bash" in block.lower():
try:
lines = block.splitlines()
start = None
for i, line in enumerate(lines):
if "bash" in line.lower():
start = i
break
if start is not None:
code_block = "\n".join(lines[start+1:])
else:
code_block = ""
except Exception as e:
code_block = ""
code_lines = code_block.splitlines()
clean_lines = [""]
# bash 工具
for line in code_lines:
stripped = line.strip()
if not (
stripped.startswith('```bash ,') or
stripped.startswith('```bash,') or
stripped.startswith('```bash') or
stripped.startswith('```') or
stripped.startswith('<bash>') or
stripped.startswith('</bash>')
):
clean_lines.append(line)
code_raw = "\n".join(clean_lines).strip()
calls.append({"name": "bash", "arguments": {"command": code_raw}})
else:
obj = extract_candidate_object(block)
tool_name = obj.get("name", "")
tool_arguments = obj.get("arguments", {})
# 模型有时会将 arguments 序列化为字符串，兼容处理
if isinstance(tool_arguments, str):
try:
tool_arguments = json.loads(tool_arguments)
except Exception:
try:
import json5
tool_arguments = json5.loads(tool_arguments)
except Exception:
tool_arguments = {}
# 列表形式的搜索
if tool_name == "search":
search_query = tool_arguments.get('query', None)
if search_query is None:
raise ValueError(f"query is not found in the tool arguments: {tool_arguments}")
if isinstance(search_query, list) or isinstance(search_query, str):
calls.append({"name": "wide_search", "arguments": {"query": search_query}})
else:
raise ValueError(f"Unknown query type: {type(search_query)}")
# 列表形式的谷歌搜索
elif tool_name == "google_scholar":
search_query = tool_arguments.get('query', None)
if search_query is None:
raise ValueError(f"query is not found in the tool arguments: {tool_arguments}")
if isinstance(search_query, list) or isinstance(search_query, str):
calls.append({"name": "scholar_search", "arguments": {"query": search_query}})
else:
raise ValueError(f"Unknown query type: {type(search_query)}")
# visit 列表形式
elif tool_name == "visit":
visit_goal = tool_arguments.get('goal', None)
visit_url = tool_arguments.get('url', None)
if visit_goal is None:
raise ValueError(f"goal is not found in the tool arguments: {tool_arguments}")
if visit_url is None:
raise ValueError(f"url is not found in the tool arguments: {tool_arguments}")
if isinstance(visit_url, list) or isinstance(visit_url, str):
calls.append({"name": "wide_visit", "arguments": {"url": visit_url, "goal": visit_goal}})
else:
raise ValueError(f"Unknown url type: {type(visit_url)}")
# 文件解析 列表形式
elif tool_name == "parse_file":
files = tool_arguments.get('files', None)
if files is None:
raise ValueError(f"files is not found in the tool arguments: {tool_arguments}")
if isinstance(files, list) or isinstance(files, str):
calls.append(
{
"name": "file_wide_parse",
"arguments": {"files": _prefix_files(file_base_dirs, files, file_prefix, prefix_mode)},
}
)
else:
raise ValueError(f"Unknown url type: {type(files)}")
# 图像搜索
elif tool_name == "image_search":
search_query = tool_arguments.get('query', None)
if search_query is None:
raise ValueError(f"query is not found in the tool arguments: {tool_arguments}")
if isinstance(search_query, list) or isinstance(search_query, str):
calls.append({"name": "image_search", "arguments": {"query": search_query}})
else:
raise ValueError(f"Unknown query type: {type(search_query)}")
# 图像问答
elif tool_name == "ask_question_about_image":
image_path = tool_arguments.get("image_path", None)
question = tool_arguments.get("question", None)
if image_path is None:
raise ValueError(f"image_path is not found in the tool arguments: {tool_arguments}")
if question is None:
raise ValueError(f"question is not found in the tool arguments: {tool_arguments}")
if (isinstance(image_path, str) or isinstance(image_path, list)) and isinstance(question, str):
calls.append(
{
"name": "ask_question_about_image",
"arguments": {"image_path": _prefix_files(file_base_dirs, image_path, file_prefix, prefix_mode), "question": question},
}
)
else:
raise ValueError(
f"Unknown image_path/question type: "
f"image_path({type(image_path)}), question({type(question)})"
)
# 视频问答
elif tool_name == "ask_question_about_video":
video_path = tool_arguments.get("video_path", None)
question = tool_arguments.get("question", None)
if video_path is None:
raise ValueError(f"video_path is not found in the tool arguments: {tool_arguments}")
if question is None:
raise ValueError(f"question is not found in the tool arguments: {tool_arguments}")
if (isinstance(video_path, str) or isinstance(video_path, list)) and isinstance(question, str):
calls.append(
{
"name": "ask_question_about_video",
"arguments": {"video_path": _prefix_files(file_base_dirs, video_path, file_prefix, prefix_mode), "question": question},
}
)
else:
raise ValueError(
f"Unknown video_path/question type: "
f"video_path({type(video_path)}), question({type(question)})"
)
elif tool_name in ("execute_code", "python_interpreter"):
code = tool_arguments.get('code', None)
if code is None:
raise ValueError(f"code is not found in the tool arguments: {tool_arguments}")
code_lines = code.splitlines()
clean_lines = [""]
for line in code_lines:
stripped = line.strip()
if not (
stripped.startswith('```python ,') or
stripped.startswith('```python,') or
stripped.startswith('```python') or
stripped.startswith('```') or
stripped.startswith('<code>') or
stripped.startswith('</code>')
):
clean_lines.append(line)
code_raw = "\n".join(clean_lines).strip()
calls.append({"name": "execute_code", "arguments": {"code": code_raw}})
else:
raise ValueError(f"Unknown tool name: {tool_name}")
except Exception as e:
calls.append({"name": "parse_error_tool_call", "arguments": {"parse_error": str(e), "raw": block}})
print(f"extract_tool_calls calls: {calls}")
return calls
|