Introduction
Once agents touch production user flows, the hardest problem is changing things safely. Feature prompts, retrieval policies, ranking models, and pricing algorithms all need controlled trials, guardrails, and rapid rollback. This article delivers an Experimentation & A/B Rollout Orchestrator Agent that proposes experiments, registers them, allocates traffic, watches guardrails, and rolls forward/back only with receipts (experiment IDs, run IDs, rollout change IDs). It treats experiments as contracts, not ad-hoc flags.
The Use Case
Product and data teams run many concurrent trials: copy variants, prompt policies, search rankers, recommender models, pricing heuristics. Without discipline, results get biased (peeking), guardrails get violated (latency, safety), and rollbacks are messy. The agent encodes your experimentation policy, integrates with your flag provider and metrics store, and turns “let’s test this” into a typed, auditable change that can be promoted or reverted in seconds.
Prompt Contract (agent interface)
# file: contracts/ab_orchestrator_v1.yaml
role: "ABOrchestratorAgent"
scope: >
Create, run, and decide A/B/n experiments with pre-registered metrics and guardrails.
Ask once if critical fields are missing (treatment spec, primary metric, MDE, duration cap).
Never assert success without a receipt (exp_id, run_id, change_id).
inputs:
domain: enum["prompts","policies","ranking","pricing","ui"]
treatment_spec: object # structured definition of variant(s)
primary_metric: string # e.g., "signup_conv", "CTR", "CSAT"
guardrails: # hard stops
- {name: string, threshold: number, direction: enum["<",">"]}
mde: number # minimum detectable effect in pct points or %
max_duration_days: integer
max_concurrent: integer
output:
type: object
required: [summary, registration, allocation, monitoring, decision, citations, tool_proposals]
properties:
summary: {type: string, maxWords: 100}
registration:
type: object
required: [exp_id, primary_metric, metrics, hypothesis]
allocation:
type: object
required: [arms, traffic_split, exposure_rules]
monitoring:
type: object
required: [run_id, dashboards, checkpoints]
decision:
type: object
required: [state, rationale, receipts]
properties:
state: {type: string, enum: ["running","guardrail_stop","promote","rollback","inconclusive"]}
rationale: {type: string}
receipts: {type: array, items: string}
citations: {type: array, items: string} # policy/rule ids
tool_proposals:
type: array
items:
type: object
required: [name, args, preconditions, idempotency_key]
properties:
name:
type: string
enum: [RegisterExperiment, CreateFlag, StartRun, AllocateTraffic,
CheckMetrics, StopRun, PromoteVariant, RollbackVariant,
SnapshotDashboard, OpenTicket]
args: {type: object}
preconditions: {type: string}
idempotency_key: {type: string}
policy_id: "exp_policy.v5"
citation_rule: "Minimal-span references to primary metric, guardrail thresholds, and policy clauses."
Tool Interfaces (typed, with receipts)
# tools.py
from pydantic import BaseModel
from typing import List, Dict, Optional
class RegisterExperimentArgs(BaseModel):
domain: str
primary_metric: str
metrics: List[str]
hypothesis: str
mde: float
max_duration_days: int
class CreateFlagArgs(BaseModel):
key: str
variants: List[str] # ["control","treatmentA","treatmentB"]
class StartRunArgs(BaseModel):
exp_id: str
population: str # "all_users","new_users","us_only", etc.
run_name: str
class AllocateTrafficArgs(BaseModel):
flag_key: str
split: Dict[str, int] # {"control": 50, "treatmentA": 50}
exposure_rules: List[str]
class CheckMetricsArgs(BaseModel):
run_id: str
primary_metric: str
guardrails: List[Dict] # [{name, threshold, direction}]
lookback_hours: int
class StopRunArgs(BaseModel):
run_id: str
reason: str
class PromoteVariantArgs(BaseModel):
flag_key: str
variant: str
class RollbackVariantArgs(BaseModel):
flag_key: str
to_variant: str
class SnapshotDashboardArgs(BaseModel):
run_id: str
class OpenTicketArgs(BaseModel):
title: str
description: str
severity: str
owners: List[str]
class ToolReceipt(BaseModel):
tool: str
ok: bool
ref: str # exp_id, run_id, change_id, dashboard_snapshot_id
message: str = ""
data: Optional[Dict] = None
# adapters.py (demo logic; wire to your flag service, metrics store, and BI)
from tools import *
import uuid, random
def register_experiment(a: RegisterExperimentArgs) -> ToolReceipt:
return ToolReceipt(tool="RegisterExperiment", ok=True, ref=f"EXP-{uuid.uuid4().hex[:8]}",
data={"primary_metric": a.primary_metric, "metrics": a.metrics, "mde": a.mde})
def create_flag(a: CreateFlagArgs) -> ToolReceipt:
return ToolReceipt(tool="CreateFlag", ok=True, ref=f"FLAG-{a.key}", message="Flag created")
def start_run(a: StartRunArgs) -> ToolReceipt:
return ToolReceipt(tool="StartRun", ok=True, ref=f"RUN-{uuid.uuid4().hex[:8]}", message="Run started")
def allocate_traffic(a: AllocateTrafficArgs) -> ToolReceipt:
return ToolReceipt(tool="AllocateTraffic", ok=True, ref=f"ALLOC-{a.flag_key}", data={"split": a.split})
def check_metrics(a: CheckMetricsArgs) -> ToolReceipt:
# toy metrics & guardrails
delta = round(random.uniform(-0.04, 0.06), 3) # treatment - control
guardrail_breaches = []
for g in a.guardrails:
obs = random.uniform(0.0, 1.0)
breached = (obs > g["threshold"]) if g["direction"] == ">" else (obs < g["threshold"])
if breached: guardrail_breaches.append({"name": g["name"], "observed": round(obs,3)})
state = "ok" if not guardrail_breaches else "breach"
return ToolReceipt(tool="CheckMetrics", ok=True, ref=f"MET-{a.run_id}", data={"delta": delta, "guardrails": guardrail_breaches, "state": state})
def stop_run(a: StopRunArgs) -> ToolReceipt:
return ToolReceipt(tool="StopRun", ok=True, ref=f"STOP-{a.run_id}", message=a.reason)
def promote_variant(a: PromoteVariantArgs) -> ToolReceipt:
return ToolReceipt(tool="PromoteVariant", ok=True, ref=f"PROM-{a.variant}", message="Variant promoted")
def rollback_variant(a: RollbackVariantArgs) -> ToolReceipt:
return ToolReceipt(tool="RollbackVariant", ok=True, ref=f"RB-{a.to_variant}", message="Rolled back")
def snapshot_dashboard(a: SnapshotDashboardArgs) -> ToolReceipt:
return ToolReceipt(tool="SnapshotDashboard", ok=True, ref=f"DASH-{a.run_id}", message="Dashboard snapshotted")
def open_ticket(a: OpenTicketArgs) -> ToolReceipt:
return ToolReceipt(tool="OpenTicket", ok=True, ref="EXP-INC-1024", message="Experiment incident opened")
Join the conversation! Your thoughts help the community grow.