旧 RAG 入库代码升级到 v1：加载、切分、向量库怎么换，结果变不变

imports: new
langchain-core 1.7.0 | langchain-community 0.4.2 | langchain-text-splitters 1.1.3

load   2 documents, metadata keys ['source']
       fixtures/docs/ingestion-steps.txt: 275 chars
       fixtures/docs/query-notes.txt: 217 chars
split  7 chunks, length 60-78, metadata keys ['source', 'start_index']
       chunk digest fc3ef762639f
store  InMemoryVectorStore, 7 chunks indexed
query  迁移之后怎么确认入库行为没有变化
       1. fixtures/docs/query-notes.txt @ 150: 迁移之后要用同一批文档重新入库，
       2. fixtures/docs/ingestion-steps.txt @ 0: 入库是 RAG 的第一步。先把原
       3. fixtures/docs/ingestion-steps.txt @ 67: 文档加载阶段要处理文件编码。中文

faiss  from langchain_community.vectorstores import FAISS
       import ok
       from_documents -> ImportError: Could not import faiss python package. Please install it with `pip install faiss-gpu` (for CUDA supported GPU) or `pip install faiss-cpu` (depending on Python version).

run digest 290baf4f7e84

langchain-classic 1.0.8 | langchain-community 0.4.2

from langchain_classic.document_loaders import TextLoader
    import: ok
    warning: DeprecationWarning: langchain-community is being sunset
    warning: LangChainDeprecationWarning, suggests: from langchain_community.document_loaders import TextLoader

from langchain_classic.text_splitter import RecursiveCharacterTextSplitter
    import: ok
    warning: (none)

from langchain_classic.vectorstores import FAISS
    import: ok
    warning: LangChainDeprecationWarning, suggests: from langchain_community.vectorstores import FAISS

langchain-community 0.3.31
warnings on import: 0

langchain-community 0.4.2
warnings on import: 1
  DeprecationWarning: `langchain-community` is being sunset and is no longer actively maintained. See https://github.com/langchain-ai/langchain-community/issues/674 for details and migration guidance toward standalone integration packages.

langchain-community 0.4.2

encoding='utf-8'          RuntimeError: Error loading gbk_note.txt
                          caused by UnicodeDecodeError: 'utf-8' codec can't decode byte 0xd6 in position 0: invalid continuation byte
encoding='gbk'            ok: '中文资料常见的编码有 UTF-8 和 GBK。'
autodetect_encoding=True  ModuleNotFoundError: No module named 'chardet'
                          caused by UnicodeDecodeError: 'utf-8' codec can't decode byte 0xd6 in position 0: invalid continuation byte

"""Run the ingestion chain and print what it produced.

usage: python ingest_pipeline.py [new|old|plain]

  new    loader, splitter and FAISS come from their new import locations (default)
  old    the locations that old tutorials use (0.3.x only)
  plain  no langchain-community at all: the splitter comes from langchain-text-splitters and the
         plain-text loader is a few lines of application code on top of langchain-core

Run it in a 0.3.x and in a 1.x environment. The documents, the chunks and the ranking must be the
same in every run, and the closing digest makes that visible. The embeddings are deterministic
fake vectors, so the ranking says nothing about retrieval quality; it is only compared between
runs. The FAISS lines depend on the installed packages, so they stay out of the digest. Run it
from this directory so that the source paths are relative.
"""
import hashlib
import sys
from importlib.metadata import PackageNotFoundError, version
from pathlib import Path

from langchain_core.documents import Document
from langchain_core.embeddings import DeterministicFakeEmbedding
from langchain_core.vectorstores import InMemoryVectorStore

MODE = sys.argv[1] if len(sys.argv) > 1 else "new"
if MODE not in {"new", "old", "plain"}:
    raise SystemExit(__doc__)

if MODE == "old":
    from langchain.document_loaders import TextLoader
    from langchain.text_splitter import RecursiveCharacterTextSplitter

    FAISS_IMPORT = "from langchain.vectorstores import FAISS"
else:
    from langchain_text_splitters import RecursiveCharacterTextSplitter

    FAISS_IMPORT = "from langchain_community.vectorstores import FAISS"
    if MODE == "new":
        from langchain_community.document_loaders import TextLoader

QUERY = "迁移之后怎么确认入库行为没有变化"
lines: list[str] = []


def say(text: str) -> None:
    """Print a line and remember it for the closing digest."""
    print(text)
    lines.append(text)


def installed(name: str) -> str:
    """Return the installed version of a distribution, or '-' when it is not installed."""
    try:
        return version(name)
    except PackageNotFoundError:
        return "-"


def load_file(path: str) -> list[Document]:
    """Load one UTF-8 text file the way TextLoader does: one document, metadata {'source': path}."""
    if MODE == "plain":
        return [Document(page_content=Path(path).read_text(encoding="utf-8"), metadata={"source": path})]
    return TextLoader(path, encoding="utf-8").load()


print(f"imports: {MODE}")
print(
    f"langchain-core {installed('langchain-core')} | langchain-community {installed('langchain-community')}"
    f" | langchain-text-splitters {installed('langchain-text-splitters')}"
)
print()

docs: list[Document] = []
for path in sorted(Path("fixtures/docs").glob("*.txt")):
    docs.extend(load_file(path.as_posix()))
say(f"load   {len(docs)} documents, metadata keys {sorted(docs[0].metadata)}")
for doc in docs:
    say(f"       {doc.metadata['source']}: {len(doc.page_content)} chars")

splitter = RecursiveCharacterTextSplitter(
    chunk_size=100,
    chunk_overlap=20,
    add_start_index=True,
    separators=["\n\n", "\n", "。", "，", " ", ""],
)
chunks = splitter.split_documents(docs)
sizes = [len(c.page_content) for c in chunks]
say(f"split  {len(chunks)} chunks, length {min(sizes)}-{max(sizes)}, metadata keys {sorted(chunks[0].metadata)}")
say(f"       chunk digest {hashlib.sha1(chr(0).join(c.page_content for c in chunks).encode()).hexdigest()[:12]}")

store = InMemoryVectorStore(DeterministicFakeEmbedding(size=32))
ids = store.add_documents(chunks)
say(f"store  InMemoryVectorStore, {len(ids)} chunks indexed")

say(f"query  {QUERY}")
for rank, hit in enumerate(store.similarity_search(QUERY, k=3), start=1):
    say(f"       {rank}. {hit.metadata['source']} @ {hit.metadata['start_index']}: {hit.page_content[:16]}")

print()
if MODE == "plain":
    print(f"langchain_community imported: {'langchain_community' in sys.modules}")
else:
    print(f"faiss  {FAISS_IMPORT}")
    if MODE == "old":
        from langchain.vectorstores import FAISS  # noqa: E402  (the import itself is under test)
    else:
        from langchain_community.vectorstores import FAISS  # noqa: E402

    print("       import ok")
    try:
        FAISS.from_documents(chunks, DeterministicFakeEmbedding(size=32))
    except Exception as exc:  # noqa: BLE001  (the exact exception type is part of the output)
        print(f"       from_documents -> {type(exc).__name__}: {exc}")

print()
print(f"run digest {hashlib.sha1(chr(10).join(lines).encode()).hexdigest()[:12]}")

"""Load a GBK-encoded Chinese text file with TextLoader and show what each encoding setting does.

The file is written to a temporary directory, so the error messages contain a short relative path.
Not passing `encoding` at all is left out on purpose: it uses the operating system's default
encoding, which differs between machines and would make this output differ too.
"""
import os
import tempfile
from importlib.metadata import version
from pathlib import Path

from langchain_community.document_loaders import TextLoader

TEXT = "中文资料常见的编码有 UTF-8 和 GBK。"
CASES = [
    ("encoding='utf-8'", {"encoding": "utf-8"}),
    ("encoding='gbk'", {"encoding": "gbk"}),
    ("autodetect_encoding=True", {"autodetect_encoding": True}),
]

print(f"langchain-community {version('langchain-community')}")
print()

origin = os.getcwd()
with tempfile.TemporaryDirectory() as tmp:
    os.chdir(tmp)
    try:
        Path("gbk_note.txt").write_bytes(TEXT.encode("gbk"))
        for label, kwargs in CASES:
            try:
                docs = TextLoader("gbk_note.txt", **kwargs).load()
                print(f"{label:<26}ok: {docs[0].page_content!r}")
            except Exception as exc:  # noqa: BLE001  (the exact exception type is part of the output)
                print(f"{label:<26}{type(exc).__name__}: {exc}")
                cause = exc.__cause__ or exc.__context__
                if cause is not None:
                    print(f"{'':<26}caused by {type(cause).__name__}: {cause}")
    finally:
        os.chdir(origin)

"""Show whether importing from langchain_community emits a warning, and what it says.

The import runs inside a recording filter, so the warning is printed here instead of going to
stderr, where it would depend on the interpreter's default warning filters.
"""
import warnings
from importlib.metadata import version

with warnings.catch_warnings(record=True) as caught:
    warnings.simplefilter("always")
    from langchain_community.document_loaders import TextLoader  # noqa: F401

print(f"langchain-community {version('langchain-community')}")
print(f"warnings on import: {len(caught)}")
for item in caught:
    print(f"  {item.category.__name__}: {item.message}")

"""Import the three ingestion classes through langchain-classic and show which warnings come with it.

langchain-classic keeps the old `langchain.*` module paths alive. This shows whether it also keeps
the community code behind them, by recording what each import warns about and where the warning
says the class now lives.
"""
import re
import warnings
from importlib.metadata import version

IMPORTS = [
    "from langchain_classic.document_loaders import TextLoader",
    "from langchain_classic.text_splitter import RecursiveCharacterTextSplitter",
    "from langchain_classic.vectorstores import FAISS",
]

print(f"langchain-classic {version('langchain-classic')} | langchain-community {version('langchain-community')}")
for statement in IMPORTS:
    with warnings.catch_warnings(record=True) as caught:
        warnings.simplefilter("always")
        exec(statement)  # noqa: S102  (the statements are the constants above)
    print()
    print(statement)
    print("    import: ok")
    notes: list[str] = []
    for item in caught:
        text = str(item.message)
        if "sunset" in text:
            note = f"{item.category.__name__}: langchain-community is being sunset"
        else:
            suggestion = re.search(r"from langchain_community[\w.]* import \w+", text)
            note = f"{item.category.__name__}, suggests: {suggestion.group(0) if suggestion else text.splitlines()[0]}"
        if note not in notes:
            notes.append(note)
    for note in notes or ["(none)"]:
        print(f"    warning: {note}")