Spaces:
Running
Running
File size: 4,463 Bytes
edfeb78 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 |
from typing import List, AsyncIterator
import os
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import MultiModalMessage
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_core import Image as AGImage
from autogen_core.tools import FunctionTool
from autogen_ext.models.openai import OpenAIChatCompletionClient
from dotenv import load_dotenv, find_dotenv
from PIL import Image
import yaml
from tools import Tools
class Minerva:
"""
AI Guardian for Scam Protection using multi-agent system for analyzing images
to identify scam attempts and provide personalized scam prevention.
"""
def __init__(self, config_path: str = "config/agents.yaml"):
"""
Initialize Minerva with configuration and setup agents.
"""
self.load_environment()
self.model = self.initialize_model()
self.config = self.load_config(config_path)
self.tools = Tools()
self.agents = self.create_agents()
self.team = self.create_team()
def load_environment(self):
"""Load environment variables"""
load_dotenv(find_dotenv())
def load_config(self, config_path: str) -> dict:
"""Load agent configurations from YAML file"""
with open(config_path, 'r') as file:
return yaml.safe_load(file)
def initialize_model(self) -> OpenAIChatCompletionClient:
"""Initialize OpenAI model"""
return OpenAIChatCompletionClient(
model="gpt-4o",
api_key=os.getenv("OPENAI_API_KEY")
)
def create_agents(self) -> List[AssistantAgent]:
"""Create all required agents with their specialized roles"""
ocr_tool = FunctionTool(
self.tools.ocr,
description="Extracts text from an image path"
)
url_checker_tool = FunctionTool(
self.tools.is_url_safe,
description="Checks if a URL is safe"
)
agents = []
agents.append(AssistantAgent(
name="OCR_Specialist",
description="Extracts text from an image",
system_message=self.config['ocr_agent']['assignment'],
model_client=self.model,
#tools=[ocr_tool] # Default OCR to GPT-4o vision capabilities. Uncomment to OCR with tool calling (requires pytesseract)
))
agents.append(AssistantAgent(
name="URL_Checker",
description="Checks if a URL is safe",
system_message=self.config['url_checker_agent']['assignment'],
model_client=self.model,
tools=[url_checker_tool]
))
agents.append(AssistantAgent(
name="Content_Analyst",
description="Analyzes the text for scam patterns",
system_message=self.config['content_agent']['assignment'],
model_client=self.model,
tools=[url_checker_tool]
))
agents.append(AssistantAgent(
name="Decision_Maker",
description="Synthesizes the analyses and make final determination",
system_message=self.config['decision_agent']['assignment'],
model_client=self.model
))
agents.append(AssistantAgent(
name="Summary_Agent",
description="Generate a summary of the final determination",
system_message=self.config['summary_agent']['assignment'],
model_client=self.model
))
agents.append(AssistantAgent(
name="Language_Translation_Agent",
description="Translate the summary to the user language",
system_message=self.config['language_translation_agent']['assignment'],
model_client=self.model
))
return agents
def create_team(self) -> RoundRobinGroupChat:
"""Create a team of agents that work together in Round Robin fashion"""
return RoundRobinGroupChat(
self.agents,
max_turns=6
)
def reset(self):
"""Reset team state"""
self.team.reset()
async def analyze_image(self, image_path: str) -> AsyncIterator:
"""
Analyze an image for potential scams.
"""
pil_image = Image.open(image_path)
img = AGImage(pil_image)
mm_message = MultiModalMessage(content=[img], source="User")
return self.team.run_stream(task=mm_message) |