SDK

nlql.sdk —— Engine（应用主入口）与流式 QueryBuilder。NLQL 语句、链式构造、LLM IR 三种查询入口均通过 Engine 执行，编译到同一份 IR。

Engine 通过构造参数接收 embedder、store、reranker 等组件；OpenAI 兼容渠道直接使用 OpenAIEmbedder(base_url=...)。

sdk

High-level SDK: the Engine and the Query Builder.

content module-attribute

content = E(Path('content'))

__all__ module-attribute

__all__ = ['Engine', 'QueryBuilder', 'select', 'E', 'F', 'Meta', 'content', 'field', 'similarity', 'contains', 'length']

E

E(expr: Expr)

A fluent wrapper over an IR expression, overloading Python operators.

源代码位于： src/nlql/sdk/builder.py

def __init__(self, expr: Expr) -> None:
    self.expr = expr

__hash__ class-attribute instance-attribute

__hash__ = None

expr instance-attribute

expr = expr

__ge__

__ge__(other: Any) -> E

源代码位于： src/nlql/sdk/builder.py

def __ge__(self, other: Any) -> E:
    return E(Compare(">=", self.expr, _lift(other)))

__gt__

__gt__(other: Any) -> E

源代码位于： src/nlql/sdk/builder.py

def __gt__(self, other: Any) -> E:
    return E(Compare(">", self.expr, _lift(other)))

__le__

__le__(other: Any) -> E

源代码位于： src/nlql/sdk/builder.py

def __le__(self, other: Any) -> E:
    return E(Compare("<=", self.expr, _lift(other)))

__lt__

__lt__(other: Any) -> E

源代码位于： src/nlql/sdk/builder.py

def __lt__(self, other: Any) -> E:
    return E(Compare("<", self.expr, _lift(other)))

__eq__

__eq__(other: Any) -> E

源代码位于： src/nlql/sdk/builder.py

def __eq__(self, other: Any) -> E:  # type: ignore[override]
    return E(Compare("==", self.expr, _lift(other)))

__ne__

__ne__(other: Any) -> E

源代码位于： src/nlql/sdk/builder.py

def __ne__(self, other: Any) -> E:  # type: ignore[override]
    return E(Compare("!=", self.expr, _lift(other)))

__and__

__and__(other: Any) -> E

源代码位于： src/nlql/sdk/builder.py

def __and__(self, other: Any) -> E:
    rhs = _lift(other)
    if isinstance(self.expr, And):
        return E(And([*self.expr.operands, rhs]))
    return E(And([self.expr, rhs]))

__or__

__or__(other: Any) -> E

源代码位于： src/nlql/sdk/builder.py

def __or__(self, other: Any) -> E:
    rhs = _lift(other)
    if isinstance(self.expr, Or):
        return E(Or([*self.expr.operands, rhs]))
    return E(Or([self.expr, rhs]))

__invert__

__invert__() -> E

源代码位于： src/nlql/sdk/builder.py

def __invert__(self) -> E:
    return E(Not(self.expr))

contains

contains(text: Any) -> E

源代码位于： src/nlql/sdk/builder.py

def contains(self, text: Any) -> E:
    return E(Call("CONTAINS", [self.expr, _lift(text)]))

matches

matches(pattern: Any) -> E

源代码位于： src/nlql/sdk/builder.py

def matches(self, pattern: Any) -> E:
    return E(Call("MATCH", [self.expr, _lift(pattern)]))

like

like(pattern: Any) -> E

源代码位于： src/nlql/sdk/builder.py

def like(self, pattern: Any) -> E:
    return E(Call("LIKE", [self.expr, _lift(pattern)]))

QueryBuilder

QueryBuilder(unit: str, window: int | None = None)

Accumulates clauses and compiles to a :class:~nlql.ir.nodes.Query.

源代码位于： src/nlql/sdk/builder.py

def __init__(self, unit: str, window: int | None = None) -> None:
    self._select = Select(unit, window)
    self._let: list[Binding] = []
    self._where: list[Expr] = []
    self._order: list[OrderKey] = []
    self._limit: int | None = None

let

let(name: str, expr: Any) -> QueryBuilder

源代码位于： src/nlql/sdk/builder.py

def let(self, name: str, expr: Any) -> QueryBuilder:
    self._let.append(Binding(name, _lift(expr)))
    return self

where

where(*conditions: Any) -> QueryBuilder

源代码位于： src/nlql/sdk/builder.py

def where(self, *conditions: Any) -> QueryBuilder:
    self._where.extend(_lift(c) for c in conditions)
    return self

order_by

order_by(key: Any, *, desc: bool = False) -> QueryBuilder

源代码位于： src/nlql/sdk/builder.py

def order_by(self, key: Any, *, desc: bool = False) -> QueryBuilder:
    expr = Ref(key) if isinstance(key, str) else _lift(key)
    self._order.append(OrderKey(expr, desc=desc))
    return self

limit

limit(n: int) -> QueryBuilder

源代码位于： src/nlql/sdk/builder.py

def limit(self, n: int) -> QueryBuilder:
    self._limit = n
    return self

build

build() -> Query

源代码位于： src/nlql/sdk/builder.py

def build(self) -> Query:
    where: Expr | None
    if not self._where:
        where = None
    elif len(self._where) == 1:
        where = self._where[0]
    else:
        where = And(list(self._where))
    return Query(
        select=self._select,
        let=self._let,
        where=where,
        order_by=self._order,
        limit=self._limit,
    )

Engine

Engine(embedder: Embedder, *, store: Store | None = None, registry: Registry | None = None, granularity: str = 'chunk', normalizer: Normalizer | None = None, cache: EmbeddingCache | None = None, field_types: dict[str, TypeTag] | None = None, type_handlers: dict | None = None, reranker: Reranker | None = None, rerank_factor: int = 5, named_embedders: dict[str, Embedder] | None = None)

A retrieval engine: ingest documents, run NLQL queries, extend capabilities.

The embedder is injected — it is the extensibility point, so there are no provider-specific constructors to multiply. Any OpenAI-compatible channel is one :class:~nlql.embed.OpenAIEmbedder parameterized by base_url; any other provider is just another :class:~nlql.embed.base.Embedder implementation::

from nlql import Engine
from nlql.embed import FakeEmbedder, OpenAIEmbedder

Engine(OpenAIEmbedder(base_url="https://your-gateway/v1", api_key="sk-..."))
Engine(FakeEmbedder())                       # deterministic, offline (tests/demos)
Engine(MyCohereEmbedder(...))                # your own backend, no core change

源代码位于： src/nlql/sdk/engine.py

def __init__(
    self,
    embedder: Embedder,
    *,
    store: Store | None = None,
    registry: Registry | None = None,
    granularity: str = "chunk",
    normalizer: Normalizer | None = None,
    cache: EmbeddingCache | None = None,
    field_types: dict[str, TypeTag] | None = None,
    type_handlers: dict | None = None,
    reranker: Reranker | None = None,
    rerank_factor: int = 5,
    named_embedders: dict[str, Embedder] | None = None,
) -> None:
    self._type_handlers: dict = type_handlers or {}
    self._registry = registry if registry is not None else GLOBAL_REGISTRY.child()
    self._embedder: Embedder = (
        embedder if isinstance(embedder, CachedEmbedder) else CachedEmbedder(embedder, cache)
    )
    # Named vectors (SIMILARITY(vec.<name>, …)) each use their own embedder, cached too.
    self._named_embedders: dict[str, Embedder] = {
        name: (e if isinstance(e, CachedEmbedder) else CachedEmbedder(e))
        for name, e in (named_embedders or {}).items()
    }
    # NOTE: `is not None`, not `or` — an empty store is falsy (len 0) and `or` would
    # silently swap in a LocalStore, ignoring the injected backend.
    self._store: Store = store if store is not None else LocalStore()
    self._granularity = granularity
    self._pipeline = IngestionPipeline(
        self._embedder, registry=self._registry, normalizer=normalizer, granularity=granularity
    )
    self._executor = Executor(
        self._store,
        self._registry,
        self._embedder,
        granularity=granularity,
        field_types=field_types,
        type_handlers=self._type_handlers,
        reranker=reranker,
        rerank_factor=rerank_factor,
        named_embedders=self._named_embedders,
    )
    self._parser = NLQLParser()

store property

store: Store

registry property

registry: Registry

granularity property

granularity: str

add

add(document: Document) -> None

Ingest a single document.

源代码位于： src/nlql/sdk/engine.py

def add(self, document: Document) -> None:
    """Ingest a single document."""
    self.add_documents([document])

add_text

add_text(text: str, *, id: str | None = None, metadata: dict[str, Any] | None = None, source: str | None = None) -> str

Ingest a single text; returns the (generated or provided) document id.

源代码位于： src/nlql/sdk/engine.py

def add_text(
    self,
    text: str,
    *,
    id: str | None = None,
    metadata: dict[str, Any] | None = None,
    source: str | None = None,
) -> str:
    """Ingest a single text; returns the (generated or provided) document id."""
    doc_id = id or uuid.uuid4().hex
    self.add(Document.from_text(text, id=doc_id, metadata=metadata, source=source))
    return doc_id

add_image

add_image(image: bytes | str, *, id: str | None = None, metadata: dict[str, Any] | None = None, source: str | None = None, mime: str | None = None) -> str

Ingest a single image as a multimodal document; returns its id.

image may be raw bytes, a local file path, or an http(s) / data: URL. Needs a multimodal embedder (e.g. DoubaoVisionEmbedder or ClipEmbedder) — the image is embedded into the same space as text, so text queries retrieve it. Use granularity="chunk" for image collections.

源代码位于： src/nlql/sdk/engine.py

def add_image(
    self,
    image: bytes | str,
    *,
    id: str | None = None,
    metadata: dict[str, Any] | None = None,
    source: str | None = None,
    mime: str | None = None,
) -> str:
    """Ingest a single image as a multimodal document; returns its id.

    ``image`` may be raw ``bytes``, a local file path, or an ``http(s)`` / ``data:`` URL.
    Needs a multimodal embedder (e.g. ``DoubaoVisionEmbedder`` or ``ClipEmbedder``) — the
    image is embedded into the same space as text, so text queries retrieve it. Use
    ``granularity="chunk"`` for image collections.
    """
    if isinstance(image, (bytes, bytearray)):
        data: bytes | str = bytes(image)
    elif image.startswith(("http://", "https://", "data:")):
        data = image  # URL / data URI — the embedder fetches or decodes it
    else:
        data = Path(image).read_bytes()  # local file path → bytes
    doc_id = id or uuid.uuid4().hex
    payload = Payload(modality=Modality.IMAGE, data=data, mime=mime)
    self.add(Document(id=doc_id, payloads=[payload], metadata=metadata or {}, source=source))
    return doc_id

add_file

add_file(path: str, *, metadata: dict[str, Any] | None = None, loader: Loader | None = None) -> list[str]

Load a file (.txt / .md / .docx / .pdf / …) and ingest it.

Dispatches by extension (override with loader=). Returns the document ids. Needs the relevant extra for docx/pdf: pip install python-nlql[loaders].

源代码位于： src/nlql/sdk/engine.py

def add_file(
    self,
    path: str,
    *,
    metadata: dict[str, Any] | None = None,
    loader: Loader | None = None,
) -> list[str]:
    """Load a file (``.txt`` / ``.md`` / ``.docx`` / ``.pdf`` / …) and ingest it.

    Dispatches by extension (override with ``loader=``). Returns the document ids. Needs
    the relevant extra for docx/pdf: ``pip install python-nlql[loaders]``.
    """
    documents = load_documents(path, loader=loader, metadata=metadata)
    self.add_documents(documents)
    return [doc.id for doc in documents]

add_files

add_files(paths: Iterable[str], *, metadata: dict[str, Any] | None = None) -> list[str]

Load and ingest multiple files; returns all document ids.

源代码位于： src/nlql/sdk/engine.py

def add_files(self, paths: Iterable[str], *, metadata: dict[str, Any] | None = None) -> list[str]:
    """Load and ingest multiple files; returns all document ids."""
    ids: list[str] = []
    for path in paths:
        ids.extend(self.add_file(path, metadata=metadata))
    return ids

add_multivector

add_multivector(id: str, *, content: str, named: dict[str, str | bytes], metadata: dict[str, Any] | None = None, kind: str | None = None) -> str

Ingest one record with several named vectors, each queryable via vec.<name>.

content is the record's text (and its default vector, via the primary embedder). named maps a vector name to its source — a str (embedded as text) or image bytes (embedded via that embedder's embed_images). Configure per-name embedders with Engine(..., named_embedders={"image": ClipEmbedder(), ...}). Query a specific vector with SIMILARITY(vec.image, "…").

源代码位于： src/nlql/sdk/engine.py

def add_multivector(
    self,
    id: str,
    *,
    content: str,
    named: dict[str, str | bytes],
    metadata: dict[str, Any] | None = None,
    kind: str | None = None,
) -> str:
    """Ingest one record with several named vectors, each queryable via ``vec.<name>``.

    ``content`` is the record's text (and its default vector, via the primary embedder).
    ``named`` maps a vector name to its source — a ``str`` (embedded as text) or image
    ``bytes`` (embedded via that embedder's ``embed_images``). Configure per-name embedders
    with ``Engine(..., named_embedders={"image": ClipEmbedder(), ...})``. Query a specific
    vector with ``SIMILARITY(vec.image, "…")``.
    """
    unit = Unit(
        id=id,
        doc_id=id,
        kind=kind or self._granularity,  # type: ignore[arg-type]
        payload=Payload.text(content),
        metadata=dict(metadata or {}),
        vector=self._embedder.embed([content])[0],
        ordinal=0,
    )
    for name, data in named.items():
        embedder = self._named_embedders.get(name)
        if embedder is None:
            raise NLQLError(f"no named embedder configured for {name!r}; pass named_embedders=")
        if isinstance(data, (bytes, bytearray)):
            embed_images = getattr(embedder, "embed_images", None)
            if not callable(embed_images):
                raise NLQLError(f"named embedder {name!r} cannot embed image bytes")
            unit.vectors[name] = np.asarray(embed_images([data])[0], dtype=np.float32)
        else:
            unit.vectors[name] = np.asarray(embedder.embed([data])[0], dtype=np.float32)
    self._store.upsert([unit])
    return id

add_documents

add_documents(documents: Iterable[Document], *, batch: int = 256) -> None

Ingest documents in batches (normalize → split → embed(cache) → index).

源代码位于： src/nlql/sdk/engine.py

def add_documents(self, documents: Iterable[Document], *, batch: int = 256) -> None:
    """Ingest documents in batches (normalize → split → embed(cache) → index)."""
    buffer: list[Document] = []
    for doc in documents:
        buffer.append(doc)
        if len(buffer) >= batch:
            self._flush(buffer)
            buffer = []
    if buffer:
        self._flush(buffer)

search

search(query: str | Query, *, limit: int | None = None, reranker: Reranker | None = None, rerank_query: str | None = None) -> list[Unit]

Run an NLQL string or Query IR and return matching units.

A reranker (here or on the engine) refines the recalled candidates: recall over-fetches, then the reranker re-scores each (query, passage) pair and the top limit are returned. rerank_query overrides the text used (default: the primary SIMILARITY query).

源代码位于： src/nlql/sdk/engine.py

def search(
    self,
    query: str | Query,
    *,
    limit: int | None = None,
    reranker: Reranker | None = None,
    rerank_query: str | None = None,
) -> list[Unit]:
    """Run an NLQL string or Query IR and return matching units.

    A ``reranker`` (here or on the engine) refines the recalled candidates: recall
    over-fetches, then the reranker re-scores each ``(query, passage)`` pair and the top
    ``limit`` are returned. ``rerank_query`` overrides the text used (default: the primary
    ``SIMILARITY`` query).
    """
    ir = self._as_query(query)
    if limit is not None:
        ir.limit = limit
    return self._executor.execute(ir, reranker=reranker, rerank_query=rerank_query)

explain

explain(query: str | Query) -> dict[str, Any]

Return the query plan (parsed IR, scores, filter, recall strategy).

源代码位于： src/nlql/sdk/engine.py

def explain(self, query: str | Query) -> dict[str, Any]:
    """Return the query plan (parsed IR, scores, filter, recall strategy)."""
    plan = self._executor.plan(self._as_query(query)).explain()
    if self._executor.reranker is not None:
        plan["rerank"] = type(self._executor.reranker).__name__
    return plan

register_function

register_function(name: str, *, signature: Signature | None = None, provides_score: bool = False, pushdownable: bool = False, overwrite: bool = False) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Register a function for this engine only (instance-scoped).

源代码位于： src/nlql/sdk/engine.py

def register_function(
    self,
    name: str,
    *,
    signature: Signature | None = None,
    provides_score: bool = False,
    pushdownable: bool = False,
    overwrite: bool = False,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Register a function for this engine only (instance-scoped)."""
    return self._registry.function(
        name,
        signature=signature,
        provides_score=provides_score,
        pushdownable=pushdownable,
        overwrite=overwrite,
    )

register_splitter

register_splitter(name: str, *, overwrite: bool = False) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Register/override a splitter for this engine only.

源代码位于： src/nlql/sdk/engine.py

def register_splitter(
    self, name: str, *, overwrite: bool = False
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Register/override a splitter for this engine only."""
    return self._registry.splitter(name, overwrite=overwrite)

register_type

register_type(name: str, handler: Any = None) -> Any

Register a custom type at the instance level (shadows global). Supports both direct call and @register_type decorator mode.

源代码位于： src/nlql/sdk/engine.py

def register_type(self, name: str, handler: Any = None) -> Any:
    """Register a custom type at the instance level (shadows global).
    Supports both direct call and ``@register_type`` decorator mode."""
    from nlql.types.core import register_type as _register

    return _register(name, handler, registry=self._type_handlers)

function_schema

function_schema() -> dict[str, Any]

JSON Schema for the Query IR, with Call names constrained to this engine.

源代码位于： src/nlql/sdk/engine.py

def function_schema(self) -> dict[str, Any]:
    """JSON Schema for the Query IR, with Call names constrained to this engine."""
    return query_json_schema(self._registry.names("function"))

function_tool

function_tool(name: str = 'nlql_query', description: str | None = None) -> dict[str, Any]

An OpenAI-style tool definition an LLM can call to emit a Query IR.

源代码位于： src/nlql/sdk/engine.py

def function_tool(
    self, name: str = "nlql_query", description: str | None = None
) -> dict[str, Any]:
    """An OpenAI-style tool definition an LLM can call to emit a Query IR."""
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": description
            or "Build a semantic retrieval query as an NLQL Query IR document.",
            "parameters": self.function_schema(),
        },
    }

search_ir

search_ir(ir: dict[str, Any], *, limit: int | None = None) -> list[Unit]

Execute a Query IR document (e.g. produced by an LLM via function-calling).

源代码位于： src/nlql/sdk/engine.py

def search_ir(self, ir: dict[str, Any], *, limit: int | None = None) -> list[Unit]:
    """Execute a Query IR document (e.g. produced by an LLM via function-calling)."""
    return self.search(Query.from_dict(ir), limit=limit)

__len__

__len__() -> int

源代码位于： src/nlql/sdk/engine.py

def __len__(self) -> int:
    return len(self._store)

F

F(name: str) -> E

Reference a LET-bound alias.

源代码位于： src/nlql/sdk/builder.py

def F(name: str) -> E:  # noqa: N802
    """Reference a LET-bound alias."""
    return E(Ref(name))

Meta

Meta(key: str) -> E

源代码位于： src/nlql/sdk/builder.py

def Meta(key: str) -> E:  # noqa: N802 - deliberate DSL capitalization
    return E(Path("meta", [key]))

contains

contains(path: str | E, text: str) -> E

源代码位于： src/nlql/sdk/builder.py

def contains(path: str | E, text: str) -> E:
    target = _lift(path) if isinstance(path, E) else _path_from_dotted(path)
    return E(Call("CONTAINS", [target, Literal(text)]))

field

field(dotted: str) -> E

源代码位于： src/nlql/sdk/builder.py

def field(dotted: str) -> E:
    return E(_path_from_dotted(dotted))

length

length(path: str | E) -> E

源代码位于： src/nlql/sdk/builder.py

def length(path: str | E) -> E:
    target = _lift(path) if isinstance(path, E) else _path_from_dotted(path)
    return E(Call("LENGTH", [target]))

select

select(unit: str, window: int | None = None) -> QueryBuilder

Start building a query at the given granularity.

源代码位于： src/nlql/sdk/builder.py

def select(unit: str, window: int | None = None) -> QueryBuilder:
    """Start building a query at the given granularity."""
    return QueryBuilder(unit, window)

similarity

similarity(path: str | E, query: str) -> E

源代码位于： src/nlql/sdk/builder.py

def similarity(path: str | E, query: str) -> E:
    target = _lift(path) if isinstance(path, E) else _path_from_dotted(path)
    return E(Call("SIMILARITY", [target, Literal(query)]))