πŸŽ‰ 75% of content is free forever β€” Unlock Premium from $10/mo β†’
CW
πŸ’Ό Servicesℹ️ Aboutβœ‰οΈ ContactView Pricing Plansfrom $10

Agent Version Control: Prompt Versioning, A/B Testing & Rollback

AI AgentsAgent Version Control🟒 Free Lesson

Advertisement

Agent Version Control

Why This Matters

Version control for AI agents is critical because prompts and configurations are code that determine agent behavior. Without proper versioning, you cannot reproduce results, compare approaches, rollback failures, or maintain consistency across deployments. It enables data-driven improvements and safe experimentation.

Real-World Analogy

Think of agent version control like pharmaceutical clinical trials. You don't give an experimental drug to everyone at once. First, you test in controlled groups (A/B testing), measure results carefully (monitoring), and if something goes wrong, you stop the trial immediately (rollback). Each drug formulation is tracked with its exact ingredients (prompt versioning), and you never lose the previous approved version.

Version Control Architecture

Agent Version Control SystemVersion Storagev1.0.0 - Initial promptv1.1.0 - Improved accuracyv1.2.0 - Added toolsv1.3.0 - Currentv1.4.0-beta - TestingA/B TestingControl A50% trafficVariant B50% trafficStatistical significancep-value < 0.05Minimum sample sizeData-driven decisionsDeploymentCanary (5%)Rolling (25%)Full (100%)RollbackProgressive deliveryAutomated rollbackMonitoringPerformanceQualityCostUser FeedbackAlertsVersion LifecycleDraftDevelopmentTestingA/B testCanary5% trafficProductionFull rolloutMonitoringObserveDeprecatedArchiveRollback StrategyAutomatic RollbackError rate > thresholdLatency spike detectedQuality drop > 20%Manual RollbackOperator triggeredOne-click revertAudit trail loggedFeature FlagsInstant disableNo deployment neededGradual rolloutBlue-Green DeploymentMaintain two environmentsInstant switch between versionsZero-downtime rollback

Version Control System

import hashlib
import json
import time
import copy
import logging
from dataclasses import dataclass, field
from typing import Any, Optional
from enum import Enum

logger = logging.getLogger(__name__)

class VersionStatus(Enum):
    DRAFT = "draft"
    TESTING = "testing"
    CANARY = "canary"
    PRODUCTION = "production"
    DEPRECATED = "deprecated"

@dataclass
class PromptVersion:
    version_id: str
    prompt: str
    model: str = "gpt-4"
    parameters: dict = field(default_factory=dict)
    status: VersionStatus = VersionStatus.DRAFT
    created_at: float = field(default_factory=time.time)
    created_by: str = ""
    description: str = ""
    hash: str = ""
    metrics: dict = field(default_factory=dict)

    def __post_init__(self):
        if not self.hash:
            self.hash = hashlib.sha256(
                f"{self.prompt}:{json.dumps(self.parameters, sort_keys=True)}".encode()
            ).hexdigest()[:16]

class VersionControlSystem:
    def __init__(self):
        self.versions: dict[str, PromptVersion] = {}
        self.current_version: Optional[str] = None
        self.version_history: list[str] = []
        self.deployments: list[dict] = []

    def create_version(
        self,
        version_id: str,
        prompt: str,
        model: str = "gpt-4",
        parameters: dict = None,
        description: str = "",
        created_by: str = "",
    ) -> PromptVersion:
        version = PromptVersion(
            version_id=version_id,
            prompt=prompt,
            model=model,
            parameters=parameters or {},
            description=description,
            created_by=created_by,
        )
        self.versions[version_id] = version
        self.version_history.append(version_id)
        logger.info(f"Created version {version_id} with hash {version.hash}")
        return version

    def deploy_version(self, version_id: str, environment: str = "production"):
        if version_id not in self.versions:
            raise ValueError(f"Version {version_id} not found")
        
        version = self.versions[version_id]
        version.status = VersionStatus.PRODUCTION
        self.current_version = version_id
        
        self.deployments.append({
            "version_id": version_id,
            "environment": environment,
            "timestamp": time.time(),
            "previous_version": self.version_history[-2] if len(self.version_history) > 1 else None,
        })
        logger.info(f"Deployed version {version_id} to {environment}")

    def rollback(self, target_version: str = None) -> str:
        if not target_version:
            if len(self.version_history) < 2:
                raise ValueError("No previous version to rollback to")
            target_version = self.version_history[-2]
        
        if target_version not in self.versions:
            raise ValueError(f"Version {target_version} not found")
        
        previous = self.current_version
        self.deploy_version(target_version)
        
        self.deployments.append({
            "version_id": target_version,
            "environment": "rollback",
            "timestamp": time.time(),
            "previous_version": previous,
            "reason": "manual_rollback",
        })
        logger.info(f"Rolled back from {previous} to {target_version}")
        return target_version

    def get_version_diff(self, v1_id: str, v2_id: str) -> dict:
        v1 = self.versions.get(v1_id)
        v2 = self.versions.get(v2_id)
        if not v1 or not v2:
            raise ValueError("Version not found")
        
        return {
            "prompt_changed": v1.prompt != v2.prompt,
            "model_changed": v1.model != v2.model,
            "parameters_changed": v1.parameters != v2.parameters,
            "prompt_length_diff": len(v2.prompt) - len(v1.prompt),
            "prompt_hash_v1": v1.hash,
            "prompt_hash_v2": v2.hash,
        }

    def get_version_history(self) -> list[dict]:
        return [
            {
                "version_id": vid,
                "status": self.versions[vid].status.value,
                "created_at": self.versions[vid].created_at,
                "hash": self.versions[vid].hash,
            }
            for vid in self.version_history
        ]

A/B Testing Framework

import random
import time
import statistics
from dataclasses import dataclass, field
from typing import Any, Callable, Optional
from enum import Enum
import math

class TestStatus(Enum):
    DRAFT = "draft"
    RUNNING = "running"
    PAUSED = "paused"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class ABTest:
    test_id: str
    name: str
    control_version: str
    variant_version: str
    traffic_split: float = 0.5
    status: TestStatus = TestStatus.DRAFT
    min_sample_size: int = 1000
    significance_level: float = 0.05
    start_time: float = 0.0
    end_time: float = 0.0
    results: dict = field(default_factory=dict)

class ABTestFramework:
    def __init__(self):
        self.tests: dict[str, ABTest] = {}
        self.assignments: dict[str, dict[str, str]] = {}
        self.metrics: dict[str, dict[str, list[float]]] = {}

    def create_test(
        self,
        test_id: str,
        name: str,
        control_version: str,
        variant_version: str,
        traffic_split: float = 0.5,
        min_sample_size: int = 1000,
    ) -> ABTest:
        test = ABTest(
            test_id=test_id,
            name=name,
            control_version=control_version,
            variant_version=variant_version,
            traffic_split=traffic_split,
            min_sample_size=min_sample_size,
        )
        self.tests[test_id] = test
        self.metrics[test_id] = {"control": [], "variant": []}
        return test

    def assign_variant(self, test_id: str, user_id: str) -> str:
        test = self.tests.get(test_id)
        if not test or test.status != TestStatus.RUNNING:
            return "control"
        
        if user_id in self.assignments.get(test_id, {}):
            return self.assignments[test_id][user_id]
        
        variant = "variant" if random.random() < test.traffic_split else "control"
        if test_id not in self.assignments:
            self.assignments[test_id] = {}
        self.assignments[test_id][user_id] = variant
        return variant

    def record_metric(self, test_id: str, variant: str, value: float):
        if test_id in self.metrics:
            self.metrics[test_id][variant].append(value)

    def analyze_results(self, test_id: str) -> dict:
        test = self.tests.get(test_id)
        if not test:
            raise ValueError(f"Test {test_id} not found")
        
        control_values = self.metrics[test_id]["control"]
        variant_values = self.metrics[test_id]["variant"]
        
        if len(control_values) < 30 or len(variant_values) < 30:
            return {"status": "insufficient_data", "control_n": len(control_values), "variant_n": len(variant_values)}
        
        control_mean = statistics.mean(control_values)
        variant_mean = statistics.mean(variant_values)
        control_std = statistics.stdev(control_values) if len(control_values) > 1 else 0
        variant_std = statistics.stdev(variant_values) if len(variant_values) > 1 else 0
        
        pooled_std = math.sqrt(
            (control_std**2 / len(control_values)) + (variant_std**2 / len(variant_values))
        )
        
        if pooled_std == 0:
            z_score = 0
        else:
            z_score = (variant_mean - control_mean) / pooled_std
        
        p_value = 2 * (1 - self._normal_cdf(abs(z_score)))
        
        significant = p_value < test.significance_level
        lift = (variant_mean - control_mean) / control_mean * 100 if control_mean != 0 else 0
        
        return {
            "control_mean": control_mean,
            "variant_mean": variant_mean,
            "lift": lift,
            "p_value": p_value,
            "significant": significant,
            "control_n": len(control_values),
            "variant_n": len(variant_values),
            "winner": "variant" if significant and variant_mean > control_mean else "control",
        }

    def _normal_cdf(self, x: float) -> float:
        return 0.5 * (1 + math.erf(x / math.sqrt(2)))

    def start_test(self, test_id: str):
        test = self.tests.get(test_id)
        if test:
            test.status = TestStatus.RUNNING
            test.start_time = time.time()

    def stop_test(self, test_id: str):
        test = self.tests.get(test_id)
        if test:
            test.status = TestStatus.COMPLETED
            test.end_time = time.time()
            test.results = self.analyze_results(test_id)

Feature Flag System

import hashlib
import time
import random
from dataclasses import dataclass, field
from typing import Any, Callable
from enum import Enum

class FlagStatus(Enum):
    DISABLED = "disabled"
    ENABLED = "enabled"
    PERCENTAGE = "percentage"

@dataclass
class FeatureFlag:
    flag_id: str
    name: str
    status: FlagStatus = FlagStatus.DISABLED
    percentage: float = 0.0
    allowed_users: list[str] = field(default_factory=list)
    denied_users: list[str] = field(default_factory=list)
    created_at: float = field(default_factory=time.time)
    expires_at: float = 0.0
    description: str = ""
    metadata: dict = field(default_factory=dict)

class FeatureFlagSystem:
    def __init__(self):
        self.flags: dict[str, FeatureFlag] = {}
        self.evaluation_log: list[dict] = []

    def create_flag(
        self,
        flag_id: str,
        name: str,
        status: FlagStatus = FlagStatus.DISABLED,
        percentage: float = 0.0,
        description: str = "",
    ) -> FeatureFlag:
        flag = FeatureFlag(
            flag_id=flag_id,
            name=name,
            status=status,
            percentage=percentage,
            description=description,
        )
        self.flags[flag_id] = flag
        return flag

    def is_enabled(self, flag_id: str, user_id: str = None, context: dict = None) -> bool:
        flag = self.flags.get(flag_id)
        if not flag:
            return False
        
        if flag.expires_at and time.time() > flag.expires_at:
            return False
        
        if user_id and user_id in flag.denied_users:
            return False
        
        if user_id and user_id in flag.allowed_users:
            return True
        
        if flag.status == FlagStatus.DISABLED:
            return False
        elif flag.status == FlagStatus.ENABLED:
            return True
        elif flag.status == FlagStatus.PERCENTAGE:
            if user_id:
                hash_val = int(hashlib.md5(f"{flag_id}:{user_id}".encode()).hexdigest(), 16)
                return (hash_val % 100) < (flag.percentage * 100)
            return random.random() < flag.percentage
        
        return False

    def set_percentage(self, flag_id: str, percentage: float):
        flag = self.flags.get(flag_id)
        if flag:
            flag.percentage = max(0.0, min(1.0, percentage))
            flag.status = FlagStatus.PERCENTAGE if 0 < percentage < 1 else (
                FlagStatus.ENABLED if percentage >= 1 else FlagStatus.DISABLED
            )

    def get_all_flags(self) -> list[dict]:
        return [
            {
                "flag_id": f.flag_id,
                "name": f.name,
                "status": f.status.value,
                "percentage": f.percentage,
            }
            for f in self.flags.values()
        ]

Mathematical Foundation

Statistical Significance (A/B testing):

Where:

  • β€” Mean of variant and control
  • β€” Standard deviations
  • β€” Sample sizes

Minimum Sample Size:

Where:

  • β€” Critical value for significance level
  • β€” Power (1 - Ξ²)
  • β€” Standard deviation
  • β€” Minimum detectable effect

Lift Calculation:

Bayesian A/B Testing (Beta-Binomial):

Performance Considerations

ComponentLatency ImpactCost ImpactAccuracy Impact
Version Lookup+0.1-1msMinimalN/A
A/B Assignment+0.01-0.1msMinimalN/A
Statistical Analysis+5-50ms+ComputeHigh confidence
Feature Flag Eval+0.01-0.1msMinimalN/A

Security Considerations

  • Version history: Protect from unauthorized modifications
  • A/B test data: May contain user behavior patterns
  • Feature flags: Control access to sensitive features
  • Deployment credentials: Secure rollback triggers
  • Audit trails: Log all version changes for compliance

Interview Questions

1. Why is prompt versioning important for production agents?

Answer: Prompt versioning is critical because: 1) Prompts are codeβ€”they determine agent behavior, 2) Changes can significantly impact quality and safety, 3) Rollback capability is essential when issues arise, 4) A/B testing requires comparing specific versions, 5) Audit trails are needed for compliance, 6) Collaboration requires tracking who changed what. Without versioning, you cannot: reproduce results, compare approaches, rollback failures, or maintain consistency across deployments.

2. How does A/B testing work for prompt optimization?

Answer: A/B testing compares two prompt versions by splitting traffic: 1) Randomly assign users to control (old prompt) or variant (new prompt), 2) Measure key metrics (quality, latency, cost), 3) Run until statistical significance (p < 0.05), 4) Analyze results and deploy winner. Critical aspects: sufficient sample size (1000+), proper randomization, consistent measurement, and accounting for novelty effects. For agents: test on a subset of queries, monitor safety metrics, and ensure the variant doesn't degrade critical capabilities.

3. What is the difference between canary and blue-green deployments?

Answer: Canary deployment gradually increases traffic to the new version: 5% β†’ 25% β†’ 50% β†’ 100%, allowing monitoring at each stage. Blue-green maintains two identical environments and switches traffic instantly. Canary is better for: gradual validation, limited resource overhead, catching issues early. Blue-green is better for: instant rollback, zero-downtime deployment, testing in production-like environment. For agents: canary is preferred because LLM behavior can be unpredictable and gradual rollout reduces risk.

4. How do you implement automated rollback triggers?

Answer: Monitor key metrics and trigger rollback when thresholds are exceeded: 1) Error rate > 5% for 5 minutes, 2) Latency p95 > 2x baseline, 3) Quality score drop > 20%, 4) Cost spike > 2x normal, 5) Safety metric violation. Implementation: continuous monitoring, sliding window evaluation, cooldown periods to prevent flapping, and notification before/after rollback. Include: automatic rollback for critical issues, semi-automatic (approve with one click) for warnings, and manual for informational alerts.

5. What is the purpose of feature flags in agent deployment?

Answer: Feature flags decouple deployment from release: 1) Deploy code but keep features disabled, 2) Enable gradually by percentage or user segment, 3) Instant disable without deployment, 4) Test in production with real traffic, 5) Personalize features by user. For agents: flags can control prompt versions, tool availability, model selection, and guardrail settings. Benefits: reduced risk, faster iteration, and granular control. Use flags for: new prompts, tool changes, model upgrades, and safety features.

6. How do you handle version compatibility across agent components?

Answer: Maintain compatibility matrix: 1) Version each component independently, 2) Define compatibility contracts between versions, 3) Test integration with all compatible versions, 4) Support N-1 versions during rollout, 5) Use semantic versioning (major.minor.patch). For agents: prompt versions must be compatible with tool schemas, model versions, and context formats. Implement: version negotiation at startup, graceful degradation for incompatible versions, and migration utilities for major version changes.

7. What metrics should you track during version rollout?

Answer: Track: 1) Performance β€” Latency, throughput, error rate, 2) Quality β€” Task completion, accuracy, user satisfaction, 3) Cost β€” Token usage, API calls, compute cost, 4) Safety β€” Toxicity, bias, policy violations, 5) Operational β€” Rollback rate, deployment success, 6) Business β€” Conversion, engagement, retention. Compare version vs. baseline using statistical testing. Set up alerts for anomalies. Use dashboards to visualize rollout progress. Log all metrics for post-mortem analysis.

8. How would you design a version control system for multi-agent workflows?

Answer: Version the entire workflow graph: 1) Version each agent independently, 2) Version the workflow definition (edges, routing), 3) Maintain compatibility matrix between agent versions, 4) Support partial rollouts (version specific agents), 5) Implement workflow-level A/B testing. Challenges: stateful workflows need migration, dependencies between agents, and testing combinatorial version spaces. Solutions: version snapshots, compatibility testing, canary deployment at workflow level, and automated migration scripts for breaking changes.

Common Pitfalls

PitfallSolution
No version historyImplement version control from day one
Skipping statistical testingRequire significance before promoting
Rollback takes too longUse feature flags for instant disable
Incompatible versions deployedMaintain compatibility matrix
No monitoring during rolloutTrack key metrics with alerts
Manual deployment processesAutomate with CI/CD pipelines
Version sprawlDeprecate old versions regularly
Inconsistent testing environmentsUse identical production-like environments

KnowledgeCheck

  1. Why is prompt versioning important?

    • a) To make prompts longer
    • b) To track changes and enable rollback
    • c) To increase model speed
    • d) To reduce token usage
  2. What is the minimum sample size typically needed for A/B testing?

    • a) 10
    • b) 100
    • c) 1000
    • d) 10000
  3. What p-value indicates statistical significance in A/B testing?

    • a) p > 0.10
    • b) p < 0.05
    • c) p > 0.05
    • d) p < 0.50
  4. What is the advantage of canary deployments?

    • a) Instant full rollout
    • b) Gradual validation with limited risk
    • c) Zero resource usage
    • d) No monitoring required
  5. When should automatic rollback trigger?

    • a) On any error
    • b) When metrics exceed predefined thresholds
    • c) After 24 hours
    • d) Never automatically
  6. What is the purpose of feature flags?

    • a) To encrypt prompts
    • b) To decouple deployment from release
    • c) To increase model accuracy
    • d) To reduce costs

Answers: 1-b, 2-c, 3-b, 4-b, 5-b, 6-b

Need Expert AI Agents Help?

Get personalized tutoring, project support, or professional consulting.

Advertisement