Batch Agent Completions at Scale

pip install swarms-client python-dotenv

export SWARMS_API_KEY="your-api-key-here"

import json
import os

import requests
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("SWARMS_API_KEY")
BASE_URL = "https://api.swarms.world"

headers = {"x-api-key": API_KEY, "Content-Type": "application/json"}

LEAD_SCORER_CONFIG = {
    "agent_name": "Lead Scoring Specialist",
    "description": "Scores inbound B2B sales leads on a 0-100 fit scale.",
    "system_prompt": (
        "You are a senior B2B sales operations analyst. Given a lead profile, "
        "score the lead on a 0-100 scale based on ICP fit, buying intent, and "
        "budget signals. Respond as strict JSON: "
        '{"score": int, "tier": "A"|"B"|"C"|"D", "reason": "one sentence"}'
    ),
    "model_name": "gpt-4.1",
    "max_loops": 1,
    "max_tokens": 200,
    "temperature": 0.2,
}

def build_lead_record(i: int) -> dict:
    return {
        "lead_id": f"L{i:05d}",
        "company": f"Acme Subsidiary {i}",
        "industry": ["fintech", "healthtech", "ecommerce", "logistics"][i % 4],
        "headcount": 50 + (i % 500),
        "title": ["VP Eng", "CTO", "Director of Data", "Head of Ops"][i % 4],
        "intent_signal": "downloaded whitepaper" if i % 3 == 0 else "visited pricing",
    }


leads = [build_lead_record(i) for i in range(10_000)]

def lead_to_batch_item(lead: dict) -> dict:
    task = (
        f"Score this lead and return strict JSON.\n"
        f"Lead ID: {lead['lead_id']}\n"
        f"Company: {lead['company']}\n"
        f"Industry: {lead['industry']}\n"
        f"Headcount: {lead['headcount']}\n"
        f"Contact title: {lead['title']}\n"
        f"Intent signal: {lead['intent_signal']}"
    )
    return {"agent_config": LEAD_SCORER_CONFIG, "task": task}


batch_items = [lead_to_batch_item(lead) for lead in leads]

def run_chunk(chunk: list[dict]) -> list[dict]:
    response = requests.post(
        f"{BASE_URL}/v1/agent/batch/completions",
        headers=headers,
        json=chunk,
        timeout=900,
    )
    response.raise_for_status()
    return response.json()


CHUNK_SIZE = 500
all_results: list[dict] = []

for start in range(0, len(batch_items), CHUNK_SIZE):
    chunk = batch_items[start : start + CHUNK_SIZE]
    print(f"Submitting leads {start} - {start + len(chunk) - 1}")
    chunk_results = run_chunk(chunk)
    all_results.extend(chunk_results)

print(f"Scored {len(all_results)} leads.")

tiers: dict[str, list[dict]] = {"A": [], "B": [], "C": [], "D": []}

for lead, result in zip(leads, all_results):
    try:
        # The agent output is a JSON string inside the response envelope.
        output_text = result["outputs"][-1]["content"] if isinstance(result.get("outputs"), list) else result.get("output", "")
        parsed = json.loads(output_text)
        tier = parsed.get("tier", "D")
        tiers.setdefault(tier, []).append({**lead, **parsed})
    except (json.JSONDecodeError, KeyError, TypeError):
        tiers["D"].append({**lead, "score": 0, "tier": "D", "reason": "parse_error"})

for tier, items in tiers.items():
    print(f"Tier {tier}: {len(items)} leads")

print("Top 5 A-tier leads:")
for item in tiers["A"][:5]:
    print(f"  {item['lead_id']} | {item['company']} | {item['reason']}")