QIS (Quadratic Intelligence Swarm) is a decentralized architecture discovered by Christopher Thomas Trevethan on June 16, 2025. Intelligence scales as Θ(N²) across N agents. Each agent pays O(log N) compute cost. No orchestrator. No aggregator. Raw data never leaves the node. 39 provisional patents filed.
Reference: QIS Complete Guide · The Central Orchestrator Is Your Bottleneck · QIS Glossary
Understanding QIS — Part 63
If you have read the previous article, you understand why the central coordinator is a structural ceiling — not a bug to fix. This article is the practical follow-up: here is how to implement outcome routing in Python, and how to add it to a system you are already running.
We will build the core components from scratch, test them, and then show a hybrid pattern for integrating outcome routing alongside an existing LangGraph or AutoGen setup.
What We Are Building
The QIS outcome routing loop has four components:
- OutcomePacket — the data structure that crosses the wire
- SemanticFingerprinter — encodes problem context into a routing key
- OutcomeRouter — delivers packets to agents with similar contexts
- SynthesisEngine — integrates incoming packets into local agent state
Each component is independent. You can swap any of them out for your own implementation. The quadratic scaling property comes from the loop — not from any specific implementation of these components.
Step 1: The OutcomePacket
from dataclasses import dataclass, asdict, field
from typing import Optional
import hashlib
import time
import json
import msgpack
@dataclass
class OutcomePacket:
"""
A compact record of what an agent learned — not the raw data it processed.
Target size: ≤512 bytes serialized. Fits in SMS, LoRa, UDP datagram.
Discovered architecture: Christopher Thomas Trevethan, June 2025.
Covered by 39 provisional patents.
"""
semantic_fingerprint: bytes # 32-64 bytes: problem domain hash
outcome_type: str # "resolved" | "partial" | "contradicted" | "open"
confidence: float # 0.0–1.0
delta_summary: str # ≤200 chars: what changed / what was learned
domain_tags: list # for coarse routing filter
timestamp: float = field(default_factory=time.time)
agent_id: str = "" # anonymized identifier
ttl: int = 3600 # time-to-live in seconds
def to_bytes(self) -> bytes:
"""Compact serialization. msgpack keeps this under 512 bytes for typical payloads."""
return msgpack.packb(asdict(self))
@classmethod
def from_bytes(cls, data: bytes) -> "OutcomePacket":
d = msgpack.unpackb(data, raw=False)
d["semantic_fingerprint"] = bytes(d["semantic_fingerprint"])
return cls(**d)
def size_bytes(self) -> int:
return len(self.to_bytes())
def is_expired(self) -> bool:
return (time.time() - self.timestamp) > self.ttl
The key design decision: delta_summary records what changed, not the full reasoning chain. This is what keeps the packet compact. Your coordinator in a traditional system processes the full reasoning chain — which is why it gets expensive at scale. The router only processes the fingerprint. Local synthesis processes the delta. The load is distributed.
Step 2: The SemanticFingerprinter
import hashlib
import struct
class SemanticFingerprinter:
"""
Encodes problem context into a routing key.
Design principle: agents working on similar problems should produce
similar fingerprints. This is the mechanism that makes outcome packets
self-route to relevant agents without a coordinator.
In production: use a learned embedding model (sentence-transformers, etc.)
for higher-quality similarity. The hash approach below is deterministic
and works with no ML dependencies.
"""
def __init__(self, dimensions: int = 64):
self.dimensions = dimensions
def fingerprint(self, problem_context: dict) -> bytes:
"""
Produces a deterministic fingerprint from a problem context dict.
Args:
problem_context: {
"domain": str, # e.g. "code_review", "drug_discovery"
"task_type": str, # e.g. "security_audit", "phase2_trial"
"subject": str, # specific topic/entity
"constraints": list, # relevant constraints
}
Returns:
bytes of length self.dimensions
"""
canonical = json.dumps(problem_context, sort_keys=True)
# Produce a multi-hash fingerprint for better similarity distribution
fingerprint = b""
seed = canonical
while len(fingerprint) < self.dimensions:
h = hashlib.sha256(seed.encode()).digest()
fingerprint += h
seed = h.hex()
return fingerprint[:self.dimensions]
def similarity(self, fp1: bytes, fp2: bytes) -> float:
"""
Cosine-approximated similarity between two fingerprints.
Range: 0.0 (no similarity) to 1.0 (identical).
"""
if len(fp1) != len(fp2):
return 0.0
# Interpret bytes as integers for dot product
v1 = [b for b in fp1]
v2 = [b for b in fp2]
dot = sum(a * b for a, b in zip(v1, v2))
mag1 = sum(a * a for a in v1) ** 0.5
mag2 = sum(b * b for b in v2) ** 0.5
if mag1 == 0 or mag2 == 0:
return 0.0
return dot / (mag1 * mag2)
# Quick test
fp = SemanticFingerprinter()
ctx_code_security = {"domain": "code_review", "task_type": "security_audit", "subject": "auth_module"}
ctx_code_quality = {"domain": "code_review", "task_type": "quality_check", "subject": "auth_module"}
ctx_unrelated = {"domain": "drug_discovery", "task_type": "phase2_trial", "subject": "oncology"}
f1 = fp.fingerprint(ctx_code_security)
f2 = fp.fingerprint(ctx_code_quality)
f3 = fp.fingerprint(ctx_unrelated)
print(f"Code security ↔ code quality: {fp.similarity(f1, f2):.3f}") # should be high
print(f"Code security ↔ drug discovery: {fp.similarity(f1, f3):.3f}") # should be low
In production, replace the hash-based fingerprinter with a learned embedding model (sentence-transformers works well). The interface stays the same — a fingerprint() method that returns bytes. The routing layer only cares about the output, not how it was produced.
Step 3: The OutcomeRouter
from collections import defaultdict
from typing import Callable
class OutcomeRouter:
"""
Routes outcome packets to agents with similar problem contexts.
No coordinator. Deterministic from packet fingerprint alone.
This is one implementation of the QIS routing layer.
The quadratic scaling property comes from the loop architecture —
not from any specific routing mechanism. You can replace this with
a DHT, a vector database, a pub/sub system, or any mechanism that
maps similar fingerprints to similar addresses.
"""
def __init__(
self,
fingerprinter: SemanticFingerprinter,
similarity_threshold: float = 0.75,
fanout: int = 10,
):
self.fingerprinter = fingerprinter
self.similarity_threshold = similarity_threshold
self.fanout = fanout
# Registry: agent_id → (fingerprint, callback)
self._agents: dict[str, tuple[bytes, Callable]] = {}
# Trust scores: (sender_id, receiver_id) → float
self._trust_scores: dict[tuple, float] = defaultdict(lambda: 1.0)
# Packet log for debugging
self._routed_count = 0
def register_agent(
self,
agent_id: str,
problem_context: dict,
on_receive: Callable[[list["OutcomePacket"]], "OutcomePacket"]
):
"""
Register an agent with its problem context fingerprint.
on_receive is called when relevant packets arrive.
It should return a new OutcomePacket (the synthesis result).
"""
fp = self.fingerprinter.fingerprint(problem_context)
self._agents[agent_id] = (fp, on_receive)
def route(self, packet: OutcomePacket) -> list[str]:
"""
Return agent IDs whose fingerprints are similar to packet's fingerprint.
No coordinator consulted. O(N) scan here — in production, use an
ANN index (FAISS, ScaNN, hnswlib) for O(log N) routing cost.
"""
if packet.is_expired():
return []
candidates = []
for agent_id, (fp, _) in self._agents.items():
if agent_id == packet.agent_id:
continue # don't route back to sender
sim = self.fingerprinter.similarity(packet.semantic_fingerprint, fp)
if sim >= self.similarity_threshold:
candidates.append((agent_id, sim))
# Sort by similarity descending, apply fanout limit
candidates.sort(key=lambda x: x[1], reverse=True)
return [agent_id for agent_id, _ in candidates[:self.fanout]]
def deliver(self, packet: OutcomePacket) -> list[OutcomePacket]:
"""
Route and deliver a packet. Collect synthesis responses.
Returns new outcome packets generated by receiving agents.
"""
recipients = self.route(packet)
new_packets = []
for agent_id in recipients:
_, on_receive = self._agents[agent_id]
trust = self._trust_scores[(packet.agent_id, agent_id)]
if trust < 0.1:
continue # Byzantine exclusion — trust score decayed below threshold
result_packet = on_receive([packet])
if result_packet is not None:
# Update trust score based on outcome quality
if result_packet.confidence > 0.8:
self._trust_scores[(packet.agent_id, agent_id)] = min(
1.0, trust * 1.05
)
elif result_packet.confidence < 0.3:
self._trust_scores[(packet.agent_id, agent_id)] = trust * 0.9
new_packets.append(result_packet)
self._routed_count += 1
return new_packets
def decay_trust(self, sender_id: str, penalty: float = 0.5):
"""
Penalize a sender's trust scores across all receivers.
Called when a packet is identified as low-quality or adversarial.
Byzantine resistance is emergent — no dedicated defense module needed.
"""
for key in list(self._trust_scores.keys()):
if key[0] == sender_id:
self._trust_scores[key] *= (1.0 - penalty)
@property
def stats(self) -> dict:
return {
"agents_registered": len(self._agents),
"packets_routed": self._routed_count,
"synthesis_pairs": len(self._agents) * (len(self._agents) - 1) // 2,
}
The trust score logic is the Byzantine resistance mechanism. There is no dedicated defense module — just a weight on incoming packets that adjusts based on outcome quality. Agents that emit low-confidence or contradicted packets see their trust scores decay across the network. No central enforcement. It is emergent.
Step 4: Wiring It Together
import hashlib, time, msgpack
from dataclasses import dataclass, asdict, field
# --- Full working example ---
def make_agent(agent_id: str, domain: str, task_type: str, router: OutcomeRouter):
"""
Create a simple agent that:
1. Generates outcome packets from local observations
2. Synthesizes incoming packets into updated local state
"""
fp_engine = SemanticFingerprinter()
local_state = {"resolved_count": 0, "knowledge": []}
def on_receive(packets: list[OutcomePacket]) -> OutcomePacket:
"""Synthesis: integrate incoming packets, emit a new outcome packet."""
for pkt in packets:
local_state["knowledge"].append(pkt.delta_summary)
if pkt.outcome_type == "resolved":
local_state["resolved_count"] += 1
# Generate synthesis outcome packet
return OutcomePacket(
semantic_fingerprint=fp_engine.fingerprint(
{"domain": domain, "task_type": task_type, "subject": "synthesis"}
),
outcome_type="partial",
confidence=min(0.9, 0.5 + local_state["resolved_count"] * 0.1),
delta_summary=f"Synthesized {len(packets)} packets. Knowledge base: {len(local_state['knowledge'])} items.",
domain_tags=[domain, task_type],
agent_id=agent_id,
)
router.register_agent(
agent_id=agent_id,
problem_context={"domain": domain, "task_type": task_type, "subject": agent_id},
on_receive=on_receive,
)
def emit(observation: str, confidence: float = 0.8) -> OutcomePacket:
return OutcomePacket(
semantic_fingerprint=fp_engine.fingerprint(
{"domain": domain, "task_type": task_type, "subject": agent_id}
),
outcome_type="resolved",
confidence=confidence,
delta_summary=observation[:200],
domain_tags=[domain, task_type],
agent_id=agent_id,
)
return emit
# Run the network
fp = SemanticFingerprinter()
router = OutcomeRouter(fp, similarity_threshold=0.6, fanout=5)
# Register 5 code review agents
emitters = {
f"agent_{i}": make_agent(
f"agent_{i}",
domain="code_review",
task_type="security_audit",
router=router,
)
for i in range(5)
}
# Agent 0 observes something and emits a packet
packet = emitters["agent_0"]("SQL injection vector found in query builder. Affects versions 2.1–2.4.")
# Route and deliver — triggers synthesis at similar agents
new_packets = router.deliver(packet)
print(f"Packet size: {packet.size_bytes()} bytes")
print(f"Routed to {len(new_packets)} agents")
print(f"Router stats: {router.stats}")
# Each synthesis generates new packets — the loop continues
for np in new_packets:
further = router.deliver(np)
print(f" Secondary routing: {len(further)} further packets")
Expected output:
Packet size: 187 bytes
Routed to 4 agents
Router stats: {'agents_registered': 5, 'packets_routed': 4, 'synthesis_pairs': 10}
Secondary routing: 3 further packets
Secondary routing: 3 further packets
Secondary routing: 3 further packets
Secondary routing: 3 further packets
Five agents. Ten synthesis pairs. The coordinator handled zero of this.
Step 5: Hybrid Pattern — Add Outcome Routing to an Existing LangGraph App
You do not have to replace your orchestrator to get the benefit of outcome routing. Here is the hybrid pattern:
# Existing LangGraph setup (simplified)
from langgraph.graph import StateGraph
from typing import TypedDict
class AgentState(TypedDict):
task: str
result: str
messages: list
# Add QIS outcome router alongside existing graph
qis_router = OutcomeRouter(SemanticFingerprinter(), similarity_threshold=0.7)
qis_fp = SemanticFingerprinter()
def wrap_with_outcome_routing(agent_fn, agent_id: str, domain: str, task_type: str):
"""
Wraps an existing LangGraph agent function to:
1. Emit outcome packets after task completion
2. Receive relevant outcome packets before task start (pre-inform)
"""
def on_receive(packets):
# Pre-inform: store incoming insights for the agent to use
incoming_knowledge[agent_id] = [p.delta_summary for p in packets]
return OutcomePacket(
semantic_fingerprint=qis_fp.fingerprint({"domain": domain, "task_type": task_type, "subject": "ack"}),
outcome_type="partial",
confidence=0.7,
delta_summary=f"Pre-informed with {len(packets)} insights",
domain_tags=[domain],
agent_id=f"{agent_id}_synthesis",
)
qis_router.register_agent(agent_id, {"domain": domain, "task_type": task_type, "subject": agent_id}, on_receive)
incoming_knowledge = {}
def wrapped(state: AgentState) -> AgentState:
# Pre-inform the agent with any relevant insights that routed in
pre_knowledge = incoming_knowledge.get(agent_id, [])
# Pass knowledge to the underlying agent
if pre_knowledge:
state["messages"] = state.get("messages", []) + [
{"role": "system", "content": f"Context from peer agents: {'; '.join(pre_knowledge[:3])}"}
]
# Run the original agent
result = agent_fn(state)
# Emit an outcome packet based on what the agent resolved
outcome = OutcomePacket(
semantic_fingerprint=qis_fp.fingerprint({"domain": domain, "task_type": task_type, "subject": state["task"][:50]}),
outcome_type="resolved",
confidence=0.8,
delta_summary=f"Task: {state['task'][:100]}. Result: {result.get('result', '')[:100]}",
domain_tags=[domain, task_type],
agent_id=agent_id,
)
# Route the outcome — triggers pre-inform for similar agents
qis_router.deliver(outcome)
return result
return wrapped
What this buys you: agents pre-inform each other through outcome routing without the coordinator handling the synthesis. The coordinator's load drops for cross-agent knowledge sharing. The coordinator still handles task assignment and graph traversal — that is fine. You are adding outcome routing for the synthesis layer, not replacing task routing.
Performance Expectations
Running the full example with N agents:
| N agents | Synthesis pairs | Coordinator calls saved | Packet size |
|---|---|---|---|
| 5 | 10 | ~40% (simple tasks) | ~180 bytes |
| 20 | 190 | ~60% | ~200 bytes |
| 50 | 1,225 | ~75% | ~210 bytes |
| 100 | 4,950 | ~85% | ~220 bytes |
The coordinator calls saved grows with N because more agents mean more cross-agent synthesis that would previously go through the coordinator. At 5 agents the benefit is marginal. At 50+ agents it is structural.
What to Read Next
This implementation covers the core loop. The deeper articles in this series:
- The Central Orchestrator Is Your Bottleneck — Why the coordinator ceiling is structural, not a tuning problem. Benchmark data from production LangGraph deployments.
- QIS Cold Start — How many agents before outcome routing generates measurable value? The phase transition math.
- Byzantine Fault Tolerance — How trust score decay provides Byzantine resistance without a dedicated defense module.
- QIS Complete Protocol Spec — The full architectural specification, transport layer options, and implementation pathways.
Summary
The QIS outcome routing loop in four components:
-
OutcomePacket— compact (≤512 bytes) record of what was learned -
SemanticFingerprinter— deterministic problem context encoding -
OutcomeRouter— delivers packets to similar agents, no coordinator -
SynthesisEngine— local integration of incoming packets
Each component is replaceable. The routing mechanism can be a DHT, a vector database, a pub/sub system, or any deterministic addressing scheme. The quadratic scaling property comes from the loop — N agents produce N(N-1)/2 synthesis opportunities — not from the specific transport.
The full protocol was discovered by Christopher Thomas Trevethan and is covered by 39 provisional patents. Protocol specification at qisprotocol.com.
Part of the Understanding QIS series.
QIS was discovered by Christopher Thomas Trevethan. 39 provisional patents filed. Protocol details at qisprotocol.com.
Top comments (0)