Workflow 004: Production Deployment
Overview
| Property | Value |
|---|---|
| Workflow ID | 004 |
| Title | Production Deployment |
| Complexity | Advanced |
| Duration | 1-2 hours |
Purpose
Deploy an optimized self-evolving agent to production with proper versioning, rollback capability, and monitoring integration.
Deployment Architecture
┌─────────────────────────────────────────────────────────────────────┐
│ PRODUCTION DEPLOYMENT │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ DEVELOPMENT │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │ Self-Evolve │───►│ Validate │───►│ Export │ │ │
│ │ │ Loop │ │ Results │ │ Prompt │ │ │
│ │ └─────────────┘ └─────────────┘ └──────┬──────┘ │ │
│ └───────────────────────────────────────────────┼──────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ STAGING │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │ Deploy │───►│ A/B Test │───►│ Approve │ │ │
│ │ │ Canary │ │ vs Prod │ │ Release │ │ │
│ │ └─────────────┘ └─────────────┘ └──────┬──────┘ │ │
│ └───────────────────────────────────────────────┼──────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ PRODUCTION │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │ Deploy │───►│ Monitor │───►│ Rollback? │ │ │
│ │ │ (Blue/ │ │ Health │ │ │ │ │
│ │ │ Green) │ │ │ │ │ │ │
│ │ └─────────────┘ └─────────────┘ └─────────────┘ │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘Prerequisites
- Completed Workflow 001: Self-Evolving Loop
- Optimized prompt validated
- Staging environment configured
- Production infrastructure ready
Deployment Checklist
┌─────────────────────────────────────────────────────────────────────┐
│ PRE-DEPLOYMENT CHECKLIST │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ □ Optimization Results Validated │
│ ├─► Best prompt identified │
│ ├─► Score meets threshold (>= 0.85) │
│ ├─► Pass rate acceptable (>= 90%) │
│ └─► Results reproducible │
│ │
│ □ Testing Complete │
│ ├─► Unit tests passing │
│ ├─► Integration tests passing │
│ ├─► Edge cases validated │
│ └─► Performance benchmarks acceptable │
│ │
│ □ Infrastructure Ready │
│ ├─► API keys configured │
│ ├─► Rate limits understood │
│ ├─► Error handling implemented │
│ └─► Logging configured │
│ │
│ □ Monitoring Configured │
│ ├─► Alert thresholds set │
│ ├─► Dashboard accessible │
│ ├─► On-call notified │
│ └─► Rollback procedure documented │
│ │
└─────────────────────────────────────────────────────────────────────┘Implementation
Step 1: Create Deployment Package
python
import json
from datetime import datetime
from dataclasses import dataclass, asdict
from typing import Optional
@dataclass
class DeploymentPackage:
"""Package containing everything needed for deployment."""
version: str
prompt: str
model: str
eval_id: str
# Metadata
created_at: str
optimization_score: float
pass_rate: float
optimization_method: str
# Configuration
temperature: float = 0.7
max_tokens: int = 500
# Validation
min_score_threshold: float = 0.85
rollback_threshold: float = 0.75
def to_json(self) -> str:
return json.dumps(asdict(self), indent=2)
@classmethod
def from_json(cls, json_str: str) -> 'DeploymentPackage':
return cls(**json.loads(json_str))
def validate(self) -> tuple[bool, list[str]]:
"""Validate package before deployment."""
errors = []
if self.optimization_score < self.min_score_threshold:
errors.append(f"Score {self.optimization_score} below threshold {self.min_score_threshold}")
if not self.prompt or len(self.prompt) < 10:
errors.append("Prompt too short or empty")
if self.pass_rate < 0.75:
errors.append(f"Pass rate {self.pass_rate} too low")
return len(errors) == 0, errors
def create_deployment_package(
best_candidate: dict,
eval_id: str,
optimization_method: str = "self-evolving"
) -> DeploymentPackage:
"""Create deployment package from optimization results."""
return DeploymentPackage(
version=f"v{best_candidate.get('version', 0)}",
prompt=best_candidate["prompt"],
model=best_candidate.get("model", "gpt-5"),
eval_id=eval_id,
created_at=datetime.utcnow().isoformat(),
optimization_score=best_candidate["score"],
pass_rate=best_candidate.get("pass_rate", 0.0),
optimization_method=optimization_method,
)Step 2: Create Production Agent Service
python
from agents import Agent, Runner
import asyncio
class ProductionAgentService:
"""Service for managing production agent deployment."""
def __init__(self):
self.current_package: Optional[DeploymentPackage] = None
self.previous_package: Optional[DeploymentPackage] = None
self.agent: Optional[Agent] = None
self.is_healthy = True
def deploy(self, package: DeploymentPackage) -> bool:
"""Deploy a new agent version."""
# Validate package
valid, errors = package.validate()
if not valid:
print(f"Deployment failed: {errors}")
return False
# Store previous for rollback
if self.current_package:
self.previous_package = self.current_package
# Create new agent
self.agent = Agent(
name="ProductionSummarizationAgent",
instructions=package.prompt,
model=package.model,
)
self.current_package = package
print(f"Deployed {package.version} successfully")
return True
def rollback(self) -> bool:
"""Rollback to previous version."""
if not self.previous_package:
print("No previous version to rollback to")
return False
self.current_package, self.previous_package = (
self.previous_package,
self.current_package
)
self.agent = Agent(
name="ProductionSummarizationAgent",
instructions=self.current_package.prompt,
model=self.current_package.model,
)
print(f"Rolled back to {self.current_package.version}")
return True
async def process(self, section: str) -> str:
"""Process a request with the deployed agent."""
if not self.agent:
raise RuntimeError("No agent deployed")
result = await Runner.run(self.agent, section)
return result.final_output
def get_status(self) -> dict:
"""Get current deployment status."""
return {
"deployed_version": self.current_package.version if self.current_package else None,
"previous_version": self.previous_package.version if self.previous_package else None,
"model": self.current_package.model if self.current_package else None,
"score": self.current_package.optimization_score if self.current_package else None,
"is_healthy": self.is_healthy,
}Step 3: Implement Blue-Green Deployment
python
class BlueGreenDeployer:
"""Blue-green deployment strategy."""
def __init__(self):
self.blue = ProductionAgentService()
self.green = ProductionAgentService()
self.active = "blue" # Current active environment
@property
def active_service(self) -> ProductionAgentService:
return self.blue if self.active == "blue" else self.green
@property
def inactive_service(self) -> ProductionAgentService:
return self.green if self.active == "blue" else self.blue
async def deploy_and_test(
self,
package: DeploymentPackage,
test_sections: list[str],
eval_id: str,
min_score: float = 0.85
) -> bool:
"""Deploy to inactive environment and test before switching."""
# Deploy to inactive
inactive = self.inactive_service
if not inactive.deploy(package):
return False
print(f"Testing deployment in {self.active} environment...")
# Run validation tests
scores = []
for section in test_sections[:10]: # Test on subset
summary = await inactive.process(section)
# Evaluate
eval_run = run_eval(eval_id=eval_id, section=section, summary=summary)
run_output = poll_eval_run(eval_id=eval_id, run_id=eval_run.id)
grader_scores = parse_eval_run_output(run_output)
avg_score = calculate_grader_score(grader_scores)
scores.append(avg_score)
avg_score = sum(scores) / len(scores)
print(f"Validation score: {avg_score:.3f}")
if avg_score >= min_score:
# Switch active environment
self.active = "green" if self.active == "blue" else "blue"
print(f"Switched to {self.active} environment")
return True
else:
print(f"Validation failed (score {avg_score:.3f} < {min_score})")
return False
def instant_rollback(self):
"""Instant rollback by switching active environment."""
self.active = "green" if self.active == "blue" else "blue"
print(f"Instant rollback to {self.active} environment")Step 4: Implement Canary Deployment
python
import random
class CanaryDeployer:
"""Canary deployment with gradual traffic shift."""
def __init__(self):
self.stable = ProductionAgentService()
self.canary = ProductionAgentService()
self.canary_percentage = 0.0
def deploy_canary(self, package: DeploymentPackage):
"""Deploy new version as canary."""
self.canary.deploy(package)
self.canary_percentage = 0.05 # Start with 5%
print(f"Canary deployed at {self.canary_percentage:.0%} traffic")
def increase_canary(self, percentage: float):
"""Increase canary traffic percentage."""
self.canary_percentage = min(1.0, percentage)
print(f"Canary traffic increased to {self.canary_percentage:.0%}")
def promote_canary(self):
"""Promote canary to stable."""
self.stable.current_package = self.canary.current_package
self.stable.agent = self.canary.agent
self.canary_percentage = 0.0
print("Canary promoted to stable")
def abort_canary(self):
"""Abort canary deployment."""
self.canary_percentage = 0.0
self.canary.current_package = None
self.canary.agent = None
print("Canary aborted")
async def process(self, section: str) -> str:
"""Route request to appropriate service."""
if random.random() < self.canary_percentage and self.canary.agent:
return await self.canary.process(section)
return await self.stable.process(section)
def get_status(self) -> dict:
return {
"stable": self.stable.get_status(),
"canary": self.canary.get_status() if self.canary.current_package else None,
"canary_percentage": self.canary_percentage,
}Step 5: Production API Endpoint
python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI()
# Global deployer
deployer = BlueGreenDeployer()
class SummarizeRequest(BaseModel):
section: str
class SummarizeResponse(BaseModel):
summary: str
version: str
class DeployRequest(BaseModel):
package_json: str
@app.post("/summarize", response_model=SummarizeResponse)
async def summarize(request: SummarizeRequest):
"""Production summarization endpoint."""
try:
summary = await deployer.active_service.process(request.section)
return SummarizeResponse(
summary=summary,
version=deployer.active_service.current_package.version
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/deploy")
async def deploy(request: DeployRequest):
"""Deploy new version."""
package = DeploymentPackage.from_json(request.package_json)
# Use test sections from dataset
test_sections = [...] # Load test sections
success = await deployer.deploy_and_test(
package=package,
test_sections=test_sections,
eval_id=package.eval_id
)
if success:
return {"status": "deployed", "version": package.version}
else:
raise HTTPException(status_code=400, detail="Deployment validation failed")
@app.post("/rollback")
async def rollback():
"""Instant rollback."""
deployer.instant_rollback()
return {"status": "rolled back", "active": deployer.active}
@app.get("/status")
async def status():
"""Get deployment status."""
return {
"active_environment": deployer.active,
"blue": deployer.blue.get_status(),
"green": deployer.green.get_status(),
}Step 6: Deployment Script
python
async def run_deployment():
"""Complete deployment workflow."""
# 1. Load optimization results
with open("results/best_prompt.txt") as f:
best_prompt = f.read()
best_candidate = {
"prompt": best_prompt,
"score": 0.892,
"pass_rate": 0.95,
"version": 3,
"model": "gpt-5"
}
# 2. Create deployment package
package = create_deployment_package(
best_candidate=best_candidate,
eval_id="eval_...",
optimization_method="self-evolving"
)
# 3. Validate
valid, errors = package.validate()
if not valid:
print(f"Package validation failed: {errors}")
return False
# 4. Save package
with open("deployment/package.json", "w") as f:
f.write(package.to_json())
print("Deployment package created successfully")
print(f"Version: {package.version}")
print(f"Score: {package.optimization_score}")
print(f"Model: {package.model}")
# 5. Deploy (in production, this would trigger CI/CD)
# deployer.deploy_and_test(package, test_sections, eval_id)
return True
if __name__ == "__main__":
asyncio.run(run_deployment())Post-Deployment
┌─────────────────────────────────────────────────────────────────────┐
│ POST-DEPLOYMENT CHECKLIST │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ □ Verify Deployment │
│ ├─► Health check passing │
│ ├─► Test requests succeeding │
│ ├─► Correct version deployed │
│ └─► Logs showing expected behavior │
│ │
│ □ Monitor Initial Period (first hour) │
│ ├─► Watch error rates │
│ ├─► Monitor latency │
│ ├─► Check evaluation scores │
│ └─► Review sample outputs │
│ │
│ □ Document Deployment │
│ ├─► Record version deployed │
│ ├─► Note any issues encountered │
│ ├─► Update runbook if needed │
│ └─► Communicate to stakeholders │
│ │
└─────────────────────────────────────────────────────────────────────┘Verification Checklist
- [ ] Deployment package created and validated
- [ ] Staging tests passing
- [ ] Blue-green or canary deployment configured
- [ ] Health checks passing
- [ ] Monitoring active
- [ ] Rollback procedure tested
- [ ] Documentation updated