refactor: simplify tool outputs
This commit is contained in:
parent
599f331c35
commit
da3ea7426c
@ -2067,7 +2067,7 @@ class MainTerminal:
|
||||
"path": write_result.get("path", target_path),
|
||||
"char_count": char_count,
|
||||
"byte_size": byte_size,
|
||||
"message": f"网页内容已以纯文本保存到 {write_result.get('path', target_path)}，请使用终端命令查看（文件建议为 .txt）。"
|
||||
"message": f"网页内容已以纯文本保存到 {write_result.get('path', target_path)}，可用 read_file 的 search/extract 查看，必要时再用终端命令。"
|
||||
}
|
||||
|
||||
if isinstance(extract_result, dict) and extract_result.get("failed_results"):
|
||||
|
||||
@ -2,7 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Dict, List, Optional
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
|
||||
def format_read_file_result(result_data: Dict[str, Any]) -> str:
|
||||
@ -387,23 +387,37 @@ def _format_todo_create(result_data: Dict[str, Any]) -> str:
|
||||
return _format_failure("todo_create", result_data)
|
||||
todo = (result_data.get("todo_list") or {}).copy()
|
||||
overview = todo.get("overview") or "未命名任务"
|
||||
tasks = _summarize_todo_tasks(todo)
|
||||
return f"已创建 TODO：{overview}\n{tasks}"
|
||||
total = len(todo.get("tasks") or [])
|
||||
return f"已创建 TODO：{overview}（共 {total} 项）"
|
||||
|
||||
|
||||
def _format_todo_update_task(result_data: Dict[str, Any]) -> str:
|
||||
if not result_data.get("success"):
|
||||
return _format_failure("todo_update_task", result_data)
|
||||
message = result_data.get("message") or "任务状态已更新"
|
||||
tasks = _summarize_todo_tasks(result_data.get("todo_list"))
|
||||
return f"{message}\n{tasks}" if tasks else message
|
||||
todo = result_data.get("todo_list") or {}
|
||||
tasks = todo.get("tasks") or []
|
||||
total = len(tasks)
|
||||
done = sum(1 for t in tasks if t.get("status") == "done")
|
||||
progress_note = f"进度 {done}/{total}" if total else ""
|
||||
return f"{message}；{progress_note}".strip("；")
|
||||
|
||||
|
||||
def _format_todo_finish(result_data: Dict[str, Any]) -> str:
|
||||
if result_data.get("success"):
|
||||
message = result_data.get("message") or "待办列表已结束"
|
||||
tasks = _summarize_todo_tasks(result_data.get("todo_list"))
|
||||
return f"{message}\n{tasks}" if tasks else message
|
||||
todo = result_data.get("todo_list") or {}
|
||||
tasks = todo.get("tasks") or []
|
||||
overview = todo.get("overview") or "未命名任务"
|
||||
total = len(tasks)
|
||||
done = sum(1 for t in tasks if t.get("status") == "done")
|
||||
status_note = "正常结束" if done == total else "已结束"
|
||||
lines = [
|
||||
f"TODO 已结束（{status_note}），进度 {done}/{total}",
|
||||
f"概述：{overview}",
|
||||
]
|
||||
if tasks:
|
||||
lines.append(_summarize_todo_tasks(todo))
|
||||
return "\n".join(lines)
|
||||
if result_data.get("requires_confirmation"):
|
||||
remaining = result_data.get("remaining") or []
|
||||
remain_note = ", ".join(remaining) if remaining else "未知"
|
||||
@ -416,10 +430,22 @@ def _format_todo_finish_confirm(result_data: Dict[str, Any]) -> str:
|
||||
return _format_failure("todo_finish_confirm", result_data)
|
||||
message = result_data.get("message") or "已处理 TODO 完结确认"
|
||||
todo = result_data.get("todo_list") or {}
|
||||
if todo.get("forced_finish"):
|
||||
reason = todo.get("forced_reason") or "未提供原因"
|
||||
message = f"{message}（强制结束，原因：{reason}）"
|
||||
return message
|
||||
overview = todo.get("overview") or "未命名任务"
|
||||
tasks = todo.get("tasks") or []
|
||||
total = len(tasks)
|
||||
done = sum(1 for t in tasks if t.get("status") == "done")
|
||||
forced = todo.get("forced_finish")
|
||||
reason = todo.get("forced_reason")
|
||||
status_note = "强制结束" if forced else "已结束"
|
||||
lines = [
|
||||
f"{message}（{status_note}，进度 {done}/{total}）",
|
||||
f"概述：{overview}",
|
||||
]
|
||||
if forced and reason:
|
||||
lines.append(f"强制结束原因：{reason}")
|
||||
if tasks:
|
||||
lines.append(_summarize_todo_tasks(todo))
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def _format_update_memory(result_data: Dict[str, Any]) -> str:
|
||||
@ -505,6 +531,74 @@ def _summarize_output_block(output: Optional[str], truncated: Optional[bool]) ->
|
||||
return f"{meta}\n```\n{output}\n```"
|
||||
|
||||
|
||||
def _preview_text(text: str, limit: int) -> Tuple[str, bool]:
|
||||
"""返回截断预览及是否截断标记。"""
|
||||
if text is None:
|
||||
return "", False
|
||||
if len(text) <= limit:
|
||||
return text, False
|
||||
return text[:limit], True
|
||||
|
||||
|
||||
def _format_terminal_snapshot(result_data: Dict[str, Any]) -> str:
|
||||
if not result_data.get("success"):
|
||||
return _format_failure("terminal_snapshot", result_data)
|
||||
session = result_data.get("session") or "default"
|
||||
requested = result_data.get("line_limit")
|
||||
requested_note = f"{requested} 行" if requested is not None else "全部"
|
||||
actual = result_data.get("lines_returned")
|
||||
actual_note = f"{actual} 行" if actual is not None else "未知行数"
|
||||
truncated = result_data.get("truncated")
|
||||
trunc_note = "已截断" if truncated else "未截断"
|
||||
output = result_data.get("output") or ""
|
||||
header = f"会话 {session}：请求 {requested_note}，实际返回 {actual_note}（{trunc_note}）。"
|
||||
body = _summarize_output_block(output, truncated)
|
||||
return f"{header}\n{body}"
|
||||
|
||||
|
||||
def _format_extract_webpage(result_data: Dict[str, Any]) -> str:
|
||||
if not result_data.get("success"):
|
||||
return _format_failure("extract_webpage", result_data)
|
||||
url = result_data.get("url") or "目标网页"
|
||||
content = result_data.get("content") or ""
|
||||
length = len(content)
|
||||
preview, truncated = _preview_text(content, 800)
|
||||
header = f"提取完成：{url}，长度 {length} 字符。"
|
||||
if not content:
|
||||
return f"{header} 内容为空。"
|
||||
note = "（截断预览）" if truncated else "（未截断）"
|
||||
return "\n".join([f"{header}{note}", "```", preview, "```"])
|
||||
|
||||
|
||||
def _format_ocr_image(result_data: Dict[str, Any]) -> str:
|
||||
if not result_data.get("success"):
|
||||
return _format_failure("ocr_image", result_data)
|
||||
content = result_data.get("content") or ""
|
||||
length = len(content)
|
||||
preview, truncated = _preview_text(content, 800)
|
||||
note = "（截断预览）" if truncated else "（未截断）"
|
||||
header = f"OCR 完成，长度 {length} 字符{note}"
|
||||
if not content:
|
||||
return f"{header}；未返回可识别文本。"
|
||||
return "\n".join([header, "```", preview, "```"])
|
||||
|
||||
|
||||
def _format_trigger_easter_egg(result_data: Dict[str, Any]) -> str:
|
||||
if not result_data.get("success"):
|
||||
return _format_failure("trigger_easter_egg", result_data)
|
||||
effect = (result_data.get("effect") or "").lower()
|
||||
duration = result_data.get("duration_seconds") or result_data.get("duration")
|
||||
if effect == "flood":
|
||||
return "大水即将淹没屏幕！预计持续 45 秒，不过这期间你和用户还可以继续对话。"
|
||||
if effect == "snake":
|
||||
dur_text = f"{duration} 秒" if duration else "约 200 秒"
|
||||
return f"发光贪吃蛇来访，约 {dur_text} 后或吃满 20 个苹果离场；动画不挡操作。"
|
||||
message = result_data.get("message")
|
||||
if message:
|
||||
return message
|
||||
return f"已触发彩蛋：{effect or '未知效果'}。"
|
||||
|
||||
|
||||
def _format_command_result(label: str, result_data: Dict[str, Any]) -> str:
|
||||
command = result_data.get("command") or ""
|
||||
return_code = result_data.get("return_code")
|
||||
@ -556,11 +650,15 @@ TOOL_FORMATTERS = {
|
||||
"create_folder": _format_create_folder,
|
||||
"focus_file": _format_focus_file,
|
||||
"unfocus_file": _format_unfocus_file,
|
||||
"terminal_snapshot": _format_terminal_snapshot,
|
||||
"terminal_session": _format_terminal_session,
|
||||
"terminal_input": _format_terminal_input,
|
||||
"sleep": _format_sleep,
|
||||
"run_command": _format_run_command,
|
||||
"run_python": _format_run_python,
|
||||
"extract_webpage": _format_extract_webpage,
|
||||
"ocr_image": _format_ocr_image,
|
||||
"trigger_easter_egg": _format_trigger_easter_egg,
|
||||
"todo_create": _format_todo_create,
|
||||
"todo_update_task": _format_todo_update_task,
|
||||
"todo_finish": _format_todo_finish,
|
||||
|
||||
Loading…
Reference in New Issue
Block a user