#!/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), "primary_source_alignment": _primary_source_alignment(), "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([ "", "## 主流實務來源證據", "", "| 實務 | 官方來源 | AWOOOI Gate | Agent 分工 |", "|---|---|---|---|", ]) for item in payload["primary_source_alignment"]: lines.append( f"| {item['practice']} | {item['source']} | `{item['awoooi_gate']}` | " f"{item['agent_assignment']} |" ) 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 _primary_source_alignment() -> list[dict[str, str]]: return [ { "practice": "OpenAI Agents SDK：專家協作、tool execution、approvals、state 由產品掌控", "source": "https://developers.openai.com/api/docs/guides/agents", "awoooi_gate": "sandbox_orchestration_no_write", "agent_assignment": "OpenClaw 負責 policy guard；MarketRadar 追版本；Hermes 產審核包。", }, { "practice": "NVIDIA Nemotron 3 Ultra / NeMo：長任務 Agent、profiling、evaluation、MCP / A2A 互通", "source": "https://developer.nvidia.com/blog/nvidia-nemotron-3-ultra-powers-faster-more-efficient-reasoning-for-long-running-agents/", "awoooi_gate": "nemotron_replay_evaluator_only", "agent_assignment": "NemoTron 只做離線 replay / evaluator / smoke gate，不接 production routing。", }, { "practice": "LangGraph：durable execution、human-in-the-loop、stateful workflow runtime", "source": "https://docs.langchain.com/oss/python/langgraph/overview", "awoooi_gate": "incident_workflow_kernel_replay_first", "agent_assignment": "OpenClaw 仲裁狀態轉移；Hermes 記錄 replay 證據與交接原因。", }, { "practice": "MCP：標準化 agent-to-tool / resource / prompt 連接，且需明確 user consent", "source": "https://modelcontextprotocol.io/specification/2025-06-18", "awoooi_gate": "read_only_tool_registry_before_write_adapter", "agent_assignment": "MarketRadar 監控 SDK / spec；Critic 檢查資料權限與 tool safety。", }, { "practice": "A2A：跨框架 Agent 溝通、委派與互通；MCP 處理工具、A2A 處理 Agent 對 Agent", "source": "https://a2a-protocol.org/latest/", "awoooi_gate": "agent_to_agent_interop_watch_only", "agent_assignment": "OpenClaw 設定協作邊界；Hermes 彙整 handoff 記錄；NemoTron 比對輸出。", }, { "practice": "OpenTelemetry GenAI：Agent / LLM / MCP trace 語意慣例，支援可觀測與稽核", "source": "https://opentelemetry.io/docs/specs/semconv/registry/attributes/gen-ai/", "awoooi_gate": "trace_semconv_mapping_before_runtime_export", "agent_assignment": "Critic 定義稽核欄位；MarketRadar 追語意規範版本；Hermes 產日週月報。", }, ] 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())