Spaces:
Sleeping
Sleeping
| from typing import Dict, List | |
| import json | |
| from agents import AlternativeAgent, SimilarityAgent, ConflictAgent, InteractionAgent | |
| from rdflib import Graph, Namespace | |
| import logging | |
| class DrugInteractionAnalyzer: | |
| """ | |
| Analyzer class to handle drug interactions, alternatives, similarities, and conflicts using separate agents. | |
| """ | |
| def __init__(self, owl_file_path: str): | |
| """ | |
| Initialize the DrugInteractionAnalyzer with the ontology file and instantiate agents. | |
| Args: | |
| owl_file_path (str): Path to the RDF/XML ontology file. | |
| """ | |
| self.graph = Graph() | |
| try: | |
| self.graph.parse(owl_file_path, format="xml") | |
| self.namespace = Namespace("http://www.example.org/DrugInteraction.owl#") | |
| logging.info(f"Successfully loaded ontology from {owl_file_path}") | |
| except Exception as e: | |
| logging.error(f"Failed to load ontology file: {e}") | |
| raise ValueError(f"Failed to load ontology file: {e}") | |
| # Initialize agents | |
| self.alternative_agent = AlternativeAgent(self.graph, self.namespace) | |
| self.similarity_agent = SimilarityAgent(self.graph, self.namespace) | |
| self.conflict_agent = ConflictAgent(self.graph, self.namespace) | |
| self.interaction_agent = InteractionAgent(self.graph, self.namespace) | |
| logging.info("Initialized all agents successfully.") | |
| def analyze_drugs(self, drug_names: List[str]) -> Dict[str, Dict[str, List[Dict[str, str]]]]: | |
| """ | |
| Analyze multiple drugs using separate agents and return structured results. | |
| Args: | |
| drug_names (List[str]): List of drug names to analyze. | |
| Returns: | |
| Dict[str, Dict[str, List[Dict[str, str]]]]: Structured analysis results. | |
| """ | |
| results = {} | |
| for drug in drug_names: | |
| drug_results = {} | |
| logging.info(f"Analyzing drug: {drug}") | |
| # Alternatives | |
| try: | |
| alternatives = self.alternative_agent.get_alternatives(drug) | |
| drug_results["alternatives"] = alternatives | |
| logging.debug(f"Alternatives for {drug}: {alternatives}") | |
| except Exception as e: | |
| logging.warning(f"Failed to get alternatives for {drug}: {e}") | |
| drug_results["alternatives"] = [] | |
| # Similarities | |
| try: | |
| similarities = self.similarity_agent.get_similarities(drug) | |
| drug_results["similarities"] = similarities | |
| logging.debug(f"Similarities for {drug}: {similarities}") | |
| except Exception as e: | |
| logging.warning(f"Failed to get similarities for {drug}: {e}") | |
| drug_results["similarities"] = [] | |
| # Conflicts | |
| try: | |
| conflicts = self.conflict_agent.get_conflicts(drug) | |
| drug_results["conflicts"] = conflicts | |
| logging.debug(f"Conflicts for {drug}: {conflicts}") | |
| except Exception as e: | |
| logging.warning(f"Failed to get conflicts for {drug}: {e}") | |
| drug_results["conflicts"] = [] | |
| # Interactions | |
| try: | |
| interactions = self.interaction_agent.get_interactions(drug) | |
| drug_results["interactions"] = interactions | |
| logging.debug(f"Interactions for {drug}: {interactions}") | |
| except Exception as e: | |
| logging.warning(f"Failed to get interactions for {drug}: {e}") | |
| drug_results["interactions"] = [] | |
| results[drug] = drug_results | |
| logging.info(f"Completed analysis for {drug}") | |
| logging.info("Completed analysis for all drugs.") | |
| return results | |