demeter
Autonomous Hydroponic Intelligencegit clone https://github.com/maydayv7/demeter
agent/sub_agents/Supervisor.py (12969B)
1import os
2import json
3import numpy as np
4from langchain_openai import AzureChatOpenAI
5from langchain_core.messages import SystemMessage, HumanMessage
6from langgraph.graph import StateGraph, END
7from agent.tools.actuation import convert_targets_to_actions
8
9from agent.Marl.bandit import ContextualBandit
10from agent.Marl.strategies import STRATEGIES, NUM_ACTIONS
11from agent.Qdrant.Store import store_fmu
12from agent.sub_agents.water_and_atmospheric_dependencies.physics_engine import predict_outcome
13
14# 🛡️ GUARDRAILS
15from agent.guardrails.validation import validate_plan, detect_hard_violations, create_validation_report
16
17from dotenv import load_dotenv
18
19load_dotenv()
20
21# --- NEW TOOLS DEFINITION ---
22def check_cross_domain_conflicts(atmos, water):
23 conflicts = []
24
25 # 1. Thermal Shock Check
26 air_t = atmos.get('air_temp', 25)
27 water_t = water.get('water_temp', 20)
28 if abs(air_t - water_t) > 10:
29 conflicts.append(f"CRITICAL: Thermal Shock Risk. Air ({air_t}C) and Water ({water_t}C) delta > 10C.")
30
31 # 2. Transpiration vs Uptake Check
32 # High VPD (Dry) + High EC (Salty) = Burn Risk
33 rh = atmos.get('humidity', 60)
34 ec = water.get('ec', 1.0)
35 if rh < 50 and ec > 2.0:
36 conflicts.append(f"STRESS: Low Humidity ({rh}%) + High EC ({ec}) will cause Tip Burn.")
37
38 return conflicts
39
40def validate_hard_limits(plan):
41 """
42 Validates plan against hard limits using guardrails.
43 Returns list of violations.
44 """
45 has_violations, violations = detect_hard_violations(plan)
46
47 if has_violations:
48 print(f"🚫 HARD LIMIT VIOLATIONS DETECTED:")
49 for v in violations:
50 print(f" {v}")
51
52 # Also check for obvious physics conflicts
53 additional_conflicts = []
54
55 if plan.get('air_temp', 25) - plan.get('water_temp', 20) > 10:
56 additional_conflicts.append("⚠️ Thermal Shock Risk: Air/Water temp delta > 10°C")
57
58 if plan.get('humidity', 60) < 40 and plan.get('ec', 1.0) > 2.5:
59 additional_conflicts.append("⚠️ Burn Risk: Low humidity + high EC")
60
61 if plan.get('humidity', 60) > 85:
62 additional_conflicts.append("🚫 Mold Risk: Humidity > 85%")
63
64 return violations + additional_conflicts
65
66# --- STATE DEFINITION ---
67from typing import TypedDict, Optional, Dict, Any, List
68
69class SupervisorState(TypedDict):
70 # Inputs
71 atmos_plan: Dict[str, Any]
72 water_plan: Dict[str, Any]
73 strategy_advice: str # Kept as advice, not law
74
75 # Processing
76 merged_plan: Dict[str, Any]
77 review_notes: List[str]
78 simulation_health: float
79
80 # Output
81 final_decision: str # "APPROVE" or "REJECT"
82 critique: str # Feedback for sub-agents if Rejected
83
84API_KEY = os.environ.get("AZURE_OPENAI_API_KEY")
85ENDPOINT = os.environ.get("AZURE_OPENAI_ENDPOINT")
86DEPLOYMENT_NAME = os.environ.get("AZURE_OPENAI_DEPLOYMENT_NAME", "gpt-4.1")
87API_VERSION = os.environ.get("AZURE_OPENAI_API_VERSION", "2024-12-01-preview")
88
89class SupervisorAgent:
90 def __init__(self, researcher_agent=None):
91 self.name = "Supervisor"
92 self.bandit = ContextualBandit(n_actions=NUM_ACTIONS, feature_dim=519)
93
94 if API_KEY and ENDPOINT:
95 self.model = AzureChatOpenAI(
96 azure_endpoint=ENDPOINT,
97 api_key=API_KEY,
98 api_version=API_VERSION,
99 deployment_name=DEPLOYMENT_NAME,
100 temperature=0.0 # Zero temp for strict judging
101 )
102
103 self.app = self._build_graph()
104
105 def _build_graph(self):
106 workflow = StateGraph(SupervisorState)
107
108 # 1. Merge: Combine the two JSONs
109 workflow.add_node("merge", self.node_merge)
110
111 # 2. Review: Run the 3 Tools (Conflicts, Limits, Physics)
112 workflow.add_node("review", self.node_review)
113
114 # 3. Judge: LLM decides if the issues are fatal
115 workflow.add_node("judge", self.node_judge)
116
117 # Flow
118 workflow.set_entry_point("merge")
119 workflow.add_edge("merge", "review")
120 workflow.add_edge("review", "judge")
121 workflow.add_edge("judge", END)
122
123 return workflow.compile()
124
125 # --- NODE FUNCTIONS ---
126
127 def node_merge(self, state):
128 print(" 🔗 Supervisor Merging Plans...")
129 # Simple dictionary merge
130 merged = {**state['atmos_plan'], **state['water_plan']}
131 return {"merged_plan": merged}
132
133 def node_review(self, state):
134 print(" 🔍 Supervisor Running Unit Tests...")
135 plan = state['merged_plan']
136 notes = []
137
138 # Tool 1: Conflict Check
139 conflicts = check_cross_domain_conflicts(state['atmos_plan'], state['water_plan'])
140 if conflicts:
141 notes.extend(conflicts)
142
143 # Tool 2: Limit Check
144 limits = validate_hard_limits(plan)
145 if limits:
146 notes.extend(limits)
147
148 # Tool 3: Physics Simulator
149 # Comparing plan vs itself as a snapshot for now
150 health = 100
151
152 if health < 90:
153 notes.append(f"SIMULATION FAIL: Predicted health drops to {health}%. ")
154
155 return {"review_notes": notes, "simulation_health": health}
156
157 def node_judge(self, state):
158 """
159 The LLM looks at the automated test results and makes the final call.
160 """
161 print(" ⚖️ Supervisor Judging...")
162
163 if not state['review_notes']:
164 # No issues found by tools
165 return {"final_decision": "APPROVE", "critique": "Plan looks solid."}
166
167 # If issues exist, ask LLM if they are fatal or acceptable trade-offs
168 prompt = f"""
169 You are the Quality Assurance Supervisor.
170
171 PROPOSED PLAN: {state['merged_plan']}
172
173 AUTOMATED TEST FAILURES:
174 {json.dumps(state['review_notes'], indent=2)}
175
176 ADVISORY STRATEGY: {state['strategy_advice']}
177
178 TASK:
179 1. BIAS: You should almost always APPROVE.
180 2. ONLY 'REJECT' if the plan is physically impossible or immediately fatal (e.g., pH < 3.0, Water Temp > 40°C).
181 3. IGNORE 'Simulation Fail' warnings if the Strategy justifies the extreme values (e.g., 'Flush' requires low EC).
182 4. Treat "Risk" warnings as acceptable trade-offs for the strategy.
183 OUTPUT JSON: {{ "verdict": "APPROVE" or "REJECT", "critique": "Explanation..." }}
184 """
185
186 print("Supervisor Prompt:\n", prompt)
187
188 try:
189 response = self.model.invoke([HumanMessage(content=prompt)])
190 content = response.content.replace("```json", "").replace("```", "").strip()
191 result = json.loads(content)
192
193 return {
194 "final_decision": result.get("verdict", "REJECT"),
195 "critique": result.get("critique", "Automated tests failed.")
196 }
197 except:
198 # Default to reject if unsafe
199 return {"final_decision": "REJECT", "critique": "Plan failed automated safety checks."}
200
201
202
203
204 # --- ENTRY POINT ---
205
206 def synthesize_plan(self, atmos_plan, water_plan, fmu, history, strategy_info):
207 strategy_name, _, action_idx = strategy_info
208
209 initial_state = {
210 "atmos_plan": atmos_plan,
211 "water_plan": water_plan,
212 "strategy_advice": strategy_name,
213 "merged_plan": {},
214 "review_notes": [],
215 "simulation_health": 0.0,
216 "final_decision": "",
217 "critique": ""
218 }
219
220 result = self.app.invoke(initial_state)
221 final_targets = result.get("merged_plan", {})
222
223 current_sensors = fmu.metadata.get('sensors', {})
224
225 # 🛡️ GUARDRAIL CHECK: Validate before executing
226 print(f"[{self.name}] 🛡️ Running guardrail validation...")
227 validation = validate_plan(final_targets)
228
229 if validation["severity"] == "CRITICAL":
230 print(create_validation_report(final_targets))
231 print(f"[{self.name}] ⚠️ CRITICAL VIOLATIONS - Clamping to bounds...")
232 final_targets = validation["bounded_plan"]
233
234 if validation["warnings"]:
235 print(f"[{self.name}] ⚠️ Warnings: {', '.join(validation['warnings'])}")
236
237 print(f"[{self.name}] ⚙️ Converting Targets to Actuator Commands...")
238
239 sensor_vals = [
240 float(current_sensors.get("pH", 0.0)),
241 float(current_sensors.get("EC", 0.0)),
242 float(current_sensors.get("temp", 0.0)),
243 float(current_sensors.get("humidity", 0.0))
244 ]
245
246 if hasattr(fmu, 'vector') and len(fmu.vector) == 512:
247 if isinstance(fmu.vector, list):
248 fmu.vector.extend(sensor_vals)
249 else:
250 import numpy as np
251 fmu.vector = np.concatenate((fmu.vector, sensor_vals)).tolist()
252
253 # Calculate physical actions
254 physical_action_obj = convert_targets_to_actions(current_sensors, final_targets)
255
256 # Convert Pydantic model to Dict for JSON serialization
257 final_payload = physical_action_obj.dict()
258
259 # Log it
260 print(f"[{self.name}] 🚜 Activating Hardware: {final_payload}")
261
262 # Store in FMU
263 fmu.metadata["action_taken"] = str(final_payload)
264 fmu.metadata["bandit_action_id"] = action_idx
265 fmu.metadata["strategic_intent"] = strategy_name
266
267 if "image_b64" in fmu.metadata: del fmu.metadata["image_b64"]
268 store_fmu(fmu)
269
270 return final_payload
271
272 # --- ADVISORY ONLY (Not Enforced) ---
273 def get_strategic_goal(self, fmu):
274 # (Same as before, but treated as advice now)
275 sensors = fmu.metadata.get('sensors', {})
276 fmu_vector = fmu.vector
277 vis_vec1 = np.array(fmu_vector) if isinstance(fmu_vector, list) else fmu_vector
278 vis_vec = vis_vec1[:512] if len(vis_vec1) >= 512 else None
279 if vis_vec is None or len(vis_vec) == 0: vis_vec = np.zeros(516)
280
281 s_vec = np.array([
282 (float(sensors.get('pH', 6.0)) - 6.0) / 2.0,
283 float(sensors.get('EC', 1.0)) / 3.0,
284 float(sensors.get('temp', 25.0)) / 40.0
285 ])
286 context_vector = np.concatenate([vis_vec, s_vec])
287
288 # print("Context Vector for Bandit:", context_vector.shape)
289
290 action_idx, _ = self.bandit.select_action(context_vector)
291 strategy_name = STRATEGIES[action_idx]
292
293 return strategy_name, "Advisory Only", int(action_idx)
294
295 def learn_from_outcome(self, fmu, outcome_info):
296 """
297 Bandit Learning: Update the model based on action outcome.
298
299 Args:
300 fmu: The FMU object containing metadata about the previous action
301 outcome_info: Either:
302 - Current plant health (0-100) from simulator, OR
303 - Reward score (-1.0 to 1.0) from judge
304 """
305 # Retrieve the action that was taken in the previous cycle
306 prev_action_idx = fmu.metadata.get("bandit_action_id")
307 if prev_action_idx is None:
308 return # No previous action to learn from
309
310 prev_action_idx = int(prev_action_idx)
311
312 # Build the context vector (same as get_strategic_goal)
313 sensors = fmu.metadata.get('sensors', {})
314 fmu_vector = fmu.vector
315 vis_vec1 = np.array(fmu_vector) if isinstance(fmu_vector, list) else fmu_vector
316 vis_vec = vis_vec1[:512] if len(vis_vec1) >= 512 else None
317 if vis_vec is None or len(vis_vec) == 0: vis_vec = np.zeros(512)
318
319 s_vec = np.array([
320 (float(sensors.get('pH', 6.0)) - 6.0) / 2.0,
321 float(sensors.get('EC', 1.0)) / 3.0,
322 float(sensors.get('temp', 25.0)) / 40.0
323 ])
324 context_vector = np.concatenate([vis_vec, s_vec])
325
326 # Handle reward: Can be health (0-100) or judge reward (-1 to 1)
327 if isinstance(outcome_info, dict):
328 reward = float(outcome_info.get("reward", 0.0))
329 else:
330 # Assume it's health (0-100), convert to normalized reward
331 health_val = float(outcome_info)
332 if health_val >= 85:
333 reward = 1.0 # Excellent
334 elif health_val >= 70:
335 reward = health_val / 100.0 # Good
336 elif health_val >= 50:
337 reward = (health_val / 100.0) * 0.5 # Mediocre
338 else:
339 reward = -1.0 # Terrible
340
341 # Update the bandit model with this outcome
342 self.bandit.update(context_vector, prev_action_idx, reward)
343
344 strategy_name = STRATEGIES.get(prev_action_idx, "UNKNOWN")
345 print(f"[{self.name}] 🧠 Bandit Learning: {strategy_name} (Action {prev_action_idx}) → Reward {reward:.2f}")
346
347 # Save the updated model
348 self.bandit.save()