Source code for langchain_community.chains.graph_qa.neptune_sparql
"""Question answering over an RDF or OWL graph using SPARQL."""from__future__importannotationsfromtypingimportAny,Dict,List,Optionalfromlangchain.chains.baseimportChainfromlangchain.chains.llmimportLLMChainfromlangchain_core._api.deprecationimportdeprecatedfromlangchain_core.callbacks.managerimportCallbackManagerForChainRunfromlangchain_core.language_modelsimportBaseLanguageModelfromlangchain_core.prompts.baseimportBasePromptTemplatefromlangchain_core.prompts.promptimportPromptTemplatefrompydanticimportFieldfromlangchain_community.chains.graph_qa.promptsimportSPARQL_QA_PROMPTfromlangchain_community.graphsimportNeptuneRdfGraphINTERMEDIATE_STEPS_KEY="intermediate_steps"SPARQL_GENERATION_TEMPLATE="""Task: Generate a SPARQL SELECT statement for querying a graph database.For instance, to find all email addresses of John Doe, the following query in backticks would be suitable:```PREFIX foaf: <http://xmlns.com/foaf/0.1/>SELECT ?emailWHERE {{ ?person foaf:name "John Doe" . ?person foaf:mbox ?email .}}```Instructions:Use only the node types and properties provided in the schema.Do not use any node types and properties that are not explicitly provided.Include all necessary prefixes.Examples:Schema:{schema}Note: Be as concise as possible.Do not include any explanations or apologies in your responses.Do not respond to any questions that ask for anything else than for you to construct a SPARQL query.Do not include any text except the SPARQL query generated.The question is:{prompt}"""SPARQL_GENERATION_PROMPT=PromptTemplate(input_variables=["schema","prompt"],template=SPARQL_GENERATION_TEMPLATE)
[docs]defextract_sparql(query:str)->str:"""Extract SPARQL code from a text. Args: query: Text to extract SPARQL code from. Returns: SPARQL code extracted from the text. """query=query.strip()querytoks=query.split("```")iflen(querytoks)==3:query=querytoks[1]ifquery.startswith("sparql"):query=query[6:]elifquery.startswith("<sparql>")andquery.endswith("</sparql>"):query=query[8:-9]returnquery
[docs]@deprecated(since="0.3.15",removal="1.0",alternative_import="langchain_aws.create_neptune_sparql_qa_chain",)classNeptuneSparqlQAChain(Chain):"""Chain for question-answering against a Neptune 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. Example: .. code-block:: python chain = NeptuneSparqlQAChain.from_llm( llm=llm, graph=graph ) response = chain.invoke(query) """graph:NeptuneRdfGraph=Field(exclude=True)sparql_generation_chain:LLMChainqa_chain:LLMChaininput_key:str="query"#: :meta private:output_key:str="result"#: :meta private:top_k:int=10return_intermediate_steps:bool=False"""Whether or not to return the intermediate steps along with the final answer."""return_direct:bool=False"""Whether or not to return the result of querying the graph directly."""extra_instructions:Optional[str]=None"""Extra instructions by the appended to the query generation prompt."""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[self.input_key]@propertydefoutput_keys(self)->List[str]:_output_keys=[self.output_key]return_output_keys
[docs]@classmethoddeffrom_llm(cls,llm:BaseLanguageModel,*,qa_prompt:BasePromptTemplate=SPARQL_QA_PROMPT,sparql_prompt:BasePromptTemplate=SPARQL_GENERATION_PROMPT,examples:Optional[str]=None,**kwargs:Any,)->NeptuneSparqlQAChain:"""Initialize from LLM."""qa_chain=LLMChain(llm=llm,prompt=qa_prompt)template_to_use=SPARQL_GENERATION_TEMPLATEifexamples:template_to_use=template_to_use.replace("Examples:","Examples: "+examples)sparql_prompt=PromptTemplate(input_variables=["schema","prompt"],template=template_to_use)sparql_generation_chain=LLMChain(llm=llm,prompt=sparql_prompt)returncls(# type: ignore[call-arg]qa_chain=qa_chain,sparql_generation_chain=sparql_generation_chain,examples=examples,**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]intermediate_steps:List=[]generated_sparql=self.sparql_generation_chain.run({"prompt":prompt,"schema":self.graph.get_schema},callbacks=callbacks)# Extract SPARQLgenerated_sparql=extract_sparql(generated_sparql)_run_manager.on_text("Generated SPARQL:",end="\n",verbose=self.verbose)_run_manager.on_text(generated_sparql,color="green",end="\n",verbose=self.verbose)intermediate_steps.append({"query":generated_sparql})context=self.graph.query(generated_sparql)ifself.return_direct:final_result=contextelse:_run_manager.on_text("Full Context:",end="\n",verbose=self.verbose)_run_manager.on_text(str(context),color="green",end="\n",verbose=self.verbose)intermediate_steps.append({"context":context})result=self.qa_chain({"prompt":prompt,"context":context},callbacks=callbacks,)final_result=result[self.qa_chain.output_key]chain_result:Dict[str,Any]={self.output_key:final_result}ifself.return_intermediate_steps:chain_result[INTERMEDIATE_STEPS_KEY]=intermediate_stepsreturnchain_result