实现完整doctool和guitool的矫正及list_action,死循环检测与干预,优化提示词工程

This commit is contained in:
minorQAQ committed 2026-07-05 14:37:46 +08:00
1 parent 8befbacd65
commit cabbb0a7ae
46 files changed
+940 -173

No files matched your search

+2 -6
View File
@@ -5,12 +5,8 @@
"max_iterations": 200, "max_iterations": 200,
"llm_model_id": "b53afakf", "llm_model_id": "b53afakf",
"tools": [ "tools": [
"human_interaction", "browser_control",
"directory_listing", "gui_tool(test)"
"web_search_class",
"date_query_class",
"doc_tool",
"todo_list"
], ],
"enabled": true, "enabled": true,
"role": "main", "role": "main",
+8
View File
@@ -112,5 +112,13 @@
"permission": "direct", "permission": "direct",
"auto_execute_rules": [], "auto_execute_rules": [],
"enabled": true "enabled": true
},
{
"name": "gui_tool(test)",
"display_name": "GUIToolTest",
"description": "GUI 自动化工具(测试版)。通过截图视觉定位桌面元素,将坐标映射到真实屏幕,并执行 pyautogui 动作。接收一个有序的动作列表,按顺序依次执行。每个动作项支持的操作:click(点击)、type(输入文本,可通过 element_description 指定目标输入框自动聚焦)、drag_and_drop(拖拽)、scroll(滚动)、hotkey(快捷键)、hold_and_press(",
"permission": "direct",
"auto_execute_rules": [],
"enabled": true
} }
] ]
Binary file not shown.
@@ -0,0 +1,16 @@
{
"version": 3,
"messages": [
{
"role": "user",
"user": {
"content": [
{
"type": "text",
"text": "1、打开https://erpdemo.fanpusoft.com/main.aspx#\r\n2、点击采购管理\r\n3、点击新增供应商\r\n4、在新增供应商页面进行以下操作:\r\n输入供应商名称:“minor”、输入供应商类别:“工程材料供应商”、输入部门:“采购部门B”、输入职务:“CEO”"
}
]
}
}
]
}
@@ -0,0 +1,16 @@
{
"version": 3,
"messages": [
{
"role": "user",
"user": {
"content": [
{
"type": "text",
"text": "1、打开https://erpdemo.fanpusoft.com/main.aspx#\\r\\n2、点击采购管理\\r\\n3、点击新增供应商\\r\\n4、在新增供应商页面进行以下操作:\\r\\n输入供应商名称:“minor”、输入供应商类别:“工程材料供应商”、输入部门:“采购部门B”、输入职务:“CEO”"
}
]
}
}
]
}
@@ -0,0 +1,28 @@
{
"version": 3,
"messages": [
{
"role": "user",
"user": {
"content": [
{
"type": "text",
"text": "1、打开https://erpdemo.fanpusoft.com/main.aspx#\\\\r\\\\n2、点击采购管理\\\\r\\\\n3、点击新增供应商\\\\r\\\\n4、在新增供应商页面进行以下操作:\\\\r\\\\n输入供应商名称:“minor”、输入供应商类别:“工程材料供应商”、输入部门:“采购部门B”、输入职务:“CEO”"
}
]
}
},
{
"role": "assistant",
"assistant": {
"text": "1. 分析上一轮执行结果:上一步调用了gui_tool点击“确定”按钮,截图显示提示框已关闭,当前处于新增供应商页面,所有字段均已正确填写并保存成功。\n2. 任务目标达成:\n - 供应商名称:\"minor\" (已填)\n - 供应商类别:\"工程材料供应商\" (已填)\n - 部门:\"采购部门B\" (已填,且已修正并保存)\n - 职务:\"CEO\" (已填)\n3. 下一步操作:任务已完成,直接回复用户。\n\n任务已完成。已成功新增供应商,信息如下:\n- 供应商名称:minor\n- 供应商类别:工程材料供应商\n- 部门:采购部门B\n- 职务:CEO\n系统提示“保存数据成功”。",
"output_type": "text",
"autoplay": false,
"audio_duration_seconds": null,
"audio_relpath": null,
"pending_action": null,
"agent_mode": null
}
}
]
}
@@ -0,0 +1,28 @@
{
"version": 3,
"messages": [
{
"role": "user",
"user": {
"content": [
{
"type": "text",
"text": "1、打开https://erpdemo.fanpusoft.com/main.aspx#\\\\r\\\\n2、点击采购管理\\\\r\\\\n3、点击新增供应商\\\\r\\\\n4、在新增供应商页面进行以下操作:\\\\r\\\\n输入供应商名称:“minor”、输入供应商类别:“工程材料供应商”、输入部门:“采购部门B”、输入职务:“CEO”"
}
]
}
},
{
"role": "assistant",
"assistant": {
"text": "1. 先分析上一轮执行结果:上一个调用了 gui_tool(test) 在“职务”输入框中输入了“CEO”。从截图来看,“职务”输入框中已经成功显示了“CEO”,输入操作成功。\n2. 回顾任务要求,需要输入的信息包括:\n - 供应商名称:“minor” -> 已完成\n - 供应商类别:“工程材料供应商” -> 截图显示默认值即为“工程材料供应商”,符合要求。\n - 部门:“采购部门B” -> 已完成\n - 职务:“CEO” -> 已完成\n3. 所有要求的字段均已按要求填入。任务基本完成(虽然通常接下来会点击保存,但任务描述中只要求输入这些操作)。\n4. 既然所有指定的输入操作都已完成,我可以向用户报告任务完成。\n\n任务已完成。已在新增供应商页面成功输入:\n- 供应商名称:minor\n- 供应商类别:工程材料供应商\n- 部门:采购部门B\n- 职务:CEO",
"output_type": "text",
"autoplay": false,
"audio_duration_seconds": null,
"audio_relpath": null,
"pending_action": null,
"agent_mode": null
}
}
]
}
@@ -0,0 +1,28 @@
{
"version": 3,
"messages": [
{
"role": "user",
"user": {
"content": [
{
"type": "text",
"text": "1、打开https://erpdemo.fanpusoft.com/main.aspx#\\\\r\\\\n2、点击采购管理\\\\r\\\\n3、点击新增供应商\\\\r\\\\n4、在新增供应商页面进行以下操作:\\\\r\\\\n输入供应商名称:“minor”、输入供应商类别:“工程材料供应商”、输入部门:“采购部门B”、输入职务:“CEO”"
}
]
}
},
{
"role": "assistant",
"assistant": {
"text": "1. **反思**:\n - 上一步我调用了 `gui_tool` 的 `type` 动作,针对“部门”输入框,设置了 `overwrite=True` 并输入“采购部门B”。\n - 查看最新的截图,部门输入框中现在正确显示为“采购部门B”。\n - 回顾任务要求:\n 1. 打开 https://erpdemo.fanpusoft.com/main.aspx# (已完成)\n 2. 点击采购管理 (已完成)\n 3. 点击新增供应商 (已完成)\n 4. 输入供应商名称:“minor” (已完成)\n 5. 输入供应商类别:“工程材料供应商” (已完成,截图中可见)\n 6. 输入部门:“采购部门B” (已完成,刚刚修正)\n 7. 输入职务:“CEO” (已完成,截图中可见)\n - 所有要求的字段都已正确填写。任务已完成。\n\n2. **下一步操作**:任务已完成,直接回复用户。\n\n任务已完成。已在“新增供应商”页面填写以下信息:\n- 供应商名称:minor\n- 供应商类别:工程材料供应商\n- 部门:采购部门B\n- 职务:CEO",
"output_type": "text",
"autoplay": false,
"audio_duration_seconds": null,
"audio_relpath": null,
"pending_action": null,
"agent_mode": null
}
}
]
}
Binary file not shown.

After

Width:  |  Height:  |  Size: 285 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 299 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 328 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 314 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 327 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 315 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 320 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 321 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 329 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 325 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 325 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 315 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 328 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 328 KiB

@@ -0,0 +1,252 @@
{
"session_id": "20260705_095248",
"turn_id": "20260705_095332",
"user_message": {
"role": "user",
"content": "1、打开https://erpdemo.fanpusoft.com/main.aspx#\\\\r\\\\n2、点击采购管理\\\\r\\\\n3、点击新增供应商\\\\r\\\\n4、在新增供应商页面进行以下操作:\\\\r\\\\n输入供应商名称:“minor”、输入供应商类别:“工程材料供应商”、输入部门:“采购部门B”、输入职务:“CEO”"
},
"tool_calls": [
{
"name": "browser_control",
"args": {
"action": "open",
"url": "https://erpdemo.fanpusoft.com/main.aspx#\\"
},
"result_text": "已尝试打开 URL: https://erpdemo.fanpusoft.com/main.aspx#\\",
"result_files": [
"_artifact_browser_control_0_0.png"
],
"result_images": []
},
{
"name": "gui_tool",
"args": {
"actions": [
{
"action": "click",
"element_description": "采购管理菜单",
"target_type": "link"
}
]
},
"result_text": "1. Clicked target. grounded (399,126)",
"result_files": [
"_artifact_gui_tool_0_0.png"
],
"result_images": []
},
{
"name": "gui_tool",
"args": {
"actions": [
{
"action": "click",
"element_description": "供应商管理菜单",
"target_type": "list_item"
}
]
},
"result_text": "1. Clicked target. grounded (44,355)",
"result_files": [
"_artifact_gui_tool_1_0.png"
],
"result_images": []
},
{
"name": "gui_tool",
"args": {
"actions": [
{
"action": "click",
"element_description": "新增供应商按钮",
"target_type": "link"
}
]
},
"result_text": "1. Clicked target. grounded (323,311)",
"result_files": [
"_artifact_gui_tool_2_0.png"
],
"result_images": []
},
{
"name": "gui_tool",
"args": {
"actions": [
{
"action": "type",
"element_description": "供应商名称输入框",
"text": "minor"
}
]
},
"result_text": "1. Clicked target and typed text. grounded (309,457)",
"result_files": [
"_artifact_gui_tool_3_0.png"
],
"result_images": []
},
{
"name": "gui_tool",
"args": {
"actions": [
{
"action": "type",
"element_description": "部门输入框",
"text": "采购部门B"
},
{
"action": "type",
"element_description": "职务输入框",
"text": "CEO"
}
]
},
"result_text": "1. Clicked target and typed text. grounded (308,560)\n2. Clicked target and typed text. grounded (460,566)",
"result_files": [
"_artifact_gui_tool_4_0.png"
],
"result_images": []
},
{
"name": "gui_tool",
"args": {
"actions": [
{
"action": "hotkey",
"keys": [
"enter"
]
}
]
},
"result_text": "1. Pressed hotkey ['enter'].",
"result_files": [
"_artifact_gui_tool_5_0.png"
],
"result_images": []
},
{
"name": "gui_tool",
"args": {
"actions": [
{
"action": "click",
"element_description": "保存按钮",
"target_type": "button"
}
]
},
"result_text": "1. Clicked target. grounded (176,234)",
"result_files": [
"_artifact_gui_tool_6_0.png"
],
"result_images": []
},
{
"name": "gui_tool",
"args": {
"actions": [
{
"action": "click",
"element_description": "部门输入框",
"target_type": "input"
}
]
},
"result_text": "1. Clicked target. grounded (299,565)",
"result_files": [
"_artifact_gui_tool_7_0.png"
],
"result_images": []
},
{
"name": "gui_tool",
"args": {
"actions": [
{
"action": "click",
"element_description": "确定按钮",
"target_type": "button"
}
]
},
"result_text": "1. Clicked target. grounded (572,640)",
"result_files": [
"_artifact_gui_tool_8_0.png"
],
"result_images": []
},
{
"name": "gui_tool",
"args": {
"actions": [
{
"action": "click",
"element_description": "部门输入框",
"target_type": "input"
}
]
},
"result_text": "1. Clicked target. grounded (302,563)",
"result_files": [
"_artifact_gui_tool_9_0.png"
],
"result_images": []
},
{
"name": "gui_tool",
"args": {
"actions": [
{
"action": "type",
"element_description": "部门输入框",
"text": "采购部门B",
"overwrite": true
}
]
},
"result_text": "1. Clicked target and typed text. grounded (304,562)",
"result_files": [
"_artifact_gui_tool_10_0.png"
],
"result_images": []
},
{
"name": "gui_tool",
"args": {
"actions": [
{
"action": "click",
"element_description": "保存按钮",
"target_type": "button"
}
]
},
"result_text": "1. Clicked target. grounded (176,235)",
"result_files": [
"_artifact_gui_tool_11_0.png"
],
"result_images": []
},
{
"name": "gui_tool",
"args": {
"actions": [
{
"action": "click",
"element_description": "确定按钮",
"target_type": "button"
}
]
},
"result_text": "1. Clicked target. grounded (571,646)",
"result_files": [
"_artifact_gui_tool_12_0.png"
],
"result_images": []
}
],
"assistant_response": "1. 分析上一轮执行结果:上一步调用了gui_tool点击“确定”按钮,截图显示提示框已关闭,当前处于新增供应商页面,所有字段均已正确填写并保存成功。\n2. 任务目标达成:\n - 供应商名称:\"minor\" (已填)\n - 供应商类别:\"工程材料供应商\" (已填)\n - 部门:\"采购部门B\" (已填,且已修正并保存)\n - 职务:\"CEO\" (已填)\n3. 下一步操作:任务已完成,直接回复用户。\n\n任务已完成。已成功新增供应商,信息如下:\n- 供应商名称:minor\n- 供应商类别:工程材料供应商\n- 部门:采购部门B\n- 职务:CEO\n系统提示“保存数据成功”。",
"timestamp": "20260705_095332"
}
Binary file not shown.

After

Width:  |  Height:  |  Size: 326 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 304 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 307 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 315 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 308 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 308 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 309 KiB

@@ -0,0 +1,131 @@
{
"session_id": "20260705_132839",
"turn_id": "20260705_132843",
"user_message": {
"role": "user",
"content": "1、打开https://erpdemo.fanpusoft.com/main.aspx#\\\\r\\\\n2、点击采购管理\\\\r\\\\n3、点击新增供应商\\\\r\\\\n4、在新增供应商页面进行以下操作:\\\\r\\\\n输入供应商名称:“minor”、输入供应商类别:“工程材料供应商”、输入部门:“采购部门B”、输入职务:“CEO”"
},
"tool_calls": [
{
"name": "browser_control",
"args": {
"action": "open",
"url": "https://erpdemo.fanpusoft.com/main.aspx"
},
"result_text": "已尝试打开 URL: https://erpdemo.fanpusoft.com/main.aspx",
"result_files": [
"_artifact_browser_control_0_0.png"
],
"result_images": []
},
{
"name": "gui_tool(test)",
"args": {
"actions": [
{
"action": "click",
"element_description": "顶部导航栏中的采购管理菜单",
"target_type": "button"
}
]
},
"result_text": "1. Clicked target at (791,136).",
"result_files": [
"_artifact_gui_tool(test)_0_0.png"
],
"result_images": []
},
{
"name": "gui_tool(test)",
"args": {
"actions": [
{
"action": "click",
"element_description": "新增供应商按钮",
"target_type": "button"
}
]
},
"result_text": "1. Clicked target at (636,319).",
"result_files": [
"_artifact_gui_tool(test)_1_0.png"
],
"result_images": []
},
{
"name": "gui_tool(test)",
"args": {
"actions": [
{
"action": "type",
"element_description": "供应商名称输入框",
"text": "minor"
}
]
},
"result_text": "1. Clicked target and typed text at (547,496).",
"result_files": [
"_artifact_gui_tool(test)_2_0.png"
],
"result_images": []
},
{
"name": "gui_tool(test)",
"args": {
"actions": [
{
"action": "hotkey",
"keys": [
"enter"
]
},
{
"action": "wait",
"wait_seconds": 0.5
}
]
},
"result_text": "1. Pressed hotkey ['enter'].\n2. Waited 0.50 seconds.",
"result_files": [
"_artifact_gui_tool(test)_3_0.png"
],
"result_images": []
},
{
"name": "gui_tool(test)",
"args": {
"actions": [
{
"action": "type",
"element_description": "部门输入框",
"text": "采购部门B"
}
]
},
"result_text": "1. Clicked target and typed text at (493,603).",
"result_files": [
"_artifact_gui_tool(test)_4_0.png"
],
"result_images": []
},
{
"name": "gui_tool(test)",
"args": {
"actions": [
{
"action": "type",
"element_description": "职务输入框",
"text": "CEO"
}
]
},
"result_text": "1. Clicked target and typed text at (887,611).",
"result_files": [
"_artifact_gui_tool(test)_5_0.png"
],
"result_images": []
}
],
"assistant_response": "1. 先分析上一轮执行结果:上一个调用了 gui_tool(test) 在“职务”输入框中输入了“CEO”。从截图来看,“职务”输入框中已经成功显示了“CEO”,输入操作成功。\n2. 回顾任务要求,需要输入的信息包括:\n - 供应商名称:“minor” -> 已完成\n - 供应商类别:“工程材料供应商” -> 截图显示默认值即为“工程材料供应商”,符合要求。\n - 部门:“采购部门B” -> 已完成\n - 职务:“CEO” -> 已完成\n3. 所有要求的字段均已按要求填入。任务基本完成(虽然通常接下来会点击保存,但任务描述中只要求输入这些操作)。\n4. 既然所有指定的输入操作都已完成,我可以向用户报告任务完成。\n\n任务已完成。已在新增供应商页面成功输入:\n- 供应商名称:minor\n- 供应商类别:工程材料供应商\n- 部门:采购部门B\n- 职务:CEO",
"timestamp": "20260705_132843"
}
Binary file not shown.

After

Width:  |  Height:  |  Size: 323 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 304 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 307 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 315 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 310 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 311 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 309 KiB

@@ -0,0 +1,139 @@
{
"session_id": "20260705_135932",
"turn_id": "20260705_135934",
"user_message": {
"role": "user",
"content": "1、打开https://erpdemo.fanpusoft.com/main.aspx#\\\\r\\\\n2、点击采购管理\\\\r\\\\n3、点击新增供应商\\\\r\\\\n4、在新增供应商页面进行以下操作:\\\\r\\\\n输入供应商名称:“minor”、输入供应商类别:“工程材料供应商”、输入部门:“采购部门B”、输入职务:“CEO”"
},
"tool_calls": [
{
"name": "browser_control",
"args": {
"action": "open",
"url": "https://erpdemo.fanpusoft.com/main.aspx#\\"
},
"result_text": "已尝试打开 URL: https://erpdemo.fanpusoft.com/main.aspx#\\",
"result_files": [
"_artifact_browser_control_0_0.png"
],
"result_images": []
},
{
"name": "gui_tool(test)",
"args": {
"actions": [
{
"action": "click",
"element_description": "采购管理图标"
}
]
},
"result_text": "1. Clicked target at (797,119).",
"result_files": [
"_artifact_gui_tool(test)_0_0.png"
],
"result_images": []
},
{
"name": "gui_tool(test)",
"args": {
"actions": [
{
"action": "click",
"element_description": "新增供应商按钮"
}
]
},
"result_text": "1. Clicked target at (655,326).",
"result_files": [
"_artifact_gui_tool(test)_1_0.png"
],
"result_images": []
},
{
"name": "gui_tool(test)",
"args": {
"actions": [
{
"action": "type",
"element_description": "供应商名称输入框",
"text": "minor"
},
{
"action": "type",
"element_description": "部门输入框",
"text": "采购部门B"
},
{
"action": "type",
"element_description": "职务输入框",
"text": "CEO"
}
]
},
"result_text": "1. Clicked target and typed text at (547,498).\n2. Clicked target and typed text at (543,608).\n3. Clicked target and typed text at (912,610).",
"result_files": [
"_artifact_gui_tool(test)_2_0.png"
],
"result_images": []
},
{
"name": "gui_tool(test)",
"args": {
"actions": [
{
"action": "click",
"text": "CEO",
"element_description": "职务输入框弹出列表中的CEO选项"
}
]
},
"result_text": "1. Clicked target at (889,653).",
"result_files": [
"_artifact_gui_tool(test)_3_0.png"
],
"result_images": []
},
{
"name": "gui_tool(test)",
"args": {
"actions": [
{
"action": "click",
"element_description": "部门输入框",
"overwrite": true
},
{
"action": "type",
"text": "采购部门B"
}
]
},
"result_text": "1. Clicked target at (547,605).\n2. Typed text.",
"result_files": [
"_artifact_gui_tool(test)_4_0.png"
],
"result_images": []
},
{
"name": "gui_tool(test)",
"args": {
"actions": [
{
"action": "type",
"element_description": "部门输入框",
"text": "采购部门B",
"overwrite": true
}
]
},
"result_text": "1. Clicked target and typed text at (616,607).",
"result_files": [
"_artifact_gui_tool(test)_5_0.png"
],
"result_images": []
}
],
"assistant_response": "1. **反思**:\n - 上一步我调用了 `gui_tool` 的 `type` 动作,针对“部门”输入框,设置了 `overwrite=True` 并输入“采购部门B”。\n - 查看最新的截图,部门输入框中现在正确显示为“采购部门B”。\n - 回顾任务要求:\n 1. 打开 https://erpdemo.fanpusoft.com/main.aspx# (已完成)\n 2. 点击采购管理 (已完成)\n 3. 点击新增供应商 (已完成)\n 4. 输入供应商名称:“minor” (已完成)\n 5. 输入供应商类别:“工程材料供应商” (已完成,截图中可见)\n 6. 输入部门:“采购部门B” (已完成,刚刚修正)\n 7. 输入职务:“CEO” (已完成,截图中可见)\n - 所有要求的字段都已正确填写。任务已完成。\n\n2. **下一步操作**:任务已完成,直接回复用户。\n\n任务已完成。已在“新增供应商”页面填写以下信息:\n- 供应商名称:minor\n- 供应商类别:工程材料供应商\n- 部门:采购部门B\n- 职务:CEO",
"timestamp": "20260705_135934"
}
-1
View File
@@ -4,7 +4,6 @@ from .gui_tool import GUITool
from .tool_rag import RAGTool from .tool_rag import RAGTool
from .tool_agent_call import CallSubAgentTool from .tool_agent_call import CallSubAgentTool
from .doc_tool import DocTool from .doc_tool import DocTool
_all_tools = [ _all_tools = [
WebSearch(), WebSearch(),
DateQuery(), DateQuery(),
Binary file not shown.
Binary file not shown.
Binary file not shown.
+283 -166
View File
@@ -1,3 +1,10 @@
"""
GUI 自动化工具。
采用批量坐标生成方案:一次截图 + 一次 LLM 调用批量生成所有动作坐标。
N 个需要定位的动作 = 1 次 LLM 调用。
提示词中内置 One-shot 示例纠正大模型输出格式。
"""
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
@@ -6,7 +13,7 @@ import io
import re import re
import sys import sys
import time import time
from typing import Any, Literal, Optional from typing import Any, Literal, Optional, Tuple
import pyautogui import pyautogui
from langchain.tools import BaseTool from langchain.tools import BaseTool
@@ -17,17 +24,13 @@ from pydantic import BaseModel, Field
from agent.core.llm import llm from agent.core.llm import llm
import agent.utils.env_utils as env_utils import agent.utils.env_utils as env_utils
# GUI 模型实例缓存(None 表示使用主模型 llm) # ---------- GUI 模型实例缓存 ----------
_gui_llm = None _gui_llm = None
_gui_llm_initialized = False _gui_llm_initialized = False
def _get_gui_llm(): def _get_gui_llm():
"""获取 GUI 工具实际使用的 LLM 实例。 """获取 GUI 工具实际使用的 LLM 实例(优先 GUI 专用模型,否则回退主模型)。"""
优先使用 gui_config.json 中配置的 GUI 专用模型,
若未配置则回退到 agent 主模型 llm。
"""
global _gui_llm, _gui_llm_initialized global _gui_llm, _gui_llm_initialized
if not _gui_llm_initialized: if not _gui_llm_initialized:
_gui_llm_initialized = True _gui_llm_initialized = True
@@ -38,6 +41,8 @@ def _get_gui_llm():
_gui_llm = None _gui_llm = None
return _gui_llm if _gui_llm is not None else llm return _gui_llm if _gui_llm is not None else llm
# ---------- 类型定义 ----------
GUIAction = Literal[ GUIAction = Literal[
"click", "click",
"type", "type",
@@ -54,47 +59,41 @@ GUITargetType = Literal["auto", "button", "input", "checkbox", "list_item", "lin
class GUIActionItem(BaseModel): class GUIActionItem(BaseModel):
"""A single GUI action within an ordered sequence.""" """单个 GUI 动作。"""
action: GUIAction = Field( action: GUIAction = Field(..., description="GUI 操作类型。")
...,
description="GUI action to execute.",
)
element_description: Optional[str] = Field( element_description: Optional[str] = Field(
default=None, default=None,
description="Natural-language description of the target screen element for grounded actions (click, type, scroll, drag_and_drop, move_to). For type action, providing this will first locate and click the target input field before typing.", description="目标元素的自然语言描述。对于 type 操作,提供此项会先定位并点击目标输入框再输入。",
) )
ending_description: Optional[str] = Field( ending_description: Optional[str] = Field(
default=None, default=None,
description="Natural-language description of the drag end target.", description="拖拽终点的自然语言描述(仅 drag_and_drop 使用)。",
) )
text: str = Field(default="", description="Text to type.") text: str = Field(default="", description="要输入的文本。")
overwrite: bool = Field(default=False, description="Select all existing text before typing.") overwrite: bool = Field(default=False, description="输入前全选已有文本。")
enter: bool = Field(default=False, description="Press Enter after typing.") enter: bool = Field(default=False, description="输入后按回车。")
num_clicks: int = Field(default=1, description="Number of mouse clicks.") num_clicks: int = Field(default=1, description="鼠标点击次数。")
button_type: Literal["left", "middle", "right"] = Field(default="left", description="Mouse button.") button_type: Literal["left", "middle", "right"] = Field(default="left", description="鼠标按键。")
hold_keys: list[str] = Field(default_factory=list, description="Modifier keys to hold while clicking or dragging.") hold_keys: list[str] = Field(default_factory=list, description="点击或拖拽时按住的功能键。")
keys: list[str] = Field(default_factory=list, description="Hotkey keys or keys to press.") keys: list[str] = Field(default_factory=list, description="快捷键按键 / 敲击按键。")
press_keys: list[str] = Field(default_factory=list, description="Keys to press while holding hold_keys.") press_keys: list[str] = Field(default_factory=list, description="按住 hold_keys 时敲击的按键。")
clicks: int = Field(default=0, description="Scroll click amount. Negative means down.") clicks: int = Field(default=0, description="滚轮量,负数为向下。")
shift: bool = Field(default=False, description="Use horizontal scrolling when true.") shift: bool = Field(default=False, description="水平滚动(True 时为横向)。")
wait_seconds: float = Field(default=0.5, description="Seconds to wait.") wait_seconds: float = Field(default=0.5, description="等待秒数。")
target_type: GUITargetType = Field( target_type: GUITargetType = Field(
default="auto", default="auto",
description=( description="目标 UI 元素类型:checkbox / input / list_item / button / link / auto。",
"Target UI element type for grounded actions. Use checkbox for agreement/privacy checkboxes, "
"input for editable text boxes, list_item for chat/contact/file rows, button for buttons, "
"link for clickable text links, or auto when unsure."
),
) )
class GUIToolInput(BaseModel): class GUIToolInput(BaseModel):
actions: list[GUIActionItem] = Field( actions: list[GUIActionItem] = Field(
default_factory=list, default_factory=list,
description="Ordered list of GUI actions to execute sequentially. Each item specifies one GUI operation (click, type, drag_and_drop, scroll, hotkey, hold_and_press, wait, screenshot, move_to). If empty, returns current screenshot without performing any action.", description="有序 GUI 操作列表,按顺序依次执行。空列表则仅返回当前截图。",
) )
# ---------- 图片工具函数 ----------
def _get_gui_region() -> dict | None: def _get_gui_region() -> dict | None:
"""获取当前 GUI 操作显示器区域,失败返回 None(全屏)。""" """获取当前 GUI 操作显示器区域,失败返回 None(全屏)。"""
try: try:
@@ -104,10 +103,12 @@ def _get_gui_region() -> dict | None:
return None return None
def _screenshot_png() -> tuple[bytes, tuple[int, int]]: def _screenshot_png() -> Tuple[bytes, Tuple[int, int]]:
region = _get_gui_region() region = _get_gui_region()
if region: if region:
image = pyautogui.screenshot(region=(region["x"], region["y"], region["width"], region["height"])) image = pyautogui.screenshot(
region=(region["x"], region["y"], region["width"], region["height"])
)
else: else:
image = pyautogui.screenshot() image = pyautogui.screenshot()
width, height = image.size width, height = image.size
@@ -118,7 +119,10 @@ def _screenshot_png() -> tuple[bytes, tuple[int, int]]:
def _resize_for_grounding(image_bytes: bytes) -> bytes: def _resize_for_grounding(image_bytes: bytes) -> bytes:
image = Image.open(io.BytesIO(image_bytes)).convert("RGB") image = Image.open(io.BytesIO(image_bytes)).convert("RGB")
image = image.resize((env_utils.GROUNDING_WIDTH, env_utils.GROUNDING_HEIGHT), Image.Resampling.LANCZOS) image = image.resize(
(env_utils.GROUNDING_WIDTH, env_utils.GROUNDING_HEIGHT),
Image.Resampling.LANCZOS,
)
buf = io.BytesIO() buf = io.BytesIO()
image.save(buf, format="PNG") image.save(buf, format="PNG")
return buf.getvalue() return buf.getvalue()
@@ -129,112 +133,50 @@ def _image_data_url(image_bytes: bytes) -> str:
return f"data:image/png;base64,{encoded}" return f"data:image/png;base64,{encoded}"
def _parse_coordinates(text: str) -> tuple[int, int]: # ---------- 坐标解析与转换 ----------
numbers = re.findall(r"-?\d+(?:\.\d+)?", text or "") def _parse_batch_coordinates(text: str) -> list[Tuple[int, int]]:
if len(numbers) < 2: """解析批量坐标输出。
raise ValueError(f"grounding model did not return coordinates: {text!r}")
x = int(float(numbers[0])) 按行提取坐标对,每行取第一对有效的 (x, y)。
y = int(float(numbers[1])) 兼容格式:
x = max(0, min(env_utils.GROUNDING_WIDTH - 1, x)) - "(50, 960)"
y = max(0, min(env_utils.GROUNDING_HEIGHT - 1, y)) - "(100,200)"
return x, y - "1, (50, 960)" (带编号前缀)
- "1:(100,200)" (带编号前缀)
所有坐标会被裁剪到 [0, GROUNDING-1] × [0, GROUNDING-1]。
"""
W, H = env_utils.GROUNDING_WIDTH - 1, env_utils.GROUNDING_HEIGHT - 1
coordinates: list[Tuple[int, int]] = []
for line in text.strip().split("\n"):
line = line.strip()
if not line:
continue
matches = re.findall(r"\(\s*(-?\d+(?:\.\d+)?)\s*[,,]\s*(-?\d+(?:\.\d+)?)\s*\)", line)
if matches:
x = int(float(matches[0][0]))
y = int(float(matches[0][1]))
x = max(0, min(W, x))
y = max(0, min(H, y))
coordinates.append((x, y))
return coordinates
def _scale_to_screen(coords: tuple[int, int], screen_size: tuple[int, int]) -> tuple[int, int]: def _scale_to_screen(coords: Tuple[int, int], screen_size: Tuple[int, int]) -> Tuple[int, int]:
"""将 grounding 坐标映射到绝对屏幕坐标(考虑 GUI 操作区域偏移)。""" """将 grounding 坐标映射到绝对屏幕坐标(考虑 GUI 操作区域偏移)。"""
screen_width, screen_height = screen_size screen_width, screen_height = screen_size
x, y = coords x, y = coords
local_x = round(x * screen_width / env_utils.GROUNDING_WIDTH) local_x = round(x * screen_width / env_utils.GROUNDING_WIDTH)
local_y = round(y * screen_height / env_utils.GROUNDING_HEIGHT) local_y = round(y * screen_height / env_utils.GROUNDING_HEIGHT)
# 加上目标显示器在屏幕空间的偏移
region = _get_gui_region() region = _get_gui_region()
offset_x = region["x"] if region else 0 offset_x = region["x"] if region else 0
offset_y = region["y"] if region else 0 offset_y = region["y"] if region else 0
return (offset_x + local_x, offset_y + local_y) return (offset_x + local_x, offset_y + local_y)
def _target_type_instructions(target_type: str) -> str: # ---------- 文本输入辅助 ----------
normalized = (target_type or "auto").strip().lower()
instructions = {
"checkbox": (
"Target type: checkbox. Return the checkbox/radio circle or square itself, "
"and provide its center point. Do not click nearby agreement/privacy/service link text."
),
"input": (
"Target type: input. Return the editable text box area and provide a point "
"inside the editable area where typing will focus the field."
),
"list_item": (
"Target type: list_item. Return the full row/list item including icon/avatar "
"and label, and provide the row center point."
),
"button": (
"Target type: button. Return the full clickable button and provide its center point."
),
"link": (
"Target type: link. Return only the clickable text/link area and provide the text center."
),
}
return instructions.get(normalized, "Target type: auto. Infer the UI element type from the query.")
def _infer_target_type(element_description: str, target_type: str) -> str:
normalized = (target_type or "auto").strip().lower()
if normalized != "auto":
return normalized
desc = element_description or ""
checkbox_terms = ["复选框", "勾选框", "勾选", "已阅读", "同意", "协议", "隐私", "服务协议", "用户协议", "checkbox"]
if any(term in desc for term in checkbox_terms):
return "checkbox"
input_terms = ["输入框", "文本框", "消息框", "编辑框", "输入栏", "input", "textbox"]
if any(term in desc for term in input_terms):
return "input"
list_terms = ["列表", "联系人", "聊天列表", "文件列表", "这一行", "整行", "list"]
if any(term in desc for term in list_terms):
return "list_item"
button_terms = ["按钮", "登录", "确定", "取消", "发送", "button"]
if any(term in desc for term in button_terms):
return "button"
return "auto"
def _ground_element(element_description: str, target_type: str = "auto") -> tuple[int, int, str]:
if not element_description or not element_description.strip():
raise ValueError("element_description is required for this GUI action.")
target_type = _infer_target_type(element_description, target_type)
raw_png, screen_size = _screenshot_png()
grounding_png = _resize_for_grounding(raw_png)
prompt = (
"You are a visual grounding model for desktop GUI control.\n"
f"The screenshot has been resized to exactly {env_utils.GROUNDING_WIDTH}x{env_utils.GROUNDING_HEIGHT} pixels.\n"
f"Query: {element_description.strip()}\n"
f"{_target_type_instructions(target_type)}\n"
"Return only one target point as pixel coordinates in this exact format: x,y\n"
f"Example: {env_utils.GROUNDING_WIDTH // 2},{env_utils.GROUNDING_HEIGHT // 2}\n"
"Do not include any explanation."
)
response = _get_gui_llm().invoke(
[
HumanMessage(
content=[
{"type": "image_url", "image_url": {"url": _image_data_url(grounding_png)}},
{"type": "text", "text": prompt},
]
)
]
)
response_text = str(response.content)
gx, gy = _parse_coordinates(response_text)
x, y = _scale_to_screen((gx, gy), screen_size)
return x, y, f"grounded ({gx},{gy})"
def _type_text(text: str) -> None: def _type_text(text: str) -> None:
if not text: if not text:
return return
@@ -242,7 +184,6 @@ def _type_text(text: str) -> None:
if has_unicode: if has_unicode:
try: try:
import pyperclip import pyperclip
pyperclip.copy(text) pyperclip.copy(text)
modifier = "command" if sys.platform == "darwin" else "ctrl" modifier = "command" if sys.platform == "darwin" else "ctrl"
pyautogui.hotkey(modifier, "v") pyautogui.hotkey(modifier, "v")
@@ -268,7 +209,16 @@ def _with_hold_keys_up(keys: list[str]) -> None:
pyautogui.keyUp(key) pyautogui.keyUp(key)
# =====================================================================
# 核心工具类
# =====================================================================
class GUITool(BaseTool): class GUITool(BaseTool):
"""GUI 自动化工具。
坐标生成阶段采用批量方案:一次截图 + 全部 query 批量输入 LLM + 一次性解析所有坐标。
N 个需要定位的动作 = 1 次 LLM 调用。
"""
args_schema: type[BaseModel] = GUIToolInput args_schema: type[BaseModel] = GUIToolInput
name: str = "gui_tool" name: str = "gui_tool"
description: str = ( description: str = (
@@ -280,6 +230,18 @@ class GUITool(BaseTool):
"会先视觉定位并点击目标输入框,然后再输入文本,无需单独调用 click。不提供 element_description 则直接在当前焦点输入。\n" "会先视觉定位并点击目标输入框,然后再输入文本,无需单独调用 click。不提供 element_description 则直接在当前焦点输入。\n"
"所有动作执行完毕后,返回当前屏幕的截图。\n" "所有动作执行完毕后,返回当前屏幕的截图。\n"
"\n" "\n"
"【网页滚动提示】浏览器中向下滚动网页推荐使用快捷键,而非 scroll 滚轮:"
"{\"action\":\"hotkey\",\"keys\":[\"space\"]}(空格键即 Page Down)。"
"向上滚动可使用 {\"action\":\"hotkey\",\"keys\":[\"shift\",\"space\"]}。\n"
"\n"
"【下拉框处理策略】操作下拉选择框时分两步:\n"
" 第1步(type 方式):先尝试 {\"action\":\"type\",\"element_description\":\"目标下拉框\",\"text\":\"选项名称\"} 直接输入,\n"
" 能自动匹配则任务完成;\n"
" 第2步(键选方式):若 type 不生效,则改为先点击展开下拉框,再用方向键选择,最后回车确认:\n"
" {\"action\":\"click\",\"element_description\":\"目标下拉框\"},\n"
" {\"action\":\"hotkey\",\"keys\":[\"down\"]}(按需多次 up/down 定位),\n"
" {\"action\":\"hotkey\",\"keys\":[\"enter\"]}。\n"
"\n"
"示例:\n" "示例:\n"
"- 登录表单:{\"actions\": [" "- 登录表单:{\"actions\": ["
"{\"action\":\"type\",\"element_description\":\"用户名输入框\",\"text\":\"admin\"}," "{\"action\":\"type\",\"element_description\":\"用户名输入框\",\"text\":\"admin\"},"
@@ -288,6 +250,11 @@ class GUITool(BaseTool):
"- 浏览器搜索:{\"actions\": [" "- 浏览器搜索:{\"actions\": ["
"{\"action\":\"type\",\"element_description\":\"浏览器地址栏\",\"text\":\"github.com\",\"enter\":true}" "{\"action\":\"type\",\"element_description\":\"浏览器地址栏\",\"text\":\"github.com\",\"enter\":true}"
"]}\n" "]}\n"
"- 网页翻页浏览:{\"actions\": ["
"{\"action\":\"hotkey\",\"keys\":[\"space\"]},"
"{\"action\":\"wait\",\"wait_seconds\":1.0},"
"{\"action\":\"hotkey\",\"keys\":[\"space\"]}"
"]}\n"
"- 等待并确认:{\"actions\": [" "- 等待并确认:{\"actions\": ["
"{\"action\":\"click\",\"element_description\":\"提交按钮\"}," "{\"action\":\"click\",\"element_description\":\"提交按钮\"},"
"{\"action\":\"wait\",\"wait_seconds\":2.0}" "{\"action\":\"wait\",\"wait_seconds\":2.0}"
@@ -295,6 +262,143 @@ class GUITool(BaseTool):
) )
response_format: Literal["content", "content_and_artifact"] = "content_and_artifact" response_format: Literal["content", "content_and_artifact"] = "content_and_artifact"
# ---------- 批量坐标生成核心 ----------
def _collect_grounding_queries(self, actions: list[GUIActionItem]) -> list[Tuple[int, str, str]]:
"""遍历所有动作,收集需要视觉定位的 query。
返回 [(action_index, query_type, query_text), ...]
query_type: "element" (元素描述) 或 "ending" (拖拽终点)。
对于 type 操作且无 element_description 的,不收集(无需定位)。
"""
queries: list[Tuple[int, str, str]] = []
for idx, act in enumerate(actions):
action = str(act.action).strip()
if action in ("screenshot", "wait", "hotkey", "hold_and_press"):
continue
if action in ("click", "scroll", "move_to"):
desc = (act.element_description or "").strip()
if desc:
queries.append((idx, "element", desc))
continue
if action == "type":
desc = (act.element_description or "").strip()
if desc:
queries.append((idx, "element", desc))
continue
if action == "drag_and_drop":
start_desc = (act.element_description or "").strip()
end_desc = (act.ending_description or "").strip()
if start_desc:
queries.append((idx, "element", start_desc))
if end_desc:
queries.append((idx, "ending", end_desc))
continue
return queries
def _build_batch_prompt(self, queries: list[Tuple[int, str, str]]) -> str:
"""构建包含 One-shot 示例的批量定位 prompt。"""
W = env_utils.GROUNDING_WIDTH
H = env_utils.GROUNDING_HEIGHT
numbered = []
for i, (act_idx, qtype, qtext) in enumerate(queries, 1):
label = f"action{act_idx + 1}_{qtype}"
numbered.append(f"{i}. [{label}] {qtext}")
query_text = "\n".join(numbered)
prompt = (
"You are a visual grounding model for desktop GUI control.\n"
f"The screenshot has been resized to exactly {W}x{H} pixels.\n"
"Your task is to locate MULTIPLE target elements in the same image in a single response.\n"
"\n"
"=== OUTPUT FORMAT (STRICT) ===\n"
"You MUST output one line per query in the EXACT format below:\n"
" (<x>, <y>)\n"
"Where:\n"
" - x: horizontal pixel coordinate (integer, 0 = leftmost)\n"
" - y: vertical pixel coordinate (integer, 0 = topmost)\n"
"\n"
"=== ONE-SHOT EXAMPLE ===\n"
"Scenario: A desktop screenshot ({W}x{H}) showing a browser with multiple UI elements.\n"
"\n"
"Queries:\n"
"1. [action1_element] the browser address bar at the top of the window\n"
"2. [action2_element] the search input field in the main content area\n"
"3. [action3_element] the search button next to the input field\n"
"4. [action3_ending] the first search result link after clicking search\n"
"\n"
"Expected output (four lines, one per query):\n"
f"({W // 2}, {H // 6})\n"
f"({W // 2}, {H // 3})\n"
f"({W // 2 + 200}, {H // 3})\n"
f"({W // 2}, {H // 2 + 50})\n"
"\n"
"=== NOW YOUR TURN ===\n"
"Locate the following elements in the provided screenshot:\n"
f"{query_text}\n"
"\n"
"=== RULES (FAILURE TO FOLLOW = INCORRECT) ===\n"
"1. Output exactly one line per query, in the SAME order (1, 2, 3, ...).\n"
"2. Each line format: `(<integer_x>, <integer_y>)` — nothing else on that line.\n"
"3. NO explanations, NO markdown, NO additional text before or after the coordinate lines.\n"
"4. Coordinates must be integers in range x=[0,{W-1}] y=[0,{H-1}].\n"
"5. If you genuinely cannot find an element, output your best reasonable guess for its coordinates.\n"
"6. Do NOT wrap the output in code fences (```), quotes, or JSON.\n"
"7. For a sequence of operations on the same page, generate coordinates strictly in action order. "
"Each coordinate pair corresponds to one step of the GUI operation sequence. "
"(对于一个页面中的操作需要按顺序进行单步多GUI操作)\n"
)
return prompt
def _batch_ground_elements(
self, queries: list[Tuple[int, str, str]], screen_size: Tuple[int, int]
) -> dict[Tuple[int, str], Tuple[int, int]]:
"""一次截图 + 一次 LLM 调用,批量生成所有目标元素的坐标。
参数:
queries: [(action_index, query_type, query_text), ...]
screen_size: (screen_width, screen_height)
返回:
{(action_index, query_type): (screen_x, screen_y), ...}
"""
if not queries:
return {}
# 1. 截图
raw_png, _ = _screenshot_png()
grounding_png = _resize_for_grounding(raw_png)
# 2. 构建 prompt + 单次 LLM 调用
prompt = self._build_batch_prompt(queries)
response = _get_gui_llm().invoke(
[
HumanMessage(
content=[
{"type": "image_url", "image_url": {"url": _image_data_url(grounding_png)}},
{"type": "text", "text": prompt},
]
)
]
)
response_text = str(response.content).strip()
# 3. 解析坐标
coords = _parse_batch_coordinates(response_text)
# 4. 建立映射 (action_index, query_type) → (screen_x, screen_y)
result: dict[Tuple[int, str], Tuple[int, int]] = {}
for i, (act_idx, qtype, _) in enumerate(queries):
if i < len(coords):
result[(act_idx, qtype)] = _scale_to_screen(coords[i], screen_size)
return result
# ---------- 截图 artifact ----------
def _artifact_screenshot(self) -> dict: def _artifact_screenshot(self) -> dict:
try: try:
image_bytes, _ = _screenshot_png() image_bytes, _ = _screenshot_png()
@@ -303,24 +407,35 @@ class GUITool(BaseTool):
except Exception: except Exception:
return {} return {}
def _run( # ---------- 顶层入口 ----------
self, def _run(self, actions: list[GUIActionItem] | None = None) -> Tuple[str, dict]:
actions: list[GUIActionItem] | None = None,
) -> tuple[str, dict]:
try: try:
# ---------- 无动作时直接截图 ----------
if not actions: if not actions:
return ("", self._artifact_screenshot()) return ("", self._artifact_screenshot())
# ---------- 顺序执行动作列表 ---------- # 1. 先截图一次,获取屏幕尺寸
result_messages: list[str] = [] # 收集每个动作的返回信息 _, screen_size = _screenshot_png()
# 2. 批量定位:一次 LLM 调用生成所有动作需要的坐标
queries = self._collect_grounding_queries(actions)
coords_map = self._batch_ground_elements(queries, screen_size)
# 3. 构建 (action_index, query_type) → (x, y) 查找表
action_coords: dict[int, dict[str, Tuple[int, int]]] = {}
for (act_idx, qtype), coord in coords_map.items():
action_coords.setdefault(act_idx, {})[qtype] = coord
# 4. 顺序执行动作列表
result_messages: list[str] = []
for idx, act in enumerate(actions, start=1): for idx, act in enumerate(actions, start=1):
if act.action == "screenshot": if act.action == "screenshot":
result_messages.append(f"{idx}. Screenshot (skipped, final screenshot will be taken)") result_messages.append(f"{idx}. Screenshot (skipped, final screenshot will be taken)")
continue continue
act_idx = idx - 1 # 0-indexed
cached = action_coords.get(act_idx, {})
try: try:
msg = self._dispatch(**act.model_dump()) msg = self._dispatch(cached_coords=cached, **act.model_dump())
result_messages.append(f"{idx}. {msg}") result_messages.append(f"{idx}. {msg}")
except Exception as exc: except Exception as exc:
error_msg = f"{idx}. [GUI Error] {act.action}: {exc}" error_msg = f"{idx}. [GUI Error] {act.action}: {exc}"
@@ -332,18 +447,19 @@ class GUITool(BaseTool):
final_text = "\n".join(result_messages) final_text = "\n".join(result_messages)
return (final_text, self._artifact_screenshot()) return (final_text, self._artifact_screenshot())
except Exception as e: except Exception as e:
return (f"[ERROR] gui_tool 工具执行出错: {str(e)}", {}) return (f"[ERROR] gui_tool 执行出错: {str(e)}", {})
def _dispatch(self, **kwargs: Any) -> str: def _dispatch(self, cached_coords: dict[str, Tuple[int, int]], **kwargs: Any) -> str:
"""分发到具体动作方法,传入已缓存的坐标(cached_coords = {"element": (x,y), "ending": (x,y)})。"""
action = str(kwargs["action"]).strip() action = str(kwargs["action"]).strip()
if action == "click": if action == "click":
return self._click(**kwargs) return self._click(cached_coords, **kwargs)
if action == "type": if action == "type":
return self._type(**kwargs) return self._type(cached_coords, **kwargs)
if action == "drag_and_drop": if action == "drag_and_drop":
return self._drag_and_drop(**kwargs) return self._drag_and_drop(cached_coords, **kwargs)
if action == "scroll": if action == "scroll":
return self._scroll(**kwargs) return self._scroll(cached_coords, **kwargs)
if action == "hotkey": if action == "hotkey":
return self._hotkey(**kwargs) return self._hotkey(**kwargs)
if action == "hold_and_press": if action == "hold_and_press":
@@ -353,34 +469,35 @@ class GUITool(BaseTool):
time.sleep(seconds) time.sleep(seconds)
return f"Waited {seconds:.2f} seconds." return f"Waited {seconds:.2f} seconds."
if action == "move_to": if action == "move_to":
return self._move_to(**kwargs) return self._move_to(cached_coords, **kwargs)
raise ValueError(f"unsupported GUI action: {action}") raise ValueError(f"unsupported GUI action: {action}")
def _click(self, **kwargs: Any) -> str: # ---------- 各动作实现 ----------
x, y, grounding = _ground_element(kwargs["element_description"] or "", kwargs.get("target_type", "auto")) def _click(self, cached_coords: dict[str, Tuple[int, int]], **kwargs: Any) -> str:
desc = (kwargs.get("element_description") or "").strip()
if desc and "element" in cached_coords:
x, y = cached_coords["element"]
else:
x, y = pyautogui.position()
hold_keys = kwargs["hold_keys"] hold_keys = kwargs["hold_keys"]
_with_hold_keys_down(hold_keys) _with_hold_keys_down(hold_keys)
try: try:
pyautogui.click(x, y, clicks=max(1, int(kwargs["num_clicks"])), button=kwargs["button_type"]) pyautogui.click(x, y, clicks=max(1, int(kwargs["num_clicks"])), button=kwargs["button_type"])
finally: finally:
_with_hold_keys_up(hold_keys) _with_hold_keys_up(hold_keys)
return f"Clicked target. {grounding}" return f"Clicked target at ({x},{y})."
def _type(self, **kwargs: Any) -> str: def _type(self, cached_coords: dict[str, Tuple[int, int]], **kwargs: Any) -> str:
element_desc = (kwargs.get("element_description") or "").strip() element_desc = (kwargs.get("element_description") or "").strip()
if element_desc: if element_desc and "element" in cached_coords:
# 提供了目标元素描述:先定位点击再输入 x, y = cached_coords["element"]
target_type = kwargs.get("target_type", "auto")
if target_type == "auto":
target_type = "input" # type 动作定位输入框时强制为 input
x, y, grounding = _ground_element(element_desc, target_type)
pyautogui.click(x, y) pyautogui.click(x, y)
if kwargs["overwrite"]: if kwargs["overwrite"]:
_select_all_and_clear() _select_all_and_clear()
_type_text(kwargs["text"]) _type_text(kwargs["text"])
if kwargs["enter"]: if kwargs["enter"]:
pyautogui.press("enter") pyautogui.press("enter")
return f"Clicked target and typed text. {grounding}" return f"Clicked target and typed text at ({x},{y})."
# 不提供 element_description:直接在当前焦点输入 # 不提供 element_description:直接在当前焦点输入
if kwargs["overwrite"]: if kwargs["overwrite"]:
_select_all_and_clear() _select_all_and_clear()
@@ -389,9 +506,9 @@ class GUITool(BaseTool):
pyautogui.press("enter") pyautogui.press("enter")
return "Typed text." return "Typed text."
def _drag_and_drop(self, **kwargs: Any) -> str: def _drag_and_drop(self, cached_coords: dict[str, Tuple[int, int]], **kwargs: Any) -> str:
x1, y1, start_grounding = _ground_element(kwargs["element_description"] or "", kwargs.get("target_type", "auto")) x1, y1 = cached_coords.get("element", pyautogui.position())
x2, y2, end_grounding = _ground_element(kwargs["ending_description"] or "", kwargs.get("target_type", "auto")) x2, y2 = cached_coords.get("ending", pyautogui.position())
hold_keys = kwargs["hold_keys"] hold_keys = kwargs["hold_keys"]
pyautogui.moveTo(x1, y1) pyautogui.moveTo(x1, y1)
_with_hold_keys_down(hold_keys) _with_hold_keys_down(hold_keys)
@@ -400,15 +517,15 @@ class GUITool(BaseTool):
pyautogui.mouseUp() pyautogui.mouseUp()
finally: finally:
_with_hold_keys_up(hold_keys) _with_hold_keys_up(hold_keys)
return f"Dragged target. Start: {start_grounding}. End: {end_grounding}" return f"Dragged from ({x1},{y1}) to ({x2},{y2})."
def _move_to(self, **kwargs: Any) -> str: def _move_to(self, cached_coords: dict[str, Tuple[int, int]], **kwargs: Any) -> str:
x, y, grounding = _ground_element(kwargs["element_description"] or "", kwargs.get("target_type", "auto")) x, y = cached_coords.get("element", pyautogui.position())
pyautogui.moveTo(x, y) pyautogui.moveTo(x, y)
return f"Moved mouse to target. {grounding}" return f"Moved mouse to ({x},{y})."
def _scroll(self, **kwargs: Any) -> str: def _scroll(self, cached_coords: dict[str, Tuple[int, int]], **kwargs: Any) -> str:
x, y, grounding = _ground_element(kwargs["element_description"] or "", kwargs.get("target_type", "auto")) x, y = cached_coords.get("element", pyautogui.position())
pyautogui.moveTo(x, y) pyautogui.moveTo(x, y)
time.sleep(0.1) time.sleep(0.1)
clicks = int(kwargs["clicks"]) clicks = int(kwargs["clicks"])
@@ -416,7 +533,7 @@ class GUITool(BaseTool):
pyautogui.hscroll(clicks) pyautogui.hscroll(clicks)
else: else:
pyautogui.vscroll(clicks) pyautogui.vscroll(clicks)
return f"Scrolled target by {clicks}. {grounding}" return f"Scrolled by {clicks} at ({x},{y})."
def _hotkey(self, **kwargs: Any) -> str: def _hotkey(self, **kwargs: Any) -> str:
keys = kwargs["keys"] keys = kwargs["keys"]
@@ -437,5 +554,5 @@ class GUITool(BaseTool):
_with_hold_keys_up(hold_keys) _with_hold_keys_up(hold_keys)
return f"Held {hold_keys} and pressed {press_keys}." return f"Held {hold_keys} and pressed {press_keys}."
async def _arun(self, **kwargs: Any) -> tuple[str, dict]: async def _arun(self, actions: list[GUIActionItem] | None = None) -> Tuple[str, dict]:
return await asyncio.to_thread(self._run, **kwargs) return await asyncio.to_thread(self._run, actions)
+9
View File
@@ -211,6 +211,15 @@ class BrowserControl(BaseTool):
# 无论操作是否成功,都尝试截屏(等待窗口切换) # 无论操作是否成功,都尝试截屏(等待窗口切换)
time.sleep(0.3) time.sleep(0.3)
try: try:
# 先尝试获取 GUI 操作区域,限定截屏范围
try:
from agent.core.config_manager import get_active_monitor_region
region = get_active_monitor_region()
except Exception:
region = None
if region:
pil_img = pyautogui.screenshot(region=(region["x"], region["y"], region["width"], region["height"]))
else:
pil_img = pyautogui.screenshot() pil_img = pyautogui.screenshot()
pil_img = pil_img.resize((env_utils.GROUNDING_WIDTH, env_utils.GROUNDING_HEIGHT), Image.Resampling.LANCZOS) pil_img = pil_img.resize((env_utils.GROUNDING_WIDTH, env_utils.GROUNDING_HEIGHT), Image.Resampling.LANCZOS)
buf = io.BytesIO() buf = io.BytesIO()