Multi-agent swarms sound exotic until you count the parts. A useful research swarm is a planner that breaks a question into subquestions, several workers that answer them in parallel, and a synthesizer that merges the results into one document. That is a to-do list, a thread pool and a prompt. You do not need a framework to build it, and at current off-peak pricing you do not need a budget either.
This tutorial builds the whole thing with the Python standard library. The only external requirement is a DeepSeek API key. Off-peak rates for DeepSeek-V4.1-Flash are fifteen cents per million input tokens and sixty cents per million output tokens, with cached input drastically cheaper, so a six-agent research run typically lands in the cents. We will put a hard ceiling on it anyway, because that is how you keep a swarm from surprising you.
Step 1: the only dependency you need
The DeepSeek API is OpenAI-compatible, so a single HTTP call does the work. Stdlib urllib is enough. Two details matter: set a timeout so one hung request cannot stall the pool, and read the token usage from the response so you can track spend.
import json, os, time, urllib.request, urllib.error
API = 'https://api.deepseek.com/chat/completions'
KEY = os.environ['DEEPSEEK_API_KEY']
MODEL = 'deepseek-v4.1-flash'
def call(messages, max_tokens=1200, retries=3):
body = json.dumps({'model': MODEL, 'messages': messages,
'max_tokens': max_tokens,
'temperature': 0.3}).encode()
for attempt in range(retries):
req = urllib.request.Request(API, data=body, headers={
'Content-Type': 'application/json',
'Authorization': 'Bearer ' + KEY})
try:
with urllib.request.urlopen(req, timeout=180) as r:
d = json.loads(r.read())
return d['choices'][0]['message']['content'], d['usage']
except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError):
if attempt == retries - 1:
raise
time.sleep(2 ** attempt)
Three short lines of discipline live in that function. Exponential backoff handles rate limits without a retry library. Returning usage gives us a cost meter. Raising on the final attempt instead of returning an empty string keeps silent failures out of the final report.
Step 2: a planner that produces a to-do list
Asking one model to 'research everything' produces a shallow essay. Asking it to split the question into four independent, non-overlapping subquestions produces four useful answers. The constraint that matters is independence: overlapping subquestions waste tokens and produce a synthesis that repeats itself.
PLANNER = '''You are a research planner. Break the user question into 4
independent subquestions that can be answered in parallel. No overlap.
Return JSON only: {"subquestions": ["...", "...", "...", "..."]}'''
def plan(question):
text, usage = call([{'role': 'system', 'content': PLANNER},
{'role': 'user', 'content': question}], max_tokens=400)
text = text.strip().removeprefix('```json').removesuffix('```').strip()
return json.loads(text)['subquestions'], usage
Stripping the markdown fence before json.loads is the difference between a script that runs and one that crashes on the second call. Keep the planner at low temperature and a small token cap; planning is cheap, and a planner that writes essays is a planner you are paying by the foot.
Step 3: four workers in parallel, with a cost cap
Python's concurrent.futures is exactly the right tool here. Cap the worker count so you do not trip a rate limit, and check the running cost before each dispatch so the swarm stops cleanly instead of draining your balance. The tracker below is deliberately a tiny class with a lock, which is all a single-process script needs.
import threading
from concurrent.futures import ThreadPoolExecutor
PRICES = {'in': 0.15 / 1_000_000, 'out': 0.60 / 1_000_000} # off-peak USD/token
class Budget:
def __init__(self, cap):
self.cap, self.spent, self.lock = cap, 0.0, threading.Lock()
def charge(self, usage):
with self.lock:
self.spent += usage['prompt_tokens'] * PRICES['in'] \\
+ usage['completion_tokens'] * PRICES['out']
return self.spent
BUDGET = Budget(cap=0.25)
WORKER = '''Answer the subquestion using only well-established knowledge.
Write 150-250 words. Be specific, no preamble, no hedging.'''
def research(subq):
text, usage = call([{'role': 'system', 'content': WORKER},
{'role': 'user', 'content': subq}])
spent = BUDGET.charge(usage)
print(f' done: {subq[:48]}... spend=${spent:.4f}')
return {'subquestion': subq, 'answer': text}
Run the workers with a pool of four and collect results as they finish. If spend crosses the cap, stop scheduling new work and keep the answers you already have. A partial report that costs twenty cents beats a complete one that costs five dollars you did not sanction.
def run_workers(subquestions, workers=4):
results = []
with ThreadPoolExecutor(max_workers=workers) as pool:
futures = [pool.submit(research, q) for q in subquestions]
for f in futures:
if BUDGET.spent > BUDGET.cap:
f.cancel()
continue
results.append(f.result())
return results
Step 4: a synthesizer that cites its inputs
The synthesizer is where most swarms quietly undo their own work. Left alone, a merge step rewrites everything into generic prose and invents facts that were never in the worker output. The fix is a strict instruction to use only the supplied material and to name which subquestion each claim came from. That also makes the output auditable.
SYNTH = '''Merge the research notes below into one report with:
1. A 3-sentence executive summary
2. One section per subquestion, each prefixed with its subquestion
3. A final section listing disagreements or gaps between notes
Use ONLY the notes. If a fact is not in the notes, omit it.'''
def synthesize(question, results):
notes = '\\n\\n'.join(f"Q: {r['subquestion']}\\nA: {r['answer']}" for r in results)
text, usage = call([{'role': 'system', 'content': SYNTH},
{'role': 'user', 'content': question + '\\n\\nNotes:\\n' + notes}],
max_tokens=2000)
BUDGET.charge(usage)
return text
Step 5: wire it together and check it
One function, one print of the cost, one file out. The assertions below are the whole test suite, and they are enough: they fail if the planner returns the wrong shape, if any worker came back empty, or if the report is not real text.
def swarm(question, out='report.md'):
subs, usage = plan(question)
BUDGET.charge(usage)
assert isinstance(subs, list) and len(subs) >= 3, subs
results = run_workers(subs)
assert results and all(r['answer'].strip() for r in results)
report = synthesize(question, results)
open(out, 'w').write('# ' + question + '\\n\\n' + report + '\\n')
print(f'total spend: ${BUDGET.spent:.4f} -> {out}')
return report
if __name__ == '__main__':
swarm('What changed in agent security tooling this quarter?')
What to tune, and what will break
- Worker count: four to six is the sweet spot. More workers mostly increase rate-limit retries and the amount of near-duplicate text the synthesizer has to deduplicate.
- Prompt caching: keep the system prompts byte-identical across runs. Static prefixes hit the cache and cut input cost sharply; a timestamp inside the system message throws that away.
- Context growth: give each worker only its own subquestion. Workers that see the whole conversation produce overlapping answers and burn tokens restating the brief.
- Failure mode to watch: if two subquestions are secretly the same question, the report repeats a section. Tighten the planner instruction to 'non-overlapping' and, if it persists, deduplicate subquestions by similarity before dispatch.
Rough cost for a real run: a small planner call, four worker calls of a few hundred tokens each, and one larger synthesis, which lands well under ten cents off-peak at current rates. That is the practical change of 2026 in one script: work that used to justify a framework, a queue and a budget review now fits in one file, runs in under a minute, and tells you exactly what it spent.
