Spaces:
Running
Running
File size: 4,462 Bytes
edfeb78 202ed7a edfeb78 202ed7a edfeb78 1057d18 edfeb78 1057d18 edfeb78 1057d18 edfeb78 254da27 edfeb78 c0c8229 edfeb78 202ed7a edfeb78 202ed7a edfeb78 c0c8229 edfeb78 c0c8229 edfeb78 d37f373 edfeb78 d37f373 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 130 131 |
from typing import List, AsyncIterator
import os
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import TextMentionTermination
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 an Agentic Workflow
"""
def __init__(self, config_path: str = "config/agents.yaml"):
"""
Initialize Minerva with agents and tools
"""
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 and tools"""
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="Link_Checker",
description="Checks if a Link is safe",
system_message=self.config['link_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
))
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"""
termination = TextMentionTermination("NO_TEXT_FOUND")
return RoundRobinGroupChat(
self.agents,
max_turns=6,
termination_condition=termination
)
async def reset(self):
"""Reset team state"""
await self.team.reset()
async def analyze(self, image: Image) -> AsyncIterator:
"""
Analyze an image for potential scams.
"""
img = AGImage(image)
mm_message = MultiModalMessage(content=[img], source="User")
return self.team.run_stream(task=mm_message) |