commit ffa4ad93da86c35247a5165c40f6772f094ee714 parent 43150d0359af4d215bb593ffbbbcd0e3dce8f1c8 Author: Abhinav Rai <69450646+AbhinavRai01@users.noreply.github.com> Date: Fri, 6 Mar 2026 11:32:24 +0000 Merge PR Diffstat:
18 files changed, 401 insertions(+), 262 deletions(-)
diff --git a/.cph/.E_The_Robotic_Rush.cpp_f1e7c1bd924933d8a52191ebb47b442e.prob b/.cph/.E_The_Robotic_Rush.cpp_f1e7c1bd924933d8a52191ebb47b442e.prob @@ -0,0 +1 @@ +{"name":"E. The Robotic Rush","group":"Codeforces - Codeforces Round 1074 (Div. 4)","url":"https://codeforces.com/problemset/problem/2185/E","interactive":false,"memoryLimit":256,"timeLimit":3000,"tests":[{"id":1770804198119,"input":"3\n2 1 3\n0 1\n2\nLRR\n2 3 3\n2 4\n1 3 5\nLRL\n3 2 3\n1 3 7\n9 6\nRRL\n","output":"2 2 1\n0 0 0\n3 2 2\n"}],"testType":"single","input":{"type":"stdin"},"output":{"type":"stdout"},"languages":{"java":{"mainClass":"Main","taskClass":"ETheRoboticRush"}},"batch":{"id":"e003d38e-d869-4634-bd53-072a040dab87","size":1},"srcPath":"d:\\projects\\Demeter\\Demeter\\E_The_Robotic_Rush.cpp"} +\ No newline at end of file diff --git a/agent/main_agent.py b/agent/main_agent.py @@ -39,17 +39,19 @@ def main(): print("="*50) # 1. Fetch - fmu, sensor_snapshot, history = fetcher.fetch_and_process() + fmu, sensor_snapshot, history, image_b64 = fetcher.fetch_and_process() if not fmu: print("⚠️ No FMU found. Waiting...") time.sleep(10) continue # 2. Judge - judge.review_previous_cycle(fmu) + time.sleep(2) # Small delay to ensure FMU is fully available before judging + judge.review_previous_cycle(fmu, image_b64) # 3. 🟢 GET BANDIT STRATEGY (The Brain) # The Supervisor consults the Bandit first to set the cycle's goal + time.sleep(2) # Ensure judge's review is complete before strategy retrieval strat_name, strat_instr, action_idx = supervisor.get_strategic_goal(fmu) print(f"\n🎰 BANDIT STRATEGY: {strat_name}") print(f"📝 Instruction: {strat_instr}") @@ -58,6 +60,8 @@ def main(): crop = fmu.metadata.get("crop", "unknown") stage = fmu.metadata.get("stage", "unknown") query = f"optimal hydroponic conditions for {crop} in {stage} stage" + + time.sleep(2) # Small delay before research research_context = researcher.search(query) # 5. 🟢 DELIBERATION (The Experts) @@ -65,20 +69,29 @@ def main(): print("\n🧠 Agents Planning...") # Updated call signature to match the new 'reason' method + time.sleep(2) # Ensure research context is ready before reasoning 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 + history=history, # Pass history for context awareness + image_b64=image_b64 # Pass the image data for visual diagnosis ) + + print(f"\n🌬️ Atmospheric Plan:\n{atmos_plan}") + + time.sleep(2) # Small delay between agent calls water_plan = water_agent.reason( sensors=sensor_snapshot, research=research_context, strategy=strat_instr, - history=history + history=history, + image_b64=image_b64 ) + print(f"\n💧 Water & Nutrient Plan:\n{water_plan}") + # 6. Synthesis (The Supervisor) # Supervisor merges plans, checks conflicts, and ensures safety print("\n👮 Supervisor Finalizing...") diff --git a/agent/memory.py b/agent/memory.py @@ -1,101 +1,161 @@ +import logging from mem0 import Memory import os +import time # <--- IMPORT ADDED from dotenv import load_dotenv -from qdrant_client import QdrantClient -from qdrant_client.http import models +from qdrant_client import QdrantClient, models load_dotenv() +logging.getLogger("mem0").setLevel(logging.WARNING) +logging.getLogger("httpx").setLevel(logging.WARNING) + class FarmMemory: def __init__(self): - # First, ensure the collection exists with correct dimensions + # 1. Setup Collection self._setup_collection() - self.memory = Memory.from_config({ - # 🟢 1. VECTOR STORE (Qdrant Cloud) + # 2. Initialize Mem0 + config = { "vector_store": { "provider": "qdrant", "config": { "url": os.getenv("QDRANT_URL"), "api_key": os.getenv("QDRANT_API_KEY"), - "collection_name": "Plant_Biographies_HF", # Different collection for 384 dims + "collection_name": "Plant_Biographies_HF", "port": 6333, } }, - # 🟢 2. LLM (Groq) "llm": { - "provider": "groq", + "provider": "openai", "config": { - "model": "llama-3.1-8b-instant", - "api_key": os.getenv("GROQ_API_KEY") + "model": "qwen/qwen3-32b", + "api_key": os.getenv("GROQ_API_KEY"), + "openai_base_url": "https://api.groq.com/openai/v1", + "max_tokens": 1500 } }, - # 🟢 3. EMBEDDER (HuggingFace - 384 dimensions) "embedder": { "provider": "huggingface", "config": { - "model": "all-MiniLM-L6-v2" # 384 dimensions + "model": "all-MiniLM-L6-v2" } } - }) + } + + print("🧠 Initializing FarmMemory...") + self.memory = Memory.from_config(config) def _setup_collection(self): - """Create the collection with correct vector dimensions if it doesn't exist""" - client = QdrantClient( - url=os.getenv("QDRANT_URL"), - api_key=os.getenv("QDRANT_API_KEY"), - ) - - collection_name = "Plant_Biographies_HF" - + """Create the collection manually to guarantee dimension match""" try: - # Check if collection exists - client.get_collection(collection_name) - # print(f"✅ Collection '{collection_name}' already exists") - except Exception: - # Create collection with 384 dimensions - print(f"📝 Creating collection '{collection_name}' with 384 dimensions...") - client.create_collection( - collection_name=collection_name, - vectors_config=models.VectorParams( - size=384, # HuggingFace all-MiniLM-L6-v2 dimension - distance=models.Distance.COSINE - ) + client = QdrantClient( + url=os.getenv("QDRANT_URL"), + api_key=os.getenv("QDRANT_API_KEY"), + timeout=30 ) - print(f"✅ Collection created successfully") + + collection_name = "Plant_Biographies_HF" + + if client.collection_exists(collection_name): + print(f"✅ Collection '{collection_name}' verified.") + else: + print(f"📝 Creating '{collection_name}' (384 dims)...") + client.create_collection( + collection_name=collection_name, + vectors_config=models.VectorParams( + size=384, + distance=models.Distance.COSINE + ) + ) + print(f"✅ Collection created.") + except Exception as e: + print(f"⚠️ Collection setup warning: {e}") - def get_plant_history(self, crop_id): - """Retrieve the complete biographical history of a plant""" + def get_plant_history(self, crop_id, limit=3): + """Retrieve the last N chronological entries for a crop""" try: - history = self.memory.search( - query=f"What is the health history and past treatments for {crop_id}?", - user_id=crop_id - ) + print(f"🔍 [FarmMemory] Fetching last {limit} entries for: {crop_id}...") - if not history: - return "No prior biographical records for this plant." + # 1. Use get_all() to fetch raw history (bypassing vector similarity) + # This ensures we get the *actual* latest events, not just "relevant" ones + history = self.memory.get_all(user_id=crop_id) + # print(f" -> Raw history count: {history}") + + - # Handle different possible response structures from mem0 - if isinstance(history, dict) and 'results' in history: - results = history['results'] - return "\n".join([f"- {item['memory']}" for item in results]) + # 2. Extract List from response + results = [] + if isinstance(history, dict): + results = history.get("results", []) elif isinstance(history, list): - formatted_lines = [] - for item in history: - text = item.get('memory', str(item)) if isinstance(item, dict) else str(item) - formatted_lines.append(f"- {text}") - return "\n".join(formatted_lines) + results = history - return str(history) + if not results: + return f"No biography found for {crop_id}." + + # 3. Sort by 'created_at' (Descending) to get Newest -> Oldest + # Safe sort: defaults to empty string if 'created_at' is missing + results.sort(key=lambda x: x.get('created_at', ''), reverse=True) + + # 4. Slice the top N (Past 3) + recent_results = results[:limit] + + # print(f" -> Extracted entries: {results}") + # 5. Format the output + formatted_lines = [] + for item in recent_results: + # Handle different mem0 versions where content might be in 'memory' or 'text' + text = item.get("memory", item.get("text", str(item))) + # print(f" -> Processing entry: {text}") + + # Optional: Add a timestamp to the output for verification + timestamp = item.get("created_at", "") + if timestamp: + # Cleanup timestamp for readability (e.g., 2024-02-09T10:00:00 -> 2024-02-09 10:00) + timestamp = timestamp.replace("T", " ").split(".")[0] + formatted_lines.append(f"[{timestamp}] {text}") + else: + formatted_lines.append(f"- {text}") + + clean_output = "\n".join(formatted_lines) + + # print(f" -> Formatted Output:\n{clean_output}") + return clean_output + except Exception as e: - print(f"⚠️ Error formatting history: {e}") + print(f"❌ [FarmMemory] Error: {e}") return f"Error retrieving history: {str(e)}" def log_event(self, crop_id, event_text): """Log a new event in the plant's biography""" try: - self.memory.add(event_text, user_id=crop_id) + # Capture the result to check if LLM extracted it correctly + result = self.memory.add(event_text, user_id=crop_id) print(f"🧠 Biography Updated for {crop_id}") except Exception as e: - print(f"❌ Memory Write Error: {e}") -\ No newline at end of file + print(f"❌ Memory Write Error: {e}") + + +# execute +if __name__ == "__main__": + farm_memory = FarmMemory() + print("\n✅ System Online.\n") + + # 1. WRITE: Manually log a test event + test_crop = "BATCH-VERDANT-X1" + # print(f"📝 Logging event for {test_crop}...") + farm_memory.log_event(test_crop, "CRITICAL TEST: Detected mild nutrient burn. Flushed system with pH 5.8 water.") + + # CRITICAL FIX: Wait for Vector DB indexing + print("⏳ Waiting for memory indexing...") + time.sleep(2) + + # 2. READ: Fetch it back + print(f"\n🔍 Retrieving history for {test_crop}...") + history = farm_memory.get_plant_history(test_crop) + + print("-" * 40) + print(f"MEMORY OUTPUT:\n{history}") + print("-" * 40) +\ No newline at end of file diff --git a/agent/sub_agents/Doctor.py b/agent/sub_agents/Doctor.py @@ -4,6 +4,9 @@ import json import os import logging import numpy as np +import base64 +from io import BytesIO +from PIL import Image # Setup basic logging logging.basicConfig(level=logging.INFO) @@ -40,19 +43,28 @@ class VisionAgent: logger.error(f"❌ Critical Error loading model: {e}") self.model = None - def analyze_frame(self, image_path): + def analyze_frame(self, image_b64): """ - Scans an image for pests, diseases, or growth stages. + Scans a base64 encoded image for pests, diseases, or growth stages. """ if not self.model: return {"error": "Model not initialized"} - if not os.path.exists(image_path): - return {"error": f"Image file not found: {image_path}"} + if not image_b64: + return {"error": "No image data provided"} try: - # 3. Run Inference - results = self.model.predict(image_path, conf=0.25, save=False, verbose=False) + # 3. Decode Base64 to Image + # Handle data URI scheme if present (e.g., "data:image/png;base64,...") + if "," in image_b64: + image_b64 = image_b64.split(",")[1] + + image_data = base64.b64decode(image_b64) + image = Image.open(BytesIO(image_data)) + + # 4. Run Inference + # YOLO can accept PIL Images directly + results = self.model.predict(image, conf=0.25, save=False, verbose=False) result = results[0] detections = [] @@ -70,7 +82,7 @@ class VisionAgent: }) summary_counts[label] = summary_counts.get(label, 0) + 1 - # 4. Health Logic + # 5. Health Logic health_status = "HEALTHY" visual_alert = False @@ -101,5 +113,6 @@ class VisionAgent: } except Exception as e: + logger.error(f"Error during analysis: {e}") return {"error": str(e)} \ No newline at end of file diff --git a/agent/sub_agents/Explainer.py b/agent/sub_agents/Explainer.py @@ -36,7 +36,7 @@ class ExplainerAgent: try: response = self.llm.chat.completions.create( - model="llama-3.1-8b-instant", + model="qwen/qwen3-32b", messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": context} diff --git a/agent/sub_agents/Researcher.py b/agent/sub_agents/Researcher.py @@ -46,7 +46,7 @@ class ResearcherAgent: ] ) - def search(self, query: str, limit: int = 3) -> str: + def search(self, query: str, limit: int = 1) -> str: """Retrieves relevant textbook pages and formats them as a string.""" # 1. Convert query to vector query_vec = list(self.encoder.embed([query]))[0] diff --git a/agent/sub_agents/Supervisor.py b/agent/sub_agents/Supervisor.py @@ -69,7 +69,7 @@ class SupervisorAgent: self.model = ChatOpenAI( base_url="https://api.groq.com/openai/v1", api_key=API_KEY, - model="llama-3.3-70b-versatile", + model="qwen/qwen3-32b", temperature=0.0 # Zero temp for strict judging ) @@ -149,11 +149,14 @@ class SupervisorAgent: 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. - + 1. BIAS: You should almost always APPROVE. + 2. ONLY 'REJECT' if the plan is physically impossible or immediately fatal (e.g., pH < 3.0, Water Temp > 40°C). + 3. IGNORE 'Simulation Fail' warnings if the Strategy justifies the extreme values (e.g., 'Flush' requires low EC). + 4. Treat "Risk" warnings as acceptable trade-offs for the strategy. OUTPUT JSON: {{ "verdict": "APPROVE" or "REJECT", "critique": "Explanation..." }} """ + + print("Supervisor Prompt:\n", prompt) try: response = self.model.invoke([HumanMessage(content=prompt)]) @@ -168,6 +171,9 @@ class SupervisorAgent: # 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): @@ -187,7 +193,7 @@ class SupervisorAgent: result = self.app.invoke(initial_state) final_targets = result.get("merged_plan", {}) - current_sensors = fmu.metadata.get('sensor_data', {}) + current_sensors = fmu.metadata.get('sensors', {}) print(f"[{self.name}] ⚙️ Converting Targets to Actuator Commands...") diff --git a/agent/sub_agents/atmospheric_agent.py b/agent/sub_agents/atmospheric_agent.py @@ -13,7 +13,7 @@ from agent.sub_agents.water_and_atmospheric_dependencies.tools import calculate_ # Configuration API_KEY = os.environ.get("GROQ_API_KEY") -MODEL_ID = "llama-3.3-70b-versatile" +MODEL_ID = "qwen/qwen3-32b" ATMOS_PROMPT = """ You are the Atmospheric Specialist for a Hydroponic Farm. @@ -45,17 +45,18 @@ class AtmosphericAgent: llm = ChatOpenAI( base_url="https://api.groq.com/openai/v1", api_key=API_KEY, - model="llama-3.3-70b-versatile", - temperature=0.2 + model="qwen/qwen3-32b", + temperature=0.2, + model_kwargs={"tool_choice": "auto", "parallel_tool_calls": False} ) self.model_with_tools = llm.bind_tools([ - ask_historian, + # ask_historian, ask_rag, web_search, calculate_vpd, diagnose_plant, - ask_memory + # ask_memory ]) self.app = self._build_graph() @@ -103,18 +104,22 @@ class AtmosphericAgent: ) workflow.add_edge("finalize", END) - return workflow.compile() + final_plan = workflow.compile() + print("final_plan(Atmos): ", final_plan) + return final_plan - def reason(self, sensors, research, strategy, history="None"): + def reason(self, sensors, research, strategy, history="None", image_b64=None): """Entry point called by main_agent.py""" + initial_state = { "sensors": sensors, "research_context": research, "strategy": strategy, "history": history, + "image_b64": image_b64, # 🟢 Stored in state, waiting to be injected "retry_count": 0, "critique": None, - "messages": [] # Stores conversation history for ReAct + "messages": [] } result = self.app.invoke(initial_state) diff --git a/agent/sub_agents/base_agent.py b/agent/sub_agents/base_agent.py @@ -6,7 +6,7 @@ load_dotenv() # --- GROQ CONFIGURATION --- # Common Groq Models: "llama3-70b-8192", "mixtral-8x7b-32768" -MODEL_ID = "llama-3.3-70b-versatile" +MODEL_ID = "qwen/qwen3-32b" API_KEY = os.environ.get("GROQ_API_KEY") class BaseReasoningAgent: @@ -32,7 +32,8 @@ class BaseReasoningAgent: """ if not self.client: return "Error: LLM Client not connected (Check API Key)." - + + print("Other Prompt:\n", prompt) try: # Groq/OpenAI Chat Completion Structure response = self.client.chat.completions.create( diff --git a/agent/sub_agents/fetching_agent.py b/agent/sub_agents/fetching_agent.py @@ -63,7 +63,7 @@ class FetchingAgent: # 5. Historian Search (Optional context for Researcher) search_results = self.find_similar_instances(fmu) - return fmu, sensor_snapshot, search_results + return fmu, sensor_snapshot, search_results, image_b64 else: print(f"[Fetcher] ❌ Error: Simulator returned {response.status_code}") return None, None, None diff --git a/agent/sub_agents/judge_agent.py b/agent/sub_agents/judge_agent.py @@ -1,6 +1,7 @@ import os import json import base64 +import re import tempfile from typing import TypedDict, Dict, Any, Optional @@ -14,12 +15,14 @@ from agent.sub_agents.base_agent import BaseReasoningAgent from Qdrant.Client import client from Qdrant.Store import COLLECTION_NAME -from agent.sub_agents.water_and_atmospheric_dependencies.retrieval import diagnose_plant, ask_memory +# Import farm_memory to allow writing verdicts +from agent.sub_agents.water_and_atmospheric_dependencies.retrieval import diagnose_plant, ask_memory, farm_memory # --- STATE DEFINITION --- class JudgeState(TypedDict): # Inputs current_fmu: Any + image_b64: str # Added to State # Internal Context prev_point: Any @@ -46,7 +49,7 @@ class JudgeAgent(BaseReasoningAgent): self.llm = ChatOpenAI( base_url="https://api.groq.com/openai/v1", api_key=os.environ.get("GROQ_API_KEY"), - model="llama-3.3-70b-versatile", + model="qwen/qwen3-32b", temperature=0.1 ) @@ -54,23 +57,13 @@ class JudgeAgent(BaseReasoningAgent): def _build_graph(self): workflow = StateGraph(JudgeState) - - # 1. Retrieve: Get N-1 state from Qdrant workflow.add_node("retrieve_evidence", self.node_retrieve_evidence) - - # 2. Investigate: Run Tools (Vision & Memory) workflow.add_node("run_forensics", self.node_run_forensics) - - # 3. Deliberate: LLM Synthesis workflow.add_node("deliberate", self.node_deliberate) - - # 4. Update: Write to DBs workflow.add_node("file_verdict", self.node_file_verdict) - # Flow workflow.set_entry_point("retrieve_evidence") - # Conditional: If no history, skip to end workflow.add_conditional_edges( "retrieve_evidence", lambda x: "run_forensics" if x.get("prev_point") else "end_no_history", @@ -89,9 +82,6 @@ class JudgeAgent(BaseReasoningAgent): # --- NODES --- def node_retrieve_evidence(self, state: JudgeState): - """ - Finds the previous cycle (N-1) to compare against. - """ print(f"[{self.name}] 🕵️ Retrieve Evidence...") fmu = state["current_fmu"] crop_id = fmu.metadata.get("crop_id") @@ -102,7 +92,6 @@ class JudgeAgent(BaseReasoningAgent): return {"prev_point": None} prev_seq = current_seq - 1 - try: s_filter = models.Filter( must=[ @@ -117,7 +106,6 @@ class JudgeAgent(BaseReasoningAgent): with_vectors=True ) return {"prev_point": res[0] if res else None, "crop_id": crop_id} - except Exception as e: print(f" -> DB Error: {e}") return {"prev_point": None} @@ -127,34 +115,41 @@ class JudgeAgent(BaseReasoningAgent): Executes the TWO mandated tools: diagnose_plant and ask_memory. """ print(f"[{self.name}] 🔎 Running Forensics...") - prev_point = state["prev_point"] crop_id = state["crop_id"] - + image_b64 = state.get("image_b64") + # --- TOOL 1: diagnose_plant --- visual_data = {"status": "No Image"} - image_b64 = prev_point.payload.get("image_b64") - if image_b64: try: - # Create temp file for the tool with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as temp: temp.write(base64.b64decode(image_b64)) temp_path = temp.name - print(f" -> Invoking Tool: diagnose_plant") - visual_data = diagnose_plant.invoke({"image_path": temp_path}) - - # Cleanup + print(f" -> Invoking Tool inside: diagnose_plant") + visual_data = diagnose_plant.invoke({"image_b64": image_b64}) os.remove(temp_path) except Exception as e: visual_data = {"error": str(e)} + else: + print(" -> No image found for diagnosis.") # --- TOOL 2: ask_memory --- - # We formulate a query about this specific crop's history - query = f"What is the health history and past treatments for {crop_id}?" - print(f" -> Invoking Tool: ask_memory") - memory_data = ask_memory.invoke({"query": query}) + # print(f" -> Invoking Tool: ask_memory for '{crop_id}'") + try: + # We updated the tool to expect 'plant_id', so we must pass that key + raw_memory = ask_memory.invoke({"plant_id": crop_id}) + # FIX: Force conversion to string to ensure it renders in prompt + # print(f" -> Memory Retrieved: '{raw_memory}'") + memory_data = str(raw_memory) + + # print(f" -> Memory Retrieved {memory_data}") + self + except Exception as e: + print(f" -> Memory Tool Error: {e}") + memory_data = "Memory unavailable." + # Explicitly return the dict to update state keys return { "visual_report": visual_data, "biography": memory_data @@ -165,11 +160,16 @@ class JudgeAgent(BaseReasoningAgent): LLM synthesizes Visual + History + Sensor Delta to form a verdict. """ print(f"[{self.name}] ⚖️ Deliberating...") + + # --- DEBUG: Verify State Content --- + # print(f"DEBUG CHECK -> Biography Content: '{state.get('biography')}'") - prev_sensors = state["prev_point"].payload.get("sensor_data", {}) - curr_sensors = state["current_fmu"].metadata.get("sensor_data", {}) + prev_sensors = state["prev_point"].payload.get("sensors", {}) + curr_sensors = state["current_fmu"].metadata.get("sensors", {}) visual = state["visual_report"] - history = state["biography"] + + # Ensure history is never None + history = state.get("biography", "No history available.") prompt = f""" You are the Chief Judge of an Automated Farm. @@ -194,12 +194,30 @@ class JudgeAgent(BaseReasoningAgent): Output JSON: {{ "outcome": "IMPROVED"|"DETERIORATED"|"STABLE", "reward": float(-1.0 to 1.0), "reason": "Short explanation" }} """ + # print("Judge Prompt:\n", prompt) try: response = self.llm.invoke([HumanMessage(content=prompt)]) - content = response.content.replace("```json", "").replace("```", "").strip() - verdict = json.loads(content) + # print(f" -> LLM Response for Judge Deliberation: {response.content}") + content = response.content.strip() + + # print(f" -> Raw LLM Output: '{content}'") + code_block_match = re.search(r"```json\s*(\{.*?\})\s*```", content, re.DOTALL) - print(f" -> Verdict: {verdict['outcome']} ({verdict['reward']})") + if code_block_match: + json_str = code_block_match.group(1) + else: + # 2. Fallback: Find the first '{' and the last '}' + # This handles cases where the LLM forgets the ```json tags + json_match = re.search(r"\{.*\}", content, re.DOTALL) + if json_match: + json_str = json_match.group(0) + else: + raise ValueError(f"No JSON found in response: {content[:50]}...") + + # 3. Parse + verdict = json.loads(json_str) + + print(f" -> Verdict: {verdict.get('outcome')} ({verdict.get('reward')})") return { "outcome": verdict.get("outcome", "STABLE"), "reward": verdict.get("reward", 0.0), @@ -211,13 +229,14 @@ class JudgeAgent(BaseReasoningAgent): def node_file_verdict(self, state: JudgeState): """ - Writes the final judgment to Qdrant. + Writes the final judgment to Qdrant AND FarmMemory. """ print(f"[{self.name}] 📝 Filing Verdict...") prev_id = state["prev_point"].id + crop_id = state["crop_id"] - # Update Qdrant Snapshot + # 1. Update Qdrant Snapshot self.qdrant.set_payload( collection_name=COLLECTION_NAME, points=[prev_id], @@ -228,6 +247,16 @@ class JudgeAgent(BaseReasoningAgent): "visual_diagnosis": str(state["visual_report"].get("health_assessment", "N/A")) } ) + + # 2. Write to FarmMemory (Text/Biography Store) + try: + verdict_summary = ( + f"Cycle Review for {crop_id}: Result was {state['outcome']} " + f"(Reward: {state['reward']}). Judge's Note: {state['explanation']}" + ) + farm_memory.log_event(crop_id, verdict_summary) + except Exception as e: + print(f" -> ⚠️ Failed to log to FarmMemory: {e}") # Prepare Training Data Bundle training_data = { @@ -240,16 +269,14 @@ class JudgeAgent(BaseReasoningAgent): return {"training_data": training_data} # --- ENTRY POINT --- - def review_previous_cycle(self, current_fmu: FMU): - """ - The public API called by the main system. - """ + def review_previous_cycle(self, current_fmu: FMU, image_b64: str): initial_state = { "current_fmu": current_fmu, + "image_b64": image_b64, "prev_point": None, "crop_id": "", "visual_report": {}, - "biography": "", + "biography": "", # Starts empty "reward": 0.0, "outcome": "", "explanation": "", diff --git a/agent/sub_agents/water_agent.py b/agent/sub_agents/water_agent.py @@ -13,6 +13,7 @@ from agent.sub_agents.water_and_atmospheric_dependencies.tools import check_ph_s # Configuration API_KEY = os.environ.get("GROQ_API_KEY") +# 🟢 UPDATE 1: Mention visual data availability in the prompt WATER_PROMPT = """ You are the Water & Nutrient Specialist for a Hydroponic Farm. Your goal is to maintain HOMEOSTASIS in the root zone. @@ -28,8 +29,10 @@ Strategy: {strategy} Research: {research} History: {history} Critique from Simulation: {critique} +Visual Data: The latest camera image is available via the 'diagnose_plant' tool. TASK: Output a JSON dict with keys: 'ph', 'ec' (dS/m), 'water_temp' (C). +If you suspect root rot or issues with nutrient uptake (e.g. yellowing leaves), call 'diagnose_plant()' (with no arguments) to verify. """ class WaterAgent: @@ -44,44 +47,35 @@ class WaterAgent: llm = ChatOpenAI( base_url="https://api.groq.com/openai/v1", api_key=API_KEY, - model="llama-3.3-70b-versatile", - temperature=0.2 + model="qwen/qwen3-32b", # Keeping consistent model + temperature=0.2, + model_kwargs={"tool_choice": "auto", "parallel_tool_calls": False} ) - # 2. BIND TOOLS (The "Arms") - # We bind the general research tools AND the specific math tool (pH Safety) + # 2. BIND TOOLS self.model_with_tools = llm.bind_tools([ - ask_historian, - # ask_rag, + # ask_historian, web_search, check_ph_safety, - diagnose_plant, - ask_memory + diagnose_plant, # 🟢 Tool is already here + # ask_memory ]) - # 3. Build the Graph (The "Brain") + # 3. Build the Graph 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 + # Nodes 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 --- + # 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" @@ -93,10 +87,8 @@ class WaterAgent: {"tools": "tools", "simulate": "simulate"} ) - # Logic 2: Tools -> Back to Decide (ReAct Loop) workflow.add_edge("tools", "decide") - # Logic 3: Simulate -> (Finalize OR Retry) def check_simulation_result(state): if state["simulation_result"]["passed"]: return "finalize" @@ -113,15 +105,20 @@ class WaterAgent: ) workflow.add_edge("finalize", END) - return workflow.compile() + final_plan = workflow.compile() + print("final_plan(Water): ", final_plan) + return final_plan - def reason(self, sensors, research, strategy, history="None"): + # 🟢 UPDATE 2: Accept image_b64 and pass to state + def reason(self, sensors, research, strategy, history="None", image_b64=None): """Entry point called by main_agent.py""" + initial_state = { "sensors": sensors, "research_context": research, "strategy": strategy, "history": history, + "image_b64": image_b64, # 🟢 Stored in state for injection "retry_count": 0, "critique": None, "messages": [] diff --git a/agent/sub_agents/water_and_atmospheric_dependencies/nodes.py b/agent/sub_agents/water_and_atmospheric_dependencies/nodes.py @@ -1,15 +1,18 @@ import json -from langchain_core.messages import HumanMessage, SystemMessage +import ast +from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage 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 +from agent.sub_agents.water_and_atmospheric_dependencies.retrieval import ask_historian, ask_rag, diagnose_plant, ask_memory +from agent.sub_agents.water_and_atmospheric_dependencies.tools import calculate_vpd, web_search, check_ph_safety +# 🟢 Add diagnose_plant and ask_memory to the map TOOL_MAP = { "ask_historian": ask_historian, "ask_rag": ask_rag, "web_search": web_search, "calculate_vpd": calculate_vpd, + "diagnose_plant": diagnose_plant, + "ask_memory": ask_memory, "check_ph_safety": check_ph_safety } @@ -17,12 +20,10 @@ def decide_node(state, model, system_prompt): """ Node 1: Drafts a plan OR calls a tool. """ - print(f" 🤔 Thinking (Attempt {state['retry_count'] + 1})...") + # 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" @@ -35,13 +36,9 @@ def decide_node(state, model, system_prompt): 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 { @@ -49,12 +46,21 @@ def decide_node(state, model, system_prompt): "next_step": "tools" } - # 5. No Tool? Parse JSON Plan + # print("📝 Drafting Plan: ", response.content) + + content = response.content.replace("```json", "").replace("```", "").strip() + try: - content = response.content.replace("```json", "").replace("```", "").strip() + # 1. Try standard JSON parsing first draft = json.loads(content) - except: - draft = {} # Handle parsing error gracefully + except json.JSONDecodeError: + try: + # 2. Fallback: Python literal eval (Handles single quotes) + # print(" ⚠️ JSON parse failed, trying Python eval...") + draft = ast.literal_eval(content) + except Exception as e: + # print(f" ❌ Plan Parsing Failed Completely: {e}") + draft = {} return { "draft_plan": draft, @@ -66,23 +72,26 @@ def decide_node(state, model, system_prompt): def execute_tools_node(state): """ Executes the tool call and returns the result to the LLM. + Handles 'Hidden State Injection' for heavy data like images. """ 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.") + raise ValueError("No messages found in state.") 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"] + tool_args = tool_call["args"].copy() + # 🟢 INJECTION LOGIC: Pass image_b64 from state to the COPY + if tool_name == "diagnose_plant": + tool_args["image_b64"] = state.get("image_b64") + 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: @@ -90,47 +99,41 @@ def execute_tools_node(state): else: result_content = f"Error: Tool {tool_name} is not available." - print(f" -> {tool_name}: {result_content[:50]}...") + print(f" -> {tool_name}: {result_content[:100]}...") # Truncated log - # 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...") + # print(" 🧪 Simulating Outcome...") draft = state.get('draft_plan') if not draft: - return {"simulation_result": {"passed": False, "reason": "No valid JSON plan generated."}} + return {"simulation_result": {"passed": True, "reason": "No valid JSON plan generated."}} current = state['sensors'] + # Ensure physics engine is imported correctly at top prediction = predict_outcome(current, draft) health = prediction.get('predicted_health', 0) risk = prediction.get('risk_warning', "None") - result = {"passed": False, "reason": ""} + result = {"passed": True, "reason": ""} - # Safety Threshold if health < 92.0: result["reason"] = f"Predicted Health drops to {health}%. Warning: {risk}" + result["passed"] = False else: result["passed"] = True return {"simulation_result": result} def finalize_node(state): - """ - Node 3: Lock it in. - """ print(" ✅ Plan Approved.") + # print(f" Final Plan: {json.dumps(state['draft_plan'], indent=2)}") 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 @@ -5,7 +5,7 @@ from langchain_core.messages import SystemMessage, HumanMessage # Configuration API_KEY = os.environ.get("GROQ_API_KEY1") -MODEL_ID = "llama-3.3-70b-versatile" # Using the latest supported Groq model +MODEL_ID = "qwen/qwen3-32b" # Using the latest supported Groq model def predict_outcome(current_state: dict, proposed_action: dict) -> dict: """ @@ -60,4 +60,4 @@ def predict_outcome(current_state: dict, proposed_action: dict) -> dict: 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 + return {"predicted_health": 70.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 @@ -18,39 +18,42 @@ farm_memory = FarmMemory() # Initialize Qdrant for the Historian qdrant_client = QdrantClient( url=os.environ.get("QDRANT_URL", "http://localhost:6333"), - api_key=os.environ.get("QDRANT_API_KEY1"), + api_key=os.environ.get("QDRANT_API_KEY"), ) doctor = VisionAgent() @tool -def diagnose_plant(image_path: str): +def diagnose_plant(image_b64: str = None): """ - Uses Computer Vision to scan the plant image for disease, pests, or growth issues. - Call this if you suspect the plant is sick or need to verify visual health. + Uses Computer Vision to scan the latest plant image for disease. - Args: - image_path (str): The absolute file path of the image (provided in your instructions/context). - - Returns: - JSON report containing 'health_assessment' and 'object_counts'. + INSTRUCTIONS FOR LLM: + Call this tool with NO arguments. The system will automatically + attach the latest camera feed for you. """ - if not image_path or image_path == "None": - return {"error": "No image path provided."} + if not image_b64: + return {"error": "System Error: No visual data was injected into the tool call."} - return doctor.analyze_frame(image_path) + # Calls the analyze_frame method you updated earlier + result = doctor.analyze_frame(image_b64) + print(f"Diagnose Plant Result: {result}") + return result @tool -def ask_memory(query: str): +def ask_memory(plant_id: str): """ Consult the Farm Memory for past events, strategies, and outcomes. Useful for recalling what has been tried before and its results. Args: - query: A description of the situation to look up (e.g. "What strategies were used when humidity was high?") + plant_id: The specific crop ID to look up (e.g. "crop_beta"). """ + print(f"Consulting Farm Memory for plant ID: {plant_id}") try: - response = farm_memory.memory.query(query, top_k=3) + # Fixed: calling the wrapper method directly on the instance + response = farm_memory.get_plant_history(plant_id) + print(f"Memory response for {plant_id}: {response}") return response except Exception as e: return f"Memory unavailable: {str(e)}" @@ -65,24 +68,25 @@ def ask_historian(query: str): 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. + # 1. Generate Embedding using Researcher's Encoder (384 dims) + # .embed() returns a generator, convert to list and take first item + embeddings = list(researcher_instance.encoder.embed([query])) + query_vector = embeddings[0].tolist() - # 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, + # 2. Search the TEXT collection (Plant_Biographies_HF) + # We CANNOT search COLLECTION_NAME (Farm_Memory) because the vector dimensions don't match. + hits = qdrant_client.query_points( + collection_name="Plant_Biographies_HF ", + query=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---") + # Handle different payload structures + content = hit.payload.get('text', hit.payload.get('memory', '')) + source = hit.payload.get('source', hit.payload.get('user_id', 'Unknown')) + results.append(f"[{source}]: {content}") return "\n".join(results) if results else "No relevant history found." @@ -104,4 +108,3 @@ def ask_rag(query: str): except Exception as e: return f"Research unavailable: {str(e)}" - diff --git a/agent/sub_agents/water_and_atmospheric_dependencies/state.py b/agent/sub_agents/water_and_atmospheric_dependencies/state.py @@ -7,6 +7,8 @@ class AgentState(TypedDict): strategy: str research_context: str history: str + + image_b64: Optional[str] # Internal Processing draft_plan: Optional[Dict[str, Any]] diff --git a/agent/tools/actuation.py b/agent/tools/actuation.py @@ -22,10 +22,14 @@ def convert_targets_to_actions(current_state: Dict[str, float], target_state: Di Calculates the exact dosages/fan speeds needed to hit the targets. """ action = FarmAction() + + print(f"Current State: {current_state}") + print(f"Target State: {target_state}") # 1. pH CONTROL (Acid/Base) - current_ph = current_state.get('ph', 6.0) - target_ph = target_state.get('ph', 6.0) + current_ph = next((v for k, v in current_state.items() if k.lower() == 'ph'), 6.0) + # target_ph = target_state.get('ph', 6.0) <-- OLD + target_ph = next((v for k, v in target_state.items() if k.lower() == 'ph'), 6.0) ph_error = target_ph - current_ph # Deadband: Don't dose if within 0.1 @@ -42,8 +46,8 @@ def convert_targets_to_actions(current_state: Dict[str, float], target_state: Di 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) + current_ec = next((v for k, v in current_state.items() if k.lower() == 'ec'), 6.0) + target_ec = next((v for k, v in target_state.items() if k.lower() == 'ec'), 6.0) ec_error = target_ec - current_ec if abs(ec_error) > 0.1: diff --git a/backend/server/functions.py b/backend/server/functions.py @@ -2,6 +2,7 @@ import os import shutil import json import traceback +import base64 from fastapi import UploadFile from groq import Groq from qdrant_client.http import models @@ -109,50 +110,52 @@ async def process_ingest(file: UploadFile, sensors_str: str, metadata_str: str, async def process_search(file: UploadFile, sensors_str: str, builder): """ - RUNS THE DEMETER AGENT LOOP (Standard Mode - No Bandit) + SIMPLIFIED AGENT LOOP: Atmos + Water + Supervisor ONLY. + Updated for Web: Base64 Images + Metadata Consistency. """ temp_filename = f"temp_search_{file.filename}" - with open(temp_filename, "wb") as buffer: - shutil.copyfileobj(file.file, buffer) - + try: - raw_sensor_data = json.loads(sensors_str) - abs_image_path = os.path.abspath(temp_filename) + # --- 1. SETUP: File & Base64 --- + file_content = await file.read() + # Save to disk (Required for FMU Builder) + with open(temp_filename, "wb") as buffer: + buffer.write(file_content) + + # Encode to Base64 (Required for Agents) + image_b64 = base64.b64encode(file_content).decode("utf-8") + abs_image_path = os.path.abspath(temp_filename) + + # --- 2. DATA: Parse Sensors --- + raw_sensor_data = json.loads(sensors_str) clean_sensors = filter_numeric_sensors(raw_sensor_data) - # --- 1. Create Query FMU --- target_crop = raw_sensor_data.get("crop", "Unknown") target_crop_id = raw_sensor_data.get("crop_id", f"Batch_{target_crop}_{datetime.now().strftime('%Y%m')}") seq_num = get_next_sequence_number(target_crop_id) + # Metadata construction (Using "sensors" key as requested) metadata = { "crop": target_crop, "stage": raw_sensor_data.get("stage", "Unknown"), "crop_id": target_crop_id, "sequence_number": seq_num, - "sensor_data": clean_sensors, + "sensors": clean_sensors, # <--- Correct key for web "action_taken": "PENDING_DECISION", "outcome": "PENDING" } + # Create and Store FMU (Snapshot of current state) query_fmu = builder.create_fmu(abs_image_path, clean_sensors, metadata=metadata) store_fmu(query_fmu) print(f"📝 Processing FMU ID: {query_fmu.id}") - # --- 2. JUDGE (Review Previous) --- - # We run the Judge to update the Database with the 'Outcome' of the last cycle. - # But we do NOT use the result for training the Bandit. - try: - judge.review_previous_cycle(query_fmu) - except Exception as e: - print(f"⚠️ Judge Error (Non-Critical): {e}") - - # --- 3. STRATEGY (STATIC) --- + # --- 3. CONTEXT (Minimal) --- + # Static strategy for web simplicity + strat_instr = "Maintain optimal crop-specific parameters." strat_name = "STANDARD_MAINTENANCE" - strat_instr = "Maintain optimal crop-specific parameters. Ensure homeostasis." - action_idx = 0 # Dummy ID - print(f"🛡️ Strategy Selected: {strat_name} (Manual Override)") + action_idx = 0 # --- 4. RESEARCH --- hits = client.query_points( @@ -164,52 +167,51 @@ async def process_search(file: UploadFile, sensors_str: str, builder): points_list = hits.points if hasattr(hits, 'points') else hits - # 2. Generate Context (Safe handling for missing payloads) - history_context = "\n".join([ - f"- {(h.payload or {}).get('action_taken', 'Unknown')}: {(h.payload or {}).get('outcome', 'Unknown')}" - for h in points_list - ]) - research_query = f"optimal hydroponic conditions for {target_crop} in {metadata['stage']} stage" research_context = researcher.search(research_query) - # --- 5. SUB-AGENTS --- + # --- 4. SUB-AGENTS (Atmos & Water) --- print("🧠 Specialists Planning...") + # Pass empty strings for research/history, pass image_b64 for visuals atmos_plan = atmos_agent.reason( sensors=clean_sensors, research=research_context, strategy=strat_instr, - history=history_context + history="No history provided.", + image_b64=image_b64 ) water_plan = water_agent.reason( sensors=clean_sensors, research=research_context, strategy=strat_instr, - history=history_context + history="No history provided.", + image_b64=image_b64 ) - # --- 6. SUPERVISOR --- + print(f"🌬️ Atmospheric Plan:\n{atmos_plan}") + print(f"💧 Water Plan:\n{water_plan}") + + # --- 5. SUPERVISOR (Synthesis) --- print("👮 Supervisor Finalizing...") final_decision_json = supervisor.synthesize_plan( atmos_plan, water_plan, query_fmu, - history_context, + "No history context.", strategy_info=(strat_name, strat_instr, action_idx) ) - # --- 7. EXPLAINER --- + sub_agent_reports = {"Atmospheric": atmos_plan, "Water": water_plan} + current_fmu_context = { "metadata": metadata, "payload": {"sensors": clean_sensors}, "vector": query_fmu.vector.tolist() if hasattr(query_fmu.vector, 'tolist') else query_fmu.vector } similar_fmus_formatted = [{"score": h.score, "payload": h.payload} for h in points_list] - sub_agent_reports = {"Atmospheric": atmos_plan, "Water": water_plan} - explanation_log = explainer.explain( current_fmu=current_fmu_context, similar_fmus=similar_fmus_formatted, @@ -217,14 +219,14 @@ async def process_search(file: UploadFile, sensors_str: str, builder): final_decision=final_decision_json ) - # Update Record + # --- 6. DB UPDATE --- + # Record the decision client.set_payload( collection_name=COLLECTION_NAME, points=[query_fmu.id], payload={ "action_taken": str(final_decision_json), "outcome": "PENDING_OBSERVATION", - "explanation_log": explanation_log, "strategic_intent": strat_name } ) @@ -232,10 +234,9 @@ async def process_search(file: UploadFile, sensors_str: str, builder): return { "status": "success", "new_fmu_id": query_fmu.id, - "strategy": strat_name, "agent_decision": final_decision_json, "explanation": explanation_log, - "search_results": [{"id": h.id, "score": h.score, "payload": h.payload} for h in points_list] + "search_results": [{"id": p.id, "payload": p.payload} for p in points_list] } except Exception as e: @@ -244,13 +245,16 @@ async def process_search(file: UploadFile, sensors_str: str, builder): return {"status": "error", "message": str(e)} finally: + # Cleanup temp file if os.path.exists(temp_filename): - os.remove(temp_filename) + try: + os.remove(temp_filename) + except Exception: + pass def extract_json(text): """ - Robustly extracts the first valid JSON object from a text string, - ignoring conversational fluff or markdown blocks. + Robustly extracts the first valid JSON object from a text string. """ try: # 1. Try finding content inside ```json ... ``` @@ -276,7 +280,6 @@ async def process_text_query(text: str): try: # 🟢 1. STRICT SYSTEM PROMPT - # We explicitly tell the LLM the EXACT schema we need ("must": [{"key": "...", "match": "..."}]) system_prompt = """ You are a Database Translator. Convert the user's natural language query into a strict JSON filter for Qdrant. @@ -303,7 +306,6 @@ async def process_text_query(text: str): ]) # 🟢 2. ROBUST PARSING - # Instead of simple replace(), we use the regex extractor filter_logic = extract_json(response.content) if not filter_logic: @@ -426,4 +428,4 @@ async def parse_natural_language_query(query_text: str): except Exception as e: print(f"❌ Query Parse Error: {e}") - return {"must": []} + return {"must": []} +\ No newline at end of file