forked from Archives/langchain
Harrion/kg (#1016)
Co-authored-by: William FH <13333726+hinthornw@users.noreply.github.com>makefile-update-1
Harrison Chase
1 year ago
committed by
GitHub
parent
78abd277ff
commit
0c553d2064
@ -0,0 +1 @@
|
||||
"""Question answering over a knowledge graph."""
|
||||
@ -0,0 +1,78 @@
|
||||
"""Question answering over a graph."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Dict, List
|
||||
|
||||
from pydantic import Field
|
||||
|
||||
from langchain.chains.base import Chain
|
||||
from langchain.chains.graph_qa.prompts import ENTITY_EXTRACTION_PROMPT, PROMPT
|
||||
from langchain.chains.llm import LLMChain
|
||||
from langchain.graphs.networkx_graph import NetworkxEntityGraph, get_entities
|
||||
from langchain.llms.base import BaseLLM
|
||||
from langchain.prompts.base import BasePromptTemplate
|
||||
|
||||
|
||||
class GraphQAChain(Chain):
|
||||
"""Chain for question-answering against a graph."""
|
||||
|
||||
graph: NetworkxEntityGraph = Field(exclude=True)
|
||||
entity_extraction_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 the input keys.
|
||||
|
||||
:meta private:
|
||||
"""
|
||||
return [self.input_key]
|
||||
|
||||
@property
|
||||
def output_keys(self) -> List[str]:
|
||||
"""Return the output keys.
|
||||
|
||||
:meta private:
|
||||
"""
|
||||
_output_keys = [self.output_key]
|
||||
return _output_keys
|
||||
|
||||
@classmethod
|
||||
def from_llm(
|
||||
cls,
|
||||
llm: BaseLLM,
|
||||
qa_prompt: BasePromptTemplate = PROMPT,
|
||||
entity_prompt: BasePromptTemplate = ENTITY_EXTRACTION_PROMPT,
|
||||
**kwargs: Any,
|
||||
) -> GraphQAChain:
|
||||
"""Initialize from LLM."""
|
||||
qa_chain = LLMChain(llm=llm, prompt=qa_prompt)
|
||||
entity_chain = LLMChain(llm=llm, prompt=entity_prompt)
|
||||
|
||||
return cls(qa_chain=qa_chain, entity_extraction_chain=entity_chain, **kwargs)
|
||||
|
||||
def _call(self, inputs: Dict[str, str]) -> Dict[str, Any]:
|
||||
"""Extract entities, look up info and answer question."""
|
||||
question = inputs[self.input_key]
|
||||
|
||||
entity_string = self.entity_extraction_chain.run(question)
|
||||
|
||||
self.callback_manager.on_text(
|
||||
"Entities Extracted:", end="\n", verbose=self.verbose
|
||||
)
|
||||
self.callback_manager.on_text(
|
||||
entity_string, color="green", end="\n", verbose=self.verbose
|
||||
)
|
||||
entities = get_entities(entity_string)
|
||||
context = ""
|
||||
for entity in entities:
|
||||
triplets = self.graph.get_entity_knowledge(entity)
|
||||
context += "\n".join(triplets)
|
||||
self.callback_manager.on_text("Full Context:", end="\n", verbose=self.verbose)
|
||||
self.callback_manager.on_text(
|
||||
context, color="green", end="\n", verbose=self.verbose
|
||||
)
|
||||
result = self.qa_chain({"question": question, "context": context})
|
||||
return {self.output_key: result[self.qa_chain.output_key]}
|
||||
@ -0,0 +1,34 @@
|
||||
# flake8: noqa
|
||||
from langchain.prompts.prompt import PromptTemplate
|
||||
|
||||
_DEFAULT_ENTITY_EXTRACTION_TEMPLATE = """Extract all entities from the following text. As a guideline, a proper noun is generally capitalized. You should definitely extract all names and places.
|
||||
|
||||
Return the output as a single comma-separated list, or NONE if there is nothing of note to return.
|
||||
|
||||
EXAMPLE
|
||||
i'm trying to improve Langchain's interfaces, the UX, its integrations with various products the user might want ... a lot of stuff.
|
||||
Output: Langchain
|
||||
END OF EXAMPLE
|
||||
|
||||
EXAMPLE
|
||||
i'm trying to improve Langchain's interfaces, the UX, its integrations with various products the user might want ... a lot of stuff. I'm working with Sam.
|
||||
Output: Langchain, Sam
|
||||
END OF EXAMPLE
|
||||
|
||||
Begin!
|
||||
|
||||
{input}
|
||||
Output:"""
|
||||
ENTITY_EXTRACTION_PROMPT = PromptTemplate(
|
||||
input_variables=["input"], template=_DEFAULT_ENTITY_EXTRACTION_TEMPLATE
|
||||
)
|
||||
|
||||
prompt_template = """Use the following knowledge triplets to answer the question at the end. If you don't know the answer, just say that you don't know, don't try to make up an answer.
|
||||
|
||||
{context}
|
||||
|
||||
Question: {question}
|
||||
Helpful Answer:"""
|
||||
PROMPT = PromptTemplate(
|
||||
template=prompt_template, input_variables=["context", "question"]
|
||||
)
|
||||
@ -0,0 +1,4 @@
|
||||
"""Graph implementations."""
|
||||
from langchain.graphs.networkx_graph import NetworkxEntityGraph
|
||||
|
||||
__all__ = ["NetworkxEntityGraph"]
|
||||
@ -0,0 +1,96 @@
|
||||
"""Networkx wrapper for graph operations."""
|
||||
|
||||
from typing import List, NamedTuple, Tuple
|
||||
|
||||
KG_TRIPLE_DELIMITER = "<|>"
|
||||
|
||||
|
||||
class KnowledgeTriple(NamedTuple):
|
||||
"""A triple in the graph."""
|
||||
|
||||
subject: str
|
||||
predicate: str
|
||||
object_: str
|
||||
|
||||
@classmethod
|
||||
def from_string(cls, triple_string: str) -> "KnowledgeTriple":
|
||||
"""Create a KnowledgeTriple from a string."""
|
||||
subject, predicate, object_ = triple_string.strip().split(", ")
|
||||
subject = subject[1:]
|
||||
object_ = object_[:-1]
|
||||
return cls(subject, predicate, object_)
|
||||
|
||||
|
||||
def parse_triples(knowledge_str: str) -> List[KnowledgeTriple]:
|
||||
"""Parse knowledge triples from the knowledge string."""
|
||||
knowledge_str = knowledge_str.strip()
|
||||
if not knowledge_str or knowledge_str == "NONE":
|
||||
return []
|
||||
triple_strs = knowledge_str.split(KG_TRIPLE_DELIMITER)
|
||||
results = []
|
||||
for triple_str in triple_strs:
|
||||
try:
|
||||
kg_triple = KnowledgeTriple.from_string(triple_str)
|
||||
except ValueError:
|
||||
continue
|
||||
results.append(kg_triple)
|
||||
return results
|
||||
|
||||
|
||||
def get_entities(entity_str: str) -> List[str]:
|
||||
"""Extract entities from entity string."""
|
||||
if entity_str.strip() == "NONE":
|
||||
return []
|
||||
else:
|
||||
return [w.strip() for w in entity_str.split(",")]
|
||||
|
||||
|
||||
class NetworkxEntityGraph:
|
||||
"""Networkx wrapper for entity graph operations."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Create a new graph."""
|
||||
import networkx as nx
|
||||
|
||||
self._graph = nx.DiGraph()
|
||||
|
||||
def add_triple(self, knowledge_triple: KnowledgeTriple) -> None:
|
||||
"""Add a triple to the graph."""
|
||||
# Creates nodes if they don't exist
|
||||
# Overwrites existing edges
|
||||
if not self._graph.has_node(knowledge_triple.subject):
|
||||
self._graph.add_node(knowledge_triple.subject)
|
||||
if not self._graph.has_node(knowledge_triple.object_):
|
||||
self._graph.add_node(knowledge_triple.object_)
|
||||
self._graph.add_edge(
|
||||
knowledge_triple.subject,
|
||||
knowledge_triple.object_,
|
||||
relation=knowledge_triple.predicate,
|
||||
)
|
||||
|
||||
def delete_triple(self, knowledge_triple: KnowledgeTriple) -> None:
|
||||
"""Delete a triple from the graph."""
|
||||
if self._graph.has_edge(knowledge_triple.subject, knowledge_triple.object_):
|
||||
self._graph.remove_edge(knowledge_triple.subject, knowledge_triple.object_)
|
||||
|
||||
def get_triples(self) -> List[Tuple[str, str, str]]:
|
||||
"""Get all triples in the graph."""
|
||||
return [(u, v, d["relation"]) for u, v, d in self._graph.edges(data=True)]
|
||||
|
||||
def get_entity_knowledge(self, entity: str, depth: int = 1) -> List[str]:
|
||||
"""Get information about an entity."""
|
||||
import networkx as nx
|
||||
|
||||
# TODO: Have more information-specific retrieval methods
|
||||
if not self._graph.has_node(entity):
|
||||
return []
|
||||
|
||||
results = []
|
||||
for src, sink in nx.dfs_edges(self._graph, entity, depth_limit=depth):
|
||||
relation = self._graph[src][sink]["relation"]
|
||||
results.append(f"{src} {relation} {sink}")
|
||||
return results
|
||||
|
||||
def clear(self) -> None:
|
||||
"""Clear the graph."""
|
||||
self._graph.clear()
|
||||
@ -0,0 +1,4 @@
|
||||
"""All index utils."""
|
||||
from langchain.indexes.graph import GraphIndexCreator
|
||||
|
||||
__all__ = ["GraphIndexCreator"]
|
||||
@ -0,0 +1,30 @@
|
||||
"""Graph Index Creator."""
|
||||
from typing import Optional, Type
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
from langchain.chains.llm import LLMChain
|
||||
from langchain.graphs.networkx_graph import NetworkxEntityGraph, parse_triples
|
||||
from langchain.indexes.prompts.knowledge_triplet_extraction import (
|
||||
KNOWLEDGE_TRIPLE_EXTRACTION_PROMPT,
|
||||
)
|
||||
from langchain.llms.base import BaseLLM
|
||||
|
||||
|
||||
class GraphIndexCreator(BaseModel):
|
||||
"""Functionality to create graph index."""
|
||||
|
||||
llm: Optional[BaseLLM] = None
|
||||
graph_type: Type[NetworkxEntityGraph] = NetworkxEntityGraph
|
||||
|
||||
def from_text(self, text: str) -> NetworkxEntityGraph:
|
||||
"""Create graph index from text."""
|
||||
if self.llm is None:
|
||||
raise ValueError("llm should not be None")
|
||||
graph = self.graph_type()
|
||||
chain = LLMChain(llm=self.llm, prompt=KNOWLEDGE_TRIPLE_EXTRACTION_PROMPT)
|
||||
output = chain.predict(text=text)
|
||||
knowledge = parse_triples(output)
|
||||
for triple in knowledge:
|
||||
graph.add_triple(triple)
|
||||
return graph
|
||||
@ -0,0 +1 @@
|
||||
"""Relevant prompts for constructing indexes."""
|
||||
@ -0,0 +1,40 @@
|
||||
# flake8: noqa
|
||||
from langchain.prompts.prompt import PromptTemplate
|
||||
|
||||
_DEFAULT_ENTITY_EXTRACTION_TEMPLATE = """You are an AI assistant reading the transcript of a conversation between an AI and a human. Extract all of the proper nouns from the last line of conversation. As a guideline, a proper noun is generally capitalized. You should definitely extract all names and places.
|
||||
|
||||
The conversation history is provided just in case of a coreference (e.g. "What do you know about him" where "him" is defined in a previous line) -- ignore items mentioned there that are not in the last line.
|
||||
|
||||
Return the output as a single comma-separated list, or NONE if there is nothing of note to return (e.g. the user is just issuing a greeting or having a simple conversation).
|
||||
|
||||
EXAMPLE
|
||||
Conversation history:
|
||||
Person #1: how's it going today?
|
||||
AI: "It's going great! How about you?"
|
||||
Person #1: good! busy working on Langchain. lots to do.
|
||||
AI: "That sounds like a lot of work! What kind of things are you doing to make Langchain better?"
|
||||
Last line:
|
||||
Person #1: i'm trying to improve Langchain's interfaces, the UX, its integrations with various products the user might want ... a lot of stuff.
|
||||
Output: Langchain
|
||||
END OF EXAMPLE
|
||||
|
||||
EXAMPLE
|
||||
Conversation history:
|
||||
Person #1: how's it going today?
|
||||
AI: "It's going great! How about you?"
|
||||
Person #1: good! busy working on Langchain. lots to do.
|
||||
AI: "That sounds like a lot of work! What kind of things are you doing to make Langchain better?"
|
||||
Last line:
|
||||
Person #1: i'm trying to improve Langchain's interfaces, the UX, its integrations with various products the user might want ... a lot of stuff. I'm working with Person #2.
|
||||
Output: Langchain, Person #2
|
||||
END OF EXAMPLE
|
||||
|
||||
Conversation history (for reference only):
|
||||
{history}
|
||||
Last line of conversation (for extraction):
|
||||
Human: {input}
|
||||
|
||||
Output:"""
|
||||
ENTITY_EXTRACTION_PROMPT = PromptTemplate(
|
||||
input_variables=["history", "input"], template=_DEFAULT_ENTITY_EXTRACTION_TEMPLATE
|
||||
)
|
||||
@ -0,0 +1,25 @@
|
||||
# flake8: noqa
|
||||
from langchain.prompts.prompt import PromptTemplate
|
||||
|
||||
_DEFAULT_ENTITY_SUMMARIZATION_TEMPLATE = """You are an AI assistant helping a human keep track of facts about relevant people, places, and concepts in their life. Update the summary of the provided entity in the "Entity" section based on the last line of your conversation with the human. If you are writing the summary for the first time, return a single sentence.
|
||||
The update should only include facts that are relayed in the last line of conversation about the provided entity, and should only contain facts about the provided entity.
|
||||
|
||||
If there is no new information about the provided entity or the information is not worth noting (not an important or relevant fact to remember long-term), return the existing summary unchanged.
|
||||
|
||||
Full conversation history (for context):
|
||||
{history}
|
||||
|
||||
Entity to summarize:
|
||||
{entity}
|
||||
|
||||
Existing summary of {entity}:
|
||||
{summary}
|
||||
|
||||
Last line of conversation:
|
||||
Human: {input}
|
||||
Updated summary:"""
|
||||
|
||||
ENTITY_SUMMARIZATION_PROMPT = PromptTemplate(
|
||||
input_variables=["entity", "summary", "history", "input"],
|
||||
template=_DEFAULT_ENTITY_SUMMARIZATION_TEMPLATE,
|
||||
)
|
||||
@ -0,0 +1,37 @@
|
||||
# flake8: noqa
|
||||
|
||||
from langchain.graphs.networkx_graph import KG_TRIPLE_DELIMITER
|
||||
from langchain.prompts.prompt import PromptTemplate
|
||||
|
||||
_DEFAULT_KNOWLEDGE_TRIPLE_EXTRACTION_TEMPLATE = (
|
||||
"You are a networked intelligence helping a human track knowledge triples"
|
||||
" about all relevant people, things, concepts, etc. and integrating"
|
||||
" them with your knowledge stored within your weights"
|
||||
" as well as that stored in a knowledge graph."
|
||||
" Extract all of the knowledge triples from the text."
|
||||
" A knowledge triple is a clause that contains a subject, a predicate,"
|
||||
" and an object. The subject is the entity being described,"
|
||||
" the predicate is the property of the subject that is being"
|
||||
" described, and the object is the value of the property.\n\n"
|
||||
"EXAMPLE\n"
|
||||
"It's a state in the US. It's also the number 1 producer of gold in the US.\n\n"
|
||||
f"Output: (Nevada, is a, state){KG_TRIPLE_DELIMITER}(Nevada, is in, US)"
|
||||
f"{KG_TRIPLE_DELIMITER}(Nevada, is the number 1 producer of, gold)\n"
|
||||
"END OF EXAMPLE\n\n"
|
||||
"EXAMPLE\n"
|
||||
"I'm going to the store.\n\n"
|
||||
"Output: NONE\n"
|
||||
"END OF EXAMPLE\n\n"
|
||||
"EXAMPLE\n"
|
||||
"Oh huh. I know Descartes likes to drive antique scooters and play the mandolin.\n"
|
||||
f"Output: (Descartes, likes to drive, antique scooters){KG_TRIPLE_DELIMITER}(Descartes, plays, mandolin)\n"
|
||||
"END OF EXAMPLE\n\n"
|
||||
"EXAMPLE\n"
|
||||
"{text}"
|
||||
"Output:"
|
||||
)
|
||||
|
||||
KNOWLEDGE_TRIPLE_EXTRACTION_PROMPT = PromptTemplate(
|
||||
input_variables=["text"],
|
||||
template=_DEFAULT_KNOWLEDGE_TRIPLE_EXTRACTION_TEMPLATE,
|
||||
)
|
||||
Loading…
Reference in New Issue