# Project export: GroundTruth
This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small.
## Project metadata
- Hackathon: UC Berkeley AI Hackathon 2026
- Tagline: When a drone fails mid-mission, your fleet shouldn't wait for a human to notice. New rules let companies fly whole drone fleets unsupervised. Groundtruth's agents reassign tasks in real time.
- Devpost: https://devpost.com/software/groundtruth-z9avo5
- GitHub: https://github.com/IceyGirl424/Groundtruth
- Video: https://www.youtube.com/embed/HX2dzDOaV-A?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent
- Team: 2 GitHub contributor(s) — Lena Munad (2 commits), Claude Opus 4.8 (1 commits)
## Devpost submission (written by the team)
### Inspiration
I'm a certified drone pilot, and I've spent real hours flying missions where the actual hard part wasn't flying the drone, it was everything around it: knowing what to do the second something goes wrong mid-flight. A battery dips faster than expected, weather shifts, and suddenly you're improvising. Now imagine that same moment, but multiplied across a whole fleet of drones at once, which is exactly where the industry is headed. New FAA rules just made it legal to fly entire fleets without a pilot watching every single drone. But nobody's built the thing that handles "something just went wrong" when there's no human watching closely enough to catch it. So we built it.
### What it does
Groundtruth is an autonomous coordination layer for drone fleets. Each drone runs as its own independent AI agent, holding its own battery, position, and task queue. When something goes wrong, a drone's battery drops critical, a weather cell rolls in, you just type it into a chat: "Drone-2's battery has dropped critical, reassign its remaining tasks." The Coordinator agent queries every drone for its live status, Claude reasons over the whole fleet under hard safety rules (never hand a task to a drone that can't safely take it), and replies with a clear reassignment plan and a quality score, all live, all inside a single chat conversation. It also remembers: every resolved incident gets stored, so next time something similar happens, it recalls what worked before instead of starting from zero.
### How we built it
Each drone is a uAgents agent registered on Fetch.ai's Agentverse, reachable through ASI:One via the Agent Chat Protocol. A Coordinator agent fans out structured status requests to the whole fleet, collects every drone's response, and hands that live state to Claude (Sonnet 4.6), which returns a structured reassignment plan with a written rationale. RedisVL backs a vector memory layer, every resolved incident gets embedded and stored, and new disruptions pull up semantically similar past incidents as context, even when worded completely differently. Arize traces every single Claude call and runs an LLM-as-judge evaluator that scores each plan against our safety rules, live, in their dashboard. A lightweight dashboard renders the whole fleet and the latest plan visually, polling the same data the agents produce.
### Challenges we ran into
The drones weren't all answering at once. We told the coordinator to ask all 3 drones their status, but a setting in the framework made it only listen to one reply at a time instead of all three. Fixed by flipping that setting. The drones weren't all answering at once. We told the coordinator to ask all 3 drones their status, but a setting in the framework made it only listen to one reply at a time instead of all three. Fixed by flipping that setting. We were checking for replies too fast. We'd ask the drones, wait 2 seconds, then give up and move on, but sometimes a drone's answer hadn't arrived yet. Gave it a little more time to wait, fixed it. We were checking for replies too fast. We'd ask the drones, wait 2 seconds, then give up and move on, but sometimes a drone's answer hadn't arrived yet. Gave it a little more time to wait, fixed it. We broke our own test by accident. We wiped the memory database to test something, but the program was still running and didn't know the memory got wiped, so it kept acting like there was no memory at all, when really we'd just confused our own system. Restarted everything cleanly and it worked. We broke our own test by accident. We wiped the memory database to test something, but the program was still running and didn't know the memory got wiped, so it kept acting like there was no memory at all, when really we'd just confused our own system. Restarted everything cleanly and it worked.
### Accomplishments we're proud of
Watching the system handle a genuinely hard edge case correctly: when every drone in our fleet was simultaneously low on battery, it refused to force an unsafe reassignment and told us to dispatch backup drones instead, real judgment, not forced productivity. Everything is real and verifiable, not staged: you can watch the actual Claude reasoning trace and safety score live in Arize, and watch it recall a past incident live in the chat. Getting true concurrent multi-agent negotiation working, not just a single LLM call dressed up as "agents." A full Redis vector memory loop confirmed working end to end: store an incident, recall it later via semantic similarity, even when worded completely differently.
### What we learned
That getting multiple AI agents to genuinely negotiate with each other is a different problem than just calling an LLM in a loop, concurrency, timing, and message protocols matter as much as the reasoning itself. We also learned that giving an AI system memory of its own past decisions doesn't just make it smarter, it makes its decisions more consistent and easier to trust, which matters a lot when the decisions are safety-related.
### What's next
Voice input via Deepgram, so a fleet operator can report a disruption hands-free instead of typing, genuinely useful for someone out in the field. A HazardAgent that detects disruptions proactively from battery and weather trends, instead of waiting for a human to type one in. Scaling the fleet size and adding more disruption types (equipment failure, no-fly zone changes) to stress-test the negotiation logic further. The actual business: Groundtruth as per-drone-per-month fleet-coordination software sold directly to commercial drone operators, the same pricing model fleet management software already uses in trucking and logistics, built for a market that's about to need this badly as multi-drone operations scale.
## README (from the GitHub repository)
# Groundtruth — Swarm Tasking
**Advisory coordination infrastructure for commercial drone inspection fleets.**
When something goes wrong mid-mission — a drone's battery hits critical, a weather
cell rolls over part of a solar farm — Groundtruth renegotiates task assignments
across the fleet in real time and hands a human fleet manager a clear,
plain-English reassignment plan. It assists the operator; it does **not** fly the
aircraft. Authority over flight decisions stays with the human.
The entire workflow is demonstrable inside an **ASI:One** chat conversation: you
describe the disruption, and the coordinator queries the live fleet, reasons over
its state with Claude, recalls how similar past incidents were resolved, and
replies with a scored plan.
---
## Why this matters
Commercial drone operators (solar/utility inspection, agricultural survey,
infrastructure, post-disaster mapping) are scaling from single-drone to
multi-drone operations as BVLOS waivers expand — but coordination today is
usually a person watching several screens. Groundtruth is fleet-coordination
software, priced per-drone-per-month like fleet management in trucking/logistics.
---
## Architecture
```
ASI:One chat (the human fleet manager)
│ "Drone-2 battery critical, reassign its tasks"
▼
┌─────────────────────────────┐
│ SwarmCoordinator │ uAgents + Agent Chat Protocol
│ (mailbox agent) │ → discoverable on Agentverse / ASI:One
└─────────────────────────────┘
│ 1. fan-out StatusRequest ▲ 4. combined fleet status,
▼ to every drone │ plan, and quality score
┌──────────┐ ┌──────────┐ ┌──────────┐
│ Drone-1 │ │ Drone-2 │ │ Drone-3 │ each: battery %, position,
│ :8001 │ │ :8002 │ │ :8003 │ current task, queue, capacity
└──────────┘ └──────────┘ └──────────┘
│ 2. collect StatusResponses
▼
┌─────────────────────────────┐
│ Claude reasoning + memory │
│ │
│ • RedisVL vector memory ───▶ recall similar past incidents
│ • Claude (Sonnet 4.6) ─────▶ produce reassignment plan
│ • LLM-as-judge ────────────▶ score plan quality
│ • Arize ───────────────────▶ trace every LLM call
└─────────────────────────────┘
│ 3. store resolved incident back into memory
```
### Agents
- **`SwarmCoordinator`** (`coordinator_agent.py`) — entry point. Implements the
Agent Chat Protocol, runs as a mailbox agent (no public endpoint needed),
orchestrates everything below.
Address: `agent1qw7awftrnyz2haxmwc7frd0u2mweelukfcueeer6lg2xcqq0mvef608jgmm`
- **`DroneAgent`** (`drone_agent.py`) — one process per drone, holds its own state
(battery, position, task queue), answers status queries, accepts/rejects task
assignments under a battery-safety rule. Launch many with distinct `--id`/`--port`.
- **Shared message schemas** (`messages.py`) — typed uAgents `Model`s exchanged
between coordinator and drones (routed by schema digest, so both sides must
match exactly).
### Reasoning, memory, observability
- **`claude_reasoning.py`** — builds the prompt from live fleet state + the
disruption (+ recalled context), calls Claude for a structured plan, and
contains the LLM-as-judge evaluator. All Claude calls are traced to Arize.
- **`agent_memory.py`** — RedisVL vector index over incident embeddings
(`all-MiniLM-L6-v2`, local, 384-dim). Stores each resolved incident; retrieves
semantically similar past ones to inform new decisions.
---
## Sponsor technology
| Sponsor | How it's used | Prize track |
|---|---|---|
| **Fetch.ai** (uAgents, Agent Chat Protocol, Agentverse, ASI:One) | All agents are uAgents; coordinator is registered on Agentverse and usable directly from ASI:One; agent-to-agent messaging uses typed protocols | Fetch.ai / ASI:One |
| **Anthropic Claude** (`claude-sonnet-4-6`) | The reasoning engine that produces reassignment plans, and the LLM-as-judge that scores them | — |
| **Redis** (RedisVL + vector search) | Agent memory: incidents embedded and stored in a Redis vector index; semantic similarity search retrieves relevant past incidents as context for new decisions | Redis (Agent Memory / vector search / context retrieval) |
| **Arize AX** (OpenTelemetry + OpenInference) | Distributed tracing of every Claude call (reasoning + judge), with plan-quality evaluation scores attached as span attributes | Arize |
---
## Setup
Requires **Python 3.11+**.
```bash
# 1. Install dependencies
pip install -r requirements.txt
# 2. Create your secrets file from the template and fill it in
cp .env.example .env
# Edit .env with your real ANTHROPIC_API_KEY, REDIS_URL, ARIZE_SPACE_ID,
# ARIZE_API_KEY, and AGENTVERSE_API_TOKEN.
```
### One-time: claim the coordinator's mailbox
The coordinator uses a mailbox so it's reachable through ASI:One without a public
endpoint. The first time you run it, claim the mailbox:
1. Start the coordinator (`./run.sh`).
2. Open the inspector URL printed in `logs/coordinator.log`
(`https://agentverse.ai/inspect/?uri=...&address=...`) in **Chrome**.
3. Sign in to Agentverse and click **Connect / Create Mailbox**.
4. Restart (`./run.sh`) — the `Agent mailbox not found` warning disappears.
---
## Run
```bash
./run.sh
```
This loads secrets from `.env`, launches the coordinator (port 8000) and all
three drones (ports 8001–8003) in the background with logs under `logs/`, and
prints a summary of what's running.
Stop everything:
```bash
pkill -f coordinator_agent.py ; pkill -f drone_agent.py
```
---
## Try it
**In ASI:One (the primary workflow):** find the `SwarmCoordinator` agent and send
it a disruption, e.g. *"Drone-2's battery has dropped critical, reassign its
remaining tasks."* You'll get back the live fleet status, a reassignment plan with
rationale, and an LLM-as-judge quality score — and a 🧠 note if a similar past
incident was recalled from memory.
**Locally (no UI needed):** `chat_test_client.py` simulates an ASI:One user.
```bash
python3 chat_test_client.py --scenario 1 # battery critical (wording A)
python3 chat_test_client.py --scenario 2 # battery critical (wording B — semantically matches 1)
python3 chat_test_client.py --scenario 3 # weather exclusion zone (different scenario)
python3 chat_test_client.py --message "custom disruption text"
```
Run scenario 1 then 2 to see semantic memory recall: scenario 2 is worded
differently but recalls scenario 1 via vector similarity. Scenario 3 (weather) is
correctly judged *not* similar and reasoned from scratch.
---
## Repository layout
```
coordinator_agent.py SwarmCoordinator: chat handler, fan-out, memory, reasoning, eval
drone_agent.py DroneAgent: per-drone state + status/assignment protocol
messages.py Shared typed message schemas (coordinator <-> drones)
claude_reasoning.py Claude reasoning, LLM-as-judge, Arize tracing
agent_memory.py RedisVL vector memory (store / retrieve similar incidents)
chat_test_client.py Local test client (simulates an ASI:One user)
run.sh Launch the full stack from .env
requirements.txt Python dependencies
.env.example Template for secrets (copy to .env)
```
---
## Roadmap
- Real disruption detection (a `HazardAgent` monitoring battery/weather thresholds
and proactively notifying the coordinator) instead of disruptions described in chat.
- Live map visualization driven off the Redis incident/message log (bonus polish).
## Detected evidence (automated analysis)
Indexed codebase: 10 recognized source files, 65 KB.
- Anthropic (technology) — detected in the code
- HTML (language) — detected in the code
- Python (language) — detected in the code
- Redis (technology) — detected in the code
- AI coding agent: Claude Code — evidence: commit authorship or trailers
## Codebase structure (from repository index)
### Files (13 of 13)
```
.env.example
.gitignore
agent_memory.py
chat_test_client.py
claude_reasoning.py
coordinator_agent.py
dashboard.html
drone_agent.py
messages.py
PROJECT_CONTEXT.md
README.md
requirements.txt
run.sh
```
### Dependencies
- requirements.txt: anthropic, arize-otel, openinference-instrumentation-anthropic, redis, redisvl, sentence-transformers, uagents, uagents-core
### Recent commits (newest first)
- Add live fleet dashboard fed by the coordinator
- Add SwarmCoordinator + drone fleet: Claude reasoning, Redis vector memory, Arize tracing
- Initial commit
## Key source files (fetched from GitHub, selected and truncated for size)
### PROJECT_CONTEXT.md
```markdown
# Project: Swarm Tasking
## What we're building
An autonomous coordination system for drone fleets doing inspection/survey
work (e.g. solar farm panel inspection). When a disruption happens mid-mission
(a drone's battery drops critical, a weather cell rolls over part of the
site), the fleet renegotiates task assignments in real time instead of
waiting for a human to manually replan.
Each drone is modeled as an independent agent. A Coordinator agent receives
disruption events, queries the Drone agents for their current status
(battery, position, task queue), and uses Claude to reason over their
responses and produce a reassignment plan with a plain-English rationale.
This is pitched as advisory coordination infrastructure that assists human
fleet managers, NOT autonomous flight control. Important framing, don't let
any copy/pitch text imply we're replacing human authority over flight
decisions.
## Why this matters / who pays
Commercial drone fleet operators (solar/utility inspection, agricultural
survey, infrastructure inspection, post-disaster survey) are scaling from
single-drone to multi-drone operations as BVLOS waivers expand, but have no
real coordination software, just a person watching multiple screens. We sell
fleet-coordination software priced per-drone-per-month, similar to fleet
management software in trucking/logistics.
## Architecture
- **Agent framework**: uAgents (Fetch.ai), using the Agent Chat Protocol
(chat_protocol_spec) so agents are discoverable/usable through ASI:One.
- **Agent types** (build in this order):
1. `SwarmCoordinator` — entry point, receives chat messages via ASI:One,
orchestrates everything else. (This one exists already, see
coordinator_agent.py)
2. `DroneAgent` (one instance per simulated drone) — holds its own state
(battery %, position, current task queue), responds to status queries
from the Coordinator.
3. (Stretch) `HazardAgent` — independently monitors simulated
weather/battery thresholds and proactively notifies the Coordinator of
disruptions, rather than disruptions only being manually triggered.
- **Claude**: called from inside the Coordinator's message handler. Input:
current fleet state (from querying Drone agents) + the disruption event.
Output: a reassignment plan (which drone takes which task) plus a
human-readable rationale.
- **Redis**: stores live fleet state and logs every agent-to-agent message
exchanged (timestamp, from, to, content). This log is both our audit trail
and, later, the data source for an optional live map visualization.
- **Arize**: traces the Claude reasoning calls inside the Coordinator, plus
an evaluator that scores plan quality (did it respect battery-safety
margins, did it maximize task coverage).
- **Devin**: used in parallel for self-contained, low-ambiguity tasks only
(e.g. the drone state simulator / Redis read-write layer). NOT used for
the Claude reasoning logic or core agent message handling, that stays
[truncated — 1733 more characters]
```
### requirements.txt
```
# Core agent framework (Fetch.ai)
uagents
uagents-core
# Claude reasoning + LLM-as-judge
anthropic
# Agent memory: Redis vector search
redis
redisvl
sentence-transformers
# Observability: Arize AX tracing for LLM calls
arize-otel
openinference-instrumentation-anthropic
```
### messages.py
```python
"""
Shared message schemas exchanged between the SwarmCoordinator and DroneAgents.
These are kept in ONE module so both sides use byte-identical typed Models.
uAgents routes messages by each Model's schema digest, so the field names and
types MUST match exactly on both ends — defining them once here guarantees that
and prevents silent delivery failures from schema drift.
"""
from uagents import Model
class StatusRequest(Model):
"""Sent by the Coordinator to ask a drone for its current state."""
requester: str # who's asking (Coordinator's address), for logging
class StatusResponse(Model):
"""Sent by a Drone back to the Coordinator with its current state."""
drone_id: str
battery_pct: int
position: str # simple string for now, e.g. "Grid-B4"
current_task: str
remaining_tasks: list[str]
can_accept_more: bool # true if battery/capacity allows taking on extra tasks
class TaskAssignment(Model):
"""Sent by the Coordinator to a Drone to assign it a new task."""
task: str
reason: str # human-readable rationale, so the drone (and logs) know why
class TaskAck(Model):
"""Drone confirms it accepted (or rejected) a task assignment."""
drone_id: str
accepted: bool
note: str = ""
```
### run.sh
```shell
#!/usr/bin/env bash
#
# run.sh — launch the full Groundtruth stack (SwarmCoordinator + 3 DroneAgents)
# with all required secrets loaded from a local .env file.
#
# Usage:
# cp .env.example .env # then fill in your real values
# ./run.sh
#
set -euo pipefail
cd "$(dirname "$0")"
# ---- Load secrets from .env ----
if [ ! -f .env ]; then
echo "ERROR: .env not found."
echo "Create it from the template: cp .env.example .env (then fill in your values)"
exit 1
fi
set -a
# shellcheck disable=SC1091
source .env
set +a
# ---- Validate required secrets ----
required=(ANTHROPIC_API_KEY REDIS_URL ARIZE_SPACE_ID ARIZE_API_KEY)
missing=()
for v in "${required[@]}"; do
[ -n "${!v:-}" ] || missing+=("$v")
done
if [ "${#missing[@]}" -gt 0 ]; then
echo "ERROR: missing required values in .env: ${missing[*]}"
exit 1
fi
# AGENTVERSE_API_TOKEN is injected for completeness but not currently consumed by
# the agents (the coordinator's mailbox is claimed interactively via Agentverse).
[ -n "${AGENTVERSE_API_TOKEN:-}" ] || echo "WARN: AGENTVERSE_API_TOKEN not set (optional)."
PY="${PYTHON:-python3}"
mkdir -p logs
echo "Starting Groundtruth stack..."
# ---- Launch the 3 drones (each its own port + stable seed via --id) ----
nohup "$PY" drone_agent.py --id 1 --battery 87 --tasks "Panel A1,Panel A2,Panel A3" --port 8001 > logs/drone1.log 2>&1 &
D1=$!
nohup "$PY" drone_agent.py --id 2 --battery 64 --tasks "Panel B1,Panel B2" --port 8002 > logs/drone2.log 2>&1 &
D2=$!
nohup "$PY" drone_agent.py --id 3 --battery 91 --tasks "Panel C1,Panel C2,Panel C3,Panel C4" --port 8003 > logs/drone3.log 2>&1 &
D3=$!
# ---- Launch the coordinator ----
nohup "$PY" coordinator_agent.py > logs/coordinator.log 2>&1 &
C=$!
# ---- Wait for each agent's HTTP server to come up ----
echo "Waiting for agents to start..."
for log in logs/drone1.log logs/drone2.log logs/drone3.log logs/coordinator.log; do
for _ in $(seq 1 40); do
if grep -q "Starting server" "$log" 2>/dev/null; then break; fi
sleep 0.5
done
done
sleep 1
# ---- Summary ----
printf '\n%s\n' "================== Groundtruth is running =================="
printf '%-18s %-6s %-8s %s\n' "COMPONENT" "PORT" "PID" "LOG"
printf '%-18s %-6s %-8s %s\n' "SwarmCoordinator" "8000" "$C" "logs/coordinator.log"
printf '%-18s %-6s %-8s %s\n' "Drone-1" "8001" "$D1" "logs/drone1.log"
printf '%-18s %-6s %-8s %s\n' "Drone-2" "8002" "$D2" "logs/drone2.log"
printf '%-18s %-6s %-8s %s\n' "Drone-3" "8003" "$D3" "logs/drone3.log"
printf '%s\n' "==========================================================="
printf '\n Memory : Redis vector index "groundtruth_incidents"\n'
printf ' Tracing : Arize project "groundtruth" (https://app.arize.com)\n'
printf '\n Tail logs: tail -f logs/coordinator.log\n'
printf ' Test : %s chat_test_client.py --scenario 1 (local)\n' "$PY"
printf ' ...or chat the SwarmCoordinator directly in ASI:One.\n'
printf ' Stop all : pkill -f coordinator_agent.py ; pkill -f drone_agent.py\n\n'
```
### chat_test_client.py
```python
"""
THROWAWAY local test client — NOT part of the product.
Simulates what ASI:One does: sends a chat message (a disruption scenario) to the
SwarmCoordinator and prints whatever chat replies come back. Used to verify the
full pipeline (query → collect → memory recall → Claude → reply) without the
ASI:One UI.
Run while coordinator_agent.py (8000) and the drones are up. Pick a scenario:
python chat_test_client.py --scenario 1 # battery-critical (wording A)
python chat_test_client.py --scenario 2 # battery-critical (wording B, semantically similar to 1)
python chat_test_client.py --scenario 3 # weather exclusion zone (different scenario)
python chat_test_client.py --message "custom disruption text"
"""
import argparse
from datetime import datetime
from uuid import uuid4
from uagents import Agent, Context, Protocol
from uagents_core.contrib.protocols.chat import (
ChatAcknowledgement,
ChatMessage,
TextContent,
chat_protocol_spec,
)
COORDINATOR_ADDRESS = "agent1qw7awftrnyz2haxmwc7frd0u2mweelukfcueeer6lg2xcqq0mvef608jgmm"
# Disruption scenarios. 1 and 2 describe the SAME battery-critical situation in
# different words (should match each other in memory via semantic similarity);
# 3 is a genuinely different scenario (should NOT strongly match 1 or 2).
SCENARIOS = {
1: "Drone-2's battery has dropped critical, reassign its remaining tasks",
2: "Drone-2 is running dangerously low on power and can't finish its route, "
"hand its remaining inspections to another drone",
3: "A weather cell is moving into Grid-C creating a no-fly exclusion zone; "
"pull the affected drone and redistribute its tasks",
}
parser = argparse.ArgumentParser()
parser.add_argument("--scenario", type=int, choices=sorted(SCENARIOS), default=1)
parser.add_argument("--message", type=str, default=None,
help="Custom disruption text (overrides --scenario)")
args = parser.parse_args()
DISRUPTION_TEXT = args.message or SCENARIOS[args.scenario]
# mailbox=False + explicit local endpoint so the coordinator can reply to us directly.
client = Agent(
name="ChatTestClient",
seed="groundtruth_chat_test_client_seed",
port=8005,
endpoint=["http://127.0.0.1:8005/submit"],
mailbox=False,
)
chat_proto = Protocol(spec=chat_protocol_spec)
def text_chat(text: str) -> ChatMessage:
return ChatMessage(
timestamp=datetime.utcnow(),
msg_id=uuid4(),
content=[TextContent(type="text", text=text)],
)
@client.on_event("startup")
async def send_probe(ctx: Context):
ctx.logger.info(f"[client] Sending disruption to coordinator: {DISRUPTION_TEXT!r}")
await ctx.send(COORDINATOR_ADDRESS, text_chat(DISRUPTION_TEXT))
@chat_proto.on_message(ChatMessage)
async def on_reply(ctx: Context, sender: str, msg: ChatMessage):
for item in msg.content:
if isinstance(item, TextContent):
ctx.logger.info(f"[client] <<< REPLY from coordinator:\n{item.text}")
# ack the reply
await ctx.send(
sender,
ChatAcknowledgement(timestamp=datetime.utcnow(), acknowledged_msg_id=msg.msg_id),
)
@chat_proto.on_message(ChatAcknowledgement)
async def on_ack(ctx: Context, sender: str, msg: ChatAcknowledgement):
ctx.logger.info(f"[client] got ack for {msg.acknowledged_msg_id}")
client.include(chat_proto)
if __name__ == "__main__":
print(f"ChatTestClient address: {client.address}")
print(f"Scenario: {DISRUPTION_TEXT!r}")
client.run()
```
### drone_agent.py
```python
"""
DroneAgent
Represents a single drone in the fleet. Holds its own state (battery,
position, current task queue) and responds to status queries from the
SwarmCoordinator agent.
Run one process per drone, e.g.:
python drone_agent.py --id 1 --battery 87 --tasks "Panel A1,Panel A2,Panel A3"
python drone_agent.py --id 2 --battery 64 --tasks "Panel B1,Panel B2"
python drone_agent.py --id 3 --battery 91 --tasks "Panel C1,Panel C2,Panel C3,Panel C4"
Each instance needs a UNIQUE seed (derived from --id below) so it gets its
own stable agent address on Agentverse.
"""
import argparse
from datetime import datetime
from uuid import uuid4
from uagents import Agent, Context, Protocol
# ---- Message schemas (agent-to-agent, not chat protocol) ----
# These typed Models are defined once in messages.py and shared with the
# Coordinator so both sides use byte-identical schemas (uAgents routes by
# schema digest). Keeping them typed is what makes this "real" agent
# communication, not just string passing.
from messages import StatusRequest, StatusResponse, TaskAssignment, TaskAck
# ---- CLI args so we can launch multiple distinct drones from one file ----
parser = argparse.ArgumentParser()
parser.add_argument("--id", type=str, required=True, help="Drone identifier, e.g. 1, 2, 3")
parser.add_argument("--battery", type=int, default=100, help="Starting battery percentage")
parser.add_argument("--position", type=str, default="Grid-A1", help="Starting position")
parser.add_argument(
"--tasks",
type=str,
default="Panel 1,Panel 2,Panel 3",
help="Comma-separated initial task list",
)
parser.add_argument(
"--port",
type=int,
default=8001,
help="Local port for this drone's server (must be unique per running agent)",
)
args = parser.parse_args()
DRONE_ID = f"Drone-{args.id}"
# ---- Mutable in-memory state for this drone ----
# (Swap this for Redis later — for now it's local to the process, which is
# fine since each drone IS its own process.)
state = {
"battery_pct": args.battery,
"position": args.position,
"remaining_tasks": [t.strip() for t in args.tasks.split(",") if t.strip()],
"current_task": None,
}
if state["remaining_tasks"]:
state["current_task"] = state["remaining_tasks"].pop(0)
def can_accept_more_tasks() -> bool:
"""Simple battery-safety rule: don't accept new tasks below 30% battery."""
return state["battery_pct"] >= 30
# ---- Agent setup ----
drone_agent = Agent(
name=DRONE_ID,
seed=f"groundtruth_drone_seed_{args.id}", # unique per drone, keeps address stable across restarts
port=args.port,
endpoint=[f"http://127.0.0.1:{args.port}/submit"],
mailbox=True,
)
drone_proto = Protocol(name="DroneStatusProtocol", version="1.0")
@drone_proto.on_message(model=StatusRequest, replies=StatusResponse)
async def handle_status_request(ctx: Context, sender: str, msg: StatusRequest):
ctx.logger.info(f"[{DRONE_ID}] Status requested by {sender}")
response = StatusResponse(
drone_id=DRONE_ID,
battery_pct=state["battery_pct"],
position=state["position"],
current_task=state["current_task"] or "idle",
remaining_tasks=state["remaining_tasks"],
can_accept_more=can_accept_more_tasks(),
)
await ctx.send(sender, response)
@drone_proto.on_message(model=TaskAssignment, replies=TaskAck)
async def handle_task_assignment(ctx: Context, sender: str, msg: TaskAssignment):
ctx.logger.info(f"[{DRONE_ID}] Task assignment from {sender}: {msg.task} (reason: {msg.reason})")
if can_accept_more_tasks():
state["remaining_tasks"].append(msg.task)
ack = TaskAck(drone_id=DRONE_ID, accepted=True, note=f"Added to queue. Battery: {state['battery_pct']}%")
else:
ack = TaskAck(
drone_id=DRONE_ID,
accepted=False,
note=f"Rejected — battery too low ({state['battery_pct']}%)",
)
await ctx.send(sender, ack)
# ---- Simple background "flight" simulation: battery drains over time ----
# Gentle drain (1% every 30s) so a fresh fleet stays usable for testing without
# constant restarts.
@drone_agent.on_interval(period=30.0)
async def simulate_flight(ctx: Context):
if state["battery_pct"] > 0:
state["battery_pct"] = max(0, state["battery_pct"] - 1)
ctx.logger.info(f"[{DRONE_ID}] Battery now at {state['battery_pct']}%")
drone_agent.include(drone_proto, publish_manifest=True)
if __name__ == "__main__":
print(f"Starting {DRONE_ID}...")
print(f"Agent address: {drone_agent.address}")
print(f"Initial state: {state}")
drone_agent.run()
```
### agent_memory.py
```python
"""
agent_memory.py
Redis-backed vector memory for Groundtruth (Redis prize track: Agent Memory +
vector search + context retrieval).
After each disruption is resolved we store the disruption text + Claude's
reassignment plan as a Redis hash, indexed by a vector embedding of the
disruption text (RedisVL vector index). Before resolving a NEW disruption we run
a vector similarity search to retrieve the most similar past incidents and feed
them back into the reasoning prompt as extra context — so the coordinator
"remembers" how comparable situations were handled.
Embeddings: sentence-transformers/all-MiniLM-L6-v2 (384-dim, runs locally, no
extra API key). Vector index + query: RedisVL.
Every public function is defensive: any Redis/embedding failure is caught and
turned into a safe no-op (empty list / None) so a memory hiccup never crashes
the agent.
"""
import json
import os
import time
from redisvl.index import SearchIndex
from redisvl.query import VectorQuery
from redisvl.utils.vectorize import HFTextVectorizer
REDIS_URL = os.environ.get("REDIS_URL")
INDEX_NAME = "groundtruth_incidents"
KEY_PREFIX = "groundtruth:incident"
EMBED_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
EMBED_DIMS = 384 # all-MiniLM-L6-v2 output dimensionality
# RedisVL index schema: one hash per incident, vector-indexed on the disruption.
_SCHEMA = {
"index": {
"name": INDEX_NAME,
"prefix": KEY_PREFIX,
"storage_type": "hash",
},
"fields": [
{"name": "disruption", "type": "text"},
{"name": "plan_json", "type": "text"},
{"name": "timestamp", "type": "text"},
{
"name": "embedding",
"type": "vector",
"attrs": {
"dims": EMBED_DIMS,
"distance_metric": "cosine",
"algorithm": "flat",
"datatype": "float32",
},
},
],
}
# Lazily-initialised singletons (model load + index connect are not free).
_index = None
_vectorizer = None
def _get_vectorizer() -> HFTextVectorizer:
global _vectorizer
if _vectorizer is None:
_vectorizer = HFTextVectorizer(model=EMBED_MODEL)
return _vectorizer
def _get_index() -> SearchIndex:
global _index
if _index is None:
if not REDIS_URL:
raise RuntimeError("REDIS_URL is not set in the environment")
_index = SearchIndex.from_dict(_SCHEMA, redis_url=REDIS_URL)
# Ensure the index actually exists on EVERY access. It may be absent on a
# fresh DB or if it was dropped out-of-band — and querying a missing index
# errors out. create(overwrite=False) is a no-op when it already exists.
if not _index.exists():
_index.create(overwrite=False)
return _index
def store_incident(disruption: str, plan: dict) -> str | None:
"""Store a resolved incident (disruption + Claude plan) with a vector
embedding of the disruption text. Returns the Redis key, or None on failure.
"""
try:
embedding = _get_vectorizer().embed(disruption, as_buffer=True)
record = {
"disruption": disruption,
"plan_json": json.dumps(plan),
"timestamp": str(time.time()),
"embedding": embedding,
}
keys = _get_index().load([record])
return keys[0] if keys else None
except Exception as exc: # noqa: BLE001 - memory must never crash the agent
print(f"[agent_memory] store_incident failed: {type(exc).__name__}: {exc}")
return None
def retrieve_similar(disruption: str, top_k: int = 2) -> list[dict]:
"""Return up to top_k past incidents most similar to the given disruption.
Each result dict contains: disruption, plan_json, timestamp, vector_distance
(lower = more similar, cosine distance). Returns [] on any failure.
"""
try:
query_vec = _get_vectorizer().embed(disruption, as_buffer=True)
query = VectorQuery(
vector=query_vec,
vector_field_name="embedding",
return_fields=["disruption", "plan_json", "timestamp"],
num_results=top_k,
)
return _get_index().query(query) or []
except Exception as exc: # noqa: BLE001
print(f"[agent_memory] retrieve_similar failed: {type(exc).__name__}: {exc}")
return []
def format_memory_context(matches: list[dict]) -> str:
"""Turn retrieved past incidents into a plain-text block for the reasoning
prompt. Returns "" when there are no matches (so the prompt is unchanged for
the very first incident)."""
if not matches:
return ""
lines = ["Similar PAST incidents and how they were previously resolved:"]
for i, m in enumerate(matches, 1):
rationale = ""
try:
rationale = json.loads(m.get("plan_json", "{}")).get("rationale_summary", "")
except Exception: # noqa: BLE001
pass
try:
similarity = f" (similarity {1 - float(m['vector_distance']):.2f})"
except Exception: # noqa: BLE001
similarity = ""
lines.append(f"{i}. Disruption{similarity}: {m.get('disruption', '')}")
if rationale:
lines.append(f" Past resolution: {rationale}")
return "\n".join(lines)
if __name__ == "__main__":
# Standalone smoke test: store one incident, then retrieve it.
demo_plan = {
"reassignments": [
{"task": "Panel B2", "from_drone": "Drone-2", "to_drone": "Drone-3",
"reason": "highest battery"}
],
"unassignable_tasks": [],
"rationale_summary": "Drone-2 critical; its task moved to Drone-3 (most capacity).",
}
key = store_incident("Drone-2 battery dropped critical, reassign its tasks", demo_plan)
print("stored key:", key)
matches = retrieve_similar("a drone has low battery and needs its work moved", top_k=2)
print("retrieved:", len(matches), "match(es)")
print(format_memory_context(matches))
```
### dashboard.html
```html
Groundtruth — Fleet Dashboard
Groundtruth — Live Fleet
Autonomous coordination for commercial drone inspection fleets