import os
from typing import List, Tuple
from langchain.embeddings.openai import OpenAIEmbeddings
from langchain.vectorstores import FAISS
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
import json
from langchain.prompts import PromptTemplate  # Import PromptTemplate


def load_faq_data(data_path: str) -> List[Tuple[str, str]]:
    """
    Loads FAQ data from a JSON file.

    Args:
        data_path: Path to the JSON file.

    Returns:
        A list of tuples, where each tuple contains a question and its answer.
    """
    try:
        with open(data_path, "r", encoding="utf-8") as f:
            faq_data = json.load(f)
        if not isinstance(faq_data, list):
            raise ValueError("Expected a list of dictionaries in the JSON file.")
        for item in faq_data:
            if not isinstance(item, dict) or "question" not in item or "answer" not in item:
                raise ValueError(
                    "Each item in the list should be a dictionary with 'question' and 'answer' keys."
                )
        return [(item["question"], item["answer"]) for item in faq_data]
    except Exception as e:
        print(f"Error loading FAQ data from {data_path}: {e}")
        return []


def chunk_faq_data(faq_data: List[Tuple[str, str]]) -> List[str]:
    """
    Splits the FAQ data into chunks.  Each chunk contains one question and answer.

    Args:
        faq_data: A list of tuples, where each tuple contains a question and its answer.

    Returns:
        A list of strings, where each string is a question and answer concatenated.
    """
    return [f"Question: {q}\nAnswer: {a}" for q, a in faq_data]



def create_embeddings(chunks: List[str]) -> OpenAIEmbeddings:
    """
    Creates embeddings for the text chunks using OpenAI.

    Args:
        chunks: A list of text chunks.

    Returns:
        An OpenAIEmbeddings object.
    """
    return OpenAIEmbeddings()



def create_vector_store(chunks: List[str], embeddings: OpenAIEmbeddings) -> FAISS:
    """
    Creates a vector store from the text chunks and embeddings using FAISS.

    Args:
        chunks: A list of text chunks.
        embeddings: An OpenAIEmbeddings object.

    Returns:
        A FAISS vector store.
    """
    return FAISS.from_texts(chunks, embeddings)



def create_rag_chain(vector_store: FAISS, llm: OpenAI) -> RetrievalQA:
    """
    Creates a RAG chain using the vector store and a language model.
    Adjusted for FAQ format.

    Args:
        vector_store: A FAISS vector store.
        llm: An OpenAI language model.

    Returns:
        A RetrievalQA chain.
    """
    prompt_template = """Use the following pieces of context to answer the question.
    If you don't know the answer, just say that you don't know, don't try to make up an answer.

    Context:
    {context}

    Question:
    {question}

    Helpful Answer:"""


    PROMPT = PromptTemplate(template=prompt_template, input_variables=["context", "question"])

    return RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=vector_store.as_retriever(),
        chain_type_kwargs={"prompt": PROMPT},
        return_source_documents=True,
    )



def rag_query(rag_chain: RetrievalQA, query: str) -> str:
    """
    Queries the RAG chain.

    Args:
        rag_chain: A RetrievalQA chain.
        query: The query string.

    Returns:
        The answer from the RAG chain.
    """
    result = rag_chain(query)
    return result["result"]



def main(data_path: str, query: str) -> str:
    """
    Main function to run the RAG process with FAQ data and OpenAI.

    Args:
        data_path: Path to the JSON file.
        query: The query string.

    Returns:
        The answer to the query using RAG.
    """
    faq_data = load_faq_data(data_path)
    if not faq_data:
        return "No data loaded. Please check the data path."
    chunks = chunk_faq_data(faq_data)
    embeddings = create_embeddings(chunks)
    vector_store = create_vector_store(chunks, embeddings)
    llm = OpenAI(temperature=0)
    rag_chain = create_rag_chain(vector_store, llm)
    answer = rag_query(rag_chain, query)
    return answer



if __name__ == "__main__":
    # Example usage
    data_path = "data/faq.json"
    query = "What is the return policy?"
    answer = main(data_path, query)
    print(f"Query: {query}")
    print(f"Answer: {answer}")

Code Explanation: RAG with FAQ and OpenAI

This Python code implements a Retrieval Augmented Generation (RAG) system specifically designed to answer questions from an FAQ dataset using OpenAI’s language models. Here’s a step-by-step explanation of the code:

1. Import Libraries:

• os: Used for interacting with the operating system, specifically for accessing environment variables (like your OpenAI API key).
• typing: Used for type hinting, which improves code readability and helps with error checking.
• langchain: A framework for developing applications powered by language models. It provides modules for various tasks, including:
  • OpenAIEmbeddings: For generating numerical representations (embeddings) of text using OpenAI.
  • FAISS: For creating and managing a vector store, which allows for efficient similarity search.
  • RetrievalQA: For creating a retrieval-based question answering chain.
  • OpenAI: For interacting with OpenAI’s language models.
  • PromptTemplate: For creating reusable prompt structures.
• json: For working with JSON data, as the FAQ data is expected to be in JSON format.

2. load_faq_data(data_path):

• Loads FAQ data from a JSON file.
• It expects the JSON file to contain a list of dictionaries, where each dictionary has a "question" and an "answer" key.
• It performs error handling to ensure the file exists and the data is in the correct format.
• It returns a list of tuples, where each tuple contains a question and its corresponding answer.

3. chunk_faq_data(faq_data):

• Prepares the FAQ data for embedding.
• Each FAQ question-answer pair is treated as a single chunk.
• It formats each question-answer pair into a string like "Question: {q}\nAnswer: {a}".
• It returns a list of these formatted strings.

4. create_embeddings(chunks):

• Uses OpenAI’s OpenAIEmbeddings to convert the text chunks (from the FAQ data) into numerical vectors (embeddings).
• Embeddings capture the semantic meaning of the text.

5. create_vector_store(chunks, embeddings):

• Creates a vector store using FAISS.
• The vector store stores the text chunks along with their corresponding embeddings.
• FAISS enables efficient similarity search.

6. create_rag_chain(vector_store, llm):

• Creates the RAG chain, combining the vector store with a language model.
• It uses Langchain’s RetrievalQA chain:
  • Retrieves relevant chunks from the vector_store based on the query.
  • Feeds the retrieved chunks and the query to the llm (OpenAI).
  • The LLM generates an answer.
• It uses a custom PromptTemplate to structure the input to the LLM, telling it to answer from the context and say “I don’t know” if the answer isn’t present.
• It sets return_source_documents=True to include the retrieved source documents in the output.

7. rag_query(rag_chain, query):

• Takes the RAG chain and a user query as input.
• Runs the query against the chain to get the answer.
• Extracts the answer from the result.

8. main(data_path, query):

• Orchestrates the RAG process:
  • Loads the FAQ data.
  • Prepares the data into chunks.
  • Creates embeddings and the vector store.
  • Creates the RAG chain using OpenAI.
  • Runs the query and prints the result.

In essence, this code automates answering questions from an FAQ by:

Using a language model to generate answers based on the most relevant FAQ entries.

Loading and formatting the FAQ data.

Converting the FAQ entries into a searchable format.

To use this code with your FAQ data:

1. Create a JSON file:
   • Create a JSON file (e.g., faq.json) with your FAQ data in the following format:
   JSON[ {"question": "What is your return policy?", "answer": "We accept returns within 30 days of purchase."}, {"question": "How do I track my order?", "answer": "You can track your order using the tracking number provided in your shipping confirmation email."}, {"question": "What are your shipping costs?", "answer": "Shipping costs vary depending on the shipping method and destination."} ]
2. Replace "data/faq.json":
   • In the if __name__ == "__main__": block, replace "data/faq.json" with the actual path to your JSON file.
3. Modify the query:
   • Change the query variable to ask a question from your FAQ data.
4. Run the code:
   • Run the Python script. It will load the FAQ data, create a vector store, and answer your query.

import os
from typing import List, Tuple
from langchain.embeddings.openai import OpenAIEmbeddings
from langchain.vectorstores import FAISS
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI, HuggingFacePipeline  # Import HuggingFacePipeline
from transformers import pipeline  # Import pipeline from transformers

# Load environment variables (replace with your actual API key or use a .env file)
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"  # Remove OpenAI API key
#  No longer needed, but keep for user's reference, in case they want to switch back.

def load_data(data_path: str) -> str:
    """
    Loads data from a file.  Supports text, and markdown.  For other file types,
    add appropriate loaders.

    Args:
        data_path: Path to the data file.

    Returns:
        The loaded data as a string.
    """
    try:
        with open(data_path, "r", encoding="utf-8") as f:
            data = f.read()
        return data
    except Exception as e:
        print(f"Error loading data from {data_path}: {e}")
        return ""

def chunk_data(data: str, chunk_size: int = 1000, chunk_overlap: int = 200) -> List[str]:
    """
    Splits the data into chunks.

    Args:
        data: The data to be chunked.
        chunk_size: The size of each chunk.
        chunk_overlap: The overlap between chunks.

    Returns:
        A list of text chunks.
    """
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size, chunk_overlap=chunk_overlap
    )
    chunks = text_splitter.split_text(data)
    return chunks

def create_embeddings(chunks: List[str]) -> OpenAIEmbeddings:
    """
    Creates embeddings for the text chunks using OpenAI.

    Args:
        chunks: A list of text chunks.

    Returns:
        An OpenAIEmbeddings object.
    """
    embeddings = OpenAIEmbeddings()  #  Still using OpenAI embeddings for now,
    return embeddings                  #  but could be replaced with a local alternative.

def create_vector_store(
    chunks: List[str], embeddings: OpenAIEmbeddings
) -> FAISS:
    """
    Creates a vector store from the text chunks and embeddings using FAISS.

    Args:
        chunks: A list of text chunks.
        embeddings: An OpenAIEmbeddings object.

    Returns:
        A FAISS vector store.
    """
    vector_store = FAISS.from_texts(chunks, embeddings)
    return vector_store

def create_rag_chain(
    vector_store: FAISS,
    llm,  # Type hint as base LLM, can be either OpenAI or HuggingFacePipeline
) -> RetrievalQA:
    """
    Creates a RAG chain using the vector store and a language model.

    Args:
        vector_store: A FAISS vector store.
        llm: A language model (OpenAI or HuggingFace pipeline).

    Returns:
        A RetrievalQA chain.
    """
    rag_chain = RetrievalQA.from_chain_type(
        llm=llm, chain_type="stuff", retriever=vector_store.as_retriever()
    )
    return rag_chain

def rag_query(rag_chain: RetrievalQA, query: str) -> str:
    """
    Queries the RAG chain.

    Args:
        rag_chain: A RetrievalQA chain.
        query: The query string.

    Returns:
        The answer from the RAG chain.
    """
    answer = rag_chain.run(query)
    return answer

def main(data_path: str, query: str, use_local_llm: bool = False) -> str:
    """
    Main function to run the RAG process.  Now supports local LLMs.

    Args:
        data_path: Path to the data file.
        query: The query string.
        use_local_llm:  Flag to use a local LLM (Hugging Face).
            If False, uses OpenAI.  Defaults to False.

    Returns:
        The answer to the query using RAG.
    """
    data = load_data(data_path)
    if not data:
        return "No data loaded. Please check the data path."
    chunks = chunk_data(data)
    embeddings = create_embeddings(chunks)
    vector_store = create_vector_store(chunks, embeddings)

    if use_local_llm:
        #  Example of using a local LLM from Hugging Face.
        #  You'll need to choose a model and ensure you have the
        #  necessary libraries installed (transformers, etc.).
        #  This example uses a small, fast model; you'll likely want
        #  a larger one for better quality.  You may need to adjust
        #  the model name and device (CPU/GPU) depending on your system.
        local_llm = pipeline(
            "text-generation",
            model="distilgpt2",  #  A small, fast model for demonstration.
            device="cpu",  #  Use "cuda" for GPU if available.
            max_length=200,  #  Limit the output length.
        )
        llm = HuggingFacePipeline(pipeline=local_llm)
    else:
        llm = OpenAI(temperature=0)  # Use OpenAI if use_local_llm is False

    rag_chain = create_rag_chain(vector_store, llm)
    answer = rag_query(rag_chain, query)
    return answer

if __name__ == "__main__":
    # Example usage
    data_path = "data/my_data.txt"  # Replace with your data file
    query = "What is the main topic of this document?"
    use_local_llm = True  # Set to True to use a local LLM, False for OpenAI
    answer = main(data_path, query, use_local_llm)
    print(f"Query: {query}")
    print(f"Answer: {answer}")

Sample code to enable running the LLM locally. This will involve using a local LLM instead of OpenAI.

Key Changes:

• Imported HuggingFacePipeline and pipeline: These are needed to load and use a local LLM from Hugging Face.
• Conditional LLM Loading: The main function now takes a use_local_llm argument. It uses an if statement to choose between loading an OpenAI LLM or a local Hugging Face LLM.
• Hugging Face Pipeline Example: The code includes an example of how to load and configure a local LLM using the pipeline function from transformers. This example uses distilgpt2, a small, fast model for demonstration purposes. You’ll likely want to replace this with a more capable model.
• device Argument: The device argument in the pipeline function is set to “cpu”. If you have a GPU, change this to “cuda” for significantly faster performance.
• Removed OpenAI Key Requirement: The os.environ[“OPENAI_API_KEY”] line has been commented out because it’s no longer needed when using a local LLM. I’ve kept it in the code, commented out, as a helpful reminder for users who may want to switch back to using OpenAI.
• Added use_local_llm to main and if __name__: The main function now accepts a boolean use_local_llm argument to determine whether to use a local LLM or OpenAI. The example usage in if __name__ now includes setting this flag.

To run this code with a local LLM:

1. Install transformers: If you don’t have it already, install the transformers library: pip install transformers.
2. Choose a Model: Select a suitable LLM from Hugging Face (https://huggingface.co/models). The example code uses “distilgpt2”, but you’ll likely want a larger, more powerful model for better results. Consider models like gpt-2, gpt-j, or others that fit your hardware and needs.
3. Modify Model Name: Replace “distilgpt2” in the code with the name of the model you’ve chosen.
4. Set Device: If you have a GPU, change device=”cpu” to device=”cuda” for faster inference.
5. Data Path and Query: Make sure data_path points to your data file and that query contains the question you want to ask.
6. Run the Code: Run the Python script. The first time you run it with a new model, it will download the model files, which may take some time.

Important Considerations:

• Model Size and Hardware: Local LLMs can be very large, and running them efficiently requires significant hardware resources, especially RAM and GPU memory. Choose a model that fits your system’s capabilities.
• Dependencies: Ensure you have all the necessary libraries installed, including transformers, torch (if using a GPU), and any other dependencies required by the specific model you choose.
• Performance: Local LLMs may run slower than cloud-based LLMs like OpenAI, especially if you don’t have a powerful GPU.
• Accuracy: The accuracy and quality of the results will depend on the specific local LLM you choose. Smaller, faster models may not be as accurate as larger ones.