| import bs4 |
| from langchain_community.document_loaders import WebBaseLoader |
| from langchain_core.vectorstores import InMemoryVectorStore |
| from langchain_openai import OpenAIEmbeddings |
| from langchain_text_splitters import RecursiveCharacterTextSplitter |
| from loguru import logger |
|
|
| vector_store: InMemoryVectorStore = None |
|
|
|
|
| def load_web_content_to_vector_store(web_post_url): |
| bs4_strainer = bs4.SoupStrainer(class_=("post-title", "post-header", "post-content")) |
| loader = WebBaseLoader( |
| web_paths=(web_post_url,), |
| bs_kwargs={"parse_only": bs4_strainer}, |
| ) |
| docs = loader.load() |
| assert len(docs) == 1 |
| text_splitter = RecursiveCharacterTextSplitter( |
| chunk_size=1000, chunk_overlap=200, add_start_index=True |
| ) |
| all_splits = text_splitter.split_documents(docs) |
| logger.info(f"Split blog post into {len(all_splits)} sub-documents.") |
| document_ids = vector_store.add_documents(documents=all_splits) |
| logger.info(f"Loaded {len(document_ids)} to vector store.") |
|
|
|
|
| def initialize_vector_store(web_post_url): |
| |
| logger.info("Loading Vector Store.....") |
| embeddings = OpenAIEmbeddings(model="text-embedding-3-large") |
| global vector_store |
| vector_store = InMemoryVectorStore(embeddings) |
| logger.info("Initialized InMemoryVectorStore") |
| load_web_content_to_vector_store(web_post_url) |
| logger.info("Loaded Web Content to the vector store...") |
|
|
|
|
| def get_vector_store(): |
| return vector_store |
|
|
