Upload 11 files

agent.py

@@ -0,0 +1,28 @@
+from magentic import Agent, prompt
+from tools import ImageCaptioning, WebSearch, GoogleSheets
+
+class ResearchAgent(Agent):
+    def __init__(self, name: str, description: str):
+        super().__init__(name=name, description=description)
+        self.tools = {
+            "image_caption": ImageCaptioning(),
+            "web_search": WebSearch(),
+            "sheets": GoogleSheets()
+        }
+    
+    @prompt("Analiziraj sliko in identificiraj objekte ter dejavnike tveganja")
+    async def process_image(self, image):
+        """Obdelava slike z uporabo magentic-one promptov"""
+        return await self.tools["image_caption"].analyze(image)
+
+class UnderwritingAgent(Agent):
+    def __init__(self, name: str, description: str):
+        super().__init__(name=name, description=description)
+    
+    @prompt("Oceni tveganje in določi ustrezno kritje")
+    async def assess_risk(self, data):
+        """Ocena tveganja z uporabo magentic-one promptov"""
+        # Implementacija ocene tveganja
+        pass
+
+# Podobno implementirajte še SalesAgent in DelegationAgent

gradio_app.py

@@ -0,0 +1,24 @@
+import gradio as gr
+from main import initialize_agents
+
+def create_interface():
+    agents = initialize_agents()
+    
+    with gr.Blocks() as app:
+        gr.Markdown("# InsurTech MGA Aplikacija")
+        
+        with gr.Tab("Ocena tveganja"):
+            image_input = gr.Image()
+            text_output = gr.Textbox()
+            
+            image_input.change(
+                fn=agents["research"].process_image,
+                inputs=image_input,
+                outputs=text_output
+            )
+            
+    return app
+
+if __name__ == "__main__":
+    interface = create_interface()
+    interface.launch() 

helper.py

@@ -0,0 +1,47 @@
+import os
+from dotenv import load_dotenv
+from typing import Dict, Any
+import logging
+
+def setup_logging(logs_dir: str) -> None:
+    """Nastavitev beleženja dogodkov"""
+    if not os.path.exists(logs_dir):
+        os.makedirs(logs_dir)
+        
+    logging.basicConfig(
+        level=logging.INFO,
+        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+        handlers=[
+            logging.FileHandler(os.path.join(logs_dir, 'app.log')),
+            logging.StreamHandler()
+        ]
+    )
+
+def load_environment() -> None:
+    """Nalaganje okoljskih spremenljivk"""
+    load_dotenv()
+    required_vars = [
+        'OPENAI_API_KEY',
+        'TOGETHER_API_KEY',
+        'STRIPE_API_KEY'
+    ]
+    
+    missing = [var for var in required_vars if not os.getenv(var)]
+    if missing:
+        raise EnvironmentError(f"Manjkajoče okoljske spremenljivke: {', '.join(missing)}")
+
+def process_image_data(image_path: str) -> Dict[str, Any]:
+    """Obdelava vhodne slike"""
+    from PIL import Image
+    
+    try:
+        with Image.open(image_path) as img:
+            return {
+                "path": image_path,
+                "size": img.size,
+                "format": img.format,
+                "mode": img.mode
+            }
+    except Exception as e:
+        logging.error(f"Napaka pri obdelavi slike: {str(e)}")
+        return {}

insurance_utils.py

@@ -0,0 +1,53 @@
+from typing import Dict, Any
+from datetime import datetime, timedelta
+import json
+
+class PolicyGenerator:
+    def __init__(self):
+        self.policy_counter = 0
+        
+    def generate_policy(self, offer_data: Dict[str, Any]) -> Dict[str, Any]:
+        """Generiranje zavarovalne police"""
+        self.policy_counter += 1
+        
+        start_date = datetime.now()
+        end_date = start_date + timedelta(days=365)
+        
+        return {
+            "policy_number": f"POL-{datetime.now().year}-{self.policy_counter:04d}",
+            "start_date": start_date.isoformat(),
+            "end_date": end_date.isoformat(),
+            "premium": offer_data["premium"],
+            "coverage": offer_data["coverage"],
+            "terms": offer_data["terms"]
+        }
+    
+    def save_policy(self, policy: Dict[str, Any], filepath: str) -> None:
+        """Shranjevanje police v JSON formatu"""
+        with open(filepath, 'w', encoding='utf-8') as f:
+            json.dump(policy, f, indent=2, ensure_ascii=False)
+
+class ClaimsHandler:
+    def __init__(self):
+        self.valid_claim_types = {"damage", "theft", "liability"}
+        
+    def validate_claim(self, claim_data: Dict[str, Any]) -> bool:
+        """Preverjanje veljavnosti zahtevka"""
+        return all([
+            claim_data.get("type") in self.valid_claim_types,
+            claim_data.get("policy_number"),
+            claim_data.get("date_of_incident"),
+            claim_data.get("description")
+        ])
+    
+    def process_claim(self, claim_data: Dict[str, Any]) -> Dict[str, Any]:
+        """Obdelava zahtevka"""
+        if not self.validate_claim(claim_data):
+            return {"status": "rejected", "reason": "Neveljavni podatki zahtevka"}
+            
+        # Tukaj bi dodali logiko za oceno zahtevka
+        return {
+            "status": "processing",
+            "claim_id": f"CLM-{datetime.now().strftime('%Y%m%d-%H%M%S')}",
+            "estimated_payout": 0.0
+        }

main.py

@@ -0,0 +1,60 @@
+import asyncio
+import logging
+import os
+from autogen_core import EVENT_LOGGER_NAME, AgentId, AgentProxy, SingleThreadedAgentRuntime
+from autogen_magentic_one.agents.orchestrator import LedgerOrchestrator
+from autogen_magentic_one.utils import LogHandler, create_completion_client_from_env
+from agents.research_agent import ResearchAgent
+from agents.underwriting_agent import UnderwritingAgent
+from agents.sales_agent import SalesAgent
+from runtime.orchestrator import InsuranceOrchestrator
+
+async def main(logs_dir: str) -> None:
+    # Ustvarimo runtime
+    runtime = SingleThreadedAgentRuntime()
+    
+    # Ustvarimo client za LLM
+    client = create_completion_client_from_env(model="gpt-4")
+    
+    # Registriramo agente
+    await ResearchAgent.register(runtime, "ResearchAgent", 
+                               lambda: ResearchAgent(model_client=client))
+    research = AgentProxy(AgentId("ResearchAgent", "default"), runtime)
+    
+    await UnderwritingAgent.register(runtime, "UnderwritingAgent",
+                                   lambda: UnderwritingAgent(model_client=client))
+    underwriting = AgentProxy(AgentId("UnderwritingAgent", "default"), runtime)
+    
+    await SalesAgent.register(runtime, "SalesAgent",
+                            lambda: SalesAgent(model_client=client))
+    sales = AgentProxy(AgentId("SalesAgent", "default"), runtime)
+    
+    agent_list = [research, underwriting, sales]
+    
+    # Registriramo orchestrator
+    await InsuranceOrchestrator.register(
+        runtime,
+        "Orchestrator",
+        lambda: InsuranceOrchestrator(
+            agents=agent_list,
+            model_client=client,
+            max_rounds=30,
+            max_time=25 * 60,
+            return_final_answer=True,
+        )
+    )
+    
+    runtime.start()
+    await runtime.stop_when_idle()
+
+if __name__ == "__main__":
+    logs_dir = "logs"
+    if not os.path.exists(logs_dir):
+        os.makedirs(logs_dir)
+        
+    logger = logging.getLogger(EVENT_LOGGER_NAME)
+    logger.setLevel(logging.INFO)
+    log_handler = LogHandler(filename=os.path.join(logs_dir, "insurance_log.jsonl"))
+    logger.handlers = [log_handler]
+    
+    asyncio.run(main(logs_dir)) 

orchestrator.py

@@ -0,0 +1,22 @@
+from autogen_magentic_one.agents.orchestrator import LedgerOrchestrator
+from typing import List
+from autogen_core import AgentProxy
+
+class InsuranceOrchestrator(LedgerOrchestrator):
+    def __init__(self, agents: List[AgentProxy], **kwargs):
+        super().__init__(agents=agents, **kwargs)
+        
+    async def orchestrate_workflow(self, initial_task: dict):
+        """Koordinacija delovnega toka zavarovanja"""
+        steps = [
+            ("research", "analyze_risk"),
+            ("underwriting", "calculate_premium"),
+            ("sales", "prepare_offer")
+        ]
+        
+        results = {}
+        for agent_name, action in steps:
+            agent = self.get_agent(agent_name)
+            results[action] = await agent.execute_action(action, initial_task)
+            
+        return results 

research_agent.py

@@ -0,0 +1,29 @@
+from autogen_magentic_one.agents.base import BaseAgent
+from autogen_core.code_executor import CodeBlock
+from typing import Dict, Any
+
+class ResearchAgent(BaseAgent):
+    def __init__(self, model_client):
+        super().__init__(
+            name="ResearchAgent",
+            description="Agent za analizo slik in raziskovanje zavarovalnih primerov",
+            model_client=model_client
+        )
+        
+    async def process_image(self, image_data: Dict[str, Any]) -> Dict[str, Any]:
+        """Analiza slike in identifikacija objektov"""
+        prompt = f"""Analiziraj sliko in identificiraj:
+        1. Vse vidne objekte
+        2. Potencialne dejavnike tveganja
+        3. Stanje objektov
+        
+        Slika: {image_data}
+        """
+        
+        response = await self.model_client.complete(prompt)
+        return self._parse_response(response)
+        
+    def _parse_response(self, response: str) -> Dict[str, Any]:
+        """Pretvorba odgovora v strukturirane podatke"""
+        # Implementacija parsanja
+        return {} 

run_app.py

@@ -0,0 +1,20 @@
+import argparse
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--interface', type=str, choices=['gradio', 'streamlit'], 
+                       default='gradio', help='Izbira uporabniškega vmesnika')
+    args = parser.parse_args()
+    
+    if args.interface == 'gradio':
+        from gradio_app import create_interface
+        app = create_interface()
+        app.launch(share=True)
+    else:
+        import streamlit.cli as stcli
+        import sys
+        sys.argv = ["streamlit", "run", "streamlit_app.py"]
+        stcli.main()
+
+if __name__ == "__main__":
+    main() 

sales_agent.py

@@ -0,0 +1,31 @@
+from autogen_magentic_one.agents.base import BaseAgent
+from typing import Dict, Any
+
+class SalesAgent(BaseAgent):
+    def __init__(self, model_client):
+        super().__init__(
+            name="SalesAgent",
+            description="Agent za pripravo ponudbe in komunikacijo s strankami",
+            model_client=model_client
+        )
+        
+    async def prepare_offer(self, insurance_data: Dict[str, Any]) -> Dict[str, Any]:
+        """Priprava zavarovalne ponudbe"""
+        prompt = f"""Pripravi ponudbo za zavarovanje:
+        1. Premium: {insurance_data.get('final_premium')}
+        2. Kritje: {insurance_data.get('coverage_details')}
+        3. Posebni pogoji: {insurance_data.get('special_conditions', [])}
+        """
+        
+        response = await self.model_client.complete(prompt)
+        return self._format_offer(response, insurance_data)
+    
+    def _format_offer(self, offer_text: str, data: Dict[str, Any]) -> Dict[str, Any]:
+        """Oblikovanje končne ponudbe"""
+        return {
+            "offer_id": "OFF-" + str(hash(offer_text))[:8],
+            "premium": data.get('final_premium'),
+            "coverage": data.get('coverage_details'),
+            "terms": offer_text,
+            "valid_until": "30 days"
+        } 

tools.py

@@ -0,0 +1,67 @@
+from typing import Dict, Any
+import requests
+import stripe
+from PIL import Image
+import google.auth
+from google.oauth2 import service_account
+from googleapiclient.discovery import build
+
+class ImageCaptioning:
+    def __init__(self):
+        self.model = None  # Initialize your image captioning model
+        
+    async def analyze(self, image: Image) -> Dict[str, Any]:
+        """Analiza slike in vračanje kontekstualnih informacij"""
+        results = {
+            "objects": [],  # seznam zaznanih objektov
+            "count": {},    # število posameznih objektov
+            "risk_factors": []  # zaznani dejavniki tveganja
+        }
+        return results
+
+class WebSearch:
+    def __init__(self):
+        self.session = requests.Session()
+        
+    async def search(self, query: str) -> list:
+        """Izvajanje spletnega iskanja in luščenja podatkov"""
+        results = []
+        # Implementacija spletnega iskanja
+        return results
+
+class GoogleSheets:
+    def __init__(self):
+        self.SCOPES = ['https://www.googleapis.com/auth/spreadsheets.readonly']
+        self.creds = None
+        
+    async def get_pricing_data(self, coverage_type: str) -> Dict[str, float]:
+        """Pridobivanje podatkov o cenah iz Google Sheets"""
+        pricing_data = {}
+        # Implementacija branja podatkov
+        return pricing_data
+
+class StripePayment:
+    def __init__(self, api_key: str):
+        stripe.api_key = api_key
+        
+    async def create_payment(self, amount: float, currency: str) -> Dict[str, Any]:
+        """Ustvarjanje Stripe plačila"""
+        try:
+            payment_intent = stripe.PaymentIntent.create(
+                amount=int(amount * 100),  # Stripe uporablja najmanjše denarne enote
+                currency=currency
+            )
+            return {"status": "success", "client_secret": payment_intent.client_secret}
+        except stripe.error.StripeError as e:
+            return {"status": "error", "message": str(e)}
+
+class WeatherAPI:
+    def __init__(self):
+        self.api_key = None
+        self.base_url = "https://api.weatherapi.com/v1"
+        
+    async def get_weather_forecast(self, location: str, days: int = 7) -> Dict[str, Any]:
+        """Pridobivanje vremenske napovedi"""
+        weather_data = {}
+        # Implementacija vremenske napovedi
+        return weather_data 

underwriting_agent.py

@@ -0,0 +1,30 @@
+from autogen_magentic_one.agents.base import BaseAgent
+from typing import Dict, Any
+
+class UnderwritingAgent(BaseAgent):
+    def __init__(self, model_client):
+        super().__init__(
+            name="UnderwritingAgent",
+            description="Agent za oceno tveganja in izračun premije",
+            model_client=model_client
+        )
+        
+    async def calculate_premium(self, risk_data: Dict[str, Any]) -> Dict[str, Any]:
+        """Izračun zavarovalne premije na podlagi ocene tveganja"""
+        prompt = f"""Oceni tveganje in izračunaj premijo za:
+        1. Identificirani objekti: {risk_data.get('objects', [])}
+        2. Dejavniki tveganja: {risk_data.get('risk_factors', [])}
+        3. Lokacija: {risk_data.get('location', 'Unknown')}
+        """
+        
+        response = await self.model_client.complete(prompt)
+        return self._calculate_final_premium(response)
+    
+    def _calculate_final_premium(self, assessment: str) -> Dict[str, Any]:
+        """Izračun končne premije na podlagi ocene"""
+        return {
+            "base_premium": 0.0,
+            "risk_multiplier": 1.0,
+            "final_premium": 0.0,
+            "coverage_details": {}
+        }