A step-by-step tutorial for Google's gemini-decision-1 in the open-source Swarms framework: ask yes/no, choice and score questions with DecisionModel, get calibrated probabilities back, check images, track cost, and route Swarms agents on the answers.
Most agent systems spend a surprising share of their tokens on small judgments: which agent should take this, is this urgent, is the answer good enough to stop. A generative model answers those in prose, and your code then has to parse the prose. Google's Gemini Decision 1 (gemini-decision-1) is built for exactly these judgments. You send it some input and a set of typed questions, and it returns calibrated probabilities with zero output tokens.
The open-source Swarms framework now speaks to it through DecisionModel, the same class that already covers TypeSafe, Cloudflare and OpenAI decision models. In this tutorial you will go from one yes/no question to a router that picks which Swarms agent handles a request. Each step is a short script you can run.
The full API is in the DecisionModel reference, and a longer worked example is on the Gemini decision model example page.
| Step | What it adds |
|---|---|
| 1 to 2 | Install Swarms and set one key |
| 3 to 6 | The three question types: yes/no, choice, score |
| 7 to 8 | Several questions in one request, then thresholds in code |
| 9 | A question about an image |
| 10 to 11 | Many items at once with async, and what it cost |
| 12 | A router that sends each request to the right Swarms agent |
Gemini support in DecisionModel is on the master branch of Swarms and will ship in the next release. Until then, install from GitHub:
pip install -U git+https://github.com/kyegomez/swarms.gitThere is nothing else to install. DecisionModel calls Gemini's HTTP API directly, so you do not need the google-genai package.
Create a Gemini API key in Google AI Studio and put it in your environment or in a .env file. Swarms loads .env when it is imported.
export GEMINI_API_KEY="..."The model name picks the provider. Any name that starts with gemini-decision goes to Gemini's Decisions API and reads GEMINI_API_KEY.
Create yes_no.py:
from swarms import DecisionModel
model = DecisionModel(model_name="gemini-decision-1")
probability = model.noul(
state="The blue mug I ordered on Monday arrived cracked on Wednesday.",
instructions="Did the item arrive damaged?",
)
print(f"Damaged: {probability:.2f}")noul() returns one float: the probability that the answer is yes. For this message you should see a value close to 1.
Two things to know about that number. It is calibrated, so 0.8 is meant to be right about 80% of the time. And Gemini keeps yes/no probabilities between 0.01 and 0.99, so a check for > 0.99 never passes.
Short questions are often ambiguous. Is a customer who dislikes the colour reporting a defect? Give each side a description with criteria:
from swarms import DecisionModel
model = DecisionModel(model_name="gemini-decision-1")
probability = model.noul(
state="I returned the sweater because I prefer green over blue.",
instructions="Is the customer reporting a product defect?",
criteria={
"true": "The product itself is broken or faulty.",
"false": "A preference, sizing, shipping or pricing complaint.",
},
)
print(f"Defect: {probability:.2f}")DecisionModel uses the keys true and false for every provider and converts them to the names Gemini expects, so the same question works unchanged on another decision model.
choice() picks one option from a fixed set. The criteria map each option to a description of when it applies:
from swarms import DecisionModel
model = DecisionModel(model_name="gemini-decision-1")
answer = model.choice(
state="I was charged twice for my order.",
instructions="Which department should handle this request?",
criteria={
"billing": "Payments, invoices and refunds.",
"technical": "Problems using the product.",
"shipping": "Delivery and tracking.",
"other": "Anything outside these categories.",
},
)
print(answer["choice"], answer["confidence"])
print(answer["probabilities"])The answer is a dict with this shape (your values will differ):
{
"type": "choice",
"choice": "billing",
"probabilities": {"billing": 0.97, "technical": 0.01, "shipping": 0.01, "other": 0.01},
"confidence": 0.97
}A choice question always spreads its probability across your options, even when none fits. Include a catch-all such as other with a real description, and check confidence before acting.
score() rates the input against ordered levels, from lowest to highest:
from swarms import DecisionModel
model = DecisionModel(model_name="gemini-decision-1")
answer = model.score(
state="Export fails in Safari but works in Chrome.",
instructions="How severe is this issue for the user?",
criteria=[
"Cosmetic: appearance only, no lost functionality.",
"Workaround available: a task fails, but another way works.",
"Fully blocked: a task fails with no workaround.",
],
)
print(f"Severity: {answer['score']:.2f} on a 0-2 scale")
print(answer["probabilities"]){
"type": "score",
"score": 1.1,
"legend": {"0": "Cosmetic: ...", "1": "Workaround available: ...", "2": "Fully blocked: ..."},
"probabilities": {"0": 0.05, "1": 0.8, "2": 0.15},
"confidence": 0.8
}score is the probability-weighted level, so it can sit between two levels. With three levels the scale runs from 0 to 2. Use it to rank or filter, and read probabilities when you need a single level.
The helper methods send one question each. run() sends as many as you like about the same input, in one request. Gemini answers each question on its own, so adding a question does not change the other answers, and one request can carry up to 1,000 of them.
Create triage.py:
from swarms import DecisionModel
model = DecisionModel(model_name="gemini-decision-1")
ticket = {
"message": "The blue mug I ordered on Monday arrived cracked on Wednesday.",
"order": {"id": "8821", "age_days": 2},
}
result = model.run(
state=ticket,
questions={
"queue": {
"type": "choice",
"instructions": "Which support queue should handle this message?",
"criteria": {
"billing": "Payment, invoice or refund questions.",
"replacements": "Damaged, missing or wrong items.",
"general": "Product questions and account updates.",
},
},
"severity": {
"type": "score",
"instructions": "How severe is this issue for the customer?",
"criteria": [
"Minor: the item is usable as it is.",
"Moderate: the item is usable with a workaround.",
"Severe: the item cannot be used.",
],
},
"is_damaged": {
"type": "noul",
"instructions": "Did the item arrive damaged?",
},
"within_window": {
"type": "noul",
"instructions": "Is order.age_days 30 or less?",
},
},
)
answers = result["answers"]
queue = answers["queue"]
print(f"Queue: {queue['choice']} (confidence {queue['confidence']:.2f})")
print(f"Severity: {answers['severity']['score']:.2f} on a 0-2 scale")
print(f"Damaged: {answers['is_damaged']['noul']:.2f}")
print(f"Within window: {answers['within_window']['noul']:.2f}")The state here is a dict. DecisionModel sends it to Gemini as JSON text, which is why within_window can refer to a field by its key path, order.age_days.
The model gives you numbers. The policy stays in your code, where you can read it and test it. Add this to the end of triage.py:
approved = (
queue["choice"] == "replacements"
and queue["confidence"] >= 0.7
and answers["is_damaged"]["noul"] >= 0.8
and answers["within_window"]["noul"] >= 0.8
)
if approved:
print("Action: send a replacement")
else:
print("Action: send to a human")Ask narrow questions and combine them with ordinary if statements. Set the thresholds from labelled examples of your own traffic, weighing what a wrong approval costs against a wrong escalation.
Gemini Decision 1 is multimodal. Pass the state as a list of content blocks and it goes to Gemini unchanged. Put a photo.jpg next to the script:
import base64
from pathlib import Path
from swarms import DecisionModel
model = DecisionModel(model_name="gemini-decision-1")
photo = base64.b64encode(Path("photo.jpg").read_bytes()).decode("ascii")
probability = model.noul(
state=[
{"type": "text", "text": "Customer photo for return request #8821."},
{"type": "image", "data": photo, "mime_type": "image/jpeg"},
],
instructions="Does the photo show physical damage to the item?",
)
print(f"Visible damage: {probability:.2f}")The same list accepts audio, video and document (PDF) blocks. Always set mime_type to the real file format. Question instructions are text only.
Each call is one small HTTP request, so screening a batch is a job for arun() and asyncio.gather:
import asyncio
from swarms import DecisionModel
model = DecisionModel(model_name="gemini-decision-1")
reviews = [
"Arrived on time and works perfectly.",
"The lid does not close. I want my money back.",
"Fine for the price, though the colour is duller than the photo.",
]
questions = {
"sentiment": {
"type": "choice",
"instructions": "What is the overall sentiment of this review?",
"criteria": {
"positive": "The customer is satisfied.",
"neutral": "Mixed or no clear feeling.",
"negative": "The customer is dissatisfied.",
},
},
"wants_refund": {
"type": "noul",
"instructions": "Is the customer asking for a refund?",
},
}
async def main() -> None:
results = await asyncio.gather(
*(model.arun(review, questions) for review in reviews)
)
for review, result in zip(reviews, results):
answers = result["answers"]
print(
f"{answers['sentiment']['choice']:<9}"
f"refund {answers['wants_refund']['noul']:.2f} {review}"
)
asyncio.run(main())For a few hundred items, wrap the calls in an asyncio.Semaphore so you stay inside your rate limit. DecisionModel retries rate limits and overloads on its own, with backoff, up to max_retries times.
Every DecisionModel counts the tokens its provider reports. Add this to the end of any script above:
print(model.usage)
print(model.get_price())
print(model.calculate_cost())usage is the input and output tokens summed over every request the instance has made. get_price() returns US dollars per million tokens, which for gemini-decision-1 is 0.10 for input and 0 for output: the model produces no output tokens, so you pay for input only. calculate_cost() multiplies the two.
The shared input is billed once per request however many questions you ask, which is another reason to group questions in one run() call.
Now put it to work in a multi-agent system. A decision model makes a good router: it is fast, it costs a fraction of a generative call, and it tells you how sure it is. The agents below run on Gemini too, through the same GEMINI_API_KEY.
Create route.py:
from swarms import Agent, DecisionModel
agents = [
Agent(
agent_name="Billing-Agent",
agent_description="Refunds, invoices, failed payments and subscription changes.",
system_prompt="You resolve billing questions clearly and briefly.",
model_name="gemini/gemini-2.5-pro",
max_loops=1,
),
Agent(
agent_name="Technical-Agent",
agent_description="Bugs, API errors, integrations and outages.",
system_prompt="You debug technical problems step by step.",
model_name="gemini/gemini-2.5-pro",
max_loops=1,
),
Agent(
agent_name="Sales-Agent",
agent_description="Pricing, plan upgrades and new accounts.",
system_prompt="You answer pricing and plan questions.",
model_name="gemini/gemini-2.5-pro",
max_loops=1,
),
]
agents_by_name = {agent.agent_name: agent for agent in agents}
router = DecisionModel(model_name="gemini-decision-1")
def route(task: str) -> str:
answers = router.run(
state=task,
questions={
"agent": {
"type": "choice",
"instructions": "Which agent should handle this request?",
"criteria": {
agent.agent_name: agent.agent_description
for agent in agents
},
},
"in_scope": {
"type": "noul",
"instructions": "Is this a request a software company's support team should handle?",
},
},
)["answers"]
if answers["in_scope"]["noul"] < 0.5:
return "Declined: out of scope for support."
decision = answers["agent"]
if decision["confidence"] < 0.6:
return "Escalated to a human: the router is unsure who owns this."
print(f"Routing to {decision['choice']}")
return agents_by_name[decision["choice"]].run(task)
print(route("I was charged twice for my March invoice, can you refund one?"))The options are built from the agents themselves: each agent_name is an option and its agent_description is the description. Add an agent to the list and the router can pick it. A second question checks that the request is in scope at all, and a confidence floor sends unclear requests to a person. Only then does a generative agent run.
The same pattern works wherever a swarm needs a judgment between steps: stopping a debate when a round adds nothing new, scoring every reply in a group chat before it posts, or choosing the next worker in a hierarchy. The reference page has an example of each.
Questions and answers have the same shape for every provider, so changing the model is a one-line change:
model = DecisionModel(model_name="gemini-decision-1") # Google, GEMINI_API_KEY
model = DecisionModel(model_name="gpt-6-luna") # OpenAI, OPENAI_API_KEY
model = DecisionModel(model_name="clef") # Cloudflare Workers AI
model = DecisionModel() # TypeSafe jev-latestThe exception is media. The list of content blocks in Step 9 is Gemini's format; the Providers section of the reference shows how each provider takes images.
DecisionModel raises as httpx.HTTPStatusError with the reason in the message.queue, severity), so write the whole question in instructions.
A hands-on comparison of the three ways to run multi-agent systems without hosting them yourself: the Swarms API, Amazon Bedrock Agents and AgentCore, and Google's Vertex AI Agent Builder with ADK. Same task on each, with architectures, model choice, streaming, batch, pricing and time to first result.
Run a team of AI agents inside any no-code automation. One HTTP Request node, one JSON body, and the Swarms API returns the finished work. Step-by-step setup for n8n, Zapier and Make, plus three ready-to-paste recipes: lead enrichment, support triage and a weekly report.
A complete tutorial for the hosted Swarms MCP server at api.swarms.world/mcp. Connect Claude Code, Claude Desktop, Cursor or your own Python and TypeScript agents in two lines, learn the 23 tools, run your first agent and swarm from a chat prompt, read results and costs, and fix the errors you will hit on day one.