commit 0dcde6cc427784b12d37964149cc6cadcea53628 parent b3fc6028423798ebdd095b02fef3fb645e29728b Author: Abhinav Rai <69450646+AbhinavRai01@users.noreply.github.com> Date: Wed, 4 Mar 2026 06:02:40 +0000 Merge PR Diffstat:
24 files changed, 904 insertions(+), 237 deletions(-)
diff --git a/.gitignore b/.gitignore @@ -3,4 +3,6 @@ node_modules/ __pycache__/ /web/node_modules -Knowledge_Base -\ No newline at end of file +Knowledge_Base +/.venv +/agent/venv +\ No newline at end of file diff --git a/Qdrant/__pycache__/Client.cpython-311.pyc b/Qdrant/__pycache__/Client.cpython-311.pyc Binary files differ. diff --git a/Qdrant/__pycache__/Client.cpython-313.pyc b/Qdrant/__pycache__/Client.cpython-313.pyc Binary files differ. diff --git a/Qdrant/__pycache__/Setup.cpython-313.pyc b/Qdrant/__pycache__/Setup.cpython-313.pyc Binary files differ. diff --git a/Qdrant/__pycache__/Store.cpython-311.pyc b/Qdrant/__pycache__/Store.cpython-311.pyc Binary files differ. diff --git a/Qdrant/__pycache__/Store.cpython-313.pyc b/Qdrant/__pycache__/Store.cpython-313.pyc Binary files differ. diff --git a/Sentinel/Encoders/__pycache__/TimeSeries.cpython-311.pyc b/Sentinel/Encoders/__pycache__/TimeSeries.cpython-311.pyc Binary files differ. diff --git a/Sentinel/Encoders/__pycache__/TimeSeries.cpython-313.pyc b/Sentinel/Encoders/__pycache__/TimeSeries.cpython-313.pyc Binary files differ. diff --git a/Sentinel/Encoders/__pycache__/Vision.cpython-311.pyc b/Sentinel/Encoders/__pycache__/Vision.cpython-311.pyc Binary files differ. diff --git a/Sentinel/Encoders/__pycache__/Vision.cpython-313.pyc b/Sentinel/Encoders/__pycache__/Vision.cpython-313.pyc Binary files differ. diff --git a/Sentinel/__pycache__/agent.cpython-311.pyc b/Sentinel/__pycache__/agent.cpython-311.pyc Binary files differ. diff --git a/Sentinel/__pycache__/agent.cpython-313.pyc b/Sentinel/__pycache__/agent.cpython-313.pyc Binary files differ. diff --git a/Sentinel/__pycache__/fmu.cpython-311.pyc b/Sentinel/__pycache__/fmu.cpython-311.pyc Binary files differ. diff --git a/Sentinel/__pycache__/fmu.cpython-313.pyc b/Sentinel/__pycache__/fmu.cpython-313.pyc Binary files differ. diff --git a/agent/main_agent.py b/agent/main_agent.py @@ -2,27 +2,22 @@ import sys import os import requests import time -from pathlib import Path -# --- PATH SETUP --- -# Adds the current directory (agent/) to sys.path so we can import sub_agents and Qdrant current_dir = os.path.dirname(os.path.abspath(__file__)) sys.path.append(current_dir) -# Import Agents from sub_agents.fetching_agent import FetchingAgent -from sub_agents.judge_agent import JudgeAgent # 🆕 Import Judge +from sub_agents.judge_agent import JudgeAgent from sub_agents.atmospheric_agent import AtmosphericAgent from sub_agents.water_agent import WaterAgent from sub_agents.Researcher import ResearcherAgent from sub_agents.Supervisor import SupervisorAgent -# Simulator Action URL -# Updated to localhost to match the simulator we just created +# Update this URL to your running simulator instance SIMULATOR_ACTION_URL = "https://unexhumed-melaine-bouncingly.ngrok-free.dev/simulation/action" def main(): - print("🚀 Initializing Demeter Orchestrator (Judge-Review -> Supervisor-Store)...") + print("🚀 Initializing Demeter Orchestrator...") try: fetcher = FetchingAgent() @@ -30,7 +25,10 @@ def main(): researcher = ResearcherAgent() atmos_agent = AtmosphericAgent() water_agent = WaterAgent() - supervisor = SupervisorAgent() + # Pass researcher so Supervisor can share the RAG tools if needed + supervisor = SupervisorAgent(researcher_agent=researcher) + + print("✅ Agents Online.") except Exception as e: print(f"❌ Init Error: {e}") return @@ -40,47 +38,66 @@ def main(): print("⏱️ STARTING NEW CYCLE") print("="*50) - # 1. Fetch Reality (Seq N) + # 1. Fetch fmu, sensor_snapshot, history = fetcher.fetch_and_process() - if not fmu: - print("❌ Fetch failed. Retrying in 10s...") + print("⚠️ No FMU found. Waiting...") time.sleep(10) continue - # 2. Judge Reviews History (Updates Seq N-1) - # Does NOT store current FMU yet + # 2. Judge judge.review_previous_cycle(fmu) - # 3. Research & Reasoning + # 3. 🟢 GET BANDIT STRATEGY (The Brain) + # The Supervisor consults the Bandit first to set the cycle's goal + strat_name, strat_instr, action_idx = supervisor.get_strategic_goal(fmu) + print(f"\n🎰 BANDIT STRATEGY: {strat_name}") + print(f"📝 Instruction: {strat_instr}") + + # 4. Research crop = fmu.metadata.get("crop", "unknown") stage = fmu.metadata.get("stage", "unknown") query = f"optimal hydroponic conditions for {crop} in {stage} stage" - research_context = researcher.search(query) - print("\n🧠 Domain Agents Deliberating...") - atmos_plan = atmos_agent.reason(sensor_snapshot, research_context) - water_plan = water_agent.reason(sensor_snapshot, research_context) + # 5. 🟢 DELIBERATION (The Experts) + # We pass the strategy instruction and history to the LangGraph agents + print("\n🧠 Agents Planning...") + + # Updated call signature to match the new 'reason' method + atmos_plan = atmos_agent.reason( + sensors=sensor_snapshot, + research=research_context, + strategy=strat_instr, # Pass the instruction text (e.g. "LOWER pH...") + history=history # Pass history for context awareness + ) + + water_plan = water_agent.reason( + sensors=sensor_snapshot, + research=research_context, + strategy=strat_instr, + history=history + ) - # 4. Supervisor Decides & STORES Reality (Seq N) - # Now passes the full 'fmu' object so Supervisor can save it - print("\n👮 Supervisor Validating & Storing...") + # 6. Synthesis (The Supervisor) + # Supervisor merges plans, checks conflicts, and ensures safety + print("\n👮 Supervisor Finalizing...") final_action = supervisor.synthesize_plan( atmos_plan, water_plan, - fmu, # <--- Passing full FMU object - history + fmu, + history, + strategy_info=(strat_name, strat_instr, action_idx) ) print(f"🎯 FINAL COMMAND: {final_action}") - # 5. Execute + # 7. Execute try: requests.post(SIMULATOR_ACTION_URL, json=final_action) - print("✅ Action sent to Simulator.") + print("✅ Sent to Simulator.") except Exception as e: - print(f"❌ Connection error: {e}") + print(f"❌ Connection Error: {e}") print("\nzzz Sleeping 15s...") time.sleep(15) diff --git a/agent/sub_agents/Supervisor.py b/agent/sub_agents/Supervisor.py @@ -1,156 +1,234 @@ -# agent/sub_agents/Supervisor.py - +import os import json import numpy as np +from langchain_openai import ChatOpenAI +from langchain_core.messages import SystemMessage, HumanMessage +from langgraph.graph import StateGraph, END +from agent.tools.actuation import convert_targets_to_actions -# 👇 CORRECT IMPORTS based on your folder structure from agent.Marl.bandit import ContextualBandit from agent.Marl.strategies import STRATEGIES, NUM_ACTIONS +from agent.Qdrant.Store import store_fmu +from agent.sub_agents.water_and_atmospheric_dependencies.physics_engine import predict_outcome + +# --- NEW TOOLS DEFINITION --- +def check_cross_domain_conflicts(atmos, water): + conflicts = [] + + # 1. Thermal Shock Check + air_t = atmos.get('air_temp', 25) + water_t = water.get('water_temp', 20) + if abs(air_t - water_t) > 10: + conflicts.append(f"CRITICAL: Thermal Shock Risk. Air ({air_t}C) and Water ({water_t}C) delta > 10C.") + + # 2. Transpiration vs Uptake Check + # High VPD (Dry) + High EC (Salty) = Burn Risk + rh = atmos.get('humidity', 60) + ec = water.get('ec', 1.0) + if rh < 50 and ec > 2.0: + conflicts.append(f"STRESS: Low Humidity ({rh}%) + High EC ({ec}) will cause Tip Burn.") + + return conflicts + +def validate_hard_limits(plan): + violations = [] + # Hard limits for Lettuce/General Hydroponics + if plan.get('ph', 6.0) < 5.0: violations.append("pH < 5.0 is toxic.") + if plan.get('ph', 6.0) > 7.5: violations.append("pH > 7.5 causes lockout.") + if plan.get('ec', 1.0) > 3.0: violations.append("EC > 3.0 is too high for lettuce.") + if plan.get('humidity', 60) > 85: violations.append("Humidity > 85% guarantees mold.") + + return violations + +# --- STATE DEFINITION --- +from typing import TypedDict, Optional, Dict, Any, List + +class SupervisorState(TypedDict): + # Inputs + atmos_plan: Dict[str, Any] + water_plan: Dict[str, Any] + strategy_advice: str # Kept as advice, not law + + # Processing + merged_plan: Dict[str, Any] + review_notes: List[str] + simulation_health: float + + # Output + final_decision: str # "APPROVE" or "REJECT" + critique: str # Feedback for sub-agents if Rejected + +API_KEY = os.environ.get("GROQ_API_KEY") class SupervisorAgent: - def __init__(self, researcher): - self.llm = researcher.llm - - # Initialize Bandit (15 actions, 515 dimensions) + def __init__(self, researcher_agent=None): + self.name = "Supervisor" + # Bandit is now just an 'Advisor', not an enforcer self.bandit = ContextualBandit(n_actions=NUM_ACTIONS, feature_dim=519) + + if API_KEY: + self.model = ChatOpenAI( + base_url="https://api.groq.com/openai/v1", + api_key=API_KEY, + model="llama-3.3-70b-versatile", + temperature=0.0 # Zero temp for strict judging + ) + + self.app = self._build_graph() - def _build_context(self, fmu_vector, sensors): - """ - Combines Visual Intuition (CLIP) with Explicit Sensors. - """ - # Ensure vector is numpy - vis_vec = np.array(fmu_vector) if isinstance(fmu_vector, list) else fmu_vector + def _build_graph(self): + workflow = StateGraph(SupervisorState) + + # 1. Merge: Combine the two JSONs + workflow.add_node("merge", self.node_merge) - # Normalize sensors roughly to 0-1 range - s_vec = np.array([ - (sensors.get('pH', 6.0) - 6.0) / 2.0, - sensors.get('EC', 1.0) / 3.0, - sensors.get('temp', 25.0) / 40.0 - ]) + # 2. Review: Run the 3 Tools (Conflicts, Limits, Physics) + workflow.add_node("review", self.node_review) + + # 3. Judge: LLM decides if the issues are fatal + workflow.add_node("judge", self.node_judge) + + # Flow + workflow.set_entry_point("merge") + workflow.add_edge("merge", "review") + workflow.add_edge("review", "judge") + workflow.add_edge("judge", END) - return np.concatenate([vis_vec, s_vec]) + return workflow.compile() - def _get_strategy_instruction(self, strategy_name): + # --- NODE FUNCTIONS --- + + def node_merge(self, state): + print(" 🔗 Supervisor Merging Plans...") + # Simple dictionary merge + merged = {**state['atmos_plan'], **state['water_plan']} + return {"merged_plan": merged} + + def node_review(self, state): + print(" 🔍 Supervisor Running Unit Tests...") + plan = state['merged_plan'] + notes = [] + + # Tool 1: Conflict Check + conflicts = check_cross_domain_conflicts(state['atmos_plan'], state['water_plan']) + if conflicts: + notes.extend(conflicts) + + # Tool 2: Limit Check + limits = validate_hard_limits(plan) + if limits: + notes.extend(limits) + + # Tool 3: Physics Simulator + # (We reuse your existing prediction engine) + sim_result = predict_outcome(plan, plan) # Comparing plan vs itself as a snapshot for now + health = sim_result.get('predicted_health', 100) + + if health < 90: + notes.append(f"SIMULATION FAIL: Predicted health drops to {health}%. Risk: {sim_result.get('risk_warning')}") + + return {"review_notes": notes, "simulation_health": health} + + def node_judge(self, state): """ - Translates mathematical intent into LLM instructions. + The LLM looks at the automated test results and makes the final call. """ - instructions = { - "MAINTAIN_CURRENT": "Do NOT recommend changes. System is stable.", - "CALIBRATE_SENSORS": "Sensor readings are anomalous. Recommend hardware calibration.", - "AGGRESSIVE_PH_DOWN": "Priority: LOWER pH rapidly. Recommend strong acid buffers.", - "AGGRESSIVE_PH_UP": "Priority: RAISE pH rapidly. Recommend strong base buffers.", - "GENTLE_PH_BALANCING": "pH is drifting. Recommend gentle adjustments only.", - "INCREASE_EC_VEG": "Plant needs NITROGEN for vegetative growth.", - "INCREASE_EC_BLOOM": "Plant needs PHOSPHORUS/POTASSIUM for flowering.", - "LOWER_EC_FLUSH": "Nutrient burn detected. Recommend flushing reservoir.", - "CALMAG_BOOST": "Deficiency detected. Recommend Calcium/Magnesium supplement.", - "RAISE_TEMP_HUMIDITY": "Environment too cold/dry. Recommend heating/humidifying.", - "LOWER_TEMP_HUMIDITY": "Mold risk high. Recommend fans and dehumidifiers.", - "MAX_AIR_CIRCULATION": "Stagnant air. Recommend max fan speed.", - "FUNGAL_TREATMENT": "Fungal risk. Recommend fungicide and lower humidity.", - "PEST_ISOLATION": "Pests detected. Recommend isolation and organic pesticide.", - "PRUNE_NECROTIC_LEAVES": "Necrosis detected. Recommend pruning dead matter." - } - return instructions.get(strategy_name, "Follow standard procedures.") - - def reason(self, current_fmu, similar_fmus, sub_agent_outputs): - sensors = current_fmu['payload']['sensors'] - fmu_vector = current_fmu.get('vector') - - if fmu_vector is None: - fmu_vector = np.zeros(512) - - # 1. 🟢 GET CONTEXT - context_vector = self._build_context(fmu_vector, sensors) - - # 2. 🟢 BANDIT DECISION (The "Will") - action_idx, debug_info = self.bandit.select_action(context_vector) - strategic_intent = STRATEGIES[action_idx] - specific_order = self._get_strategy_instruction(strategic_intent) - - # 3. 🟢 IDENTIFY RELEVANT SPECIALIST (The "Physics") - if "NUTRIENT" in strategic_intent or "PH" in strategic_intent or "EC" in strategic_intent: - highlighted_report = sub_agent_outputs.get("nutrient_report", "No Report") - focus_area = "NUTRIENT SPECIALIST" - elif "TEMP" in strategic_intent or "HUMIDITY" in strategic_intent or "AIR" in strategic_intent: - highlighted_report = sub_agent_outputs.get("atmosphere_report", "No Report") - focus_area = "ATMOSPHERE SPECIALIST" - elif "PEST" in strategic_intent or "FUNGAL" in strategic_intent: - highlighted_report = "Visual analysis indicates bio-threats." - focus_area = "BIO-SECURITY" - else: - highlighted_report = "Standard operational check." - focus_area = "ALL SECTORS" - - # 4. 🟢 FORMAT HISTORY (The "Precedent") <--- NEW SECTION - history_context = "No relevant historical cases found." - if similar_fmus and len(similar_fmus) > 0: - history_lines = [] - for i, fmu in enumerate(similar_fmus): - # Extract key details from the past record - past_action = fmu['payload'].get('action_taken', 'Unknown') - past_outcome = fmu['payload'].get('outcome', 'Unknown') - score = fmu.get('score', 0.0) - history_lines.append(f"- Case #{i+1} (Match: {score:.1%}): Action '{past_action}' -> Result: '{past_outcome}'") - history_context = "\n".join(history_lines) - - # 5. 🟢 SYNTHESIS PROMPT - system_prompt = f""" - You are the Supervisor of a Hydroponic Farm. - - --- 🚨 CHAIN OF COMMAND INSTRUCTIONS 🚨 --- - - 1. STRATEGIC GOAL (From RL General): - "{strategic_intent}" -> "{specific_order}" - *This is your MANDATORY objective.* - - 2. INTELLIGENCE REPORT (From {focus_area}): - "{highlighted_report}" - *Use these specific calculations (VPD, Lockout, etc.) to justify your plan.* - - 3. HISTORICAL PRECEDENT (Retrieval Memory): - {history_context} - *Reference these past cases to support (or warn against) specific implementation details.* - - 4. FULL CONTEXT: - Other Reports: {json.dumps(sub_agent_outputs)} - - --- YOUR TASK --- - Generate a specific action plan that executes the Strategic Goal. - - CRITICAL: You must synthesize the RL Order, the Physics Report, and the History. - - If History shows the RL Strategy failed recently, mention that risk and propose a safer variation. - - If History confirms success, cite it to build confidence. - - RESPONSE FORMAT (JSON): - {{ - "decision": "Brief summary", - "reasoning": "Detailed synthesis of Logic + Physics + History...", - "risk_matrix": {{ "nutrients": 5, "climate": 5, "visuals": 5, "history": 5 }} - }} + print(" ⚖️ Supervisor Judging...") + + if not state['review_notes']: + # No issues found by tools + return {"final_decision": "APPROVE", "critique": "Plan looks solid."} + + # If issues exist, ask LLM if they are fatal or acceptable trade-offs + prompt = f""" + You are the Quality Assurance Supervisor. + + PROPOSED PLAN: {state['merged_plan']} + + AUTOMATED TEST FAILURES: + {json.dumps(state['review_notes'], indent=2)} + + ADVISORY STRATEGY: {state['strategy_advice']} + + TASK: + 1. If the failures are dangerous (Toxic pH, Thermal Shock, Low Health), REJECT the plan. + 2. If the failures are minor or necessary for the Strategy (e.g., Low Humidity required for 'Fungal Treatment'), APPROVE it. + + OUTPUT JSON: {{ "verdict": "APPROVE" or "REJECT", "critique": "Explanation..." }} """ - + try: - response = self.llm.chat.completions.create( - model="llama-3.1-8b-instant", - messages=[ - {"role": "system", "content": system_prompt}, - {"role": "user", "content": f"Sub-Agent Reports: {json.dumps(sub_agent_outputs)}"} - ], - response_format={"type": "json_object"} - ) - decision_json = json.loads(response.choices[0].message.content) + response = self.model.invoke([HumanMessage(content=prompt)]) + content = response.content.replace("```json", "").replace("```", "").strip() + result = json.loads(content) - # 🟢 Attach Metadata for Training - decision_json["strategic_intent"] = strategic_intent - decision_json["bandit_action_idx"] = int(action_idx) - - return decision_json - - except Exception as e: return { - "decision": "Error in reasoning", - "reasoning": str(e), - "strategic_intent": strategic_intent, - "bandit_action_idx": int(action_idx) - } -\ No newline at end of file + "final_decision": result.get("verdict", "REJECT"), + "critique": result.get("critique", "Automated tests failed.") + } + except: + # Default to reject if unsafe + return {"final_decision": "REJECT", "critique": "Plan failed automated safety checks."} + + # --- ENTRY POINT --- + + def synthesize_plan(self, atmos_plan, water_plan, fmu, history, strategy_info): + strategy_name, _, action_idx = strategy_info + + initial_state = { + "atmos_plan": atmos_plan, + "water_plan": water_plan, + "strategy_advice": strategy_name, + "merged_plan": {}, + "review_notes": [], + "simulation_health": 0.0, + "final_decision": "", + "critique": "" + } + + result = self.app.invoke(initial_state) + final_targets = result.get("merged_plan", {}) + + # 🟢 NEW STEP: CONVERT TARGETS TO PHYSICAL ACTIONS + current_sensors = fmu.metadata.get('sensor_data', {}) + + print(f"[{self.name}] ⚙️ Converting Targets to Actuator Commands...") + + # Calculate physical actions + physical_action_obj = convert_targets_to_actions(current_sensors, final_targets) + + # Convert Pydantic model to Dict for JSON serialization + final_payload = physical_action_obj.dict() + + # Log it + print(f"[{self.name}] 🚜 Activating Hardware: {final_payload}") + + # Store in FMU + fmu.metadata["action_taken"] = str(final_payload) + fmu.metadata["bandit_action_id"] = action_idx + fmu.metadata["strategic_intent"] = strategy_name + + if "image_b64" in fmu.metadata: del fmu.metadata["image_b64"] + store_fmu(fmu) + + return final_payload + + # --- ADVISORY ONLY (Not Enforced) --- + def get_strategic_goal(self, fmu): + # (Same as before, but treated as advice now) + sensors = fmu.metadata.get('sensor_data', {}) + fmu_vector = fmu.vector + vis_vec = np.array(fmu_vector) if isinstance(fmu_vector, list) else fmu_vector + if vis_vec is None or len(vis_vec) == 0: vis_vec = np.zeros(516) + + s_vec = np.array([ + (float(sensors.get('pH', 6.0)) - 6.0) / 2.0, + float(sensors.get('EC', 1.0)) / 3.0, + float(sensors.get('temp', 25.0)) / 40.0 + ]) + context_vector = np.concatenate([vis_vec, s_vec]) + + action_idx, _ = self.bandit.select_action(context_vector) + strategy_name = STRATEGIES[action_idx] + + return strategy_name, "Advisory Only", int(action_idx) +\ No newline at end of file diff --git a/agent/sub_agents/atmospheric_agent.py b/agent/sub_agents/atmospheric_agent.py @@ -1,49 +1,134 @@ import os -from openai import OpenAI +from langchain_openai import ChatOpenAI +from langgraph.graph import StateGraph, END + +# Graph State & Nodes +from agent.sub_agents.water_and_atmospheric_dependencies.state import AgentState +from agent.sub_agents.water_and_atmospheric_dependencies.nodes import decide_node, simulate_node, finalize_node, execute_tools_node + +# Tools +from agent.sub_agents.water_and_atmospheric_dependencies.retrieval import ask_historian, ask_rag +from agent.sub_agents.water_and_atmospheric_dependencies.tools import calculate_vpd, web_search # Configuration -MODEL_ID = "openai/gpt-oss-120b" API_KEY = os.environ.get("GROQ_API_KEY") +MODEL_ID = "llama-3.3-70b-versatile" + +ATMOS_PROMPT = """ +You are the Atmospheric Specialist for a Hydroponic Farm. +Your goal is to optimize VAPOR PRESSURE DEFICIT (VPD) and PHOTOSYNTHESIS. + +--- RULES --- +1. Target VPD: 0.8 - 1.2 kPa (Vegetative), 1.2 - 1.6 kPa (Flowering). +2. Humidity > 80% is dangerous (Mold Risk). +3. CO2 > 1500ppm is wasteful unless light is maxed out. + +--- CURRENT CONTEXT --- +Sensors: {sensors} +Strategy: {strategy} +Research: {research} +History: {history} +Critique from Simulation: {critique} + +TASK: Output a JSON dict with keys: 'air_temp' (C), 'humidity' (%), 'co2' (ppm), 'light_intensity' (umol). +""" + class AtmosphericAgent: def __init__(self): self.name = "Atmospheric Agent" + + # 1. Initialize Model if not API_KEY: - print(f"[{self.name}] ⚠️ GROQ_API_KEY missing.") - self.client = None + print(f"[{self.name}] ⚠️ No API Key found.") + self.model = None else: - self.client = OpenAI(base_url="https://api.groq.com/openai/v1", api_key=os.getenv("GROQ_API_KEY")) - - def reason(self, current_state: dict, research_context: str) -> dict: - """ - Decides on CO2, Light, Temp, Humidity changes. - Returns a dict of actions (e.g., {"CO2": 1200, "Light": 800}) - """ - print(f"[{self.name}] 🌤️ Analyzing Air conditions...") - - prompt = ( - f"You are the Atmospheric Control System.\n" - f"RESEARCH GUIDELINES:\n{research_context}\n\n" - f"CURRENT STATE:\n{current_state}\n\n" - f"TASK: Output a JSON dictionary ONLY of target values for 'co2' (ppm), 'light_intensity' (umol), " - f"'air_temp' (C), and 'humidity' (%).\n" - f"Example format: {{'co2': 1000, 'light_intensity': 600, 'air_temp': 24, 'humidity': 60}}\n" - f"Do not add markdown formatting or explanation." + llm = ChatOpenAI( + base_url="https://api.groq.com/openai/v1", + api_key=API_KEY, + model="llama-3.3-70b-versatile", + temperature=0.2 + ) + + # 2. 🟢 BIND TOOLS (The "Arms") + # We bind the general research tools AND the specific math tool (VPD) + self.model_with_tools = llm.bind_tools([ + ask_historian, + ask_rag, + web_search, + calculate_vpd + ]) + + # 3. Build the Graph (The "Brain") + self.app = self._build_graph() + + def _build_graph(self): + workflow = StateGraph(AgentState) + + # --- A. ADD NODES --- + # 1. Decide: Uses the LLM with Tools bound to it + workflow.add_node("decide", lambda state: decide_node(state, self.model_with_tools, ATMOS_PROMPT)) + + # 2. Tools: Executes the function if the LLM calls one + workflow.add_node("tools", execute_tools_node) + + # 3. Simulate: Checks physics/safety + workflow.add_node("simulate", simulate_node) + + # 4. Finalize: Formatting + workflow.add_node("finalize", finalize_node) + + # --- B. DEFINE FLOW --- + workflow.set_entry_point("decide") + + # Logic 1: Decide -> (Tools OR Simulate) + def check_decision_output(state): + # If the LLM decided to call a tool, go to tool execution + if state.get("next_step") == "tools": + return "tools" + # Otherwise, it wrote a plan, so go verify it + return "simulate" + + workflow.add_conditional_edges( + "decide", + check_decision_output, + {"tools": "tools", "simulate": "simulate"} ) + + # Logic 2: Tools -> Back to Decide (ReAct Loop) + workflow.add_edge("tools", "decide") - if not self.client: - return {"co2": 400, "light_intensity": 500} # Defaults + # Logic 3: Simulate -> (Finalize OR Retry) + def check_simulation_result(state): + if state["simulation_result"]["passed"]: + return "finalize" + elif state["retry_count"] > 3: + print(f"[{self.name}] ⚠️ Max retries reached. Forcing unsafe plan.") + return "finalize" + else: + # Loop back to fix the mistake + return "decide" - try: - response = self.client.chat.completions.create( - model=MODEL_ID, - messages=[{"role": "user", "content": prompt}] - ) - # Simple cleaning to ensure valid JSON - content = response.choices[0].message.content.replace("```json", "").replace("```", "").strip() - # In a real system, use json.loads(content) with error handling - import ast - return ast.literal_eval(content) - except Exception as e: - print(f"[{self.name}] Error: {e}") - return {} -\ No newline at end of file + workflow.add_conditional_edges( + "simulate", + check_simulation_result, + {"finalize": "finalize", "decide": "decide"} + ) + + workflow.add_edge("finalize", END) + return workflow.compile() + + def reason(self, sensors, research, strategy, history="None"): + """Entry point called by main_agent.py""" + initial_state = { + "sensors": sensors, + "research_context": research, + "strategy": strategy, + "history": history, + "retry_count": 0, + "critique": None, + "messages": [] # Stores conversation history for ReAct + } + + result = self.app.invoke(initial_state) + return result.get("final_action", {}) +\ No newline at end of file diff --git a/agent/sub_agents/water_agent.py b/agent/sub_agents/water_agent.py @@ -1,43 +1,129 @@ import os -from openai import OpenAI +from langchain_openai import ChatOpenAI +from langgraph.graph import StateGraph, END -MODEL_ID = "openai/gpt-oss-120b" +# Graph State & Nodes +from agent.sub_agents.water_and_atmospheric_dependencies.state import AgentState +from agent.sub_agents.water_and_atmospheric_dependencies.nodes import decide_node, simulate_node, finalize_node, execute_tools_node + +# Tools +from agent.sub_agents.water_and_atmospheric_dependencies.retrieval import ask_historian, ask_rag +from agent.sub_agents.water_and_atmospheric_dependencies.tools import check_ph_safety, web_search + +# Configuration API_KEY = os.environ.get("GROQ_API_KEY") +WATER_PROMPT = """ +You are the Water & Nutrient Specialist for a Hydroponic Farm. +Your goal is to maintain HOMEOSTASIS in the root zone. + +--- RULES --- +1. pH is Logarithmic. Never swing more than 0.5 in one cycle. +2. If Strategy is 'Flush', EC must drop to < 0.2 dS/m. +3. If Water Temp > 24C, you MUST recommend cooling or beneficial bacteria to prevent root rot. + +--- CURRENT CONTEXT --- +Sensors: {sensors} +Strategy: {strategy} +Research: {research} +History: {history} +Critique from Simulation: {critique} + +TASK: Output a JSON dict with keys: 'ph', 'ec' (dS/m), 'water_temp' (C). +""" + class WaterAgent: def __init__(self): self.name = "Water Agent" + + # 1. Initialize Model if not API_KEY: - self.client = None + print(f"[{self.name}] ⚠️ No API Key found.") + self.model = None else: - self.client = OpenAI(base_url="https://api.groq.com/openai/v1", api_key=os.getenv("GROQ_API_KEY")) - - def reason(self, current_state: dict, research_context: str) -> dict: - """ - Decides on pH, EC, Water Temp changes. - """ - print(f"[{self.name}] 💧 Analyzing Water solution...") - - prompt = ( - f"You are the Water Chemistry System.\n" - f"RESEARCH GUIDELINES:\n{research_context}\n\n" - f"CURRENT STATE:\n{current_state}\n\n" - f"TASK: Output a JSON dictionary ONLY of target values for 'ph', 'ec' (dS/m), and 'water_temp' (C).\n" - f"Example format: {{'ph': 5.8, 'ec': 1.5, 'water_temp': 20}}\n" - f"Do not add markdown formatting." + llm = ChatOpenAI( + base_url="https://api.groq.com/openai/v1", + api_key=API_KEY, + model="llama-3.3-70b-versatile", + temperature=0.2 + ) + + # 2. 🟢 BIND TOOLS (The "Arms") + # We bind the general research tools AND the specific math tool (pH Safety) + self.model_with_tools = llm.bind_tools([ + ask_historian, + ask_rag, + web_search, + check_ph_safety + ]) + + # 3. Build the Graph (The "Brain") + self.app = self._build_graph() + + def _build_graph(self): + workflow = StateGraph(AgentState) + + # --- A. ADD NODES --- + # 1. Decide: Uses the LLM with Tools bound to it + workflow.add_node("decide", lambda state: decide_node(state, self.model_with_tools, WATER_PROMPT)) + + # 2. Tools: Executes the function if the LLM calls one + workflow.add_node("tools", execute_tools_node) + + # 3. Simulate: Checks physics/safety + workflow.add_node("simulate", simulate_node) + + # 4. Finalize: Formatting + workflow.add_node("finalize", finalize_node) + + # --- B. DEFINE FLOW --- + workflow.set_entry_point("decide") + + # Logic 1: Decide -> (Tools OR Simulate) + def check_decision_output(state): + if state.get("next_step") == "tools": + return "tools" + return "simulate" + + workflow.add_conditional_edges( + "decide", + check_decision_output, + {"tools": "tools", "simulate": "simulate"} ) + + # Logic 2: Tools -> Back to Decide (ReAct Loop) + workflow.add_edge("tools", "decide") - if not self.client: - return {"ph": 6.0, "ec": 1.2} + # Logic 3: Simulate -> (Finalize OR Retry) + def check_simulation_result(state): + if state["simulation_result"]["passed"]: + return "finalize" + elif state["retry_count"] > 3: + print(f"[{self.name}] ⚠️ Max retries reached. Forcing unsafe plan.") + return "finalize" + else: + return "decide" - try: - response = self.client.chat.completions.create( - model=MODEL_ID, - messages=[{"role": "user", "content": prompt}] - ) - content = response.choices[0].message.content.replace("```json", "").replace("```", "").strip() - import ast - return ast.literal_eval(content) - except Exception as e: - print(f"[{self.name}] Error: {e}") - return {} -\ No newline at end of file + workflow.add_conditional_edges( + "simulate", + check_simulation_result, + {"finalize": "finalize", "decide": "decide"} + ) + + workflow.add_edge("finalize", END) + return workflow.compile() + + def reason(self, sensors, research, strategy, history="None"): + """Entry point called by main_agent.py""" + initial_state = { + "sensors": sensors, + "research_context": research, + "strategy": strategy, + "history": history, + "retry_count": 0, + "critique": None, + "messages": [] + } + + result = self.app.invoke(initial_state) + return result.get("final_action", {}) +\ No newline at end of file diff --git a/agent/sub_agents/water_and_atmospheric_dependencies/nodes.py b/agent/sub_agents/water_and_atmospheric_dependencies/nodes.py @@ -0,0 +1,136 @@ +import json +from langchain_core.messages import HumanMessage, SystemMessage +from agent.sub_agents.water_and_atmospheric_dependencies.physics_engine import predict_outcome +from agent.sub_agents.water_and_atmospheric_dependencies.retrieval import ask_historian, ask_rag +from agent.sub_agents.water_and_atmospheric_dependencies.tools import calculate_vpd, check_ph_safety, web_search +from langchain_core.messages import ToolMessage + +TOOL_MAP = { + "ask_historian": ask_historian, + "ask_rag": ask_rag, + "web_search": web_search, + "calculate_vpd": calculate_vpd, + "check_ph_safety": check_ph_safety +} + +def decide_node(state, model, system_prompt): + """ + Node 1: Drafts a plan OR calls a tool. + """ + print(f" 🤔 Thinking (Attempt {state['retry_count'] + 1})...") + + # 1. Initialize Messages if empty + messages = state.get("messages", []) + if not messages: + # First turn: Add System Prompt + User Context + messages = [SystemMessage(content=system_prompt)] + user_msg = ( + f"Current Sensors: {state['sensors']}\n" + f"Strategy: {state['strategy']}\n" + f"Research: {state['research_context']}\n" + f"History Context: {state.get('history', 'None provided')}\n" + ) + if state.get("critique"): + user_msg += f"\n\n❌ PREVIOUS SIMULATION FAILED: {state['critique']}" + + messages.append(HumanMessage(content=user_msg)) + + # 2. Invoke Model + response = model.invoke(messages) + + # 3. Update Message History + new_messages = messages + [response] + + # 4. Check for Tool Call + if response.tool_calls: + print(f" 📞 Calling Tool: {response.tool_calls[0]['name']}") + return { + "messages": new_messages, + "next_step": "tools" + } + + # 5. No Tool? Parse JSON Plan + try: + content = response.content.replace("```json", "").replace("```", "").strip() + draft = json.loads(content) + except: + draft = {} # Handle parsing error gracefully + + return { + "draft_plan": draft, + "messages": new_messages, + "next_step": "simulate", + "retry_count": state['retry_count'] + 1 + } + +def execute_tools_node(state): + """ + Executes the tool call and returns the result to the LLM. + """ + print(" ⚙️ Executing Tools...") + + # Safety check + if "messages" not in state or not state["messages"]: + raise ValueError("No messages found in state to execute tools from.") + + last_message = state["messages"][-1] + tool_results = [] + + for tool_call in last_message.tool_calls: + tool_name = tool_call["name"] + tool_args = tool_call["args"] + + if tool_name in TOOL_MAP: + try: + # Execute Tool + output = TOOL_MAP[tool_name].invoke(tool_args) + result_content = str(output) + except Exception as e: + result_content = f"Error executing {tool_name}: {e}" + else: + result_content = f"Error: Tool {tool_name} is not available." + + print(f" -> {tool_name}: {result_content[:50]}...") + + # Create Tool Message + tool_results.append(ToolMessage( + tool_call_id=tool_call["id"], + name=tool_name, + content=result_content + )) + + # Return updated history so 'decide_node' sees the answer + return {"messages": state["messages"] + tool_results} + +def simulate_node(state): + """ + Node 2: The Safety Sandbox. + """ + print(" 🧪 Simulating Outcome...") + draft = state.get('draft_plan') + + if not draft: + return {"simulation_result": {"passed": False, "reason": "No valid JSON plan generated."}} + + current = state['sensors'] + prediction = predict_outcome(current, draft) + + health = prediction.get('predicted_health', 0) + risk = prediction.get('risk_warning', "None") + + result = {"passed": False, "reason": ""} + + # Safety Threshold + if health < 92.0: + result["reason"] = f"Predicted Health drops to {health}%. Warning: {risk}" + else: + result["passed"] = True + + return {"simulation_result": result} + +def finalize_node(state): + """ + Node 3: Lock it in. + """ + print(" ✅ Plan Approved.") + return {"final_action": state['draft_plan']} +\ No newline at end of file diff --git a/agent/sub_agents/water_and_atmospheric_dependencies/physics_engine.py b/agent/sub_agents/water_and_atmospheric_dependencies/physics_engine.py @@ -0,0 +1,63 @@ +import os +import json +from langchain_openai import ChatOpenAI +from langchain_core.messages import SystemMessage, HumanMessage + +# Configuration +API_KEY = os.environ.get("GROQ_API_KEY") +MODEL_ID = "llama-3.3-70b-versatile" # Using the latest supported Groq model + +def predict_outcome(current_state: dict, proposed_action: dict) -> dict: + """ + Stateless 'What-If' Engine using Groq (LLM-based Physics). + Takes a snapshot and an action, returns the PREDICTED future state. + """ + if not API_KEY: + print(" ⚠️ Physics Engine Error: Missing GROQ_API_KEY") + return {"predicted_health": 50.0, "risk_warning": "No API Key configured"} + + # Initialize Groq Client + llm = ChatOpenAI( + base_url="https://api.groq.com/openai/v1", + api_key=API_KEY, + model=MODEL_ID, + temperature=0.1, # Low temp for consistent physics logic + max_tokens=1024 + ) + + system_prompt = ( + "You are a Hydroponic Physics Engine.\n" + "Your task is to simulate the biological and chemical reaction of a plant ecosystem " + "to a specific set of environmental changes over a 4-hour period.\n" + "BE REALISTIC. If parameters are extreme (e.g. pH < 4, Temp > 35C), predict drastic health drops." + ) + + user_prompt = ( + f"Current Sensor Readings: {json.dumps(current_state)}\n" + f"Proposed Action/Targets: {json.dumps(proposed_action)}\n\n" + f"TASK:\n" + f"1. Predict the Plant Health (0-100) after 4 hours.\n" + f"2. Identify any specific risks (Root Rot, Tip Burn, Lockout, Shock).\n" + f"OUTPUT JSON ONLY: {{ 'predicted_health': float, 'risk_warning': string }}" + ) + + try: + # Invoke Groq + response = llm.invoke([ + SystemMessage(content=system_prompt), + HumanMessage(content=user_prompt) + ]) + + # Clean and Parse JSON + content = response.content.replace("```json", "").replace("```", "").strip() + result = json.loads(content) + + # Default fallback keys if the LLM misses them + return { + "predicted_health": result.get("predicted_health", 50.0), + "risk_warning": result.get("risk_warning", "Unknown Risk") + } + + except Exception as e: + print(f" ⚠️ Physics Engine Error: {e}") + return {"predicted_health": 50.0, "risk_warning": "Simulation Connection Failed"} +\ No newline at end of file diff --git a/agent/sub_agents/water_and_atmospheric_dependencies/retrieval.py b/agent/sub_agents/water_and_atmospheric_dependencies/retrieval.py @@ -0,0 +1,67 @@ +import os +import json +from langchain.tools import tool +from qdrant_client import QdrantClient + +# Import your existing modules +from agent.sub_agents.Researcher import ResearcherAgent +from agent.Qdrant.Store import COLLECTION_NAME + +# Initialize shared clients +# Note: We rely on the existing ResearcherAgent logic for embeddings/search +researcher_instance = ResearcherAgent() + +# Initialize Qdrant for the Historian +qdrant_client = QdrantClient( + url=os.environ.get("QDRANT_URL", "http://localhost:6333"), + api_key=os.environ.get("QDRANT_API_KEY"), +) + +@tool +def ask_historian(query: str): + """ + Search the Farm's Database (History) for similar past events. + Useful for checking past mistakes, success rates, or specific scenarios. + + Args: + query: A description of the situation to look up (e.g. "What happened when pH dropped to 5.5?") + """ + try: + # 1. We use the Researcher's internal embedder to vectorize the query + # (Assuming ResearcherAgent has a method/property for this, or we use a fresh one) + # If your ResearcherAgent doesn't expose it, we can fallback to a simple keyword search + # or instantiate a lightweight SentenceTransformer here. + + # For this example, we'll assume the Researcher can give us a vector: + query_vector = researcher_instance.embed_query(query) + + hits = qdrant_client.search( + collection_name=COLLECTION_NAME, + query_vector=query_vector, + limit=3 + ) + + results = [] + for hit in hits: + payload = hit.payload + results.append(f"Outcome: {payload.get('outcome')}\nAction: {payload.get('action_taken')}\n---") + + return "\n".join(results) if results else "No relevant history found." + + except Exception as e: + return f"Historian unavailable: {str(e)}" + +@tool +def ask_rag(query: str): + """ + Consult the Research Assistant (RAG) for scientific knowledge. + Useful for finding optimal ranges, chemical interactions, or biological facts. + + Args: + query: Scientific question (e.g. "Optimal VPD for Lettuce in late flower") + """ + try: + # Reuse your existing ResearcherAgent logic + return researcher_instance.search(query) + except Exception as e: + return f"Research unavailable: {str(e)}" +\ No newline at end of file diff --git a/agent/sub_agents/water_and_atmospheric_dependencies/state.py b/agent/sub_agents/water_and_atmospheric_dependencies/state.py @@ -0,0 +1,20 @@ +from typing import List, TypedDict, Optional, Dict, Any +from langchain_core.messages import BaseMessage + +class AgentState(TypedDict): + # Inputs + sensors: Dict[str, float] + strategy: str + research_context: str + history: str + + # Internal Processing + draft_plan: Optional[Dict[str, Any]] + simulation_result: Optional[Dict[str, Any]] + critique: Optional[str] + retry_count: int + + messages: List[BaseMessage] + + # Final Output + final_action: Optional[Dict[str, Any]] +\ No newline at end of file diff --git a/agent/sub_agents/water_and_atmospheric_dependencies/tools.py b/agent/sub_agents/water_and_atmospheric_dependencies/tools.py @@ -0,0 +1,27 @@ +from langchain.tools import tool +import math + +@tool +def calculate_vpd(air_temp_c: float, humidity: float) -> float: + """Calculates Vapor Pressure Deficit (kPa). Target: 0.8-1.2 kPa.""" + svp = 0.6108 * math.exp((17.27 * air_temp_c) / (air_temp_c + 237.3)) + avp = svp * (humidity / 100.0) + return round(svp - avp, 2) + +@tool +def check_ph_safety(current_ph: float, proposed_change: float) -> str: + """Checks if a pH swing is too aggressive (>0.5 change).""" + predicted_ph = current_ph + proposed_change + swing = abs(predicted_ph - current_ph) + if swing > 0.5: + return f"DANGER: pH swing of {swing} is too high. Max safe swing is 0.5." + if predicted_ph < 5.0 or predicted_ph > 7.0: + return f"WARNING: Target pH {predicted_ph} is out of safe range (5.5-6.5)." + return "SAFE" + +# Placeholder for your Web Search (Tavily/Serper) +@tool +def web_search(query: str) -> str: + """Searches the web for specific hydroponic thresholds.""" + # Implement your search logic here (or use LangChain's TavilySearchResults) + return f"Simulated search result for: {query}" +\ No newline at end of file diff --git a/agent/tools/actuation.py b/agent/tools/actuation.py @@ -0,0 +1,80 @@ +from pydantic import BaseModel +from typing import Dict, Any + +# --- USER PROVIDED MODEL --- +class FarmAction(BaseModel): + acid_dosage_ml: float = 0.0 + base_dosage_ml: float = 0.0 + nutrient_dosage_ml: float = 0.0 + fan_speed_pct: float = 0.0 + water_refill_l: float = 0.0 + +# --- CALIBRATION CONSTANTS --- +# How much does 1ml of solution change the reservoir? +# Assuming a ~50L reservoir for this simulation +RESERVOIR_LITERS = 50.0 +PH_STRENGTH = 0.02 # 1ml changes 50L by 0.02 pH +EC_STRENGTH = 0.05 # 1ml changes 50L by 0.05 EC + +def convert_targets_to_actions(current_state: Dict[str, float], target_state: Dict[str, float]) -> FarmAction: + """ + Acts as a Proportional Controller. + Calculates the exact dosages/fan speeds needed to hit the targets. + """ + action = FarmAction() + + # 1. pH CONTROL (Acid/Base) + current_ph = current_state.get('ph', 6.0) + target_ph = target_state.get('ph', 6.0) + ph_error = target_ph - current_ph + + # Deadband: Don't dose if within 0.1 + if abs(ph_error) > 0.1: + needed_change = abs(ph_error) + # Formula: Dose = (Delta / Strength) + dose = needed_change / PH_STRENGTH + + if ph_error < 0: + # Current is too high -> Need Acid + action.acid_dosage_ml = round(dose, 2) + else: + # Current is too low -> Need Base + action.base_dosage_ml = round(dose, 2) + + # 2. EC CONTROL (Nutrients/Water) + current_ec = current_state.get('ec', 1.5) + target_ec = target_state.get('ec', 1.5) + ec_error = target_ec - current_ec + + if abs(ec_error) > 0.1: + if ec_error > 0: + # Current is too low -> Add Nutrients + dose = ec_error / EC_STRENGTH + action.nutrient_dosage_ml = round(dose, 2) + else: + # Current is too high -> Dilute with Water + # Rough heuristic: Add 1L water to drop EC by ~0.1 + dilution_needed = abs(ec_error) * 10 + action.water_refill_l = round(dilution_needed, 2) + + # 3. ATMOSPHERIC CONTROL (Fans) + # Fans cool down air and lower humidity + current_temp = current_state.get('air_temp', 25) + target_temp = target_state.get('air_temp', 25) + current_rh = current_state.get('humidity', 60) + target_rh = target_state.get('humidity', 60) + + # Simple Logic: If too hot OR too humid, ramp up fans + temp_error = current_temp - target_temp + rh_error = current_rh - target_rh + + fan_speed = 0.0 + if temp_error > 0: fan_speed += temp_error * 10 # +1C = +10% speed + if rh_error > 0: fan_speed += rh_error * 2 # +1% RH = +2% speed + + # Minimum circulation + fan_speed = max(10.0, fan_speed) + # Clamp to 100% + action.fan_speed_pct = min(100.0, round(fan_speed, 1)) + + return action +\ No newline at end of file