Files
projectAIpopular/router_system/tools.py
T
tzt 43e2bceae7 feat(v3): T26 两级智能体(大模型规划/审查 + 本地小模型执行,D7)
- run_dual 编排:规划者两阶段 JSON(plan/review,失败回喂重试一次再降级),
  redo 时裁决意见回喂执行者,交接上限 agent.max_handoffs(默认 2)
- 交接文档 agent_runs/{id}/handoff.json(智能体版交流文本:instructions/acceptance/exchanges)
- /agent 新增 executor_pool_id;规划者==执行者条目拒绝;整体 token_cap 覆盖两级调用
- ToolLoop 增 emit_final 开关(内层循环不发终态,防前端 SSE 提前收口)
- 前端:执行者选择器 + phase/message 事件渲染(阶段徽标 + 双色消息卡)
- 测试 +3(done/redo/执行者故障),全量 277 passed
- fix(tests): test_config_get_put_reset 增加设置备份/恢复隔离,防止清掉用户真实配置
2026-09-01 11:45:33 +08:00

526 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""工具调用内核 —— 让 LLM 以 OpenAI function-calling 协议操作工作区文件。
组成(对齐《实现方案_v4_模型池与工具智能体.md》D4):
- TOOLS_SPEClist_dir / read_file / write_file 三个工具的 OpenAI tools 声明
- WorkspaceTools:被"关押"在根目录内的文件工具(路径越界一律拒绝,Windows pathlib
- parse_tool_calls:解析 OpenAI 响应里的 tool_callsarguments 容错为 {}
- ToolLoop:通用智能体循环。chat_fn 注入(网关传 OpenAI 兼容客户端,测试传假实现),
本模块只负责循环编排:调用 -> 执行工具 -> 回喂结果 -> 直到模型给出最终答复。
工程约束:
- 纯标准库(router_system 零第三方依赖不变)
- 工具结果回喂前截断(防止上下文爆炸),轮数与 token 双上限(金额护栏)
- 事件回调 on_event 逐条产出过程事件(供 SSE 透出"智能体在做什么"
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any, Awaitable, Callable, Dict, List, Optional
# 回喂给模型的工具结果/读取内容上限(字符)
MAX_READ_CHARS = 8000
MAX_LIST_ENTRIES = 200
MAX_RESULT_CHARS = 8000
MAX_WRITE_CHARS = 200_000
# search_files 上限
SEARCH_MAX_MATCHES = 30
SEARCH_MAX_FILES = 400
SEARCH_MAX_FILE_BYTES = 512 * 1024
SEARCH_SKIP_DIRS = {".git", "node_modules", "__pycache__", ".venv", "venv", "dist", "build"}
# run_command 上限
SHELL_OUTPUT_CHARS = 4000
# 默认循环上限
DEFAULT_MAX_ROUNDS = 8
# OpenAI tools 声明(chat/completions 请求的 tools 参数)
TOOLS_SPEC: List[Dict[str, Any]] = [
{
"type": "function",
"function": {
"name": "list_dir",
"description": "列出工作区内目录的内容(文件与子目录,含大小)。",
"parameters": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "工作区内的相对路径,默认根目录"}
},
},
},
},
{
"type": "function",
"function": {
"name": "read_file",
"description": "读取工作区内一个文本文件的内容(过长自动截断)。",
"parameters": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "工作区内的相对路径"}
},
"required": ["path"],
},
},
},
{
"type": "function",
"function": {
"name": "write_file",
"description": "把文本内容写入(或创建/覆盖)工作区内的一个文件。",
"parameters": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "工作区内的相对路径"},
"content": {"type": "string", "description": "要写入的全文"},
},
"required": ["path", "content"],
},
},
},
{
"type": "function",
"function": {
"name": "edit_file",
"description": "对工作区内已有文件做精确替换编辑:old_string 必须在文件中恰好出现一次,"
"被替换为 new_string。适合小改动;大改用 write_file 整体重写。",
"parameters": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "工作区内的相对路径"},
"old_string": {"type": "string", "description": "要替换的原文(须唯一匹配)"},
"new_string": {"type": "string", "description": "替换后的新文"},
},
"required": ["path", "old_string", "new_string"],
},
},
},
{
"type": "function",
"function": {
"name": "search_files",
"description": "在工作区(或其子目录)内按关键词跨文件搜索文本内容,"
"返回匹配的文件/行号/行内容(自动跳过 .git、node_modules 等目录与二进制大文件)。",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索的关键词(大小写敏感)"},
"path": {"type": "string", "description": "限定的子目录,默认整个工作区"},
},
"required": ["query"],
},
},
},
{
"type": "function",
"function": {
"name": "run_command",
"description": "在工作区根目录执行一条 shell 命令并返回退出码与输出(如运行测试、查看版本)。"
"仅当系统开启 allow_shell 时可用。",
"parameters": {
"type": "object",
"properties": {
"command": {"type": "string", "description": "要执行的命令行"},
},
"required": ["command"],
},
},
},
]
TOOL_NAMES = {t["function"]["name"] for t in TOOLS_SPEC}
def browse_directories(path: str = "") -> Dict[str, Any]:
"""目录选择器的本地文件系统浏览(只列目录,不读文件内容)。
path 为空时列出 Windows 盘符(POSIX 列根目录)。返回
{"ok", "path", "parent", "dirs": [名称]};用于智能体"选择工作区"。
"""
import os
if not path or not path.strip():
if os.name == "nt":
drives = [f"{c}:\\" for c in "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
if os.path.exists(f"{c}:\\")]
return {"ok": True, "path": "", "parent": "", "dirs": drives}
return {"ok": True, "path": "/", "parent": "",
"dirs": sorted(os.listdir("/"))}
p = Path(path).resolve()
if not p.exists():
return {"ok": False, "error": f"路径不存在: {path}"}
if not p.is_dir():
return {"ok": False, "error": f"不是目录: {path}"}
dirs = []
for child in sorted(p.iterdir(), key=lambda c: c.name.lower()):
try:
if child.is_dir():
dirs.append(child.name)
except OSError:
continue # 无权限/符号链接坏点,跳过
parent = str(p.parent) if p.parent != p else ""
return {"ok": True, "path": str(p), "parent": parent, "dirs": dirs}
class ToolError(Exception):
"""工具执行失败(路径越界/不存在/参数非法)。"""
class WorkspaceTools:
"""被限制在根目录内的文件工具(智能体的"手")。
安全:所有路径先 join 再 resolve,解析结果必须仍位于根目录内
(根目录自身允许),否则抛 ToolError——防 ../ 越界与绝对路径逃逸。
"""
def __init__(self, root: str | Path,
allow_shell: bool = False, shell_timeout_s: int = 20):
self.root = Path(root).resolve()
self.root.mkdir(parents=True, exist_ok=True)
self.allow_shell = bool(allow_shell)
self.shell_timeout_s = max(1, int(shell_timeout_s))
# ---------- 路径关押 ----------
def resolve(self, rel_path: str) -> Path:
rel = (rel_path or "").strip().replace("\\", "/").lstrip("/")
p = (self.root / rel).resolve()
if p != self.root and self.root not in p.parents:
raise ToolError(f"路径越界(不允许访问工作区之外): {rel_path}")
return p
# ---------- 三个工具 ----------
def list_dir(self, rel_path: str = "") -> Dict[str, Any]:
d = self.resolve(rel_path)
if not d.exists():
return {"ok": False, "error": f"目录不存在: {rel_path}"}
if not d.is_dir():
return {"ok": False, "error": f"不是目录: {rel_path}"}
entries = []
for child in sorted(d.iterdir(), key=lambda c: (c.is_file(), c.name.lower())):
if child.is_dir():
entries.append({"name": child.name + "/", "type": "dir"})
else:
entries.append({
"name": child.name, "type": "file",
"size": child.stat().st_size,
})
if len(entries) >= MAX_LIST_ENTRIES:
entries.append({"name": f"…(超过 {MAX_LIST_ENTRIES} 项已截断)", "type": "notice"})
break
return {"ok": True, "path": rel_path or ".", "entries": entries}
def read_file(self, rel_path: str) -> Dict[str, Any]:
p = self.resolve(rel_path)
if not p.exists():
return {"ok": False, "error": f"文件不存在: {rel_path}"}
if not p.is_file():
return {"ok": False, "error": f"不是文件: {rel_path}"}
try:
text = p.read_text(encoding="utf-8")
except UnicodeDecodeError:
return {"ok": False, "error": f"非文本文件(UTF-8 解码失败): {rel_path}"}
truncated = len(text) > MAX_READ_CHARS
return {
"ok": True,
"path": rel_path,
"content": text[:MAX_READ_CHARS],
"truncated": truncated,
"total_chars": len(text),
}
def write_file(self, rel_path: str, content: str) -> Dict[str, Any]:
if len(content) > MAX_WRITE_CHARS:
return {"ok": False, "error": f"内容过长(>{MAX_WRITE_CHARS} 字符),拒绝写入"}
p = self.resolve(rel_path)
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text(content, encoding="utf-8")
return {"ok": True, "path": rel_path, "bytes_written": len(content.encode("utf-8"))}
def edit_file(self, rel_path: str, old_string: str, new_string: str) -> Dict[str, Any]:
"""精确替换编辑:old_string 必须在文件中恰好出现一次(harness 式安全编辑)。"""
if not old_string:
return {"ok": False, "error": "old_string 不能为空"}
if len(old_string) > MAX_READ_CHARS:
return {"ok": False, "error": "old_string 过长(先 read_file 分段定位)"}
p = self.resolve(rel_path)
if not p.exists() or not p.is_file():
return {"ok": False, "error": f"文件不存在: {rel_path}"}
try:
text = p.read_text(encoding="utf-8")
except UnicodeDecodeError:
return {"ok": False, "error": f"非文本文件: {rel_path}"}
count = text.count(old_string)
if count == 0:
return {"ok": False, "error": "old_string 未在文件中找到(先 read_file 核对原文)"}
if count > 1:
return {"ok": False,
"error": f"old_string 出现 {count} 次(要求唯一);请扩大上下文使其唯一"}
new_text = text.replace(old_string, new_string, 1)
p.write_text(new_text, encoding="utf-8")
return {
"ok": True, "path": rel_path,
"replaced": 1,
"changed_chars": len(new_text) - len(text),
}
def search_files(self, query: str, rel_path: str = "") -> Dict[str, Any]:
"""跨文件文本搜索(跳过依赖/构建目录与二进制大文件,限量返回)。"""
if not query:
return {"ok": False, "error": "query 不能为空"}
base = self.resolve(rel_path or "")
if not base.exists() or not base.is_dir():
return {"ok": False, "error": f"目录不存在: {rel_path}"}
matches: List[Dict[str, Any]] = []
scanned = 0
truncated = False
for p in sorted(base.rglob("*")):
if len(matches) >= SEARCH_MAX_MATCHES:
truncated = True
break
if not p.is_file():
continue
rel_parts = p.relative_to(base).parts
if any(part in SEARCH_SKIP_DIRS for part in rel_parts):
continue
try:
if p.stat().st_size > SEARCH_MAX_FILE_BYTES:
continue
text = p.read_text(encoding="utf-8")
except (OSError, UnicodeDecodeError):
continue # 二进制/不可读,跳过
scanned += 1
if scanned > SEARCH_MAX_FILES:
truncated = True
break
for lineno, line in enumerate(text.splitlines(), 1):
if query in line:
rel = Path(*p.relative_to(self.root).parts).as_posix()
matches.append({
"file": rel, "line": lineno,
"text": line.strip()[:300],
})
if len(matches) >= SEARCH_MAX_MATCHES:
truncated = True
break
return {"ok": True, "query": query, "matches": matches,
"scanned_files": scanned, "truncated": truncated}
def run_command(self, command: str) -> Dict[str, Any]:
"""在工作区根目录执行 shell 命令(默认关闭,allow_shell 开启后可用)。"""
if not self.allow_shell:
return {"ok": False,
"error": "run_command 未启用(系统设置 allow_shell 为关)。"
"请让用户在智能体页打开「允许执行命令」后重试。"}
command = (command or "").strip()
if not command:
return {"ok": False, "error": "command 不能为空"}
import subprocess
creationflags = 0x08000000 if __import__("os").name == "nt" else 0 # CREATE_NO_WINDOW
try:
proc = subprocess.run(
command, shell=True, cwd=str(self.root), capture_output=True,
timeout=self.shell_timeout_s, creationflags=creationflags,
)
out = (proc.stdout or b"").decode("utf-8", errors="replace")
err = (proc.stderr or b"").decode("utf-8", errors="replace")
combined = (out + ("\n[stderr]\n" + err if err.strip() else "")).strip()
if len(combined) > SHELL_OUTPUT_CHARS:
combined = combined[:SHELL_OUTPUT_CHARS] + "…(输出截断)"
return {"ok": True, "exit_code": proc.returncode,
"output": combined or "(无输出)", "command": command}
except subprocess.TimeoutExpired:
return {"ok": False, "error": f"命令超时(>{self.shell_timeout_s}s),已终止: {command}"}
except OSError as e:
return {"ok": False, "error": f"命令执行失败: {type(e).__name__}: {e}"}
# ---------- 统一执行入口 ----------
def execute(self, name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""按名字执行工具;任何异常折叠为 {"ok": False, "error": ...}。"""
try:
if name == "list_dir":
return self.list_dir(str(arguments.get("path", "")))
if name == "read_file":
return self.read_file(str(arguments.get("path", "")))
if name == "write_file":
return self.write_file(
str(arguments.get("path", "")), str(arguments.get("content", "")))
if name == "edit_file":
return self.edit_file(
str(arguments.get("path", "")),
str(arguments.get("old_string", "")),
str(arguments.get("new_string", "")))
if name == "search_files":
return self.search_files(
str(arguments.get("query", "")), str(arguments.get("path", "")))
if name == "run_command":
return self.run_command(str(arguments.get("command", "")))
return {"ok": False, "error": f"未知工具: {name}"}
except ToolError as e:
return {"ok": False, "error": str(e)}
except OSError as e:
return {"ok": False, "error": f"文件系统错误: {type(e).__name__}: {e}"}
def parse_tool_calls(message: Dict[str, Any]) -> List[Dict[str, Any]]:
"""从 OpenAI 响应的 message 解析 tool_calls。
返回 [{"id","name","arguments"(dict)}]arguments 非法 JSON 时容错为 {}
"""
out: List[Dict[str, Any]] = []
for tc in message.get("tool_calls") or []:
fn = tc.get("function") or {}
raw = fn.get("arguments")
if isinstance(raw, dict):
args = raw
elif isinstance(raw, str) and raw.strip():
try:
parsed = json.loads(raw)
args = parsed if isinstance(parsed, dict) else {}
except json.JSONDecodeError:
args = {}
else:
args = {}
out.append({
"id": tc.get("id") or f"call_{len(out)}",
"name": fn.get("name") or "",
"arguments": args,
})
return out
class ToolLoop:
"""通用智能体工具循环(zcode 式:思考 -> 调工具 -> 看结果 -> 再思考)。
chat_fn(messages, tools_spec) -> {"content": str|None,
"tool_calls": [ {id,name,arguments}, ... ],
"usage": {"prompt_tokens", "completion_tokens"}}
由网关注入真实 OpenAI 兼容客户端;测试注入脚本化假实现。
on_event(ev) 为可选同步回调,逐条收到过程事件:
{"type":"round","round":n}
{"type":"tool_call","round":n,"name":...,"arguments":...}
{"type":"tool_result","round":n,"name":...,"ok":...,"preview":...}
{"type":"usage","prompt_tokens":...,"completion_tokens":...}
{"type":"final","round":n,"reason":"answer"|"max_rounds"|"token_cap"|"error"}
"""
def __init__(
self,
tools: WorkspaceTools,
chat_fn: Callable[[List[Dict[str, Any]], List[Dict[str, Any]]], Awaitable[Dict[str, Any]]],
max_rounds: int = DEFAULT_MAX_ROUNDS,
token_cap: int = 0,
on_event: Optional[Callable[[Dict[str, Any]], None]] = None,
result_preview_chars: int = MAX_RESULT_CHARS,
emit_final: bool = True,
):
self.tools = tools
self.chat_fn = chat_fn
self.max_rounds = max(1, int(max_rounds))
self.token_cap = int(token_cap) # 0 = 不限
self.on_event = on_event
self.result_preview_chars = result_preview_chars
self.emit_final = emit_final # 两级模式内层循环置 False,由外层统一收尾
def _emit(self, ev: Dict[str, Any]) -> None:
if ev.get("type") == "final" and not self.emit_final:
return # 内层循环不发终态事件(外层编排负责)
if self.on_event is not None:
try:
self.on_event(ev)
except Exception:
pass # 事件回调不允许打断主循环
def _total_tokens(self, usage: Dict[str, int]) -> int:
return int(usage.get("prompt_tokens", 0)) + int(usage.get("completion_tokens", 0))
async def run(self, task: str, system: str = "") -> Dict[str, Any]:
"""执行任务直到模型给出最终答复或触顶。返回最终结果与账目。"""
messages: List[Dict[str, Any]] = []
if system:
messages.append({"role": "system", "content": system})
messages.append({"role": "user", "content": task})
total_in = 0
total_out = 0
last_content = ""
for round_no in range(1, self.max_rounds + 1):
self._emit({"type": "round", "round": round_no})
try:
resp = await self.chat_fn(messages, TOOLS_SPEC)
except Exception as e:
self._emit({"type": "final", "round": round_no, "reason": "error",
"error": f"{type(e).__name__}: {e}"})
return {"response": "", "rounds": round_no, "reason": "error",
"error": f"{type(e).__name__}: {e}",
"prompt_tokens": total_in, "completion_tokens": total_out}
usage = resp.get("usage") or {}
total_in += int(usage.get("prompt_tokens", 0))
total_out += int(usage.get("completion_tokens", 0))
self._emit({"type": "usage", "prompt_tokens": total_in,
"completion_tokens": total_out})
# 金额护栏(D6 同源):token 触顶立即停
if self.token_cap and total_in + total_out > self.token_cap:
self._emit({"type": "final", "round": round_no, "reason": "token_cap"})
return {"response": last_content, "rounds": round_no, "reason": "token_cap",
"error": f"token 熔断({total_in + total_out}/{self.token_cap}",
"prompt_tokens": total_in, "completion_tokens": total_out}
calls = resp.get("tool_calls") or []
if not calls:
self._emit({"type": "final", "round": round_no, "reason": "answer"})
return {"response": resp.get("content") or "", "rounds": round_no,
"reason": "answer", "error": None,
"prompt_tokens": total_in, "completion_tokens": total_out}
# 有工具调用:回填 assistant 消息(带原始 tool_calls 结构)+ 逐个执行
messages.append({
"role": "assistant",
"content": resp.get("content") or None,
"tool_calls": [
{
"id": c["id"],
"type": "function",
"function": {"name": c["name"],
"arguments": json.dumps(c["arguments"], ensure_ascii=False)},
}
for c in calls
],
})
for c in calls:
self._emit({"type": "tool_call", "round": round_no,
"name": c["name"], "arguments": c["arguments"]})
result = self.tools.execute(c["name"], c["arguments"])
preview = json.dumps(result, ensure_ascii=False)
if len(preview) > self.result_preview_chars:
preview = preview[:self.result_preview_chars] + "…(截断)"
self._emit({"type": "tool_result", "round": round_no,
"name": c["name"], "ok": bool(result.get("ok")),
"preview": preview})
messages.append({
"role": "tool",
"tool_call_id": c["id"],
"content": preview,
})
last_content = resp.get("content") or last_content
# 轮次耗尽:不再给工具,让模型立即总结(无 tools 的最后一次调用)
self._emit({"type": "final", "round": self.max_rounds, "reason": "max_rounds"})
try:
messages.append({"role": "user",
"content": "工具轮次已达上限。请基于以上信息立即给出最终答复,不要再调用工具。"})
resp = await self.chat_fn(messages, [])
usage = resp.get("usage") or {}
total_in += int(usage.get("prompt_tokens", 0))
total_out += int(usage.get("completion_tokens", 0))
final_text = resp.get("content") or last_content
except Exception:
final_text = last_content
return {"response": final_text, "rounds": self.max_rounds, "reason": "max_rounds",
"error": "工具轮次达上限,已强制总结",
"prompt_tokens": total_in, "completion_tokens": total_out}