DEV Community

Rubens Zimbres
Rubens Zimbres

Posted on • Originally published at Medium on

Develop a Multi-Orchestration Neighborhood Watch App with Antigravity CLI

#fbi #googleantigravity #truecrime #googlecloudplatform

I always develop sample apps focusing on the real utility for the community. I imagined the following situation: if you are a tourist, or even a person looking to rent or buy a real state property, it is useful to check the environmental factors in that area, as well as infrastructure complaints and crime rates.

These are the main reasons I developed the Neighborhood Watch App, as a valuable tool for people to understand the risks of certain US areas. Neighborhood Alert aggregates crime reports, weather alerts, 311 infrastructure complaints, and seismic data for any US address, normalizes them into a single composite risk score, and runs the aggregated collected data through Google Gemini to produce an actionable safety briefing.

The collected data pipeline is built as a multi-agent orchestration layer : each data source is an isolated agent in a dependency graph. The graph runs on a background thread, and the Streamlit UI streams live per-subagent progress while the work executes.

ARCHITECTURE


Multi-agent architecture for the Neighborhood Watch App

The application has two execution paths over the same underlying fetch methods:

  • Synchronous path (DataFetcher.fetch_all): geocode, then a 5-way parallel fan-out, then risk scoring. Used by tests and any non-UI caller.
  • Orchestrated path (utils/orchestration.start_pipeline): fetches agents in a dependency graph, executed on a background thread with live progress and the AI briefing as the final node. Used by the UI.

We have the following Components and Responsibilities:

Agent: is a named unit of work. An Orchestrator runs agents as topological units, many in parallel (APIs). Each unit executes in a thread pool and failures are isolated per node. The worker thread writes on the Progress Board. The final assembler is given by Gemini and makes sense of the data collected via APIs.

I used the following APIs:

  • Geocode for US addresses, using OpenStreetMap
  • Weather: OpenWeather maps + NWS alerts
  • NWS: National Weather Service:
  • Crime: uses FBI Data Explorer, for statistics and aggregated data
  • Infrastructure: 311 Open Data (Socrata)
  • Seismic: USGS EarthQuakes
  • Composer: Gemini

This data will generate a given risk for the area, and Gemini provides a briefing for the user. I developed the sketch project, a framework agnostic multi-agent system, then I converted it to ADK, with Antigravity CLI, w_ith the **/goal_** command. When using this command, be sure to have the complete architecture and requirements of the system in the prompt, or precede it with the /planning command, as the /goal is a fully autonomous execution command.


Antigravity CLI performing a task

The Geocoding returns coordinates (lat long) for any given US address. Weather API returns a 5-day forecast, returning a risk score. With the FBI Crime Data Explorer you have violent crimes, theft, burglary and other offenses, providing totals that will be use the severity of risk in that area.

The Socrata Open Data API has many different endpoints, each one related to a specific city. The 311 infrastrucuture complaints API provides datasets per city. The seismic app, that I also used here, in the Xtreme Weather App, uses USGS (United States Geological Survey) provides a GeoJSON with events from a 500km limit radius of the geocoded point, with minimum magnitude of 2.0 in Richter Scale, limited to the 10 most recent results.

Here, I used gemini-3.5-flash via Vertex AI to aggregate and interpret the results.


Project Structure for the Neighborhood Watch App

ANTIGRAVITY

In order to install Antigravity-cli on Linux to help you code the solution, you run:

curl -fsSL https://antigravity.google/cli/install.sh | bash
Enter fullscreen mode Exit fullscreen mode

You if are a Gemini-cli user, you can import settings in a simple manner:

agy plugin import gemini
Enter fullscreen mode Exit fullscreen mode

You can also get some skills here: https://github.com/google-gemini/gemini-skills, and add them in this path: ~/.gemini/antigravity-cli/skills/

Your settings are now in: ~/.gemini/antigravity-cli/settings.json

Simple as that. For a more detailed walkthrough, check these series of articles from Romin Irani:

Antigravity CLI Tutorial Series

… and from xbill:

Getting Started with Antigravity CLI

The Antigravity-cli documentation can be accessed here:

https://antigravity.google/docs/cli/overview

CODE SNIPPETS

Initially we define the API endpoints, cities’ total population (for the risk score), and the severity weights:

SEVERITY_WEIGHTS = {
    "Violent Crime": 10,
    "Weapons Offense": 8,
    "Theft & Burglary": 4,
    "Vehicle Incident": 3,
    "Drug Offense": 2,
    "Fraud & Deception": 2,
    "Property Damage/Vandalism": 2,
    "Trespassing": 1,
}
Enter fullscreen mode Exit fullscreen mode

Then we use the data fetcher, to obtain information from the APIs (the full code Gurhub repo is provided ahead).

def _fetch_socrata_crime(self, lat: float, lon: float, city_key: str) -> List[Dict]:
    url = SOCRATA_ENDPOINTS[city_key]
    since = (datetime.utcnow() - timedelta(days=30)).strftime("%Y-%m-%dT00:00:00.000")
    headers = {"X-App-Token": SOCRATA_TOKEN} if SOCRATA_TOKEN else {}

    if city_key == "chicago":
        params = {
            "$limit": 500,
            "$where": f"date >= '{since}'",
            "$select": "date,primary_type,description,location_description,latitude,longitude,block",
            "$order": "date DESC",
        }
    elif city_key == "new york":
        params = {
            "$limit": 500,
            "$where": f"cmplnt_fr_dt >= '{since}'",
            "$select": "cmplnt_fr_dt,ofns_desc,law_cat_cd,boro_nm,latitude,longitude",
            "$order": "cmplnt_fr_dt DESC",
        }
    else:
        params = {"$limit": 300, "$order": ":id DESC"}

    try:
        r = self._safe_get(url, params=params, headers=headers, timeout=12)
        r.raise_for_status()
        raw = r.json()

        # If the 30-day filter returned no data (often due to city data delays), fetch the latest available
        if not raw and "$where" in params:
            del params["$where"]
            params["$limit"] = 100
            r = self._safe_get(url, params=params, headers=headers, timeout=12)
            r.raise_for_status()
            raw = r.json()

    except Exception as e:
        logger.warning("[Socrata:%s] crime fetch failed: %s", city_key, e)
        return []

    incidents = []
    for row in raw:
        try:
            rlat = float(row.get("latitude") or row.get("lat") or 0)
            rlon = float(row.get("longitude") or row.get("lon") or row.get("lng") or 0)
        except (ValueError, TypeError):
            rlat, rlon = lat, lon

        raw_type = (
            row.get("primary_type") or
            row.get("ofns_desc") or
            row.get("crm_cd_desc") or
            row.get("crime_type") or
            row.get("nibrs_crime") or
            row.get("nibrs_crime_category") or
            row.get("offense_sub_category") or
            row.get("offense_parent_group") or
            row.get("offense") or
            row.get("offense_description") or
            row.get("incident_type_primary") or
            row.get("offense_type") or
            row.get("highest_nibrs_ucr_offense_description") or
            row.get("crime_category") or
            row.get("incident_category") or
            row.get("incident_description") or
            "Unknown"
        )
        if not isinstance(raw_type, str) or not raw_type.strip():
            raw_type = "Unknown"

        raw_type_upper = raw_type.upper()
        if any(x in raw_type_upper for x in ["THEFT", "BURGLARY", "LARCENY", "ROBBERY"]):
            norm_type = "Theft & Burglary"
        elif any(x in raw_type_upper for x in ["ASSAULT", "BATTERY", "HOMICIDE", "MURDER"]):
            norm_type = "Violent Crime"
        elif any(x in raw_type_upper for x in ["VEHICLE", "AUTO", "MOTOR"]):
            norm_type = "Vehicle Incident"
        elif any(x in raw_type_upper for x in ["DRUG", "NARCOTIC"]):
            norm_type = "Drug Offense"
        elif any(x in raw_type_upper for x in ["FRAUD", "DECEPTIVE", "FORGERY"]):
            norm_type = "Fraud & Deception"
        elif "TRESPASS" in raw_type_upper:
            norm_type = "Trespassing"
        elif any(x in raw_type_upper for x in ["WEAPON", "FIREARM"]):
            norm_type = "Weapons Offense"
        elif any(x in raw_type_upper for x in ["DAMAGE", "MISCHIEF", "VANDALISM"]):
            norm_type = "Property Damage/Vandalism"
        elif "UNKNOWN" not in raw_type_upper:
            norm_type = raw_type.title()
        else:
            norm_type = "Other/Unknown"

        incidents.append({
            "type": norm_type,
            "description": row.get("description") or row.get("law_cat_cd") or "",
            "date": row.get("date") or row.get("cmplnt_fr_dt") or "",
            "location": row.get("location_description") or row.get("boro_nm") or "",
            "block": row.get("block") or "",
            "lat": rlat or lat,
            "lon": rlon or lon,
        })
    return incidents
Enter fullscreen mode Exit fullscreen mode

The orchestrator.py file handles the agents:

  • First, the Geocode agent handles the user input.
  • After that, 5 agents run in parallel : the Weather agent, the NWS Alerts Agent, the Crime Agent, the 311 Infra Agent and the Seismic agent.
  • After all the agents’ tasks are complete, a risk score for the city is calculated.
  • Finally, the AI Briefing (Gemini) agent aggregates all data and provides insights to the user. 
def build_pipeline(address: str, fetcher, analyzer, board: ProgressBoard) -> Tuple[BaseAgent, Callable[[Dict[str, Any]], dict]]:

    def geo_of(deps):
        return deps["geocode"]

    def city_key_of(geo) -> str:
        return fetcher._detect_city(geo.get("city", ""), geo.get("state", ""))

    geocode = NodeAgent(
        name="geocode",
        body=lambda d: fetcher._geocode(address),
        board=board,
    )
    weather = NodeAgent(
        name="weather",
        body=lambda d: fetcher._fetch_weather(geo_of(d)["lat"], geo_of(d)["lon"]),
        node_deps=("geocode",),
        on_error=lambda d: fetcher._weather_stub(geo_of(d)["lat"], geo_of(d)["lon"]),
        board=board,
    )
    nws = NodeAgent(
        name="nws",
        body=lambda d: fetcher._fetch_nws_alerts(geo_of(d)["lat"], geo_of(d)["lon"]),
        node_deps=("geocode",),
        on_error=lambda d: [],
        board=board,
    )
    crime = NodeAgent(
        name="crime",
        body=lambda d: fetcher._fetch_crime(
            geo_of(d)["lat"], geo_of(d)["lon"], city_key_of(geo_of(d)),
            geo_of(d).get("state", ""), geo_of(d),
        ),
        node_deps=("geocode",),
        on_error=lambda d: {"incidents": [], "total_count": 0, "type_counts": {},
                            "fbi_stats": {}, "fbi_summary": {}, "period_days": 30},
        board=board,
    )
    infra = NodeAgent(
        name="infra",
        body=lambda d: fetcher._fetch_311(geo_of(d)["lat"], geo_of(d)["lon"], city_key_of(geo_of(d))),
        node_deps=("geocode",),
        on_error=lambda d: {"complaints": [], "total": 0, "categories": {}},
        board=board,
    )
    earthquakes = NodeAgent(
        name="earthquakes",
        body=lambda d: fetcher._fetch_earthquakes(geo_of(d)["lat"], geo_of(d)["lon"]),
        node_deps=("geocode",),
        on_error=lambda d: [],
        board=board,
    )
    risk = NodeAgent(
        name="risk",
        body=lambda d: fetcher._compute_risk(
            d["crime"], _merge_alerts(d["weather"], d["nws"]),
            d["infra"], d["earthquakes"], city_key_of(d["geocode"]),
        ),
        node_deps=("geocode", "weather", "nws", "crime", "infra", "earthquakes"),
        on_error=lambda d: {"overall": 0, "overall_label": "Low"},
        board=board,
    )
    ai_briefing = NodeAgent(
        name="ai_briefing",
        body=lambda d: analyzer.analyze(_core_data(d), address),
        node_deps=("geocode", "weather", "nws", "crime", "infra", "earthquakes", "risk"),
        on_error=lambda d: "AI analysis unavailable.",
        board=board,
    )

    nodes = [geocode, weather, nws, crime, infra, earthquakes, risk, ai_briefing]
    for node in nodes:
        board.init_row(node.name, node.node_deps)

    sources = ParallelAgent(name="sources", sub_agents=[weather, nws, crime, infra, earthquakes])
    root = SequentialAgent(name="pipeline", sub_agents=[geocode, sources, risk, ai_briefing])
    return root, assemble
Enter fullscreen mode Exit fullscreen mode

The Github repository for this solution is at:

GitHub - RubensZimbres/Neighborhood-Watch-App

RUNNING LOCALLY

.env file:

OPENWEATHER_API_KEY=your_openweathermap_key_here
FBI_CRIME_API_KEY=your_fbi_crime_api_key_here
SOCRATA_APP_TOKEN=your_socrata_app_token_here
GCP_PROJECT_NAME=your_gcp_project_id_here
GCP_LOCATION=us-central1
Enter fullscreen mode Exit fullscreen mode

How to run locally:

python3 -m venv venv
source venv/bin/activate

gcloud auth application-default login

streamlit run app.py
Enter fullscreen mode Exit fullscreen mode

DEPLOYMENT IN GOOGLE CLOUD 

export PROJECT_ID="your-gcp-project-id"
gcloud config set project "$PROJECT_ID"

gcloud services enable \
  run.googleapis.com \
  cloudbuild.googleapis.com \
  artifactregistry.googleapis.com \
  aiplatform.googleapis.com

gcloud run deploy neighborhoodalert \
  --source . \
  --region us-central1 \
  --allow-unauthenticated \
  --set-env-vars "OPENWEATHER_API_KEY=your_key" \
  --set-env-vars "FBI_CRIME_API_KEY=your_key" \
  --set-env-vars "SOCRATA_APP_TOKEN=your_token" \
  --set-env-vars "GCP_PROJECT_NAME=$PROJECT_ID" \
  --set-env-vars "GCP_LOCATION=us-central1" \
  --memory 512Mi \
  --cpu 1 \
  --timeout 300 \
  --max-instances 3

VAI="$(gcloud iam service-accounts list \
  --filter='displayName:Compute Engine' \
  --format='value(email)')"

gcloud projects add-iam-policy-binding "$PROJECT_ID" \
  --member "serviceAccount:$VAI" \
  --role "roles/aiplatform.user"
Enter fullscreen mode Exit fullscreen mode

Service URL: https://neighborhoodalert-xxxxx-uc.a.run.app

Acknowledgements

✨ Google ML Developer Programs and Google Developers Program supported this work by providing Google Cloud Credits✨

🔗https://developers.google.com/machinelearning 🔗

Top comments (0)