第 23 章 · 实战三：测试智能体——CI 集成与定时回归巡检
本章目标：接续第 19 章的用例产物
cases.json，实现接口自动化与 Web 自动化两类执行 Agent、聚合测试报告，并给出生产级整合清单。
20.1 执行层架构
第 19 章产出的每条用例都有 type 字段，据此路由到不同执行器：
text
cases.json
│ 按 type 路由
├── type = "api" ──▶ 接口执行 Agent ──▶ runApiTest 工具（fetch）
├── type = "ui" ──▶ Web 执行 Agent ──▶ Playwright
└── type = "mixed" ──▶ 拆分为 api + ui 两步分别执行
│
▼
TestResult[]（统一结果结构）
│
▼
报告 Agent ──▶ report.md（含失败分析）统一的结果结构是聚合报告的前提：
typescript
// executor/types.ts —— 执行结果契约
export interface TestResult {
caseId: string; // 对应 TestCase.id
status: "passed" | "failed" | "skipped";
durationMs: number; // 执行耗时
error?: string; // 失败原因（断言差异/超时/元素未找到）
evidence?: string; // 证据：接口为响应摘要，Web 为截图路径
}20.2 接口自动化执行 Agent
核心是一个 runApiTest 工具：Agent 读取用例后，把步骤翻译成 HTTP 请求参数，工具内部用 fetch 执行并断言。
typescript
// executor/apiTool.ts —— 接口测试工具
import { Type } from "typebox";
import type { AgentTool } from "@earendil-works/pi-agent-core";
import type { TestResult } from "./types.js";
export function makeApiTestTool(baseUrl: string): AgentTool {
return {
name: "run_api_test",
description:
"执行单个接口测试。发起 HTTP 请求并断言状态码与响应字段。" +
"expectedJsonPath 形如 user.name，用点号取嵌套字段；" +
"断言失败时返回实际值供分析，不要重试超过一次。",
parameters: Type.Object({
caseId: Type.String({ description: "关联的用例编号，如 TC-001" }),
method: Type.String({ description: "HTTP 方法，大写：GET/POST/PUT/DELETE" }),
path: Type.String({ description: "接口路径，如 /api/cart/items" }),
body: Type.Optional(Type.String({ description: "JSON 字符串的请求体，GET 请求留空" })),
expectStatus: Type.Number({ description: "期望的 HTTP 状态码" }),
expectJsonPath: Type.Optional(Type.String({ description: "期望断言的响应字段路径" })),
expectValue: Type.Optional(Type.String({ description: "该字段期望值（字符串化比较）" })),
}),
execute: async (_id, params) => {
const started = Date.now();
const result: TestResult = {
caseId: params.caseId, status: "passed",
durationMs: 0,
};
try {
// 发起真实请求
const res = await fetch(baseUrl + params.path, {
method: params.method,
headers: { "Content-Type": "application/json" },
body: params.body || undefined,
signal: AbortSignal.timeout(10_000), // 10 秒超时保护
});
// 断言一：状态码
if (res.status !== params.expectStatus) {
throw new Error(`状态码断言失败：期望 ${params.expectStatus}，实际 ${res.status}`);
}
// 断言二：响应字段（可选）
if (params.expectJsonPath) {
const json: any = await res.json();
const actual = params.expectJsonPath
.split(".")
.reduce((node, key) => node?.[key], json);
if (String(actual) !== params.expectValue) {
throw new Error(
`字段断言失败 ${params.expectJsonPath}：期望 ${params.expectValue}，实际 ${actual}`);
}
}
result.evidence = `HTTP ${res.status}，耗时 ${Date.now() - started}ms`;
} catch (err) {
// 失败不抛出给模型重试——记录结构化结果即可，避免无意义重跑
result.status = "failed";
result.error = (err as Error).message;
}
result.durationMs = Date.now() - started;
return {
content: [{ type: "text", text: JSON.stringify(result) }],
details: result,
};
},
};
}然后创建接口执行 Agent，让它按用例批量驱动这个工具：
typescript
// executor/apiAgent.ts
import { Agent } from "@earendil-works/pi-agent-core";
import { createModels } from "@earendil-works/pi-ai";
import type { TestCase, TestResult } from "./types.js";
import { makeApiTestTool } from "./apiTool.js";
const models = createModels();
const model = models.getModel("anthropic", "claude-sonnet-4-6")!;
export async function runApiCases(
cases: TestCase[],
baseUrl: string,
): Promise<TestResult[]> {
const results: TestResult[] = [];
const agent = new Agent({
initialState: {
systemPrompt: `你是接口测试执行器。
逐条分析给定的 api 类用例，为每条调用一次 run_api_test 工具。
把用例的步骤翻译为 method/path/body/expectStatus/expectJsonPath。
所有用例执行完后，只输出一个 JSON 数组（TestResult 结构），不要其他文字。`,
model,
tools: [makeApiTestTool(baseUrl)],
messages: [],
},
streamFn: models.streamSimple.bind(models),
});
let reply = "";
agent.subscribe((e) => {
// 从工具结果中直接收集结构化数据（比解析最终回复更可靠）
if (e.type === "tool_execution_end" && e.toolCallId) {
const r = e.result?.details as TestResult | undefined;
if (r?.caseId) results.push(r);
}
if (e.type === "message_update" && e.assistantMessageEvent.type === "text_delta") {
reply += e.assistantMessageEvent.delta;
}
});
await agent.prompt(`执行以下用例：\n${JSON.stringify(
cases.filter((c) => c.type !== "ui"), null, 2)}`);
console.log(`接口执行完成：${results.filter((r) => r.status === "passed").length}/${results.length} 通过`);
return results;
}从工具 details 收集结果
execute 返回的 details 会出现在 tool_execution_end 事件里——直接从事件收集结构化结果，比让模型在最终回复里复述一遍可靠得多（模型转述 JSON 偶尔会丢字段）。
20.3 Web 自动化执行 Agent
UI 用例交给 Playwright。策略是「Agent 生成 spec 文件 → 命令行执行 → 解析结果」：
typescript
// executor/webAgent.ts —— Web 自动化：生成并运行 Playwright spec
import { Agent } from "@earendil-works/pi-agent-core";
import { createModels } from "@earendil-works/pi-ai";
import { execFile } from "node:child_process";
import { promisify } from "node:util";
import { writeFile, readFile } from "node:fs/promises";
import type { TestCase, TestResult } from "./types.js";
const run = promisify(execFile);
const models = createModels();
const model = models.getModel("anthropic", "claude-sonnet-4-6")!;
export async function runWebCases(
cases: TestCase[],
baseUrl: string,
): Promise<TestResult[]> {
// ① 让 Agent 把 UI 用例翻译成 Playwright spec（TypeScript）
const gen = new Agent({
initialState: {
systemPrompt: `你是 Playwright 测试开发专家。
把 UI 用例翻译为 @playwright/test 的 spec 文件。
规则：
1. 每条用例一个 test()，testTitle 前缀用例编号，如 "TC-003 登录失败提示"；
2. 每个测试开头 await page.goto(baseUrl + 相对路径)；
3. 失败时自动截图：test.use({ screenshot: "only-on-failure" })；
4. 断言使用 expect + locator，禁止裸 waitForTimeout；
5. 只输出完整 spec 文件代码，不要解释。`,
model,
messages: [],
},
streamFn: models.streamSimple.bind(models),
});
let spec = "";
gen.subscribe((e) => {
if (e.type === "message_update" && e.assistantMessageEvent.type === "text_delta") {
spec += e.assistantMessageEvent.delta;
}
});
await gen.prompt(
`baseUrl 是 ${baseUrl}。用例列表：\n${JSON.stringify(
cases.filter((c) => c.type !== "api"), null, 2)}`);
// ② 落盘并执行
const specPath = "tests/generated.spec.ts";
await writeFile(specPath, spec.replace(/^```(?:typescript)?\s*/i, "").replace(/```\s*$/, ""));
try {
await run("npx", ["playwright", "test", specPath, "--reporter=json"], {
cwd: process.cwd(),
timeout: 300_000,
});
} catch (e: any) {
// playwright test 有失败用例时以非零码退出——属预期，解析 stdout 即可
console.error("存在失败用例，解析报告中…");
}
// ③ 解析 JSON 报告为统一 TestResult
const report = JSON.parse(await readFile("test-results/.json-report", "utf-8"));
return report.suites.flatMap((s: any) =>
s.specs.map((sp: any) => ({
caseId: sp.title.split(" ")[0], // "TC-003 ..." 取编号
status: sp.ok ? "passed" : "failed",
durationMs: sp.tests?.[0]?.results?.[0]?.duration ?? 0,
error: sp.tests?.[0]?.results?.[0]?.error?.message,
evidence: sp.tests?.[0]?.results?.[0]?.attachments
?.find((a: any) => a.contentType.includes("image"))?.path,
})),
);
}生成代码要过 review
让 Agent 直接生成可执行脚本很强大，但永远不要未经审查就在生产环境执行。建议：先在隔离目录生成 → 人工或 lint 规则抽查 → 再运行。第 6 章的 beforeToolCall 钩子也可以用来拦截危险操作。
20.4 报告 Agent：聚合与失败分析
typescript
// executor/report.ts —— 聚合两类结果，产出 Markdown 报告
import { Agent } from "@earendil-works/pi-agent-core";
import { createModels } from "@earendil-works/pi-ai";
import { writeFile } from "node:fs/promises";
import type { TestResult } from "./types.js";
const models = createModels();
// 报告分析用便宜的小模型即可——省成本（见 20.5）
const model = models.getModel("anthropic", "claude-haiku-4-5")!;
export async function makeReport(
apiResults: TestResult[],
webResults: TestResult[],
): Promise<string> {
const all = [...apiResults, ...webResults];
const passed = all.filter((r) => r.status === "passed").length;
// 先用纯代码生成统计表——确定性数据不劳烦模型
const table = all.map((r) =>
`| ${r.caseId} | ${r.status} | ${r.durationMs}ms | ${r.error ?? "-"} |`).join("\n");
const agent = new Agent({
initialState: {
systemPrompt: `你是测试报告撰写人。基于给定的失败用例列表：
1. 按疑似根因分组（前端缺陷/后端缺陷/环境问题/用例本身问题）；
2. 每组给出缺陷定位建议（该先查哪一层日志、复现路径）；
3. 输出 Markdown，直接可粘贴进缺陷管理工具。`,
model,
messages: [],
},
streamFn: models.streamSimple.bind(models),
});
const failed = all.filter((r) => r.status === "failed");
let analysis = "";
agent.subscribe((e) => {
if (e.type === "message_update" && e.assistantMessageEvent.type === "text_delta") {
analysis += e.assistantMessageEvent.delta;
}
});
await agent.prompt(`失败用例详情：\n${JSON.stringify(failed, null, 2)}`);
const md = [
`# 测试执行报告`,
``,
`- 总计 ${all.length} 条，通过 ${passed}（通过率 ${Math.round(passed / all.length * 100)}%）`,
`- 接口：${apiResults.length} 条 / Web：${webResults.length} 条`,
``,
`## 明细`,
``,
`| 用例 | 结果 | 耗时 | 失败原因 |`,
`|---|---|---|---|`,
table,
``,
`## 失败分析`,
``,
analysis,
].join("\n");
await writeFile("report.md", md);
console.log(`📄 报告已生成 report.md（通过率 ${Math.round(passed / all.length * 100)}%）`);
return md;
}20.5 生产整合清单
把三段脚本变成可运营的服务，还需补齐以下要素：
① HTTP 服务封装——用 Fastify 暴露触发入口：
typescript
// server.ts —— 触发完整流水线的 HTTP 入口
import Fastify from "fastify";
import { readFile } from "node:fs/promises";
const app = Fastify({ logger: true });
app.post<{ Body: { requirementPath: string; mockups?: string[]; baseUrl: string } }>(
"/qa/run",
async (req, reply) => {
const { requirementPath, mockups = [], baseUrl } = req.body;
// 幂等：用需求文档内容 hash 作为 runId，重复提交直接返回已有报告
const runId = createHash("sha256")
.update(await readFile(requirementPath))
.digest("hex")
.slice(0, 12);
// 实际项目里这里应改为异步任务队列，HTTP 立即返回 runId
const report = await runPipeline(requirementPath, mockups, baseUrl);
return { runId, report };
},
);
app.listen({ port: 3000 });② 会话持久化复跑——第 17 章的 SQLite 后端在这里发挥作用：把每次运行的各阶段输出（功能点/评审/方案/用例/结果）按 runId 落库。复跑时若某阶段输入未变化，直接读缓存跳过 LLM 调用。
③ 错误恢复与重试——区分两类失败：
- 基础设施失败（网络超时、限流 429）：指数退避重试，
agent.continue()可从断点恢复当前回合； - 逻辑失败（评审不通过、JSON 解析失败）：不盲目重试，记录后转人工。JSON 解析失败可重试一次并在 prompt 中附上上次解析错误的具体位置。
④ Token 成本控制——按阶段分配不同档位的模型：
| 阶段 | 任务特点 | 建议档位 |
|---|---|---|
| 需求解析 / 评审 | 长文本理解 + 判断 | 中档（sonnet 级） |
| 测试方案 | 策略设计 | 中高档 |
| 用例生成（含设计图） | 多模态 + 结构化 | 中高档 |
| 接口执行翻译 | 机械翻译步骤 | 低档（haiku 级） |
| 报告分析 | 模式归纳 | 低档 |
再配合 AbortSignal.timeout 防止单回合失控、对 cases.json 做分批（每批 ≤20 条）避免上下文爆炸。
本章小结
- 用例按
type路由到接口/Web 两类执行器，统一TestResult结构是聚合前提； - 接口执行：
run_api_test工具内用 fetch 断言，失败记入结果而非抛给模型重试； - Web 执行：Agent 生成 Playwright spec →
npx playwright test运行 → 解析 JSON 报告，失败自动截图留证； - 报告 = 确定性统计（代码算） + 失败根因分析（低档模型写），生成 Markdown；
- 生产化四件套：Fastify 服务入口、SQLite 会话缓存、区分基础设施/逻辑失败的重试策略、按阶段分档的模型成本控制。
🧪 随堂测验
点击你认为正确的选项。答错时会展示正确答案与原因解析。
1. 为什么从 tool_execution_end 事件的 details 收集测试结果，而不是解析模型的最终文本回复？
2. Web 自动化采用「Agent 生成 spec → 命令行执行」而不是让 Agent 逐步调用浏览器工具，主要好处是？
3. 接口测试执行中，某条用例断言失败，工具内部应该怎么处理？
4. 关于各阶段的模型选档，下列哪种分配最合理？
🛠️ 动手实践
- 为第 19 章的
cases.json补充一个 mock 接口服务（如 json-server），端到端跑通接口执行与报告生成。 - 给 Web 执行器加一个「用例失败自动重跑一次」机制，对比重跑前后的通过率变化，思考哪些失败是偶发的。
- 实现 20.5 的幂等设计：相同需求文档二次提交时直接返回缓存的报告，不触发任何 LLM 调用。
三章实战到此完结。回顾整门课程，你已经掌握了从统一 LLM API 到多 Agent 生产系统的完整链路——下一章：CI 集成与定时回归巡检把流水线接入持续集成。