Estratégias de implantação e reversão

Desafios da implantação de prompts

Implantação azul-verde de prompts

# prompt_router.py — blue/green traffic control
class PromptRouter:
    def __init__(self):
        self.slots = {
            'blue': None,   # {'prompt_id': ..., 'version': ...}
            'green': None,
        }
        self.active_slot = 'blue'

    def load_slot(self, slot, prompt_id, version, template):
        self.slots[slot] = {
            'prompt_id': prompt_id,
            'version': version,
            'template': template
        }
        print(f'Loaded {prompt_id}@{version} into {slot} slot')

    def switch_to(self, slot):
        if not self.slots[slot]:
            raise ValueError(f'Slot {slot} is empty')
        self.active_slot = slot
        info = self.slots[slot]
        print(f'Traffic now routed to {slot}: {info["prompt_id"]}@{info["version"]}')

    def get_active_template(self):
        return self.slots[self.active_slot]['template']

Divisão de tráfego para lançamento gradual

import random

class TrafficSplitter:
    def __init__(self):
        # version_weights: {version_id: percentage}
        self.version_weights = {
            'v1.1.0': 90,
            'v1.2.0': 10
        }

    def select_version(self):
        versions = list(self.version_weights.keys())
        weights = list(self.version_weights.values())
        return random.choices(versions, weights=weights, k=1)[0]

    def update_split(self, new_weights):
        assert sum(new_weights.values()) == 100, 'Weights must sum to 100'
        self.version_weights = new_weights
        print(f'Traffic split updated: {new_weights}')

# Usage
splitter = TrafficSplitter()
for _ in range(5):
    print(splitter.select_version())

# Ramp to 25%
splitter.update_split({'v1.1.0': 75, 'v1.2.0': 25})

Sinalizadores de funcionalidade para lançamento de prompts

# Feature flag-based prompt selection
BETA_USER_IDS = {'user_123', 'user_456', 'user_789'}

def select_prompt_version(user_id, prompt_id, registry):
    # Check if user is in beta cohort
    if user_id in BETA_USER_IDS:
        # Try to get a beta version tagged for this prompt
        beta = registry.get_version_by_tag(prompt_id, tag='beta')
        if beta:
            return beta
    # Default: serve active (stable) version
    return registry.get_active(prompt_id)

# Example: LaunchDarkly-style flag check
def select_prompt_ld(user_id, ld_client, registry):
    use_new = ld_client.variation(
        'use-summarize-v2', {'key': user_id}, default=False
    )
    version = 'v2.0.0' if use_new else 'v1.1.0'
    return registry.get_version(prompt_id='summarize-article', version=version)

Monitorando a qualidade da nova versão

from collections import defaultdict
import time

class VersionMonitor:
    def __init__(self):
        self.metrics = defaultdict(lambda: {
            'calls': 0, 'errors': 0,
            'latency_sum': 0, 'quality_sum': 0
        })

    def record(self, version, latency_ms, quality_score, error=False):
        m = self.metrics[version]
        m['calls'] += 1
        m['latency_sum'] += latency_ms
        m['quality_sum'] += quality_score
        if error:
            m['errors'] += 1

    def report(self, version):
        m = self.metrics[version]
        n = m['calls'] or 1
        return {
            'version': version,
            'calls': m['calls'],
            'error_rate': round(m['errors'] / n, 3),
            'avg_latency_ms': round(m['latency_sum'] / n),
            'avg_quality': round(m['quality_sum'] / n, 2)
        }

Rollback automático em caso de regressão de qualidade

ROLLBACK_THRESHOLDS = {
    'error_rate': 0.05,      # > 5% errors trigger rollback
    'avg_latency_ms': 10000, # > 10s avg latency triggers rollback
    'avg_quality': 3.5       # < 3.5 quality score triggers rollback
}

def check_and_rollback(monitor, registry, version, previous_version):
    report = monitor.report(version)
    triggers = []

    if report['error_rate'] > ROLLBACK_THRESHOLDS['error_rate']:
        triggers.append(f'error_rate={report["error_rate"]}')
    if report['avg_latency_ms'] > ROLLBACK_THRESHOLDS['avg_latency_ms']:
        triggers.append(f'latency={report["avg_latency_ms"]}ms')
    if report['avg_quality'] < ROLLBACK_THRESHOLDS['avg_quality']:
        triggers.append(f'quality={report["avg_quality"]}')

    if triggers:
        print(f'ROLLBACK TRIGGERED: {triggers}')
        registry.activate_version('summarize-article', previous_version)
        alert_oncall(f'Prompt auto-rolled back to {previous_version}: {triggers}')
        return True
    return False

Padrão de implantação canário

import time

def canary_deploy(registry, splitter, monitor, new_version, prev_version,
                  soak_minutes=30, stages=[1, 5, 25, 50, 100]):
    for pct in stages:
        splitter.update_split({
            prev_version: 100 - pct,
            new_version: pct
        })
        print(f'Stage: {pct}% canary. Soaking for {soak_minutes} min...')
        time.sleep(soak_minutes * 60)  # wait soak period

        should_rollback = check_and_rollback(
            monitor, registry, new_version, prev_version
        )
        if should_rollback:
            splitter.update_split({prev_version: 100})
            print('Canary aborted. Fully reverted to', prev_version)
            return False
        print(f'Stage {pct}% passed quality check.')

    print(f'Canary complete. {new_version} now at 100%.')
    return True

Orquestração do fluxo de implantação

# deploy_prompt.py — full pipeline
import argparse

def deploy(prompt_id, new_version, prev_version, dry_run=False):
    print(f'=== Deploying {prompt_id}: {prev_version} -> {new_version} ===')

    # 1. Validate new version exists in registry
    artifact = registry.get_version(prompt_id, new_version)
    print(f'Found artifact: {artifact["version"]}')

    # 2. Run offline eval suite
    score = run_eval_suite(artifact['template'])
    if score < 0.90:
        raise RuntimeError(f'Eval score {score} below threshold 0.90')
    print(f'Eval passed: {score}')

    if dry_run:
        print('Dry run complete. Not deploying.')
        return

    # 3. Canary deploy with automatic rollback
    success = canary_deploy(
        registry, splitter, monitor,
        new_version, prev_version,
        soak_minutes=15, stages=[1, 5, 25, 100]
    )

    print('Deployment', 'SUCCEEDED' if success else 'FAILED')

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--new', required=True)
    parser.add_argument('--prev', required=True)
    parser.add_argument('--dry-run', action='store_true')
    args = parser.parse_args()
    deploy('summarize-article', args.new, args.prev, args.dry_run)

Procedimento de rollback manual

# RUNBOOK: Manual Prompt Rollback
# Estimated time to execute: 2-3 minutes

# Step 1: Identify current and target versions
python manage.py prompt list-versions --prompt-id summarize-article
# Output:
#   v1.2.0  [ACTIVE]  2024-08-15  alice
#   v1.1.0  [stable]  2024-07-01  alice

# Step 2: Activate previous stable version
python manage.py prompt activate --prompt-id summarize-article --version v1.1.0
# Output: Activated summarize-article@v1.1.0

# Step 3: Verify traffic is serving old version
python manage.py prompt verify --prompt-id summarize-article
# Output: Active version: v1.1.0  Serving: 100%

# Step 4: Log the incident
python manage.py incident create \
  --title 'Prompt rollback: summarize-article v1.2.0 -> v1.1.0' \
  --severity P2 \
  --reason 'Quality score dropped from 4.1 to 3.2 after v1.2.0 deploy'

Configuração de implantação como código

# deployments/summarize-article.yaml
prompt_id: summarize-article
current_stable: '1.1.0'
canary_config:
  stages: [1, 5, 25, 50, 100]
  soak_minutes_per_stage: 15
rollback_thresholds:
  error_rate_max: 0.05
  latency_p95_max_ms: 8000
  quality_score_min: 3.5
feature_flags:
  beta_group: [user_123, user_456]
alerts:
  pagerduty_key: 'PD_KEY_PLACEHOLDER'
  slack_channel: '#prompt-alerts'

# Load and apply with a deploy script
import yaml

with open('deployments/summarize-article.yaml') as f:
    config = yaml.safe_load(f)

print('Deploying with config:', config['canary_config'])

Guias operacionais de plantão e análises pós-incidente

# Post-mortem template (stored in docs/post-mortems/)

## Incident: summarize-article v1.2.0 quality regression
## Date: 2024-08-15
## Severity: P2
## Duration: 47 minutes (09:15 - 10:02 UTC)

### What happened
Deployed v1.2.0 of summarize-article. At 5% canary, quality score dropped
from 4.1 to 3.0 for articles over 2000 words.

### Root cause
New prompt template removed the explicit length constraint instruction.
Long articles caused the model to generate overly verbose summaries.

### Timeline
09:15  v1.2.0 deployed to 5% canary
09:28  Quality monitor detected avg_quality < 3.5
09:29  Auto-rollback triggered to v1.1.0
09:32  Incident acknowledged by on-call
10:02  Post-mortem drafted

### Action items
- [ ] Add length regression test to eval suite
- [ ] Update canary monitoring to separate metrics by article length
- [ ] Add changelog requirement: 'length behavior' field

Verificação rápida

Resumo da estratégia de implantação