demeter
Autonomous Hydroponic Intelligencegit clone https://github.com/maydayv7/demeter
backend/server/functions.py (29927B)
1import sys
2import os
3import re
4import json
5import base64
6import shutil
7import asyncio
8import traceback
9from datetime import datetime
10from fastapi import UploadFile, HTTPException, Form
11from qdrant_client import models
12from langchain_core.messages import SystemMessage, HumanMessage
13
14from Qdrant.Store import store_fmu, COLLECTION_NAME
15from Qdrant.Client import client
16
17# 🛡️ GUARDRAILS
18from agent.guardrails.validation import sanitize_input
19
20# Import Agent instances
21from agent.sub_agents.fetching_agent import FetchingAgent
22from agent.sub_agents.judge_agent import JudgeAgent
23from agent.sub_agents.atmospheric_agent import AtmosphericAgent
24from agent.sub_agents.water_agent import WaterAgent
25from agent.sub_agents.Supervisor import SupervisorAgent
26from agent.sub_agents.Researcher import ResearcherAgent
27from agent.sub_agents.Explainer import ExplainerAgent
28
29# Global singletons to avoid re-initializing heavy models per request
30fetcher = FetchingAgent()
31judge = JudgeAgent()
32atmos_agent = AtmosphericAgent()
33water_agent = WaterAgent()
34researcher = ResearcherAgent()
35supervisor = SupervisorAgent(researcher_agent=researcher)
36explainer = ExplainerAgent()
37
38
39def filter_numeric_sensors(raw_sensors: dict):
40 wanted = {"pH", "EC", "temp", "humidity"}
41 clean = {}
42 for k, v in raw_sensors.items():
43 if k in wanted:
44 try:
45 clean[k] = float(v)
46 except:
47 pass
48 return clean
49
50
51def get_next_sequence_number(crop_id: str) -> int:
52 try:
53 count_filter = models.Filter(
54 must=[
55 models.FieldCondition(
56 key="crop_id", match=models.MatchValue(value=crop_id)
57 )
58 ]
59 )
60 count_result = client.count(
61 collection_name=COLLECTION_NAME, count_filter=count_filter
62 )
63 return count_result.count + 1
64 except:
65 return 1
66
67
68# --- ENDPOINTS ---
69async def process_ingest(
70 file: UploadFile, sensors_str: str, metadata_str: str, builder
71):
72 """
73 Handles file saving, FMU creation, and storage logic.
74 """
75 temp_filename = f"temp_{file.filename}"
76 with open(temp_filename, "wb") as buffer:
77 shutil.copyfileobj(file.file, buffer)
78
79 try:
80 raw_sensor_data = json.loads(sensors_str)
81 meta_data = json.loads(metadata_str)
82 abs_image_path = os.path.abspath(temp_filename)
83
84 clean_sensors = filter_numeric_sensors(raw_sensor_data)
85
86 # 1. Identity Logic
87 target_crop = meta_data.get("crop", "Unknown")
88 target_crop_id = meta_data.get("crop_id") or raw_sensor_data.get("crop_id")
89 if not target_crop_id:
90 target_crop_id = f"Batch_{target_crop}_{datetime.now().strftime('%Y%m')}"
91
92 seq_num = get_next_sequence_number(target_crop_id)
93 print(f"📥 Ingesting {target_crop_id} | Snapshot #{seq_num}")
94
95 # 2. Metadata Injection
96 meta_data.update(
97 {
98 "crop_id": target_crop_id,
99 "sequence_number": seq_num,
100 "sensor_data": clean_sensors,
101 "action_taken": meta_data.get("action_taken", "PENDING_ACTION"),
102 "outcome": meta_data.get("outcome", "PENDING_OBSERVATION"),
103 }
104 )
105
106 # 3. Store
107 fmu = builder.create_fmu(abs_image_path, clean_sensors, meta_data)
108 store_fmu(fmu)
109
110 return {"status": "success", "fmu_id": fmu.id}
111 finally:
112 if os.path.exists(temp_filename):
113 os.remove(temp_filename)
114
115
116async def process_search(file: UploadFile, sensors_str: str, builder):
117 """
118 SIMPLIFIED AGENT LOOP: Atmos + Water + Supervisor ONLY.
119 """
120 temp_filename = f"temp_search_{file.filename}"
121 try:
122 # --- 1. SETUP: File & Base64 ---
123 file_content = await file.read()
124
125 # Save to disk (Required for FMU Builder)
126 with open(temp_filename, "wb") as buffer:
127 buffer.write(file_content)
128
129 # Encode to Base64 (Required for Agents)
130 image_b64 = base64.b64encode(file_content).decode("utf-8")
131 abs_image_path = os.path.abspath(temp_filename)
132
133 # --- 2. DATA: Parse Sensors ---
134 raw_sensor_data = json.loads(sensors_str)
135 clean_sensors = filter_numeric_sensors(raw_sensor_data)
136
137 target_crop = raw_sensor_data.get("crop", "Unknown")
138 target_crop_id = raw_sensor_data.get("crop_id")
139 if not target_crop_id:
140 target_crop_id = f"Batch_{target_crop}_{datetime.now().strftime('%Y%m')}"
141
142 seq_num = get_next_sequence_number(target_crop_id)
143
144 # Metadata construction
145 metadata = {
146 "crop": target_crop,
147 "stage": raw_sensor_data.get("stage")
148 or raw_sensor_data.get("metadata", {}).get("stage", "seedling"),
149 "crop_id": target_crop_id,
150 "sequence_number": seq_num,
151 "sensors": clean_sensors,
152 "action_taken": "PENDING_DECISION",
153 "outcome": "PENDING",
154 }
155
156 # Create and Store FMU (Snapshot of current state)
157 query_fmu = builder.create_fmu(abs_image_path, clean_sensors, metadata=metadata)
158 store_fmu(query_fmu)
159 print(f"📝 Processing FMU ID: {query_fmu.id}")
160
161 # --- 3. CONTEXT (Minimal) ---
162 # Static strategy for web simplicity
163 strat_instr = "Maintain optimal crop-specific parameters."
164 strat_name = "STANDARD_MAINTENANCE"
165 action_idx = 0
166
167 # --- 4. RESEARCH ---
168 hits = client.query_points(
169 collection_name=COLLECTION_NAME,
170 query=query_fmu.vector,
171 limit=3,
172 with_payload=True,
173 )
174 points_list = hits.points if hasattr(hits, "points") else hits
175
176 research_query = f"optimal hydroponic conditions for {target_crop} in {metadata['stage']} stage"
177 research_context = researcher.search(research_query)
178
179 # --- 4. SUB-AGENTS (Atmos & Water) ---
180 print("🧠 Specialists Planning...")
181
182 # Pass empty strings for research/history, pass image_b64 for visuals
183 atmos_plan = atmos_agent.reason(
184 sensors=clean_sensors,
185 research=research_context,
186 strategy=strat_instr,
187 history="No history provided.",
188 image_b64=image_b64,
189 )
190
191 water_plan = water_agent.reason(
192 sensors=clean_sensors,
193 research=research_context,
194 strategy=strat_instr,
195 history="No history provided.",
196 image_b64=image_b64,
197 )
198
199 print(f"🌬️ Atmospheric Plan:\n{atmos_plan}")
200 print(f"💧 Water Plan:\n{water_plan}")
201
202 # --- 5. SUPERVISOR (Synthesis) ---
203 print("👮 Supervisor Finalizing...")
204 final_decision_json = supervisor.synthesize_plan(
205 atmos_plan,
206 water_plan,
207 query_fmu,
208 "No history context.",
209 strategy_info=(strat_name, strat_instr, action_idx),
210 )
211
212 sub_agent_reports = {"Atmospheric": atmos_plan, "Water": water_plan}
213
214 current_fmu_context = {
215 "metadata": metadata,
216 "payload": {"sensors": clean_sensors},
217 "vector": (
218 query_fmu.vector.tolist()
219 if hasattr(query_fmu.vector, "tolist")
220 else query_fmu.vector
221 ),
222 }
223
224 similar_fmus_formatted = [
225 {"score": h.score, "payload": h.payload} for h in points_list
226 ]
227
228 explanation_log = explainer.explain(
229 current_fmu=current_fmu_context,
230 similar_fmus=similar_fmus_formatted,
231 sub_agent_reports=sub_agent_reports,
232 final_decision=final_decision_json,
233 )
234
235 # --- 6. DB UPDATE ---
236 # Record the decision
237 client.set_payload(
238 collection_name=COLLECTION_NAME,
239 points=[query_fmu.id],
240 payload={
241 "action_taken": str(final_decision_json),
242 "outcome": "PENDING_OBSERVATION",
243 "strategic_intent": strat_name,
244 },
245 )
246
247 return {
248 "status": "success",
249 "new_fmu_id": query_fmu.id,
250 "agent_decision": final_decision_json,
251 "explanation": explanation_log,
252 "search_results": [
253 {"id": p.id, "score": p.score, "payload": p.payload}
254 for p in points_list
255 ],
256 }
257
258 except Exception as e:
259 print(f"❌ Pipeline Error: {e}")
260 traceback.print_exc()
261 raise HTTPException(status_code=500, detail=str(e))
262 finally:
263 # Cleanup temp file
264 if os.path.exists(temp_filename):
265 try:
266 os.remove(temp_filename)
267 except Exception:
268 pass
269
270
271async def process_cycle_stream(file: UploadFile, sensors_str: str, builder):
272 """
273 REAL-TIME AGENT LOOP: Streams step-by-step reasoning via SSE.
274 """
275 temp_filename = f"temp_stream_{file.filename}"
276 try:
277 yield f"data: {json.dumps({'agent': 'SYSTEM', 'text': '🚀 Initializing Demeter Orchestrator...'})}\n\n"
278 await asyncio.sleep(0.5)
279
280 # --- 1. SETUP ---
281 file_content = await file.read()
282 with open(temp_filename, "wb") as buffer:
283 buffer.write(file_content)
284
285 image_b64 = base64.b64encode(file_content).decode("utf-8")
286 abs_image_path = os.path.abspath(temp_filename)
287
288 yield f"data: {json.dumps({'agent': 'FETCHER', 'text': '📡 Requesting data from simulator...'})}\n\n"
289 await asyncio.sleep(0.5)
290
291 # --- 2. DATA ---
292 raw_sensor_data = json.loads(sensors_str)
293 clean_sensors = filter_numeric_sensors(raw_sensor_data)
294
295 target_crop = raw_sensor_data.get("crop", "Unknown")
296 target_crop_id = raw_sensor_data.get("crop_id")
297 if not target_crop_id:
298 target_crop_id = f"Batch_{target_crop}_{datetime.now().strftime('%Y%m')}"
299
300 seq_num = get_next_sequence_number(target_crop_id)
301
302 yield f"data: {json.dumps({'agent': 'FETCHER', 'text': f'🔢 Sequence for {target_crop_id}: {seq_num}'})}\n\n"
303 await asyncio.sleep(0.3)
304
305 metadata = {
306 "crop": target_crop,
307 "stage": raw_sensor_data.get("stage")
308 or raw_sensor_data.get("metadata", {}).get("stage", "seedling"),
309 "crop_id": target_crop_id,
310 "sequence_number": seq_num,
311 "sensors": clean_sensors,
312 "action_taken": "PENDING_DECISION",
313 "outcome": "PENDING",
314 }
315
316 query_fmu = builder.create_fmu(abs_image_path, clean_sensors, metadata=metadata)
317 store_fmu(query_fmu)
318
319 yield f"data: {json.dumps({'agent': 'FETCHER', 'text': f'🧠 FMU Created (ID: {query_fmu.id}) — Handing off to specialists.'})}\n\n"
320 await asyncio.sleep(0.5)
321
322 # --- 3. RESEARCH ---
323 yield f"data: {json.dumps({'agent': 'RESEARCHER', 'text': f'🔍 Searching knowledge base for {target_crop} {metadata['stage']} stage...'})}\n\n"
324 research_query = f"optimal hydroponic conditions for {target_crop} in {metadata['stage']} stage"
325 research_context = researcher.search(research_query)
326 await asyncio.sleep(0.5)
327 yield f"data: {json.dumps({'agent': 'RESEARCHER', 'text': ' 📚 Found relevant scientific data.'})}\n\n"
328
329 # --- 3.5 JUDGE: Review previous cycle and update bandit ---
330 yield f"data: {json.dumps({'agent': 'JUDGE', 'text': '⚖️ Judge reviewing previous cycle outcome...'})}\n\n"
331 await asyncio.sleep(0.3)
332 judge_result = judge.review_previous_cycle(query_fmu, image_b64)
333
334 # --- 3.6 BANDIT LEARNING: Update model based on previous cycle outcome ---
335 if judge_result:
336 yield f"data: {json.dumps({'agent': 'SUPERVISOR', 'text': '🧠 Supervisor learning from outcome...'})}\n\n"
337 await asyncio.sleep(0.3)
338 supervisor.learn_from_outcome(query_fmu, judge_result)
339
340 await asyncio.sleep(0.3)
341
342 # --- 4. AGENTS ---
343 strat_name, strat_instr, action_idx = supervisor.get_strategic_goal(query_fmu)
344
345 yield f"data: {json.dumps({'agent': 'BANDIT', 'text': f'🎰 BANDIT STRATEGY: {strat_name}'})}\n\n"
346 await asyncio.sleep(0.3)
347
348 yield f"data: {json.dumps({'agent': 'ATMOSPHERIC', 'text': '🌬️ Atmospheric Agent — deciding...'})}\n\n"
349 atmos_plan = atmos_agent.reason(
350 sensors=clean_sensors,
351 research=research_context,
352 strategy=strat_instr,
353 history="No history provided.",
354 image_b64=image_b64,
355 )
356 yield f"data: {json.dumps({'agent': 'ATMOSPHERIC', 'text': f' ✅ Plan Approved: {json.dumps(atmos_plan)}'})}\n\n"
357 await asyncio.sleep(0.5)
358
359 yield f"data: {json.dumps({'agent': 'WATER', 'text': '💧 Water Agent — deciding...'})}\n\n"
360 water_plan = water_agent.reason(
361 sensors=clean_sensors,
362 research=research_context,
363 strategy=strat_instr,
364 history="No history provided.",
365 image_b64=image_b64,
366 )
367 yield f"data: {json.dumps({'agent': 'WATER', 'text': f' ✅ Plan Approved: {json.dumps(water_plan)}'})}\n\n"
368 await asyncio.sleep(0.5)
369
370 # --- 5. SUPERVISOR ---
371 yield f"data: {json.dumps({'agent': 'SUPERVISOR', 'text': ' 🔗 Supervisor Merging Plans...'})}\n\n"
372 await asyncio.sleep(0.3)
373 yield f"data: {json.dumps({'agent': 'SUPERVISOR', 'text': ' ⚖️ Supervisor Judging...'})}\n\n"
374
375 final_decision_json = supervisor.synthesize_plan(
376 atmos_plan,
377 water_plan,
378 query_fmu,
379 "No history context.",
380 strategy_info=(strat_name, strat_instr, action_idx),
381 )
382
383 await asyncio.sleep(0.5)
384 yield f"data: {json.dumps({'agent': 'SUPERVISOR', 'text': ' ✅ Plan looks solid.'})}\n\n"
385
386 # --- 6. FINAL ---
387 yield f"data: {json.dumps({'agent': 'SUPERVISOR', 'text': f'🚜 Activating Hardware: {json.dumps(final_decision_json)}'})}\n\n"
388 await asyncio.sleep(0.5)
389
390 yield f"data: {json.dumps({'agent': 'SYSTEM', 'text': '✅ Sent to Simulator. Cycle complete.', 'final_action': final_decision_json, 'phase': 'done'})}\n\n"
391
392 except Exception as e:
393 yield f"data: {json.dumps({'agent': 'SYSTEM', 'text': f'❌ Error: {str(e)}', 'level': 'error'})}\n\n"
394 finally:
395 if os.path.exists(temp_filename):
396 try:
397 os.remove(temp_filename)
398 except Exception:
399 pass
400
401
402def extract_json(text):
403 """
404 Robustly extracts the first valid JSON object from text string.
405 """
406 cleaned = re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL)
407 match = re.search(r"\{.*\}", cleaned, re.DOTALL)
408 if match:
409 try:
410 return json.loads(match.group(0))
411 except json.JSONDecodeError:
412 pass
413 return None
414
415
416async def process_text_query(text: str, crop_id: str = None):
417 """
418 HYBRID FILTER ENGINE:
419 Uses LLM to extract Exact/Range/Text filters
420
421 When crop_id is provided the caller has selected a specific crop, so we
422 inject a should-match for that crop_id to bias results toward it.
423 """
424
425 # 🛡️ GUARDRAIL: Check for injection attempts and off-topic queries
426 sanitized_text, violations = sanitize_input(text)
427
428 if violations:
429 print(f"⚠️ Query Security Alert:")
430 for v in violations:
431 print(f" {v}")
432
433 if len(violations) >= 3:
434 return {
435 "status": "error",
436 "message": "❌ Query blocked: Multiple security violations detected. Please ask only farm-related questions.",
437 "violations": violations
438 }
439
440 # Use sanitized input
441 text = sanitized_text
442
443 system_prompt = """
444 You are a Database Translator for an AI Hydroponic Farm.
445 Your goal: Convert natural language queries into a precise JSON filter object.
446
447 AVAILABLE METADATA FIELDS (String):
448 - crop (e.g., "Tomato", "Lettuce", "Basil")
449 - stage (e.g., "Seedling", "Vegetative", "Flowering")
450 - outcome (e.g., "Positive", "Negative")
451
452 AVAILABLE SENSOR FIELDS (Numeric):
453 - pH (float, e.g., 5.5 to 6.5)
454 - EC (float, e.g., 1.0 to 3.0)
455 - temp (float, e.g., 20.0 to 30.0)
456 - humidity (float, e.g., 40.0 to 80.0)
457
458 AVAILABLE AGENT LOGIC FIELDS (Text/Substring):
459 - action_taken (Use this to search for specific agent decisions like "FLUSH", "INCREASE_WATER", "DECREASE_NUTRIENTS")
460 - strategic_intent (e.g., "STANDARD_MAINTENANCE", "RECOVERY")
461
462 OUTPUT SCHEMA:
463 {
464 "filters": [
465 {
466 "field": "field_name",
467 "operator": "exact" | "text" | "gt" | "lt" | "gte" | "lte",
468 "value": string_or_number
469 }
470 ]
471 }
472
473 RULES:
474 1. Use "exact" for exact string matches (crop, stage, outcome, strategic_intent).
475 2. Use "text" for partial/substring matches (CRITICAL for 'action_taken' since it contains stringified JSON records).
476 3. Use "gt", "lt", "gte", "lte" for numeric sensor comparisons.
477 4. Translate queries into English (e.g., "Tamatar" -> "Tomato", "Kharab" -> "Negative").
478 5. For queries about "similar crops" or "crops like X", extract the crop name as an exact filter on 'crop'.
479
480 EXAMPLE: "Find tomato crops in vegetative stage with pH over 6.0 where the agent flushed the tank"
481 {
482 "filters": [
483 { "field": "crop", "operator": "exact", "value": "Tomato" },
484 { "field": "stage", "operator": "exact", "value": "Vegetative" },
485 { "field": "pH", "operator": "gt", "value": 6.0 },
486 { "field": "action_taken", "operator": "text", "value": "FLUSH" }
487 ]
488 }
489
490 Return ONLY valid JSON. If no filters apply, return { "filters": [] }.
491 """
492
493 try:
494 response = supervisor.model.invoke(
495 [SystemMessage(content=system_prompt), HumanMessage(content=text)]
496 )
497
498 raw_output = response.content
499 filter_logic = extract_json(raw_output)
500
501 if not filter_logic:
502 return {
503 "status": "error",
504 "message": "Failed to parse JSON filter.",
505 "query_logic": raw_output,
506 }
507
508 qdrant_conditions = []
509 post_filters = []
510
511 if "filters" in filter_logic:
512 for rule in filter_logic["filters"]:
513 field = rule.get("field")
514 op = rule.get("operator", "exact")
515 val = rule.get("value")
516
517 if not field or val is None:
518 continue
519
520 # Text substring matches (Python post-filter)
521 if op == "text":
522 post_filters.append((field, str(val).lower()))
523 continue
524
525 # Numeric sensor fields (nested routing)
526 if field.lower() in ["ph", "ec", "temp", "humidity"]:
527 field_norm = {
528 "ph": "pH",
529 "ec": "EC",
530 "temp": "temp",
531 "humidity": "humidity",
532 }.get(field.lower(), field)
533 path1 = f"sensors.{field_norm}"
534 path2 = f"sensor_data.{field_norm}"
535
536 if op == "exact":
537 qdrant_conditions.append(
538 models.Filter(
539 should=[
540 models.FieldCondition(
541 key=path1, match=models.MatchValue(value=val)
542 ),
543 models.FieldCondition(
544 key=path2, match=models.MatchValue(value=val)
545 ),
546 ]
547 )
548 )
549 else:
550 try:
551 num_val = float(val)
552 range_kwargs = {op: num_val}
553 qdrant_conditions.append(
554 models.Filter(
555 should=[
556 models.FieldCondition(
557 key=path1,
558 range=models.Range(**range_kwargs),
559 ),
560 models.FieldCondition(
561 key=path2,
562 range=models.Range(**range_kwargs),
563 ),
564 ]
565 )
566 )
567 except ValueError:
568 pass
569 else:
570 # Exact string fields
571 qdrant_conditions.append(
572 models.FieldCondition(
573 key=field, match=models.MatchValue(value=val)
574 )
575 )
576
577 # Build Qdrant filter
578 scroll_filter = (
579 models.Filter(must=qdrant_conditions) if qdrant_conditions else None
580 )
581
582 results, _ = client.scroll(
583 collection_name=COLLECTION_NAME,
584 scroll_filter=scroll_filter,
585 limit=100,
586 with_payload=True,
587 with_vectors=False,
588 )
589
590 # Python post-filter (for text/substring fields like action_taken)
591 filtered_results = []
592 for res in results:
593 payload = res.payload or {}
594 passed = True
595 for pf_field, pf_val in post_filters:
596 if pf_val not in str(payload.get(pf_field, "")).lower():
597 passed = False
598 break
599 if passed:
600 filtered_results.append(res)
601 if len(filtered_results) >= 10:
602 break
603
604 # If a specific crop was selected, sort its results to the top
605 if crop_id:
606 filtered_results.sort(
607 key=lambda p: 0 if p.payload.get("crop_id") == crop_id else 1
608 )
609
610 return {
611 "status": "success",
612 "results": [
613 {"id": str(p.id), "score": 1.0, "payload": p.payload}
614 for p in filtered_results
615 ],
616 "query_logic": filter_logic,
617 }
618
619 except Exception as e:
620 print(f"❌ Text Search Error: {e}")
621 import traceback
622
623 traceback.print_exc()
624 return {"status": "error", "message": str(e)}
625
626
627async def process_audio_search(file: UploadFile):
628 import whisper
629
630 temp_filename = f"temp_audio_{file.filename}"
631 try:
632 with open(temp_filename, "wb") as buffer:
633 shutil.copyfileobj(file.file, buffer)
634
635 model = whisper.load_model("base")
636 result = model.transcribe(temp_filename)
637 detected_text = result["text"].strip()
638
639 # This ensures we get the same RAG/Qdrant logic as text queries
640 response_data = await process_text_query(detected_text)
641
642 # Inject the transcription so the UI can show what was heard
643 response_data["transcription"] = detected_text
644
645 return response_data
646
647 except Exception as e:
648 print(f"❌ Audio Search Error: {e}")
649 return {"status": "error", "message": str(e)}
650
651 finally:
652 # Cleanup temp file
653 if os.path.exists(temp_filename):
654 os.remove(temp_filename)
655
656
657async def process_ask_query(query: str, context: str, language: str):
658 """
659 Answers natural language questions about the farm using pre-built context
660 from the frontend.
661 """
662 try:
663 # 🛡️ GUARDRAIL: Check for injection attempts and off-topic queries
664 sanitized_query, violations = sanitize_input(query)
665
666 if violations:
667 print(f"⚠️ Query Security Alert:")
668 for v in violations:
669 print(f" {v}")
670
671 if len(violations) >= 3:
672 return {
673 "status": "error",
674 "message": " Question blocked: Multiple security violations detected. Please ask only farm-related questions.",
675 "violations": violations
676 }
677
678 # Use sanitized input
679 query = sanitized_query
680
681 lang_instr = (
682 "Respond entirely in Hindi."
683 if language == "hi"
684 else "Respond entirely in English."
685 )
686
687 system_prompt = f"""You are Demeter Intelligence — an expert AI agronomist embedded in a hydroponic farm management system.
688
689ROLE:
690- Answer questions about crop health, sensor readings, agent decisions, and farm trends
691- Compare crops when asked, citing their crop_id
692- Give actionable recommendations grounded in the data
693- Be concise: lead with the direct answer, then explain
694- SCOPE: Answer ONLY farm-related questions. Politely decline off-topic queries.
695
696REASONING:
697Wrap your internal reasoning in <thinking>...</thinking> before your answer.
698Keep thinking brief — focus on which crops are relevant and what the data says.
699
700FARM DATA:
701{context}
702
703LANGUAGE: {lang_instr}"""
704
705 response = supervisor.model.invoke(
706 [SystemMessage(content=system_prompt), HumanMessage(content=query)]
707 )
708
709 raw_text = response.content
710 thinking = ""
711 answer = raw_text
712
713 # Support both <thinking> and <think> tags
714 think_match = re.search(
715 r"<think(?:ing)?>(.*?)</think(?:ing)?>", raw_text, re.DOTALL
716 )
717 if think_match:
718 thinking = think_match.group(1).strip()
719 answer = re.sub(
720 r"<think(?:ing)?>(.*?)</think(?:ing)?>", "", raw_text, flags=re.DOTALL
721 ).strip()
722
723 return {"status": "success", "thinking": thinking, "answer": answer}
724
725 except Exception as e:
726 traceback.print_exc()
727 return {"status": "error", "message": str(e)}
728
729
730async def process_similar_crops(crop_id: str, crop_name: str, payload_json: str):
731 """
732 Find cosine-similar crops using the latest stored vector for crop_id.
733 Falls back to a zero-padded sensor vector if no stored vector is found.
734 """
735 import json as _json
736 import numpy as np
737
738 try:
739 # Step 1: Scroll all points for this crop, requesting vectors
740 points, _ = client.scroll(
741 collection_name=COLLECTION_NAME,
742 scroll_filter=models.Filter(
743 must=[
744 models.FieldCondition(
745 key="crop_id",
746 match=models.MatchValue(value=crop_id),
747 )
748 ]
749 ),
750 limit=100,
751 with_payload=True,
752 with_vectors=True,
753 )
754
755 query_vector = None
756
757 if points:
758 # Pick the point with the highest sequence_number
759 best = max(points, key=lambda p: p.payload.get("sequence_number", 0))
760 v = best.vector
761 if v is not None:
762 # Handle named-vector collections (dict) vs plain list
763 if isinstance(v, dict):
764 v = next(iter(v.values()))
765 if len(v) == 516:
766 query_vector = list(v)
767 else:
768 print(
769 f"[SimilarCrops] Unexpected vector length {len(v)} for {crop_id}"
770 )
771
772 # Step 2: Sensor-only fallback — build a 516-dim vector
773 # Vision dims (0–511) stay zero; sensor dims (512–515) are filled in
774 if query_vector is None:
775 print(
776 f"[SimilarCrops] No usable stored vector for {crop_id}, using sensor fallback"
777 )
778 payload = _json.loads(payload_json) if payload_json else {}
779 raw_sensors = payload.get("sensor_data") or payload.get("sensors") or {}
780
781 SENSOR_ORDER = ["pH", "EC", "temp", "humidity"]
782 SENSOR_DEFAULTS = {"pH": 6.0, "EC": 1.5, "temp": 23.0, "humidity": 65.0}
783
784 full_vec = np.zeros(516, dtype=np.float32)
785 for i, key in enumerate(SENSOR_ORDER):
786 raw = raw_sensors.get(key)
787 val = raw[-1] if isinstance(raw, list) and raw else raw
788 try:
789 full_vec[512 + i] = (
790 float(val) if val is not None else SENSOR_DEFAULTS[key]
791 )
792 except (TypeError, ValueError):
793 full_vec[512 + i] = SENSOR_DEFAULTS[key]
794
795 query_vector = full_vec.tolist()
796
797 # Step 3: Vector search, excluding the source crop
798 results = client.query_points(
799 collection_name=COLLECTION_NAME,
800 query=query_vector,
801 query_filter=models.Filter(
802 must_not=[
803 models.FieldCondition(
804 key="crop_id",
805 match=models.MatchValue(value=crop_id),
806 )
807 ]
808 ),
809 limit=6,
810 with_payload=True,
811 with_vectors=False,
812 )
813
814 hits = results.points if hasattr(results, "points") else results
815
816 return {
817 "status": "success",
818 "results": [
819 {
820 "id": str(r.id),
821 "score": round(float(r.score), 4),
822 "payload": r.payload,
823 }
824 for r in hits
825 ],
826 }
827
828 except Exception as e:
829 traceback.print_exc()
830 return {"status": "error", "message": str(e), "results": []}