A project that uses Langchain to construct SPARQL queries against SPARQL endpoints
import os 
from langchain.chat_models import ChatOpenAI
from langchain.chains import GraphSparqlQAChain
#from langchain.graphs import RdfGraph


from langsmith import Client
from langchain.agents import AgentType, initialize_agent, load_tools
from langchain.callbacks.tracers.langchain import wait_for_all_tracers

from rdfgraph import RdfGraph
from sparqlchain import MyGraphSparqlQAChain

import asyncio
from dotenv import load_dotenv
load_dotenv()

os.environ["LANGCHAIN_TRACING_V2"] = "false"
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"
os.environ["LANGCHAIN_PROJECT"] = "sparql"
#client = Client()

graph = RdfGraph(
    query_endpoint="https://bio2rdf.org/sparql",
    standard="owl",
    local_copy="mygraph.ttl",
)
#graph.load_schema()
#schema = graph.get_schema


chain = MyGraphSparqlQAChain.from_llm(
    ChatOpenAI(temperature=0), graph=graph, verbose=True
)
context ="""

# """
# Query: Which drugs can be used to treat hypertension? Return the drug name and indication for the first 5 results
# Answer: Answer: The drugs that can be used to treat hypertension are Latanoprost, Mecamylamine, Guanethidine, Pargyline, and Valsartan
# Query What is the mechanism of action for Lepirudin? Return the text that describes the mechanism of action. 
# Query: What are the functions of the target Prothrombin? Only retrieve 5 results.
# Query: What is the specific function of the target Prothrombin?
# Answer: The specific function of the target Prothrombin is to convert fibrinogen to fibrin and activate factors V, VII, VIII, XIII, and protein C. Additionally, when in complex with thrombomodulin, Prothrombin activates protein C.
# Query: What are the synonyms of the target Prothrombin?
# Answer: The synonyms of the target Prothrombin are Thrombin, Activated Factor II [IIa], Coagulation factor II, EC 3.4.21.5, and Prothrombin precursor
# Query: What species is the target Prothrombin from?
# Answer: The target Prothrombin is from the species Homo sapiens.

query = """
Query: What is the mechanism of action for Lepirudin?
"""
context = """
Context: Use Bio2RDF to answer the question. 

Schema:
Only use the following tuples to create the SPARQL query. The statements are written as a tuple (t1, rel, t2), where t1 is related to t2 by relation rel.
PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
PREFIX owl: <http://www.w3.org/2002/07/owl#>
PREFIX xsd: <http://www.w3.org/2001/XMLSchema#>
PREFIX dct: <http://purl.org/dc/terms/>
PREFIX dv: <http://bio2rdf.org/drugbank_vocabulary:>
PREFIX tv: <http://bio2rdf.org/taxonomy_vocabulary:>

dv:Drug dct:title xsd:Literal .
dv:Drug dv:category dv:Category .
dv:Category dct:title xsd:Literal .
dv:Drug dv:mechanism-of-action dv:Mechanism-of-action .
dv:Mechanism-of-action dct:description xsd:Literal .
dv:Drug dv:indication dv:Indication .
dv:Indication dct:description xsd:Literal .
dv:Drug dv:target dv:Target .
dv:Target dct:title xsd:Literal.
dv:Target dv:synonym xsd:Literal .
dv:Target dv:specific-function xsd:Literal .
dv:Target dv:go-function xsd:Literal .
dv:Target dv:go-component xsd:Literal .
dv:Target dv:cellular-location xsd:Literal .
dv:Target dv:x-taxonomy tv:Resource .
tv:Resource tv:scientific-name xsd:Literal .

Instructions:
Use the schema to guide the development of the SPARQL query. 
Use RDF variables such as ?drug ?target ?indication instead of prefixed names in SPARQL queries.
Apply text search on the objects of dct:title and dct:description. 
Do not use CONTAINS directly on RDF nodes.
Return text variables in a SELECT query.
Apply the DISTINCT on variables in a SELECT query.

Only use the results from the database query to answer the question.
"""

print(f'Query: {query}')
answer = chain.run( query + context )
print(f'Answer: {answer}')

src/query.py
micheldumontier/sparql-langchain
from __future__ import annotations

from typing import (
    TYPE_CHECKING,
    List,
    Optional,
)

if TYPE_CHECKING:
    import rdflib


prefixes = {
    "owl": """PREFIX owl: <http://www.w3.org/2002/07/owl#>\n""",
    "rdf": """PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>\n""",
    "rdfs": """PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>\n""",
    "xsd": """PREFIX xsd: <http://www.w3.org/2001/XMLSchema#>\n""",
    "dct": """PREFIX dct: <http://purl.org/dc/terms/>\n"""
}

from_restriction = """FROM <http://bio2rdf.org/drugbank_resource:bio2rdf.dataset.drugbank.R3>\n"""

cls_query_owl = prefixes["rdfs"] + prefixes['dct'] + (
"""
SELECT ?entity ?label
{
  {{
        SELECT DISTINCT ?entity 
        FROM_RESTRICTION
        { ?s ?entity ?o .}
  }}
  ?entity dct:title ?label .
}
""".replace("FROM_RESTRICTION",from_restriction)
)

op_query_owl = (prefixes["rdfs"]+ prefixes["owl"] + prefixes['dct'] + (
"""
SELECT ?entity ?label
{
  {{
        SELECT DISTINCT ?entity 
        FROM_RESTRICTION
        {?entity a owl:ObjectProperty .}
  }}
  ?entity dct:title ?label .
}
""".replace("FROM_RESTRICTION",from_restriction)
))

dp_query_owl = (prefixes["rdfs"] + prefixes["owl"] + prefixes['dct'] + (
"""
SELECT ?entity ?label
{
  {{
        SELECT DISTINCT ?entity 
        FROM_RESTRICTION
        {?entity a owl:DatatypeProperty .}
  }}
  ?entity dct:title ?label .
}
""".replace("FROM_RESTRICTION",from_restriction)
))


class RdfGraph:
    """
    RDFlib wrapper for graph operations.
    Modes:
    * local: Local file - can be queried and changed
    * online: Online file - can only be queried, changes can be stored locally
    * store: Triple store - can be queried and changed if update_endpoint available
    Together with a source file, the serialization should be specified.
    """

    def __init__(
        self,
        source_file: Optional[str] = None,
        serialization: Optional[str] = "ttl",
        query_endpoint: Optional[str] = None,
        update_endpoint: Optional[str] = None,
        standard: Optional[str] = "rdf",
        local_copy: Optional[str] = None,
    ) -> None:
        """
        Set up the RDFlib graph

        :param source_file: either a path for a local file or a URL
        :param serialization: serialization of the input
        :param query_endpoint: SPARQL endpoint for queries, read access
        :param update_endpoint: SPARQL endpoint for UPDATE queries, write access
        :param standard: RDF, RDFS, or OWL
        :param local_copy: new local copy for storing changes
        """
        self.source_file = source_file
        self.serialization = serialization
        self.query_endpoint = query_endpoint
        self.update_endpoint = update_endpoint
        self.standard = standard
        self.local_copy = local_copy

        try:
            import rdflib
            from rdflib.graph import DATASET_DEFAULT_GRAPH_ID as default
            from rdflib.plugins.stores import sparqlstore
        except ImportError:
            raise ValueError(
                "Could not import rdflib python package. "
                "Please install it with `pip install rdflib`."
            )
        if self.standard not in (supported_standards := ("rdf", "rdfs", "owl")):
            raise ValueError(
                f"Invalid standard. Supported standards are: {supported_standards}."
            )

        if (
            not source_file
            and not query_endpoint
            or source_file
            and (query_endpoint or update_endpoint)
        ):
            raise ValueError(
                "Could not unambiguously initialize the graph wrapper. "
                "Specify either a file (local or online) via the source_file "
                "or a triple store via the endpoints."
            )

        if source_file:
            if source_file.startswith("http"):
                self.mode = "online"
            else:
                self.mode = "local"
                if self.local_copy is None:
                    self.local_copy = self.source_file
            self.graph = rdflib.Graph()
            self.graph.parse(source_file, format=self.serialization)

        if query_endpoint:
            self.mode = "store"
            if not update_endpoint:
                self._store = sparqlstore.SPARQLStore()
                self._store.open(query_endpoint)
            else:
                self._store = sparqlstore.SPARQLUpdateStore()
                self._store.open((query_endpoint, update_endpoint))
            self.graph = rdflib.Graph(self._store) #, identifier=default)

        # Verify that the graph was loaded
        #if not len(self.graph):
        #    raise AssertionError("The graph is empty.")

        # Set schema
        self.schema = ""
        #self.load_schema()

    @property
    def get_schema(self) -> str:
        """
        Returns the schema of the graph database.
        """
        return self.schema

    def query(
        self,
        query: str,
    ) -> List[rdflib.query.ResultRow]:
        """
        Query the graph.
        """
        from rdflib.exceptions import ParserError
        from rdflib.query import ResultRow

        try:
            res = self.graph.query(query)
        except ParserError as e:
            raise ValueError("Generated SPARQL statement is invalid\n" f"{e}")
        return [r for r in res if isinstance(r, ResultRow)]

    def update(
        self,
        query: str,
    ) -> None:
        """
        Update the graph.
        """
        from rdflib.exceptions import ParserError

        try:
            self.graph.update(query)
        except ParserError as e:
            raise ValueError("Generated SPARQL statement is invalid\n" f"{e}")
        if self.local_copy:
            self.graph.serialize(
                destination=self.local_copy, format=self.local_copy.split(".")[-1]
            )
        else:
            raise ValueError("No target file specified for saving the updated file.")

    @staticmethod
    def _get_local_name(iri: str) -> str:
        if "#" in iri:
            local_name = iri.split("#")[-1]
        elif "/" in iri:
            local_name = iri.split("/")[-1]
            if ":" in local_name:
                local_name = local_name.split(":")[-1]
        else:
            raise ValueError(f"Unexpected IRI '{iri}', contains neither '#' nor '/'.")
        return local_name

    def _res_to_str(self, res: rdflib.query.ResultRow, var: str) -> str:
        return (
            "<"
            + str(res[var])
            + "> ("
            # + self._get_local_name(res[var])
            # + ", "
            + str(res["label"])
            + ")"
        )

    def load_schema(self) -> None:
        """
        Load the graph schema information.
        """

        if self.standard == "owl":
            clss = self.query(cls_query_owl)
            ops = self.query(op_query_owl)
            dps = self.query(dp_query_owl)
            self.schema = (
                f"In the following, each IRI is followed its label in parentheses.\n "
                f"The OWL graph supports the following node types:\n"
                f'{", ".join([self._res_to_str(r, "entity") for r in clss])}\n'
                f"The OWL graph supports the following object properties, "
                f"i.e., relationships between objects:\n"
                f'{", ".join([self._res_to_str(r, "entity") for r in ops])}\n'
                f"The OWL graph supports the following data properties, "
                f"i.e., relationships between objects and literals:\n"
                f'{", ".join([self._res_to_str(r, "entity") for r in dps])}\n'
            )
        else:
            raise ValueError(f"Mode '{self.standard}' is currently not supported.")

src/rdfgraph.py
micheldumontier/sparql-langchain
"""
Question answering over an RDF or OWL graph using SPARQL.
"""
from __future__ import annotations

from typing import Any, Dict, List, Optional
#import pydantic as pydantic_v1
from pydantic import Field
from langchain.prompts.prompt import PromptTemplate
from langchain.callbacks.manager import CallbackManagerForChainRun
from langchain.chains.base import Chain
from langchain.chains.graph_qa.prompts import (
    SPARQL_GENERATION_SELECT_PROMPT,
    SPARQL_GENERATION_UPDATE_PROMPT,
    SPARQL_INTENT_PROMPT,
    SPARQL_QA_PROMPT,
)
from langchain.chains.llm import LLMChain
#from langchain.graphs.rdf_graph import RdfGraph
from rdfgraph import RdfGraph
from langchain.prompts.base import BasePromptTemplate
from langchain.schema.language_model import BaseLanguageModel

class MyGraphSparqlQAChain(Chain):
    """
    Chain for question-answering against an RDF or OWL graph by generating
    SPARQL statements.
    """

    graph: RdfGraph = Field(exclude=True)
    sparql_generation_select_chain: LLMChain
    sparql_generation_update_chain: LLMChain
    sparql_intent_chain: LLMChain
    qa_chain: LLMChain
    input_key: str = "query"  #: :meta private:
    output_key: str = "result"  #: :meta private:

    @property
    def input_keys(self) -> List[str]:
        return [self.input_key]

    @property
    def output_keys(self) -> List[str]:
        _output_keys = [self.output_key]
        return _output_keys

    @classmethod
    def from_llm(
        cls,
        llm: BaseLanguageModel,
        *,
        qa_prompt: BasePromptTemplate = SPARQL_QA_PROMPT,
        sparql_select_prompt: BasePromptTemplate = SPARQL_GENERATION_SELECT_PROMPT,
        sparql_update_prompt: BasePromptTemplate = SPARQL_GENERATION_UPDATE_PROMPT,
        sparql_intent_prompt: BasePromptTemplate = SPARQL_INTENT_PROMPT,
        **kwargs: Any,
    ) -> MyGraphSparqlQAChain:
        """Initialize from LLM."""
        qa_chain = LLMChain(llm=llm, prompt=qa_prompt)
        sparql_generation_select_chain = LLMChain(llm=llm, prompt=sparql_select_prompt)
        sparql_generation_update_chain = LLMChain(llm=llm, prompt=sparql_update_prompt)
        sparql_intent_chain = LLMChain(llm=llm, prompt=sparql_intent_prompt)

        return cls(
            qa_chain=qa_chain,
            sparql_generation_select_chain=sparql_generation_select_chain,
            sparql_generation_update_chain=sparql_generation_update_chain,
            sparql_intent_chain=sparql_intent_chain,
            **kwargs,
        )

    def _call(
        self,
        inputs: Dict[str, Any],
        run_manager: Optional[CallbackManagerForChainRun] = None,
    ) -> Dict[str, str]:
        """
        Generate SPARQL query, use it to retrieve a response from the gdb and answer
        the question.
        """
        _run_manager = run_manager or CallbackManagerForChainRun.get_noop_manager()
        callbacks = _run_manager.get_child()
        prompt = inputs[self.input_key]

        _intent = self.sparql_intent_chain.run({"prompt": prompt}, callbacks=callbacks)
        intent = _intent.strip()

        if "SELECT" not in intent and "UPDATE" not in intent:
            raise ValueError(
                "I am sorry, but this prompt seems to fit none of the currently "
                "supported SPARQL query types, i.e., SELECT and UPDATE."
            )
        elif intent.find("SELECT") != -1:
            sparql_generation_chain = self.sparql_generation_select_chain
            intent = "SELECT"
        else:
            sparql_generation_chain = self.sparql_generation_update_chain
            intent = "UPDATE"

        _run_manager.on_text("Identified intent:", end="\n", verbose=self.verbose)
        _run_manager.on_text(intent, color="green", end="\n", verbose=self.verbose)

        generated_sparql = sparql_generation_chain.run(
            {"prompt": prompt, "schema": self.graph.get_schema}, callbacks=callbacks
        )

        _run_manager.on_text("Generated SPARQL:", end="\n", verbose=self.verbose)
        _run_manager.on_text(
            generated_sparql, color="green", end="\n", verbose=self.verbose
        )

        if intent == "SELECT":
            context = self.graph.query(generated_sparql)

            _run_manager.on_text("Full Context:", end="\n", verbose=self.verbose)
            _run_manager.on_text(
                str(context), color="green", end="\n", verbose=self.verbose
            )
            result = self.qa_chain(
                {"prompt": prompt, "context": context},
                callbacks=callbacks,
            )
            res = result[self.qa_chain.output_key]
        elif intent == "UPDATE":
            self.graph.update(generated_sparql)
            res = "Successfully inserted triples into the graph."
        else:
            raise ValueError("Unsupported SPARQL query type.")
        return {self.output_key: res}


src/sparqlchain.py
micheldumontier/sparql-langchain