JOJO
e4fdc82319
sub_agent/stats.py 的统计摘要（调用/读写/搜索/网页/指令次数、运行时长、交付目录）与 tool_result_formatter/agent_context.py 的子智能体相关格式化函数此前仍是硬编码中文，英文界面显示中文。新增 sub_agent_stats.* 9 个 key、扩展 fmt_agent2.* 30 个 key；补齐后引用完整性 0 缺失、0 占位符漂移，冒烟 6/6，中英双语自检通过。
72 lines
2.6 KiB
Python
72 lines
2.6 KiB
Python
"""子智能体统计摘要与消息构建工具。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from typing import Any, Dict, Optional
|
|
|
|
from modules.i18n import tr
|
|
|
|
|
|
class SubAgentStatsMixin:
|
|
"""提供子智能体任务耗时、统计摘要与结果消息组装能力。"""
|
|
|
|
@staticmethod
|
|
def _compute_elapsed_seconds(task: Dict) -> Optional[int]:
|
|
try:
|
|
created_at = float(task.get("created_at") or 0)
|
|
updated_at = float(task.get("updated_at") or time.time())
|
|
except (TypeError, ValueError):
|
|
return None
|
|
if created_at <= 0:
|
|
return None
|
|
return int(round(max(0.0, updated_at - created_at)))
|
|
|
|
@staticmethod
|
|
def _coerce_stat_int(value: Any) -> int:
|
|
try:
|
|
return max(0, int(value))
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
|
|
def _build_stats_summary(self, stats: Optional[Dict[str, Any]]) -> str:
|
|
if not isinstance(stats, dict):
|
|
stats = {}
|
|
api_calls = self._coerce_stat_int(stats.get("api_calls") or stats.get("turn_count"))
|
|
files_read = self._coerce_stat_int(stats.get("files_read"))
|
|
write_files = self._coerce_stat_int(stats.get("write_files"))
|
|
edit_files = self._coerce_stat_int(stats.get("edit_files"))
|
|
searches = self._coerce_stat_int(stats.get("searches"))
|
|
web_pages = self._coerce_stat_int(stats.get("web_pages"))
|
|
commands = self._coerce_stat_int(stats.get("commands"))
|
|
lines = [
|
|
tr("sub_agent_stats.api_calls", n=api_calls),
|
|
tr("sub_agent_stats.files_read", n=files_read),
|
|
tr("sub_agent_stats.write_files", n=write_files),
|
|
tr("sub_agent_stats.edit_files", n=edit_files),
|
|
tr("sub_agent_stats.searches", n=searches),
|
|
tr("sub_agent_stats.web_pages", n=web_pages),
|
|
tr("sub_agent_stats.commands", n=commands),
|
|
]
|
|
return "\n".join(lines)
|
|
|
|
def _compose_sub_agent_message(
|
|
self,
|
|
*,
|
|
prefix: str,
|
|
stats_summary: str,
|
|
summary: str,
|
|
deliverables_dir: Optional[str] = None,
|
|
duration_seconds: Optional[int] = None,
|
|
) -> str:
|
|
parts = [prefix]
|
|
if stats_summary:
|
|
parts.append(stats_summary)
|
|
if duration_seconds is not None:
|
|
parts.append(tr("sub_agent_stats.duration", n=duration_seconds))
|
|
if summary:
|
|
parts.append(summary)
|
|
if deliverables_dir:
|
|
parts.append(tr("sub_agent_stats.deliverables", dir=deliverables_dir))
|
|
return "\n\n".join(parts)
|