Skip to content

NLQL API Reference

NLQL

Main NLQL engine interface.

This is the primary entry point for users to execute NLQL queries.

The design follows an explicit adapter pattern - users must provide a concrete adapter implementation for their data source. This ensures clear separation of concerns and makes the system highly extensible.

Example

from nlql import NLQL from nlql.adapters import MemoryAdapter

Create adapter with data

adapter = MemoryAdapter() adapter.add_text("AI agents are autonomous systems") adapter.add_text("Machine learning powers modern AI")

Initialize NLQL with explicit adapter

nlql = NLQL(adapter=adapter)

Execute query

results = nlql.execute("SELECT CHUNK WHERE CONTAINS('AI') LIMIT 5") for result in results: ... print(result.content)

Source code in src/nlql/api.py
class NLQL:
    """Main NLQL engine interface.

    This is the primary entry point for users to execute NLQL queries.

    The design follows an explicit adapter pattern - users must provide
    a concrete adapter implementation for their data source. This ensures
    clear separation of concerns and makes the system highly extensible.

    Example:
        >>> from nlql import NLQL
        >>> from nlql.adapters import MemoryAdapter
        >>>
        >>> # Create adapter with data
        >>> adapter = MemoryAdapter()
        >>> adapter.add_text("AI agents are autonomous systems")
        >>> adapter.add_text("Machine learning powers modern AI")
        >>>
        >>> # Initialize NLQL with explicit adapter
        >>> nlql = NLQL(adapter=adapter)
        >>>
        >>> # Execute query
        >>> results = nlql.execute("SELECT CHUNK WHERE CONTAINS('AI') LIMIT 5")
        >>> for result in results:
        ...     print(result.content)
    """

    def __init__(
        self,
        adapter: BaseAdapter,
        embedding_provider: EmbeddingProvider | None = None,
        config: NLQLConfig | None = None,
        meta_registry: MetaFieldRegistry | None = None,
    ) -> None:
        """Initialize NLQL engine.

        Args:
            adapter: Data source adapter (required). Users must explicitly
                    provide an adapter implementation for their data source.
            embedding_provider: Optional custom embedding provider
            config: Optional configuration
            meta_registry: Optional META field registry

        Raises:
            TypeError: If adapter is not a BaseAdapter instance

        Example:
            >>> from nlql import NLQL
            >>> from nlql.adapters import MemoryAdapter
            >>>
            >>> adapter = MemoryAdapter()
            >>> nlql = NLQL(adapter=adapter)
        """
        if not isinstance(adapter, BaseAdapter):
            raise TypeError(
                f"adapter must be an instance of BaseAdapter, got {type(adapter)}. "
                f"Please create an explicit adapter (e.g., MemoryAdapter, ChromaAdapter)."
            )

        # Initialize components
        self._parser = NLQLParser()
        self._config = config or NLQLConfig.default()

        # Instance-level registries (optional, falls back to global if None)
        self._function_registry: FunctionRegistry | None = None
        self._operator_registry: OperatorRegistry | None = None
        self._embedding_provider: EmbeddingProvider | None = embedding_provider

        # Create execution context
        self._context = ExecutionContext(
            adapter=adapter,
            config=self._config,
            meta_registry=meta_registry,
            embedding_provider=embedding_provider,
            function_registry=self._function_registry,
            operator_registry=self._operator_registry,
        )

        self._executor = Executor(self._context)

    def execute(self, query: str) -> list[Result]:
        """Execute an NLQL query.

        Args:
            query: NLQL query string

        Returns:
            List of query results

        Raises:
            NLQLParseError: If query parsing fails
            NLQLExecutionError: If query execution fails
        """
        # Parse query
        ast = self._parser.parse(query)

        # Execute query
        results = self._executor.execute(ast)

        return results

    @property
    def config(self) -> NLQLConfig:
        """Get the current configuration."""
        return self._config

    @property
    def adapter(self) -> BaseAdapter:
        """Get the current adapter."""
        return self._context.adapter

    def register_function(self, name: str) -> Callable[[FunctionImpl], FunctionImpl]:
        """Register a custom function to this NLQL instance.

        This creates an instance-level function registry if it doesn't exist,
        and registers the function only for this instance. Instance-level
        registrations take precedence over global registrations.

        Args:
            name: Function name (e.g., "word_count")

        Returns:
            Decorator function

        Example:
            >>> nlql = NLQL(adapter=adapter)
            >>> @nlql.register_function("CUSTOM")
            ... def my_func(text: str) -> int:
            ...     return len(text)
        """
        # Create instance-level registry if it doesn't exist
        if self._function_registry is None:
            self._function_registry = FunctionRegistry()
            self._context.function_registry = self._function_registry

        def decorator(func: FunctionImpl) -> FunctionImpl:
            self._function_registry.register(name, func)
            return func

        return decorator

    def register_operator(self, name: str) -> Callable[[OperatorFunc], OperatorFunc]:
        """Register a custom operator to this NLQL instance.

        This creates an instance-level operator registry if it doesn't exist,
        and registers the operator only for this instance. Instance-level
        registrations take precedence over global registrations.

        Args:
            name: Operator name (must be UPPERCASE, e.g., "CUSTOM_OP")

        Returns:
            Decorator function

        Example:
            >>> nlql = NLQL(adapter=adapter)
            >>> @nlql.register_operator("CUSTOM_OP")
            ... def my_op(text: str, param: str) -> bool:
            ...     return param in text
        """
        # Create instance-level registry if it doesn't exist
        if self._operator_registry is None:
            self._operator_registry = OperatorRegistry()
            self._context.operator_registry = self._operator_registry

        def decorator(func: OperatorFunc) -> OperatorFunc:
            self._operator_registry.register(name, func)
            return func

        return decorator

    def register_embedding_provider(self, provider: EmbeddingProvider) -> EmbeddingProvider:
        """Register a custom embedding provider to this NLQL instance.

        This registers the embedding provider only for this instance.
        Instance-level registrations take precedence over global registrations.

        Args:
            provider: Embedding provider function

        Returns:
            The same provider function (for decorator usage)

        Example:
            >>> nlql = NLQL(adapter=adapter)
            >>> @nlql.register_embedding_provider
            ... def my_embedding(texts: list[str]) -> list[list[float]]:
            ...     return [[0.1, 0.2] for _ in texts]
        """
        self._embedding_provider = provider
        self._context.embedding_provider = provider
        return provider

config property

Get the current configuration.

adapter property

Get the current adapter.

__init__(adapter, embedding_provider=None, config=None, meta_registry=None)

Initialize NLQL engine.

Parameters:

Name Type Description Default
adapter BaseAdapter

Data source adapter (required). Users must explicitly provide an adapter implementation for their data source.

required
embedding_provider EmbeddingProvider | None

Optional custom embedding provider

None
config NLQLConfig | None

Optional configuration

None
meta_registry MetaFieldRegistry | None

Optional META field registry

None

Raises:

Type Description
TypeError

If adapter is not a BaseAdapter instance

Example

from nlql import NLQL from nlql.adapters import MemoryAdapter

adapter = MemoryAdapter() nlql = NLQL(adapter=adapter)

Source code in src/nlql/api.py
def __init__(
    self,
    adapter: BaseAdapter,
    embedding_provider: EmbeddingProvider | None = None,
    config: NLQLConfig | None = None,
    meta_registry: MetaFieldRegistry | None = None,
) -> None:
    """Initialize NLQL engine.

    Args:
        adapter: Data source adapter (required). Users must explicitly
                provide an adapter implementation for their data source.
        embedding_provider: Optional custom embedding provider
        config: Optional configuration
        meta_registry: Optional META field registry

    Raises:
        TypeError: If adapter is not a BaseAdapter instance

    Example:
        >>> from nlql import NLQL
        >>> from nlql.adapters import MemoryAdapter
        >>>
        >>> adapter = MemoryAdapter()
        >>> nlql = NLQL(adapter=adapter)
    """
    if not isinstance(adapter, BaseAdapter):
        raise TypeError(
            f"adapter must be an instance of BaseAdapter, got {type(adapter)}. "
            f"Please create an explicit adapter (e.g., MemoryAdapter, ChromaAdapter)."
        )

    # Initialize components
    self._parser = NLQLParser()
    self._config = config or NLQLConfig.default()

    # Instance-level registries (optional, falls back to global if None)
    self._function_registry: FunctionRegistry | None = None
    self._operator_registry: OperatorRegistry | None = None
    self._embedding_provider: EmbeddingProvider | None = embedding_provider

    # Create execution context
    self._context = ExecutionContext(
        adapter=adapter,
        config=self._config,
        meta_registry=meta_registry,
        embedding_provider=embedding_provider,
        function_registry=self._function_registry,
        operator_registry=self._operator_registry,
    )

    self._executor = Executor(self._context)

execute(query)

Execute an NLQL query.

Parameters:

Name Type Description Default
query str

NLQL query string

required

Returns:

Type Description
list[Result]

List of query results

Raises:

Type Description
NLQLParseError

If query parsing fails

NLQLExecutionError

If query execution fails

Source code in src/nlql/api.py
def execute(self, query: str) -> list[Result]:
    """Execute an NLQL query.

    Args:
        query: NLQL query string

    Returns:
        List of query results

    Raises:
        NLQLParseError: If query parsing fails
        NLQLExecutionError: If query execution fails
    """
    # Parse query
    ast = self._parser.parse(query)

    # Execute query
    results = self._executor.execute(ast)

    return results

register_function(name)

Register a custom function to this NLQL instance.

This creates an instance-level function registry if it doesn't exist, and registers the function only for this instance. Instance-level registrations take precedence over global registrations.

Parameters:

Name Type Description Default
name str

Function name (e.g., "word_count")

required

Returns:

Type Description
Callable[[FunctionImpl], FunctionImpl]

Decorator function

Example

nlql = NLQL(adapter=adapter) @nlql.register_function("CUSTOM") ... def my_func(text: str) -> int: ... return len(text)

Source code in src/nlql/api.py
def register_function(self, name: str) -> Callable[[FunctionImpl], FunctionImpl]:
    """Register a custom function to this NLQL instance.

    This creates an instance-level function registry if it doesn't exist,
    and registers the function only for this instance. Instance-level
    registrations take precedence over global registrations.

    Args:
        name: Function name (e.g., "word_count")

    Returns:
        Decorator function

    Example:
        >>> nlql = NLQL(adapter=adapter)
        >>> @nlql.register_function("CUSTOM")
        ... def my_func(text: str) -> int:
        ...     return len(text)
    """
    # Create instance-level registry if it doesn't exist
    if self._function_registry is None:
        self._function_registry = FunctionRegistry()
        self._context.function_registry = self._function_registry

    def decorator(func: FunctionImpl) -> FunctionImpl:
        self._function_registry.register(name, func)
        return func

    return decorator

register_operator(name)

Register a custom operator to this NLQL instance.

This creates an instance-level operator registry if it doesn't exist, and registers the operator only for this instance. Instance-level registrations take precedence over global registrations.

Parameters:

Name Type Description Default
name str

Operator name (must be UPPERCASE, e.g., "CUSTOM_OP")

required

Returns:

Type Description
Callable[[OperatorFunc], OperatorFunc]

Decorator function

Example

nlql = NLQL(adapter=adapter) @nlql.register_operator("CUSTOM_OP") ... def my_op(text: str, param: str) -> bool: ... return param in text

Source code in src/nlql/api.py
def register_operator(self, name: str) -> Callable[[OperatorFunc], OperatorFunc]:
    """Register a custom operator to this NLQL instance.

    This creates an instance-level operator registry if it doesn't exist,
    and registers the operator only for this instance. Instance-level
    registrations take precedence over global registrations.

    Args:
        name: Operator name (must be UPPERCASE, e.g., "CUSTOM_OP")

    Returns:
        Decorator function

    Example:
        >>> nlql = NLQL(adapter=adapter)
        >>> @nlql.register_operator("CUSTOM_OP")
        ... def my_op(text: str, param: str) -> bool:
        ...     return param in text
    """
    # Create instance-level registry if it doesn't exist
    if self._operator_registry is None:
        self._operator_registry = OperatorRegistry()
        self._context.operator_registry = self._operator_registry

    def decorator(func: OperatorFunc) -> OperatorFunc:
        self._operator_registry.register(name, func)
        return func

    return decorator

register_embedding_provider(provider)

Register a custom embedding provider to this NLQL instance.

This registers the embedding provider only for this instance. Instance-level registrations take precedence over global registrations.

Parameters:

Name Type Description Default
provider EmbeddingProvider

Embedding provider function

required

Returns:

Type Description
EmbeddingProvider

The same provider function (for decorator usage)

Example

nlql = NLQL(adapter=adapter) @nlql.register_embedding_provider ... def my_embedding(texts: list[str]) -> list[list[float]]: ... return [[0.1, 0.2] for _ in texts]

Source code in src/nlql/api.py
def register_embedding_provider(self, provider: EmbeddingProvider) -> EmbeddingProvider:
    """Register a custom embedding provider to this NLQL instance.

    This registers the embedding provider only for this instance.
    Instance-level registrations take precedence over global registrations.

    Args:
        provider: Embedding provider function

    Returns:
        The same provider function (for decorator usage)

    Example:
        >>> nlql = NLQL(adapter=adapter)
        >>> @nlql.register_embedding_provider
        ... def my_embedding(texts: list[str]) -> list[list[float]]:
        ...     return [[0.1, 0.2] for _ in texts]
    """
    self._embedding_provider = provider
    self._context.embedding_provider = provider
    return provider

NLQLConfig dataclass

Configuration for NLQL engine.

Attributes:

Name Type Description
default_limit int | None

Default LIMIT value if not specified in query. Applied when query does not have explicit LIMIT clause.

enable_caching bool

Whether to enable query result caching. Reserved for future implementation.

debug_mode bool

Enable debug logging and verbose error messages. When True, logs execution steps and detailed error information.

custom_settings dict[str, Any]

Additional user-defined settings. Reserved for future extensibility.

Source code in src/nlql/config.py
@dataclass
class NLQLConfig:
    """Configuration for NLQL engine.

    Attributes:
        default_limit: Default LIMIT value if not specified in query.
                      Applied when query does not have explicit LIMIT clause.
        enable_caching: Whether to enable query result caching.
                       Reserved for future implementation.
        debug_mode: Enable debug logging and verbose error messages.
                   When True, logs execution steps and detailed error information.
        custom_settings: Additional user-defined settings.
                        Reserved for future extensibility.
    """

    default_limit: int | None = None
    enable_caching: bool = False  # TODO: Not implemented yet
    debug_mode: bool = False
    custom_settings: dict[str, Any] = field(default_factory=dict)  # TODO: Not implemented yet

    @classmethod
    def default(cls) -> "NLQLConfig":
        """Create a default configuration."""
        return cls()

default() classmethod

Create a default configuration.

Source code in src/nlql/config.py
@classmethod
def default(cls) -> "NLQLConfig":
    """Create a default configuration."""
    return cls()

Result dataclass

Represents a single query result.

Attributes:

Name Type Description
content str

The text content of this result

score float | None

Relevance score (e.g., similarity score), if applicable

metadata dict[str, Any]

Additional metadata from the data source

unit str

The text unit type (DOCUMENT, CHUNK, SENTENCE, SPAN)

source_id str | None

Optional identifier for the source document/chunk

Source code in src/nlql/result.py
@dataclass
class Result:
    """Represents a single query result.

    Attributes:
        content: The text content of this result
        score: Relevance score (e.g., similarity score), if applicable
        metadata: Additional metadata from the data source
        unit: The text unit type (DOCUMENT, CHUNK, SENTENCE, SPAN)
        source_id: Optional identifier for the source document/chunk
    """

    content: str
    score: float | None = None
    metadata: dict[str, Any] = field(default_factory=dict)
    unit: str = "CHUNK"
    source_id: str | None = None

    def __repr__(self) -> str:
        score_str = f"{self.score:.3f}" if self.score is not None else "N/A"
        return f"Result(unit={self.unit}, score={score_str}, content={self.content[:50]}...)"