SYSTEM OVERVIEW

Project Summary

This repository implements Transit 2.0 — a scalable, agentic, graph-based supply chain resilience platform. The system transforms a legacy linear file-based pipeline into a production-grade architecture featuring real-time data ingestion (GDACS RSS), distributed task queues (Celery/Redis), a Neo4j graph database for supply chain network modeling, a LangGraph agentic loop for intelligent decision-making, and Streamlit dashboards with pydeck supply chain graph visualization.

High-Level Architecture

                    ┌────────────────────────────────────────────┐
                    │          LIVE INGESTION ENGINE             │
                    │  GDACS RSS → Redis Queue → Celery Tasks   │
                    │              (10 min polling)              │
                    └──────────────────┬─────────────────────────┘
                                       │
                                       ▼
                    ┌────────────────────────────────────────────┐
                    │         LANGGRAPH AGENTIC LOOP             │
                    │  ┌────────┐   ┌──────────┐   ┌─────────┐  │
                    │  │ Scout  │──▶│ Analyst  │──▶│Simulator│  │
                    │  │ (LLM)  │   │ (Risk)   │   │(Neo4j+  │  │
                    │  └────────┘   └──────────┘   │ OSRM)   │  │
                    │       │           │          └─────────┘  │
                    │       ▼           ▼                       │
                    │  LOW→END   MEDIUM→SIM   HIGH/CRIT→ALL    │
                    └──────────────────────────────────────────┘
                                       │
                                       ▼
        ┌──────────────────────────────┴──────────────────────────┐
        │                       DATA LAYER                       │
        │  ┌────────────┐   ┌──────────────┐   ┌──────────────┐  │
        │  │   Redis    │   │    Neo4j     │   │   Shared     │  │
        │  │  (Queue)   │   │  (Graph DB)  │   │  Exchange    │  │
        │  └────────────┘   └──────────────┘   │  (JSON)      │  │
        │                                      └──────────────┘  │
        └─────────────────────────────────────────────────────────┘
                                       │
                                       ▼
        ┌─────────────────────────────────────────────────────────┐
        │                  STREAMLIT DASHBOARDS                   │
        │  ┌────────────────────────┐  ┌──────────────────────┐  │
        │  │   Main Dashboard       │  │ Neo4j Graph Viz      │  │
        │  │   (Port 8501)          │  │ (Port 8502)          │  │
        │  └────────────────────────┘  └──────────────────────┘  │
        └─────────────────────────────────────────────────────────┘

Key Modules and Responsibilities

01_scout_module

chaos_trigger.py
- Generates synthetic disruption events such as flash floods, port strikes, road accidents, cyclones, and bridge closures.
- Outputs shared_exchange/signal.json.
scout_agent.py
- Loads the raw event signal.
- Uses LLM inference via Groq (or a fallback mock) to convert the raw signal into a normalized perception object.
- Normalized scout output keys include description, location, severity, disruption_type, agent_thoughts, and raw_signal.
dashboard.py
- Main Streamlit control room (port 8501).
- Integrates the LangGraph agentic pipeline: one-click chaos injection triggers run_pipeline() which runs Scout→Analyst→Simulator with conditional routing.
- Features: live disruption alert card, pydeck supply chain map with multi-modal routing, agent intelligence feed, live risk dashboard with ROI/savings metrics, Telegram integration sidebar, demo sequence animation.

02_analyst_module

analyst_agent.py
- Reads shared_exchange/scout_output.json and the shipment database shipments.json.
- Filters shipments affected by the disruption based on route logic.
- Computes risk metrics, total exposure, and a risk level.
- Attempts a Groq LLM call for narrative analysis; otherwise falls back to rule-based JSON.
- Writes shared_exchange/analyst_output.json.
shipments.json
- Contains 15 sample shipments with fields like id, value, origin, destination, route, cargo, weight_kg, eta_days, and priority.

03_intel_module

simulator_agent.py
- Uses OSRM HTTP route queries to estimate route distance and duration.
- Caches OSRM responses for 300 seconds.
- Runs Monte Carlo simulations for modes: Road_bypass, Rail, and Air.
- Queries Neo4j graph database for downstream domino effects via Cypher (query_neo4j_downstream()).
- Adds carbon scoring and downstream domino-effect damage estimates.
- Calls a Groq-based simulation oracle to recommend a transport mode and route.
strategist_agent.py
- Loads historical events from past_events.json.
- Uses a Groq LLM prompt to match current disruptions to past events and produce a bias_factor and strategic lesson.
capacity_matcher.py
- Matches unused cargo space against local vendor demand in shared_exchange/local_vendors.json.
- Generates a monetized cargo manifest at shared_exchange/open_cargo_manifest.json.
intel_coordinator.py
- Legacy pipeline orchestrator (file-watch based). Superseded by LangGraph for new flows.

04_manager_module

manager_agent.py
- Telegram alert dispatch, driver confirmation polling, voice alerts (gTTS).
- Reads intel and analyst outputs, calculates ROI, generates audio briefings.
- Writes shared_exchange/final_results.json.

Root-Level Transit 2.0 Modules

live_ingestion.py
- Polls the GDACS RSS feed every 10 minutes for India-relevant natural disasters.
- Falls back to mock events when no live GDACS events are detected.
- Pushes events to Redis queue (transit.disruptions) and publishes to Redis pub/sub (transit.events).
- Defines a Celery task process_disruption() that chains Scout→Analyst.
graph_builder.py
- Reads historical_shipments.csv (500 rows).
- Creates 25 City nodes and 500 SHIPMENT_TO relationships in Neo4j with aggregated properties.
- Provides query_downstream() and query_downstream_details() Cypher query functions.
langgraph_workflow.py
- Defines DisruptionState TypedDict for shared agent memory.
- Two LangChain @tool functions: query_neo4j(city, cargo_type, max_hops) and query_osrm(origin, destination, waypoint).
- Builds a LangGraph StateGraph with Scout→Analyst→Simulator nodes.
- Conditional routing: HIGH/CRITICAL severity routes to Analyst+Simulator, MEDIUM to Simulator only, LOW ends the pipeline.
- Entry point: run_pipeline(raw_event) returns full DisruptionState.
neo4j_viz.py
- Streamlit app (port 8502) for visualizing the Neo4j supply chain graph.
- Uses pydeck with PathLayer (shipment routes), ScatterplotLayer (city nodes), and TextLayer (city labels).
- City selector dropdown, max hops slider, Redis live disruption feed integration.

Data Flow

A. Real-Time Ingestion Path (Live)

live_ingestion.py polls GDACS RSS every 10 minutes.
Parsed events are pushed to Redis queue (transit.disruptions) and published to Redis pub/sub (transit.events).
Celery worker (--pool=solo on Windows) picks up process_disruption tasks.
Each task runs run_scout_agent() then run_analyst_agent(), writing results to shared_exchange/.

B. LangGraph Agentic Path (Dashboard)

User clicks INJECT CHAOS EVENT in the dashboard (or trigger_chaos() generates signal.json).
run_pipeline(signal) executes the LangGraph state graph:
- Scout Node: calls run_scout_agent() → returns severity, location, description.
- Router: if HIGH/CRITICAL → Analyst; MEDIUM → Simulator only; LOW → end.
- Analyst Node: writes scout output to disk, calls run_analyst_agent() → returns affected shipments, value at risk.
- Simulator Node: calls run_simulator() with Neo4j downstream queries and OSRM routing.
Results are written to shared_exchange/intel_output.json and shared_exchange/final_results.json.
Streamlit UI reads these files and renders the disruption alert card, agent feed, metrics, and map.

C. Neo4j Graph Query Path

graph_builder.py populates Neo4j from historical_shipments.csv.
simulator_agent.py / LangGraph tool calls query_neo4j(city, cargo_type, max_hops).
Cypher variable-length path query finds downstream cities within N hops.
Results feed into domino effect damage calculations.

Shared Data Contracts

`shared_exchange/signal.json`

Raw chaos event input (from chaos_trigger or live_ingestion).

`shared_exchange/scout_output.json`

Structured perception from the Scout (location, severity, description).

`shared_exchange/analyst_output.json`

Risk analytics, affected shipments, total value at risk, recommended action.

`shared_exchange/intel_output.json`

Simulation-driven route recommendation, confidence score, alternative modes.

`shared_exchange/final_results.json`

Final manager decision: recommended route, ROI/savings, executive summary.

`shared_exchange/open_cargo_manifest.json`

Market matching of surplus capacity to local vendors.

Redis Channels

transit.disruptions — list/queue for Celery task consumption.
transit.events — pub/sub channel for real-time dashboard updates.

Key Algorithms and Logic

1. Affected Shipment Filtering

02_analyst_module/analyst_agent.py matches disruption location and route logic to flag shipments.

Kochi events focus on NH-66 shipments.
Routes are matched using ROUTE_TO_CITIES and disruption location checks.
Risk scores combine priority, severity, and cargo sensitivity.

2. Neo4j Downstream Query

graph_builder.py Cypher query:

MATCH (origin:City {name: $city})
MATCH path = (origin)-[:SHIPMENT_TO*1..{max_hops}]->(downstream:City)
WHERE downstream.name <> $city
RETURN downstream.name AS city, ...

Used by simulator_agent.py and the LangGraph query_neo4j tool to find supply chain cascade effects.

3. Monte Carlo Simulation

03_intel_module/simulator_agent.py simulates transport modes:

Air: fast, low variance
Rail: medium time, low variance
Road_bypass: uses OSRM live route data when available

Returns mean time, std deviation, on-time probability, reliability score, carbon impact, cascading damage.

4. LangGraph Conditional Routing

HIGH/CRITICAL severity → Scout → Analyst → Simulator (full pipeline).
MEDIUM severity → Scout → Simulator only (skip analyst).
LOW severity → Scout → END (no further processing).

5. Domino Effect Damage Modeling

Downstream damage per cargo type using DOWNSTREAM_NETWORK. Delayed electronics can halt Bangalore assembly lines (₹15,000/hour). Neo4j graph queries extend this with real supply chain topology.

6. Dynamic Load Pooling

The capacity matcher sells unused truck space to vendors along route waypoints via greedy allocation.

Tech Stack

Technology	Version	Purpose
Python	3.11	Core language
Celery	5.3.6	Distributed task queue
Redis	(Docker)	Message broker & pub/sub
Neo4j	(Docker)	Graph database
LangGraph	1.0.10	Agentic workflow graph
LangChain Groq	1.1.2	LLM tool binding
Groq API	Llama 3	LLM inference
OSRM	Public API	Live road routing
Streamlit	Latest	Dashboards
pydeck	Latest	Deck.gl graph visualization
NumPy	-	Monte Carlo statistics
gTTS	-	Text-to-speech alerts
Docker	-	Containerization
Kubernetes	-	Deployment (k8s/)

Deployment and Run Instructions

Prerequisites

Python 3.11+
Docker Desktop (for Redis and Neo4j)
Groq API Key (free at groq.com)

1. Docker Services

# Redis (message broker)
docker run -d --name transit-redis -p 6379:6379 redis:7

# Neo4j (graph database)
docker run -d --name transit-neo4j -p 7687:7687 -p 7474:7474 \
  -e NEO4J_AUTH=neo4j/transit123 neo4j:5

2. Environment Setup

pip install -r requirements.txt
echo "GROQ_API_KEY=your_key_here" > .env

3. Populate Neo4j

python graph_builder.py

4. Start Celery Worker (Windows)

python -m celery -A live_ingestion.celery_app worker --pool=solo --loglevel=info

5. Start Live Ingestion

python live_ingestion.py

6. Start Dashboards

python -m streamlit run 01_scout_module/dashboard.py --server.port 8501 --server.headless true
python -m streamlit run neo4j_viz.py --server.port 8502 --server.headless true

7. Run LangGraph Pipeline (Standalone)

python langgraph_workflow.py

Pipeline Check

debug_pipeline.py runs Scout and Analyst stages and validates shared output files.

Notes and Observations

The legacy file-based pipeline (through shared_exchange/) is superseded by the LangGraph agentic loop for new flows.
Groq LLM calls are used in the Scout, Analyst, Strategist, and Simulator agents.
Celery 5.3.6 is required (5.6.3 has a fast_trace_task bug on Windows). Use --pool=solo on Windows.
Neo4j variable-length path patterns (*1..{max_hops}) must use f-strings (Cypher doesn't parameterize path bounds).
The dashboard at 01_scout_module/dashboard.py integrates the LangGraph pipeline via run_pipeline().
routes_dict.py and generate_routes.py define static rail route geometry for route modeling.

Important Files

Core Transit 2.0 (New)

live_ingestion.py — GDACS polling, Redis queue, Celery dispatch
graph_builder.py — CSV → Neo4j graph population + Cypher queries
langgraph_workflow.py — LangGraph agentic loop (Scout→Analyst→Simulator)
neo4j_viz.py — Neo4j supply chain graph Streamlit visualization
historical_shipments.csv — 500 shipment records for Neo4j

Legacy Modules (Maintained)

01_scout_module/chaos_trigger.py
01_scout_module/scout_agent.py
01_scout_module/dashboard.py
02_analyst_module/analyst_agent.py
02_analyst_module/shipments.json
03_intel_module/strategist_agent.py
03_intel_module/simulator_agent.py
03_intel_module/capacity_matcher.py
04_manager_module/manager_agent.py

Supporting

routes_dict.py / generate_routes.py — Rail route geometry
debug_pipeline.py — Pipeline validation
Dockerfile — Container build
k8s/ — Kubernetes deployment configs

This document captures the Transit 2.0 system's data flow, component roles, and end-to-end technology stack for the current codebase.

FilesExpand file tree

SYSTEM_OVERVIEW.md

Latest commit

History