|
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}")
|
|
|
|
|
|
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}")
|
|
|
|
|
|
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"] = []
|
|
|
|
|
|
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"] = []
|
|
|
|
|
|
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"] = []
|
|
|
|
|
|
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
|
|
|