init
This commit is contained in:
904
demo_with_skills.py
Normal file
904
demo_with_skills.py
Normal file
@@ -0,0 +1,904 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""使用 llama.cpp 的 OpenAI 兼容接口创建带有 skills 的 ReActAgent 示例。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
from datetime import datetime
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import time
|
||||
from typing import Any
|
||||
from agentscope.agent import ReActAgent
|
||||
from agentscope.formatter import OpenAIChatFormatter
|
||||
from agentscope.memory import InMemoryMemory
|
||||
from agentscope.message import ImageBlock, Msg, TextBlock, URLSource
|
||||
from agentscope.model import OpenAIChatModel
|
||||
from agentscope.tool import ToolResponse, Toolkit, execute_python_code
|
||||
|
||||
LLAMA_CPP_BASE_URL = os.getenv("LLAMA_CPP_BASE_URL", "http://yyplab.site:8033/v1")
|
||||
LLAMA_CPP_MODEL_NAME = os.getenv(
|
||||
"LLAMA_CPP_MODEL_NAME",
|
||||
"Qwen3.5-35B-A3B-UD-Q8_K_XL.gguf",
|
||||
)
|
||||
LLAMA_CPP_API_KEY = os.getenv("LLAMA_CPP_API_KEY", "EMPTY")
|
||||
WORKSPACE_DIR = Path(__file__).parent.resolve()
|
||||
SKILLS_DIR = Path(__file__).parent / "skills"
|
||||
SCREENSHOT_DIR = Path(__file__).parent / "artifacts" / "screenshots"
|
||||
DEFAULT_GUI_MAX_STEPS = 60
|
||||
|
||||
|
||||
def _load_pyautogui():
|
||||
"""延迟导入 pyautogui，避免缺依赖时脚本启动失败。"""
|
||||
try:
|
||||
import pyautogui # type: ignore
|
||||
except ImportError as exc:
|
||||
raise RuntimeError(
|
||||
"缺少 pyautogui，请先执行 `pip install -r requirements.txt`。",
|
||||
) from exc
|
||||
|
||||
pyautogui.FAILSAFE = True
|
||||
pyautogui.PAUSE = 0.15
|
||||
return pyautogui
|
||||
|
||||
|
||||
def _ensure_screenshot_dir() -> Path:
|
||||
"""确保截图目录存在。"""
|
||||
SCREENSHOT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
return SCREENSHOT_DIR
|
||||
|
||||
|
||||
def normalize_text(value: Any, default: str = "") -> str:
|
||||
"""将模型传入的参数规范化为字符串。"""
|
||||
if value is None:
|
||||
return default
|
||||
if isinstance(value, str):
|
||||
stripped = value.strip()
|
||||
return stripped if stripped else default
|
||||
return str(value)
|
||||
|
||||
|
||||
def normalize_float(value: Any, default: float) -> float:
|
||||
"""将模型传入的参数规范化为浮点数。"""
|
||||
if value is None:
|
||||
return default
|
||||
if isinstance(value, str):
|
||||
stripped = value.strip()
|
||||
if not stripped:
|
||||
return default
|
||||
return float(stripped)
|
||||
return float(value)
|
||||
|
||||
|
||||
def normalize_int(value: Any, default: int) -> int:
|
||||
"""将模型传入的参数规范化为整数。"""
|
||||
if value is None:
|
||||
return default
|
||||
if isinstance(value, str):
|
||||
stripped = value.strip()
|
||||
if not stripped:
|
||||
return default
|
||||
return int(float(stripped))
|
||||
return int(value)
|
||||
|
||||
|
||||
def normalize_keys(value: Any) -> list[str]:
|
||||
"""将模型传入的按键参数规范化为字符串列表。"""
|
||||
if value is None:
|
||||
return []
|
||||
if isinstance(value, str):
|
||||
stripped = value.strip()
|
||||
return [stripped] if stripped else []
|
||||
if isinstance(value, (list, tuple)):
|
||||
return [normalize_text(item) for item in value if normalize_text(item)]
|
||||
normalized = normalize_text(value)
|
||||
return [normalized] if normalized else []
|
||||
|
||||
|
||||
def make_tool_response(text: str, metadata: dict[str, Any] | None = None) -> ToolResponse:
|
||||
"""构造符合 AgentScope 要求的工具响应。"""
|
||||
return ToolResponse(
|
||||
content=[
|
||||
TextBlock(
|
||||
type="text",
|
||||
text=text,
|
||||
),
|
||||
],
|
||||
metadata=metadata,
|
||||
)
|
||||
|
||||
|
||||
def make_success_response(summary: str, **metadata: Any) -> ToolResponse:
|
||||
"""构造成功响应。"""
|
||||
payload = "\n".join(
|
||||
[summary, *(f"{key}={value}" for key, value in metadata.items())],
|
||||
)
|
||||
return make_tool_response(payload, metadata=metadata or None)
|
||||
|
||||
|
||||
def make_error_response(action: str, exc: Exception) -> ToolResponse:
|
||||
"""构造失败响应，避免工具异常直接打断会话。"""
|
||||
return make_tool_response(
|
||||
f"action={action}\nstatus=error\nerror={type(exc).__name__}: {exc}",
|
||||
metadata={
|
||||
"action": action,
|
||||
"status": "error",
|
||||
"error": f"{type(exc).__name__}: {exc}",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def make_multimodal_response(
|
||||
summary: str,
|
||||
image_path: str,
|
||||
metadata: dict[str, Any] | None = None,
|
||||
) -> ToolResponse:
|
||||
"""构造包含截图图片块的工具响应，供多模态模型直接观察。"""
|
||||
return ToolResponse(
|
||||
content=[
|
||||
TextBlock(
|
||||
type="text",
|
||||
text=summary,
|
||||
),
|
||||
ImageBlock(
|
||||
type="image",
|
||||
source=URLSource(
|
||||
type="url",
|
||||
url=image_path,
|
||||
),
|
||||
),
|
||||
],
|
||||
metadata=metadata,
|
||||
)
|
||||
|
||||
|
||||
def read_local_text_file(path: str) -> ToolResponse:
|
||||
"""读取工作区内的文本文件，供 agent 查看 skill 说明。"""
|
||||
try:
|
||||
requested_path = Path(normalize_text(path)).expanduser().resolve()
|
||||
if WORKSPACE_DIR not in requested_path.parents and requested_path != WORKSPACE_DIR:
|
||||
return make_tool_response(
|
||||
"action=read_local_text_file\nstatus=error\nerror=Path is outside the workspace.",
|
||||
metadata={
|
||||
"action": "read_local_text_file",
|
||||
"status": "error",
|
||||
"path": str(requested_path),
|
||||
},
|
||||
)
|
||||
|
||||
if not requested_path.is_file():
|
||||
return make_tool_response(
|
||||
"action=read_local_text_file\nstatus=error\nerror=File does not exist.",
|
||||
metadata={
|
||||
"action": "read_local_text_file",
|
||||
"status": "error",
|
||||
"path": str(requested_path),
|
||||
},
|
||||
)
|
||||
|
||||
content = requested_path.read_text(encoding="utf-8")
|
||||
return make_tool_response(
|
||||
"\n".join(
|
||||
[
|
||||
"action=read_local_text_file",
|
||||
"status=ok",
|
||||
f"path={requested_path}",
|
||||
content,
|
||||
],
|
||||
),
|
||||
metadata={
|
||||
"action": "read_local_text_file",
|
||||
"status": "ok",
|
||||
"path": str(requested_path),
|
||||
},
|
||||
)
|
||||
except Exception as exc:
|
||||
return make_error_response("read_local_text_file", exc)
|
||||
|
||||
|
||||
def save_desktop_screenshot(label: str = "observe") -> ToolResponse:
|
||||
"""截取当前桌面并保存到本地，用于 GUI 操作前后的观察与确认。"""
|
||||
try:
|
||||
pyautogui = _load_pyautogui()
|
||||
screenshot_dir = _ensure_screenshot_dir()
|
||||
label = normalize_text(label, default="observe")
|
||||
safe_label = "".join(
|
||||
char if char.isalnum() or char in "-_" else "_" for char in label
|
||||
)
|
||||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
|
||||
screenshot_path = screenshot_dir / f"{timestamp}_{safe_label}.png"
|
||||
pyautogui.screenshot(str(screenshot_path))
|
||||
return make_multimodal_response(
|
||||
summary=(
|
||||
"action=save_desktop_screenshot\n"
|
||||
"status=ok\n"
|
||||
f"path={screenshot_path}\n"
|
||||
f"label={label}\n"
|
||||
"note=The attached image is the latest desktop screenshot."
|
||||
),
|
||||
image_path=str(screenshot_path),
|
||||
metadata={
|
||||
"action": "save_desktop_screenshot",
|
||||
"status": "ok",
|
||||
"path": str(screenshot_path),
|
||||
"label": label,
|
||||
},
|
||||
)
|
||||
except Exception as exc:
|
||||
return make_error_response("save_desktop_screenshot", exc)
|
||||
|
||||
|
||||
def get_mouse_position() -> ToolResponse:
|
||||
"""获取当前鼠标坐标。"""
|
||||
try:
|
||||
pyautogui = _load_pyautogui()
|
||||
x_pos, y_pos = pyautogui.position()
|
||||
return make_success_response(
|
||||
"action=get_mouse_position\nstatus=ok",
|
||||
x=x_pos,
|
||||
y=y_pos,
|
||||
)
|
||||
except Exception as exc:
|
||||
return make_error_response("get_mouse_position", exc)
|
||||
|
||||
|
||||
def left_click(x: int, y: int, clicks: int = 1, interval: float = 0.2) -> ToolResponse:
|
||||
"""在指定坐标执行鼠标左键单击。"""
|
||||
try:
|
||||
x = normalize_int(x, default=0)
|
||||
y = normalize_int(y, default=0)
|
||||
clicks = normalize_int(clicks, default=1)
|
||||
interval = normalize_float(interval, default=0.2)
|
||||
pyautogui = _load_pyautogui()
|
||||
pyautogui.click(x=x, y=y, clicks=clicks, interval=interval, button="left")
|
||||
return make_success_response(
|
||||
"action=left_click\nstatus=ok",
|
||||
x=x,
|
||||
y=y,
|
||||
clicks=clicks,
|
||||
interval=interval,
|
||||
)
|
||||
except Exception as exc:
|
||||
return make_error_response("left_click", exc)
|
||||
|
||||
|
||||
def type_text(text: str, interval: float = 0.03) -> ToolResponse:
|
||||
"""向当前焦点控件输入文本。"""
|
||||
try:
|
||||
text = normalize_text(text, default="")
|
||||
if not text:
|
||||
return make_tool_response(
|
||||
"action=type_text\nstatus=error\nerror=Empty text is not allowed.",
|
||||
metadata={
|
||||
"action": "type_text",
|
||||
"status": "error",
|
||||
"error": "Empty text is not allowed.",
|
||||
},
|
||||
)
|
||||
interval = normalize_float(interval, default=0.03)
|
||||
pyautogui = _load_pyautogui()
|
||||
pyautogui.write(text, interval=interval)
|
||||
return make_success_response(
|
||||
"action=type_text\nstatus=ok",
|
||||
length=len(text),
|
||||
interval=interval,
|
||||
)
|
||||
except Exception as exc:
|
||||
return make_error_response("type_text", exc)
|
||||
|
||||
|
||||
def press_keys(keys: list[str], interval: float = 0.1) -> ToolResponse:
|
||||
"""依次按下多个按键。"""
|
||||
try:
|
||||
keys = normalize_keys(keys)
|
||||
if not keys:
|
||||
return make_tool_response(
|
||||
"action=press_keys\nstatus=error\nerror=At least one key is required.",
|
||||
metadata={
|
||||
"action": "press_keys",
|
||||
"status": "error",
|
||||
"error": "At least one key is required.",
|
||||
},
|
||||
)
|
||||
interval = normalize_float(interval, default=0.1)
|
||||
pyautogui = _load_pyautogui()
|
||||
pyautogui.press(keys, interval=interval)
|
||||
return make_success_response(
|
||||
"action=press_keys\nstatus=ok",
|
||||
keys=keys,
|
||||
interval=interval,
|
||||
)
|
||||
except Exception as exc:
|
||||
return make_error_response("press_keys", exc)
|
||||
|
||||
|
||||
def hotkey(*keys: str) -> ToolResponse:
|
||||
"""按组合键，例如 Ctrl+A、Ctrl+C。"""
|
||||
try:
|
||||
keys = tuple(normalize_keys(keys))
|
||||
if not keys:
|
||||
return make_tool_response(
|
||||
"action=hotkey\nstatus=error\nerror=At least one key is required.",
|
||||
metadata={
|
||||
"action": "hotkey",
|
||||
"status": "error",
|
||||
"error": "At least one key is required.",
|
||||
},
|
||||
)
|
||||
pyautogui = _load_pyautogui()
|
||||
pyautogui.hotkey(*keys)
|
||||
return make_success_response(
|
||||
"action=hotkey\nstatus=ok",
|
||||
keys=list(keys),
|
||||
)
|
||||
except Exception as exc:
|
||||
return make_error_response("hotkey", exc)
|
||||
|
||||
|
||||
def wait_seconds(seconds: float = 1.0) -> ToolResponse:
|
||||
"""等待指定秒数，给界面渲染或响应留出时间。"""
|
||||
try:
|
||||
seconds = normalize_float(seconds, default=1.0)
|
||||
time.sleep(seconds)
|
||||
return make_success_response(
|
||||
"action=wait_seconds\nstatus=ok",
|
||||
seconds=seconds,
|
||||
)
|
||||
except Exception as exc:
|
||||
return make_error_response("wait_seconds", exc)
|
||||
|
||||
|
||||
def get_active_window_title() -> ToolResponse:
|
||||
"""获取当前活动窗口标题。"""
|
||||
try:
|
||||
pyautogui = _load_pyautogui()
|
||||
title = ""
|
||||
|
||||
get_title = getattr(pyautogui, "getActiveWindowTitle", None)
|
||||
if callable(get_title):
|
||||
title = get_title() or ""
|
||||
|
||||
return make_success_response(
|
||||
"action=get_active_window_title\nstatus=ok",
|
||||
title=title,
|
||||
)
|
||||
except Exception as exc:
|
||||
return make_error_response("get_active_window_title", exc)
|
||||
|
||||
|
||||
def get_clipboard_text() -> ToolResponse:
|
||||
"""读取 Windows 剪贴板文本，用于确认输入结果。"""
|
||||
try:
|
||||
completed = subprocess.run(
|
||||
["powershell", "-NoProfile", "-Command", "Get-Clipboard"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
encoding="utf-8",
|
||||
)
|
||||
status = "ok" if completed.returncode == 0 else "error"
|
||||
return make_tool_response(
|
||||
"\n".join(
|
||||
[
|
||||
"action=get_clipboard_text",
|
||||
f"status={status}",
|
||||
f"text={completed.stdout.strip()}",
|
||||
f"stderr={completed.stderr.strip()}",
|
||||
],
|
||||
),
|
||||
metadata={
|
||||
"action": "get_clipboard_text",
|
||||
"status": status,
|
||||
"text": completed.stdout.strip(),
|
||||
"stderr": completed.stderr.strip(),
|
||||
},
|
||||
)
|
||||
except Exception as exc:
|
||||
return make_error_response("get_clipboard_text", exc)
|
||||
|
||||
|
||||
def paste_text(text: str) -> ToolResponse:
|
||||
"""通过剪贴板粘贴文本，降低输入法对 ASCII 输入的干扰。"""
|
||||
try:
|
||||
text = normalize_text(text, default="")
|
||||
if not text:
|
||||
return make_tool_response(
|
||||
"action=paste_text\nstatus=error\nerror=Empty text is not allowed.",
|
||||
metadata={
|
||||
"action": "paste_text",
|
||||
"status": "error",
|
||||
"error": "Empty text is not allowed.",
|
||||
},
|
||||
)
|
||||
|
||||
set_clipboard = subprocess.run(
|
||||
["powershell", "-NoProfile", "-Command", "Set-Clipboard -Value @'\n" + text + "\n'@"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
encoding="utf-8",
|
||||
)
|
||||
if set_clipboard.returncode != 0:
|
||||
return make_tool_response(
|
||||
"\n".join(
|
||||
[
|
||||
"action=paste_text",
|
||||
"status=error",
|
||||
f"stderr={set_clipboard.stderr.strip()}",
|
||||
],
|
||||
),
|
||||
metadata={
|
||||
"action": "paste_text",
|
||||
"status": "error",
|
||||
"stderr": set_clipboard.stderr.strip(),
|
||||
},
|
||||
)
|
||||
|
||||
pyautogui = _load_pyautogui()
|
||||
pyautogui.hotkey("ctrl", "v")
|
||||
return make_success_response(
|
||||
"action=paste_text\nstatus=ok",
|
||||
length=len(text),
|
||||
)
|
||||
except Exception as exc:
|
||||
return make_error_response("paste_text", exc)
|
||||
|
||||
|
||||
def register_gui_tools(toolkit: Toolkit) -> None:
|
||||
"""注册本地 Windows GUI 操作工具。"""
|
||||
gui_tools = [
|
||||
save_desktop_screenshot,
|
||||
get_mouse_position,
|
||||
left_click,
|
||||
type_text,
|
||||
paste_text,
|
||||
press_keys,
|
||||
hotkey,
|
||||
wait_seconds,
|
||||
get_active_window_title,
|
||||
get_clipboard_text,
|
||||
]
|
||||
|
||||
for tool_func in gui_tools:
|
||||
toolkit.register_tool_function(tool_func)
|
||||
|
||||
|
||||
def register_local_skills(toolkit: Toolkit) -> None:
|
||||
"""自动注册 skills 目录下的全部本地 skills。"""
|
||||
for skill_dir in sorted(SKILLS_DIR.iterdir()):
|
||||
if skill_dir.is_dir() and (skill_dir / "SKILL.md").exists():
|
||||
toolkit.register_agent_skill(str(skill_dir))
|
||||
|
||||
|
||||
def strip_thinking_from_msg(msg: Msg) -> Msg:
|
||||
"""移除消息中的 thinking block。"""
|
||||
all_blocks = msg.get_content_blocks()
|
||||
visible_blocks = [
|
||||
block for block in all_blocks if block.get("type") != "thinking"
|
||||
]
|
||||
|
||||
if len(visible_blocks) == len(all_blocks):
|
||||
return msg
|
||||
|
||||
filtered_msg_dict = msg.to_dict()
|
||||
filtered_msg_dict["content"] = visible_blocks
|
||||
return Msg.from_dict(filtered_msg_dict)
|
||||
|
||||
|
||||
def keep_visible_text_blocks(msg: Msg) -> Msg | None:
|
||||
"""仅保留适合展示给用户的自然语言文本块。"""
|
||||
visible_blocks = []
|
||||
|
||||
for block in msg.get_content_blocks():
|
||||
if block.get("type") != "text":
|
||||
continue
|
||||
|
||||
text = block.get("text", "")
|
||||
if isinstance(text, str) and text.strip():
|
||||
visible_blocks.append(block)
|
||||
|
||||
if not visible_blocks:
|
||||
return None
|
||||
|
||||
filtered_msg_dict = msg.to_dict()
|
||||
filtered_msg_dict["content"] = visible_blocks
|
||||
return Msg.from_dict(filtered_msg_dict)
|
||||
|
||||
|
||||
class SilentThinkingFormatter(OpenAIChatFormatter):
|
||||
"""在发给 OpenAI 兼容接口前移除 thinking block。"""
|
||||
|
||||
async def format(self, msgs: list[Msg], **kwargs: Any) -> list[dict[str, Any]]:
|
||||
sanitized_msgs = [strip_thinking_from_msg(msg) for msg in msgs]
|
||||
return await super().format(sanitized_msgs, **kwargs)
|
||||
|
||||
|
||||
class SilentThinkingReActAgent(ReActAgent):
|
||||
"""仅打印可见文本，隐藏 ThinkingBlock 内容。"""
|
||||
|
||||
def __init__(self, *args: Any, **kwargs: Any) -> None:
|
||||
self.hide_thinking = kwargs.pop("hide_thinking", True)
|
||||
self.debug_model_output = kwargs.pop("debug_model_output", False)
|
||||
super().__init__(*args, **kwargs)
|
||||
self._react_action_cache: dict[str, str] = {}
|
||||
self._react_observation_cache: dict[str, str] = {}
|
||||
self._suppress_max_iter_summary = False
|
||||
|
||||
@staticmethod
|
||||
def _summarize_text(text: str, limit: int = 240) -> str:
|
||||
"""压缩长文本，避免控制台输出过长。"""
|
||||
normalized = " ".join(text.split())
|
||||
if len(normalized) <= limit:
|
||||
return normalized
|
||||
return f"{normalized[:limit]}..."
|
||||
|
||||
def _format_tool_input(self, tool_input: Any) -> str:
|
||||
"""将工具输入格式化为简短可读文本。"""
|
||||
try:
|
||||
serialized = json.dumps(tool_input, ensure_ascii=False)
|
||||
except TypeError:
|
||||
serialized = str(tool_input)
|
||||
return self._summarize_text(serialized)
|
||||
|
||||
def _format_tool_output(self, output: Any) -> str:
|
||||
"""将工具输出格式化为简短可读文本。"""
|
||||
if isinstance(output, list):
|
||||
text_parts = []
|
||||
for item in output:
|
||||
if isinstance(item, dict) and item.get("type") == "text":
|
||||
text_parts.append(item.get("text", ""))
|
||||
else:
|
||||
text_parts.append(str(item))
|
||||
return self._summarize_text("\n".join(text_parts))
|
||||
|
||||
return self._summarize_text(str(output))
|
||||
|
||||
def _print_raw_model_output(self, msg: Msg) -> None:
|
||||
"""打印模型原始返回内容，便于调试多模态观察链路。"""
|
||||
if msg.role != "assistant":
|
||||
return
|
||||
|
||||
try:
|
||||
raw_content = json.dumps(
|
||||
msg.get_content_blocks(),
|
||||
ensure_ascii=False,
|
||||
indent=2,
|
||||
)
|
||||
except TypeError:
|
||||
raw_content = str(msg.get_content_blocks())
|
||||
|
||||
print("===== Raw Model Output =====")
|
||||
print(raw_content)
|
||||
print("===== End Raw Model Output =====")
|
||||
|
||||
async def _print_react_process(self, msg: Msg, last: bool) -> None:
|
||||
"""打印 ReAct 中的 Action / Observation 步骤。"""
|
||||
if not last:
|
||||
return
|
||||
|
||||
for block in msg.get_content_blocks():
|
||||
block_type = block.get("type")
|
||||
|
||||
if block_type == "tool_use":
|
||||
tool_name = block.get("name", "unknown_tool")
|
||||
tool_input = self._format_tool_input(block.get("input", {}))
|
||||
cache_key = f"{msg.id}:{block.get('id', tool_name)}"
|
||||
cache_value = f"{tool_name} {tool_input}"
|
||||
if self._react_action_cache.get(cache_key) == cache_value:
|
||||
continue
|
||||
self._react_action_cache[cache_key] = cache_value
|
||||
print(f"{msg.name}[Action]: {cache_value}")
|
||||
|
||||
elif block_type == "tool_result":
|
||||
tool_name = block.get("name", "unknown_tool")
|
||||
tool_output = self._format_tool_output(block.get("output", ""))
|
||||
cache_key = f"{msg.id}:{block.get('id', tool_name)}"
|
||||
cache_value = f"{tool_name} -> {tool_output}"
|
||||
if self._react_observation_cache.get(cache_key) == cache_value:
|
||||
continue
|
||||
self._react_observation_cache[cache_key] = cache_value
|
||||
print(f"{msg.name}[Observation]: {cache_value}")
|
||||
|
||||
async def print(self, msg: Msg, last: bool = True, speech=None) -> None:
|
||||
"""展示自然语言回复，并打印可读的 ReAct 过程。"""
|
||||
await self._print_react_process(msg, last=last)
|
||||
|
||||
if self.debug_model_output and last:
|
||||
self._print_raw_model_output(msg)
|
||||
|
||||
filtered_msg = msg
|
||||
if self.hide_thinking:
|
||||
filtered_msg = strip_thinking_from_msg(filtered_msg)
|
||||
|
||||
filtered_msg = keep_visible_text_blocks(filtered_msg)
|
||||
if filtered_msg is None:
|
||||
return
|
||||
|
||||
await super().print(filtered_msg, last=last, speech=speech)
|
||||
|
||||
async def _summarizing(self) -> Msg:
|
||||
"""在 GUI 单步模式下，达到上限时不要生成误导性总结。"""
|
||||
if self._suppress_max_iter_summary:
|
||||
return Msg(self.name, "", "assistant")
|
||||
return await super()._summarizing()
|
||||
|
||||
|
||||
def looks_like_gui_task(text: str) -> bool:
|
||||
"""粗略判断用户请求是否属于 GUI 自动化任务。"""
|
||||
gui_keywords = [
|
||||
"打开",
|
||||
"浏览器",
|
||||
"chrome",
|
||||
"edge",
|
||||
"bing",
|
||||
"窗口",
|
||||
"桌面",
|
||||
"点击",
|
||||
"输入",
|
||||
"回车",
|
||||
"搜索",
|
||||
"gui",
|
||||
]
|
||||
lowered = text.lower()
|
||||
return any(keyword in text or keyword in lowered for keyword in gui_keywords)
|
||||
|
||||
|
||||
def should_auto_continue_gui(reply: Msg | None) -> bool:
|
||||
"""判断 GUI 任务回复是否表现为中途暂停而非真正完成。"""
|
||||
if reply is None:
|
||||
return False
|
||||
|
||||
text = (reply.get_text_content() or "").strip()
|
||||
if not text:
|
||||
return True
|
||||
|
||||
stop_markers = [
|
||||
"是否需要我继续",
|
||||
"是否需要我继续执行",
|
||||
"下一步计划",
|
||||
"当前任务状态总结",
|
||||
"当前问题",
|
||||
"我将",
|
||||
"准备继续",
|
||||
"让我尝试",
|
||||
"继续执行",
|
||||
"尚未成功",
|
||||
"未成功",
|
||||
"仍然是",
|
||||
"仍为",
|
||||
"当前活动窗口是",
|
||||
"需要先打开",
|
||||
"尝试打开",
|
||||
"尝试关闭",
|
||||
"回到桌面",
|
||||
"<tool_call>",
|
||||
]
|
||||
done_markers = [
|
||||
"任务已完成",
|
||||
"已经完成",
|
||||
"已成功",
|
||||
"搜索结果",
|
||||
"已打开 bing",
|
||||
"已输入test",
|
||||
"已回车搜索",
|
||||
"明确无法继续",
|
||||
"无法继续",
|
||||
]
|
||||
blocked_markers = [
|
||||
"需要用户",
|
||||
"请用户",
|
||||
"权限不足",
|
||||
"无法定位",
|
||||
"无法识别",
|
||||
"无法访问",
|
||||
"明确受阻",
|
||||
]
|
||||
|
||||
if any(marker in text for marker in done_markers):
|
||||
return False
|
||||
|
||||
if any(marker in text for marker in blocked_markers):
|
||||
return False
|
||||
|
||||
if any(marker in text for marker in stop_markers):
|
||||
return True
|
||||
|
||||
# For GUI tasks, default to continue unless the reply clearly indicates
|
||||
# success or a real blocker. This prevents the agent from stopping after
|
||||
# an intermediate summary.
|
||||
return True
|
||||
|
||||
|
||||
def register_optional_skill_access_tools(toolkit: Toolkit) -> None:
|
||||
"""注册受限的本地文本读取工具，避免 agent 使用 shell 绕开 GUI。"""
|
||||
toolkit.register_tool_function(read_local_text_file)
|
||||
print("已注册可选工具: read_local_text_file")
|
||||
|
||||
|
||||
def build_agent(show_thinking: bool, debug_model_output: bool) -> ReActAgent:
|
||||
"""创建一个带有示例 skill 的 ReActAgent。"""
|
||||
toolkit = Toolkit()
|
||||
toolkit.register_tool_function(execute_python_code)
|
||||
register_optional_skill_access_tools(toolkit)
|
||||
register_gui_tools(toolkit)
|
||||
register_local_skills(toolkit)
|
||||
|
||||
agent_cls = SilentThinkingReActAgent
|
||||
formatter = (
|
||||
OpenAIChatFormatter(promote_tool_result_images=True)
|
||||
if show_thinking
|
||||
else SilentThinkingFormatter(promote_tool_result_images=True)
|
||||
)
|
||||
|
||||
agent = agent_cls(
|
||||
name="Jarvis",
|
||||
sys_prompt=(
|
||||
"你是一个名为 Jarvis 的助手。请优先使用已注册的 skill 完成任务。"
|
||||
"对于 GUI 自动化任务，不要在中途停下来询问用户是否继续；"
|
||||
"你应当持续执行观察、操作、复查循环，直到任务完成或明确无法继续。"
|
||||
"每一轮只做一个最小必要动作，并在动作后立刻重新观察。"
|
||||
"不要预设固定的长操作顺序，不要在旧假设上连续执行多步。"
|
||||
),
|
||||
model=OpenAIChatModel(
|
||||
model_name=LLAMA_CPP_MODEL_NAME,
|
||||
api_key=LLAMA_CPP_API_KEY,
|
||||
stream=True,
|
||||
client_kwargs={
|
||||
"base_url": LLAMA_CPP_BASE_URL,
|
||||
"timeout": 60,
|
||||
},
|
||||
generate_kwargs={
|
||||
"temperature": 0.7,
|
||||
},
|
||||
),
|
||||
max_iters=8,
|
||||
hide_thinking=not show_thinking,
|
||||
debug_model_output=debug_model_output,
|
||||
formatter=formatter,
|
||||
toolkit=toolkit,
|
||||
memory=InMemoryMemory(),
|
||||
)
|
||||
|
||||
return agent
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
"""解析命令行参数。"""
|
||||
parser = argparse.ArgumentParser(description="AgentScope skills demo")
|
||||
parser.add_argument(
|
||||
"--skip-agent-run",
|
||||
action="store_true",
|
||||
help="只打印 skill 注册结果，不调用远程模型。",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--message",
|
||||
help="发送给 ReActAgent 的单轮消息；不传时进入交互式对话。",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--show-thinking",
|
||||
action="store_true",
|
||||
help="在控制台显示模型 thinking 内容。默认隐藏。",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--user-name",
|
||||
default="user",
|
||||
help="交互式对话中使用的用户名。",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--debug-model-output",
|
||||
action="store_true",
|
||||
help="打印每轮 assistant 的原始模型返回内容，便于调试截图观察链路。",
|
||||
)
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
async def chat_once(agent: ReActAgent, user_name: str, user_input: str) -> None:
|
||||
"""发送一条用户消息。"""
|
||||
msg = Msg(
|
||||
name=user_name,
|
||||
content=user_input,
|
||||
role="user",
|
||||
)
|
||||
|
||||
is_gui_task = looks_like_gui_task(user_input)
|
||||
original_max_iters = agent.max_iters
|
||||
original_suppress = getattr(agent, "_suppress_max_iter_summary", False)
|
||||
|
||||
if is_gui_task:
|
||||
agent.max_iters = 1
|
||||
if hasattr(agent, "_suppress_max_iter_summary"):
|
||||
agent._suppress_max_iter_summary = True
|
||||
|
||||
try:
|
||||
reply = await agent(msg)
|
||||
|
||||
if not is_gui_task:
|
||||
return
|
||||
|
||||
for _ in range(DEFAULT_GUI_MAX_STEPS):
|
||||
if not should_auto_continue_gui(reply):
|
||||
return
|
||||
|
||||
follow_up = Msg(
|
||||
name=user_name,
|
||||
content=(
|
||||
"继续执行当前 GUI 任务，不要停下来询问是否继续。"
|
||||
"请继续按照观察、操作、复查的闭环执行，直到任务完成或明确无法继续。"
|
||||
"如果你刚才输出的是计划、状态总结，或原样的 <tool_call> 文本，请立即把它落实为真实工具调用并继续。"
|
||||
"下一轮只做一个最小必要动作，并在动作后立刻截图复查。"
|
||||
"这一轮最多只允许一次真实工具动作，不要连续做多步。"
|
||||
),
|
||||
role="user",
|
||||
)
|
||||
reply = await agent(follow_up)
|
||||
finally:
|
||||
agent.max_iters = original_max_iters
|
||||
if hasattr(agent, "_suppress_max_iter_summary"):
|
||||
agent._suppress_max_iter_summary = original_suppress
|
||||
|
||||
|
||||
async def chat_loop(agent: ReActAgent, user_name: str) -> None:
|
||||
"""通过命令行输入与模型持续对话。"""
|
||||
print("\n===== 交互式对话已启动 =====")
|
||||
print("输入 /exit 结束对话，输入 /help 查看提示。")
|
||||
|
||||
while True:
|
||||
try:
|
||||
user_input = await asyncio.to_thread(input, f"\n{user_name}> ")
|
||||
except (EOFError, KeyboardInterrupt):
|
||||
print("\n对话结束。")
|
||||
return
|
||||
|
||||
user_input = user_input.strip()
|
||||
|
||||
if not user_input:
|
||||
continue
|
||||
|
||||
if user_input in {"/exit", "/quit"}:
|
||||
print("对话结束。")
|
||||
return
|
||||
|
||||
if user_input == "/help":
|
||||
print("直接输入消息即可和模型对话。使用 /exit 或 /quit 退出。")
|
||||
continue
|
||||
|
||||
try:
|
||||
await chat_once(agent, user_name, user_input)
|
||||
except Exception as exc:
|
||||
print(f"本轮对话执行失败：{type(exc).__name__}: {exc}")
|
||||
print("你可以继续输入下一条消息，或输入 /exit 退出。")
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
args = parse_args()
|
||||
agent = build_agent(
|
||||
show_thinking=args.show_thinking,
|
||||
debug_model_output=args.debug_model_output,
|
||||
)
|
||||
|
||||
print("===== 已注册 skill 提示词 =====")
|
||||
print(agent.toolkit.get_agent_skill_prompt())
|
||||
|
||||
print("\n===== 当前模型配置 =====")
|
||||
print(f"base_url={LLAMA_CPP_BASE_URL}")
|
||||
print(f"model_name={LLAMA_CPP_MODEL_NAME}")
|
||||
print(f"show_thinking={args.show_thinking}")
|
||||
print(f"debug_model_output={args.debug_model_output}")
|
||||
|
||||
if args.skip_agent_run:
|
||||
print("\n===== 已跳过远程模型调用 =====")
|
||||
return
|
||||
|
||||
if args.message:
|
||||
print("\n===== 单轮对话模式 =====")
|
||||
await chat_once(agent, args.user_name, args.message)
|
||||
return
|
||||
|
||||
await chat_loop(agent, args.user_name)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user