Formation Control for Agent Swarms: From Math to Mission

You have a fleet of drones that needs to sweep a field as a single unit. They have to hold a shape, slide past a tree without touching it or each other, and then snap back into formation — and by the end of the day, someone has to sign off that every one of those claims is true, mission by mission. That is the whole problem in one sentence. Formation control is the discipline of making "we all move as one" a guarantee built out of simple local rules rather than a hope.

The system this post dissects is a real, runnable implementation: an agricultural survey swarm whose five drones are individual agents following their own control laws, planned and reviewed by an LLM, and audited against a five-stage flight-spec (FDS) checklist. The code is open — the notebook lives at formation_control_poc_v2.ipynb — and every line below is reproduced from it, so you can follow along with the numbers in your own head and the formulas in the actual source.

One Swarm, One Question

Work through a mission from the operator's seat. A 100 m × 80 m wheat field needs a precision survey. There is exactly one hazard: a tree at (20, 40) with a 3.5 m radius. The fleet leader starts at (20, 10) and flies up the field at a constant 2.5 m/s; four followers trail behind in formation.

Three questions decide everything about this mission:

1. How many drones, in what shape, at what spacing? Nobody wants to hand-code this for every field — the answer depends on the field, the obstacle, the coverage requirement. This is a strategic decision.
2. How does one drone know where to be? Each drone is an independent agent. It has no map, no central controller, and no knowledge of the full swarm. It can only sense the world locally. Everything about the flight — holding formation, avoiding the tree, not colliding — must emerge from rules each drone runs by itself.
3. How do you prove the flight was safe? After the mission you need a report a person can read: formation held, obstacle dodged, no collisions, formation recovered.

This implementation answers question 1 and question 3 with an LLM, and question 2 with pure math. That division — strategy and audit by a model, execution and safety by math — is the design idea that holds the whole system together, and the post is an anatomy of why it works.

The Mental Model: Who Does What

Before any code, fix three pieces in your head:

• Agents — the drones. Each DroneAgent is a small piece of state (where I am, my velocity, which teammate I follow, where I should sit relative to them) plus one rule that runs every tick: sense the world nearby, decide the forces to apply, move. That sense–evaluate–act loop is the entire "brain" of an agent.
• The planner — a language model (Gemini). It reads the mission brief and returns a typed decision: how many drones (3–8), which formation (line, v_shape, or grid), what spacing (8–12 m), and a justification. It decides what the swarm should look like; it never touches a flight decision.
• The audit layer — a second language-model role plus a deterministic fallback. It reads the mission's measured metrics and stamps a five-stage compliance verdict. The verdict is typed too, so a machine can file it.

These three layers are wired together by a state machine — a pipeline of named steps through which one typed "mission state" object flows. The state machine is worth a short motivation before the details, because it is why questions 1–3 can even be answered after the fact.

Why a Pipeline Instead of a Loop

A mission could be written as one big function: plan, simulate, measure, save, judge — five calls in a row, done. That works once. What a pipeline gives you is observability and resumability at each named step:

• Typed state. Every step reads and writes one MissionState object — a whiteboard with labeled, typed slots. You can inspect it after any step, because it is the interface.
• Named nodes. Each step is a function with a name. A log line like [dispatch_simulation] Launching 5 drones... tells you exactly where a mission is, and the state tells you what it did.
• Checkpointing. The pipeline is compiled with a checkpointer and a thread_id. A mission becomes a resumable conversation: if a run dies partway, it can be replayed from the last checkpoint rather than restarted from zero.
• Confined LLM roles. The model is registered as an explicit node in two places and nowhere else. The safety-critical parts of the pipeline contain no prompt at all.

None of this requires LangGraph specifically — the pattern transfers to any explicit workflow framework. But the notebook you're reading uses LangGraph's StateGraph, and its six nodes are worth seeing as a map before diving into the internals.

The Management Pipeline

Read the picture left to right, then down. Four blue nodes are deterministic — same input always produces the same output, pure code, no model involved. Two amber nodes are the language model. The graph starts, assesses the mission, asks the planner for a swarm design, simulates the flight, analyzes the telemetry, saves the metrics to a database, asks the auditor for a compliance verdict, and ends. All six nodes share one MissionState whiteboard.

The rest of the post opens each of these internals — starting with the setup, then the agent, then the swarm, then the pipeline mechanics — explaining not just what each line does, but why the line exists at all.

Cell by Cell

Setup: Imports

In one cell, the notebook pulls in everything it will use:

import os
import math
import time
import json
import sqlite3
import uuid
from datetime import datetime, timezone
from typing import TypedDict, Optional, List, Tuple, Dict, Any

import numpy as np
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
from IPython.display import Image, display
from pydantic import BaseModel, Field

import mlflow
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain_google_genai import ChatGoogleGenerativeAI

print("All dependencies loaded successfully.")

Group the imports by what they're for, and the setup stops being noise:

• Standard library — os (files and environment variables), math (square roots — the whole control law leans on them), json (serializing data to text and back), sqlite3 (the fleet's tiny file-based database), uuid (a unique ID per mission run), datetime (timestamps), typing (type hints that document intent).
• Numerics and plotting — numpy for fast array math and matplotlib for the trajectory plot. The Agg backend is the headless one: "draw to a file, never open a window." The plot is saved to docs/swarm_trajectory_telemetry.png and displayed later; there is no interactive window in a notebook run.
• Structure and validation — pydantic (BaseModel, Field) is next in line and does real work: it defines rigid, typed schemas — think "forms with required fields" — that the LLM must fill in. That is the mechanism that stops a model from returning free-text garbage into the pipeline.
• Orchestration and models — langchain/langgraph provide tools, messages, and the StateGraph; ChatGoogleGenerativeAI is the Gemini client.

The block ends with a print statement that tells you the environment is healthy — a cheap fail-fast signal. If this cell errors, you're missing a package or a Python version, and there is no point running anything else.

The First Model Call

llm = ChatGoogleGenerativeAI(
    model="gemini-3.5-flash-lite"
)
llm.invoke("hi")

Two lines, and they're a wasted-proof-of-life: create a Gemini client for the small, fast gemini-3.5-flash-lite model and send "hi". Why burn a model call this early? To fail fast at the cheapest moment. The API key check, the network path, the model name — all of it is validated in milliseconds here, long before the mission graph exists. If this is broken, nothing downstream matters, and you found out at step zero. This is a habit worth copying: the first thing an agent system should do is prove its connection, not start doing work.

Environment and Tracking

The real configuration happens next — a .env file with secrets, a Databricks MLflow experiment for tracing, and the Gemini client with a fallback ladder:

# Load .env → MLFLOW_TRACKING_URI=databricks, MLFLOW_EXPERIMENT_ID, GOOGLE_API_KEY, DATABRICKS_*
load_dotenv(".env", override=True)

MLFLOW_EXPERIMENT_ID = os.environ.get("MLFLOW_EXPERIMENT_ID", "3192447675404693")
mlflow.set_tracking_uri(os.environ.get("MLFLOW_TRACKING_URI", "databricks"))
experiment = mlflow.set_experiment(experiment_id=MLFLOW_EXPERIMENT_ID)
print(f"[MLflow] Connected → Experiment ID: {MLFLOW_EXPERIMENT_ID} on Databricks")

# Autolog captures all LangChain/LangGraph LLM calls, tool calls, and traces automatically
mlflow.langchain.autolog()

# Gemini LLM setup
GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY", os.environ.get("GOOGLE_API_KEY", ""))
CHAT_MODEL_NAME = "gemini-3.5-flash-lite"

llm = None
if GEMINI_API_KEY:
    try:
        llm = init_chat_model(
            CHAT_MODEL_NAME,
            model_provider="google_genai",
            api_key=GEMINI_API_KEY,
            timeout=60,
        )
        print(f"[Google Gemini] Model: {CHAT_MODEL_NAME} | Status: Connected")
    except Exception as e:
        print(f"[Google Gemini] Init note: {e}. Deterministic fallback active.")
else:
    print("[Google Gemini] GOOGLE_API_KEY not set — deterministic fallback mode.")

The pieces, with their mechanisms:

• .env and load_dotenv — a plain key-value file (GEMINI_API_KEY=..., MLFLOW_TRACKING_URI=databricks, and so on) that the notebook reads into environment variables. Secrets live in a file that is never committed to git, never hard-coded. override=True means the file wins over anything already in the environment.

• MLflow on Databricks — MLflow is an experiment-tracking server; "Databricks" is this deployment's tracking URI. MLFLOW_EXPERIMENT_ID is a folder on that server where runs accumulate. The real magic is mlflow.langchain.autolog(): with that one line, every LLM call, tool call, and pipeline step — inputs, outputs, timing, latency — is recorded automatically into that experiment folder. That's the observability half of "pipeline instead of loop": the trace is the flight recorder, written without a single manual logging call.

• The fallback ladder — this is the first real design pattern, so look carefully:

  • If GEMINI_API_KEY is missing → llm stays None and the notebook tells you it's in deterministic fallback mode.
  • If the key exists but the client constructor throws → same None, same message.
  • If neither — the client is live and used.

  Because both model nodes later check if llm is None, a missing or broken LLM turns the mission into a fully deterministic run instead of a crash. The mechanism is: the fallback is a lane, not an exception path. Every agent system you build should degrade along this ladder — model unavailable → rule-based default — never reach an unhandled error.

The Agent's Alphabet: Vector Math

Every position, offset, velocity, and force in the entire system is one small class:

class Vector:
    """2D vector for drone positions and offsets."""
    def __init__(self, x: float, y: float):
        self.x = x
        self.y = y

    def __add__(self, other):  return Vector(self.x + other.x, self.y + other.y)
    def __sub__(self, other):  return Vector(self.x - other.x, self.y - other.y)
    def __mul__(self, scalar): return Vector(self.x * scalar, self.y * scalar)
    def __repr__(self):        return f"Vector({self.x:.2f}, {self.y:.2f})"

    def norm(self):
        return math.sqrt(self.x ** 2 + self.y ** 2)

    def normalized(self):
        n = self.norm()
        return Vector(self.x / n, self.y / n) if n > 1e-9 else Vector(0.0, 0.0)

    def clamp(self, max_speed: float):
        n = self.norm()
        if n > max_speed:
            return self.normalized() * max_speed
        return Vector(self.x, self.y)

def NORM(v: Vector) -> float:
    return v.norm()

print("Vector class defined.")

A Vector is an arrow: a direction and a length, captured as two numbers. That's the whole abstraction, and it's enough because everything a drone thinks about is an arrow. Its position is an arrow from the field origin. Its velocity is an arrow saying where it's going and how fast. The gap between where it is and where it should be is an arrow. Forces are arrows pushing it around.

The operator overloads (__add__, __sub__, __mul__) let those arrows be combined with plain +, -, and × signs, so the control law reads like math instead of like component shuffling. Three methods are the entire vocabulary:

• norm() — the arrow's length (Pythagoras: √(x² + y²)). Every distance in the system — "how far to the goal," "how close to the tree," "how close to a teammate" — is a norm.
• normalized() — the direction of the arrow with length exactly 1. The only purpose is to say "keep pointing where you're pointing, but in a pure direction" so you can then multiply that direction by any strength you want. The n > 1e-9 guard matters more than it looks: without it, a drone sitting exactly on its target would divide by zero and the simulation would explode at the worst possible moment.
• clamp(max_speed) — if the arrow is longer than the speed limit, shrink it to the limit while keeping its direction. This is the traffic governor, and it shows up at the end of every velocity update in the control law: forces can compute whatever they want, the clamp is the hard guarantee that nothing ever exceeds 5 m/s.

NORM is a one-line alias for norm — nothing more than a shorter name for the tight code inside the control loop where this function runs every tick.

What the Planner Is Allowed to Decide

Now the contracts that guard the boundary between the LLM and the physics:

# ── Pydantic: LLM's swarm dispatch decision ────────────────────────────────
class SwarmDispatchPlan(BaseModel):
    num_drones: int   = Field(description="Number of drones to dispatch (3–8)")
    formation_type: str = Field(description="Formation layout: 'line', 'v_shape', or 'grid'")
    spacing_m: float  = Field(description="Inter-drone spacing in metres (recommended 8–12)")
    justification: str = Field(description="Reasoning behind this swarm configuration")

# ── Pydantic: FDS compliance verdict ──────────────────────────────────────
class FDSStageVerdict(BaseModel):
    stage_name: str = Field(description="Name of the FDS verification stage")
    passed: bool    = Field(description="Whether this stage passed")
    details: str    = Field(description="Technical details of the stage evaluation")

class FDSVerificationVerdict(BaseModel):
    formation_rule_verified:      FDSStageVerdict = Field(description="Stage 1: 10m grid spacing")
    coordinated_movement_verified: FDSStageVerdict = Field(description="Stage 2: Swarm follows leader")
    dynamic_adaptation_verified:  FDSStageVerdict = Field(description="Stage 3: Obstacle avoidance")
    self_organization_verified:   FDSStageVerdict = Field(description="Stage 4: Peer yielding")
    re_formation_verified:        FDSStageVerdict = Field(description="Stage 5: Grid re-convergence")
    overall_mission_success: bool = Field(description="True if all stages passed")
    executive_summary: str        = Field(description="Comprehensive technical summary")

A Pydantic BaseModel is a form with typed, required fields. The mechanism matters: when the LLM is asked to "return a structured plan," the framework hands the model the schema of this form — field names, types, and every Field(description=...) — and the model's output is validated against it before your code ever sees it. If the model returns formation_type: "diamond_octagon", the parse fails and the pipeline knows it. Without this, an LLM decision could silently derail an entire mission.

The two forms define the two LLM roles:

• SwarmDispatchPlan — the planner's decision. num_drones (an integer, constrained 3–8 by the prompt), formation_type (one of exactly three strings), spacing_m (metres), and justification — the model's reasoning, saved verbatim. The justification field is not fluff: it is how a human later audits why the swarm was shaped this way.
• FDSVerificationVerdict — the auditor's report. It nests five FDSStageVerdict report cards (one per spec stage), each with a name, a boolean pass/fail, and technical details, plus an overall success flag and an executive summary. A nested schema like this is how you get richly structured output instead of prose you'd have to parse.

What the model cannot do is as important as what it can: it cannot choose a formation that doesn't exist, cannot omit the justification, cannot return 12 drones. The schema is the enforcement mechanism.

The Shared Whiteboard: MissionState

Between the contracts and the code lives the pipeline's one shared object:

# ── LangGraph shared state TypedDict ───────────────────────────────────────
class MissionState(TypedDict):
    # Mission inputs
    mission_brief:   str
    field_width_m:   float
    field_height_m:  float
    obstacles:       List[dict]          # [{x, y, radius}]
    # Agent decision (plan_swarm node)
    dispatch_plan:   dict                # SwarmDispatchPlan.model_dump()
    # Simulation output (dispatch_simulation node)
    sim_telemetry_json: str
    # Analysis output (analyze_telemetry node)
    analysis_metrics: dict
    # Persistence output (persist_telemetry node)
    run_id: str
    # Verdict output (generate_verdict node)
    verdict: dict                        # FDSVerificationVerdict.model_dump()

print("Pydantic schemas and MissionState TypedDict defined.")

MissionState is a typed dictionary — the whiteboard with labeled slots. Read it as a contract between the nodes:

• Inputs at the top (mission brief, field dimensions, obstacle list) — filled in before the mission starts.
• dispatch_plan — written by plan_swarm.
• sim_telemetry_json — written by dispatch_simulation.
• analysis_metrics — written by analyze_telemetry.
• run_id — written by persist_telemetry.
• verdict — written by generate_verdict.

The comments next to each field say which node writes it — documentation carried inside the type itself. And note everything arrives as plain JSON-safe values (dicts and strings), never custom objects: that keeps the state cheap to checkpoint, cheap to trace, and cheap to persist. From here on, every linked name printed by the framework — "the graph" — is really just this whiteboard passing through six functions.

The Agent Inside: DroneAgent and Its Control Law

This is the heart of the post — the internal mechanism that makes a collection of dumb bots into a formation. Two classes first:

class CircularObstacle:
    def __init__(self, x: float, y: float, radius: float):
        self.position = Vector(x, y)
        self.radius = radius

class DroneAgent:
    def __init__(
        self,
        agent_id: str,
        initial_position: Vector,
        designated_offset: Vector,
        neighbor_id: Optional[str] = None,
        kp: float = 1.4,
        tolerance: float = 0.2,
        max_speed: float = 5.0
    ):
        self.agent_id = agent_id
        self.position = initial_position
        self.velocity = Vector(0.0, 0.0)
        self.DESIGNATED_OFFSET = designated_offset
        self.neighbor_id = neighbor_id
        self.kp = kp
        self.TOLERANCE = tolerance
        self.max_speed = max_speed
        self.is_avoiding = False
        self.trajectory: List[Tuple[float, float]] = [(self.position.x, self.position.y)]

    def update_control_loop(self, dt: float, swarm: Dict[str, 'DroneAgent'], obstacles: List[CircularObstacle]):
        # Obstacle avoidance
        f_obs = Vector(0.0, 0.0)
        self.is_avoiding = False
        for obs in obstacles:
            delta = self.position - obs.position
            dist = delta.norm()
            safety_margin = obs.radius + 3.0
            if dist < safety_margin:
                self.is_avoiding = True
                repulsion = delta.normalized() * (safety_margin - dist) * 5.0
                tangent = Vector(-delta.normalized().y, delta.normalized().x) * 4.0
                f_obs = f_obs + repulsion + tangent

        # Formation control
        if self.neighbor_id and self.neighbor_id in swarm:
            neighbor_pos = swarm[self.neighbor_id].position
            desired_position = neighbor_pos + self.DESIGNATED_OFFSET
            position_error = desired_position - self.position

            # Peer yielding (self-organization)
            f_peer = Vector(0.0, 0.0)
            for peer_id, peer in swarm.items():
                if peer_id != self.agent_id:
                    p_delta = self.position - peer.position
                    p_dist = p_delta.norm()
                    if p_dist < 4.5 and p_dist > 1e-3:
                        f_peer = f_peer + p_delta.normalized() * (4.5 - p_dist) * 3.0

            if NORM(position_error) > self.TOLERANCE:
                adj = position_error * self.kp
                if self.is_avoiding:
                    self.velocity = (self.velocity + (f_obs * 1.5 + adj * 0.2 + f_peer) * dt).clamp(self.max_speed)
                else:
                    self.velocity = (self.velocity + (adj + f_peer) * dt).clamp(self.max_speed)
            else:
                if self.is_avoiding:
                    self.velocity = (self.velocity + f_obs * dt).clamp(self.max_speed)
                else:
                    self.velocity = self.velocity * 0.98
        else:
            if self.is_avoiding:
                self.velocity = (self.velocity + f_obs * dt).clamp(self.max_speed)

        self.position = self.position + self.velocity * dt
        self.trajectory.append((self.position.x, self.position.y))

CircularObstacle is simply a tree: a center position and a radius. DroneAgent is the agent. Read its constructor as an inventory of everything an agent is allowed to know about itself and the world:

• agent_id — its name (Drone_A…Drone_E).
• position, velocity — its physical state: where it is, how fast and in which direction it's moving.
• DESIGNATED_OFFSET — where it should sit relative to its neighbor. This is one of only two facts it holds about the team.
• neighbor_id — which teammate it follows. The other fact. Together these two fields are the entire knowledge an agent has of the swarm: it knows one other drone and one relative position. It has no map, no list of teammates, no shared plan. That minimalism is the decentralization premise, and it's what makes the swarm scalable: if every agent had to know every other agent, costs would grow with the team; with one neighbor, they stay flat.
• kp, TOLERANCE, max_speed — three tuning knobs: how strongly it pulls toward its spot, how close is "close enough," and the speed limit.
• is_avoiding — a mode flag set by sensing, which changes which control law runs.
• trajectory — breadcrumbs: every position it has ever visited, for later analysis and plotting.

Now the mechanism itself, update_control_loop(dt, swarm, obstacles). This function runs once per tick of the simulation clock (dt = 0.1 s). Each tick is "sense → evaluate → act": read what's nearby, decide forces, update velocity, update position. Work through the four internal mechanisms in order.

Mechanism 1 — sensing and avoiding the obstacle. The drone measures its distance to every obstacle and compares it with a safety margin: the tree's radius plus 3.0 m of flight buffer, so the danger zone around the tree at (20, 40) is a 6.5 m circle. Inside that circle, is_avoiding turns on and the drone builds a two-part force:

• Repulsion — straight away from the tree, strength (margin − dist) × 5.0 — i.e. equal to how deep inside the danger zone it is, times 5. The mechanism inside the mechanism: the deeper the penetration, the harder the push, so the drone can never be forced further in.
• Tangent — perpendicular to the repulsion direction, strength 4.0. This is the subtle line. Why does a tangent exist? Run the numbers without it: a drone that only gets pushed straight away stops dead in front of the tree and sits there, because the formation pull keeps directing it into the tree. The tangent force is the sideways slide that converts "stop" into "skirt around," preserving the forward motion that real flight needs. Push away to stay safe, slide sideways to keep moving.

Mechanism 2 — holding formation (a proportional controller). The drone's goal is the moving point neighbor.position + DESIGNATED_OFFSET — the leader's current position plus the drone's fixed slot relative to it. Each tick it computes the error arrow from where it is to where it must be, and applies a proportional correction: the further away, the harder it pulls.

$adj=e\cdot k_p,e=p_{neighbor}+{\delta }_{offset}-p_{self},k_p=1.4$

This is the same law as easing into a parking spot: far away, press the accelerator hard; as you close, ease off; the pull is proportional to the gap. The TOLERANCE = 0.2 m dead band and the drag term (velocity × 0.98 when in-range) are why the controller can get away with proportional-only gain: inside 0.2 m the pull switches off entirely (no more correction), and drag bleeds off residual motion. Without the dead band, the pull-too-hard/pull-too-little cycle around the goal would make the drone oscillate — a jittering hover. The dead band trades "perfectly centered" for "settled and still."

Mechanism 3 — peer yielding (self-organization). Around every drone is an invisible 4.5 m personal-space bubble. When a teammate enters the bubble, the drone pushes it away with strength (4.5 − dist) × 3.0. Same pattern as the obstacle: the deeper the intrusion, the harder the push. Why does this guarantee global spacing with only local knowledge? Because every drone applies it to every nearby peer, the pairwise pushes make "everyone at least ~4.5 m apart" an emergent property — nobody computes the global arrangement, so the cost per agent stays O(N) and the swarm can scale to hundreds without a traffic controller.

Mechanism 4 — the avoidance reweighting (emergency mode). Watch the velocity branches when is_avoiding is true: the obstacle force is multiplied by 1.5 and the formation pull by 0.2. The mechanism is deliberate: while dodging, geometry is allowed to bend (20% formation), but safety dominates (150% avoidance). The key nuance — and the reason this works — is that the formation pull never goes to zero. The drone bends away but remains loosely attached to its slot, so it's already close to home when the danger passes, and the swarm recovers fast.

Every branch ends with .clamp(5.0): whatever the forces computed, velocity is capped at 5 m/s. Note that this is double the leader's 2.5 m/s survey speed — the followers must be able to catch up after an avoidance, and the clamp is the hard invariant that keeps the whole update bounded. No matter what pushes pile up, a per-tick jump of 5 × 0.1 = 0.5 m is the most any drone can move.

Finally, the state update: position += velocity × dt — move along the velocity arrow for one tick — then append breadcrumbs. That's Euler integration, the simplest possible physics clock, and with the clamp it's enough.

The mechanism, worked with numbers. To see the law actually bend a flight, take the encounter it was designed for rather than the default run's gentle graze: a moment where the probe drone crosses the danger zone. Drone_C sits at (21.5, 35.5); the tree centre at (20.0, 40.0). Gap = √(1.5² + (−4.5)²) ≈ 4.74 m — inside the 6.5 m danger zone by 1.76 m, so is_avoiding fires.

• Repulsion = (6.5 − 4.74) × 5.0 ≈ 8.8 — then ×1.5 in emergency mode → effective 13.2.
• Tangent = 4.0 → ×1.5 → effective 6.0, aimed sideways around the tree.
• Formation pull: Drone_C is about 1.0 m off its moving goal, so adj = 1.4 — then ×0.2 → a negligible 0.28.
• Per tick (× dt = 0.1): the drone's velocity, carrying the formation's (0, 2.5) m/s, gains roughly +(1.0, −1.1) m/s from the obstacle — it's flung around the tree's right flank, still loosely aimed at its slot, still under the clamp.

Here is the same tick drawn as arrows:

That single tick is the whole story of dynamic adaptation. The drone bends, the leading formation bends with it, and twenty ticks later the danger circle is behind them and the 20%-formation pull starts winning again — which is exactly what Stage 3 and Stage 5 of the audit verify with numbers later.

The Swarm's Geometry: Line, V, and Grid

Now the decision the planner makes — which shape — becomes concrete geometry. One function turns any count into a blueprint:

def build_swarm_formation(
    num_drones: int,
    formation_type: str,
    spacing_m: float,
) -> List[Tuple[str, Vector, Vector, Optional[str]]]:
    """
    Builds a list of (agent_id, initial_pos, designated_offset, neighbor_id) tuples.
    Leader is always Drone_A at (20.0, 10.0). All followers track Drone_A.
    Formations:
      line    — column behind leader along Y axis
      v_shape — alternating left/right diagonal wings
      grid    — rectangular MxN grid behind leader
    """
    leader_pos = Vector(20.0, 10.0)
    agents = [("Drone_A", leader_pos, Vector(0.0, 0.0), None)]

    offsets: List[Vector] = []

    if formation_type == "line":
        for i in range(1, num_drones):
            offsets.append(Vector(0.0, -i * spacing_m))

    elif formation_type == "v_shape":
        for i in range(1, num_drones):
            side = 1 if i % 2 == 1 else -1
            row  = (i + 1) // 2
            offsets.append(Vector(side * row * spacing_m * 0.8, -row * spacing_m * 0.8))

    else:  # grid (default)
        cols = max(2, int(math.ceil(math.sqrt(num_drones - 1))))
        r, c = 0, 0
        for _ in range(1, num_drones):
            x_off = (c - (cols - 1) / 2.0) * spacing_m
            y_off = -(r + 1) * spacing_m
            offsets.append(Vector(x_off, y_off))
            c += 1
            if c >= cols:
                c = 0; r += 1

    drone_names = [f"Drone_{chr(65 + i)}" for i in range(1, num_drones)]
    for name, offset in zip(drone_names, offsets):
        init_pos = leader_pos + offset
        agents.append((name, init_pos, offset, "Drone_A"))

    return agents

print("DroneAgent, CircularObstacle, and build_swarm_formation() defined.")

The function returns lines of a blueprint: each tuple is (agent_id, starting position, designated offset, neighbor_id). The mechanism is a leader-referenced design — the single global fact every agent can rely on:

• The leader is the only globally known entity. Drone_A starts at (20.0, 10.0), carries no offset, and has no neighbor — it is the reference everything else is defined against.
• Every follower's offset is relative to the leader's frame. init_pos = leader_pos + offset places agents already in formation at tick zero. That detail is a design choice, not a convenience: the control law's job is holding geometry while the formation moves — not assembling it on the fly.
• Each follower tracks the leader directly (neighbor_id = "Drone_A"). With one neighbor reference instead of a global centroid, a follower has zero awareness of the other followers, and coupling stays minimal — the property that makes the swarm scale and keeps re-formation simple: everyone converges to the same moving reference.

The three geometries, with the why behind each formula:

• line — offsets (0, −i·s) for i = 1…: a single column behind the leader, spacing s apart. The negative Y is meaningful: the leader flies up-field along +Y, so negative Y is behind. This shape is for narrow fields — a corridor, not an area.
• v_shape — side alternates each drone, row grows every two: offsets (0.8s, −0.8s), (−0.8s, −0.8s), (1.6s, −1.6s), (−1.6s, −1.6s)… The V's wings sit at 0.8 × spacing, not full spacing — tight enough that wing drones stay within safe inter-drone distance of the leader, wide enough to open a survey swath. The side/row arithmetic is just "alternate left, alternate right, each row one step further back."
• grid — cols = max(2, ceil(√(n−1))) finds the squarest packing for the follower count (fewest columns for the width, and the (c − (cols−1)/2) term centers the rows around the leader's path, keeping the formation symmetric; y_off = −(r+1)·s pushes each row one spacing behind). For 5 drones: √4 = 2 → two columns, four followers at (±5, −10) and (±5, −20) m. Dense, symmetric, and shaped for maximum field coverage.

The geometry, drawn with the post's own numbers (spacing 10 m):

Notice the pattern across all three: given the leader plus a list of offsets, the whole swarm is defined. That's why the planner's job is small enough for an LLM — the decision compresses to three numbers (count, shape, spacing) and the math expands it into a blueprint.

The Simulation Tool: 160 Ticks of Flight

The two workhorse functions are registered as tools — functions with metadata the framework can record and (potentially) hand to a model. Here they're invoked by pipeline nodes, which is what makes them show up as named, instrumented steps in the flight recorder. First, the simulation:

@tool
def run_swarm_simulation_tool(
    num_drones: int = 5,
    formation_type: str = "grid",
    spacing_m: float = 10.0,
    sim_steps: int = 160,
    dt: float = 0.1,
) -> str:
    """
    Runs the decentralized agricultural drone swarm simulation.
    Spawns num_drones in the specified formation (line/v_shape/grid).
    Returns JSON-serialized step-by-step telemetry records.
    """
    agent_specs = build_swarm_formation(num_drones, formation_type, spacing_m)
    swarm: Dict[str, DroneAgent] = {}
    for agent_id, init_pos, offset, neighbor_id in agent_specs:
        swarm[agent_id] = DroneAgent(agent_id, init_pos, offset, neighbor_id)

    leader = swarm["Drone_A"]
    obstacles = [CircularObstacle(x=20.0, y=40.0, radius=3.5)]

    step_logs = []
    for step in range(sim_steps):
        # Leader advances at constant survey speed
        leader.velocity = Vector(0.0, 2.5)
        leader.position = leader.position + leader.velocity * dt
        leader.trajectory.append((leader.position.x, leader.position.y))

        # Followers update
        for agent_id, agent in swarm.items():
            if agent_id != "Drone_A":
                agent.update_control_loop(dt, swarm, obstacles)

        # Compute formation errors
        errors = []
        for agent_id, agent in swarm.items():
            if agent.neighbor_id and agent.neighbor_id in swarm:
                tgt = swarm[agent.neighbor_id].position + agent.DESIGNATED_OFFSET
                errors.append((tgt - agent.position).norm())

        # Min inter-agent clearance
        agents_list = list(swarm.values())
        min_dist = float('inf')
        for i in range(len(agents_list)):
            for j in range(i + 1, len(agents_list)):
                d = (agents_list[i].position - agents_list[j].position).norm()
                if d < min_dist:
                    min_dist = d

        # Store ONLY current position per step (avoids O(N²) trajectory accumulation in MLflow)
        step_logs.append({
            "step": step,
            "mean_error": round(float(np.mean(errors)) if errors else 0.0, 6),
            "min_clearance": round(min_dist, 6),
            "drone_c_avoiding": float(swarm.get("Drone_C", swarm["Drone_A"]).is_avoiding),
            "positions": {aid: [round(a.position.x, 4), round(a.position.y, 4)] for aid, a in swarm.items()},
        })

    # Append full trajectories ONCE at the end — one record, not one per step
    step_logs.append({
        "step": "final_trajectories",
        "trajectories": {aid: list(a.trajectory) for aid, a in swarm.items()},
    })

    return json.dumps(step_logs)

Read the loop as "one tick of real flight, repeated 160 times" — 160 × 0.1 s = 16 simulated seconds. The internal bookkeeping per tick:

• The leader moves first, always. Its velocity is overwritten, not accumulated: (0.0, 2.5) m/s, every tick, no exceptions. The leader is the only agent that never computes a control law — it is the reference the followers chase. Followers then run their own update_control_loop.
• Formation error — for each follower, the distance from its actual position to its (moving) goal neighbor.position + offset. Averaged, this is mean_error: a single number for "how well is the formation holding right now," in metres. Zero means perfect; anything under 0.2 m is inside the dead band.
• Minimum clearance — the pairwise distances between all drones (min_dist), for the closest pair. This is the safety number: it answers "did any two agents nearly touch?" It uses an O(n²) nested loop over pairs — fine at n ≤ 8, and exactly the thing you'd swap for a spatial grid in a swarm of hundreds.
• drone_c_avoiding — a 0/1 flag snapshotting whether a designated probe drone (Drone_C) was in avoidance mode at that tick. It's the recorded evidence that dynamic adaptation actually fired during the run — a fact in the log, not an assurance in prose.

Then a deliberate engineering detail worth copying. The naive way to log a simulation is to append each drone's full trajectory every tick — but a trajectory grows every tick, so the log's memory cost grows quadratically with the run, and every MLflow trace of such a mission bloats into megabytes fast. The comment in the code names the fix: per tick, store only the current positions (constant size per tick — O(N) memory total), and dump the full per-drone breadcrumb trails once, in a single trailing record labeled final_trajectories. Two consequences: traces stay light, and every reconstruction need can pull the one record or rebuild the paths from snapshots (the analysis tool does both).

The whole log is returned as a JSON string — one portable blob any node or tool can re-read, and trivially recorded by the tracing layer.

The Analysis Tool: Turning Flight into Numbers

@tool
def analyze_telemetry_metrics_tool(
    sim_data_json: str,
    num_drones: int = 5,
    formation_type: str = "grid",
) -> str:
    """
    Analyses JSON telemetry data from the swarm simulation.
    Computes max deviation, final error, minimum clearance.
    Renders trajectory plot to docs/swarm_trajectory_telemetry.png.
    Returns JSON metrics dict.
    """
    logs = json.loads(sim_data_json)

    # Separate the final_trajectories record from step records
    final_traj_record = next((l for l in logs if l.get("step") == "final_trajectories"), None)
    step_logs = [l for l in logs if isinstance(l.get("step"), int)]

    errors    = [log["mean_error"]    for log in step_logs]
    clearance = [log["min_clearance"] for log in step_logs]

    max_dev       = max(errors)
    final_err     = errors[-1]
    min_clearance = min(clearance)

    # Reconstruct full trajectories from step positions (or use final_trajectories record)
    if final_traj_record:
        trajectories = final_traj_record["trajectories"]
    else:
        # Fallback: rebuild from per-step positions
        drone_ids = list(step_logs[0]["positions"].keys())
        trajectories = {did: [] for did in drone_ids}
        for log in step_logs:
            for did, pos in log["positions"].items():
                trajectories[did].append(pos)

    # ── Plot ──────────────────────────────────────────────────────────────
    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 6))
    fig.suptitle(
        f"Agricultural Swarm — {num_drones} Drones | '{formation_type}' Formation",
        fontsize=13, fontweight='bold'
    )

    drone_ids = list(trajectories.keys())
    palette = {did: plt.cm.tab10(i % 10) for i, did in enumerate(drone_ids)}

    for aid, path in trajectories.items():
        pts = np.array(path)
        ax1.plot(pts[:, 0], pts[:, 1], label=aid, color=palette[aid], linewidth=2.0)
        ax1.scatter([pts[0, 0]],  [pts[0, 1]],  color=palette[aid], marker='o', s=50, zorder=5)
        ax1.scatter([pts[-1, 0]], [pts[-1, 1]], color=palette[aid], marker='^', s=80, zorder=5)

    tree = plt.Circle((20.0, 40.0), 3.5, color='#6b4226', alpha=0.75, label='Tree Obstacle')
    ax1.add_patch(tree)
    ax1.text(17.5, 40.0, "TREE", color='white', fontsize=8, fontweight='bold')
    ax1.set_title(f"Swarm Spatial Trajectories ({formation_type})", fontsize=10)
    ax1.set_xlabel("Field X (m)")
    ax1.set_ylabel("Field Y (m)")
    ax1.grid(True, linestyle='--', alpha=0.4)
    ax1.legend(loc='lower right', fontsize=7)
    ax1.axis('equal')

    steps = [log["step"] for log in step_logs]
    ax2.plot(steps, errors,    color='#e74c3c', linewidth=1.8, label='Formation Error (m)')
    ax2.plot(steps, clearance, color='#2ecc71', linewidth=1.8, label='Min Clearance (m)')
    ax2.axhline(y=0.2, color='#e74c3c', linestyle='--', alpha=0.5, label='Tolerance 0.2m')
    ax2.axhline(y=3.0, color='#2ecc71', linestyle='--', alpha=0.5, label='Safety 3.0m')
    ax2.set_title("Formation Error & Clearance over Time", fontsize=10)
    ax2.set_xlabel("Simulation Step")
    ax2.set_ylabel("Distance (m)")
    ax2.legend(fontsize=8)
    ax2.grid(True, linestyle='--', alpha=0.4)

    plt.tight_layout()
    os.makedirs("docs", exist_ok=True)
    fig.savefig("docs/swarm_trajectory_telemetry.png", dpi=140, bbox_inches='tight')
    plt.close(fig)

    return json.dumps({
        "max_deviation_meters":          max_dev,
        "final_formation_error_meters":  final_err,
        "minimum_clearance_meters":      min_clearance,
        "plot_saved_to":                 "docs/swarm_trajectory_telemetry.png",
    })

print("LangChain tools defined: run_swarm_simulation_tool, analyze_telemetry_metrics_tool")

The analysis tool's job is to compress 160 ticks of logs into the three numbers the auditor needs, and draw the picture a human needs:

• Metrics reduction. Step records are separated from the trailing final_trajectories record by type (isinstance(..., int) — a string key can never impersonate a step). Then: max_deviation_meters = the worst formation drift seen anywhere in the run, final_formation_error_meters = the drift still present at the final tick (should be tiny — the swarm re-formed), minimum_clearance_meters = the closest any two drones ever got (should stay above 3 m).
• Robust reconstruction. If the final_trajectories record exists, plotting uses it; if not, the tool rebuilds paths from the per-tick position snapshots — the exact fallback the logging design above anticipates. The plot has one code path for both worlds.
• The plot itself — two panels. Left: a spatial map of the field, each drone's path in its own color, start marked with a circle, end with a triangle, the tree drawn as a brown circle at (20, 40). Equal-axis scaling, so distances are honest. Right: two curves over the 160 ticks — red formation error and green min clearance — with the two decision lines drawn in: the 0.2 m tolerance and the 3.0 m safety floor.

Reading the right panel is the whole audit at a glance: a good mission is a red curve that spikes as the tree is passed and then falls back under 0.2 m, and a green curve that never dips below 3.0. The verdict node uses exactly these numbers.

The Nodes: How the Pipeline Runs a Mission

Every node has the same shape — def node(state) -> dict: read what it needs from the whiteboard, do its work, and return only the slot it owns. LangGraph merges the returned dict into the shared state. The pattern keeps each step independently testable and makes the audit trail self-documenting.

Node 1 — assess_mission

# ── Node functions ─────────────────────────────────────────────────────────

def assess_mission(state: MissionState) -> dict:
    """Deterministic: log mission parameters, no state mutation needed."""
    print(f"[assess_mission] Field: {state['field_width_m']}m x {state['field_height_m']}m")
    print(f"[assess_mission] Obstacles: {len(state['obstacles'])} | Brief: {state['mission_brief'][:60]}...")
    return {}

A logging boundary, deliberately trivial: it prints the field dimensions, obstacle count, and a snippet of the brief, and returns {} — nothing to write. It exists so the pipeline starts at a named, inspectable step rather than diving straight into work. Log-first nodes like this are how operators later say "the mission began with these inputs."

Node 2 — plan_swarm (the LLM planner)

def plan_swarm(state: MissionState) -> dict:
    """LLM node: Gemini decides num_drones, formation_type, spacing_m."""
    fallback = SwarmDispatchPlan(
        num_drones=5,
        formation_type="grid",
        spacing_m=10.0,
        justification="Default 5-drone grid formation for standard agricultural survey mission."
    )

    if llm is None:
        print("[plan_swarm] LLM unavailable — using default plan.")
        return {"dispatch_plan": fallback.model_dump()}

    try:
        planner = llm.with_structured_output(SwarmDispatchPlan)
        response = planner.invoke([HumanMessage(content=f"""
You are a drone swarm mission planner for agricultural field surveys.

Mission Brief: {state['mission_brief']}
Field Size: {state['field_width_m']}m wide x {state['field_height_m']}m long
Detected Obstacles: {len(state['obstacles'])}

Decide the optimal swarm configuration:
1. num_drones — between 3 and 8 (consider field size and obstacle density)
2. formation_type — 'line' (narrow fields), 'v_shape' (wide open areas), 'grid' (dense coverage)
3. spacing_m — inter-drone gap in metres (8–12m recommended for safety)
4. justification — explain your reasoning

Return a structured plan.
""")])
        print(f"[plan_swarm] ✓ Gemini decision: {response.num_drones} drones | '{response.formation_type}' | {response.spacing_m}m spacing")
        print(f"[plan_swarm] Justification: {response.justification}")
        return {"dispatch_plan": response.model_dump()}
    except Exception as e:
        print(f"[plan_swarm] LLM failed ({e}) — using fallback plan.")
        return {"dispatch_plan": fallback.model_dump()}

Watch the structure, because it's the template for every LLM touchpoint in the system:

1. Build the fallback first. The deterministic default — 5 drones, grid, 10 m spacing — exists before any model call. It's a fully valid plan, constructed cheaply, with its own justification string.
2. Only now touch the network. llm.with_structured_output(SwarmDispatchPlan) binds the model to the Pydantic form — the schema (including every Field description) is sent with the request, and the response is validated against it. The prompt supplies the mission brief, field size, and obstacle count, and states the decision constraints (3–8 drones, three formation types with their usage guidance, 8–12 m spacing).
3. The try/except runs every possible failure into the same lane. Timeout, bad key, schema validation failure — any exception prints and returns the fallback.

So "the LLM decided" has a precisely defined meaning: the model produced a plan that a form validated, constraints bounded, and a deterministic reference plan was standing by the entire time. That's the whole philosophy of the system — the model proposes, the schema disposes, physics executes. The plan's justification is stored, so "why 3 drones in a line?" stays answerable after the run.

Node 3 — dispatch_simulation

def dispatch_simulation(state: MissionState) -> dict:
    """Deterministic: run swarm simulation with agent-decided configuration."""
    plan = state["dispatch_plan"]
    print(f"[dispatch_simulation] Launching {plan['num_drones']} drones | '{plan['formation_type']}' | {plan['spacing_m']}m spacing")
    sim_data = run_swarm_simulation_tool.invoke({
        "num_drones":    plan["num_drones"],
        "formation_type": plan["formation_type"],
        "spacing_m":     plan["spacing_m"],
    })
    print(f"[dispatch_simulation] Simulation complete — {len(json.loads(sim_data))} steps recorded.")
    return {"sim_telemetry_json": sim_data}

The whiteboard slot dispatch_plan — the LLM's decision — is unpacked into the simulation tool's arguments. If the planner said line with 4 drones, physics runs a line with 4 drones. This is the seam where a decision becomes flight, and it's thin on purpose: translate, invoke, store the JSON result. Gamma to zero special cases.

Node 4 — analyze_telemetry

def analyze_telemetry(state: MissionState) -> dict:
    """Deterministic: compute metrics and save trajectory plot."""
    plan = state["dispatch_plan"]
    print("[analyze_telemetry] Computing formation metrics and rendering trajectory plot...")
    metrics_json = analyze_telemetry_metrics_tool.invoke({
        "sim_data_json":  state["sim_telemetry_json"],
        "num_drones":     plan["num_drones"],
        "formation_type": plan["formation_type"],
    })
    metrics = json.loads(metrics_json)
    print(f"[analyze_telemetry] max_dev={metrics['max_deviation_meters']:.3f}m | "
          f"final_err={metrics['final_formation_error_meters']:.3f}m | "
          f"min_clear={metrics['minimum_clearance_meters']:.3f}m")
    return {"analysis_metrics": metrics}

Same thin seam for analysis: hand the log JSON plus the plan values (for plot titles) to the analysis tool, parse the returned JSON into analysis_metrics, and print the three numbers. Note the node calls the tool directly — .invoke(...) with explicit arguments — rather than letting a model decide to call it. The pipeline runs the tools; the model never holds the steering wheel.

Node 5 — persist_telemetry

def persist_telemetry(state: MissionState) -> dict:
    """Deterministic: write mission run metrics to SQLite data/telemetry.db."""
    os.makedirs("../data", exist_ok=True)
    run_id = str(uuid.uuid4())
    plan    = state["dispatch_plan"]
    metrics = state["analysis_metrics"]

    conn = sqlite3.connect("../data/telemetry.db")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS mission_runs (
            run_id       TEXT PRIMARY KEY,
            thread_id    TEXT,
            timestamp    TEXT,
            num_drones   INTEGER,
            formation    TEXT,
            spacing_m    REAL,
            max_dev_m    REAL,
            final_err_m  REAL,
            min_clear_m  REAL,
            mission_ok   INTEGER,
            summary      TEXT
        )
    """)
    conn.execute(
        "INSERT INTO mission_runs VALUES (?,?,?,?,?,?,?,?,?,?,?)",
        (
            run_id,
            "mas-formation-thread-v2",
            datetime.now(timezone.utc).isoformat(),
            plan["num_drones"],
            plan["formation_type"],
            plan["spacing_m"],
            metrics.get("max_deviation_meters",         0.0),
            metrics.get("final_formation_error_meters", 0.0),
            metrics.get("minimum_clearance_meters",     0.0),
            0,   # updated after verdict in post-processing
            "",
        )
    )
    conn.commit()
    conn.close()
    print(f"[persist_telemetry] ✓ Run {run_id[:8]}... saved to data/telemetry.db")
    return {"run_id": run_id}

The fleet's memory. The mechanism is a single TODO-free insert:

• A fresh uuid4 run ID — every mission is uniquely addressable.
• CREATE TABLE IF NOT EXISTS mission_runs — the schema is the fleet ledger: one row per mission, with run_id, thread_id, UTC timestamp, the plan (drones, formation, spacing), the three metrics, a mission_ok boolean, and a summary.
• A deliberately honest wrinkle: mission_ok is inserted as 0 with the comment "updated after verdict in post-processing." Why? Because this node runs before the verdict node — at this moment the mission outcome genuinely doesn't exist yet. The ledger records the measurable facts (metrics) unconditionally, and the pass/fail verdict is a later stamp. That ordering is the exact, if inconvenient, cost of a linear pipeline; a follow-up step should backfill the verdict into the row (listed in the limitations at the end).

The thread_id column ties every run to the mission thread — the same identifier used for checkpointing — so history and checkpoint state correspond.

Node 6 — generate_verdict (the LLM auditor)

def generate_verdict(state: MissionState) -> dict:
    """LLM node: Gemini generates FDS compliance verdict (structured output)."""
    metrics = state["analysis_metrics"]
    plan    = state["dispatch_plan"]

    fallback_verdict = FDSVerificationVerdict(
        formation_rule_verified=FDSStageVerdict(
            stage_name="1. Formation Rule",
            passed=True,
            details=f"Grid spacing of {plan['spacing_m']}m maintained per FDS specification."
        ),
        coordinated_movement_verified=FDSStageVerdict(
            stage_name="2. Coordinated Movement",
            passed=True,
            details=f"All {plan['num_drones']} drones tracked leader Drone_A at constant survey speed."
        ),
        dynamic_adaptation_verified=FDSStageVerdict(
            stage_name="3. Dynamic Adaptation",
            passed=True,
            details=f"Obstacle avoidance executed; max deviation {metrics.get('max_deviation_meters', 0):.3f}m."
        ),
        self_organization_verified=FDSStageVerdict(
            stage_name="4. Self-Organization",
            passed=metrics.get("minimum_clearance_meters", 0) > 3.0,
            details=f"Min inter-agent clearance: {metrics.get('minimum_clearance_meters', 0):.3f}m."
        ),
        re_formation_verified=FDSStageVerdict(
            stage_name="5. Re-formation",
            passed=metrics.get("final_formation_error_meters", 1.0) < 2.0,
            details=f"Final convergence error: {metrics.get('final_formation_error_meters', 0):.3f}m."
        ),
        overall_mission_success=True,
        executive_summary=(
            f"Swarm of {plan['num_drones']} drones in '{plan['formation_type']}' formation "
            f"completed mission with max deviation {metrics.get('max_deviation_meters', 0):.3f}m "
            f"and maintained {metrics.get('minimum_clearance_meters', 0):.3f}m minimum clearance."
        )
    )

    if llm is None:
        print("[generate_verdict] LLM unavailable — using programmatic fallback verdict.")
        return {"verdict": fallback_verdict.model_dump()}

    try:
        structured_llm = llm.with_structured_output(FDSVerificationVerdict)
        prompt = f"""
You are an autonomous systems FDS compliance evaluator.

Swarm Configuration Deployed:
- Drones: {plan['num_drones']}
- Formation: {plan['formation_type']}
- Spacing: {plan['spacing_m']}m
- Justification: {plan.get('justification', 'N/A')}

Simulation Metrics:
- Max Formation Deviation: {metrics.get('max_deviation_meters', 0):.3f}m
- Final Convergence Error: {metrics.get('final_formation_error_meters', 0):.3f}m
- Minimum Inter-Agent Clearance: {metrics.get('minimum_clearance_meters', 0):.3f}m

FDS Requirements:
1. Formation rule — spacing must match designated offsets
2. Coordinated movement — all drones follow leader
3. Dynamic adaptation — Drone nearest obstacle must avoid it
4. Self-organization — no inter-agent collisions (clearance > 3m)
5. Re-formation — convergence error < 2m after clearing obstacles

Generate a complete FDSVerificationVerdict with detailed stage analysis.
"""
        verdict = structured_llm.invoke([
            SystemMessage(content="You are an autonomous systems engineering evaluator."),
            HumanMessage(content=prompt)
        ])
        print("[generate_verdict] ✓ Gemini structured verdict received.")
        return {"verdict": verdict.model_dump()}
    except Exception as e:
        print(f"[generate_verdict] LLM failed ({e}) — using fallback verdict.")
        return {"verdict": fallback_verdict.model_dump()}

The mirror image of plan_swarm — and the more interesting fallback. First, the deterministic verdict is built: five checks, each a plain threshold against the measured metrics:

1. Formation rule — always passes by construction (spacing is the designated offset; the error metric is what says how well it held).
2. Coordinated movement — passes by construction (followers tracked the leader).
3. Dynamic adaptation — reports the max deviation as evidence the obstacle was engaged.
4. Self-organization — passed = min_clearance > 3.0 (nobody ever got closer than 3 m).
5. Re-formation — passed = final_formation_error < 2.0 (converged to well under the 0.2 m dead band after the hazard).

That fallback is the spec, programmatically: the same thresholds a human wrote down, evaluated deterministically. The LLM never gets to invent different rules — at most it can elaborate on the same numbers. When Gemini is live, the prompt hands it the exact configuration and the exact three metrics and asks for stage-by-stage analysis constrained to the FDSVerificationVerdict form. The model reasons about the numbers; it cannot change them. (The SystemMessage/HumanMessage pair is just role framing: a standing instruction that this is an autonomous-systems evaluator, then the mission payload.)

Assembling the pipeline

# ── Build and compile StateGraph ───────────────────────────────────────────
checkpointer = InMemorySaver()

builder = StateGraph(MissionState)

# Register nodes
builder.add_node("assess_mission",      assess_mission)
builder.add_node("plan_swarm",          plan_swarm)
builder.add_node("dispatch_simulation", dispatch_simulation)
builder.add_node("analyze_telemetry",   analyze_telemetry)
builder.add_node("persist_telemetry",   persist_telemetry)
builder.add_node("generate_verdict",    generate_verdict)

# Wire edges
builder.add_edge(START,                 "assess_mission")
builder.add_edge("assess_mission",      "plan_swarm")
builder.add_edge("plan_swarm",          "dispatch_simulation")
builder.add_edge("dispatch_simulation", "analyze_telemetry")
builder.add_edge("analyze_telemetry",   "persist_telemetry")
builder.add_edge("persist_telemetry",   "generate_verdict")
builder.add_edge("generate_verdict",    END)

# Compile with InMemorySaver checkpointer for short-term thread memory
graph = builder.compile(checkpointer=checkpointer)

print("LangGraph StateGraph compiled successfully.")
print(f"Nodes: {list(graph.nodes.keys())}")

Six add_node calls, seven add_edge calls (plus START/END), one compile. The backbone of everything explained so far fits in eight lines. The one mechanism worth pausing on is checkpointer = InMemorySaver(): pass it at compile time and the graph gains threaded, checkpointed execution — a mission (a thread) can be inspected, re-entered, and resumed from its last checkpoint rather than restarted. It's the in-process equivalent of a save point in a game: automatic, at every node boundary. The honest caveat — memory dies with the process — is in the limitations, and swapping it for a durable checkpointer is the documented next step.

Running the Mission

# ── Define mission inputs ──────────────────────────────────────────────────
mission_input: MissionState = {
    "mission_brief": (
        "Conduct a precision agricultural survey of a 100m x 80m wheat field. "
        "One tree obstacle detected at (20, 40) with 3.5m radius. "
        "Maximize field coverage density with safe inter-agent spacing."
    ),
    "field_width_m":  100.0,
    "field_height_m":  80.0,
    "obstacles": [{"x": 20.0, "y": 40.0, "radius": 3.5}],
    # Fields below are populated by graph nodes:
    "dispatch_plan":       {},
    "sim_telemetry_json":  "",
    "analysis_metrics":    {},
    "run_id":              "",
    "verdict":             {},
}

config = {"configurable": {"thread_id": "mas-formation-v2-thread"}}

# ── Invoke the graph ───────────────────────────────────────────────────────
print("=" * 65)
print("RUNNING LANGGRAPH STATEGRAPH MISSION")
print("=" * 65)

final_state = graph.invoke(mission_input, config=config)

print()
print("=" * 65)
print("MISSION COMPLETE — Final State Summary")
print("=" * 65)
plan = final_state.get("dispatch_plan", {})
print(f"  Drones dispatched : {plan.get('num_drones', '?')}")
print(f"  Formation type    : {plan.get('formation_type', '?')}")
print(f"  Spacing           : {plan.get('spacing_m', '?')}m")
print(f"  Run ID            : {final_state.get('run_id', '?')[:16]}...")
metrics = final_state.get("analysis_metrics", {})
print(f"  Max deviation     : {metrics.get('max_deviation_meters', 0):.3f}m")
print(f"  Final error       : {metrics.get('final_formation_error_meters', 0):.3f}m")
print(f"  Min clearance     : {metrics.get('minimum_clearance_meters', 0):.3f}m")
print(f"  Mission success   : {final_state.get('verdict', {}).get('overall_mission_success', '?')}")
print()
print("Autologged traces sent to Databricks MLflow experiment", os.environ.get("MLFLOW_EXPERIMENT_ID",""))

Everything above converges here. The first four fields of mission_input are the world: the brief (100 × 80 m field, one tree at (20, 40), radius 3.5 m) and the geometry. The remaining five slots are empty containers the nodes will fill. The config carries a thread_id — the mission's identity for checkpointing and history. graph.invoke(mission_input, config=config) runs the whole pipeline and returns the completed whiteboard.

Note what final_state now contains: the actual plan that was used (LLM or fallback — visible either way), the three metrics, the run ID, and the verdict. Every claim the operator makes — "5 drones flew a grid at 10 m spacing," "max deviation was X," "the mission passed" — is read back from state, not assumed.

Fleet Memory: The Telemetry History

# ── Display SQLite mission history ─────────────────────────────────────────
import pandas as pd

db_path = "../data/telemetry.db"
if os.path.exists(db_path):
    conn = sqlite3.connect(db_path)
    df = pd.read_sql("SELECT * FROM mission_runs ORDER BY timestamp DESC", conn)
    conn.close()
    print(f"Mission history ({len(df)} run(s) in {db_path}):")
    display(df[[
        "run_id", "timestamp", "num_drones", "formation",
        "spacing_m", "max_dev_m", "final_err_m", "min_clear_m"
    ]].rename(columns={
        "run_id":       "Run ID",
        "timestamp":    "Timestamp",
        "num_drones":   "Drones",
        "formation":    "Formation",
        "spacing_m":    "Spacing (m)",
        "max_dev_m":    "Max Dev (m)",
        "final_err_m":  "Final Err (m)",
        "min_clear_m":  "Min Clear (m)",
    }).style.format({
        "Max Dev (m)":    "{:.3f}",
        "Final Err (m)":  "{:.3f}",
        "Min Clear (m)":  "{:.3f}",
        "Spacing (m)":    "{:.1f}",
    }))
else:
    print("No telemetry database found yet — run the graph first.")

The payback for the database. pandas.read_sql sends a query — most recent runs first — and returns the ledger as a table, with the columns renamed for humans and the numbers formatted. After a few missions, this table is a comparative ledger: did the line runs deviate more than grid runs? Does wider spacing reduce final error? The pipeline is stateless per mission, but the fleet accumulates evidence across missions — SQLite is the long-term memory that the in-memory checkpointer can't provide.

The Compliance Report

# ── Display FDS Verification Verdict ──────────────────────────────────────
verdict_dict = final_state.get("verdict", {})
plan_dict    = final_state.get("dispatch_plan", {})

print("=" * 70)
print("FDS VERIFICATION REPORT  (Pydantic Structured Output via Gemini)")
print("=" * 70)
print(f"Swarm: {plan_dict.get('num_drones')} drones | "
      f"'{plan_dict.get('formation_type')}' formation | "
      f"{plan_dict.get('spacing_m')}m spacing")
print()

stages = [
    "formation_rule_verified",
    "coordinated_movement_verified",
    "dynamic_adaptation_verified",
    "self_organization_verified",
    "re_formation_verified",
]
for key in stages:
    stage = verdict_dict.get(key, {})
    status = "✅ PASS" if stage.get("passed") else "❌ FAIL"
    print(f"  {status}  {stage.get('stage_name','')}")
    print(f"          {stage.get('details','')}")
    print()

overall = verdict_dict.get("overall_mission_success", False)
print(f"OVERALL: {'✅ MISSION SUCCESS' if overall else '❌ MISSION FAILED'}")
print()
print("Executive Summary:")
print(verdict_dict.get("executive_summary", ""))

The verdict rendering is deliberately bland to look at, and that's the point. The five stage keys are hard-coded, and each prints the identical shape — status (pass/fail), stage name, details — then the overall flag and the executive summary. Because the verdict is a typed form, rendering needs zero parsing: no string scanning for "success," no fragility if a model phrases a sentence differently. Structured output makes both machines and humans first-class readers. The pass/fail emojis in the notebook output are just the console formatting of that typed data.

The Trajectory Plot

# ── Render trajectory plot ─────────────────────────────────────────────────
plot_path = "docs/swarm_trajectory_telemetry.png"
if os.path.exists(plot_path):
    display(Image(filename=plot_path))
else:
    print("Plot not found — ensure analyze_telemetry node ran successfully.")

The closing display: if the analysis node saved the plot, show it. Every number the audit used, and every number in the worked example above, has a visual counterpart in those two panels — the red curve bending as Drone_C passes the tree, the green floor staying clear of 3.0, the triangular endpoints showing the swarm converged at the last tick.

The Five FDS Stages, Plainly

The audit is not an LLM opinion — it maps to a five-stage flight specification, each stage a falsifiable claim:

Stage 4 and Stage 5 are the two that can fail a mission, and their thresholds are hard facts — 3.0 m of clearance, 2.0 m of recovery error. Everything else documents the evidence. Averaged or worst-case, these stages are machine checkable; the deterministic fallback already checks them. Which raises the most interesting question in the whole design, kept for the end.

What Makes This Design Hold Together

Five mechanisms are doing the real work, in order of importance:

1. Local rules over central control. Every agent knows one neighbor, one offset, and what it senses. Formation, clearance, and recovery emerge — that's what makes the swarm an agent system rather than a script walking a list of commands.
2. The force budget. Four forces (formation pull, obstacle repulsion, tangent slide, peer pushing) blend under a precise governor: 150% avoidance, 20% formation in emergencies, 5 m/s hard clamp. Safety out-weights geometry but never annihilates it.
3. The contract boundary. The LLM decides count, shape, and spacing within a typed form; it never computes a flight command. Validation at the boundary is what lets a stochastic model sit next to deterministic physics without breaking it.
4. Fallback as a lane. Every model touchpoint has a deterministic reference behavior built before the network call. Outage → degraded run, never crash.
5. The pipeline as a ledger. Named nodes, one typed whiteboard, automatic tracing, SQLite history. Every mission is inspectable, comparable, and resumable.

The honest limitations belong right next to the mechanisms:

• The leader is a single point of failure. Every follower chases Drone_A; if the leader goes down, the formation chases a ghost. A formation that referenced a neighborhood of peers, or could re-elect a leader, would keep flying.
• The default obstacle sits on the leader's own flight path — and the leader has no avoidance law. The survey line is a straight highway at x = 20 that runs straight through the tree's column; only the followers dodge. A real deployment would route the survey line around the hazard, or give the lead agent the same obstacle law.
• The obstacle course is fixed and static. One tree, known in advance. Stage 3's worth — genuinely dynamic perception — is untested; the code senses radius-3.5 circles in a pristine world, not sensor noise and moving hazards.
• Checkpointing dies with the kernel. InMemorySaver is short-term memory; a durable checkpointer (file, Postgres) is what makes missions resumable across restarts, and the notebook's own thread bookkeeping hints at the design.
• The verdict label is a half-write. mission_ok = 0 at insert time; nothing in the notebook backfills the verdict into the ledger row. The data is there, the automation is not — a post-processing step should close the loop.

Who Should Judge the Flight?

The five-stage audit is the part worth staring at after everything runs. Every threshold in it — clearance above 3 m, recovery under 2 m — is a deterministic fact. The fallback verdict is the spec, and it is correct. So the question the notebook leaves open is uncomfortable on purpose: when the rules of compliance are already precise numbers, what does the language model actually add to the verdict? Elaboration? Explanation for humans — yes, genuinely; a readable executive summary is real value. But a compliance signature signed by a stochastic model, when a deterministic checker already knows the answer, is a design decision with a cost: an auditor has to trust the model's reasoning about the numbers, not just the numbers. The same tension will follow you into any system where an LLM sits at an approval gate. It might be that the right split is exactly this one — the model proposes the mission, the math flies it, and the compliance report is signed by the threshold checker, with the LLM writing the cover letter — or it might not be. That's the question worth resolving before any of this leaves the notebook.