xst
300 lines
11 KiB
Python
300 lines
11 KiB
Python
"""
|
|
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
|
|
|
|
|
|
def build_agent_invocation_context(base_name: str, agent_id: str) -> dict[str, Any]:
|
|
"""为 AI Agent 调用构建完整的上下文注入字典。"""
|
|
from fleet_manifest import ZONE_ORDER, load_manifest_safe
|
|
|
|
info = AGENT_REGISTRY.get(agent_id, {})
|
|
zone = info.get("zone", "")
|
|
|
|
context: dict[str, Any] = {
|
|
"base_name": base_name,
|
|
"agent_id": agent_id,
|
|
"agent_zone": zone,
|
|
"repo_root": str(REPO_ROOT),
|
|
"output_dir": str(REPO_ROOT / "output"),
|
|
"manifests_dir": str(REPO_ROOT / "output" / "manifests"),
|
|
"knowledge_base_dir": str(REPO_ROOT / "knowledge_base"),
|
|
"fleet_config_path": str(REPO_ROOT / "fleet_config.yml"),
|
|
"PROJECT_PROFILE": str(REPO_ROOT / "knowledge_base" / "00_project" / "project_profile.md"),
|
|
"FLEET_CONFIG": str(REPO_ROOT / "fleet_config.yml"),
|
|
}
|
|
|
|
for prev_zone in ZONE_ORDER:
|
|
manifest = load_manifest_safe(base_name, prev_zone)
|
|
if manifest is None:
|
|
continue
|
|
context[f"manifest_{prev_zone}"] = manifest
|
|
for key, value in manifest.items():
|
|
if key.endswith("_file") or key.endswith("_files"):
|
|
context[key] = value
|
|
if prev_zone == zone:
|
|
break
|
|
|
|
return context
|
|
|
|
|
|
def get_agent_input_files(base_name: str, agent_id: str) -> list[str]:
|
|
"""返回 Agent 需要的所有输入文件路径列表。"""
|
|
from fleet_manifest import ZONE_ORDER, load_manifest_safe
|
|
|
|
info = AGENT_REGISTRY.get(agent_id, {})
|
|
zone = info.get("zone", "")
|
|
input_files: list[str] = []
|
|
|
|
for prev_zone in ZONE_ORDER:
|
|
if prev_zone == zone:
|
|
break
|
|
manifest = load_manifest_safe(base_name, prev_zone)
|
|
if manifest is None:
|
|
continue
|
|
for key in manifest:
|
|
if key.endswith("_file") or key.endswith("_files"):
|
|
value = manifest[key]
|
|
if isinstance(value, str):
|
|
input_files.append(value)
|
|
elif isinstance(value, list):
|
|
input_files.extend(v for v in value if isinstance(v, str))
|
|
|
|
kb_files = [
|
|
str(REPO_ROOT / "knowledge_base" / "00_project" / "project_profile.md"),
|
|
str(REPO_ROOT / "knowledge_base" / "01_standards" / "test_case_template.md"),
|
|
str(REPO_ROOT / "knowledge_base" / "01_standards" / "review_checklist.md"),
|
|
str(REPO_ROOT / "knowledge_base" / "01_standards" / "definition_of_done.md"),
|
|
str(REPO_ROOT / "fleet_config.yml"),
|
|
]
|
|
input_files.extend(kb_files)
|
|
|
|
return sorted(set(input_files))
|