log-diagnosis¶
第5章 · Coding Agent 与代码生成 · 配套项目
chapter5/log-diagnosis
项目说明¶
实验 5-8：生产日志的智能诊断系统¶
配套《深入理解 AI Agent》第 5 章「代码作为生成式能力 —— Agent 执行日志自动分析和问题诊断」。
目的¶
生产环境的 Agent 会产生大量轨迹日志（trajectory）。从中识别问题、定位根因、构建回归测试成本很高。 本实验让一个诊断 Agent 自动完成这条流水线：
读轨迹集合 + 架构文档 + PRD → 定位问题、生成结构化报告 → 生成回归测试用例 → 重放框架真正执行验证 → (mock) 通过 MCP 对接 GitHub 创建 Issue。
诊断流水线¶
data/trajectories.jsonl (含已知问题的生产轨迹)
data/architecture.md (系统架构) ┐
data/PRD.md (产品需求) ├─► [LLM] diagnose() 结构化问题报告(优先级/模块/描述/建议)
┘ │
▼
[LLM] gen_test_cases() 回归测试用例(引用轨迹ID+交互轮次)
│
▼
replay.py 重放框架 ── 对同一输入重放被测系统并断言
(A) 未修复系统 → FAIL(复现bug)
(B) 修复后系统 → PASS(验证修复)
│
▼
github_mcp.py (mock) 渲染并打印/落盘 GitHub Issue
diagnoser.py：诊断 Agent，两次真实调用 OpenAI（默认 gpt-5.6-luna，JSON 模式）。sut.py：被测系统的确定性仿真器。fixed=False复现线上 bug，fixed=True模拟修复后行为。replay.py：回归测试重放框架。取轨迹输入 → 重放sut→ 在新轨迹上求值断言（内置 4 种断言 DSL）。github_mcp.py：GitHub Issue 创建，默认 mock（打印 + 写output/github_issues.json）。
预置的已知问题（Agent 应能定位）¶
| 轨迹 | 问题 | 违反 PRD | 定位模块 |
|---|---|---|---|
| T-1001 / T-1002 | 退款前跳过了强制的 verify_refund_eligibility 校验 |
R1 (P0) | order_service |
| T-1002 | process_refund 反复失败、无退避、且最终误报成功 |
R2 (P0) | payment_service |
| T-1003 | check_stock 延迟 8300ms 超时未降级 |
R3 (P1) | inventory_service |
| T-1004 | 正常轨迹（对照组，无问题） | — | — |
运行¶
pip install -r requirements.txt
cp env.example .env # 填入 OPENAI_API_KEY（模型默认 gpt-5.6-luna）；未配置时设 OPENROUTER_API_KEY 自动改走 OpenRouter
python demo.py # 完整流程（两次真实 LLM 调用）
demo.py 一次跑完：读轨迹 → 诊断报告 → 回归测试用例 → 重放执行(通过/失败) → (mock) GitHub Issue。
常用参数（python demo.py -h 查看全部）：
--smoke：免 API 快速自检，跳过 LLM，用内置诊断结果仅跑重放框架 + GitHub mock，验证管道是否端到端连通（全绿退出码 0）。适合无 Key 环境或 CI。--model gpt-5.6：临时覆盖模型（等价于设置OPENAI_MODEL）。--data-dir DIR：换用自己的输入目录（需含trajectories.jsonl+architecture.md+PRD.md，默认data/）。--output FILE：mock GitHub Issue 的落盘路径（默认output/github_issues.json）。--create-issue：经 MCP 在真实仓库创建 Issue（需GITHUB_TOKEN+GITHUB_REPO，见下节；缺失时自动回退 mock）。--no-github：跳过步骤 4，不生成 GitHub Issue。
真实运行输出（节选）¶
诊断阶段，Agent 定位到全部 3 个预置问题：
[问题 1] 未进行退款资格校验
优先级 : P0 模块: order_service PRD: R1
轨迹 : ['T-1001', 'T-1002'] 关键轮次: [3]
[问题 2] 支付重试机制未正确实现
优先级 : P0 模块: payment_service PRD: R2
[问题 3] 库存查询延迟未降级处理
优先级 : P1 模块: inventory_service PRD: R3
回归测试用例被重放框架真正执行（先复现 bug、再验证修复）：
(A) 对『线上未修复』系统重放 —— 期望复现 bug（FAIL）
[FAIL] RT-001 (T-1001) 工具 verify_refund_eligibility 缺失
[FAIL] RT-002 (T-1002) process_refund 调用 3 次, 失败 3 次, 末次失败
[FAIL] RT-003 (T-1003) check_stock 最大延迟 8300ms, 阈值 5000ms
(B) 对『修复后』系统重放 —— 期望修复被验证（PASS）
[PASS] RT-001 (T-1001) 工具 verify_refund_eligibility 出现
[PASS] RT-002 (T-1002) process_refund 调用 2 次, 失败 1 次, 末次成功
[PASS] RT-003 (T-1003) check_stock 最大延迟 400ms, 阈值 5000ms
小结：复现 bug 3/3 条；修复后通过 3/3 条。
mock GitHub Issue 打印并写入 output/github_issues.json，示例：
title : [P0][order_service] 未进行退款资格校验
labels : ['module:order_service', 'priority:critical', 'auto-diagnosis']
body : ## 问题描述 ... ## 关联回归测试用例 - RT-001 (轨迹 T-1001 第 3 轮) ...
回归测试断言 DSL（replay 框架内置）¶
Agent 生成的测试用例须使用以下断言之一，框架可自动求值：
step_present{tool}：某工具必须出现（如强制前置校验）。tool_succeeds{tool}：某工具最终成功、且不存在"多次失败后误报成功"。latency_under{tool, threshold_ms}：某工具单次延迟低于阈值。final_status_is{value}：任务最终状态等于给定值。
如何适配/扩展¶
- 换模型：设置
OPENAI_MODEL（或python demo.py --model <名称>）。diagnoser.py默认gpt-5.6-luna，均走 JSON 模式；更强模型对复杂/隐性问题更稳。 - 换供应商：本项目用官方
openaiSDK，只需再设OPENAI_BASE_URL指向兼容 OpenAI 接口的服务（如 Moonshot / 火山方舟 / 本地 vLLM），配合该供应商的OPENAI_API_KEY与OPENAI_MODEL即可，无需改代码。例如： - 换日志：把你自己的生产轨迹按
data/trajectories.jsonl的结构（trajectory_id / task / task_input / turns[] / final_status，turns内含module/tool/input/output/status/latency_ms）落盘替换即可；同时更新data/architecture.md与data/PRD.md作为诊断依据。若轨迹字段不同，sut.py（重放桩）与replay.py（断言求值）按新字段小幅调整。 - 接入真实 GitHub MCP：见下一节，通过
GITHUB_TOKEN+GITHUB_REPO+--create-issue把mock=False接通。
接入真实 GitHub MCP（需 token，默认 mock）¶
本实验默认 mock，--create-issue 才会真正联网。真实创建的实现已内置在
github_mcp._create_issues_via_mcp()：通过 MCP 客户端（mcp SDK，stdio）连接官方
GitHub MCP Server，逐个调用其 create_issue 工具，传入 build_issue() 生成的
title / body / labels / assignees。启用步骤：
- 准备一个 GitHub Personal Access Token（
repo权限），写入.env的GITHUB_TOKEN； 并设置目标仓库GITHUB_REPO=owner/repo。 - 确保本机可启动官方 GitHub MCP Server。默认启动命令用官方 Docker 镜像
ghcr.io/github/github-mcp-server；可用GITHUB_MCP_COMMAND覆盖为任意暴露create_issue工具的 MCP Server（token 经GITHUB_PERSONAL_ACCESS_TOKEN注入其环境）。 pip install mcp（仅此路径需要；mock/自检不需要）。- 运行
python demo.py --create-issue。缺少GITHUB_TOKEN/GITHUB_REPO时会打印提示并 自动回退 mock，避免误联网。
局限¶
- 被测系统
sut.py是确定性仿真，用于让回归测试可真正重放、给出稳定的通过/失败；真实场景下重放需对接实际系统或录制/回放的依赖桩。 - 诊断质量取决于 LLM；gpt-5.6-luna 在本数据集能稳定定位全部 3 类预置问题（R1 校验缺失 / R2 重试误报 / R3 延迟未降级），步骤 1 诊断稳定；但它倾向把 R1「校验缺失」按 T-1001、T-1002 各报一条，故常输出 4 条问题（而非上文示例合并成的 3 条）。步骤 2 生成断言时，它稳定地为「支付重试」问题选用
final_status_is:failed（两次实跑均如此）而非上文示例的tool_succeeds——因修复后的被测系统会重试成功（final_status=success），该断言在修复后重放中判 FAIL，使修复后通过数降为 3/4 而非满绿。此外它偶尔给step_present/latency_under的工具名加上模块前缀（如order_service.verify_refund_eligibility），与重放框架按裸工具名匹配不符，会进一步压低修复后通过数（两次实跑分别得到 3/4 与 0/4）。--smoke内置用例则确定性给出 3/3。 - GitHub 创建默认 mock，不联网；
--create-issue才经真实 MCP Server 联网创建（需 token + repo + 可用的 GitHub MCP Server）。 - 轨迹格式为简化示意，生产环境轨迹字段更丰富（token 用量、子 Agent 调用树等）。
源代码¶
demo.py¶
"""
demo.py —— 实验 5-8：生产日志的智能诊断系统（全流程演示）
流水线：
读轨迹集合 + 架构 + PRD
-> [LLM] 诊断：定位问题、结构化报告(优先级/模块/描述/建议)
-> [LLM] 生成回归测试用例(引用轨迹ID+交互轮次)
-> 重放框架真正执行：先复现 bug(FAIL)，再验证修复(PASS)
-> (mock) 通过 MCP 对接 GitHub 创建 Issue
运行：
cp env.example .env && 填入 OPENAI_API_KEY
python demo.py # 完整流程（两次真实 LLM 调用，GitHub 步骤默认 mock）
python demo.py --smoke # 快速自检：跳过 LLM，用内置用例仅跑重放+GitHub mock
python demo.py --model gpt-5.6 # 临时切换模型
python demo.py --create-issue # 真正经 MCP 创建 GitHub Issue（需 GITHUB_TOKEN+GITHUB_REPO）
python demo.py -h # 查看全部参数
换供应商/模型：设置 OPENAI_BASE_URL + OPENAI_MODEL（见 README『如何适配/扩展』）。
"""
import argparse
import json
import os
import sys
try:
from dotenv import load_dotenv
load_dotenv()
except Exception:
pass
from diagnoser import Diagnoser
import replay
import github_mcp
HERE = os.path.dirname(os.path.abspath(__file__))
DATA = os.path.join(HERE, "data")
DEFAULT_OUTPUT = os.path.join(HERE, "output", "github_issues.json")
# --smoke 自检用的内置样例：与 LLM 在本数据集上的稳定产物一致，
# 使得无需联网/无需 API Key 即可验证 重放框架 + GitHub mock 的端到端管道。
_CANNED_PROBLEMS = [
{"title": "未进行退款资格校验", "priority": "P0", "module": "order_service",
"description": "退款前缺失强制的 verify_refund_eligibility 校验。", "prd_ref": "R1",
"trajectory_ids": ["T-1001", "T-1002"], "focus_turns": [3]},
{"title": "支付重试机制未正确实现", "priority": "P0", "module": "payment_service",
"description": "process_refund 反复失败、无退避、且最终误报成功。", "prd_ref": "R2",
"trajectory_ids": ["T-1002"], "focus_turns": [7]},
{"title": "库存查询延迟未降级处理", "priority": "P1", "module": "inventory_service",
"description": "check_stock 延迟 8300ms 超时未降级。", "prd_ref": "R3",
"trajectory_ids": ["T-1003"], "focus_turns": [3]},
]
_CANNED_TEST_CASES = [
{"test_id": "RT-001", "trajectory_id": "T-1001", "focus_turn": 3,
"description": "退款前必须先做资格校验",
"assertion": {"type": "step_present", "params": {"tool": "verify_refund_eligibility"}}},
{"test_id": "RT-002", "trajectory_id": "T-1002", "focus_turn": 7,
"description": "process_refund 应最终成功且无『多次失败后误报成功』",
"assertion": {"type": "tool_succeeds", "params": {"tool": "process_refund"}}},
{"test_id": "RT-003", "trajectory_id": "T-1003", "focus_turn": 3,
"description": "check_stock 延迟应低于 5000ms",
"assertion": {"type": "latency_under", "params": {"tool": "check_stock", "threshold_ms": 5000}}},
]
def _read(data_dir, name):
with open(os.path.join(data_dir, name), "r", encoding="utf-8") as f:
return f.read()
def _traj_path(data_dir):
return os.path.join(data_dir, "trajectories.jsonl")
def _hr(title):
print("\n" + "=" * 70)
print(title)
print("=" * 70)
def _replay_and_issues(problems, test_cases, do_github=True,
traj_path=None, out_path=DEFAULT_OUTPUT, create_issue=False):
"""步骤 3/4：对同一输入重放被测系统并断言，再生成 GitHub Issue（默认 mock）。
对 fixed=False / fixed=True 各重放一次，演示同一条回归用例的
『失败(复现bug)』与『通过(验证修复)』。返回 (复现数, 验证数)。
"""
traj_path = traj_path or replay._DATA
_hr("步骤 3｜重放框架真正执行测试用例")
print("(A) 对『线上未修复』系统重放 —— 期望复现 bug（FAIL）")
buggy = replay.run_suite(test_cases, fixed=False, path=traj_path)
for r in buggy:
flag = "PASS" if r["passed"] else "FAIL"
print(f" [{flag}] {r['test_id']} ({r.get('trajectory_id')}) {r['detail']}")
print("\n(B) 对『修复后』系统重放 —— 期望修复被验证（PASS）")
fixed = replay.run_suite(test_cases, fixed=True, path=traj_path)
for r in fixed:
flag = "PASS" if r["passed"] else "FAIL"
print(f" [{flag}] {r['test_id']} ({r.get('trajectory_id')}) {r['detail']}")
reproduced = sum(1 for r in buggy if not r["passed"])
verified = sum(1 for r in fixed if r["passed"])
print(f"\n 小结：复现 bug {reproduced}/{len(buggy)} 条；修复后通过 {verified}/{len(fixed)} 条。")
if do_github:
token, repo = os.getenv("GITHUB_TOKEN"), os.getenv("GITHUB_REPO")
if create_issue and token and repo:
_hr(f"步骤 4｜通过 MCP 对接 GitHub 在 {repo} 真实创建 Issue")
github_mcp.create_issues(problems, test_cases, mock=False,
out_path=out_path, repo=repo, token=token)
else:
if create_issue:
print("\n[提示] --create-issue 需要 GITHUB_TOKEN 与 GITHUB_REPO(owner/repo)，"
"当前缺失，已回退到 mock。")
_hr("步骤 4｜通过 MCP 对接 GitHub 创建 Issue（mock，不联网）")
github_mcp.create_issues(problems, test_cases, mock=True, out_path=out_path)
return reproduced, verified
def run_smoke(data_dir=DATA, out_path=DEFAULT_OUTPUT):
"""快速自检：不调用 LLM，用内置样例仅跑 重放框架 + GitHub mock 的端到端管道。
退出码：管道全绿(复现全部 + 验证全部)返回 0，否则返回 3。
"""
_hr("自检模式（--smoke）：跳过 LLM，用内置诊断结果验证重放+GitHub mock 管道")
reproduced, verified = _replay_and_issues(
_CANNED_PROBLEMS, _CANNED_TEST_CASES,
traj_path=_traj_path(data_dir), out_path=out_path)
n = len(_CANNED_TEST_CASES)
ok = reproduced == n and verified == n
print(f"\n自检结果：{'OK' if ok else 'FAILED'}（复现 {reproduced}/{n}，验证 {verified}/{n}）")
return 0 if ok else 3
def run_full(model=None, do_github=True, data_dir=DATA,
out_path=DEFAULT_OUTPUT, create_issue=False):
"""完整流程：真实调用 OpenAI 诊断并生成回归用例，再重放执行。"""
if not (os.getenv("OPENAI_API_KEY") or os.getenv("OPENROUTER_API_KEY")):
print("错误：未设置 OPENAI_API_KEY（或 OPENROUTER_API_KEY 兜底），请 cp env.example .env 后填入"
"（或用 python demo.py --smoke 免 API 自检）。")
sys.exit(1)
# ---------- 0. 读取输入 ----------
architecture = _read(data_dir, "architecture.md")
prd = _read(data_dir, "PRD.md")
trajectories = list(replay.load_trajectories(_traj_path(data_dir)).values())
_hr(f"步骤 0｜读取输入：{len(trajectories)} 条生产轨迹 + 架构文档 + PRD")
for t in trajectories:
print(f" - {t['trajectory_id']}: {t['task']}（{len(t['turns'])} 轮）")
agent = Diagnoser(model=model) if model else Diagnoser()
print(f" 使用模型：{agent.model}")
# ---------- 1. 诊断：定位问题 ----------
_hr("步骤 1｜Agent 诊断（真实调用 OpenAI）：定位问题并生成结构化报告")
problems = agent.diagnose(architecture, prd, trajectories)
if not problems:
print("未诊断出问题（异常）。")
sys.exit(2)
for i, p in enumerate(problems, 1):
print(f"\n[问题 {i}] {p.get('title', '')}")
print(f" 优先级 : {p.get('priority')} 模块: {p.get('module')} PRD: {p.get('prd_ref')}")
print(f" 轨迹 : {p.get('trajectory_ids')} 关键轮次: {p.get('focus_turns')}")
print(f" 描述 : {p.get('description')}")
print(f" 建议 : {p.get('suggestion')}")
# ---------- 2. 生成回归测试用例 ----------
_hr("步骤 2｜Agent 生成回归测试用例（真实调用 OpenAI）：引用轨迹ID + 交互轮次")
test_cases = agent.gen_test_cases(problems)
for tc in test_cases:
print(f" {tc.get('test_id')} 轨迹={tc.get('trajectory_id')} "
f"轮次={tc.get('focus_turn')} 断言={json.dumps(tc.get('assertion'), ensure_ascii=False)}")
print(f" 说明: {tc.get('description')}")
# ---------- 3/4. 重放执行 + GitHub Issue（默认 mock） ----------
_replay_and_issues(problems, test_cases, do_github=do_github,
traj_path=_traj_path(data_dir), out_path=out_path,
create_issue=create_issue)
_hr("完成｜读轨迹 -> 诊断报告 -> 回归测试用例 -> (mock) GitHub Issue 全流程跑通")
def main():
parser = argparse.ArgumentParser(
description="实验 5-8：生产日志的智能诊断系统（读轨迹->诊断->回归测试->GitHub Issue）",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="示例：\n"
" python demo.py 完整流程（需 OPENAI_API_KEY）\n"
" python demo.py --smoke 免 API 快速自检（仅重放+GitHub mock）\n"
" python demo.py --model gpt-5.6 临时切换模型\n"
" python demo.py --data-dir ./mine 换用自己的轨迹/架构/PRD 目录\n"
" python demo.py --create-issue 经 MCP 真实创建 Issue（需 GITHUB_TOKEN+GITHUB_REPO）\n"
"换供应商：设置 OPENAI_BASE_URL + OPENAI_MODEL 环境变量。")
parser.add_argument("--smoke", action="store_true",
help="快速自检：跳过 LLM，用内置样例仅跑重放框架+GitHub mock（无需 API Key）")
parser.add_argument("--model", default=None,
help="临时覆盖模型（等价于设置 OPENAI_MODEL；默认 gpt-5.6-luna）")
parser.add_argument("--data-dir", default=DATA, metavar="DIR",
help="输入目录：轨迹日志 trajectories.jsonl + architecture.md + PRD.md（默认 data/）")
parser.add_argument("--output", default=DEFAULT_OUTPUT, metavar="FILE",
help="GitHub Issue（mock）落盘路径（默认 output/github_issues.json）")
parser.add_argument("--create-issue", action="store_true",
help="经 MCP 在真实仓库创建 Issue（需 GITHUB_TOKEN 与 GITHUB_REPO；默认 mock 不联网）")
parser.add_argument("--no-github", action="store_true",
help="跳过步骤 4（既不 mock 也不创建 GitHub Issue）")
args = parser.parse_args()
if args.smoke:
sys.exit(run_smoke(data_dir=args.data_dir, out_path=args.output))
run_full(model=args.model, do_github=not args.no_github,
data_dir=args.data_dir, out_path=args.output,
create_issue=args.create_issue)
if __name__ == "__main__":
main()
diagnoser.py¶
"""
diagnoser.py —— 诊断 Agent（真实调用 OpenAI）
两个阶段，均为真实 LLM 调用：
1) diagnose() 读轨迹集合 + 架构 + PRD -> 结构化问题报告(优先级/模块/描述/建议)
2) gen_test_cases() 基于问题报告 -> 生成可被 replay.py 自动执行的回归测试用例
默认模型 gpt-5.6-luna，输出走 JSON 模式，尽量稳定可解析。
"""
import json
import os
from typing import Dict, Any, List
from openai import OpenAI
MODEL = os.getenv("OPENAI_MODEL", "gpt-5.6-luna")
# --- 通用 OpenRouter 兜底 ---
OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
def _map_to_openrouter_model(model: str) -> str:
"""把直连模型名映射为 OpenRouter 上的 id（非可映射 id 统一兜底到当前廉价旗舰）。"""
if not model or "/" in model:
return model or "openai/gpt-5.6-luna"
m = model.lower()
if m.startswith(("gpt-", "o1", "o3", "o4")):
return "openai/" + model
if m.startswith("claude"):
if "haiku" in m:
return "anthropic/claude-haiku-4.5"
if "sonnet" in m:
return "anthropic/claude-sonnet-4.6"
return "anthropic/claude-opus-4.8"
if m.startswith("gemini"):
return "google/" + model
return "openai/gpt-5.6-luna"
# 供 LLM 生成测试用例时使用的断言 DSL 说明（须与 replay.py 保持一致）
_ASSERTION_SPEC = """可用断言类型（assertion.type 只能取以下之一）：
- "step_present" params: {"tool": <工具名>} 该工具必须在轨迹中出现
- "tool_succeeds" params: {"tool": <工具名>} 该工具最终成功且无"多次失败后误报成功"
- "latency_under" params: {"tool": <工具名>, "threshold_ms": <整数>} 该工具单次延迟须低于阈值
- "final_status_is" params: {"value": "success"|"failed"} 任务最终状态必须等于给定值"""
class Diagnoser:
def __init__(self, model: str = MODEL):
# 通用 OpenRouter 兜底：无直连 key，或默认 gpt-5.x（直连需组织实名认证）时改走 OpenRouter。
api_key = os.getenv("OPENAI_API_KEY")
base_url = os.getenv("OPENAI_BASE_URL")
orkey = os.getenv("OPENROUTER_API_KEY")
prefer_or = bool(orkey) and (model or "").lower().startswith("gpt-5")
if prefer_or or (not api_key and orkey):
api_key, base_url, model = orkey, OPENROUTER_BASE_URL, _map_to_openrouter_model(model)
# timeout / max_retries：让偶发的网络/SSL 抖动自动重试，不至于整轮崩溃
kw = {"timeout": 60.0, "max_retries": 3}
if api_key:
kw["api_key"] = api_key
if base_url:
kw["base_url"] = base_url
self.client = OpenAI(**kw)
self.model = model
# 推理模型（gpt-5 / o 系列等）不接受 temperature=0。
self._temp = (1 if any(k in (model or "").lower()
for k in ("gpt-5", "o1", "o3", "o4", "thinking", "reasoner", "kimi-k3"))
else 0)
# ---------- 阶段一：诊断 ----------
def diagnose(self, architecture: str, prd: str,
trajectories: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
traj_text = json.dumps(trajectories, ensure_ascii=False, indent=2)
system = (
"你是资深的 Agent 系统诊断专家。给定系统架构文档、PRD 与一组生产轨迹，"
"你要判断每条轨迹的执行流程是否符合架构与 PRD 的要求，识别问题模式、定位到具体模块，"
"输出结构化问题报告。只报告确有证据的问题，不要臆造。"
)
user = f"""# 系统架构
{architecture}
# PRD
{prd}
# 生产轨迹集合（JSON）
{traj_text}
# 任务
逐条核对轨迹与 PRD/架构，找出偏离项。以 JSON 输出，结构：
{{
"problems": [
{{
"title": "一句话问题标题",
"priority": "P0|P1|P2|P3",
"module": "涉及的模块名(取架构中的模块)",
"description": "问题描述，引用具体轨迹与轮次作为证据",
"suggestion": "可操作的改进建议",
"trajectory_ids": ["涉及的轨迹ID"],
"focus_turns": [关键交互轮次的index],
"prd_ref": "对应的PRD条目(如 R1/R2/R3)",
"suggested_assignee": "建议负责人(可留空)"
}}
]
}}
只输出 JSON。"""
resp = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "system", "content": system},
{"role": "user", "content": user}],
response_format={"type": "json_object"},
temperature=self._temp,
)
data = json.loads(resp.choices[0].message.content)
return data.get("problems", [])
# ---------- 阶段二：生成回归测试用例 ----------
def gen_test_cases(self, problems: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
prob_text = json.dumps(problems, ensure_ascii=False, indent=2)
system = (
"你是测试工程师。基于诊断出的问题，为每个问题生成一条回归测试用例，"
"用例引用问题轨迹 ID 与关键交互轮次，并给出一个可被自动重放框架求值的断言。"
)
user = f"""# 已诊断的问题
{prob_text}
# 断言 DSL
{_ASSERTION_SPEC}
# 任务
为每个问题生成 1 条回归测试用例。断言应表达"修复后系统应满足的正确行为"
（例如：退款前应出现 verify_refund_eligibility；process_refund 应最终成功；check_stock 延迟应 < 5000ms）。
以 JSON 输出：
{{
"test_cases": [
{{
"test_id": "RT-001",
"trajectory_id": "引用的问题轨迹ID",
"focus_turn": 关键轮次index,
"description": "该用例验证什么",
"assertion": {{"type": "...", "params": {{...}}}}
}}
]
}}
只输出 JSON。"""
resp = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "system", "content": system},
{"role": "user", "content": user}],
response_format={"type": "json_object"},
temperature=self._temp,
)
data = json.loads(resp.choices[0].message.content)
return data.get("test_cases", [])
github_mcp.py¶
"""
github_mcp.py —— GitHub Issue 创建（默认 mock，可选真实 MCP）
- mock（默认）：把"创建 Issue"渲染成将要提交的 Issue 结构，打印并写入本地文件，
不联网、不需要 token。
- 真实（mock=False，需 GITHUB_TOKEN + GITHUB_REPO）：通过 MCP 协议连接官方
GitHub MCP Server（stdio），调用其 `create_issue` 工具在真实仓库创建 Issue。
出于安全，真实创建须由 demo.py 的 --create-issue 显式开启。
"""
import json
import os
import shlex
from datetime import datetime
from typing import Dict, Any, List
_OUT = os.path.join(os.path.dirname(__file__), "output", "github_issues.json")
# 官方 GitHub MCP Server 的默认启动命令（可用 GITHUB_MCP_COMMAND 覆盖）。
# 默认用 Docker 运行官方镜像；也可换成任何暴露 create_issue 工具的 MCP Server。
_DEFAULT_MCP_COMMAND = (
"docker run -i --rm -e GITHUB_PERSONAL_ACCESS_TOKEN ghcr.io/github/github-mcp-server")
# 优先级 -> GitHub label 的映射
_PRIORITY_LABEL = {"P0": "priority:critical", "P1": "priority:high",
"P2": "priority:medium", "P3": "priority:low"}
def build_issue(problem: Dict[str, Any], test_cases: List[Dict[str, Any]]) -> Dict[str, Any]:
"""把一条诊断问题 + 关联回归测试用例，渲染成 GitHub Issue 结构。"""
prio = problem.get("priority", "P2")
module = problem.get("module", "unknown")
related = [tc for tc in test_cases
if tc.get("trajectory_id") in problem.get("trajectory_ids", [])]
body_lines = [
f"## 问题描述\n{problem.get('description', '')}",
f"\n## 涉及模块\n`{module}`",
f"\n## 优先级\n{prio}",
f"\n## 改进建议\n{problem.get('suggestion', '')}",
f"\n## 相关生产轨迹\n" + ", ".join(problem.get("trajectory_ids", []) or ["(无)"]),
]
if related:
body_lines.append("\n## 关联回归测试用例")
for tc in related:
body_lines.append(
f"- `{tc.get('test_id')}` (轨迹 {tc.get('trajectory_id')} "
f"第 {tc.get('focus_turn')} 轮): {tc.get('description', '')}")
return {
"title": f"[{prio}][{module}] {problem.get('title', problem.get('description', ''))[:60]}",
"body": "\n".join(body_lines),
"labels": [f"module:{module}", _PRIORITY_LABEL.get(prio, "priority:medium"),
"auto-diagnosis"],
"assignees": [problem.get("suggested_assignee", "")] if problem.get("suggested_assignee") else [],
}
def create_issues(problems: List[Dict[str, Any]], test_cases: List[Dict[str, Any]],
mock: bool = True, out_path: str = _OUT,
repo: str = None, token: str = None) -> List[Dict[str, Any]]:
"""为每条问题创建 Issue。
mock=True（默认）：打印 + 落盘到 out_path，不联网。
mock=False：通过 GitHub MCP Server 在 repo（owner/repo）真实创建 Issue，需 token。
"""
issues = [build_issue(p, test_cases) for p in problems]
if mock:
os.makedirs(os.path.dirname(out_path) or ".", exist_ok=True)
payload = {"created_at": datetime.now().isoformat(),
"mode": "mock", "issues": issues}
with open(out_path, "w", encoding="utf-8") as f:
json.dump(payload, f, ensure_ascii=False, indent=2)
print(f"\n[github_mcp:mock] 已将 {len(issues)} 个 Issue 写入 {out_path}")
for i, iss in enumerate(issues, 1):
print(f"\n----- Mock GitHub Issue #{i} -----")
print(f"title : {iss['title']}")
print(f"labels : {iss['labels']}")
print("body :")
for ln in iss["body"].splitlines():
print(" " + ln)
else:
if not token or not repo:
raise RuntimeError(
"真实创建需 GITHUB_TOKEN 与 GITHUB_REPO(owner/repo)，见 README。")
created = _create_issues_via_mcp(issues, repo=repo, token=token)
print(f"\n[github_mcp] 通过 MCP 在 {repo} 创建了 {len(created)} 个 Issue：")
for url in created:
print(f" {url}")
return issues
def _create_issues_via_mcp(issues: List[Dict[str, Any]], repo: str, token: str) -> List[str]:
"""通过 stdio 连接官方 GitHub MCP Server，逐个调用 create_issue 工具。
返回创建成功的 Issue URL 列表。需要已安装 `mcp` Python SDK 与可用的
GitHub MCP Server（默认 Docker 镜像，可用 GITHUB_MCP_COMMAND 覆盖启动命令）。
"""
import asyncio
try:
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
except ImportError as e: # pragma: no cover - 依赖缺失时给出清晰指引
raise RuntimeError(
"缺少 MCP 客户端：pip install mcp（并确保 GitHub MCP Server 可启动）") from e
owner, _, name = repo.partition("/")
if not owner or not name:
raise RuntimeError(f"GITHUB_REPO 需形如 owner/repo，收到：{repo!r}")
cmd = shlex.split(os.getenv("GITHUB_MCP_COMMAND", _DEFAULT_MCP_COMMAND))
params = StdioServerParameters(
command=cmd[0], args=cmd[1:],
env={**os.environ, "GITHUB_PERSONAL_ACCESS_TOKEN": token})
async def _run() -> List[str]:
urls: List[str] = []
async with stdio_client(params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
for iss in issues:
result = await session.call_tool("create_issue", {
"owner": owner, "repo": name,
"title": iss["title"], "body": iss["body"],
"labels": iss["labels"],
"assignees": iss["assignees"],
})
# MCP 工具返回文本内容；尽力提取 Issue URL，否则回退为原始文本。
text = "".join(getattr(c, "text", "") for c in result.content)
url = text
try:
url = json.loads(text).get("html_url", text)
except Exception:
pass
urls.append(url)
return urls
return asyncio.run(_run())
replay.py¶
"""
replay.py —— 回归测试重放框架
输入：Agent 生成的回归测试用例（结构化 JSON），引用轨迹 ID 与交互轮次。
过程：从原始轨迹取出该任务的输入，喂给被测系统（sut.run_task）重放，
在重放产生的新轨迹上求值断言，给出 通过/失败。
一个测试用例的结构（Agent 需按此 DSL 生成）：
{
"test_id": "RT-001",
"trajectory_id": "T-1001", # 引用的问题轨迹
"focus_turn": 3, # 关键交互轮次（问题所在）
"description": "退款前必须先做资格校验",
"assertion": {"type": "step_present", "params": {"tool": "verify_refund_eligibility"}}
}
支持的断言类型（replay 框架内置，可被自动求值）：
- step_present params.tool 某工具在轨迹中必须出现（如强制前置校验）
- tool_succeeds params.tool 某工具最终必须成功、且不得出现连续失败后误报成功
- latency_under params.tool, threshold_ms 某工具单次延迟必须低于阈值
- final_status_is params.value 任务最终状态必须等于给定值
"""
import json
import os
from typing import Dict, Any, List, Tuple
import sut
_DATA = os.path.join(os.path.dirname(__file__), "data", "trajectories.jsonl")
def load_trajectories(path: str = _DATA) -> Dict[str, Dict[str, Any]]:
"""读取生产轨迹集合，按 trajectory_id 索引。"""
out = {}
with open(path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
t = json.loads(line)
out[t["trajectory_id"]] = t
return out
# ---------------- 断言求值器 ----------------
def _tool_turns(traj: Dict[str, Any], tool: str) -> List[Dict[str, Any]]:
return [t for t in traj.get("turns", []) if t.get("tool") == tool]
def _eval_assertion(assertion: Dict[str, Any], traj: Dict[str, Any]) -> Tuple[bool, str]:
a_type = assertion.get("type")
params = assertion.get("params", {})
if a_type == "step_present":
tool = params.get("tool") or params.get("step")
ok = len(_tool_turns(traj, tool)) > 0
return ok, f"工具 {tool} {'出现' if ok else '缺失'}"
if a_type == "tool_succeeds":
tool = params.get("tool")
calls = _tool_turns(traj, tool)
if not calls:
return False, f"{tool} 未被调用"
n_err = sum(1 for c in calls if c.get("status") == "error")
last_ok = calls[-1].get("status") == "success"
# 修复标准：最终成功，且不存在"多次失败后仍误报成功"（>=2 次失败视为未正确处理）
ok = last_ok and n_err < 2
return ok, f"{tool} 调用 {len(calls)} 次, 失败 {n_err} 次, 末次{'成功' if last_ok else '失败'}"
if a_type == "latency_under":
tool = params.get("tool")
thr = params.get("threshold_ms") or params.get("threshold")
if thr is None:
return False, "latency_under 断言缺失阈值设置"
try:
thr = float(thr)
except (TypeError, ValueError):
return False, f"latency_under 阈值非法: {thr!r}"
calls = _tool_turns(traj, tool)
if not calls:
return False, f"{tool} 未被调用"
latencies = []
for c in calls:
lat = c.get("latency_ms")
if lat is None:
lat = 0
try:
latencies.append(float(lat))
except (TypeError, ValueError):
latencies.append(0.0)
worst = max(latencies) if latencies else 0.0
ok = worst < thr
return ok, f"{tool} 最大延迟 {worst}ms, 阈值 {thr}ms"
if a_type == "final_status_is":
want = params.get("value")
ok = traj.get("final_status") == want
return ok, f"final_status={traj.get('final_status')}, 期望={want}"
return False, f"未知断言类型: {a_type}"
def run_test_case(tc: Dict[str, Any], trajectories: Dict[str, Any],
fixed: bool) -> Dict[str, Any]:
"""对单条测试用例：取原始轨迹输入 -> 重放被测系统 -> 求值断言。"""
tid = tc.get("trajectory_id")
src = trajectories.get(tid)
if src is None:
return {"test_id": tc.get("test_id"), "passed": False,
"detail": f"引用的轨迹 {tid} 不存在"}
replayed = sut.run_task(src["task_input"], fixed=fixed)
passed, detail = _eval_assertion(tc.get("assertion", {}), replayed)
return {
"test_id": tc.get("test_id"),
"trajectory_id": tid,
"focus_turn": tc.get("focus_turn"),
"passed": passed,
"detail": detail,
"replay_mode": "fixed" if fixed else "buggy",
}
def run_suite(test_cases: List[Dict[str, Any]], fixed: bool,
path: str = _DATA) -> List[Dict[str, Any]]:
"""跑完整套测试用例，返回结果列表。"""
trajectories = load_trajectories(path)
results = []
for tc in test_cases:
try:
results.append(run_test_case(tc, trajectories, fixed))
except Exception as e: # 单条用例出错不影响整套
results.append({"test_id": tc.get("test_id"), "passed": False,
"detail": f"用例执行异常: {e}",
"replay_mode": "fixed" if fixed else "buggy"})
return results
sut.py¶
"""
sut.py —— System Under Test（被测系统的确定性仿真）
回归测试的核心是"用相同输入重放，断言修复后系统能产生正确行为"。
这里用一个**确定性**的仿真器来扮演线上 Agent 系统：
- run_task(task_input, fixed=False)：复现线上（有 bug）的行为，
产出的轨迹会带上和生产轨迹一致的三类已知问题。
- run_task(task_input, fixed=True)：模拟"修复后"的系统，
正确执行前置校验 / 重试退避 / 库存降级。
replay.py 会分别对 fixed=False / fixed=True 重放同一输入，
从而演示同一条回归测试用例的"失败(复现bug)"与"通过(验证修复)"。
轨迹结构与 data/trajectories.jsonl 完全一致，便于对比。
"""
from typing import Dict, Any
def run_task(task_input: Dict[str, Any], fixed: bool = False) -> Dict[str, Any]:
"""给定任务输入，确定性地跑一遍被测系统，返回一条轨迹。"""
intent = task_input.get("intent")
order_id = task_input.get("order_id", "UNKNOWN")
turns = []
idx = 0
def add(**kw):
nonlocal idx
kw["index"] = idx
idx += 1
turns.append(kw)
# 0. 用户输入 & 意图识别
add(role="user", content=f"task={intent}, order={order_id}")
add(role="assistant", module="intent_parser", content=f"意图={intent}")
final_status = "success"
if intent == "refund":
# 查询订单
add(role="tool", module="order_service", tool="query_order",
input={"order_id": order_id},
output={"status": task_input.get("order_status", "paid")},
status="success", latency_ms=210)
# R1：退款前置资格校验（仅修复版本执行）
if fixed:
add(role="tool", module="order_service", tool="verify_refund_eligibility",
input={"order_id": order_id},
output={"eligible": True}, status="success", latency_ms=120)
# R2：支付重试 + 退避
if task_input.get("payment_flaky") and not fixed:
# 线上 bug：无退避，连续失败后误报成功
for _ in range(3):
add(role="tool", module="payment_service", tool="process_refund",
input={"order_id": order_id},
output={"error": "gateway_timeout"},
status="error", latency_ms=3000)
add(role="assistant", module="payment_service",
content="多次失败，仍按成功结束（bug）")
final_status = "success" # 误报成功
elif task_input.get("payment_flaky") and fixed:
# 修复：一次失败后带退避重试成功
add(role="tool", module="payment_service", tool="process_refund",
input={"order_id": order_id},
output={"error": "gateway_timeout"}, status="error", latency_ms=1500)
add(role="assistant", module="payment_service", content="退避 800ms 后重试")
add(role="tool", module="payment_service", tool="process_refund",
input={"order_id": order_id, "retry": 1},
output={"refund_id": "R-OK"}, status="success", latency_ms=600)
else:
add(role="tool", module="payment_service", tool="process_refund",
input={"order_id": order_id},
output={"refund_id": "R-OK"}, status="success", latency_ms=540)
elif intent == "order_status":
add(role="tool", module="order_service", tool="query_order",
input={"order_id": order_id},
output={"status": "paid", "sku": task_input.get("sku")},
status="success", latency_ms=220)
# R3：库存查询延迟
if task_input.get("slow_inventory") and not fixed:
# 线上 bug：超时仍阻塞等待，不降级
add(role="tool", module="inventory_service", tool="check_stock",
input={"sku": task_input.get("sku")},
output={"stock": 12}, status="success", latency_ms=8300)
elif task_input.get("slow_inventory") and fixed:
# 修复：超过阈值走降级路径，快速返回
add(role="tool", module="inventory_service", tool="check_stock",
input={"sku": task_input.get("sku"), "degraded": True},
output={"stock": "cached:12", "degraded": True},
status="success", latency_ms=400)
else:
add(role="tool", module="inventory_service", tool="check_stock",
input={"sku": task_input.get("sku")},
output={"stock": 5}, status="success", latency_ms=300)
# R4：通知用户
add(role="tool", module="notification_service", tool="notify_user",
input={"final_status": final_status},
output={"sent": True}, status="success", latency_ms=60)
return {
"trajectory_id": f"REPLAY::{order_id}::{'fixed' if fixed else 'buggy'}",
"task_input": task_input,
"final_status": final_status,
"turns": turns,
}
test_github_mcp_bare_output.py¶
import json
import os
import github_mcp
def test_create_issues_bare_filename(tmp_path, monkeypatch):
"""out_path 为不带目录的裸文件名时，落盘不应崩溃（dirname 为空串）。"""
monkeypatch.chdir(tmp_path)
problems = [{"priority": "P1", "module": "refund",
"title": "退款未校验", "description": "描述"}]
issues = github_mcp.create_issues(problems, [], mock=True,
out_path="issues.json")
assert os.path.exists("issues.json")
with open("issues.json", encoding="utf-8") as f:
payload = json.load(f)
assert payload["issues"] == issues
def test_create_issues_with_directory(tmp_path):
"""带目录的 out_path 保持原行为：自动创建父目录。"""
out = tmp_path / "sub" / "issues.json"
github_mcp.create_issues([], [], mock=True, out_path=str(out))
assert out.exists()
test_replay_latency.py¶
from replay import _eval_assertion
def test_latency_under_null_latency_ms():
assertion = {
"type": "latency_under",
"params": {"tool": "test_tool", "threshold": 150},
}
traj = {
"turns": [
{"tool": "test_tool", "latency_ms": None},
{"tool": "test_tool", "latency_ms": 120},
]
}
ok, msg = _eval_assertion(assertion, traj)
assert ok
assert "最大延迟 120.0ms" in msg
def test_latency_under_missing_threshold():
assertion = {"type": "latency_under", "params": {"tool": "test_tool"}}
traj = {"turns": [{"tool": "test_tool", "latency_ms": 100}]}
ok, msg = _eval_assertion(assertion, traj)
assert not ok
assert "断言缺失阈值设置" in msg
data/trajectories.jsonl¶
{"trajectory_id": "T-1001", "task": "用户要求对订单 A1001 退款", "task_input": {"intent": "refund", "order_id": "A1001", "order_status": "paid", "payment_flaky": false, "slow_inventory": false}, "final_status": "success", "turns": [{"index": 0, "role": "user", "content": "我要退款，订单号 A1001"}, {"index": 1, "role": "assistant", "module": "intent_parser", "content": "意图=refund, order_id=A1001"}, {"index": 2, "role": "tool", "module": "order_service", "tool": "query_order", "input": {"order_id": "A1001"}, "output": {"status": "paid", "amount": 199}, "status": "success", "latency_ms": 210}, {"index": 3, "role": "assistant", "module": "order_service", "content": "订单已支付，直接发起退款", "note": "未调用 verify_refund_eligibility"}, {"index": 4, "role": "tool", "module": "payment_service", "tool": "process_refund", "input": {"order_id": "A1001", "amount": 199}, "output": {"refund_id": "R-9001"}, "status": "success", "latency_ms": 540}, {"index": 5, "role": "tool", "module": "notification_service", "tool": "notify_user", "input": {"msg": "退款成功"}, "output": {"sent": true}, "status": "success", "latency_ms": 60}]}
{"trajectory_id": "T-1002", "task": "用户要求对订单 A1002 退款", "task_input": {"intent": "refund", "order_id": "A1002", "order_status": "paid", "payment_flaky": true, "slow_inventory": false}, "final_status": "success", "turns": [{"index": 0, "role": "user", "content": "订单 A1002 退款"}, {"index": 1, "role": "assistant", "module": "intent_parser", "content": "意图=refund, order_id=A1002"}, {"index": 2, "role": "tool", "module": "order_service", "tool": "query_order", "input": {"order_id": "A1002"}, "output": {"status": "paid", "amount": 88}, "status": "success", "latency_ms": 230}, {"index": 3, "role": "assistant", "module": "order_service", "content": "订单已支付，直接发起退款", "note": "未调用 verify_refund_eligibility"}, {"index": 4, "role": "tool", "module": "payment_service", "tool": "process_refund", "input": {"order_id": "A1002", "amount": 88}, "output": {"error": "gateway_timeout"}, "status": "error", "latency_ms": 3010}, {"index": 5, "role": "tool", "module": "payment_service", "tool": "process_refund", "input": {"order_id": "A1002", "amount": 88}, "output": {"error": "gateway_timeout"}, "status": "error", "latency_ms": 3005}, {"index": 6, "role": "tool", "module": "payment_service", "tool": "process_refund", "input": {"order_id": "A1002", "amount": 88}, "output": {"error": "gateway_timeout"}, "status": "error", "latency_ms": 3020}, {"index": 7, "role": "assistant", "module": "payment_service", "content": "重试多次失败，按成功结束", "note": "无退避；最终误报成功"}, {"index": 8, "role": "tool", "module": "notification_service", "tool": "notify_user", "input": {"msg": "退款成功"}, "output": {"sent": true}, "status": "success", "latency_ms": 55}]}
{"trajectory_id": "T-1003", "task": "用户查询订单 A1003 的库存/发货状态", "task_input": {"intent": "order_status", "order_id": "A1003", "sku": "SKU-77", "payment_flaky": false, "slow_inventory": true}, "final_status": "success", "turns": [{"index": 0, "role": "user", "content": "A1003 还有货吗，什么时候发"}, {"index": 1, "role": "assistant", "module": "intent_parser", "content": "意图=order_status, order_id=A1003"}, {"index": 2, "role": "tool", "module": "order_service", "tool": "query_order", "input": {"order_id": "A1003"}, "output": {"status": "paid", "sku": "SKU-77"}, "status": "success", "latency_ms": 240}, {"index": 3, "role": "tool", "module": "inventory_service", "tool": "check_stock", "input": {"sku": "SKU-77"}, "output": {"stock": 12}, "status": "success", "latency_ms": 8300, "note": "延迟超过 5000ms 阈值，未降级"}, {"index": 4, "role": "tool", "module": "notification_service", "tool": "notify_user", "input": {"msg": "有货，1天内发货"}, "output": {"sent": true}, "status": "success", "latency_ms": 70}]}
{"trajectory_id": "T-1004", "task": "用户查询订单 A1004 状态（正常轨迹，对照组）", "task_input": {"intent": "order_status", "order_id": "A1004", "sku": "SKU-12", "payment_flaky": false, "slow_inventory": false}, "final_status": "success", "turns": [{"index": 0, "role": "user", "content": "A1004 到哪了"}, {"index": 1, "role": "assistant", "module": "intent_parser", "content": "意图=order_status, order_id=A1004"}, {"index": 2, "role": "tool", "module": "order_service", "tool": "query_order", "input": {"order_id": "A1004"}, "output": {"status": "shipped", "sku": "SKU-12"}, "status": "success", "latency_ms": 200}, {"index": 3, "role": "tool", "module": "inventory_service", "tool": "check_stock", "input": {"sku": "SKU-12"}, "output": {"stock": 5}, "status": "success", "latency_ms": 300}, {"index": 4, "role": "tool", "module": "notification_service", "tool": "notify_user", "input": {"msg": "已发货"}, "output": {"sent": true}, "status": "success", "latency_ms": 65}]}