liqiang
b119135836
Build latest book artifacts / build (push) Canceled after 0s
dependency resolution / resolve (3.11) (push) Canceled after 0s
dependency resolution / resolve (3.13) (push) Canceled after 0s
deploy-pages / build (push) Canceled after 0s
deploy-pages / deploy (push) Canceled after 0s
i18n consistency check / check (push) Canceled after 0s
provider adoption tests / test (chapter2/context-compression) (push) Canceled after 0s
provider adoption tests / test (chapter2/prompt-injection) (push) Canceled after 0s
provider adoption tests / test (chapter2/system-hint) (push) Canceled after 0s
provider adoption tests / test (chapter3/log-sanitization) (push) Canceled after 0s
web-search-agent tests / test (push) Canceled after 0s
web-search-agent tests / agentbook (push) Canceled after 0s
219 lines
11 KiB
Python
219 lines
11 KiB
Python
"""
|
||
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()
|