把 LLM 成本降低 10 倍：先用决策模型做筛选

pip install -U swarms
# or
uv pip install -U swarms

export OPENAI_API_KEY="sk-..."
export TYPESAFE_API_KEY="..."

python -c "import swarms; print(swarms.__version__)"

from swarms import DecisionModel

# Reads TYPESAFE_API_KEY and calls TypeSafe's jev-latest by default.
model = DecisionModel()

result = model.run(
    state={
        "ideal_customer": "Data teams at companies with 200 to 5,000 employees",
        "lead": {
            "employees": 900,
            "contact_role": "Head of Data Platform",
            "message": "Our dbt models broke three times last month. Can we see a demo next week?",
        },
    },
    questions={
        "request": {
            "type": "choice",
            "instructions": "What is the `lead` asking for?",
            "criteria": {
                "evaluation": "A demo, a trial, pricing or a vendor evaluation",
                "question": "A product or technical question",
                "other": "Anything else",
            },
        },
        "fit": {
            "type": "score",
            "instructions": "How well does the `lead` match the `ideal_customer`?",
            "criteria": ["Not a fit", "Partial fit", "Strong fit"],
        },
        "decision_maker": {
            "type": "noul",
            "instructions": "The contact leads or manages a data team.",
        },
    },
)

answers = result["answers"]
print(answers["request"]["choice"], answers["request"]["confidence"])
print(answers["fit"]["score"], answers["fit"]["probabilities"])
print(answers["decision_maker"]["noul"])
print(result["usage"], model.calculate_cost(result["usage"])["total_cost"])

python first_call.py

import random

IDEAL_CUSTOMER = {
    "product": "Monitoring and alerting for data pipelines (Airflow, dbt, Spark)",
    "best_fit": [
        "Company with 200 to 5,000 employees",
        "Runs its own data pipelines in production",
        "Contact leads or manages a data or platform team",
        "Based in North America or Europe",
    ],
}

QUESTIONS = {
    "request": {
        "type": "choice",
        "instructions": "What is the `lead` asking for?",
        "criteria": {
            "evaluation": "A demo, a trial, pricing or a vendor evaluation",
            "question": "A product or technical question",
            "pitch": "Selling us a service, or looking for a job",
            "other": "Students, unsubscribes, spam and anything else",
        },
    },
    "fit": {
        "type": "score",
        "instructions": "How well does the company in the `lead` match the `ideal_customer`?",
        "criteria": [
            "Not a fit",
            "Weak fit",
            "Partial fit",
            "Strong fit",
            "Ideal customer",
        ],
    },
    "intent": {
        "type": "noul",
        "instructions": "The `lead` describes a concrete pipeline problem or a buying timeline.",
    },
    "decision_maker": {
        "type": "noul",
        "instructions": "The contact in the `lead` leads or manages a data or platform team.",
    },
}

WEIGHTS = {"fit": 0.45, "intent": 0.30, "decision_maker": 0.25}

MESSAGES = [
    ("We run about 400 Airflow DAGs and silent failures keep reaching our dashboards. We want a tool in place this quarter.", 3),
    ("Our dbt models broke three times last month before anyone noticed. Can we see a demo next week?", 3),
    ("We are evaluating pipeline monitoring vendors for a Q3 rollout across 12 Spark jobs. What does pricing look like at our size?", 3),
    ("How do you compare to the alerting we already get from Datadog?", 8),
    ("Do you support Dagster, or only Airflow?", 8),
    ("Saw your talk at a meetup. Interested, but no timeline yet.", 8),
    ("I'm a student writing a thesis on data quality. Is there a free license?", 10),
    ("We are an SEO agency and can get your site to page one of Google.", 12),
    ("Are you hiring data engineers? My resume is attached.", 10),
    ("Please remove me from your mailing list.", 10),
    ("Just browsing.", 12),
]
ROLES = [
    "VP of Data",
    "Head of Data Platform",
    "Data Engineering Manager",
    "Senior Data Engineer",
    "Analytics Engineer",
    "Founder",
    "Marketing Coordinator",
    "Recruiter",
    "Student",
]
INDUSTRIES = [
    "Fintech",
    "E-commerce",
    "Healthcare",
    "Logistics",
    "Media",
    "SaaS",
    "Marketing agency",
    "University",
]
COUNTRIES = [
    "United States",
    "Germany",
    "United Kingdom",
    "Canada",
    "Netherlands",
    "Brazil",
    "India",
    "Australia",
]
EMPLOYEES = [8, 25, 60, 150, 400, 900, 2500, 8000, 30000]


def make_leads(count: int, seed: int = 7) -> list:
    """
    Generate synthetic inbound leads, most of them poor fits.

    Args:
        count: Number of leads.
        seed: Random seed, so every run screens the same pool.

    Returns:
        Lead dictionaries.
    """
    rng = random.Random(seed)
    messages, weights = zip(*MESSAGES)
    return [
        {
            "id": f"L{index:04d}",
            "company_industry": rng.choice(INDUSTRIES),
            "employees": rng.choice(EMPLOYEES),
            "country": rng.choice(COUNTRIES),
            "contact_role": rng.choice(ROLES),
            "message": rng.choices(messages, weights)[0],
        }
        for index in range(count)
    ]


def lead_score(answers: dict) -> float:
    """
    Combine the decision model's answers into one score.

    Args:
        answers: Answers keyed by question id.

    Returns:
        A score from 0 to 1, or 0 when the lead is not a buyer.
    """
    if answers["request"]["choice"] in ("pitch", "other"):
        return 0.0
    fit_levels = len(QUESTIONS["fit"]["criteria"]) - 1
    values = {
        "fit": answers["fit"]["score"] / fit_levels,
        "intent": answers["intent"]["noul"],
        "decision_maker": answers["decision_maker"]["noul"],
    }
    return sum(WEIGHTS[key] * values[key] for key in WEIGHTS)

import asyncio
import json
import math
import time

import litellm

from swarms import Agent, DecisionModel, run_agents_with_different_tasks

from leads import IDEAL_CUSTOMER, QUESTIONS, lead_score, make_leads

NUM_LEADS = 200
SHORTLIST_SHARE = 0.05
DEEP_WORK_MODEL = "gpt-5.4"


async def screen_all(model: DecisionModel, leads: list) -> list:
    """
    Screen every lead with the decision model, 20 requests at a time.

    Args:
        model: Decision model that answers the screening questions.
        leads: Leads to screen.

    Returns:
        One result per lead with its score and latency.
    """
    limit = asyncio.Semaphore(20)

    async def screen(lead: dict) -> dict:
        async with limit:
            started = time.perf_counter()
            response = await model.arun(
                state={"ideal_customer": IDEAL_CUSTOMER, "lead": lead},
                questions=QUESTIONS,
            )
            return {
                "lead": lead,
                "score": lead_score(response["answers"]),
                "seconds": time.perf_counter() - started,
            }

    return await asyncio.gather(*(screen(lead) for lead in leads))


def llm_cost(model_name: str, usage: dict) -> float:
    """
    Price an agent's token usage with litellm's price table.

    Args:
        model_name: Model the agent ran on.
        usage: The agent's usage dictionary.

    Returns:
        Cost in US dollars.
    """
    prompt_cost, completion_cost = litellm.cost_per_token(
        model=model_name,
        prompt_tokens=usage["input_tokens"],
        completion_tokens=usage["output_tokens"],
    )
    return prompt_cost + completion_cost


leads = make_leads(NUM_LEADS)
model = DecisionModel()

started = time.perf_counter()
screened = asyncio.run(screen_all(model, leads))
screen_wall = time.perf_counter() - started
screen_cost = model.calculate_cost()

shortlist_size = math.ceil(NUM_LEADS * SHORTLIST_SHARE)
shortlist = sorted(screened, key=lambda r: r["score"], reverse=True)[
    :shortlist_size
]

print(f"Screened {NUM_LEADS} leads in {screen_wall:.1f} s")
print(
    f"  {screen_cost['input_tokens']} input tokens, "
    f"${screen_cost['total_cost']:.5f} total, "
    f"${screen_cost['total_cost'] / NUM_LEADS:.7f} per lead"
)
print(
    f"  {sum(r['seconds'] for r in screened) / NUM_LEADS:.2f} s per request"
)
print(f"\nTop 5 of the {shortlist_size}-lead shortlist:")
for result in shortlist[:5]:
    lead = result["lead"]
    print(
        f"  {result['score']:.2f}  {lead['contact_role']}, "
        f"{lead['employees']} employees, {lead['country']}: "
        f"{lead['message'][:50]}..."
    )

writers = [
    Agent(
        agent_name=f"Account-Executive-{r['lead']['id']}",
        system_prompt=(
            "You are an account executive. For the lead you are given, write "
            "a three-bullet account brief (why they fit, what to ask, what "
            "could block the deal), then a short first reply that refers to "
            "their message."
        ),
        model_name=DEEP_WORK_MODEL,
        max_loops=1,
        output_type="final",
        print_on=False,
    )
    for r in shortlist
]
started = time.perf_counter()
briefs = run_agents_with_different_tasks(
    [
        (
            writer,
            f"Ideal customer: {json.dumps(IDEAL_CUSTOMER)}\n\n"
            f"Lead: {json.dumps(r['lead'])}",
        )
        for writer, r in zip(writers, shortlist)
    ]
)
deep_wall = time.perf_counter() - started
deep_cost = sum(llm_cost(DEEP_WORK_MODEL, w.usage) for w in writers)

print(
    f"\nDeep work on {shortlist_size} leads with {DEEP_WORK_MODEL}: "
    f"{deep_wall:.1f} s, ${deep_cost:.4f}, "
    f"${deep_cost / shortlist_size:.4f} per lead"
)
print(
    f"Pipeline total: ${screen_cost['total_cost'] + deep_cost:.4f} "
    f"for {NUM_LEADS} leads"
)
print(f"\nBrief for {shortlist[0]['lead']['id']}:\n\n{briefs[0]}")

python screen.py

import json
import random
import time

import litellm

from swarms import Agent, DecisionModel

from leads import IDEAL_CUSTOMER, QUESTIONS, make_leads

SAMPLE_SIZE = 3
SCREEN_MODEL = "gpt-5.4-mini"

JSON_PROMPT = (
    "You screen inbound sales leads against an ideal customer profile. "
    "Reply with JSON only: "
    '{"request": "evaluation" | "question" | "pitch" | "other", '
    '"fit": 0-4 (0 not a fit, 4 ideal customer), '
    '"intent": 0-1 (the lead describes a concrete pipeline problem or buying timeline), '
    '"decision_maker": 0-1 (the contact leads or manages a data or platform team)}'
)
SCREENERS = {
    "LLM, JSON only": JSON_PROMPT,
    "LLM, JSON with reasons": JSON_PROMPT
    + ' Add a "reasons" field with one sentence per answer.',
}

sample = random.Random(1).sample(make_leads(200), SAMPLE_SIZE)
results = {}

model = DecisionModel()
seconds = []
for lead in sample:
    started = time.perf_counter()
    model.run(
        state={"ideal_customer": IDEAL_CUSTOMER, "lead": lead},
        questions=QUESTIONS,
    )
    seconds.append(time.perf_counter() - started)
results[f"Decision model ({model.model_name})"] = (
    model.calculate_cost()["total_cost"],
    sum(seconds),
)

for name, system_prompt in SCREENERS.items():
    cost, seconds = 0.0, 0.0
    for lead in sample:
        # A fresh agent per lead keeps earlier leads out of the context.
        screener = Agent(
            agent_name="Screener",
            system_prompt=system_prompt,
            model_name=SCREEN_MODEL,
            max_loops=1,
            output_type="final",
            print_on=False,
        )
        started = time.perf_counter()
        screener.run(
            f"Ideal customer: {json.dumps(IDEAL_CUSTOMER)}\n\n"
            f"Lead: {json.dumps(lead)}"
        )
        seconds += time.perf_counter() - started
        prompt_cost, completion_cost = litellm.cost_per_token(
            model=SCREEN_MODEL,
            prompt_tokens=screener.usage["input_tokens"],
            completion_tokens=screener.usage["output_tokens"],
        )
        cost += prompt_cost + completion_cost
    results[f"{name} ({SCREEN_MODEL})"] = (cost, seconds)

baseline_cost = results[f"Decision model ({model.model_name})"][0]
print(f"Per lead, measured on {SAMPLE_SIZE} leads:")
for name, (cost, seconds) in results.items():
    per_lead = cost / SAMPLE_SIZE
    print(
        f"  {name:<40} ${per_lead:.7f}  {seconds / SAMPLE_SIZE:.2f} s  "
        f"{cost / baseline_cost:5.1f}x cost  "
        f"${per_lead * 100_000:,.2f} per 100k leads"
    )

python compare.py