RandomZ / app /ingest /parser_pdf.py
StormShadow308's picture
feat(experiment-lab): rewrite RAG pipeline using LangChain framework
49f0cfb
Raw
History Blame Contribute Delete
1.98 kB
"""Parse .pdf files using the LangChain PyMuPDFLoader.
LangChain replaces the manual pymupdf font-size heuristics and line-by-line
block extraction with ``PyMuPDFLoader`` which returns one LangChain
``Document`` per page. Page-level metadata (``page``, ``source``) is
preserved automatically.
Compared to master branch:
- ✅ Simpler code — no font-size heading heuristics
- ✅ Standard LangChain interface with reliable per-page Documents
- ✅ File handle automatically managed by the loader
- ⚠️ Heading detection not performed (heading levels not in metadata)
"""
import logging
from pathlib import Path
from langchain_community.document_loaders import PyMuPDFLoader
from langchain_core.documents import Document
logger = logging.getLogger(__name__)
def parse_pdf(file_path: Path) -> list[Document]:
"""Load a ``.pdf`` file and return its pages as LangChain Documents.
Uses ``langchain_community.document_loaders.PyMuPDFLoader`` which
extracts text from each page and attaches ``page``, ``source``,
``total_pages``, and ``file_path`` metadata.
Args:
file_path: Absolute path to a ``.pdf`` file.
Returns:
List of :class:`langchain_core.documents.Document` objects, one per
PDF page. Empty pages are included with empty ``page_content``.
Raises:
FileNotFoundError: If ``file_path`` does not exist.
ValueError: If the file cannot be opened as a valid PDF.
Example::
docs = parse_pdf(Path("/uploads/survey.pdf"))
print(f"Loaded {len(docs)} pages")
"""
if not file_path.exists():
raise FileNotFoundError(f"File not found: {file_path}")
try:
loader = PyMuPDFLoader(str(file_path))
docs = loader.load()
except Exception as exc:
raise ValueError(f"Cannot open PDF '{file_path}': {exc}") from exc
logger.debug(
"Parsed %d page(s) from %s via LangChain PyMuPDFLoader", len(docs), file_path.name
)
return docs