210577de28
Some checks failed
Code Review / ai-code-review (push) Successful in 13s
CD Pipeline / tests (push) Successful in 1m39s
CD Pipeline / build-and-deploy (push) Successful in 4m35s
CD Pipeline / post-deploy-checks (push) Successful in 1m51s
Ansible / Reboot Recovery Contract / validate (push) Has been cancelled
435 lines
19 KiB
Python
435 lines
19 KiB
Python
#!/usr/bin/env python3
|
||
"""Build the AWOOOI AI technology radar readback artifact."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
from collections import defaultdict
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
|
||
def build_radar(
|
||
*,
|
||
technology_watch: dict[str, Any],
|
||
agent_market_radar: dict[str, Any],
|
||
evidence_commit: str,
|
||
generated_at: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Build a read-only AI technology radar readback from committed evidence."""
|
||
_require_schema(technology_watch, "ai_technology_watch_report_v1", "technology_watch")
|
||
_require_schema(agent_market_radar, "ai_agent_market_radar_readback_v1", "agent_market_radar")
|
||
|
||
watch_summary = technology_watch.get("summary") or {}
|
||
agent_summary = agent_market_radar.get("summary") or {}
|
||
source_count = int(watch_summary.get("source_count", 0))
|
||
failure_count = int(watch_summary.get("source_failure_count", 0))
|
||
success_rate = _success_rate(source_count, failure_count)
|
||
|
||
return {
|
||
"schema_version": "ai_technology_radar_readback_v1",
|
||
"generated_at": generated_at or datetime.now(timezone.utc).isoformat(),
|
||
"source_scope": {
|
||
"technology_watch_report": "docs/evaluations/ai_technology_watch_report_2026-06-25.json",
|
||
"technology_source_registry": "docs/ai/ai-technology-watch-sources.v1.json",
|
||
"agent_market_radar_readback": "docs/operations/ai-agent-market-radar-readback.snapshot.json",
|
||
"gitea_main_evidence_basis_commit": evidence_commit,
|
||
"scope_note": "本讀回只整合已提交的只讀來源監控、AI Agent 市場雷達與治理 gate；不包含 raw chat history、secret、session 或本機工作視窗內容。",
|
||
},
|
||
"summary": {
|
||
"overall_completion_percent": float(agent_summary.get("overall_completion_percent", 42.2)),
|
||
"ai_technology_radar_completion_percent": success_rate,
|
||
"technology_count": int(watch_summary.get("technology_count", 0)),
|
||
"technology_area_count": int(watch_summary.get("technology_area_count", 0)),
|
||
"source_count": source_count,
|
||
"changed_technologies": int(watch_summary.get("changed_technologies", 0)),
|
||
"review_queue_count": int(watch_summary.get("review_queue_count", 0)),
|
||
"source_failures": failure_count,
|
||
"high_priority_count": int(watch_summary.get("high_priority_count", 0)),
|
||
"rolling_update_status": "near_real_time_watch_ready_integration_gated",
|
||
},
|
||
"policy": {
|
||
"read_only": True,
|
||
"raw_chat_history_synced": False,
|
||
"sdk_installation_approved": False,
|
||
"paid_api_calls_approved": False,
|
||
"production_routing_approved": False,
|
||
"telegram_send_approved": False,
|
||
"model_provider_switch_approved": False,
|
||
"host_write_approved": False,
|
||
"openclaw_replacement_approved": False,
|
||
},
|
||
"technology_area_counts": dict(technology_watch.get("technology_area_counts") or {}),
|
||
"technology_domains": _technology_domains(technology_watch),
|
||
"high_priority_review_queue": _high_priority_review_queue(technology_watch),
|
||
"professional_agent_roles": _professional_agent_roles(),
|
||
"rolling_update_controls": _rolling_update_controls(technology_watch),
|
||
"integration_candidates": _integration_candidates(technology_watch),
|
||
"priority_workplan": _priority_workplan(),
|
||
"blocked_gates": [
|
||
"sdk_installation_approved=false",
|
||
"paid_api_calls_approved=false",
|
||
"production_routing_approved=false",
|
||
"telegram_send_approved=false",
|
||
"model_provider_switch_approved=false",
|
||
"host_write_approved=false",
|
||
"openclaw_replacement_approved=false",
|
||
"replay_shadow_canary_gate_required=true",
|
||
"cost_and_data_boundary_review_required=true",
|
||
],
|
||
"report_contract": {
|
||
"api_endpoint": "/api/v1/agents/ai-technology-radar-readback",
|
||
"frontend_target": "/zh-TW/governance?tab=agent-market",
|
||
"schedule_workflow": ".gitea/workflows/ai-technology-watch.yaml",
|
||
"schedule_enabled": True,
|
||
"schedule_cron_utc": "0 */6 * * *",
|
||
"near_real_time": "每 6 小時讀取 primary sources，偵測主流 AI 技術版本、文件與 release 變更。",
|
||
"daily": "每日彙整變更、來源失敗、審核佇列與可自動處理項目。",
|
||
"weekly": "每週做技術 scorecard，決定 sandbox / replay / adapter design 優先級。",
|
||
"monthly": "每月進行策略 review，決定納入 roadmap、維持 watch-only 或移出監控。",
|
||
"human_review_required_for": [
|
||
"新 SDK / package / MCP server 安裝",
|
||
"付費 API 或 token 上限變更",
|
||
"模型 provider / 生產路由切換",
|
||
"Telegram Bot 即時發送或審批按鈕策略變更",
|
||
"主機、K8s、workflow、Nginx、secret 或資料層寫入",
|
||
"OpenClaw 生產決策核心替換、拆分或降級",
|
||
],
|
||
"agent_auto_allowed_for": [
|
||
"官方來源只讀監控",
|
||
"版本與文件 hash 比對",
|
||
"審核佇列分類",
|
||
"繁中 no-send 報告草稿",
|
||
"離線 replay fixture 準備",
|
||
"低風險文件與讀回 snapshot 更新提案",
|
||
],
|
||
},
|
||
}
|
||
|
||
|
||
def render_markdown(payload: dict[str, Any]) -> str:
|
||
"""Render a Traditional Chinese operator report."""
|
||
summary = payload["summary"]
|
||
lines = [
|
||
"# AI 技術雷達與滾動更新讀回",
|
||
"",
|
||
f"- 產生時間：`{payload['generated_at']}`",
|
||
f"- 整體治理完成度：`{summary['overall_completion_percent']}%`",
|
||
f"- AI 技術雷達來源成功率：`{summary['ai_technology_radar_completion_percent']}%`",
|
||
f"- 監控技術項目：`{summary['technology_count']}`",
|
||
f"- 技術領域：`{summary['technology_area_count']}`",
|
||
f"- 官方 / primary sources：`{summary['source_count']}`",
|
||
f"- 來源失敗：`{summary['source_failures']}`",
|
||
f"- 需要審核變更：`{summary['changed_technologies']}`",
|
||
f"- 高優先級項目：`{summary['high_priority_count']}`",
|
||
f"- 滾動更新狀態：`{summary['rolling_update_status']}`",
|
||
"",
|
||
"## 技術領域覆蓋",
|
||
"",
|
||
"| 技術領域 | 技術數 | 高優先級 | 需要審核 | 代表技術 |",
|
||
"|---|---:|---:|---:|---|",
|
||
]
|
||
for domain in payload["technology_domains"]:
|
||
lines.append(
|
||
f"| `{domain['technology_area']}` | `{domain['technology_count']}` | "
|
||
f"`{domain['high_priority_count']}` | `{domain['changed_count']}` | "
|
||
f"{', '.join(domain['representative_technologies'])} |"
|
||
)
|
||
|
||
lines.extend([
|
||
"",
|
||
"## 高優先級審核佇列",
|
||
"",
|
||
"| 技術 | 領域 | 優先級 | Gate | 下一步 |",
|
||
"|---|---|---|---|---|",
|
||
])
|
||
for item in payload["high_priority_review_queue"]:
|
||
lines.append(
|
||
f"| {item['display_name']} | `{item['technology_area']}` | "
|
||
f"`{item['evaluation_priority']}` | `{item['gate_status']}` | {item['next_gate']} |"
|
||
)
|
||
|
||
lines.extend([
|
||
"",
|
||
"## Agent 專業分工",
|
||
"",
|
||
"| Agent | 專業角色 | 自動化範圍 | 需要審核的邊界 |",
|
||
"|---|---|---|---|",
|
||
])
|
||
for role in payload["professional_agent_roles"]:
|
||
lines.append(
|
||
f"| {role['agent']} | {role['professional_role']} | "
|
||
f"{role['auto_scope']} | {role['review_boundary']} |"
|
||
)
|
||
|
||
lines.extend([
|
||
"",
|
||
"## 滾動更新控制",
|
||
"",
|
||
"| 節奏 | Agent 可自動做什麼 | 輸出 | Gate |",
|
||
"|---|---|---|---|",
|
||
])
|
||
for control in payload["rolling_update_controls"]:
|
||
lines.append(
|
||
f"| {control['cadence']} | {control['agent_auto_action']} | "
|
||
f"{control['output']} | `{control['gate']}` |"
|
||
)
|
||
|
||
lines.extend([
|
||
"",
|
||
"## 優先工作清單",
|
||
"",
|
||
"| 順序 | 工作 | 優先級 | 自動化模式 | 完成定義 |",
|
||
"|---:|---|---|---|---|",
|
||
])
|
||
for item in payload["priority_workplan"]:
|
||
lines.append(
|
||
f"| {item['order']} | {item['work_item']} | `{item['priority']}` | "
|
||
f"`{item['automation_mode']}` | {item['done_definition']} |"
|
||
)
|
||
|
||
lines.extend([
|
||
"",
|
||
"## 仍被 Gate 擋下",
|
||
"",
|
||
])
|
||
for gate in payload["blocked_gates"]:
|
||
lines.append(f"- `{gate}`")
|
||
lines.append("")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _require_schema(payload: dict[str, Any], schema_version: str, label: str) -> None:
|
||
if payload.get("schema_version") != schema_version:
|
||
raise ValueError(f"{label}: expected {schema_version}")
|
||
|
||
|
||
def _success_rate(source_count: int, failure_count: int) -> float:
|
||
if source_count <= 0:
|
||
return 0.0
|
||
return round(((source_count - failure_count) / source_count) * 100, 1)
|
||
|
||
|
||
def _technology_domains(report: dict[str, Any]) -> list[dict[str, Any]]:
|
||
grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
||
for row in report.get("technologies") or []:
|
||
grouped[str(row.get("technology_area") or "uncategorized")].append(row)
|
||
|
||
domains = []
|
||
for area, rows in sorted(grouped.items()):
|
||
high_priority = [row for row in rows if row.get("evaluation_priority") in {"p0", "p1"}]
|
||
changed = [row for row in rows if row.get("changed")]
|
||
domains.append({
|
||
"technology_area": area,
|
||
"technology_count": len(rows),
|
||
"high_priority_count": len(high_priority),
|
||
"changed_count": len(changed),
|
||
"representative_technologies": [
|
||
str(row.get("display_name") or row.get("technology_id"))
|
||
for row in rows[:4]
|
||
],
|
||
})
|
||
return domains
|
||
|
||
|
||
def _high_priority_review_queue(report: dict[str, Any]) -> list[dict[str, Any]]:
|
||
rows = [
|
||
row
|
||
for row in report.get("technologies") or []
|
||
if row.get("evaluation_priority") in {"p0", "p1"} and row.get("changed")
|
||
]
|
||
return [
|
||
{
|
||
"technology_id": row.get("technology_id"),
|
||
"display_name": row.get("display_name"),
|
||
"technology_area": row.get("technology_area"),
|
||
"evaluation_priority": row.get("evaluation_priority"),
|
||
"gate_status": "scorecard_required_before_integration",
|
||
"next_gate": "刷新 scorecard，若涉及 SDK/API/route/Telegram/host write 則送人工審核。",
|
||
"requires_cost_approval": bool(row.get("requires_cost_approval")),
|
||
"requires_dependency_approval": bool(row.get("requires_dependency_approval")),
|
||
"requires_security_review": bool(row.get("requires_security_review")),
|
||
}
|
||
for row in rows[:12]
|
||
]
|
||
|
||
|
||
def _professional_agent_roles() -> list[dict[str, Any]]:
|
||
return [
|
||
{
|
||
"agent": "OpenClaw",
|
||
"professional_role": "生產決策仲裁者、風險分級與最後 policy guard",
|
||
"auto_scope": "維持現有 production baseline、讀取 replay / shadow 評分、拒絕無證據替換",
|
||
"review_boundary": "任何取代、降級、生產路由切換都必須通過 replay / shadow / canary 與人工批准。",
|
||
},
|
||
{
|
||
"agent": "NemoTron",
|
||
"professional_role": "離線回放評估者、模型能力比較、合約輸出 smoke gate",
|
||
"auto_scope": "只讀 request pack、比對候選輸出、產生 replay scorecard 草稿",
|
||
"review_boundary": "不得自行呼叫外部 NIM/API、不得讀 labels 作答、不得進生產路由。",
|
||
},
|
||
{
|
||
"agent": "Hermes",
|
||
"professional_role": "知識管理、RAG 整理、報告草稿與長期技能庫維護",
|
||
"auto_scope": "整理 primary source 摘要、建立 no-send 日週月報、準備人審包",
|
||
"review_boundary": "不得同步 raw chat history、不得保存 secret、不得直接發 Telegram live report。",
|
||
},
|
||
{
|
||
"agent": "MarketRadar",
|
||
"professional_role": "AI 技術市場雷達、版本監控、來源失敗偵測",
|
||
"auto_scope": "每 6 小時只讀 primary sources、產生 freshness / review queue",
|
||
"review_boundary": "不得自動新增 SDK、不得自動修改 provider route 或 workflow 行為。",
|
||
},
|
||
{
|
||
"agent": "Critic / Reviewer",
|
||
"professional_role": "獨立審核、反例檢查、整合風險評分",
|
||
"auto_scope": "檢查政策旗標、來源可靠性、成本與資安風險",
|
||
"review_boundary": "只能輸出 blocked / candidate / owner_review，不得直接執行寫入。",
|
||
},
|
||
]
|
||
|
||
|
||
def _rolling_update_controls(report: dict[str, Any]) -> list[dict[str, Any]]:
|
||
cadence = report.get("cadence") or {}
|
||
return [
|
||
{
|
||
"cadence": "每 6 小時",
|
||
"agent_auto_action": "讀取官方文件、PyPI、npm、GitHub release、primary source hash。",
|
||
"output": "AI 技術 watch report、來源失敗清單、review queue。",
|
||
"gate": "read_only_only",
|
||
"cadence_source": cadence.get("near_real_time_watch", ""),
|
||
},
|
||
{
|
||
"cadence": "每日",
|
||
"agent_auto_action": "依 business applicability、成本、依賴、資安、AWOOOI fit 分類。",
|
||
"output": "日報摘要與中低風險自動處理建議。",
|
||
"gate": "no_send_report_until_delivery_gate",
|
||
"cadence_source": cadence.get("daily_triage", ""),
|
||
},
|
||
{
|
||
"cadence": "每週",
|
||
"agent_auto_action": "刷新 scorecard，決定 sandbox / replay / adapter design 優先級。",
|
||
"output": "週報、優先序、候選整合審查包。",
|
||
"gate": "scorecard_required_before_replay",
|
||
"cadence_source": cadence.get("weekly_scorecard", ""),
|
||
},
|
||
{
|
||
"cadence": "每月",
|
||
"agent_auto_action": "彙整趨勢，提出 roadmap / watch-only / retire 建議。",
|
||
"output": "月報與策略審核包。",
|
||
"gate": "human_review_for_strategy_or_production_change",
|
||
"cadence_source": cadence.get("monthly_strategy_review", ""),
|
||
},
|
||
]
|
||
|
||
|
||
def _integration_candidates(report: dict[str, Any]) -> list[dict[str, Any]]:
|
||
return [
|
||
{
|
||
"technology_id": row.get("technology_id"),
|
||
"display_name": row.get("display_name"),
|
||
"technology_area": row.get("technology_area"),
|
||
"integration_surface": row.get("integration_surface"),
|
||
"awoooi_role": row.get("awoooi_role"),
|
||
"changed": bool(row.get("changed")),
|
||
"decision": row.get("decision"),
|
||
"recommended_actions": row.get("recommended_actions") or [],
|
||
}
|
||
for row in report.get("technologies") or []
|
||
]
|
||
|
||
|
||
def _priority_workplan() -> list[dict[str, Any]]:
|
||
return [
|
||
{
|
||
"order": 1,
|
||
"priority": "P0",
|
||
"work_item": "AI 技術雷達 primary source 監控產品化",
|
||
"automation_mode": "agent_auto_read_only",
|
||
"done_definition": "API、snapshot、Markdown、schema、測試與 production readback 都能顯示技術領域、來源與 Gate。",
|
||
},
|
||
{
|
||
"order": 2,
|
||
"priority": "P0",
|
||
"work_item": "近即時版本 / release / docs 變更偵測",
|
||
"automation_mode": "agent_auto_schedule_read_only",
|
||
"done_definition": "每 6 小時可跑 watch；失敗來源會進日報，不會自動整合。",
|
||
},
|
||
{
|
||
"order": 3,
|
||
"priority": "P0",
|
||
"work_item": "OpenClaw / Hermes / NemoTron / MarketRadar 專業分工與成長紀錄",
|
||
"automation_mode": "agent_auto_read_model_human_review_for_write",
|
||
"done_definition": "每個 Agent 的角色、輸出、學習寫回與限制都能被前端讀回。",
|
||
},
|
||
{
|
||
"order": 4,
|
||
"priority": "P1",
|
||
"work_item": "AI 技術 scorecard 與 sandbox / replay 優先級",
|
||
"automation_mode": "agent_propose_owner_review",
|
||
"done_definition": "高優先級變更先進 scorecard，再進 no-cost/no-write sandbox 或 replay 計畫。",
|
||
},
|
||
{
|
||
"order": 5,
|
||
"priority": "P1",
|
||
"work_item": "Telegram Bot 報告與高風險審核橋接",
|
||
"automation_mode": "blocked_until_telegram_send_gate",
|
||
"done_definition": "低中風險只告警回報；高風險需 owner approval 後才可發送或執行。",
|
||
},
|
||
{
|
||
"order": 6,
|
||
"priority": "P2",
|
||
"work_item": "新 AI 技術探索與 watchlist 擴充",
|
||
"automation_mode": "agent_auto_discover_human_classify",
|
||
"done_definition": "GitHub topic / package registry / 官方 blog 可提出候選，但加入正式 watchlist 前需審核。",
|
||
},
|
||
]
|
||
|
||
|
||
def load_json(path: Path) -> dict[str, Any]:
|
||
with path.open(encoding="utf-8") as handle:
|
||
payload = json.load(handle)
|
||
if not isinstance(payload, dict):
|
||
raise ValueError(f"{path}: expected JSON object")
|
||
return payload
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="Build AI technology radar readback.")
|
||
parser.add_argument("--technology-watch", required=True)
|
||
parser.add_argument("--agent-market-radar", required=True)
|
||
parser.add_argument("--evidence-commit", required=True)
|
||
parser.add_argument("--output", required=True)
|
||
parser.add_argument("--markdown-output", required=True)
|
||
parser.add_argument("--markdown-alias-output")
|
||
args = parser.parse_args()
|
||
|
||
payload = build_radar(
|
||
technology_watch=load_json(Path(args.technology_watch)),
|
||
agent_market_radar=load_json(Path(args.agent_market_radar)),
|
||
evidence_commit=args.evidence_commit,
|
||
)
|
||
Path(args.output).write_text(
|
||
json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
|
||
encoding="utf-8",
|
||
)
|
||
markdown = render_markdown(payload)
|
||
Path(args.markdown_output).write_text(markdown, encoding="utf-8")
|
||
if args.markdown_alias_output:
|
||
Path(args.markdown_alias_output).write_text(markdown, encoding="utf-8")
|
||
print(
|
||
"AI_TECHNOLOGY_RADAR_READBACK_OK "
|
||
f"overall={payload['summary']['overall_completion_percent']}% "
|
||
f"technology={payload['summary']['technology_count']} "
|
||
f"sources={payload['summary']['source_count']} "
|
||
f"changed={payload['summary']['changed_technologies']} "
|
||
f"failures={payload['summary']['source_failures']}"
|
||
)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|