# Project export: HiveSense 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: Non-invasive AI for beehive health: It listens and looks to catch disease, instead of killing ~300 bees per bee hive - Devpost: https://devpost.com/software/hivesense - GitHub: https://github.com/Dee-1862/HiveSense - Video: https://www.youtube.com/embed/7uJ9bPZAkqw?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Team: 2 GitHub contributor(s) — Dee-1862 (16 commits), Lance Streuber (2 commits) ## Devpost submission (written by the team) ### Inspiration To check a hive for Varroa mites, the standard test is an alcohol wash, wherein a beekeeper scoops out about 300 bees and kills them to get a single mite reading. It is lethal, invasive, slow, and it only samples one moment in time. Bees are already under huge pressure worldwide and the main tool we have for watching their health works by killing them. We wanted to flip that, such that we read a colony's health from the outside, continuously, without opening the hive or harming a single bee. ### What it does HiveSense watches bee colonies using two non-invasive signals, which includes the sound inside the hive and a camera at the entrance. Bees walk through a clear 3D-printed entrance tunnel where a vision model reads Varroa load and a mic captures traffic and acoustic stress. A fleet of AI agents turns those signals into per-hive verdicts (Varroa, queenless, swarm risk), and when the signals disagree the system does not guess but it flags needs a human and asks the beekeeper to inspect. A coordinator watches the whole apiary for patterns no single hive can see, and you can ask it in plain English ("how are my bees?") through an ASI:One chat agent. ### How we built it Vision (Varroa): a fine-tuned ViViT video transformer reads 32-frame clips of bees in the tunnel (0.986 accuracy on the VD2 dataset, paper figure plus our local check). Acoustic: RandomForest models on MFCC and spectral-shape features for colony strength (0.646 balanced accuracy, hive-held-out, matching the published baseline), a bee/no-bee gate, and a Ferrari-frequency rule for swarming. Agents (Fetch.ai uAgents): seven hive agents, one per colony. Each runs the cheap acoustic check every cycle, then decides whether the reading is ambiguous enough to spend the expensive vision test, then reconciles the two. A "Godfather" coordinator fans them in (7 to 1), finds regional Varroa and neighbour robbing, and speaks the official ASI:One chat protocol. Durable action (Orkes Agentspan): a Value-of-Information gate that acts on its own when confident and only pauses for a durable human approval on a genuine close call. Redis Stack memory (the part we are proudest of): every hive reading, sound and sight, is fused into one vector and stored in Redis. Using RediSearch HNSW, a single k-NN query recalls the most similar past states, giving the agents a memory ("this looks like a reading we confirmed last week was a false alarm") before they ever alarm the keeper. Redis also holds the live state (RedisJSON), rolling metrics (RedisTimeSeries), and pushes updates to the dashboard. One vector means one index and one query, where keeping the modalities separate needs two of each. The fused vector is a lightweight early fusion: $$ v = \text{L2}\big(\text{L2}(a_{\text{acoustic}}) \oplus \text{L2}(a_{\text{vision}})\big) $$ and retrieval ranks past readings by cosine similarity. A Vite dashboard shows the apiary live and the whole stack degrades gracefully, with no Redis it falls back to a file store and with no LLM key it falls back to deterministic logic. ### Challenges we ran into Some things are not learnable so we did not use them. Across about 100 MSPB inspections, zero hives crossed the 3 percent Varroa treatment threshold, so the acoustic Varroa label was single-class. We dropped it and kept Varroa as a vision task rather than ship a fake model. Hidden data leakage. Our queenless model looked great within-hive (0.88) but collapsed cross-hive (0.18): it had learned which colony it was hearing, not queen state. Hive-held-out validation exposed it, and we reported it honestly instead of hiding it. ### What we learned Honest evaluation is a feature, not a footnote. Hive-held-out cross-validation, reporting both within-hive and cross-hive scores, and saying plainly what the data cannot support made the project stronger and more trustworthy. On a remote (cloud) Redis, a single tiny read is network-bound and not faster than a local file, so we do not claim that. The real wins are capabilities the file store cannot do at all (filtered vector search for retrieval-augmented reasoning, time-series retention, pub/sub) and doing retrieval in one query instead of two (about 2x fewer round-trips, measured live). A single fused multimodal vector is a genuinely better unit of memory than two separate vectors, with one index, one query, and a shared space to search across. Accomplishments we are proud of A working, end-to-end, honest system: real vision and acoustic models, a real multi-agent fleet, a durable human-in-the-loop action layer, and a Redis-powered multimodal memory that we benchmarked live rather than hand-waved, all wrapped in a dashboard and queryable in plain English. ### What's next Swap the concatenation fusion for a fully bound space using ImageBind, so you could search by sound and retrieve by sight and compose modality vectors into a "Varroa-stress" direction. Then take the entrance tunnel from prototype to a weatherproof field unit on real hives. Fetch.ai • Public ASI: One Shared Chat Session: https://asi1.ai/shared-chat/2357edca-5e40-42bc-9407-1c6172289f4b • Agentverse Profile(s): https://agentverse.ai/agents/details/agent1qt5wrurzefxsk0y50yaw29awtn5n9fwl6jhqtrth6pzpafufy95ak0kktlf/profile ## README (from the GitHub repository) # HiveSense - Non-Invasive Beehive Monitoring A multimodal (acoustic + vision) pipeline for assessing beehive health *without opening the hive*, exposed as a live **Fetch.ai uAgent fleet** with an **ASI:One chat interface** and a real-time dashboard. The guiding principle of this project is **honest evaluation**: every model is validated **hive-held-out** (no colony appears in both train and test), and we report what the data can and *cannot* support rather than inflated within-hive numbers. ![HiveSense dashboard overview](images/dashboard-overview.png) ## The entrance tunnel (hardware prototype) HiveSense is *non-invasive*: instead of opening the hive, bees are filmed and recorded as they walk through a clear entrance tunnel, which is where the vision model reads Varroa load and the entrance mic captures traffic and sound. Below are the 3D-printed tunnel prototype and a field test with bees actually passing through it. | Tunnel prototype | Field testing | | :---: | :---: | | ![Entrance tunnel prototype](images/tunnel_prototype.png) | ![Tunnel field testing with live bees](images/tunnel_testing.png) | ## What actually works (honest results) All acoustic models are RandomForests on handcrafted features (13 MFCC + 9 spectral-shape descriptors, or 20 MSPB audio channels). Metrics are **balanced accuracy**, the only honest split in brackets. | Signal | Modality | Dataset | Model | Honest performance | Status | | :--- | :--- | :--- | :--- | :--- | :--- | | **Colony strength** (population) | Acoustic | MSPB | RandomForest | **0.646** hive-held-out (baseline 0.658) | shipped (`models/rf_population.pkl`) | | **Input gate** (bee / noBee) | Acoustic | To-bee | RandomForest | 0.58 hive-held-out / 0.69 within-hive | weak cross-hive (only 6 colonies) | | **Queenless** | Acoustic | To-bee | RandomForest | 0.18 hive-held-out / 0.88 within-hive | confounded with colony identity | | **Varroa mites** | Vision | VD2 | ViViT-B (Vit4V, pretrained) | 0.986 acc / 0.988 F1 (paper); locally verified | usable (`models/Vit4V_model.pth`) | | **Swarm** | Acoustic | - | Ferrari frequency rule | deterministic | rule, no training | **Key findings (see the notebooks for the evidence):** - **Population is the one solid acoustic model** - it matches the published MSPB SVM baseline. - **Queenless looks great within-hive (0.88) but collapses cross-hive (0.18)** - a textbook case of the model learning *which colony* rather than *queen state*, because each To-bee colony is recorded in only one state. Demonstrating it properly needs many colonies recorded before/after dequeening. ## Validation methodology (why hive-held-out, not a single train/test split) Every acoustic model is evaluated with **hive-held-out cross-validation** (`GroupShuffleSplit` / `GroupKFold` grouped on hive id): no colony ever appears in both the training and test folds, so the score measures generalisation to **colonies the model has never heard**, which is the only thing a deployed fleet actually faces. We deliberately do **not** use one fixed train/test split, for two reasons: 1. **Sample size.** With 53 hives (MSPB) or just 6 (To-bee), a single split is luck-driven; repeated grouped CV gives a mean and a standard deviation, which is a far more honest estimate than one number from one arbitrary partition. 2. **Leakage detection.** Reporting *both* within-hive and hive-held-out scores exposes identity leakage. The queenless model is the clearest example: it scores ~0.88 within-hive but ~0.18 hive-held-out, which proves it was memorising *which colony* rather than learning *queen state* (each To-bee colony is recorded in only one state). A single random split would have hidden this and produced a misleadingly good number. The shipped `.pkl` models are then refit on **all** labelled data (standard practice: validate with CV, deploy the model trained on everything); the stored metric is always the hive-held-out score, never the optimistic within-hive one. ## System architecture A strict **gate** runs before any acoustic health check, and acoustic capabilities are kept separate from vision capabilities (they detect different things at different scales). ```mermaid flowchart TD AudioInput[Raw Audio Input] --> Gate{Input Gate
Bee vs noBee} Gate -- "noBee (noise)" --> Discard[Discard / Halt] Gate -- "Bee (valid)" --> Feats[MFCC + Spectral-Shape Features] Feats --> QueenModel[Queenless Classifier
acoustic, RF] Feats --> PopModel[Colony-Strength Classifier
acoustic, RF] Feats --> Swarm[Swarm: Ferrari frequency rule] ImageInput[Raw Image Input] --> VisionModel[Vision Model vit4v] VisionModel --> VisualMite[Visual Varroa Detection] VisualMite -.-> Verdict((Fused
Hive Verdict)) QueenModel -.-> Verdict PopModel -.-> Verdict Swarm -.-> Verdict ``` ## Agentic layer (Fetch.ai uAgents) The models are wrapped into a **multi-agent system**: a Host-Worker pattern per hive, scaling up to a fleet-level coordinator that speaks the official ASI:One chat protocol. ```mermaid flowchart TD Feed["Apiary feed (simulated, 24h+ continuous)"] subgraph Hives ["7 hive reasoning agents (hive_agent.py)"] H["each hive: acoustic (always-on) -> decide vision -> reconcile -> Verdict"] end Feed --> Hives Hives -- "Verdict per hive" --> GF{"Godfather: coordinator.py + godfather.py"} GF -- "cross-hive: regional varroa, robbing, priorities" --> Store[("data/verdicts.json (shared store)")] Store --> API["api_server.py: /api/status + /api/apiary"] API --> Dash["Dashboard (frontend)"] Store --> Chat["asi1_agent.py: ASI:One chat protocol"] User((Beekeeper)) -- "how are my bees?" --> Chat Chat -- "answer + godfather summary" --> User Store --> HD["HiveDoctor: VoI gate (Orkes Agentspan)"] HD -- "auto-act / auto-hold (confident)" --> Store HD -- "close call: durable approval pause" --> User User -- "approve / decline" --> HD ``` Sensing (uAgents) writes verdicts; the **action** layer (Agentspan) reads them and decides whether to act alone or pause for the beekeeper. The VoI decision flow is below in the Agentspan section. ![Dashboard: per-hive detail and live operations log](images/dashboard-pt2.png) ### Agent roles 1. **Hive reasoning agents (7, one per hive)** - [`hive_agent.py`](src/agents/hive_agent.py) + [`reasoning.py`](src/agents/reasoning.py) + [`tools.py`](src/agents/tools.py). Each runs the cheap acoustic detector every cycle, then *decides* (ASI:One asi1, with a deterministic fallback) whether the reading is ambiguous enough to spend the expensive tunnel-vision test. It reconciles the two: acoustic and vision agree -> confident verdict; they clash -> it does not guess, it sets `needs_human` and asks a beekeeper. Models are called as tools; it emits a `Verdict`. 2. **The Godfather (fleet coordinator)** - [`coordinator.py`](src/agents/coordinator.py) + [`godfather.py`](godfather.py). Looks across all 7 hives for what no single hive can see: regional Varroa pressure, neighbour robbing (influx at one hive vs outflux next door, by position), and a prioritised beekeeper action list. It writes the shared verdict store and serves `GET /api/status`. 3. **ASI:One chat agent** - [`asi1_agent.py`](asi1_agent.py). Publishes the official chat protocol (`uagents_core.contrib.protocols.chat`) so the apiary is queryable from ASI:One ("how are my bees?"), answering from the live verdicts plus the Godfather's summary. Its LLM brain is switchable (ASI:One / Gemini / Claude) and degrades to a deterministic, data-only answer if no key is set. The point of this layer is that the beekeeper never has to read sensor data: they get a plain-language morning report and can ask about any hive in their own words, like texting a knowledgeable friend.

HiveSense plain-language chat: a morning report and follow-up questions abo [README truncated for size] ## Detected evidence (automated analysis) Indexed codebase: 49 recognized source files, 354 KB. - CSS (language) — detected in the code - HTML (language) — detected in the code - Hugging Face (technology) — detected in the code - JavaScript (language) — detected in the code - OpenAI (technology) — detected in the code - Python (language) — detected in the code - PyTorch (technology) — detected in the code - Redis (technology) — detected in the code ## Codebase structure (from repository index) ### Files (59 of 59) ``` .env.example .gitignore api_server.py asi1_agent_hosted.py asi1_agent.py asi1_client.py bench/redis_bench.py demo_hive_doctor.py eda/mspb_eda.ipynb eda/tobee_gate_queen_eda.ipynb explain.py frontend/index.html frontend/package.json frontend/scripts/make_media.sh frontend/src/api.js frontend/src/main.js frontend/src/style.css frontend/vite.config.js godfather.py hardware/Bee Monitor Stand Phone Holder V1.3mf hardware/Bee Monitor Stand V1.2.3mf hardware/LICENSE hardware/phone blank.3mf hive_state.py live_feed.py README.md requirements.txt scripts/redis_show.py scripts/redis_smoke.py scripts/seed_redis_unimodal.py seed_apiary.py seed_voi_demo.py src/agents/connect_mailbox.py src/agents/coordinator.py src/agents/feed.py src/agents/hive_agent.py src/agents/reasoning.py src/agents/run_coordinator.py src/agents/run_fleet.py src/agents/schema.py src/agents/tools.py src/agentspan/__init__.py src/agentspan/hive_doctor.py src/agentspan/runs.py src/agentspan/server.py src/agentspan/voi.py src/embedding.py src/imagebind_embed.py src/mspb_loader.py src/run_demo.py src/store/__init__.py src/store/base.py src/store/file_store.py src/store/redis_store.py src/tobee_loader.py src/train_gate_queenless.py src/train_population.py src/train_varroa_acoustic.py src/vit4v_infer.py ``` ### Dependencies - frontend/package.json: @geoman-io/leaflet-geoman-free@^2.19.3, leaflet@^1.9.4, vite@^8.0.16 - requirements.txt: agentspan@>=0.1.0, jupyter@>=1.0.0, librosa@>=0.10.0, matplotlib@>=3.7.0, numpy@>=1.24.0, openai@>=1.0.0, openpyxl@>=3.1.0, pandas@>=2.0.0, Pillow@>=10.0.0, python-dotenv@>=1.0.0, redis@>=5.0.0, scikit-learn@>=1.2.0, scipy@>=1.10.0, seaborn@>=0.12.0, torch@>=2.0.0, transformers@==4.44.2, uagents@>=0.11.0, uagents-core@>=0.4.7 ### Recent commits (newest first) - Updated Readme - Updated ReadMe - Update .gitignore - Stop tracking files now covered by .gitignore - Improved readme - Merge branch 'master' of https://github.com/Dee-1862/BEEAgentic - Final Commit - Merge pull request #1 from Dee-1862/add-hardware - Add 'hardware/' from commit 'cc4f3d0f938f51ed88027c69fbf353c86923d0cb' - Improved Frontend + Redis Optimization - Final Updated UI - Add files via upload - Simplified the frontend - Initial commit - Optimizing the multi agent orchestration workflow Draft 1 - Integrated the agentic pipeline - Non-invasive beehive health monitoring: acoustic + vision ML (RandomForest + ViViT) - Initial Draft ## Key source files (fetched from GitHub, selected and truncated for size) ### requirements.txt ``` librosa>=0.10.0 scikit-learn>=1.2.0 pandas>=2.0.0 numpy>=1.24.0 jupyter>=1.0.0 matplotlib>=3.7.0 seaborn>=0.12.0 scipy>=1.10.0 openpyxl>=3.1.0 torch>=2.0.0 transformers==4.44.2 openai>=1.0.0 python-dotenv>=1.0.0 uagents>=0.11.0 uagents-core>=0.4.7 # Redis Stack integration (RedisJSON + TimeSeries + RediSearch vectors + Pub/Sub). # Only needed when running with USE_REDIS=1; the file-store demo path needs none of these. redis>=5.0.0 Pillow>=10.0.0 # LSB steganography carrier (unimodal packing) # Orkes Agentspan: durable, human-in-the-loop runtime for the HiveDoctor agent # (src/agentspan). Optional - the Value-of-Information gate runs without it; with the # server up (`agentspan server start`) the approval pause executes on the durable engine. agentspan>=0.1.0 ``` ### frontend/package.json ``` { ). Serves exactly what the dashboard's api.js already calls: GET /api/status ApiaryStatusResponse shape (header data-link) GET /api/treatments {treatments: [run, ...]} (HiveDoctor runs) POST /api/treatment/start {hive} -> hive_doctor.start(hive) POST /api/treatment/respond {id, approve, note} -> hive_doctor.respond(...) GET /api/advise?hive=..&... {advice} (plain-language, VoI-grounded) Run: python -m src.agentspan.server (PORT env overrides 8000) """ from __future__ import annotations import json import os import sys from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import urlparse, parse_qs _ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) if _ROOT not in sys.path: sys.path.insert(0, _ROOT) from src.agentspan import hive_doctor, runs # noqa: E402 import hive_state # noqa: E402 # Per-dashboard-hive sensor signals. Chosen so the Value-of-Information gate produces a # realistic SPREAD - it asks only on genuine close calls and stays quiet otherwise: # B1/A3/C3 -> close call on mites -> ASK (durable approval gate) # C1 (queenless), B3 (swarm) -> confident -> AUTO-ACT # A1/A2/B2/C2 -> clearly calm -> AUTO-HOLD (just watch) SIGNALS = { "A1": dict(acoustic_stress=0.08, vision_mite_rate=0.000, vision_ran=True), # calm -> HOLD "A2": dict(acoustic_stress=0.10, vision_mite_rate=0.000, vision_ran=True), # calm -> HOLD "A3": dict(acoustic_stress=0.55, vision_mite_rate=0.000, vision_ran=False), # vision skipped, raised sound -> ASK "B1": dict(acoustic_stress=0.72, vision_mite_rate=0.038, vision_ran=True), # mites near the line -> ASK "B2": dict(acoustic_stress=0.09, vision_mite_rate=0.000, vision_ran=True), # calm -> HOLD "B3": dict(acoustic_stress=0.10, vision_mite_rate=0.000, vision_ran=True, swarm_alert=True), # swarm flag -> AUTO-ACT "C1": dict(acoustic_stress=0.10, vision_mite_rate=0.000, vision_ran=True, queenless_alert=True), # queenless -> AUTO-ACT "C2": dict(acoustic_stress=0.10, vision_mite_rate=0.000, vision_ran=True), # calm -> HOLD "C3": dict(acoustic_stress=0.50, vision_mite_rate=0.000, vision_ran=False), # vision skipped -> ASK } HIVES = list(SIGNALS) def _now() -> str: return datetime.now(timezone.utc).isoformat() def _seed_state() -> None: """Write the demo signals into the shared store as verdicts, so hive_doctor's compute_decision (which reads hive_state) runs the gate on real, varied inputs.""" verdicts = {} for code, s in SIGNALS.items(): verdicts[code] = [{ "hive_id": code, "acoustic_stress": s.get("acoustic_stress", 0.0), "vision_mite_rate": s.get("vision_mite_rate", 0.0), "vision_ran": s.get("vision_ran", False), "queenless_alert": s.get("queenless_alert", False), "swarm_alert": s.get("swarm_alert", False), "timestamp": _now(), }] try: hive_state.save_verdicts(verdicts) except Exception as e: # pragma: no cover print("warn: could not seed hive_state:", e) def _reset_registry() -> None: """Start each demo from a clean slate of treatment runs.""" try: runs._save({}) # one-time reset of data/treatments.json except Exception: pass # Live-link payload: kept identical to the mock so the header behaves the same. def _status_payload() -> dict: t = _now() return {"hives": { "hive3": [ {"hive_id": "hive3", "varroa_status": "watch", "queenless_alert": False, "swarm_alert": False, "traffic": 12, "position": [0, 0], "timestamp": t}, {"hive_id": "hive3", "varroa_status": "alert", "queenless_alert": False, "swarm_alert": False, "traffic": -82, "position": [0, 0], "timestamp": t}, ], "hive5": [ {"hive_id": "hive5", "varroa_status": "clear", "queenless_alert": False, "swarm_alert": True, "traffic": 64, "position": [3, 0], "timestamp": t}, ], }} def _advice(params: dict) -> str: """Plain-language advice grounded in the SAME VoI computation (no LLM needed, so it works keyless; swap for the Gemini advisor by setting GEMINI_API_KEY + reasoning.py).""" hive = (params.get("hive", ["?"])[0]).upper() plan = hive_doctor.compute_decision(hive) if hive in SIGNALS else None if not plan: return f"Hive {hive}: no live signals to advise on yet." lead = next((g for g in plan["gates"] if g["condition"] == plan["lead_condition"]), plan["gates"][0]) tail = {"ask": "It's a close call, so the beekeeper's judgement is worth the interruption.", "auto_act": "Confident enough to handle it without interrupting you.", "auto_hold": "All calm - just keep watching, nothing to do."}[plan["decision"]] return f"{plan['headline']} {lead['explain']} {tail}" class Handler(BaseHTTPRequestHandler): def _send(self, code: int, obj) -> None: body = json.dumps(obj).encode() self.send_response(code) self.send_header("content-type", "application/json") self.send_header("access-control-allow-origin", "*") self.send_header("access-control-allow-methods", "G [truncated — 2442 more characters] ``` ### asi1_client.py ```python """ Local test client for asi1_agent.py (from the Fetch.ai ASI:One guide). Sends one ChatMessage to the HiveSense agent and prints its reply - lets you test the agent WITHOUT ASI:One/Agentverse. Run the agent first (python asi1_agent.py), then: python asi1_client.py AI_AGENT_ADDRESS is the address of asi1_agent.py (derived from its seed). If you change the agent's seed, update this address (run: python -c "from uagents import Agent; print(Agent(seed='').address)"). """ from datetime import datetime from uuid import uuid4 from uagents import Agent, Context from uagents_core.contrib.protocols.chat import ( ChatMessage, ChatAcknowledgement, TextContent, ) AI_AGENT_ADDRESS = "agent1qt5wrurzefxsk0y50yaw29awtn5n9fwl6jhqtrth6pzpafufy95ak0kktlf" QUESTION = "How do I tell if my hive has a Varroa mite problem?" agent = Agent( name="hivesense-test-client", seed="hivesense-asi1-client-seed-v1", port=8002, endpoint=["http://127.0.0.1:8002/submit"], ) @agent.on_event("startup") async def send_message(ctx: Context): ctx.logger.info(f"Asking the agent: {QUESTION}") await ctx.send(AI_AGENT_ADDRESS, ChatMessage( timestamp=datetime.now(), msg_id=uuid4(), content=[TextContent(type="text", text=QUESTION)], )) @agent.on_message(ChatAcknowledgement) async def handle_ack(ctx: Context, sender: str, msg: ChatAcknowledgement): ctx.logger.info(f"Acknowledged by {sender} for {msg.acknowledged_msg_id}") @agent.on_message(ChatMessage) async def handle_reply(ctx: Context, sender: str, msg: ChatMessage): for item in msg.content: if isinstance(item, TextContent): ctx.logger.info(f"Reply from agent:\n{item.text}") if __name__ == "__main__": agent.run() ``` ### demo_hive_doctor.py ```python """ HiveDoctor demo - the agent that knows when NOT to ask (Value-of-Information gate). python demo_hive_doctor.py # show the VoI decision for every hive python demo_hive_doctor.py A3 # one hive, then the approval gate if it asks AGENTSPAN_LIVE=1 python demo_hive_doctor.py A3 # run on the live Agentspan engine # (after: agentspan server start) The gate (src/agentspan/voi.py) weighs the value of the beekeeper's input against the cost of interrupting them, so it stays quiet when confident and asks only on close calls. When it asks, Agentspan holds the pause durably. Grounded in "Value of Information: A Framework for Human-Agent Communication" (arXiv 2601.06407, Jan 2026). """ import sys import hive_state from src.agentspan import hive_doctor as hd def main(): args = sys.argv[1:] if args: plan = hd.run_cli(args[0]) if plan["decision"] != "ask": return # confident: the agent handled it without us run = hd.start(args[0]) if run["status"] != "awaiting_approval": return print("\n--- human-in-the-loop gate (Agentspan would hold this durably) ---") ans = input("Treat this hive? [y/N] ").strip().lower() out = hd.respond(run["id"], approve=ans in ("y", "yes"), note="decided from CLI") print("resolved:", out["status"], "->", out.get("result")) return verdicts = hive_state.load_verdicts() if not verdicts: print("No verdicts in the store yet. Run: python seed_apiary.py") return print(f"HiveDoctor VoI sweep over {len(verdicts)} hives " f"(Agentspan {'available' if hd.HAS_AGENTSPAN else 'not installed - gate still runs'}):\n") tag = {"ask": "ASK YOU", "auto_act": "ACT ", "auto_hold": "WATCH"} for hive_id in verdicts: p = hd.compute_decision(hive_id) print(f"[{tag[p['decision']]}] {p['headline']}") if __name__ == "__main__": main() ``` ### hive_state.py ```python """ Shared apiary state for the dashboard, fleet and ASI:One agent. The hive fleet / coordinator writes the latest verdicts here; the dashboard API and the ASI:One agent read them back. This file is the only coupling between the chat agent and the monitoring fleet - a simple shared store, no network needed. The actual storage is now pluggable (see src/store): with USE_REDIS=1 it is backed by Redis Stack, otherwise by data/verdicts.json. The functions below keep their old signatures so every existing caller (coordinator, godfather, api_server, asi1_agent) works unchanged - they just delegate to the selected backend. """ import os from src.store import get_store ROOT = os.path.dirname(os.path.abspath(__file__)) def load_verdicts(): """Return {hive_id: [verdict, ...]} from the shared store, or {} if unavailable.""" return get_store().load_verdicts() def save_verdicts(verdicts): """Write the verdicts store (called by the coordinator as verdicts arrive).""" return get_store().save_verdicts(verdicts) def append_verdict(verdict: dict, max_points: int = 96): """Append one live verdict to its hive's rolling history (keeps the 24h seed).""" return get_store().append_verdict(verdict, max_points) def apiary_summary(verdicts=None): """Plain-text status of every hive, flagging any that need a human inspection.""" verdicts = load_verdicts() if verdicts is None else verdicts if not verdicts: return "No live hive data is available yet (the monitoring fleet may be offline)." lines, needs = [], [] for hid, hist in sorted(verdicts.items()): v = hist[-1] if isinstance(hist, list) and hist else hist if not isinstance(v, dict): continue flag = "" if v.get("needs_human"): needs.append(hid) flag = f" [NEEDS INSPECTION: {v.get('reason', '')}]" lines.append( f"- Hive {hid}: varroa={v.get('varroa_status', '?')}, " f"queenless={v.get('queenless_alert', '?')}, swarm={v.get('swarm_alert', '?')}, " f"net_traffic={v.get('traffic', '?')}{flag}" ) header = "Current apiary status (live):" if needs: header += f" {len(needs)} hive(s) need inspection: {', '.join(needs)}." return header + "\n" + "\n".join(lines) ``` ### seed_voi_demo.py ```python """ Give the apiary a spread of severities so the Value-of-Information gate visibly does its job: stay quiet when confident, ask only on close calls. Each hive's LATEST acoustic/vision reading is set to a designed scenario (the rest of its 24h history is left untouched). Re-runnable and reversible (data/verdicts.json is tracked in git: `git checkout data/verdicts.json` to undo). This only changes simulated demo data, not any model. python seed_voi_demo.py # then: python demo_hive_doctor.py (or open the dashboard) Expected outcome with the paper's default costs (varroa c_fn=150, c_fp=15, c_ask=3): - very calm hives -> AUTO-HOLD (agent watches, never bothers you) - clearly infested hives -> AUTO-ACT (agent treats on its own, no interruption) - genuine close calls -> ASK YOU (durable Agentspan approval pause) """ import hive_state from src.agentspan import hive_doctor as hd # (acoustic_stress, vision_mite_rate, vision_ran) per hive - chosen to span the VoI regimes SCENARIOS = { "A1": (0.01, 0.00, False), # dead calm -> auto-hold "C1": (0.02, 0.00, False), # calm -> auto-hold "B1": (0.85, 0.12, True), # both sensors high -> auto-act (clearly infested) "A2": (0.80, 0.11, True), # both sensors high -> auto-act "A3": (0.75, 0.00, True), # sound says yes, camera says no (disagree) -> ask "B2": (0.55, 0.00, False), # middling sound only -> ask "B3": (0.60, 0.04, True), # mixed, just over the line -> ask } def main(): verdicts = hive_state.load_verdicts() if not verdicts: print("No verdicts yet. Run: python seed_apiary.py") return for hive_id, (ac, vr, ran) in SCENARIOS.items(): hist = verdicts.get(hive_id) if not hist: continue latest = dict(hist[-1]) latest["acoustic_stress"] = ac latest["vision_mite_rate"] = vr latest["vision_ran"] = ran hist[-1] = latest verdicts[hive_id] = hist hive_state.save_verdicts(verdicts) print("Seeded VoI demo scenarios. Decisions now:") tag = {"ask": "ASK YOU", "auto_act": "ACT", "auto_hold": "WATCH"} for hive_id in SCENARIOS: p = hd.compute_decision(hive_id) g = p["varroa"] print(f" {hive_id}: p={g['p']:.2f} EVPI=${g['evpi']:.2f} vs ${g['c_ask']}" f" -> {tag[p['decision']]}") if __name__ == "__main__": main() ``` ### live_feed.py ```python """ Live apiary feed - extends the seeded 24h history in real time. Every FEED_INTERVAL seconds it appends one new verdict per hive (continuing each hive's storyline, with diurnal + noisy traffic) to data/verdicts.json, and prints the Godfather's apiary-wide read. The api_server re-reads the file per request, so the dashboard updates live; the ASI:One agent's answers change too. HONEST FRAMING: the sensor FEED is simulated (no live 7-hive hardware), but the agent logic and ML models are real. Present as "real agents/models on a simulated apiary feed." Run (own terminal): python live_feed.py (Ctrl+C to stop) FEED_INTERVAL=5 python live_feed.py Needs the seed first: python seed_apiary.py """ import os import time import math import random from datetime import datetime import hive_state import godfather from seed_apiary import _verdict, POSITIONS INTERVAL = float(os.getenv("FEED_INTERVAL", "10")) # Each hive's steady "regime" = where the 24h seed left it. The feed keeps them here # (so the demo narrative is stable) while traffic and acoustic signals move believably. REGIME = { "A1": {"varroa": "clear"}, "A2": {"varroa": "clear"}, "C1": {"varroa": "clear"}, "A3": {"varroa": "alert", "reason": "Acoustic stress and visible mites agree: treat this week."}, "B1": {"varroa": "watch", "needs_human": True, "reason": "Signals disagree: acoustic=stressed but vision=clear. Please inspect and confirm."}, "B2": {"varroa": "clear", "swarm": True, "reason": "Swarm spike detected; heads-up to the yard."}, "B3": {"varroa": "clear", "queenless": True, "reason": "Queenless roar detected."}, } def _diurnal(now): """Rough day/night foraging curve: low at night, peak midday.""" h = now.hour + now.minute / 60.0 return int(45 * max(0.0, math.sin(math.pi * h / 24.0))) def next_verdict(hive, now): r = REGIME[hive] traffic = _diurnal(now) + random.randint(-8, 8) if r.get("swarm"): traffic = -75 + random.randint(-6, 6) # mass outflux during a swarm elif r.get("needs_human"): traffic = 60 + random.randint(-6, 6) # the clash hive runs hot return _verdict( hive, now, varroa=r.get("varroa", "clear"), queenless=r.get("queenless", False), swarm=r.get("swarm", False), traffic=traffic, needs_human=r.get("needs_human", False), reason=r.get("reason", ""), ) def main(): print(f"Live feed: appending 1 verdict/hive every {INTERVAL:.0f}s. Ctrl+C to stop.") try: while True: now = datetime.now() for hive in POSITIONS: hive_state.append_verdict(next_verdict(hive, now)) head = godfather.apiary_analysis(hive_state.load_verdicts())["headline"] print(f"[{now:%H:%M:%S}] {head}") time.sleep(INTERVAL) except KeyboardInterrupt: print("\nlive feed stopped.") if __name__ == "__main__": main() ``` ### asi1_agent_hosted.py ```python """ HiveSense - HOSTED Agentverse agent (no local mailbox, no 401). Deploy on Agentverse so it runs on their infra and ASI:One can reach it directly: 1. agentverse.ai -> Agents -> + New Agent -> Blank Agent (Hosted). 2. Paste this whole file into the editor. 3. In the agent's Secrets, add: ASI_ONE_API_KEY = 4. Click Run. Then use "Chat with Agent" / ASI:One. Hosted agents have a restricted package set, so this uses only `requests` for the LLM call (not the openai SDK). Live local hive data can't be read from Agentverse, so a representative apiary snapshot is embedded; the live dashboard shows the real-time data. """ import os from datetime import datetime from uuid import uuid4 import requests from uagents import Agent, Context, Protocol from uagents_core.contrib.protocols.chat import ( ChatAcknowledgement, ChatMessage, EndSessionContent, TextContent, chat_protocol_spec, ) ASI_ONE_API_KEY = os.environ.get("ASI_ONE_API_KEY", "") # Representative apiary snapshot (hosted agents can't read your local files). APIARY_SNAPSHOT = ( "Current apiary snapshot: 7 hives. A3 is on Varroa ALERT (treat this week). " "B1 needs inspection (acoustic stress but vision sees no mites - signals disagree). " "B2 shows a swarm signal. B3 reads queenless. A1, A2, C1 are healthy." ) SYSTEM_PROMPT = ( "You are the HiveSense apiary assistant: an expert on beehive health, Varroa mites, " "queen status, swarming, and bee acoustics. Use the apiary snapshot below to answer " "questions about the user's hives; for general bee questions, answer from expertise. " "Keep replies concise and practical.\n\n" + APIARY_SNAPSHOT ) def ask_llm(question: str) -> str: if not ASI_ONE_API_KEY: return ("(No ASI_ONE_API_KEY secret set.) " + APIARY_SNAPSHOT) try: r = requests.post( "https://api.asi1.ai/v1/chat/completions", headers={"Authorization": f"Bearer {ASI_ONE_API_KEY}", "Content-Type": "application/json"}, json={"model": "asi1", "messages": [{"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": question}], "max_tokens": 2048}, timeout=30, ) r.raise_for_status() return r.json()["choices"][0]["message"]["content"] except Exception as e: return f"I could not reach the language model right now. {APIARY_SNAPSHOT} (error: {e})" agent = Agent(name="hivesense", seed=os.environ.get("AGENT_SEED", "hivesense-hosted-seed-v1")) protocol = Protocol(spec=chat_protocol_spec) @protocol.on_message(ChatMessage) async def handle_message(ctx: Context, sender: str, msg: ChatMessage): await ctx.send(sender, ChatAcknowledgement( timestamp=datetime.now(), acknowledged_msg_id=msg.msg_id)) question = "".join(i.text for i in msg.content if isinstance(i, TextContent)) ctx.logger.info(f"Chat query: {question}") answer = ask_llm(question) await ctx.send(sender, ChatMessage( timestamp=datetime.utcnow(), msg_id=uuid4(), content=[TextContent(type="text", text=answer), EndSessionContent(type="end-session")], )) @protocol.on_message(ChatAcknowledgement) async def handle_ack(ctx: Context, sender: str, msg: ChatAcknowledgement): pass agent.include(protocol, publish_manifest=True) if __name__ == "__main__": agent.run() ``` ### godfather.py ```python """ The Godfather: apiary-level orchestration logic. Each hive agent decides about its OWN hive. The godfather looks across ALL hives and produces the apiary-wide picture no single hive can see: regional Varroa spread, neighbour robbing (influx at one hive vs outflux next door), and a prioritised action list for the beekeeper. Pure functions over the verdict store, so the live feed, the API server, and the ASI:One agent all share one brain. """ NEIGHBOR_DIST = 10.0 # metres; closer hives can rob each other FLOW_THRESHOLD = 50 # net bees/cycle that counts as a real surge or drain def _latest(verdicts): out = {} for h, hist in verdicts.items(): v = hist[-1] if isinstance(hist, list) and hist else hist if isinstance(v, dict): out[h] = v return out def _dist(p, q): if not p or not q or len(p) < 2 or len(q) < 2: return float("inf") return ((p[0] - q[0]) ** 2 + (p[1] - q[1]) ** 2) ** 0.5 def apiary_analysis(verdicts: dict) -> dict: """Return the apiary-wide picture: counts, emergent patterns, priorities, headline.""" latest = _latest(verdicts) n = len(latest) alerts = [h for h, v in latest.items() if v.get("varroa_status") == "alert"] watches = [h for h, v in latest.items() if v.get("varroa_status") == "watch"] needs_human = [h for h, v in latest.items() if v.get("needs_human")] queenless = [h for h, v in latest.items() if v.get("queenless_alert")] swarming = [h for h, v in latest.items() if v.get("swarm_alert")] # emergent: regional varroa pressure across the yard emergent = [] if len(alerts) + len(watches) >= 2: emergent.append(f"Regional Varroa pressure across {sorted(alerts + watches)} - treat the row, not one hive.") # emergent: possible robbing between neighbours (influx vs outflux) robbing = [] hids = list(latest) for i in range(len(hids)): for j in range(i + 1, len(hids)): a, b = latest[hids[i]], latest[hids[j]] if _dist(a.get("position"), b.get("position")) > NEIGHBOR_DIST: continue ta, tb = a.get("traffic", 0), b.get("traffic", 0) if ta >= FLOW_THRESHOLD and tb <= -FLOW_THRESHOLD: robbing.append((hids[j], hids[i])) # (robbed, robber) elif tb >= FLOW_THRESHOLD and ta <= -FLOW_THRESHOLD: robbing.append((hids[i], hids[j])) for robbed, robber in robbing: emergent.append(f"Possible robbing: {robbed} draining while neighbour {robber} surges.") # prioritised actions (most urgent first) priorities = [] for h in needs_human: priorities.append({"hive": h, "action": "inspect", "why": latest[h].get("reason", "needs inspection")}) for h in alerts: priorities.append({"hive": h, "action": "treat varroa", "why": "mite load over the economic threshold"}) for h in queenless: priorities.append({"hive": h, "action": "requeen", "why": "queenless signature"}) for h in swarming: priorities.append({"hive": h, "action": "swarm control", "why": "pre-swarm signal"}) parts = [] if alerts: parts.append(f"{len(alerts)} on Varroa alert ({', '.join(sorted(alerts))})") if needs_human: parts.append(f"{len(needs_human)} need inspection ({', '.join(sorted(needs_human))})") if queenless: parts.append(f"{len(queenless)} queenless ({', '.join(sorted(queenless))})") if swarming: parts.append(f"{len(swarming)} swarming ({', '.join(sorted(swarming))})") healthy = n - len(set(alerts + watches + needs_human + queenless + swarming)) headline = (f"Apiary: {n} hives, {healthy} healthy. " + ("; ".join(parts) + "." if parts else "All hives nominal.") + (f" Top priority: {priorities[0]['hive']} ({priorities[0]['action']})." if priorities else "")) return { "n_hives": n, "healthy": healthy, "alerts": sorted(alerts), "watches": sorted(watches), "needs_human": sorted(needs_human), "queenless": sorted(queenless), "swarming": sorted(swarming), "emergent": emergent, "priorities": priorities, "headline": headline, } if __name__ == "__main__": import hive_state a = apiary_analysis(hive_state.load_verdicts()) print(a["headline"]) for e in a["emergent"]: print(" -", e) for p in a["priorities"]: print(f" * {p['hive']}: {p['action']} ({p['why']})") ``` [39 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]