Automated A/B Testing Pipeline for AI‑Generated Sales Collateral with OpenClaw on UBOS

# Clone the starter repo
git clone https://github.com/ubos-tech/openclaw-agent-starter.git
cd openclaw-agent-starter

# Install dependencies
npm install

# Register the agent (replace placeholders)
curl -X POST https://api.ubos.tech/agents/register \
  -H "Authorization: Bearer $UBOS_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
        "name": "sales-collateral-agent",
        "image": "ubos/openclaw-agent:latest",
        "env": {
          "OPENAI_API_KEY": "'$OPENAI_API_KEY'",
          "CHROMA_DB_URL": "'$CHROMA_DB_URL'"
        }
      }'

ubos agents list | grep sales-collateral-agent

import csv, requests, os

UBOS_QUEUE_URL = os.getenv('UBOS_QUEUE_URL')
API_TOKEN = os.getenv('UBOS_API_TOKEN')

def push_to_queue(record):
    headers = {'Authorization': f'Bearer {API_TOKEN}'}
    response = requests.post(UBOS_QUEUE_URL, json=record, headers=headers)
    response.raise_for_status()

def ingest(csv_path):
    with open(csv_path, newline='') as csvfile:
        reader = csv.DictReader(csvfile)
        for row in reader:
            push_to_queue(row)

if __name__ == "__main__":
    ingest('prospects.csv')

import os, openai, uuid, requests

openai.api_key = os.getenv('OPENAI_API_KEY')
CHROMA_DB_URL = os.getenv('CHROMA_DB_URL')
RESULT_ENDPOINT = f"{CHROMA_DB_URL}/vectors"

PROMPT_TEMPLATE = """
You are a senior B2B sales copywriter. Write two distinct email bodies (Variant A and Variant B) for the following prospect:
Name: {name}
Company: {company}
Industry: {industry}
Pain point: {pain_point}
"""

def generate_variants(prospect):
    prompt = PROMPT_TEMPLATE.format(**prospect)
    response = openai.ChatCompletion.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.7,
        n=1
    )
    # Assume the model returns a string with "Variant A:" and "Variant B:" sections
    text = response.choices[0].message.content
    variant_a, variant_b = text.split("Variant B:")
    return {
        "id": str(uuid.uuid4()),
        "prospect_id": prospect["id"],
        "variant_a": variant_a.replace("Variant A:", "").strip(),
        "variant_b": variant_b.strip()
    }

def store_in_chroma(record):
    payload = {
        "id": record["id"],
        "metadata": {
            "prospect_id": record["prospect_id"],
            "variant_a": record["variant_a"],
            "variant_b": record["variant_b"]
        },
        "embedding": []  # Optional: add vector embedding if needed
    }
    requests.post(RESULT_ENDPOINT, json=payload).raise_for_status()

if __name__ == "__main__":
    # Example prospect payload from queue
    example = {
        "id": "123",
        "name": "Jane Doe",
        "company": "Acme Corp",
        "industry": "Manufacturing",
        "pain_point": "low inventory turnover"
    }
    result = generate_variants(example)
    store_in_chroma(result)

import os, requests, time
from datetime import datetime

TELEGRAM_BOT_TOKEN = os.getenv('TELEGRAM_BOT_TOKEN')
TELEGRAM_CHAT_ID = os.getenv('TELEGRAM_CHAT_ID')
CHROMA_DB_URL = os.getenv('CHROMA_DB_URL')
METRICS_ENDPOINT = f"{CHROMA_DB_URL}/metrics"

def send_telegram(message):
    url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
    payload = {"chat_id": TELEGRAM_CHAT_ID, "text": message}
    requests.post(url, json=payload).raise_for_status()

def record_metric(variant, clicked):
    payload = {
        "variant": variant,
        "clicked": clicked,
        "timestamp": datetime.utcnow().isoformat()
    }
    requests.post(METRICS_ENDPOINT, json=payload).raise_for_status()

def orchestrate():
    # Pull all pending vectors
    vectors = requests.get(f"{CHROMA_DB_URL}/vectors?status=pending").json()
    for vec in vectors:
        # Send Variant A
        send_telegram(f"🅰️ Variant A for {vec['metadata']['prospect_id']}:\n{vec['metadata']['variant_a']}")
        time.sleep(2)  # simulate delay
        # Send Variant B
        send_telegram(f"🅱️ Variant B for {vec['metadata']['prospect_id']}:\n{vec['metadata']['variant_b']}")
        # In a real scenario, you would capture click events via a webhook.
        # Here we mock random clicks:
        import random
        record_metric("A", random.choice([True, False]))
        record_metric("B", random.choice([True, False]))
        # Mark vector as processed
        requests.patch(f"{CHROMA_DB_URL}/vectors/{vec['id']}", json={"status": "processed"}).raise_for_status()

if __name__ == "__main__":
    orchestrate()

version: "3.8"
services:
  ingestion:
    build: ./ingestion
    environment:
      - UBOS_QUEUE_URL=${UBOS_QUEUE_URL}
      - UBOS_API_TOKEN=${UBOS_API_TOKEN}
    restart: unless-stopped

  generation:
    build: ./generation
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - CHROMA_DB_URL=${CHROMA_DB_URL}
    restart: unless-stopped

  orchestrator:
    build: ./orchestrator
    environment:
      - TELEGRAM_BOT_TOKEN=${TELEGRAM_BOT_TOKEN}
      - TELEGRAM_CHAT_ID=${TELEGRAM_CHAT_ID}
      - CHROMA_DB_URL=${CHROMA_DB_URL}
    restart: unless-stopped

# Log in to UBOS
ubos login --api-key $UBOS_API_KEY

# Initialize a new project (optional)
ubos project init sales-ab-test

# Deploy the compose file
ubos deploy -f docker-compose.yml --env-file .env

# Verify deployment
ubos services list