mirror of
https://github.com/hanzoai/chat-rag-api.git
synced 2026-10-30 18:45:41 +00:00
zeekay
74110d03f8
Extracted from the deployed chat-rag-api:0.7.8-hanzo image; adds a Qdrant vector store targeting hanzoai/vector as a drop-in for pgvector. - app/services/vector_store/qdrant_vector.py: QdrantVector(QdrantVectorStore) with the id helpers (get_all_ids/get_filtered_ids/get_documents_by_ids), delete-by-file_id, and the route's Mongo-style file_id filter translated to native Qdrant filters. file_id stored in payload metadata (unique point ids). - factory.get_vector_store: mode='qdrant' (lazy import; pg/atlas deploys unaffected). - config: VectorDBType.QDRANT + QDRANT_URL/QDRANT_API_KEY/QDRANT_DISTANCE/ EMBEDDINGS_DIM(1536) + construction branch. - health: qdrant_health_check (client.get_collections()). - requirements: langchain-qdrant==0.2.0, qdrant-client==1.12.1. Verified: pytest tests/test_qdrant_vector.py (4 tests) against in-memory Qdrant; factory builds QdrantVector with all route methods present.
94 lines
3.1 KiB
Python
94 lines
3.1 KiB
Python
# app/utils/document_loader.py
|
|
from typing import List, Optional
|
|
|
|
from langchain_core.documents import Document
|
|
|
|
from app.config import known_source_ext, PDF_EXTRACT_IMAGES, CHUNK_OVERLAP
|
|
from langchain_community.document_loaders import (
|
|
TextLoader,
|
|
PyPDFLoader,
|
|
CSVLoader,
|
|
Docx2txtLoader,
|
|
UnstructuredEPubLoader,
|
|
UnstructuredMarkdownLoader,
|
|
UnstructuredXMLLoader,
|
|
UnstructuredRSTLoader,
|
|
UnstructuredExcelLoader,
|
|
UnstructuredPowerPointLoader,
|
|
)
|
|
|
|
def get_loader(filename: str, file_content_type: str, filepath: str):
|
|
file_ext = filename.split(".")[-1].lower()
|
|
known_type = True
|
|
|
|
if file_ext == "pdf":
|
|
loader = PyPDFLoader(filepath, extract_images=PDF_EXTRACT_IMAGES)
|
|
elif file_ext == "csv":
|
|
loader = CSVLoader(filepath)
|
|
elif file_ext == "rst":
|
|
loader = UnstructuredRSTLoader(filepath, mode="elements")
|
|
elif file_ext == "xml":
|
|
loader = UnstructuredXMLLoader(filepath)
|
|
elif file_ext == "pptx":
|
|
loader = UnstructuredPowerPointLoader(filepath)
|
|
elif file_ext == "md":
|
|
loader = UnstructuredMarkdownLoader(filepath)
|
|
elif file_content_type == "application/epub+zip":
|
|
loader = UnstructuredEPubLoader(filepath)
|
|
elif (
|
|
file_content_type
|
|
== "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
|
|
or file_ext in ["doc", "docx"]
|
|
):
|
|
loader = Docx2txtLoader(filepath)
|
|
elif file_content_type in [
|
|
"application/vnd.ms-excel",
|
|
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
] or file_ext in ["xls", "xlsx"]:
|
|
loader = UnstructuredExcelLoader(filepath)
|
|
elif file_content_type == "application/json" or file_ext == "json":
|
|
loader = TextLoader(filepath, autodetect_encoding=True)
|
|
elif file_ext in known_source_ext or (
|
|
file_content_type and file_content_type.find("text/") >= 0
|
|
):
|
|
loader = TextLoader(filepath, autodetect_encoding=True)
|
|
else:
|
|
loader = TextLoader(filepath, autodetect_encoding=True)
|
|
known_type = False
|
|
|
|
return loader, known_type, file_ext
|
|
|
|
def clean_text(text: str) -> str:
|
|
"""
|
|
Remove NUL (0x00) characters from a string.
|
|
|
|
:param text: The original text with potential NUL characters.
|
|
:return: Cleaned text without NUL characters.
|
|
"""
|
|
return text.replace("\x00", "")
|
|
|
|
def process_documents(documents: List[Document]) -> str:
|
|
processed_text = ""
|
|
last_page: Optional[int] = None
|
|
doc_basename = ""
|
|
|
|
for doc in documents:
|
|
if "source" in doc.metadata:
|
|
doc_basename = doc.metadata["source"].split("/")[-1]
|
|
break
|
|
|
|
processed_text += f"{doc_basename}\n"
|
|
|
|
for doc in documents:
|
|
current_page = doc.metadata.get("page")
|
|
if current_page and current_page != last_page:
|
|
processed_text += f"\n# PAGE {doc.metadata['page']}\n\n"
|
|
last_page = current_page
|
|
|
|
new_content = doc.page_content
|
|
if processed_text.endswith(new_content[:CHUNK_OVERLAP]):
|
|
processed_text += new_content[CHUNK_OVERLAP:]
|
|
else:
|
|
processed_text += new_content
|
|
|
|
return processed_text.strip() |