""" Agentic QE Fleet — Agent 加载和上下文注入模块 负责: 1. 加载 Agent prompt 文件 2. 注入战区上下文（前序 manifest 数据） 3. 渲染最终可执行的 Agent prompt """ from __future__ import annotations import re from pathlib import Path from typing import Any REPO_ROOT = Path(__file__).resolve().parent.parent AGENTS_DIR = REPO_ROOT / "agents" # Agent 注册表: agent_id → (战区, 文件路径, 描述) AGENT_REGISTRY: dict[str, dict[str, Any]] = { # ── Prepare ── "document-parser": { "zone": "prepare", "path": "prepare/document_parser.md", "name": "文档解析专家", "description": "解析原始需求文档为标准 Markdown，自动识别技术方案", }, "knowledge-activator": { "zone": "prepare", "path": "prepare/knowledge_activator.md", "name": "知识激活专家", "description": "按需求内容自动激活术语/规则/历史缺陷/最佳实践", }, # ── Analyze ── "requirement-analyzer": { "zone": "analyze", "path": "analyze/requirement_analyzer.md", "name": "需求分析专家", "description": "结构化需求模型 + 歧义标注", }, "conflict-detector": { "zone": "analyze", "path": "analyze/conflict_detector.md", "name": "冲突检测专家", "description": "语义级历史需求规则冲突检测", }, "risk-assessor": { "zone": "analyze", "path": "analyze/risk_assessor.md", "name": "风险评估专家", "description": "多维度风险矩阵量化", }, # ── Design ── "test-strategist": { "zone": "design", "path": "design/test_strategist.md", "name": "测试策略师", "description": "分层测试策略 + 优先级矩阵", }, "testpoint-designer": { "zone": "design", "path": "design/testpoint_designer.md", "name": "测试点设计师", "description": "全面测试点矩阵 + 来源标注", }, "case-designer": { "zone": "design", "path": "design/case_designer.md", "name": "用例设计师", "description": "可执行测试用例 + 双验证预期结果", }, "data-builder": { "zone": "design", "path": "design/data_builder.md", "name": "数据构造师", "description": "精确测试数据集构造", }, # ── Execute ── "web-executor": { "zone": "execute", "path": "execute/web_executor.md", "name": "Web 自动化执行师", "description": "PC Web 端 Playwright 自动化测试执行 + 截图采集", }, "mobile-executor": { "zone": "execute", "path": "execute/mobile_executor.md", "name": "移动端自动化执行师", "description": "Android/iOS APP Appium 自动化测试执行 + 截图采集", }, "result-reporter": { "zone": "execute", "path": "execute/result_reporter.md", "name": "测试结果报告师", "description": "截图对比 + AI 视觉验证 + 测试结论报告", }, # ── Review ── "case-reviewer": { "zone": "review", "path": "review/case_reviewer.md", "name": "用例评审师", "description": "用例质量/规范性/可执行性逐项评审", }, "coverage-auditor": { "zone": "review", "path": "review/coverage_auditor.md", "name": "覆盖率审计师", "description": "需求→测试点→用例三级追溯覆盖审计", }, "quality-gatekeeper": { "zone": "review", "path": "review/quality_gatekeeper.md", "name": "质量门禁裁决官", "description": "三级裁决: PASS / PASS_WITH_FIX / BLOCKED", }, # ── Monitor ── "execution-analyst": { "zone": "monitor", "path": "monitor/execution_analyst.md", "name": "执行结果分析师", "description": "失败归类 + 模式识别 + 根因推测", }, "knowledge-curator": { "zone": "monitor", "path": "monitor/knowledge_curator.md", "name": "知识沉淀师", "description": "自动回写知识库 + 去重保护", }, } def get_agent_path(agent_id: str) -> Path: """返回 Agent prompt 文件的绝对路径。""" if agent_id not in AGENT_REGISTRY: raise ValueError(f"未知 Agent: {agent_id}，可用: {list(AGENT_REGISTRY)}") return AGENTS_DIR / AGENT_REGISTRY[agent_id]["path"] def load_agent_prompt(agent_id: str) -> str: """加载 Agent prompt 原始内容。""" path = get_agent_path(agent_id) if not path.exists(): raise FileNotFoundError(f"Agent prompt 不存在: {path}") return path.read_text(encoding="utf-8") def list_zone_agents(zone: str) -> list[str]: """列出指定战区的所有 Agent ID。""" return sorted([ agent_id for agent_id, info in AGENT_REGISTRY.items() if info["zone"] == zone ]) def list_all_agents() -> dict[str, list[str]]: """按战区分组列出所有 Agent。""" result: dict[str, list[str]] = {} for zone in ["prepare", "analyze", "design", "execute", "review", "monitor"]: result[zone] = list_zone_agents(zone) return result def inject_context(prompt: str, context: dict[str, Any]) -> str: """向 prompt 注入运行时上下文变量。 支持的占位符: {{BASE_NAME}} → 需求基础名 {{MANIFEST_DIR}} → manifest 目录 {{PREPARE_MANIFEST}} → prepare manifest 路径 {{ANALYZE_MANIFEST}} → analyze manifest 路径 {{DESIGN_MANIFEST}} → design manifest 路径 {{PROJECT_PROFILE}} → 项目画像路径 {{FLEET_CONFIG}} → fleet_config.yml 路径 """ replacements = { "{{BASE_NAME}}": context.get("base_name", ""), "{{MANIFEST_DIR}}": str(REPO_ROOT / "output" / "manifests"), "{{PREPARE_MANIFEST}}": str(REPO_ROOT / "output" / "manifests" / f"{context.get('base_name', '')}_prepare.json"), "{{ANALYZE_MANIFEST}}": str(REPO_ROOT / "output" / "manifests" / f"{context.get('base_name', '')}_analyze.json"), "{{DESIGN_MANIFEST}}": str(REPO_ROOT / "output" / "manifests" / f"{context.get('base_name', '')}_design.json"), "{{PROJECT_PROFILE}}": str(REPO_ROOT / "knowledge_base" / "00_project" / "project_profile.md"), "{{FLEET_CONFIG}}": str(REPO_ROOT / "fleet_config.yml"), "{{OUTPUT_DIR}}": str(REPO_ROOT / "output"), "{{KNOWLEDGE_BASE_DIR}}": str(REPO_ROOT / "knowledge_base"), "{{REQUIREMENT_FILE}}": context.get("requirement_file", ""), } result = prompt for placeholder, value in replacements.items(): result = result.replace(placeholder, value) return result def render_agent_prompt(agent_id: str, context: dict[str, Any]) -> str: """加载并渲染 Agent prompt（注入上下文）。""" raw = load_agent_prompt(agent_id) return inject_context(raw, context) def get_agent_frontmatter(agent_id: str) -> dict[str, Any]: """提取 Agent prompt 的 YAML frontmatter。""" content = load_agent_prompt(agent_id) frontmatter: dict[str, Any] = {} if content.startswith("---"): parts = content.split("---", 2) if len(parts) >= 3: for line in parts[1].strip().split("\n"): line = line.strip() if ":" in line: key, value = line.split(":", 1) frontmatter[key.strip()] = value.strip() return frontmatter def get_agent_short_description(agent_id: str) -> str: """返回 Agent 的一句话描述。""" info = AGENT_REGISTRY.get(agent_id, {}) return info.get("description", agent_id) def validate_all_agents() -> dict[str, Any]: """验证所有 Agent prompt 文件是否存在且非空。""" result: dict[str, Any] = {"valid": True, "missing": [], "empty": [], "total": len(AGENT_REGISTRY)} for agent_id in AGENT_REGISTRY: path = get_agent_path(agent_id) if not path.exists(): result["missing"].append(agent_id) result["valid"] = False elif path.stat().st_size == 0: result["empty"].append(agent_id) result["valid"] = False return result