demeter
Autonomous Hydroponic Intelligencegit clone https://github.com/maydayv7/demeter
agent/sub_agents/judge_agent.py (10760B)
1import os
2import json
3import base64
4import re
5import tempfile
6from typing import TypedDict, Dict, Any, Optional
7
8from langchain_openai import AzureChatOpenAI
9from langchain_core.messages import SystemMessage, HumanMessage
10from langgraph.graph import StateGraph, END
11from qdrant_client import models
12from dotenv import load_dotenv
13
14load_dotenv()
15
16from Sentinel.fmu import FMU
17from agent.sub_agents.base_agent import BaseReasoningAgent
18from Qdrant.Client import client
19from Qdrant.Store import COLLECTION_NAME
20
21# Import farm_memory to allow writing verdicts
22from agent.sub_agents.water_and_atmospheric_dependencies.retrieval import diagnose_plant, ask_memory, farm_memory
23
24# --- STATE DEFINITION ---
25class JudgeState(TypedDict):
26 # Inputs
27 current_fmu: Any
28 image_b64: str # Added to State
29
30 # Internal Context
31 prev_point: Any
32 crop_id: str
33
34 # Forensic Evidence
35 visual_report: Dict
36 biography: Any
37
38 # Verdict
39 reward: float
40 outcome: str
41 explanation: str
42
43 # Output
44 training_data: Optional[Dict]
45
46class JudgeAgent(BaseReasoningAgent):
47 def __init__(self):
48 super().__init__(name="Judge Agent")
49 self.qdrant = client
50
51 # LLM for the "Deliberation" phase
52 self.llm = AzureChatOpenAI(
53 azure_endpoint=os.environ.get("AZURE_OPENAI_ENDPOINT"),
54 api_key=os.environ.get("AZURE_OPENAI_API_KEY"),
55 api_version=os.environ.get("AZURE_OPENAI_API_VERSION", "2024-12-01-preview"),
56 deployment_name=os.environ.get("AZURE_OPENAI_DEPLOYMENT_NAME", "gpt-4.1"),
57 temperature=0.1
58 )
59
60 self.app = self._build_graph()
61
62 def _build_graph(self):
63 workflow = StateGraph(JudgeState)
64 workflow.add_node("retrieve_evidence", self.node_retrieve_evidence)
65 workflow.add_node("run_forensics", self.node_run_forensics)
66 workflow.add_node("deliberate", self.node_deliberate)
67 workflow.add_node("file_verdict", self.node_file_verdict)
68
69 workflow.set_entry_point("retrieve_evidence")
70
71 workflow.add_conditional_edges(
72 "retrieve_evidence",
73 lambda x: "run_forensics" if x.get("prev_point") else "end_no_history",
74 {
75 "run_forensics": "run_forensics",
76 "end_no_history": END
77 }
78 )
79
80 workflow.add_edge("run_forensics", "deliberate")
81 workflow.add_edge("deliberate", "file_verdict")
82 workflow.add_edge("file_verdict", END)
83
84 return workflow.compile()
85
86 # --- NODES ---
87
88 def node_retrieve_evidence(self, state: JudgeState):
89 print(f"[{self.name}] 🕵️ Retrieve Evidence...")
90 fmu = state["current_fmu"]
91 crop_id = fmu.metadata.get("crop_id")
92 current_seq = fmu.metadata.get("sequence_number", 1)
93
94 if current_seq <= 1:
95 print(" -> First cycle. No history to judge.")
96 return {"prev_point": None}
97
98 prev_seq = current_seq - 1
99 try:
100 s_filter = models.Filter(
101 must=[
102 models.FieldCondition(key="crop_id", match=models.MatchValue(value=crop_id)),
103 models.FieldCondition(key="sequence_number", match=models.MatchValue(value=prev_seq))
104 ]
105 )
106 res, _ = self.qdrant.scroll(
107 collection_name=COLLECTION_NAME,
108 scroll_filter=s_filter,
109 limit=1,
110 with_vectors=True
111 )
112 return {"prev_point": res[0] if res else None, "crop_id": crop_id}
113 except Exception as e:
114 print(f" -> DB Error: {e}")
115 return {"prev_point": None}
116
117 def node_run_forensics(self, state: JudgeState):
118 """
119 Executes the TWO mandated tools: diagnose_plant and ask_memory.
120 """
121 print(f"[{self.name}] 🔎 Running Forensics...")
122 crop_id = state["crop_id"]
123 image_b64 = state.get("image_b64")
124
125 # --- TOOL 1: diagnose_plant ---
126 visual_data = {"status": "No Image"}
127 if image_b64:
128 try:
129 with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as temp:
130 temp.write(base64.b64decode(image_b64))
131 temp_path = temp.name
132
133 print(f" -> Invoking Tool inside: diagnose_plant")
134 visual_data = diagnose_plant.invoke({"image_b64": image_b64})
135 os.remove(temp_path)
136 except Exception as e:
137 visual_data = {"error": str(e)}
138 else:
139 print(" -> No image found for diagnosis.")
140
141 # --- TOOL 2: ask_memory ---
142 # print(f" -> Invoking Tool: ask_memory for '{crop_id}'")
143 try:
144 # We updated the tool to expect 'plant_id', so we must pass that key
145 raw_memory = ask_memory.invoke({"plant_id": crop_id})
146 # FIX: Force conversion to string to ensure it renders in prompt
147 # print(f" -> Memory Retrieved: '{raw_memory}'")
148 memory_data = str(raw_memory)
149
150 # print(f" -> Memory Retrieved {memory_data}")
151 self
152 except Exception as e:
153 print(f" -> Memory Tool Error: {e}")
154 memory_data = "Memory unavailable."
155
156 # Explicitly return the dict to update state keys
157 return {
158 "visual_report": visual_data,
159 "biography": memory_data
160 }
161
162 def node_deliberate(self, state: JudgeState):
163 """
164 LLM synthesizes Visual + History + Sensor Delta to form a verdict.
165 """
166 print(f"[{self.name}] ⚖️ Deliberating...")
167
168 # --- DEBUG: Verify State Content ---
169 # print(f"DEBUG CHECK -> Biography Content: '{state.get('biography')}'")
170
171 prev_sensors = state["prev_point"].payload.get("sensors", {})
172 curr_sensors = state["current_fmu"].metadata.get("sensors", {})
173 visual = state["visual_report"]
174
175 # Ensure history is never None
176 history = state.get("biography", "No history available.")
177
178 prompt = f"""
179 You are the Chief Judge of an Automated Farm.
180 Evaluate the result of the LAST ACTION based on the transition from State N-1 to State N.
181
182 --- EVIDENCE ---
183 PREVIOUS SENSORS (N-1): {prev_sensors}
184 CURRENT SENSORS (N): {curr_sensors}
185
186 VISUAL AUTOPSY (Doctor's Report):
187 {json.dumps(visual, indent=2)}
188
189 PLANT BIOGRAPHY (Past Issues):
190 {history}
191
192 --- RUBRIC ---
193 1. IF Doctor found "DISEASE_DETECTED": Reward = -1.0 (Critical Failure).
194 2. IF Doctor found "HEALTHY" AND Sensors moved closer to targets: Reward = 0.5 to 1.0.
195 3. IF Sensors drifted away from targets: Reward = -0.5.
196
197 TASK:
198 Output JSON: {{ "outcome": "IMPROVED"|"DETERIORATED"|"STABLE", "reward": float(-1.0 to 1.0), "reason": "Short explanation" }}
199 """
200
201 # print("Judge Prompt:\n", prompt)
202 try:
203 response = self.llm.invoke([HumanMessage(content=prompt)])
204 # print(f" -> LLM Response for Judge Deliberation: {response.content}")
205 content = response.content.strip()
206
207 # print(f" -> Raw LLM Output: '{content}'")
208 code_block_match = re.search(r"```json\s*(\{.*?\})\s*```", content, re.DOTALL)
209
210 if code_block_match:
211 json_str = code_block_match.group(1)
212 else:
213 # 2. Fallback: Find the first '{' and the last '}'
214 # This handles cases where the LLM forgets the ```json tags
215 json_match = re.search(r"\{.*\}", content, re.DOTALL)
216 if json_match:
217 json_str = json_match.group(0)
218 else:
219 raise ValueError(f"No JSON found in response: {content[:50]}...")
220
221 # 3. Parse
222 verdict = json.loads(json_str)
223
224 print(f" -> Verdict: {verdict.get('outcome')} ({verdict.get('reward')})")
225 return {
226 "outcome": verdict.get("outcome", "STABLE"),
227 "reward": verdict.get("reward", 0.0),
228 "explanation": verdict.get("reason", "Analysis complete.")
229 }
230 except Exception as e:
231 print(f" -> Deliberation Failed: {e}")
232 return {"outcome": "ERROR", "reward": 0.0, "explanation": "Judge LLM failed."}
233
234 def node_file_verdict(self, state: JudgeState):
235 """
236 Writes the final judgment to Qdrant AND FarmMemory.
237 """
238 print(f"[{self.name}] 📝 Filing Verdict...")
239
240 prev_id = state["prev_point"].id
241 crop_id = state["crop_id"]
242
243 # 1. Update Qdrant Snapshot
244 self.qdrant.set_payload(
245 collection_name=COLLECTION_NAME,
246 points=[prev_id],
247 payload={
248 "outcome": f"{state['outcome']} | Reward: {state['reward']}",
249 "reward_score": state['reward'],
250 "explanation_log": state["explanation"],
251 "visual_diagnosis": str(state["visual_report"].get("health_assessment", "N/A"))
252 }
253 )
254
255 # 2. Write to FarmMemory (Text/Biography Store)
256 try:
257 verdict_summary = (
258 f"Cycle Review for {crop_id}: Result was {state['outcome']} "
259 f"(Reward: {state['reward']}). Judge's Note: {state['explanation']}"
260 )
261 farm_memory.log_event(crop_id, verdict_summary)
262 except Exception as e:
263 print(f" -> ⚠️ Failed to log to FarmMemory: {e}")
264
265 # Prepare Training Data Bundle
266 training_data = {
267 "reward": state['reward'],
268 "prev_action_idx": state["prev_point"].payload.get("bandit_action_id"),
269 "prev_vector": state["prev_point"].vector,
270 "prev_sensors": state["prev_point"].payload.get("sensor_data", {})
271 }
272
273 return {"training_data": training_data}
274
275 # --- ENTRY POINT ---
276 def review_previous_cycle(self, current_fmu: FMU, image_b64: str):
277 initial_state = {
278 "current_fmu": current_fmu,
279 "image_b64": image_b64,
280 "prev_point": None,
281 "crop_id": "",
282 "visual_report": {},
283 "biography": "", # Starts empty
284 "reward": 0.0,
285 "outcome": "",
286 "explanation": "",
287 "training_data": None
288 }
289
290 result = self.app.invoke(initial_state)
291 return result.get("training_data")