🌾 Farm-AI Enterprise Platform

Low-Latency Swarm-ReAct Voice Platform & Databricks Lakehouse Data Engine

Farm-AI Enterprise is an enterprise-grade agricultural AI system that combines Swarm-ReAct Hybrid Agents, Real-Time Voice Streaming, Databricks PySpark Lakehouse Data Pipelines, and MLflow MLOps Model Registries to deliver sub-second intelligence to agricultural ecosystems.

---

🏗️ Centralized Master Planner Architecture

graph TD
    UserQuery((Farmer Request)) --> PlannerNode["KnowledgeSupportAgent (Master Planner)"]

    subgraph "1. Planning Phase (Centralized)"
        PlannerNode -->|Analyzes Query| PlanGen["Task Plan Generator"]
        PlanGen -->|Generates or Updates| TaskMD["session_state: task.md"]
        TaskMD -->|Step Dependency Classifier| Dispatcher{Execution Type?}
    end

    subgraph "2. Execution Phase (Swarm Workers)"
        Dispatcher -->|Parallel Execution| ParallelGroup["Parallel Workers"]
        ParallelGroup --> WeatherNode[Weather Agent]
        ParallelGroup --> MarketNode[Market Intelligence Agent]

        Dispatcher -->|Sequential Chain| ChainA[Farmer Profile Context]
        ChainA -->|Output to Input| ChainB[Plant Disease / Knowledge Agent]
    end

    subgraph "3. Synthesis & Completion Phase"
        WeatherNode --> SynthesizerNode["KnowledgeSupportAgent (Synthesizer)"]
        MarketNode --> SynthesizerNode
        ChainB --> SynthesizerNode
        
        SynthesizerNode -->|Marks Completed Tasks in task.md| Output[Final Response + Updated task.md]
    end

Loading

---

⚡ Core Technical Innovations

1. 🧠 Centralized Swarm-ReAct Master Planner (KnowledgeSupportAgent)

• Master Orchestrator: Replaces single-point supervisor routers with a centralized Master Planner that analyzes intent, generates dynamic session execution plans (task.md), and coordinates agent swarms.
• Dynamic Session Planning (task.md): Automatically breaks complex user queries into explicit parallel or sequential sub-tasks:
  • Parallel Tasks: Executes independent agents simultaneously (e.g. weather forecast + market_intelligence commodity rates).
  • Sequential Chains: Pass output context forward when Task B depends on Task A.

2. 🚀 Ultra-Low Latency Optimization Suite (<200ms TTFT)

• Zero-Hop Context Injection: Injects user profile & memory narratives directly into prompts, removing extra graph node turns.
• 5-Min RAM TTL Context Cache: Eliminates MongoDB read latency by caching active user context in memory (<1ms).
• 15-Min Semantic Query & 10-Min Tool Caching: Serves repeated vector search and web tool calls in <5ms.
• Streaming SSE Endpoint (/api/chat/stream): Delivers tokens continuously to user interfaces for a sub-200ms Time-To-First-Token (TTFT).

3. 🧱 Databricks PySpark Medallion Data Engineering Pipeline

• Bronze Layer (Raw Ingestion): Ingests raw mandi price CSVs, PDFs, and API payloads preserving immutable audit histories.
• Silver Layer (Cleaned & Standardized): PySpark jobs perform schema enforcement, deduplication, date normalization, and spatial geo-hashing.
• Gold Layer (Feature Store): Computes 30-day moving price averages, market volatility scores, and risk metrics synced to MongoDB.

4. 📈 Databricks MLflow Model Registry & MLOps

• Automated experiment tracking, metric logging (RMSE/MAE), and model registry deployment for Random Forest commodity price forecasters (FarmAI_Crop_Price_Forecaster).

5. 📡 Apache Kafka Real-Time Telemetry Streaming

• Consumes high-throughput IoT field sensor streams (soil moisture, temperature) using PySpark Structured Streaming with 10-minute event-time watermarking to handle late-arriving telemetry.

---

🛠️ Execution & Deployment Guide

# 1. Install Dependencies
uv sync
cd frontend && npm install && cd ..

# 2. Configure Environment Variables
cp backend/.env.example .env

# 3. Launch Enterprise Application Stack
./run_farm_ai.sh

🧪 Data Pipeline Execution Commands

# PySpark Medallion Lakehouse Pipeline
uv run python backend/spark_pipeline/medallion_mandi_etl.py

# Databricks MLflow Model Training & Registry
uv run python backend/spark_pipeline/mlflow_price_forecaster.py

# PySpark Kafka Structured Streaming Ingest
uv run python backend/spark_pipeline/kafka_streaming_ingest.py

---

📜 License

Distributed under the MIT License.