Swarms for Manufacturing

import os
from swarms_client import SwarmsClient

client = SwarmsClient(api_key=os.environ["SWARMS_API_KEY"])

# A MultiAgentRouter: the router reads the ticket and picks one specialist.
# Each specialist runs on the model that suits its work and returns a draft
# for the line lead or engineer to approve.
ticket = (
    "Ticket 4471, Line 2, Station 3 (paint booth), 14:20. Operator reports "
    "orange peel on 14 consecutive door panels, lot DP-2291. Booth humidity "
    "reading 71 percent against a 55 to 65 percent spec. Units moved to the hold rack."
)

response = client.swarms.run(
    name="Line 2 Ticket Desk",
    description="Route shop-floor tickets to the quality, maintenance or supply specialist.",
    swarm_type="MultiAgentRouter",
    task=ticket,
    agents=[
        {
            "agent_name": "Quality Engineer",
            "description": "Defects, nonconformances, holds and dispositions.",
            "system_prompt": (
                "You are a quality engineer on an automotive paint and assembly line. "
                "For a defect ticket, state the likely cause from the evidence given, "
                "the containment (what to hold, what to inspect), and a draft disposition "
                "(use as is, rework, scrap) marked FOR QUALITY MANAGER APPROVAL."
            ),
            "model_name": "claude-sonnet-5-5",  # careful reasoning on dispositions
            "max_loops": 1,
        },
        {
            "agent_name": "Maintenance Planner",
            "description": "Equipment faults, sensor flags and work orders.",
            "system_prompt": (
                "You are a maintenance planner. For an equipment ticket, draft a work "
                "order: asset, symptom, probable failure mode, parts likely needed, "
                "priority P1 to P3 with the reason, and the shift window you suggest."
            ),
            "model_name": "gpt-5.4",
            "max_loops": 1,
        },
        {
            "agent_name": "Supply Analyst",
            "description": "Late lots, shortages and supplier notices.",
            "system_prompt": (
                "You are a supply chain analyst. For a material ticket, state days of "
                "cover from the inventory figures given, the options (expedite, "
                "alternate supplier, reschedule) and draft the note to the supplier."
            ),
            "model_name": "gemini-2.5-flash",  # fast, cheap for volume tickets
            "max_loops": 1,
        },
    ],
    max_loops=1,
)

print(response)

curl -X POST https://api.swarms.world/v1/swarm/completions \
  -H "x-api-key: $SWARMS_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Line 2 Ticket Desk",
    "description": "Route shop-floor tickets to the quality, maintenance or supply specialist.",
    "swarm_type": "MultiAgentRouter",
    "task": "Ticket 4472, Line 2, Station 1 (stamping press P-2), 15:05. Vibration sensor on the main drive bearing reads 7.1 mm/s RMS, up from 4.2 mm/s last week. Alarm threshold is 7.5 mm/s. Press is running. 2,400 panels left on the shift order.",
    "agents": [
      {
        "agent_name": "Quality Engineer",
        "system_prompt": "You are a quality engineer. For a defect ticket, state likely cause, containment and a draft disposition marked FOR QUALITY MANAGER APPROVAL.",
        "model_name": "claude-sonnet-5-5",
        "max_loops": 1
      },
      {
        "agent_name": "Maintenance Planner",
        "system_prompt": "You are a maintenance planner. For an equipment ticket, draft a work order with asset, symptom, probable failure mode, parts, priority P1 to P3 with reason, and a suggested shift window.",
        "model_name": "gpt-5.4",
        "max_loops": 1
      },
      {
        "agent_name": "Supply Analyst",
        "system_prompt": "You are a supply chain analyst. For a material ticket, state days of cover, the options, and draft the supplier note.",
        "model_name": "gemini-2.5-flash",
        "max_loops": 1
      }
    ],
    "max_loops": 1
  }'

# The router sends this ticket to the Maintenance Planner. The response has
# "output" (the drafted work order) and "usage" (input_tokens, output_tokens,
# total_tokens), and the run appears in Swarms Cloud with the route it took.

// Supplier risk review as a ConcurrentWorkflow: three agents score the same
// supplier from different evidence at the same time, and the sourcing team
// gets all three reads in one response.
const SWARMS_API = "https://api.swarms.world/v1/swarm/completions"

type Supplier = {
  name: string
  part: string
  onTimePct12m: number
  ppm12m: number
  openScars: number
  notes: string
}

const scorer = (name: string, model: string, lens: string) => ({
  agent_name: name,
  system_prompt:
    "You assess suppliers for a tier-one automotive plant. Your lens: " + lens + " " +
    "Give a risk rating LOW, MEDIUM or HIGH on the first line, then the evidence you used " +
    "and the one question the sourcing team should ask this supplier.",
  model_name: model,
  max_loops: 1,
})

export async function reviewSupplier(s: Supplier) {
  const res = await fetch(SWARMS_API, {
    method: "POST",
    headers: {
      "x-api-key": process.env.SWARMS_API_KEY!,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({
      name: "Supplier Risk Review",
      description: "Delivery, quality and financial reads on one supplier, in parallel.",
      swarm_type: "ConcurrentWorkflow",
      task:
        "Supplier: " + s.name + ", part family: " + s.part + ".\n" +
        "On-time delivery, trailing 12 months: " + s.onTimePct12m + " percent.\n" +
        "Defects, trailing 12 months: " + s.ppm12m + " PPM. Open SCARs: " + s.openScars + ".\n" +
        "Notes: " + s.notes,
      agents: [
        scorer("Delivery Agent", "gpt-4.1-mini", "on-time history, lead time changes and expedite frequency."),
        scorer("Quality Agent", "claude-haiku-5-5", "PPM trend, open corrective actions and repeat failure modes."),
        scorer("Financial Agent", "gpt-5.4", "signs of financial stress in the notes and any public information you were given."),
      ],
      max_loops: 1,
    }),
  })

  if (!res.ok) throw new Error("Swarms API " + res.status + ": " + (await res.text()))
  const data = await res.json()
  return { review: data.output as string, usage: data.usage }
}

reviewSupplier({
  name: "Northfield Fasteners",
  part: "M8 and M10 flange bolts",
  onTimePct12m: 91.4,
  ppm12m: 380,
  openScars: 2,
  notes: "Two late lots in the last quarter, both attributed to a plating subcontractor change. Payment terms extended from 30 to 60 days at their request in August.",
}).then((r) => {
  console.log(r.review)
  console.log(r.usage)
})

import os
import sys

import requests

BASE = "https://api.swarms.world"
HEADERS = {"x-api-key": os.environ["SWARMS_API_KEY"], "Content-Type": "application/json"}

# The nightly job reviews every asset flagged by the condition monitoring
# system. Before it starts, check the Swarms Cloud credit balance and the
# day's usage so a plant does not find out about an empty account at 03:00.
credits = requests.get(f"{BASE}/v1/account/credits", headers=HEADERS, timeout=30).json()
metrics = requests.get(f"{BASE}/v1/metrics/summary", headers=HEADERS, timeout=30).json()

print("credit balance:", credits)
print("usage summary:", metrics)

MIN_CREDITS = float(os.environ.get("MIN_CREDITS", "25"))


def balance_of(doc) -> float:
    """Find the first numeric credit figure in the response, whatever the nesting."""
    if isinstance(doc, dict):
        for key in ("credits", "credit", "balance", "remaining_credits"):
            if isinstance(doc.get(key), (int, float)):
                return float(doc[key])
        for value in doc.values():
            found = balance_of(value)
            if found is not None:
                return found
    return None


balance = balance_of(credits)
if balance is not None and balance < MIN_CREDITS:
    print(f"balance {balance:.2f} below {MIN_CREDITS:.2f}, skipping the nightly review")
    sys.exit(1)

with open("flagged_assets_tonight.txt", encoding="utf-8") as f:
    flagged = f.read()

review = {
    "name": "Nightly Maintenance Review",
    "description": "Vibration, thermal and history reads on each flagged asset, aggregated.",
    "swarm_type": "MixtureOfAgents",
    "task": "Review each flagged asset below and recommend a priority and window for the planner.\n\n" + flagged,
    "agents": [
        {"agent_name": "Vibration Agent", "system_prompt": "You read vibration trends (RMS, peak, spectrum notes) and name the probable failure mode.", "model_name": "gpt-4.1-mini", "max_loops": 1},
        {"agent_name": "Thermal Agent", "system_prompt": "You read bearing and motor temperatures against ambient and rating and flag abnormal rise.", "model_name": "claude-haiku-5-5", "max_loops": 1},
        {"agent_name": "History Agent", "system_prompt": "You read the maintenance history and note repeat failures and overdue PMs.", "model_name": "gemini-2.5-flash", "max_loops": 1},
        {"agent_name": "Aggregator", "system_prompt": "You are the reliability engineer. Combine the three reads into one recommendation per asset: priority P1 to P3, suggested window, parts. Mark it FOR PLANNER REVIEW.", "model_name": "gpt-5.4", "max_loops": 1},
    ],
    "max_loops": 1,
}

resp = requests.post(f"{BASE}/v1/swarm/completions", headers=HEADERS, json=review, timeout=600)
resp.raise_for_status()
data = resp.json()
print(data["output"])
print(data["usage"])