CHROMA_PATH = "chroma" DATA_PATH = "data" from fastapi import FastAPI import argparse import os import shutil from langchain_community.document_loaders.pdf import PyPDFDirectoryLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain.schema.document import Document from get_embedding_function import get_embedding_function from langchain_community.vectorstores import Chroma from langchain.prompts import ChatPromptTemplate from langchain_community.llms.ollama import Ollama from pydantic import BaseModel PROMPT_TEMPLATE = """ Answer the question based only on the following context: {context} --- Answer the question based on the above context: {question} """ app = FastAPI() from langchain_community.embeddings.ollama import OllamaEmbeddings from langchain_community.embeddings.bedrock import BedrockEmbeddings # def get_embedding_function(): # embeddings = BedrockEmbeddings( # credentials_profile_name="default", region_name="us-east-1" # ) embeddings = OllamaEmbeddings(model="nomic-embed-text") return embeddings def load_documents(): document_loader = PyPDFDirectoryLoader(DATA_PATH) return document_loader.load() def split_documents(documents: list[Document]): text_splitter = RecursiveCharacterTextSplitter( chunk_size=800, chunk_overlap=80, length_function=len, is_separator_regex=False, ) return text_splitter.split_documents(documents) def add_to_chroma(chunks: list[Document]): # Load the existing database. db = Chroma( persist_directory=CHROMA_PATH, embedding_function=get_embedding_function() ) # Calculate Page IDs. chunks_with_ids = calculate_chunk_ids(chunks) # Add or Update the documents. existing_items = db.get(include=[]) # IDs are always included by default existing_ids = set(existing_items["ids"]) print(f"Number of existing documents in DB: {len(existing_ids)}") # Only add documents that don't exist in the DB. new_chunks = [] for chunk in chunks_with_ids: if chunk.metadata["id"] not in existing_ids: new_chunks.append(chunk) if len(new_chunks): print(f"👉 Adding new documents: {len(new_chunks)}") new_chunk_ids = [chunk.metadata["id"] for chunk in new_chunks] db.add_documents(new_chunks, ids=new_chunk_ids) db.persist() else: print("✅ No new documents to add") def calculate_chunk_ids(chunks): # This will create IDs like "data/monopoly.pdf:6:2" # Page Source : Page Number : Chunk Index last_page_id = None current_chunk_index = 0 for chunk in chunks: source = chunk.metadata.get("source") page = chunk.metadata.get("page") current_page_id = f"{source}:{page}" # If the page ID is the same as the last one, increment the index. if current_page_id == last_page_id: current_chunk_index += 1 else: current_chunk_index = 0 # Calculate the chunk ID. chunk_id = f"{current_page_id}:{current_chunk_index}" last_page_id = current_page_id # Add it to the page meta-data. chunk.metadata["id"] = chunk_id return chunks def clear_database(): if os.path.exists(CHROMA_PATH): shutil.rmtree(CHROMA_PATH) return {""} @app.get("/") def greet_json(): return {"Hello": "World!"} class QueryRequest(BaseModel): query_text: str @app.get("/train") def train(): # Check if the database should be cleared (using the --clear flag). # parser = argparse.ArgumentParser() # parser.add_argument("--reset", action="store_true", help="Reset the database.") # args = parser.parse_args() # if args.reset: # print("✨ Clearing Database") # clear_database() # Create (or update) the data store. documents = load_documents() chunks = split_documents(documents) add_to_chroma(chunks) @app.get("/cleardb") def cleardb(): # Check if the database should be cleared (using the --clear flag). # parser = argparse.ArgumentParser() # parser.add_argument("--reset", action="store_true", help="Reset the database.") # args = parser.parse_args() # if args.reset: print("✨ Clearing Database") clear_database() @app.get("/query") def query(request: QueryRequest): query_text = request.query_text # Prepare the DB. embedding_function = get_embedding_function() db = Chroma(persist_directory=CHROMA_PATH, embedding_function=embedding_function) # Search the DB. results = db.similarity_search_with_score(query_text, k=5) context_text = "\n\n---\n\n".join([doc.page_content for doc, _score in results]) prompt_template = ChatPromptTemplate.from_template(PROMPT_TEMPLATE) prompt = prompt_template.format(context=context_text, question=query_text) # print(prompt) model = Ollama(model="mistral") response_text = model.invoke(prompt) sources = [doc.metadata.get("id", None) for doc, _score in results] formatted_response = f"Response: {response_text}\nSources: {sources}" print(formatted_response) return response_text