"""Question answering over an RDF or OWL graph using SPARQL."""from__future__importannotationsfromtypingimportAny,Dict,List,Optionalfromlangchain.chains.baseimportChainfromlangchain.chains.llmimportLLMChainfromlangchain_core.callbacksimportCallbackManagerForChainRunfromlangchain_core.language_modelsimportBaseLanguageModelfromlangchain_core.prompts.baseimportBasePromptTemplatefrompydanticimportFieldfromlangchain_community.chains.graph_qa.promptsimport(SPARQL_GENERATION_SELECT_PROMPT,SPARQL_GENERATION_UPDATE_PROMPT,SPARQL_INTENT_PROMPT,SPARQL_QA_PROMPT,)fromlangchain_community.graphs.rdf_graphimportRdfGraph
[docs]classGraphSparqlQAChain(Chain):"""Question-answering against an RDF or OWL graph by generating SPARQL statements. *Security note*: Make sure that the database connection uses credentials that are narrowly-scoped to only include necessary permissions. Failure to do so may result in data corruption or loss, since the calling code may attempt commands that would result in deletion, mutation of data if appropriately prompted or reading sensitive data if such data is present in the database. The best way to guard against such negative outcomes is to (as appropriate) limit the permissions granted to the credentials used with this tool. See https://python.langchain.com/docs/security for more information. """graph:RdfGraph=Field(exclude=True)sparql_generation_select_chain:LLMChainsparql_generation_update_chain:LLMChainsparql_intent_chain:LLMChainqa_chain:LLMChainreturn_sparql_query:bool=Falseinput_key:str="query"#: :meta private:output_key:str="result"#: :meta private:sparql_query_key:str="sparql_query"#: :meta private:allow_dangerous_requests:bool=False"""Forced user opt-in to acknowledge that the chain can make dangerous requests. *Security note*: Make sure that the database connection uses credentials that are narrowly-scoped to only include necessary permissions. Failure to do so may result in data corruption or loss, since the calling code may attempt commands that would result in deletion, mutation of data if appropriately prompted or reading sensitive data if such data is present in the database. The best way to guard against such negative outcomes is to (as appropriate) limit the permissions granted to the credentials used with this tool. See https://python.langchain.com/docs/security for more information. """def__init__(self,**kwargs:Any)->None:"""Initialize the chain."""super().__init__(**kwargs)ifself.allow_dangerous_requestsisnotTrue:raiseValueError("In order to use this chain, you must acknowledge that it can make ""dangerous requests by setting `allow_dangerous_requests` to `True`.""You must narrowly scope the permissions of the database connection ""to only include necessary permissions. Failure to do so may result ""in data corruption or loss or reading sensitive data if such data is ""present in the database.""Only use this chain if you understand the risks and have taken the ""necessary precautions. ""See https://python.langchain.com/docs/security for more information.")@propertydefinput_keys(self)->List[str]:"""Return the input keys. :meta private: """return[self.input_key]@propertydefoutput_keys(self)->List[str]:"""Return the output keys. :meta private: """_output_keys=[self.output_key]return_output_keys
[docs]@classmethoddeffrom_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,)->GraphSparqlQAChain:"""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)returncls(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_managerorCallbackManagerForChainRun.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"inintentand"UPDATE"notinintent:sparql_generation_chain=self.sparql_generation_select_chainintent="SELECT"elif"UPDATE"inintentand"SELECT"notinintent:sparql_generation_chain=self.sparql_generation_update_chainintent="UPDATE"else:raiseValueError("I am sorry, but this prompt seems to fit none of the currently ""supported SPARQL query types, i.e., SELECT and 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)ifintent=="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]elifintent=="UPDATE":self.graph.update(generated_sparql)res="Successfully inserted triples into the graph."else:raiseValueError("Unsupported SPARQL query type.")chain_result:Dict[str,Any]={self.output_key:res}ifself.return_sparql_query:chain_result[self.sparql_query_key]=generated_sparqlreturnchain_result