文档加载与解析：先统一正文和元数据，再进入知识库

python -m pip install "langchain-core>=1,<2" beautifulsoup4 pypdf

import json
from pathlib import Path
from bs4 import BeautifulSoup
from langchain_core.documents import Document
from pypdf import PdfReader


def base_meta(path: Path) -> dict:
    return {
        "source": path.name,
        "file_type": path.suffix.lower(),
        "version": "2026-08-27",
        "access_scope": "training",
    }


def load_one(path: Path) -> list[Document]:
    meta = base_meta(path)
    suffix = path.suffix.lower()
    if suffix in {".txt", ".md"}:
        text = path.read_text(encoding="utf-8")
        return [Document(page_content=text, metadata=meta)]
    if suffix == ".json":
        data = json.loads(path.read_text(encoding="utf-8"))
        text = "\n".join(f"{key}: {value}" for key, value in data.items())
        return [Document(page_content=text, metadata=meta)]
    if suffix == ".html":
        soup = BeautifulSoup(path.read_text(encoding="utf-8"), "html.parser")
        for tag in soup(["script", "style", "nav"]):
            tag.decompose()
        title = soup.title.get_text(strip=True) if soup.title else path.stem
        return [Document(
            page_content=soup.get_text("\n", strip=True),
            metadata={**meta, "title": title},
        )]
    if suffix == ".pdf":
        pages = []
        for index, page in enumerate(PdfReader(path).pages, start=1):
            pages.append(Document(
                page_content=page.extract_text() or "",
                metadata={**meta, "page": index},
            ))
        return pages
    raise ValueError(f"unsupported type: {suffix}")


def load_directory(root: Path) -> tuple[list[Document], list[dict]]:
    documents, errors = [], []
    for path in sorted(root.iterdir()):
        if not path.is_file():
            continue
        try:
            loaded = load_one(path)
            documents.extend(doc for doc in loaded if doc.page_content.strip())
        except Exception as error:
            errors.append({"source": path.name, "error": type(error).__name__})
    return documents, errors


docs, errors = load_directory(Path("sample_docs"))
print([(d.metadata, len(d.page_content)) for d in docs])
print("errors:", errors)