Graph Workflows for Production Pipelines

pip install requests python-dotenv

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",
}

[BuyerResearcher]   ──┐
[CompanyResearcher] ──┼──> [PitchSynthesizer] ──> [LeadScorer]
[TriggerResearcher] ──┘

agents = [
    {
        "agent_name": "BuyerResearcher",
        "description": "Researches the individual buyer and their role.",
        "system_prompt": (
            "You are a sales researcher specializing in buyer personas. Given a "
            "name, title, and company, produce: (1) role responsibilities, "
            "(2) likely KPIs, (3) recent public statements or content, "
            "(4) seniority and likely budget authority. Be concise."
        ),
        "model_name": "gpt-4.1",
        "max_tokens": 3000,
        "temperature": 0.2,
        "max_loops": 1,
    },
    {
        "agent_name": "CompanyResearcher",
        "description": "Researches the target company.",
        "system_prompt": (
            "You are a company research analyst. Given a company name, produce: "
            "(1) what they do in one sentence, (2) headcount and stage, "
            "(3) likely tech stack, (4) recent funding or M&A. Be concise."
        ),
        "model_name": "gpt-4.1",
        "max_tokens": 3000,
        "temperature": 0.2,
        "max_loops": 1,
    },
    {
        "agent_name": "TriggerResearcher",
        "description": "Finds a recent trigger event worth referencing.",
        "system_prompt": (
            "You are a trigger-event researcher. Given a company, find ONE "
            "recent (last 60 days) public event that creates a natural opening "
            "for outbound: a hire, a launch, a layoff, a funding round, a "
            "press release. Output a single trigger with date and source."
        ),
        "model_name": "gpt-4.1",
        "max_tokens": 2000,
        "temperature": 0.3,
        "max_loops": 1,
    },
    {
        "agent_name": "PitchSynthesizer",
        "description": "Synthesizes buyer + company + trigger into a 1-line hook.",
        "system_prompt": (
            "You are a senior SDR. Given buyer research, company research, and "
            "a trigger event, write a single outbound opener of at most 35 "
            "words that (a) references the trigger, (b) connects it to a KPI "
            "the buyer owns, (c) ends with a clear question. No fluff, no "
            "'hope you're doing well.'"
        ),
        "model_name": "gpt-4.1",
        "max_tokens": 1500,
        "temperature": 0.5,
        "max_loops": 1,
    },
    {
        "agent_name": "LeadScorer",
        "description": "Scores the lead 1-100 with a confidence band.",
        "system_prompt": (
            "You are a revenue operations analyst. Given the buyer profile, "
            "the company profile, the trigger event, and the synthesized "
            "pitch, return a JSON object with fields: score (1-100), "
            "confidence ('low'|'medium'|'high'), tier ('A'|'B'|'C'), and "
            "one_line_rationale. Return ONLY the JSON object."
        ),
        "model_name": "gpt-4.1",
        "max_tokens": 800,
        "temperature": 0.1,
        "max_loops": 1,
    },
]

edges = [
    {"source": "BuyerResearcher",   "target": "PitchSynthesizer"},
    {"source": "CompanyResearcher", "target": "PitchSynthesizer"},
    {"source": "TriggerResearcher", "target": "PitchSynthesizer"},
    {"source": "PitchSynthesizer",  "target": "LeadScorer"},
]

LEAD = {
    "buyer_name": "Jamie Chen",
    "buyer_title": "VP of Engineering",
    "company": "Northwind Analytics",
}

task = (
    f"Lead: {LEAD['buyer_name']}, {LEAD['buyer_title']} at {LEAD['company']}. "
    "Run the full enrichment pipeline: buyer research, company research, "
    "trigger event search, pitch synthesis, and final lead scoring."
)

workflow_input = {
    "name": "SDR-Enrichment-Pipeline",
    "description": (
        "Fan-out enrichment DAG. Three researchers run in parallel, converge "
        "into a pitch synthesizer, then feed a lead scorer."
    ),
    "agents": agents,
    "edges": edges,
    "entry_points": ["BuyerResearcher", "CompanyResearcher", "TriggerResearcher"],
    "end_points": ["LeadScorer"],
    "max_loops": 1,
    "task": task,
    "auto_compile": True,
    "verbose": False,
}

response = requests.post(
    f"{BASE_URL}/v1/graph-workflow/completions",
    headers=headers,
    json=workflow_input,
    timeout=300,
)
response.raise_for_status()
result = response.json()

outputs = result.get("outputs", {})

stages = [
    ("PARALLEL RESEARCH", ["BuyerResearcher", "CompanyResearcher", "TriggerResearcher"]),
    ("SYNTHESIS",         ["PitchSynthesizer"]),
    ("SCORING",           ["LeadScorer"]),
]

for stage_label, names in stages:
    print(f"\n{'=' * 60}\n{stage_label}\n{'=' * 60}")
    for name in names:
        out = outputs.get(name, "")
        if isinstance(out, list):
            out = "\n".join(str(x) for x in out)
        print(f"\n[{name}]\n{str(out)[:400]}")

# The scorer returned JSON — parse it
import re
score_raw = outputs.get("LeadScorer", "")
if isinstance(score_raw, list):
    score_raw = " ".join(str(x) for x in score_raw)

match = re.search(r"\{.*\}", str(score_raw), re.DOTALL)
if match:
    lead_score = json.loads(match.group(0))
    print(f"\nFinal score: {lead_score['score']} ({lead_score['tier']}-tier, "
          f"{lead_score['confidence']} confidence)")

usage = result.get("usage", {})
print(f"\nTotal cost: ${usage.get('token_cost', 0):.4f}")
print(f"Execution time: {result.get('execution_time', 0):.1f}s")

agents[2]["max_loops"] = 2  # TriggerResearcher gets one retry