LangChainとOpenAIによる質問応答

!python3 -m pip install -qU langchain langchain-elasticsearch openai==0.28.1 tiktoken jq

from langchain_elasticsearch import ElasticsearchStore
from langchain.embeddings.openai import OpenAIEmbeddings
from getpass import getpass

# https://www.elastic.co/search-labs/tutorials/install-elasticsearch/elastic-cloud#finding-your-cloud-id
ELASTIC_CLOUD_ID = getpass("Elastic Cloud ID: ")

# https://www.elastic.co/search-labs/tutorials/install-elasticsearch/elastic-cloud#creating-an-api-key
ELASTIC_API_KEY = getpass("Elastic Api Key: ")

# https://platform.openai.com/api-keys
OPENAI_API_KEY = getpass("OpenAI API key: ")

embeddings = OpenAIEmbeddings(openai_api_key=OPENAI_API_KEY)

vector_store = ElasticsearchStore(
    es_cloud_id=ELASTIC_CLOUD_ID,
    es_api_key=ELASTIC_API_KEY,
    index_name="workplace_index",
    embedding=embeddings,
)

from urllib.request import urlopen
from langchain.llms import OpenAI
import json

url = "https://raw.githubusercontent.com/elastic/elasticsearch-labs/main/example-apps/chatbot-rag-app/data/data.json"

response = urlopen(url)
data = json.load(response)

with open("temp.json", "w") as json_file:
    json.dump(data, json_file)

from langchain.document_loaders import JSONLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter


def metadata_func(record: dict, metadata: dict) -> dict:
    metadata["name"] = record.get("name")
    metadata["summary"] = record.get("summary")
    metadata["url"] = record.get("url")
    metadata["category"] = record.get("category")
    metadata["updated_at"] = record.get("updated_at")

    return metadata


# For more loaders https://python.langchain.com/docs/modules/data_connection/document_loaders/
# And 3rd party loaders https://python.langchain.com/docs/modules/data_connection/document_loaders/#third-party-loaders
loader = JSONLoader(
    file_path="temp.json",
    jq_schema=".[]",
    content_key="content",
    metadata_func=metadata_func,
)

text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=512, chunk_overlap=256
)
docs = loader.load_and_split(text_splitter=text_splitter)

documents = vector_store.from_documents(
    docs,
    embeddings,
    index_name="workplace_index",
    es_cloud_id=ELASTIC_CLOUD_ID,
    es_api_key=ELASTIC_API_KEY,
)

from langchain.schema.runnable import RunnablePassthrough
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser

retriever = vector_store.as_retriever()

llm = OpenAI(openai_api_key=OPENAI_API_KEY)

ANSWER_PROMPT = ChatPromptTemplate.from_template(
    """You are an assistant for question-answering tasks. Use the following pieces of retrieved context to answer the question. If you don't know the answer, just say that you don't know. Be as verbose and educational in your response as possible.

    context: {context}
    Question: "{question}"
    Answer:
    """
)

chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | ANSWER_PROMPT
    | llm
    | StrOutputParser()
)

ans = chain.invoke("what is the nasa sales team?")

print("---- Answer ----")
print(ans)

from langchain.schema.runnable import RunnablePassthrough
from langchain.prompts import ChatPromptTemplate, PromptTemplate
from langchain.schema.output_parser import StrOutputParser
from langchain.schema import format_document

retriever = vector_store.as_retriever()

llm = OpenAI(openai_api_key=OPENAI_API_KEY)

ANSWER_PROMPT = ChatPromptTemplate.from_template(
    """
You are an assistant for question-answering tasks. Use the following pieces of retrieved context to answer the question. If you don't know the answer, just say that you don't know. Be as verbose and educational in your response as possible.
Each passage has a SOURCE which is the title of the document. When answering, cite source name of the passages you are answering from below the answer, on a new line, with a prefix of "SOURCE:".


context: {context}
Question: "{question}"
Answer:
"""
)

DOCUMENT_PROMPT = PromptTemplate.from_template(
    """
---
SOURCE: {name}
{page_content}
---
"""
)


def _combine_documents(
    docs, document_prompt=DOCUMENT_PROMPT, document_separator="\n\n"
):
    doc_strings = [format_document(doc, document_prompt) for doc in docs]
    return document_separator.join(doc_strings)


_context = {
    "context": retriever | _combine_documents,
    "question": RunnablePassthrough(),
}

chain = _context | ANSWER_PROMPT | llm | StrOutputParser()

ans = chain.invoke("what is the nasa sales team?")

print("---- Answer ----")
print(ans)

from langchain.schema.runnable import RunnableMap
from langchain.prompts import ChatPromptTemplate, PromptTemplate
from langchain.schema import format_document
from operator import itemgetter

retriever = vector_store.as_retriever()

llm = OpenAI(openai_api_key=OPENAI_API_KEY)

ANSWER_PROMPT = ChatPromptTemplate.from_template(
    """
You are an assistant for question-answering tasks. Use the following pieces of retrieved context to answer the question. If you don't know the answer, just say that you don't know. Be as verbose and educational in your response as possible.
Each passage has a SOURCE which is the title of the document. When answering, cite source name of the passages you are answering from below the answer, on a new line, with a prefix of "SOURCE:".

context: {context}
Question: {question}
Answer:

"""
)

DOCUMENT_PROMPT = PromptTemplate.from_template(
    """
---
SOURCE: {name}
{page_content}
---
"""
)


def _combine_documents(
    docs, document_prompt=DOCUMENT_PROMPT, document_separator="\n\n"
):
    doc_strings = [format_document(doc, document_prompt) for doc in docs]
    return document_separator.join(doc_strings)


retrieved_documents = RunnableMap(
    docs=itemgetter("question") | retriever,
    question=itemgetter("question"),
)

_context = {
    "context": lambda x: _combine_documents(x["docs"]),
    "question": lambda x: x["question"],
}

answer = {
    "answer": _context | ANSWER_PROMPT | llm,
    "docs": itemgetter("docs"),
}

chain = retrieved_documents | answer

ans = chain.invoke({"question": "what is the nasa sales team?"})

print("---- Answer ----")
print(ans["answer"])
print()
print("---- Documents ----")
for doc in ans["docs"]:
    print(doc.metadata["name"])
    print(doc.page_content)
    print("----")

from langchain.schema.runnable import RunnableMap
from langchain.prompts import ChatPromptTemplate, PromptTemplate
from langchain.schema import format_document
from operator import itemgetter

retriever = vector_store.as_retriever()

llm = OpenAI(openai_api_key=OPENAI_API_KEY)

ANSWER_PROMPT = ChatPromptTemplate.from_template(
    """
You are an assistant for question-answering tasks. Use the following pieces of retrieved context to answer the question. If you don't know the answer, just say that you don't know. Be as verbose and educational in your response as possible.
Each passage has a SOURCE which is the title of the document. When answering, cite source name of the passages you are answering from below the answer, on a new line, with a prefix of "SOURCE:".

context:
{context}

Question: {question}
Answer:
"""
)

DOCUMENT_PROMPT = PromptTemplate.from_template(
    """
---
SOURCE: {name}
{page_content}
---
"""
)

CONDENSE_QUESTION_PROMPT = PromptTemplate.from_template(
    """Given the following conversation and a follow up question, rephrase the follow up question to be a standalone question, in its original language.

Chat History:
{chat_history}
Follow Up Input: {question}
"""
)

standalone_question = RunnableMap(
    standalone_question=RunnablePassthrough.assign(
        chat_history=lambda x: _format_chat_history(x["chat_history"])
    )
    | CONDENSE_QUESTION_PROMPT
    | llm
    | StrOutputParser(),
)


def _format_chat_history(chat_history) -> str:
    buffer = ""
    for dialogue_turn in chat_history:
        human = "Human: " + dialogue_turn[0]
        ai = "Assistant: " + dialogue_turn[1]
        buffer += "\n" + "\n".join([human, ai])
    return buffer


def _combine_documents(
    docs, document_prompt=DOCUMENT_PROMPT, document_separator="\n\n"
):
    doc_strings = [format_document(doc, document_prompt) for doc in docs]
    return document_separator.join(doc_strings)


retrieved_documents = RunnableMap(
    docs=itemgetter("standalone_question") | retriever,
    question=itemgetter("standalone_question"),
)

_context = {
    "context": lambda x: _combine_documents(x["docs"]),
    "question": lambda x: x["question"],
}

answer = {
    "answer": _context | ANSWER_PROMPT | llm,
    "docs": itemgetter("docs"),
}

chain = standalone_question | retrieved_documents | answer

ans = chain.invoke(
    {
        "question": "What are their objectives?",
        "chat_history": [
            "What is the nasa sales team?",
            "The sales team of NASA consists of Laura Martinez, the Area "
            "Vice-President of North America, and Gary Johnson, the Area "
            "Vice-President of South America."
            "SOURCE: Sales Organization Overview",
        ],
    }
)

print("---- Answer ----")
print(ans["answer"])