Ingestion

nlql.ingest —— 写入管线：Normalizer → 可插拔 Splitter → Embedder（带缓存）→ Indexer。内置规则句子分割（中/英/日 + CJK 标点），可选 pysbd（nlql[segment]）。

ingest

Ingestion: normalization, pluggable splitting, and the write-time pipeline.

Importing this package registers the default SENTENCE and CHUNK splitters.

__all__ module-attribute

__all__ = ['IngestionPipeline', 'Normalizer', 'DefaultNormalizer', 'split_sentences', 'split_chunks', 'detect_language', 'LanguageRouter', 'make_pysbd_splitter']

LanguageRouter

LanguageRouter(routes: dict[str, Splitter], default: Splitter)

A splitter that dispatches to a per-language sub-splitter by detected script.

源代码位于： src/nlql/ingest/language.py

def __init__(self, routes: dict[str, Splitter], default: Splitter) -> None:
    self._routes = dict(routes)
    self._default = default

__call__

__call__(text: str) -> list[str]

源代码位于： src/nlql/ingest/language.py

def __call__(self, text: str) -> list[str]:
    return self._routes.get(detect_language(text), self._default)(text)

DefaultNormalizer

NFC-normalize, collapse spaces/tabs and blank lines, trim each line.

normalize

normalize(text: str) -> str

源代码位于： src/nlql/ingest/normalize.py

def normalize(self, text: str) -> str:
    text = unicodedata.normalize("NFC", text)
    text = self._INLINE_WS.sub(" ", text)
    text = self._MULTI_NL.sub("\n", text)
    text = "\n".join(line.strip() for line in text.split("\n"))
    return text.strip()

Normalizer

Bases: Protocol

normalize

normalize(text: str) -> str

源代码位于： src/nlql/ingest/normalize.py

def normalize(self, text: str) -> str: ...

IngestionPipeline

IngestionPipeline(embedder: Embedder, *, registry: Registry = GLOBAL_REGISTRY, normalizer: Normalizer | None = None, granularity: str = 'sentence')

Turns documents into embedded units at a configured base granularity.

源代码位于： src/nlql/ingest/pipeline.py

def __init__(
    self,
    embedder: Embedder,
    *,
    registry: Registry = GLOBAL_REGISTRY,
    normalizer: Normalizer | None = None,
    granularity: str = "sentence",
) -> None:
    self._embedder = embedder
    self._registry = registry
    self._normalizer = normalizer or DefaultNormalizer()
    self._granularity = granularity

granularity property

granularity: str

process

process(documents: Iterable[Document]) -> list[Unit]

Normalize, split, and embed documents into units (vectors attached).

Text payloads are split into segments; image payloads become one unit each and are embedded via the embedder's embed_images when it is multimodal (otherwise they are skipped). Both paths share the same index — only the vector's origin differs.

源代码位于： src/nlql/ingest/pipeline.py

def process(self, documents: Iterable[Document]) -> list[Unit]:
    """Normalize, split, and embed documents into units (vectors attached).

    Text payloads are split into segments; image payloads become one unit each and are
    embedded via the embedder's ``embed_images`` when it is multimodal (otherwise they
    are skipped). Both paths share the same index — only the vector's origin differs.
    """
    splitter = self._splitter()
    can_embed_images = supports_images(self._embedder)
    units: list[Unit] = []
    text_units: list[Unit] = []
    texts: list[str] = []
    image_units: list[Unit] = []
    images: list[bytes | str] = []

    for doc in documents:
        ordinal = 0
        for payload in doc.payloads:
            if payload.is_text:
                for segment in splitter(self._normalizer.normalize(payload.as_text)):
                    unit = Unit(
                        id=f"{doc.id}#{self._granularity}:{ordinal}",
                        doc_id=doc.id,
                        kind=self._granularity,  # type: ignore[arg-type]
                        payload=Payload.text(segment),
                        metadata=dict(doc.metadata),  # copy so units never share/mutate
                        ordinal=ordinal,
                    )
                    units.append(unit)
                    text_units.append(unit)
                    texts.append(segment)
                    ordinal += 1
            elif payload.modality is Modality.IMAGE and can_embed_images:
                unit = Unit(
                    id=f"{doc.id}#{self._granularity}:{ordinal}",
                    doc_id=doc.id,
                    kind=self._granularity,  # type: ignore[arg-type]
                    payload=payload,
                    metadata=dict(doc.metadata),
                    ordinal=ordinal,
                )
                units.append(unit)
                image_units.append(unit)
                images.append(payload.data)
                ordinal += 1
            # else: unsupported modality → skipped

    if texts:
        for unit, vector in zip(text_units, self._embedder.embed(texts), strict=True):
            unit.vector = vector
    if images:
        image_vectors = self._embedder.embed_images(images)  # type: ignore[attr-defined]
        for unit, vector in zip(image_units, image_vectors, strict=True):
            unit.vector = vector
    return units

detect_language

detect_language(text: str) -> str

Coarse script detection: "cjk" if the text is substantially CJK, else "latin".

源代码位于： src/nlql/ingest/language.py

def detect_language(text: str) -> str:
    """Coarse script detection: ``"cjk"`` if the text is substantially CJK, else ``"latin"``."""
    cjk = sum(1 for ch in text if "　" <= ch <= "鿿" or "＀" <= ch <= "￯")
    total = sum(1 for ch in text if not ch.isspace())
    return "cjk" if total and cjk / total > 0.2 else "latin"

make_pysbd_splitter

make_pysbd_splitter(language: str = 'en') -> Splitter

A robust sentence splitter backed by pysbd (handles abbreviations).

Needs pip install pysbd. Register it as SENTENCE (globally or per engine) to replace the rule-based default.

源代码位于： src/nlql/ingest/language.py

def make_pysbd_splitter(language: str = "en") -> Splitter:
    """A robust sentence splitter backed by pysbd (handles abbreviations).

    Needs ``pip install pysbd``. Register it as ``SENTENCE`` (globally or per engine) to
    replace the rule-based default.
    """
    try:
        import pysbd
    except ImportError as e:  # pragma: no cover - exercised only without pysbd
        raise NLQLError("pysbd not installed: pip install pysbd") from e

    segmenter = pysbd.Segmenter(language=language, clean=False)

    def split(text: str) -> list[str]:
        return [s.strip() for s in segmenter.segment(text) if s and s.strip()]

    return split

split_chunks

split_chunks(text: str, max_chars: int = 1000) -> list[str]

Group whole sentences into chunks no longer than max_chars characters.

源代码位于： src/nlql/ingest/splitters.py

def split_chunks(text: str, max_chars: int = 1000) -> list[str]:
    """Group whole sentences into chunks no longer than ``max_chars`` characters."""
    text = text.strip()
    if not text:
        return []
    chunks: list[str] = []
    current = ""
    for sentence in split_sentences(text):
        if current and len(current) + len(sentence) + 1 > max_chars:
            chunks.append(current)
            current = sentence
        else:
            current = f"{current} {sentence}".strip() if current else sentence
    if current:
        chunks.append(current)
    return chunks or [text]

split_sentences

split_sentences(text: str) -> list[str]

Split text into sentences using punctuation and newline boundaries.

源代码位于： src/nlql/ingest/splitters.py

def split_sentences(text: str) -> list[str]:
    """Split text into sentences using punctuation and newline boundaries."""
    text = text.strip()
    if not text:
        return []
    parts = _SENTENCE_BOUNDARY.split(text)
    return [p.strip() for p in parts if p and p.strip()]