Spaces:
Sleeping
Sleeping
File size: 3,869 Bytes
4704777 |
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 |
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
|