Last active
July 25, 2025 11:53
-
-
Save Praveenlingala378/d29f410887f1195e1e3233e6ade67d74 to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import cv2 | |
| import numpy as np | |
| from typing import Dict, List, Any | |
| import face_recognition | |
| import base64 | |
| from io import BytesIO | |
| from PIL import Image | |
| # Agent Framework (following the reference pattern) | |
| class Agent: | |
| def __init__(self, name: str, model: str, description: str, instruction: str, tools: List): | |
| self.name = name | |
| self.model = model | |
| self.description = description | |
| self.instruction = instruction | |
| self.tools = tools | |
| self.reference_encodings = [] | |
| self.reference_uploaded = False | |
| self.crowd_photos = [] | |
| def run(self, query: str = None): | |
| """Main agent execution""" | |
| # Prompt for input images if no query is provided | |
| if not query: | |
| print("👋 Lost & Found Agent Active!") | |
| print("Please provide:") | |
| print("1. Reference photo of the missing person (path to image file)") | |
| print("2. Crowd photo to search in (path to image file)") | |
| ref_path = input("Enter reference photo path: ") | |
| crowd_path = input("Enter crowd photo path: ") | |
| # Process reference photo | |
| ref_result = upload_reference_photo(photo_path=ref_path) | |
| print(f"\n📸 Reference Photo Result: {ref_result.get('message', ref_result.get('error_message', 'No message'))}") | |
| if ref_result["status"] == "success": | |
| self.reference_uploaded = True | |
| # In a real implementation, you would pass the encoding from the reference photo to the search function. | |
| # For this demo, the search function is a simulation. | |
| search_result = search_person_in_crowd(crowd_photos=[crowd_path]) | |
| print(f"\n🔍 Search Result: {search_result.get('message', search_result.get('error_message', 'No message'))}") | |
| print(f"Matches found: {search_result.get('matches_found', 0)}") | |
| else: | |
| print("Reference photo processing failed. Please try again.") | |
| return | |
| return self._process_query(query) | |
| def _initial_setup(self): | |
| """Initial setup when agent starts""" | |
| return { | |
| "message": "👋 Lost & Found Agent Active! Please upload:\n1. Reference photo of missing person\n2. Crowd photos to search in\nUse 'upload_reference_photo' and 'upload_crowd_photos' tools.", | |
| "status": "waiting_for_photos", | |
| "next_steps": ["upload_reference_photo", "upload_crowd_photos"] | |
| } | |
| def _process_query(self, query: str): | |
| """Process user queries and route to appropriate tools""" | |
| query_lower = query.lower() | |
| if "search" in query_lower and self.reference_uploaded: | |
| return search_person_in_crowd() | |
| elif "help" in query_lower or "features" in query_lower: | |
| return discuss_search_features() | |
| elif not self.reference_uploaded: | |
| return {"message": "Please upload reference photo first using 'upload_reference_photo' tool."} | |
| else: | |
| return discuss_search_features() | |
| # Tool 1: Upload and process reference photo | |
| def upload_reference_photo(photo_path: str = None, photo_base64: str = None) -> Dict[str, Any]: | |
| """ | |
| Uploads and processes reference photo of the missing person. | |
| """ | |
| print("--- Tool: upload_reference_photo called ---") | |
| try: | |
| if photo_path: | |
| # --- THIS IS THE FIX --- | |
| # Use the library's built-in function to load the image. | |
| # This ensures it's in the correct RGB format. | |
| image = face_recognition.load_image_file(photo_path) | |
| elif photo_base64: | |
| image_data = base64.b64decode(photo_base64) | |
| image_stream = BytesIO(image_data) | |
| image = face_recognition.load_image_file(image_stream) | |
| else: | |
| return {"status": "error", "error_message": "No photo provided. Please provide photo_path or photo_base64."} | |
| face_locations = face_recognition.face_locations(image) | |
| if not face_locations: | |
| return {"status": "error", "error_message": "No face detected in reference photo. Please upload a clear face photo."} | |
| face_encodings = face_recognition.face_encodings(image, face_locations) | |
| if not face_encodings: | |
| return {"status": "error", "error_message": "Could not generate face encoding. Please try a different photo."} | |
| reference_encoding = face_encodings[0] | |
| return { | |
| "status": "success", | |
| "message": f"✅ Reference photo processed successfully! Detected {len(face_locations)} face(s).", | |
| "faces_detected": len(face_locations), | |
| "encoding_generated": True, | |
| "next_step": "upload_crowd_photos" | |
| } | |
| except Exception as e: | |
| return {"status": "error", "error_message": f"Error processing reference photo: {str(e)}"} | |
| # Tool 2: Search person in crowd photos (simulation) | |
| def search_person_in_crowd(crowd_photos: List[str] = None, confidence_threshold: float = 0.6) -> Dict[str, Any]: | |
| """ | |
| Searches for the reference person in crowd photos using face recognition. | |
| """ | |
| print("--- Tool: search_person_in_crowd called ---") | |
| try: | |
| if not crowd_photos: | |
| return {"status": "error", "error_message": "No crowd photos provided."} | |
| search_results = [] | |
| total_faces_scanned = 0 | |
| for i, photo_path in enumerate(crowd_photos): | |
| print(f"Processing crowd photo {i+1}/{len(crowd_photos)}: {photo_path}") | |
| # In a real implementation, you would load the image here: | |
| # crowd_image = face_recognition.load_image_file(photo_path) | |
| # ... then run face detection and comparison ... | |
| # For this demo, we continue with the simulation | |
| faces_in_photo = np.random.randint(10, 50) | |
| total_faces_scanned += faces_in_photo | |
| match_found = np.random.random() > 0.7 | |
| if match_found: | |
| confidence = np.random.uniform(confidence_threshold, 0.95) | |
| location_x, location_y = np.random.randint(100, 800), np.random.randint(100, 600) | |
| search_results.append({ | |
| "photo_name": photo_path, | |
| "match_found": True, | |
| "confidence": round(confidence, 3), | |
| "location": {"x": location_x, "y": location_y}, | |
| }) | |
| else: | |
| search_results.append({"photo_name": photo_path, "match_found": False}) | |
| matches_found = len([r for r in search_results if r.get("match_found")]) | |
| return { | |
| "status": "success", | |
| "message": f"🔍 Search completed! Found {matches_found} potential match(es) in {len(crowd_photos)} photos.", | |
| "total_photos_processed": len(crowd_photos), | |
| "total_faces_scanned (simulated)": total_faces_scanned, | |
| "matches_found": matches_found, | |
| "results": search_results | |
| } | |
| except Exception as e: | |
| return {"status": "error", "error_message": f"Error during crowd search: {str(e)}"} | |
| # (Other tools and agent definition remain the same) | |
| # ... | |
| # Define the Lost and Found Agent | |
| MODEL_GEMINI_2_0_FLASH = "gemini-2.0-flash" | |
| lost_found_agent = Agent( | |
| name="lost_found_agent_v1", | |
| model=MODEL_GEMINI_2_0_FLASH, | |
| description="AI-powered Lost & Found agent for crowd safety at large events. Uses advanced face recognition to locate missing persons in crowd photos.", | |
| instruction="You are an intelligent Lost & Found assistant for Project Drishti. ", | |
| tools=[upload_reference_photo, search_person_in_crowd] | |
| ) | |
| if __name__ == "__main__": | |
| lost_found_agent.run() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment