CI/CD 整合與持續改進：讓 Harness 自動跑
從 GitHub Actions 整合、門檻設定到生產環境回饋與測試集演化，建立完整的自動化持續改進系統。
CI/CD 整合與持續改進：讓 Harness 自動跑
上一篇我們談了回歸測試與失敗分析，理解了如何檢測退步、分類失敗、找出根因。
但這些能力如果只靠手動執行，就無法持續發揮價值。
真正的 Harness，應該整合進 CI/CD 流程，每次修改都自動跑測試、自動比較、自動阻止退步。
這一篇，我們要談 Harness 的最後一哩路：CI/CD 整合與持續改進。
我們會從 GitHub Actions 整合、自動化流程設計、門檻設定，一路談到生產環境回饋與測試集演化，並用 Python 實作一個完整的自動化系統。
一、為什麼 Harness 要整合 CI/CD？
手動執行的問題
如果 Harness 只能手動執行，會遇到這些問題：
- 忘記跑：修改了 Prompt，但忘了跑測試就發布。
- 不一致：每個人跑的測試集不同，結果無法比較。
- 不及時：等到發布後才發現問題，已經影響使用者。
- 無法阻止：即使發現退步，也沒有機制阻止合併。
- 無法追蹤：沒有歷史記錄，不知道問題是什麼時候引入的。
CI/CD 整合的價值
把 Harness 整合進 CI/CD，可以：
| 價值 | 說明 |
|---|---|
| 自動化 | 每次提交都自動跑測試 |
| 一致性 | 所有人都跑同一套測試 |
| 及時性 | 在合併前就發現問題 |
| 阻止退步 | 未通過測試就阻止合併 |
| 可追蹤 | 每次執行都有記錄 |
| 持續改進 | 生產回饋自動加入測試集 |
完整的 CI/CD 流程
開發者提交程式碼
↓
觸發 CI Pipeline
↓
┌─────────────────────────────┐
│ 1. 單元測試 │
│ 2. Harness 測試 │
│ 3. 回歸檢測 │
│ 4. 產生報告 │
│ 5. 判斷是否通過 │
└─────────────────────────────┘
↓
通過 → 允許合併 → 部署
↓
不通過 → 阻止合併 → 通知開發者
↓
生產環境運行
↓
收集失敗案例 → 加入測試集 → 回到 CI二、GitHub Actions 整合
GitHub Actions 是最常見的 CI/CD 工具。讓我們從這裡開始。
基本工作流程
# .github/workflows/harness.yml
name: Agent Harness
on:
pull_request:
branches: [main]
push:
branches: [main]
jobs:
test:
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Setup Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install dependencies
run: |
pip install -r requirements.txt
- name: Run unit tests
run: |
pytest tests/unit/ -v
- name: Run Harness
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: |
python run_harness.py \
--test-suite default \
--baseline v1.0.0 \
--output reports/harness_report.json
- name: Check for regressions
run: |
python check_regressions.py \
--report reports/harness_report.json \
--threshold 0.1
- name: Upload report
if: always()
uses: actions/upload-artifact@v4
with:
name: harness-report
path: reports/
- name: Comment on PR
if: github.event_name == 'pull_request'
uses: actions/github-script@v7
with:
script: |
const fs = require('fs');
const report = JSON.parse(fs.readFileSync('reports/harness_report.json'));
const body = formatReport(report);
github.rest.issues.createComment({
issue_number: context.issue.number,
owner: context.repo.owner,
repo: context.repo.repo,
body: body
});Harness 執行腳本
# run_harness.py
import argparse
import json
import sys
from datetime import datetime
from harness import Harness
from agent import CompleteAgent
from test_case_manager import TestCaseManager
from baseline_manager import BaselineManager
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--test-suite", default="default")
parser.add_argument("--baseline", default=None)
parser.add_argument("--output", default="harness_report.json")
parser.add_argument("--parallel", action="store_true")
parser.add_argument("--max-workers", type=int, default=4)
args = parser.parse_args()
# 載入測試案例
case_manager = TestCaseManager("test_cases")
case_manager.load(args.test_suite)
test_cases = case_manager.list_all()
print(f"載入 {len(test_cases)} 個測試案例")
# 建立 Agent
agent = CompleteAgent(user_id="ci_test")
# 建立 Harness
harness = Harness(
agent=agent,
parallel=args.parallel,
max_workers=args.max_workers,
verbose=True,
)
# 執行測試
report = harness.run(test_cases)
# 如果有基準線，做回歸檢測
if args.baseline:
baseline_manager = BaselineManager("baselines")
baseline = baseline_manager.load(args.baseline)
from regression import RegressionDetector
detector = RegressionDetector(threshold=0.1)
diff = detector.detect(
baseline.results,
report["details"],
test_cases,
)
report["regression"] = diff
report["baseline_version"] = args.baseline
# 加入 metadata
report["metadata"] = {
"timestamp": datetime.now().isoformat(),
"test_suite": args.test_suite,
"total_cases": len(test_cases),
"commit_sha": get_commit_sha(),
"branch": get_branch(),
}
# 儲存報告
with open(args.output, "w", encoding="utf-8") as f:
json.dump(report, f, ensure_ascii=False, indent=2)
print(f"\n報告已儲存至 {args.output}")
# 判斷是否通過
if not check_pass(report):
sys.exit(1)
def get_commit_sha() -> str:
import subprocess
try:
return subprocess.check_output(
["git", "rev-parse", "HEAD"]
).decode().strip()[:8]
except Exception:
return "unknown"
def get_branch() -> str:
import subprocess
try:
return subprocess.check_output(
["git", "rev-parse", "--abbrev-ref", "HEAD"]
).decode().strip()
except Exception:
return "unknown"
def check_pass(report: dict) -> bool:
"""判斷是否通過"""
if report["summary"]["pass_rate"] < 0.8:
print(f"通過率過低：{report['summary']['pass_rate']}")
return False
if "regression" in report:
reg = report["regression"]
if reg["new_failure_count"] > 0:
print(f"發現 {reg['new_failure_count']} 個新增失敗")
return False
critical = [
r for r in reg["regressions"]
if r["severity"] == "critical"
]
if critical:
print(f"發現 {len(critical)} 個 critical 退步")
return False
print("所有檢查通過")
return True
if __name__ == "__main__":
main()回歸檢查腳本
# check_regressions.py
import argparse
import json
import sys
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--report", required=True)
parser.add_argument("--threshold", type=float, default=0.1)
parser.add_argument("--fail-on-regression", action="store_true", default=True)
args = parser.parse_args()
with open(args.report) as f:
report = json.load(f)
if "regression" not in report:
print("沒有回歸資料，跳過檢查")
return
reg = report["regression"]
print(f"退步數量：{reg['regression_count']}")
print(f"新增失敗：{reg['new_failure_count']}")
print(f"改進數量：{reg['improvement_count']}")
if reg["new_failure_count"] > 0:
print("\n新增失敗案例：")
for failure in reg["new_failures"]:
print(f" - {failure['case_id']}: "
f"{failure['baseline_score']} → {failure['current_score']}")
sys.exit(1)
critical = [
r for r in reg["regressions"]
if r["severity"] in ("critical", "high")
]
if critical:
print(f"\n嚴重退步（{len(critical)} 個）：")
for r in critical:
print(f" [{r['severity']}] {r['case_id']}: {r['delta']}")
sys.exit(1)
print("\n沒有嚴重退步")
if __name__ == "__main__":
main()PR 留言格式化
// scripts/format_report.js
function formatReport(report) {
const summary = report.summary;
const regression = report.regression;
let body = '## 🤖 Agent Harness 報告\n\n';
// 摘要
body += '### 摘要\n\n';
body += '| 指標 | 數值 |\n';
body += '|------|------|\n';
body += `| 總案例數 | ${summary.total} |\n`;
body += `| 通過 | ${summary.passed} |\n`;
body += `| 失敗 | ${summary.failed} |\n`;
body += `| 通過率 | ${(summary.pass_rate * 100).toFixed(1)}% |\n`;
body += `| 平均分數 | ${summary.avg_score.toFixed(3)} |\n`;
body += `| 平均延遲 | ${summary.avg_latency.toFixed(2)}s |\n`;
// 回歸
if (regression) {
body += '\n### 回歸檢測\n\n';
body += '| 指標 | 數值 |\n';
body += '|------|------|\n';
body += `| 退步 | ${regression.regression_count} |\n`;
body += `| 改進 | ${regression.improvement_count} |\n`;
body += `| 新增失敗 | ${regression.new_failure_count} |\n`;
if (regression.regressions.length > 0) {
body += '\n### 退步案例\n\n';
body += '| 案例 | 嚴重程度 | 基準線 | 當前 | 差異 |\n';
body += '|------|----------|--------|------|------|\n';
for (const reg of regression.regressions.slice(0, 10)) {
body += `| ${reg.case_id} | ${reg.severity} | `;
body += `${reg.baseline_score.toFixed(3)} | `;
body += `${reg.current_score.toFixed(3)} | `;
body += `${reg.delta.toFixed(3)} |\n`;
}
}
}
// 失敗案例
if (report.failures && report.failures.length > 0) {
body += '\n### 失敗案例\n\n';
for (const failure of report.failures.slice(0, 5)) {
body += `- **${failure.case_id}**：${failure.score.toFixed(3)}\n`;
if (failure.error) {
body += ` - 錯誤：${failure.error}\n`;
}
}
}
// 建議
body += '\n---\n';
if (summary.pass_rate >= 0.9 && (!regression || regression.new_failure_count === 0)) {
body += '**建議：可以合併**\n';
} else {
body += '**建議：請修正後再合併**\n';
}
return body;
}三、門檻設定：什麼時候該阻止合併？
不是所有退步都需要阻止合併。我們需要設定合理的門檻。
門檻的類型
| 門檻 | 說明 | 預設值 |
|---|---|---|
| 通過率門檻 | 整體通過率的最低標準 | 80% |
| 新增失敗門檻 | 允許的新增失敗數 | 0 |
| 嚴重退步門檻 | 允許的 critical/high 退步數 | 0 |
| 分數下降門檻 | 單一案例允許的分數下降 | 0.1 |
| 延遲門檻 | 允許的平均延遲增加 | 20% |
| 成本門檻 | 允許的平均成本增加 | 30% |
門檻設定策略
不同類型的專案，有不同的門檻策略：
@dataclass
class GatePolicy:
"""門檻政策"""
name: str
# 通過率
min_pass_rate: float = 0.8
# 新增失敗
max_new_failures: int = 0
# 嚴重退步
max_critical_regressions: int = 0
max_high_regressions: int = 2
# 分數
max_score_drop: float = 0.1
max_avg_score_drop: float = 0.05
# 效率
max_latency_increase: float = 0.2
max_cost_increase: float = 0.3
# 對抗案例
adversarial_must_pass: bool = True
# 嚴格政策（生產環境）
STRICT_POLICY = GatePolicy(
name="strict",
min_pass_rate=0.95,
max_new_failures=0,
max_critical_regressions=0,
max_high_regressions=0,
max_avg_score_drop=0.02,
adversarial_must_pass=True,
)
# 標準政策（一般開發）
STANDARD_POLICY = GatePolicy(
name="standard",
min_pass_rate=0.85,
max_new_failures=0,
max_critical_regressions=0,
max_high_regressions=1,
max_avg_score_drop=0.05,
)
# 寬鬆政策（實驗階段）
LENIENT_POLICY = GatePolicy(
name="lenient",
min_pass_rate=0.7,
max_new_failures=2,
max_critical_regressions=1,
max_high_regressions=3,
max_avg_score_drop=0.1,
)門檻檢查器
class GateChecker:
"""門檻檢查器"""
def __init__(self, policy: GatePolicy):
self.policy = policy
def check(self, report: dict) -> dict:
"""檢查是否通過門檻"""
violations = []
# 1. 通過率
pass_rate = report["summary"]["pass_rate"]
if pass_rate < self.policy.min_pass_rate:
violations.append({
"rule": "min_pass_rate",
"expected": f">= {self.policy.min_pass_rate}",
"actual": pass_rate,
"severity": "high",
})
# 2. 回歸檢查
if "regression" in report:
reg = report["regression"]
if reg["new_failure_count"] > self.policy.max_new_failures:
violations.append({
"rule": "max_new_failures",
"expected": f"<= {self.policy.max_new_failures}",
"actual": reg["new_failure_count"],
"severity": "critical",
"details": reg["new_failures"],
})
critical = sum(
1 for r in reg["regressions"]
if r["severity"] == "critical"
)
high = sum(
1 for r in reg["regressions"]
if r["severity"] == "high"
)
if critical > self.policy.max_critical_regressions:
violations.append({
"rule": "max_critical_regressions",
"expected": f"<= {self.policy.max_critical_regressions}",
"actual": critical,
"severity": "critical",
})
if high > self.policy.max_high_regressions:
violations.append({
"rule": "max_high_regressions",
"expected": f"<= {self.policy.max_high_regressions}",
"actual": high,
"severity": "high",
})
if "summary" in reg:
avg_drop = reg["summary"].get("net_change", 0)
if avg_drop < -self.policy.max_avg_score_drop:
violations.append({
"rule": "max_avg_score_drop",
"expected": f">= -{self.policy.max_avg_score_drop}",
"actual": avg_drop,
"severity": "high",
})
# 3. 對抗案例
if self.policy.adversarial_must_pass:
adversarial_failures = self._check_adversarial(report)
if adversarial_failures:
violations.append({
"rule": "adversarial_must_pass",
"expected": "全部通過",
"actual": f"{len(adversarial_failures)} 個失敗",
"severity": "critical",
"details": adversarial_failures,
})
return {
"passed": len(violations) == 0,
"violations": violations,
"policy": self.policy.name,
"recommendation": self._recommend(violations),
}
def _check_adversarial(self, report: dict) -> list:
"""檢查對抗案例"""
failures = []
for detail in report.get("details", []):
if detail.get("is_adversarial") and not detail.get("passed"):
failures.append(detail)
return failures
def _recommend(self, violations: list) -> str:
"""產生建議"""
if not violations:
return "所有檢查通過，可以合併"
critical = [v for v in violations if v["severity"] == "critical"]
if critical:
return f"發現 {len(critical)} 個關鍵問題，必須修復後才能合併"
return f"發現 {len(violations)} 個問題，建議修復後再合併"使用範例
# 載入報告
with open("reports/harness_report.json") as f:
report = json.load(f)
# 選擇政策
policy = STRICT_POLICY # 或 STANDARD_POLICY
# 檢查
checker = GateChecker(policy)
result = checker.check(report)
print(f"政策：{result['policy']}")
print(f"通過：{result['passed']}")
print(f"建議：{result['recommendation']}")
if result["violations"]:
print("\n違規項目：")
for v in result["violations"]:
print(f" [{v['severity']}] {v['rule']}: "
f"預期 {v['expected']}，實際 {v['actual']}")
if not result["passed"]:
sys.exit(1)四、生產環境回饋：從真實使用者學習
CI/CD 中的測試是「離線」的。
真正的價值來自生產環境的回饋。
回饋的來源
| 來源 | 說明 | 收集方式 |
|---|---|---|
| 失敗案例 | Agent 執行失敗的案例 | 自動記錄 |
| 低評分回饋 | 使用者評分低的案例 | 評分系統 |
| 投訴 | 使用者主動投訴 | 客服系統 |
| 異常行為 | 延遲、成本異常 | 監控系統 |
| 邊界探索 | 使用者輸入的新模式 | 輸入分析 |
生產環境追蹤器
class ProductionTracker:
"""生產環境追蹤器"""
def __init__(self, storage_path: str = "production_feedback"):
self.storage_path = storage_path
os.makedirs(storage_path, exist_ok=True)
self.buffer = []
def track_invocation(
self,
user_input: str,
output: str,
trace: dict = None,
success: bool = True,
error: str = None,
metadata: dict = None,
):
"""追蹤一次呼叫"""
record = {
"timestamp": datetime.now().isoformat(),
"user_input": user_input,
"output": output[:500] if output else "",
"success": success,
"error": error,
"trace_id": trace.get("trace_id") if trace else None,
"duration": trace.get("total_duration") if trace else None,
"metadata": metadata or {},
}
self.buffer.append(record)
if len(self.buffer) >= 100:
self.flush()
def track_feedback(
self,
trace_id: str,
rating: int,
comment: str = "",
):
"""追蹤使用者回饋"""
record = {
"timestamp": datetime.now().isoformat(),
"type": "feedback",
"trace_id": trace_id,
"rating": rating,
"comment": comment,
}
self.buffer.append(record)
def flush(self):
"""寫入緩衝區"""
if not self.buffer:
return
date = datetime.now().strftime("%Y%m%d")
path = os.path.join(self.storage_path, f"{date}.jsonl")
with open(path, "a", encoding="utf-8") as f:
for record in self.buffer:
f.write(json.dumps(record, ensure_ascii=False) + "\n")
self.buffer = []
def collect_failures(self, days: int = 7) -> list:
"""收集最近 N 天的失敗案例"""
from datetime import timedelta
cutoff = datetime.now() - timedelta(days=days)
failures = []
for filename in os.listdir(self.storage_path):
if not filename.endswith(".jsonl"):
continue
path = os.path.join(self.storage_path, filename)
with open(path, "r", encoding="utf-8") as f:
for line in f:
try:
record = json.loads(line)
record_time = datetime.fromisoformat(record["timestamp"])
if record_time < cutoff:
continue
if record.get("success") is False:
failures.append(record)
if record.get("type") == "feedback" and record.get("rating", 5) <= 2:
failures.append(record)
except (json.JSONDecodeError, KeyError):
continue
return failures
def collect_low_rating(self, days: int = 7) -> list:
"""收集低評分案例"""
return [
f for f in self.collect_failures(days)
if f.get("type") == "feedback" and f.get("rating", 5) <= 2
]自動標註
收集到生產案例後，需要自動標註，才能加入測試集。
class AutoLabeler:
"""自動標註器"""
def __init__(self, llm_client):
self.llm = llm_client
def label_failure(self, record: dict) -> dict:
"""為失敗案例自動標註"""
user_input = record.get("user_input", "")
error = record.get("error", "")
prompt = f"""請分析以下 Agent 失敗案例，並標註：
使用者輸入：{user_input}
錯誤訊息：{error}
輸出：{record.get('output', '')}
請以 JSON 格式輸出：
{{
"category": "execution/output/behavior/safety",
"subtype": "具體問題類型",
"severity": "critical/high/medium/low",
"expected_behavior": "期望的行為",
"tags": ["標籤1", "標籤2"]
}}
只輸出 JSON。"""
try:
response = self.llm([{"role": "user", "content": prompt}])
content = response.choices[0].message.content
import re
json_match = re.search(r"\{.*\}", content, re.DOTALL)
if json_match:
return json.loads(json_match.group(0))
except Exception:
pass
return {
"category": "unknown",
"subtype": "unknown",
"severity": "medium",
"expected_behavior": "正確處理此輸入",
"tags": ["production"],
}
def label_batch(self, records: list) -> list:
"""批次標註"""
labeled = []
for record in records:
label = self.label_failure(record)
labeled.append({**record, "label": label})
return labeled加入測試集
class TestCaseEvolver:
"""測試集演化器"""
def __init__(self, case_manager, auto_labeler: AutoLabeler):
self.case_manager = case_manager
self.labeler = auto_labeler
def evolve_from_production(
self,
days: int = 7,
min_severity: str = "medium",
max_cases: int = 20,
) -> dict:
"""從生產環境演化測試集"""
tracker = ProductionTracker()
failures = tracker.collect_failures(days)
severity_order = {"critical": 0, "high": 1, "medium": 2, "low": 3}
min_level = severity_order.get(min_severity, 2)
labeled = self.labeler.label_batch(failures)
filtered = [
f for f in labeled
if severity_order.get(f["label"]["severity"], 3) <= min_level
]
filtered.sort(key=lambda f: severity_order.get(f["label"]["severity"], 3))
filtered = filtered[:max_cases]
added = []
for record in filtered:
case_id = f"prod_{record.get('trace_id', 'unknown')[:8]}"
if self.case_manager.get(case_id):
continue
from test_case import TestCase
case = TestCase(
id=case_id,
name=f"生產案例：{record['user_input'][:30]}",
input={"query": record["user_input"]},
expected_behavior=record["label"]["expected_behavior"],
tags=record["label"]["tags"] + ["production"],
category=record["label"]["category"],
severity=record["label"]["severity"],
metadata={
"source": "production",
"discovered_at": record["timestamp"],
"original_error": record.get("error"),
},
)
self.case_manager.add(case)
added.append(case)
if added:
self.case_manager.save("default")
return {
"collected": len(failures),
"filtered": len(filtered),
"added": len(added),
"new_case_ids": [c.id for c in added],
}使用範例
evolver = TestCaseEvolver(
case_manager=TestCaseManager("test_cases"),
auto_labeler=AutoLabeler(llm_client=call_llm),
)
result = evolver.evolve_from_production(
days=7,
min_severity="medium",
max_cases=20,
)
print(f"收集到 {result['collected']} 個失敗案例")
print(f"過濾後 {result['filtered']} 個")
print(f"新增 {result['added']} 個測試案例")五、持續改進的完整迴圈
把 CI/CD、生產回饋、測試集演化整合起來，形成一個完整的持續改進迴圈。
┌─────────────────────────────────────────────────────────┐
│ │
│ 1. 開發者修改 Prompt / Skill / Agent │
│ ↓ │
│ 2. 提交到 Git │
│ ↓ │
│ 3. CI 觸發 Harness │
│ ↓ │
│ 4. 跑測試集、產生報告 │
│ ↓ │
│ 5. 回歸檢測、門檻檢查 │
│ ↓ │
│ 6. 通過 → 合併 → 部署 │
│ 不通過 → 阻止合併 → 通知開發者 │
│ ↓ │
│ 7. 生產環境運行 │
│ ↓ │
│ 8. 收集失敗案例、使用者回饋 │
│ ↓ │
│ 9. 自動標註、加入測試集 │
│ ↓ │
│ 10. 回到步驟 1，持續改進 │
│ │
└─────────────────────────────────────────────────────────┘完整實作
class ContinuousImprovementSystem:
"""持續改進系統"""
def __init__(
self,
harness: Harness,
case_manager: TestCaseManager,
baseline_manager: BaselineManager,
gate_checker: GateChecker,
tracker: ProductionTracker = None,
evolver: TestCaseEvolver = None,
):
self.harness = harness
self.case_manager = case_manager
self.baseline_manager = baseline_manager
self.gate_checker = gate_checker
self.tracker = tracker or ProductionTracker()
self.evolver = evolver
def run_ci(self, baseline_version: str = None) -> dict:
"""執行 CI 流程"""
test_cases = self.case_manager.list_all()
report = self.harness.run(test_cases)
if baseline_version:
baseline = self.baseline_manager.load(baseline_version)
from regression import RegressionDetector
detector = RegressionDetector(threshold=0.1)
diff = detector.detect(
baseline.results,
report["details"],
test_cases,
)
report["regression"] = diff
gate_result = self.gate_checker.check(report)
report["gate"] = gate_result
return report
def run_production_feedback(self, days: int = 7) -> dict:
"""執行生產回饋流程"""
if not self.evolver:
return {"error": "未設定 evolver"}
failures = self.tracker.collect_failures(days)
labeled = self.evolver.labeler.label_batch(failures)
result = self.evolver.evolve_from_production(
days=days,
min_severity="medium",
)
return {
"collected": len(failures),
"added_to_test_suite": result["added"],
"new_cases": result["new_case_ids"],
}
def weekly_cycle(self) -> dict:
"""每週迴圈"""
feedback_result = self.run_production_feedback(days=7)
test_cases = self.case_manager.list_all()
new_baseline = self.baseline_manager.create_baseline(
version=f"weekly_{datetime.now().strftime('%Y%m%d')}",
agent=self.harness.agent,
test_cases=test_cases,
harness=self.harness,
)
return {
"feedback": feedback_result,
"new_baseline": new_baseline.version,
"total_test_cases": len(test_cases),
}使用範例
system = ContinuousImprovementSystem(
harness=harness,
case_manager=case_manager,
baseline_manager=baseline_manager,
gate_checker=GateChecker(STRICT_POLICY),
evolver=TestCaseEvolver(case_manager, AutoLabeler(call_llm)),
)
report = system.run_ci(baseline_version="v1.0.0")
if not report["gate"]["passed"]:
print("門檻未通過，阻止合併")
for v in report["gate"]["violations"]:
print(f" - {v['rule']}: {v['actual']}")
sys.exit(1)
print("CI 通過")
weekly = system.weekly_cycle()
print(f"從生產環境新增 {weekly['feedback']['added_to_test_suite']} 個測試案例")
print(f"新基準線：{weekly['new_baseline']}")六、可視化與監控
CI/CD 的結果需要被可視化，讓團隊成員一眼就能看出問題。
產生趨勢圖
def generate_trend_chart(reports_dir: str, output_path: str):
"""產生趨勢圖"""
import matplotlib.pyplot as plt
from datetime import datetime
dates = []
pass_rates = []
avg_scores = []
regression_counts = []
for filename in sorted(os.listdir(reports_dir)):
if not filename.endswith(".json"):
continue
with open(os.path.join(reports_dir, filename)) as f:
report = json.load(f)
timestamp = report.get("metadata", {}).get("timestamp", "")
if timestamp:
dates.append(datetime.fromisoformat(timestamp))
pass_rates.append(report["summary"]["pass_rate"])
avg_scores.append(report["summary"]["avg_score"])
reg = report.get("regression", {})
regression_counts.append(reg.get("regression_count", 0))
fig, axes = plt.subplots(3, 1, figsize=(12, 10))
axes[0].plot(dates, pass_rates, marker="o", color="green")
axes[0].set_title("通過率趨勢")
axes[0].set_ylabel("Pass Rate")
axes[0].axhline(y=0.8, color="red", linestyle="--", label="門檻")
axes[0].legend()
axes[1].plot(dates, avg_scores, marker="o", color="blue")
axes[1].set_title("平均分數趨勢")
axes[1].set_ylabel("Avg Score")
axes[2].bar(dates, regression_counts, color="orange")
axes[2].set_title("退步數量趨勢")
axes[2].set_ylabel("Regressions")
plt.tight_layout()
plt.savefig(output_path, dpi=100)
plt.close()
print(f"趨勢圖已儲存至 {output_path}")監控儀表板
def generate_dashboard(reports_dir: str) -> dict:
"""產生儀表板資料"""
reports = []
for filename in sorted(os.listdir(reports_dir)):
if filename.endswith(".json"):
with open(os.path.join(reports_dir, filename)) as f:
reports.append(json.load(f))
if not reports:
return {}
latest = reports[-1]
return {
"current": {
"pass_rate": latest["summary"]["pass_rate"],
"avg_score": latest["summary"]["avg_score"],
"avg_latency": latest["summary"].get("avg_latency", 0),
"total_cases": latest["summary"]["total"],
},
"regression": {
"count": latest.get("regression", {}).get("regression_count", 0),
"new_failures": latest.get("regression", {}).get("new_failure_count", 0),
},
"trend": {
"pass_rate_change": (
latest["summary"]["pass_rate"] -
reports[0]["summary"]["pass_rate"]
),
"avg_score_change": (
latest["summary"]["avg_score"] -
reports[0]["summary"]["avg_score"]
),
},
"history": [
{
"timestamp": r["metadata"]["timestamp"],
"pass_rate": r["summary"]["pass_rate"],
"avg_score": r["summary"]["avg_score"],
}
for r in reports[-10:]
],
}七、最佳實踐
1. 從簡單開始
不要一開始就建立完整的 CI/CD 系統。
# 第一階段：只跑測試
- name: Run Harness
run: python run_harness.py
# 第二階段：加入回歸檢測
- name: Check regressions
run: python check_regressions.py
# 第三階段：加入門檻檢查
- name: Gate check
run: python check_gate.py --policy strict
# 第四階段：加入生產回饋
- name: Weekly evolution
run: python evolve_test_cases.py2. 設定合理的門檻
# 差：門檻太嚴格，什麼都阻止
STRICT = GatePolicy(min_pass_rate=1.0, max_new_failures=0)
# 好：根據嚴重程度區分
REASONABLE = GatePolicy(
min_pass_rate=0.85,
max_new_failures=0,
max_critical_regressions=0,
max_high_regressions=2,
)3. 快速反饋
CI 應該在幾分鐘內完成，而不是幾小時。
# 差：跑所有測試
harness.run(all_test_cases) # 1000 個案例
# 好：分層執行
harness.run(smoke_cases) # PR 時只跑 50 個
harness.run(all_test_cases) # 合併後跑完整4. 並行執行
harness = Harness(agent, parallel=True, max_workers=8)5. 快取結果
class CachedRunner:
def _run_single(self, case):
cache_key = hash(case.input)
if cache_key in self.cache:
return self.cache[cache_key]
# ...6. 清楚的報告
報告應該包含摘要、退步詳情、失敗原因、建議。
7. 自動通知
def notify_on_failure(report: dict, channel: str = "slack"):
"""失敗時通知"""
if not report["gate"]["passed"]:
send_slack_message(
channel="#agent-ci",
text=f"Harness 未通過：{report['gate']['recommendation']}",
)8. 版本控制
git add test_cases/ baselines/ policies/
git commit -m "更新測試集與基準線"9. 定期審查
每月審查測試集是否過時、門檻是否合理、基準線是否更新。
10. 從失敗中學習
def on_ci_failure(report: dict):
save_failure_log(report)
analyze_root_cause(report)
generate_improvement_suggestions(report)
add_regression_test(report)八、常見的陷阱
1. CI 太慢
分層執行，PR 時只跑 smoke tests，合併後跑完整測試。
2. 門檻太嚴格
只阻止嚴重退步，不是任何退步都阻止。
3. 沒有基準線
沒有基準線，無法比較。每個版本都建立基準線。
4. 測試集停滯
每週從生產環境加入新案例，讓測試集持續成長。
5. 沒有通知
CI 失敗時自動通知，不要讓問題被忽略。
6. 忽略成本
分層評分，簡單案例用規則式，複雜案例才用 LLM。
7. 沒有歷史記錄
每次執行都儲存報告，建立歷史記錄。
8. 手動執行
讓 CI 自動觸發，不要依賴手動執行。
九、總結：從手動到自動，從自動到持續
讓我們回顧這一篇的核心：
- 為什麼整合 CI/CD：自動化、一致性、及時性、阻止退步、可追蹤、持續改進。
- GitHub Actions 整合：工作流程、執行腳本、回歸檢查、PR 留言。
- 門檻設定：通過率、新增失敗、嚴重退步、分數下降、延遲、成本。
- 生產環境回饋：收集失敗案例、使用者回饋、自動標註、加入測試集。
- 持續改進迴圈：開發 → 提交 → CI → Harness → 門檻檢查 → 部署 → 生產回饋 → 測試集演化 → 回到開發。
- 可視化與監控：趨勢圖、儀表板。
- 最佳實踐：從簡單開始、合理門檻、快速反饋、並行執行、快取結果、清楚報告、自動通知、版本控制、定期審查、從失敗中學習。
- 常見陷阱：CI 太慢、門檻太嚴格、沒有基準線、測試集停滯、沒有通知、忽略成本、沒有歷史、手動執行。
CI/CD 整合讓 Harness 從「手動工具」變成「自動化流程」。
持續改進讓 Harness 從「靜態測試」變成「動態演化」。
有了這兩者，你的 Agent 系統就能：
- 每次修改都自動驗證
- 發現退步就阻止合併
- 從生產環境持續學習
- 測試集不斷成長
- 品質持續提升
這就是 Harness 的最終目標：讓 Agent 從「能跑」到「可靠」，從「可靠」到「越來越好」。
系列回顧
| 篇號 | 主題 | 核心概念 |
|---|---|---|
| 1 | 什麼是 Agent Harness？ | 定位、與測試/評估/可觀測性的差異、核心組件 |
| 2 | 測試案例設計 | 案例類型、正常/邊界/對抗、組織與管理 |
| 3 | 自動化執行與評分 | Runner、Evaluator、規則式評分、報告產生 |
| 4 | LLM-as-Judge 與 Jev | 推理式評估、決策式評估、三層架構 |
| 5 | 回歸測試與失敗分析 | 基準線、差異檢測、失敗分類、根因分析 |
| 6 | CI/CD 整合與持續改進 | GitHub Actions、門檻設定、生產回饋、測試集演化 |
結語：品質不是一次性的，而是持續的
Harness 系列到這裡告一個段落。
我們從「什麼是 Harness」開始，一路談到測試案例設計、自動化執行、LLM 評分、回歸檢測、失敗分析，最後整合進 CI/CD。
這條路的核心思想是：
品質不是一次性的檢查，而是持續的過程。
沒有 Harness，你只能在發布後祈禱 Agent 不出錯。
有了 Harness，你能在每次修改前就知道會不會退步。
沒有 CI/CD，你只能手動跑測試，偶爾漏掉。
有了 CI/CD，每次提交都自動驗證，不會遺漏。
沒有生產回饋，你的測試集永遠停在開發時的樣子。
有了生產回饋，測試集持續成長，涵蓋越來越多的真實場景。
這就是持續改進的力量。