agentic-ai-engineering/course/labs/L5-cicd/starter.py

90 lines
2.5 KiB
Python

"""
Lab 5.10: CI/CD Pipeline
Create a golden dataset and automated regression gate.
"""
import json
import sys
from typing import Any
# TODO 1: Import or define your agent
# from my_agent import run_agent
# TODO 2: Define the CI/CD pipeline
class AgentCIPipeline:
"""
CI/CD pipeline for agents.
Stages:
1. CHECKOUT — latest agent config
2. EVAL — run against golden dataset
3. GATE — pass@k >= threshold?
4. PROMOTE — deploy if pass
5. ROLLBACK — revert if fail
"""
def __init__(self, golden_dataset_path: str, threshold: float = 0.8):
self.golden_dataset = json.load(open(golden_dataset_path))
self.threshold = threshold
def run_eval(self) -> dict:
"""
Run agent against golden dataset.
Returns pass@1, pass@3, pass@5 scores.
"""
pass # TODO
def check_gate(self, eval_results: dict) -> bool:
"""
Check if eval results pass the gate.
Returns True if pass@3 >= threshold.
"""
pass # TODO
def promote(self):
"""Promote new agent config to production."""
# TODO: Copy new config to production path
# TODO: Restart agent service
# TODO: Log deployment
pass
def rollback(self, reason: str):
"""Rollback to previous agent config."""
# TODO: Restore previous config
# TODO: Log rollback reason
# TODO: Alert operator
pass
def run(self):
"""Execute the full CI/CD pipeline."""
print("=" * 50)
print("Agent CI/CD Pipeline")
print("=" * 50)
print("\n[CHECKOUT] Loading agent config...")
# TODO: load current config
print("\n[EVAL] Running golden dataset...")
results = self.run_eval()
print(f" pass@1: {results['pass@1']:.1%}")
print(f" pass@3: {results['pass@3']:.1%}")
print(f"\n[GATE] Threshold: {self.threshold:.0%}")
if self.check_gate(results):
print(" ✓ Gate passed. Promoting to production...")
self.promote()
print(" ✓ Deployment complete")
else:
print(" ✗ Gate failed. Rolling back...")
self.rollback(f"pass@3 {results['pass@3']:.1%} < threshold {self.threshold:.0%}")
print(" ✓ Rollback complete")
sys.exit(1)
if __name__ == "__main__":
pipeline = AgentCIPipeline("golden_dataset.json", threshold=0.8)
pipeline.run()