用 Strands Agents 和 Gemini 检测不稳定测试并定位根因

flaky-test-detector/
├── .github/workflows/
│   └── flaky-report.yml   # weekly analysis + result files
├── agents/
│   └── agent.py           # optional AI agent: root-cause analysis of reports
├── docs/
│   └── how-it-works.svg   # diagram used in this README
├── flaky_detector/
│   ├── __init__.py
│   ├── __main__.py        # CLI
│   ├── parsers.py         # JUnit XML parser
│   ├── github.py          # fetches JUnit artifacts from GitHub Actions
│   ├── scoring.py         # flakiness scoring and ranking
│   └── report.py          # result files (JSON / CSV / Markdown)
├── scripts/
│   └── generate_junit.py  # generates sample JUnit XML files
├── results/               # sample JUnit XML files (git-ignored)
├── reports/               # generated reports and AI analyses
├── tests/
├── pyproject.toml         # project metadata and dev dependencies (uv)
├── requirements.txt
└── README.md

class Outcome(str， Enum)：
    PASSED = "passed"
    FAILED = "failed"
    ERROR = "error"
    SKIPPED = "skipped"


@dataclass(frozen=True)
class TestResult：
    """单次 CI 运行中的单次测试执行。"""
    __test__ = False  # 防止 pytest 尝试收集这个类

    name： str  # "<classname>::<name>"
    outcome: Outcome
    commit_sha: str | None
    run_id: str | None
    time: float = 0.0
    message: str | None = None
    source: str | None = None  # 结果来自的文件

    @property
    def passed(self) -> bool:
        return self.outcome is Outcome.PASSED

@dataclass(frozen=True)
class TestScore:
    """单个测试的不稳定性评分。"""
    __test__ = False  # 防止 pytest 尝试收集这个类

    name: str
    commits: int  # 该测试有可用数据的 commit 数
    flaky_commits: int  # 该测试结果不一致的 commit 数
    commits_where_flaky: tuple[str, ...] = field(default=())
    error_messages: tuple[str, ...] = field(default=())  # 去重后的失败消息，按首次出现顺序

    @property
    def score(self) -> float:
        return self.flaky_commits / self.commits if self.commits else 0.0


class FlakinessScorer:
    def score(self, results: Iterable[TestResult]) -> list[TestScore]:
        outcomes: dict[str, dict[str, set[bool]]] = defaultdict(lambda: defaultdict(set))
        messages: dict[str, dict[str, None]] = defaultdict(dict)  # 有序、去重
        for r in results:
            if not r.commit_sha or r.outcome.value == "skipped":
                continue
            outcomes[r.name][r.commit_sha].add(r.passed)
            if not r.passed and r.message:
                messages[r.name][r.message] = None
        scores = []
        for name, by_commit in outcomes.items():
            flaky = tuple(sorted(sha for sha, seen in by_commit.items() if len(seen) > 1))
            scores.append(
                TestScore(
                    name=name,
                    commits=len(by_commit),
                    flaky_commits=len(flaky),
                    commits_where_flaky=flaky,
                    error_messages=tuple(messages[name]),
                )
            )
        return scores

{
  "name": "tests.checkout.test_cart::test_9",
  "commits": 10,
  "flaky_commits": 3,
  "commits_where_flaky": [
    "gen0007d154385",
    "gen05893e77ff9",
    "gen059095256f8"
  ],
  "error_messages": [
    "AssertionError: timeout waiting for gateway",
    "ConnectionError: connection reset by peer"
  ]
}

def __init__(self, threshold: float = 0.1, min_commits: int = 5) -> None:
    if not 0.0 <= threshold <= 1.0:
        raise ValueError("threshold must be between 0 and 1")
    if min_commits < 1:
        raise ValueError("min_commits must be >= 1")
    self.threshold = threshold
    self.min_commits = min_commits

python -m flaky_detector analyze ./results/*.xml --threshold 0.1

{
  "generated_at": "2026-11-10T19:32:23+00:00",
  "source": "local",
  "threshold": 0.1,
  "min_commits": 5,
  "results_parsed": 19644,
  "tests_analyzed": 109,
  "flaky_count": 12,
  "flaky_rate": 0.1101,
  "tests": [
    {
      "name": "tests.checkout.test_cart::test_9",
      "score": 0.7,
      "commits": 60,
      "flaky_commits": 42,
      "is_flaky": true,
      "commits_where_flaky": [
        "gen0007d154385",
        "gen059095256f8"
      ],
      "error_messages": [
        "AssertionError: timeout waiting for gateway",
        "ConnectionError: connection reset by peer",
        "AssertionError: race condition on shared state",
        AssertionError: expected 200, got 503,
        TimeoutError: operation timed out after 30s
      ]
    }
  ]
}

from strands import Agent
from strands.models.gemini import GeminiModel

model = GeminiModel(
    client_args={"api_key": api_key},
    model_id="gemini-2.5-flash",
)

agent = Agent(
    model=model,
    tools=[fetch_json_report, export_analysis],
    name="Flaky Test Analysis Agent",
    description="这个智能体分析不稳定测试报告，并提供关于不稳定测试、其分数、趋势的洞察，同时分析错误消息以提供关于不稳定根因的洞察。当被要求总结不稳定测试报告时，它会使用 fetch_json_report 工具获取报告数据，然后提供总结，说明有多少测试不稳定、哪些最严重，以及基于错误消息的可能根因。",
    system_prompt="你是一个有用的助手，负责分析不稳定测试报告。在需要时使用 fetch_json_report 工具获取报告数据。始终阅读每个不稳定测试的 error_messages，并推断可能的根因（超时、竞态条件、网络/网关不稳定、共享状态等），说明每个原因影响哪些测试以及你的置信度。分析完成后，调用一次 export_analysis，传入 summary、root_causes 和 recommendations 以保存 Markdown 报告，然后告诉用户文件路径。",
)

def main() -> None:
    parser = argparse.ArgumentParser(description=用 AI 智能体分析不稳定的测试报告。)
    parser。add_argument("period"， nargs="?"， default="last_24_hours"， choices=list(PERIODS)，
                        help="要分析的报告周期（默认：%(default)s）")
    args = parser。parse_args()
    label = args。period。replace("_"， " ")
    agent(f"总结来自 {label} 的不稳定测试报告（period={args.period}）：有多少测试不稳定，哪些最严重？完成后，分析错误信息以推断不稳定的可能根本原因，并给出修复建议。然后用 export_analysis 工具创建报告文件，再告诉我写入的 Markdown 文件路径。"
          "how many tests are flaky and which are the worst? After finishing, analyze the error messages to infer the possible root causes of flakiness and give recommendations to fix them. And create the report file with export_analysis tool, then tell me the path of the Markdown file that was written.")


if __name__ == "__main__"：
    main()