原文：https://dev.to/m4rri4nne/detecting-flaky-tests-and-finding-their-root-causes-with-strands-agents-and-gemini-4kan（作者 @m4rri4nne）
几年前（最多三年前），QA 工程师面临的最大挑战之一是：如何说服业务方和团队，让我们在自动化测试上投入资源。如今有了 AI，这个问题基本解决了——我们不用花太多时间编写自动化测试，因为有 AI 替我们来做。但一个新的挑战随之出现：flaky 测试（结果时好时坏、不稳定的测试）。本文我会讲清楚什么是 flaky 测试、我如何构建了一个工具来识别它们，以及我怎样用 AI 跑了第一轮分析。
什么是 flaky 测试，为什么它很重要
在深入讲解决方案之前，先弄清楚什么是 flaky 测试，以及我们为什么该在意它。
一句话概括：flaky 测试就是那种在代码和测试本身都没有任何改动的情况下，既会通过也会失败的自动化测试。
既然它们最终都能通过，为什么还要监控？反正结果是过的。
这可能是你的第一反应，但一旦理解了 flaky 测试是什么，你就会发现：在大型项目里，它会破坏团队对自家产品和测试的信任，真正的失败开始被忽视，CI 的耗时和每条流水线的成本也会往上涨。
项目速览
基于这些顾虑，我做了这个项目。代码已在 GitHub 开源：flaky-test-detector。
第一版草稿里，我只打算解析测试结果、按单个测试和提交（commit）打分，然后手动做分析：
第一版之后我想到：为什么不接入一个 AI 来帮我做分析？带着这个想法，我新增了支持分析和对接 AI 模型的功能。如今这个方案长这样：
项目的目录结构如下：
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
flaky 测试的打分与排名
先来看项目如何给 flaky 测试打分和排名。流程从接收一份 XML 格式的 JUnit 报告开始（这是大多数测试框架的标准输出，每个被执行的测试对应一个 <testcase>），由 parsers.py 负责把文件里的信息解析成合适的格式：
class Outcome(str, Enum):
PASSED = "passed"
FAILED = "failed"
ERROR = "error"
SKIPPED = "skipped"
@dataclass(frozen=True)
class TestResult:
"""A single test execution in a single CI run."""
__test__ = False # keep pytest from trying to collect this class
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 # file the result came from
@property
def passed(self) -> bool:
return self.outcome is Outcome.PASSED
如果文件不是合法的 JUnit XML，就会抛出异常；如果合法，解析继续进行。调用 scoring.py 时，它会使用 FlakinessScorer 类，由这个类负责把 TestResult 对象转换成 TestScore 对象。所有结果先按测试名分组，再按 commit 分组，评分器对每个（测试，commit）组合检查其结果。以下情况会被忽略：
skipped的测试，因为它们说明不了任何 flaky 特征；- 来自缺少
commit_sha属性（commit id）的 JUnit 文件的结果，因为它们彼此之间无法比较。
failed 和 error 都算作失败。
@dataclass(frozen=True)
class TestScore:
"""Flakiness score of a single test."""
__test__ = False # keep pytest from trying to collect this class
name: str
commits: int # commits with usable data for this test
flaky_commits: int # commits where the test had mixed outcomes
commits_where_flaky: tuple[str, ...] = field(default=())
error_messages: tuple[str, ...] = field(default=()) # distinct failure messages, first seen first
@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) # ordered, de-duplicated
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
当同一个 commit 下的结果同时是 {True, False} 时——也就是同一份代码库上这个测试既失败又通过——脚本就把该测试标记为 flaky。随后它会生成 Ranker 所需的数据，格式如下：
{
"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"
]
}
调用 FlakinessRanker 类时，它利用这份数据对测试进行排名。它只考虑至少有 5 个 commit 数据的测试（min_commits），这样分数才可信；当某个测试的分数达到 0.1 或更高（threshold）时，就将其标记为 flaky。两个值都可以调整：
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
分数的算法是：flaky_commits（同一个测试既通过又失败的 commit 数量）除以流水线中该测试可用的 commits 总数。
某个测试在 10 个 commit 上运行过，其中 3 个 commit 上它既通过又失败。
在这个例子里，commits 是 10，flaky_commits 是 3，所以分数是 0.3。
第一版实现已经能看到初步结果了。我构造了几份 JUnit 报告，运行下面的命令后得到了这样的输出：
python -m flaky_detector analyze ./results/*.xml --threshold 0.1
每一行依次显示分数、测试名，以及 flaky commit 数占该测试有数据的 commit 总数的比例。
报告生成与 Agent 创建
前面提到过，把基础功能做完之后我又想：为什么不继续改进呢？第一步是把结果生成报告。因为我已经在考虑接入 AI agent，报告就需要一种适合它的格式。报告由 report.py 生成，支持 JSON、CSV 和 Markdown 三种格式（通过 --output-dir 选项指定）。本文重点介绍 JSON 格式，因为 AI agent 用的正是它：
{
"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"
]
}
]
}
构建这个 AI agent 用的是 Strands Agents SDK。我第一次接触这个 SDK 是在 Train the Trainers 工作坊上（感谢 AWS EMEA 的邀请），由于配置 agent 的方式非常简单，我决定在这个项目里使用它。
本项目中我使用的是 GeminiModel（API key 从 .env 文件的 GOOGLE_API_KEY 环境变量读取），当然你也可以换成其他模型，参见 Model Providers 文档。
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="This agent analyzes flaky test reports and provides insights on flaky tests, their scores, trends, and it analyzes the error messages to provide insights on the root causes of flakiness. And when it is asked to summarize the flaky test report, it will use the fetch_json_report tool to retrieve the report data and then provide a summary of how many tests are flaky and which are the worst and the possible root causes based on the error messages.",
system_prompt="You are a helpful assistant that analyzes flaky test reports. Use the fetch_json_report tool to retrieve report data when needed. Always read the error_messages of each flaky test and infer the possible root causes (timeouts, race conditions, network/gateway instability, shared state, etc.), saying which tests each cause affects and how confident you are. After you finish analysing, call export_analysis once with summary, root_causes and recommendations to save the Markdown report, then tell the user the file path.",
)
构建 agent 时，我们需要给它提供工具，它才能干活。这个应用我写了两个工具：
fetch_json_report：获取已生成的 JSON 报告。它负责按时间戳或时间段查找文件，并把结果按从旧到新的顺序合并在一次响应中返回。export_analysis：把 AI 做出的分析保存为 Markdown 文件。
有了这两个工具，就可以调用 agent 来执行分析了：
def main() -> None:
parser = argparse.ArgumentParser(description="Analyze flaky test reports with an AI agent.")
parser.add_argument("period", nargs="?", default="last_24_hours", choices=list(PERIODS),
help="period of reports to analyze (default: %(default)s)")
args = parser.parse_args()
label = args.period.replace("_", " ")
agent(f"Summarize the flaky test reports from the {label} (period={args.period}): "
"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()
时间段作为参数传入（例如 python agents/agent.py last_7_days），agent 会给出类似这样的报告：Flaky Tests Analysis - Last 24 Hours。
有了这些能做什么
好了，现在可以识别 flaky 测试，甚至对它们做分析了。那这些能用来干什么？首先，你可以开始在团队里落地。监控测试并把 flaky 测试标记出来，能提升团队对自己所构建内容的信任、降低 CI 成本，也能避免真正的 bug 因为测试在第二次或第三次执行时通过了而被忽略。
什么时候该跑这套流程？取决于团队的工作方式：
- 如果团队有持续部署，每天或每周都发版，那么每周跑一次就很合适。如果低环境里经常出现 PR 合并困难，跑一次也有助于搞清楚到底发生了什么。
- 如果没有持续部署、每月才发一次版，建议每次部署时都跑，并且始终对比最近几次执行的结果，看数量有没有上涨。
- AI 分析是可选的。你可以只跑打分和排名两个步骤，等一个周期结束——一个月甚至一个季度——再让 agent 给你反馈。
局限与集成
- 目前 JUnit XML 文件必须带有属性
<property name="commit_sha" value=""/>（commit id）和属性<property name="run_id" value=""/>。由于这套方案是围绕 GitHub Actions 构建的，用到其他框架和流水线时可能需要改动。当结果从 GitHub Actions 获取时（--repo），这两个值直接取自 GitHub run 本身。 - 已经有一个可直接集成 GitHub Actions 的 workflow（
.github/workflows/flaky-report.yml）。它每周一运行一次，生成报告并作为 artifact 上传。如果你打算把它加进自己的 workflow，可能需要调整，且被分析的仓库必须把 JUnit XML 文件作为 workflow artifact 发布出来。 - AI 模型可以配置成你喜欢的任何一个。记得查看 Model Providers 文档，按你的场景做适配。
- 另一个建议是接入 Grafana 这类报表工具，这样你就能看到 flaky 测试在不同时间点的表现，以及它们是否在减少。
结语
flaky 测试不只是烦人：它们会侵蚀团队对测试的信任，还会把真正的失败藏起来。度量它们是处理它们的第一步，而每个测试、每个 commit 一个简单分数就足够起步了。AI 分析能让你先看到可能的根因，但它不能替代人工复核结果，因为基础设施导致的失败看起来也可能像 flaky。试一试，看看你自己的流水线表现如何！
原文：https://dev.to/m4rri4nne/detecting-flaky-tests-and-finding-their-root-causes-with-strands-agents-and-gemini-4kan（作者 @m4rri4nne）
