AOP-wiki as a new database + text categorization

examples/aop_categorization/prompt_specification.txt

@@ -0,0 +1,3 @@
+The text describes an Adverse Outcome Pathway (AOP). Classify the AOP according to its target organ / system.
+
+Category other includes AOPs not associated with a specific organ system, including changes in body weight, decreased survival, and other nonspecific toxicity.

pyproject.toml

@@ -33,6 +33,7 @@ dependencies = [
     "tenacity",
     "boto3>=1.43.2",
     "pillow",
+    "sparqlwrapper",
 ]
 description = "Tools to support data mining for the development of (q)AOPs"
 keywords = ["AOP"," toxicology"," data mining"," ai"," machine learning"," data analysis"]

src/aoptk/literature/categorize_text.py

@@ -0,0 +1,20 @@
+from __future__ import annotations
+from abc import ABC
+from abc import abstractmethod
+
+
+class CategorizeText(ABC):
+    """Abstract base class for categorizing text."""
+
+    @abstractmethod
+    def categorize_text(self, text: str, categories: list[str]) -> str | None:
+        """Categorize the given text into one of the specified categories.
+
+        Args:
+            text (str): The text to categorize.
+            categories (list[str]): The list of available categories.
+
+        Returns:
+            str | None: The categorized label or None if no match is found.
+        """
+        ...

src/aoptk/literature/databases/aop_wiki.py

@@ -0,0 +1,54 @@
+from SPARQLWrapper import JSON
+from SPARQLWrapper import SPARQLWrapper
+from aoptk.literature.abstract import Abstract
+from aoptk.literature.get_abstract import GetAbstract
+from aoptk.literature.id import ID
+
+
+class AOPWiki(GetAbstract):
+    """Class to extract data from AOP-Wiki."""
+
+    endpoint_url = "https://aopwiki.rdf.bigcat-bioinformatics.org/sparql"
+    prefix = """PREFIX aop: <http://aopkb.org/aop_ontology#>
+PREFIX dc: <http://purl.org/dc/elements/1.1/>
+PREFIX dcterms: <http://purl.org/dc/terms/>"""
+    abstract_query = (
+        prefix
+        + """
+
+SELECT ?AOP ?AOPTitle ?abstract
+
+WHERE {
+  ?AOP a aop:AdverseOutcomePathway ;
+    dc:title ?AOPTitle .
+  OPTIONAL { ?AOP dcterms:abstract ?abstract. }
+}
+"""
+    )
+
+    def __init__(
+        self,
+    ):
+        self.sparql = SPARQLWrapper(self.endpoint_url)
+        self.sparql.setReturnFormat(JSON)
+
+    def get_abstracts(self) -> list[Abstract]:
+        """Get abstracts of AOPs from AOP-Wiki.
+
+        Returns:
+            list[Abstract]: List of abstracts.
+        """
+        self.sparql.setQuery(self.abstract_query)
+        results = self.sparql.query().convert()
+
+        abstracts: list[Abstract] = []
+        for result in results["results"]["bindings"]:
+            aop_id = result.get("AOP", {}).get("value")
+            title = result.get("AOPTitle", {}).get("value", "")
+            abstract = result.get("abstract", {}).get("value", "")
+            if abstract == "":
+                continue
+            text = f"{title}\n\n{abstract}".strip()
+            abstracts.append(Abstract(id=ID(aop_id), text=text))
+
+        return abstracts

src/aoptk/prompts/categorize_text_prompt.txt

@@ -0,0 +1,14 @@
+Task:
+Categorize the provided text into exactly one category from the following list:
+{{ categories }}
+
+{{ specification_categorization_prompt }}
+
+Output requirements:
+1. Return exactly one category label. Treat all category names as fixed labels, not words. Do not modify, stem, normalize, correct, interpret, paraphrase, or transform labels in any way. Output the label exactly as provided, character-for-character.
+2. Do not include explanations, labels, punctuation, or additional text.
+3. If there is no fitting category, return exactly:
+"none"
+
+Text:
+{{ text }}

src/aoptk/text_generation_api.py

@@ -57,8 +57,10 @@ class TextGenerationAPI(
     convert_pdf_scan_prompt_template: str = "convert_pdf_scan_prompt.txt"
     convert_image_prompt_template: str = "convert_image_prompt.txt"
     find_relevant_publications_prompt_template: str = "find_relevant_publications_prompt.txt"
+    categorize_text_prompt_template: str = "categorize_text_prompt.txt"
 
     specification_relationship_text_prompt: str = ""
+    specification_categorization_prompt: str = ""
 
     def __init__(
         self,
@@ -465,3 +467,21 @@ def find_relevant_publications(self, question: str, text: str) -> bool | None:
             if response == "no":
                 return False
         return None
+
+    def categorize_text(self, text: str, categories: list[str]) -> str | None:
+        """Categorize the given text into one of the specified categories.
+
+        Args:
+            text (str): The text to categorize.
+            categories (list[str]): The list of available categories.
+
+        Returns:
+            str | None: The categorized label or None if no match is found.
+        """
+        if response := self._prompt(
+            self._render_prompt(self.categorize_text_prompt_template, text=text, categories=", ".join(categories)),
+        ).lower():
+            if response == "none":
+                return None
+            return response
+        return None

tests/test_aop_wiki.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+from aoptk.literature.databases.aop_wiki import AOPWiki
+from aoptk.literature.get_abstract import GetAbstract
+from aoptk.literature.id import ID
+
+
+def test_can_create():
+    """Test that AOPWiki can be instantiated."""
+    actual = AOPWiki()
+    assert actual is not None
+
+
+def test_implements_interface():
+    """Test that AOPWiki implements GetAbstract interface."""
+    assert issubclass(AOPWiki, GetAbstract)
+
+
+def test_get_abstracts():
+    """Test that AOP 38 has the correct title."""
+    abstracts = AOPWiki().get_abstracts()
+    aop_38 = next(abstract for abstract in abstracts if abstract.id == ID("https://identifiers.org/aop/38"))
+    min_length_of_abstract = 2000
+    min_number_of_abstracts = 300
+    assert aop_38.text.startswith("Protein Alkylation leading to Liver Fibrosis")
+    assert len(aop_38.text) > min_length_of_abstract
+    assert len(abstracts) > min_number_of_abstracts

tests/test_text_generation.py

@@ -277,3 +277,25 @@ def test_find_relevant_publications(question: str, text: str, expected: bool):
     """Test that find_relevant_publications method finds relevant publications."""
     actual = TextGenerationAPI().find_relevant_publications(question=question, text=text)
     assert actual == expected
+
+
+@pytest.mark.openai
+@pytest.mark.parametrize(
+    ("text", "categories", "expected"),
+    [
+        (
+            "This text is about liver.",
+            ["Liver", "Heart", "Kidney"],
+            "liver",
+        ),
+        (
+            "This text is not about any organ.",
+            ["Liver", "Heart", "Kidney"],
+            None,
+        ),
+    ],
+)
+def test_categorize_text(text: str, categories: list[str], expected: str | None):
+    """Test that categorize_text method categorizes text correctly."""
+    actual = TextGenerationAPI().categorize_text(text=text, categories=categories)
+    assert actual == expected