operonx.providers¶

LLM, embedding, reranker, and ONNX provider ops. Provider backends are loaded lazily — a tier-1 install (pip install operonx) can import operonx.providers without pulling openai / httpx / numpy / torch. Missing-dep errors surface only when the corresponding backend is actually accessed.

See Pick an extra on the installation page for which extra each provider needs.

Provider ops¶

The core provider op types. Each exposes an Op.of(...) classmethod for concise construction with explicit keyword args — that's the recommended style.

LLMOp ¶

LLMOp(
    resource: Optional[Union[str, List[str]]] = None,
    ratios: Optional[List[float]] = None,
    fallback: Optional[List[str]] = None,
    batch_mode: bool = False,
    seed: Optional[int] = None,
    fields: Optional[List[str]] = None,
    parser: Optional[str] = None,
    validators: Optional[Dict[str, List[Any]]] = None,
    max_retries: int = 0,
    retry_hint: bool = True,
    inputs: Dict[str, Any] = None,
    outputs: Dict[str, Any] = None,
    **kwargs: Any,
)

Bases: BaseOp

Op that formats a prompt and calls a language model via ResourceHub.

Give it exactly one of prompt= or messages=.

prompt= is a template, and every string in it is str.format_map-substituted with the non-reserved kwargs:

str — one user message: "Hello {name}".
dict with system / user keys — the standard two-message call.

messages= is a conversation: a full OpenAI messages array, passed through untouched. Use it whenever the content is data rather than a template — an agent's history, a multimodal block assembled upstream, anything past two messages.

The split exists because formatting a conversation is destructive. Any brace in any message — a tool returning {"city": "Hanoi"}, a user pasting CSS, the model's own tool-call arguments — becomes a template variable that does not exist, and the run dies on the next model call. prompt= used to accept a list, which made that the default outcome for every agent.

Inputs

prompt (str | dict): Message template. Mutually exclusive with messages. messages (list): Ready OpenAI messages array, never formatted. temperature (float): Sampling temperature. Default: 0.0. max_tokens (int): Max output tokens. Default: None. tools (list): Tool/function definitions. Default: None. tool_choice (str | dict): Tool selection strategy. Default: None. response_format (dict): Structured output format. Default: None. (any): Template variables ({var} placeholders).

Outputs

content (str): Generated text. role (str): Message role (usually "assistant"). finish_reason (str): Stop reason ("stop", "tool_calls", ...). model_used (str): Actual model that served the request. tool_calls (list): Tool-call objects (empty list when absent). usage (dict): Flat token-cost metrics. extras (dict): Bag of uncommon fields (thinking_content, refusal, logprobs).

Example::

llm = LLMOp.of(
    resource="gpt-4o",
    prompt={"system": "You are {role}.", "user": "{query}"},
    role="helpful", query=PARENT["query"],
)

chat = LLMOp.of(resource="gpt-4o", messages=history["messages"])

Initialize LLMOp.

Parameters:

Name	Type	Description	Default
`resource`	`Optional[Union[str, List[str]]]`	Resource key(s) for LLM in ResourceHub. - Single string: "gpt-4" - List for load balancing: ["gpt-4", "claude-3"]	`None`
`ratios`	`Optional[List[float]]`	Weight ratios for load balancing. Must sum to 1.0.	`None`
`fallback`	`Optional[List[str]]`	Fallback resource keys tried in order on hard failures — refusals from the model, provider-side content filtering, or exhausted transport retries (the underlying SDK gave up). NOT triggered by parse/validator failures — those use `max_retries`.	`None`
`batch_mode`	`bool`	Use OpenAI Batch API (50% cheaper).	`False`
`seed`	`Optional[int]`	Optional seed for load balancing RNG.	`None`
`fields`	`Optional[List[str]]`	Optional list of `"path.to.value: type"` extraction schemas (see `operonx.providers.parsing.ExtractField`). When set, the LLM response is parsed inline; each field becomes a top-level output of this op.	`None`
`parser`	`Optional[str]`	Parser format when `fields` is set: `"xml"`, `"json"`, or `"yaml"`. Defaults to `"xml"` when `fields` is provided.	`None`
`validators`	`Optional[Dict[str, List[Any]]]`	Optional per-field allow-list validators applied after extraction. Format: `{"field_name": [allowed_value, ...]}`. A value prefixed with `@` in the list is used as a default when the extracted value doesn't match the allow-list.	`None`
`max_retries`	`int`	Max semantic retries when the parser or validators report an error. Default 0 (no retry — first parse failure surfaces as `error` in the output). Transport errors are the SDK's responsibility and NOT counted here.	`0`
`retry_hint`	`bool`	When True (default) and retrying, append the previous LLM response and a "that failed — , try again" user turn so the model sees what went wrong.	`True`
`inputs`	`Dict[str, Any]`	Input variable mappings.	`None`
`outputs`	`Dict[str, Any]`	Output variable mappings.	`None`
`**kwargs`	`Any`	Additional keyword arguments for BaseOp.	`{}`

Source code in operonx/providers/ops/llm.py

def __init__(
    self,
    resource: Optional[Union[str, List[str]]] = None,
    ratios: Optional[List[float]] = None,
    fallback: Optional[List[str]] = None,
    batch_mode: bool = False,
    seed: Optional[int] = None,
    # ── Structured-output layer (merged from ask()/ParserOp in 1.0.0) ───
    fields: Optional[List[str]] = None,
    parser: Optional[str] = None,
    validators: Optional[Dict[str, List[Any]]] = None,
    max_retries: int = 0,
    retry_hint: bool = True,
    inputs: Dict[str, Any] = None,
    outputs: Dict[str, Any] = None,
    **kwargs: Any,
):
    """Initialize LLMOp.

    Args:
        resource: Resource key(s) for LLM in ResourceHub.
            - Single string: "gpt-4"
            - List for load balancing: ["gpt-4", "claude-3"]
        ratios: Weight ratios for load balancing. Must sum to 1.0.
        fallback: Fallback resource keys tried in order on **hard** failures
            — refusals from the model, provider-side content filtering, or
            exhausted transport retries (the underlying SDK gave up). NOT
            triggered by parse/validator failures — those use ``max_retries``.
        batch_mode: Use OpenAI Batch API (50% cheaper).
        seed: Optional seed for load balancing RNG.
        fields: Optional list of ``"path.to.value: type"`` extraction schemas
            (see ``operonx.providers.parsing.ExtractField``). When set, the
            LLM response is parsed inline; each field becomes a top-level
            output of this op.
        parser: Parser format when ``fields`` is set: ``"xml"``, ``"json"``,
            or ``"yaml"``. Defaults to ``"xml"`` when ``fields`` is provided.
        validators: Optional per-field allow-list validators applied after
            extraction. Format: ``{"field_name": [allowed_value, ...]}``.
            A value prefixed with ``@`` in the list is used as a default
            when the extracted value doesn't match the allow-list.
        max_retries: Max **semantic** retries when the parser or validators
            report an error. Default 0 (no retry — first parse failure
            surfaces as ``error`` in the output). Transport errors are the
            SDK's responsibility and NOT counted here.
        retry_hint: When True (default) and retrying, append the previous
            LLM response and a "that failed — <error>, try again" user turn
            so the model sees what went wrong.
        inputs: Input variable mappings.
        outputs: Output variable mappings.
        **kwargs: Additional keyword arguments for BaseOp.
    """
    kwargs.setdefault("bound", "io")
    super().__init__(**kwargs)

    self.batch_mode = batch_mode
    self.contain_generation = True
    self.fallback = fallback
    self._rng = random.Random(seed)

    # Structured-output config (merged from ParserOp).
    if fields and not parser:
        parser = "xml"
    if parser and not fields:
        raise TypeError(
            "LLMOp(parser=...) requires fields=[...] — parser has no work "
            "to do without extraction schemas."
        )
    if validators and not fields:
        raise TypeError(
            "LLMOp(validators=...) requires fields=[...] — nothing to "
            "validate without extracted values."
        )
    if max_retries < 0:
        raise ValueError(f"max_retries must be >= 0, got {max_retries}")
    self.fields = fields
    self.parser = parser
    self.validators = validators
    self.max_retries = max_retries
    self.retry_hint = retry_hint
    self._extract_fields = [ExtractField.from_string(s) for s in fields] if fields else None

    # Validate resource + ratios
    if isinstance(resource, list):
        self.resource = resource
        self.ratios = ratios or [1.0 / len(resource)] * len(resource)
        if len(self.ratios) != len(self.resource):
            raise ValueError(
                f"ratios length ({len(self.ratios)}) must match "
                f"resource length ({len(self.resource)})"
            )
        if abs(sum(self.ratios) - 1.0) > 0.01:
            raise ValueError(f"ratios must sum to 1.0, got {sum(self.ratios)}")
    else:
        self.resource = resource
        self.ratios = [1.0] if resource else None

    # Fixed LLM knobs
    input_schema = {
        "prompt": Param(type=(str, dict), default=None),
        "messages": Param(type=list, default=None),
        "temperature": Param(type=float, default=0.0),
        "max_tokens": Param(type=int, default=None),
        "tools": Param(type=list, default=None),
        "tool_choice": Param(type=(str, dict), default=None),
        "response_format": Param(type=dict, default=None),
        "top_p": Param(type=float, default=None),
        "stop": Param(type=(str, list), default=None),
        "frequency_penalty": Param(type=float, default=None),
        "presence_penalty": Param(type=float, default=None),
        "seed": Param(type=int, default=None),
        "logprobs": Param(type=bool, default=None),
        "top_logprobs": Param(type=int, default=None),
        "n": Param(type=int, default=None),
        "user": Param(type=str, default=None),
    }

    output_schema = {
        "role": Param(type=str, default="assistant"),
        "content": Param(type=str, required=True),
        # Streaming only: False on a token delta, True on the frame
        # that repeats the accumulated content. Batch calls are always
        # final. See ``_stream_final``.
        "final": Param(type=bool, default=True),
        "finish_reason": Param(type=str, default=None),
        "model_used": Param(type=str, required=True),
        "tool_calls": Param(type=list, default=[]),
        "usage": Param(type=dict, default={}),
        "extras": Param(type=dict, default={}),
    }
    # When fields=[...] is set, extend the output schema with one Param
    # per extracted field and an ``error`` field (parse/validate error
    # string, or None on success). Callers wire the individual fields
    # through refs the same way they used to wire ParserOp outputs.
    if self._extract_fields:
        for f in self._extract_fields:
            output_schema[f.output_key] = Param(default=None)
        output_schema["error"] = Param(type=str, default=None)

    # Checked on the *raw* mapping: _normalize_params wraps each value
    # in a Param, so a literal list is no longer a list by then.
    _check_prompt_inputs(inputs or {})

    normalized_inputs = self._normalize_params(inputs)
    normalized_outputs = self._normalize_params(outputs)

    # Wildcard PARENT inference: pull template var names from a static prompt
    if "__FORWARD_WILDCARD__" in normalized_inputs:
        for var in self._infer_wildcard_vars(normalized_inputs["__FORWARD_WILDCARD__"]):
            if var not in input_schema:
                input_schema[var] = Param(type=Any, required=False, default=None)

    # Non-reserved user inputs are template variables
    for key in normalized_inputs:
        if key not in RESERVED_KEYS and key != "__FORWARD_WILDCARD__":
            if key not in input_schema:
                input_schema[key] = Param(type=Any, required=False, default=None)

    self.inputs = self._merge_params(input_schema, normalized_inputs)
    self.outputs = self._merge_params(output_schema, normalized_outputs)

    # Lazy-initialized from ResourceHub on first use
    self._llms: List["BaseLLM"] = []
    self._fallback_llms: List["BaseLLM"] = []
    self._batch_coordinator = None
    self._initialized = False

    # Core: stream → _stream_core, else → _generate_core
    if self.stream:
        self._set_core(self._stream_core)
    else:
        self._set_core(self._generate_core)

Attributes¶

specific_metadata `property` ¶

specific_metadata: Dict[str, Any]

Return LLM + prompt metadata dictionary.

Methods:¶

warmup ¶

warmup() -> None

Eagerly initialize LLM backends on engine startup.

Source code in operonx/providers/ops/llm.py

def warmup(self) -> None:
    """Eagerly initialize LLM backends on engine startup."""
    self._ensure_initialized()

normalize_trace_io ¶

normalize_trace_io(inputs: Dict[str, Any], outputs: Dict[str, Any]) -> tuple

Wrap OpenAI chat-format multimodal blocks as Media for tracing.

Prompt is formatted to messages first (using the current template vars), then multimodal image/audio blocks are wrapped in Media for the trace-time view. Real op state is untouched.

Source code in operonx/providers/ops/llm.py

def normalize_trace_io(self, inputs: Dict[str, Any], outputs: Dict[str, Any]) -> tuple:
    """Wrap OpenAI chat-format multimodal blocks as ``Media`` for tracing.

    Prompt is formatted to messages first (using the current template vars),
    then multimodal image/audio blocks are wrapped in ``Media`` for the
    trace-time view. Real op state is untouched.
    """
    prompt = inputs.get("prompt")
    raw_messages = inputs.get("messages")
    if prompt is not None or raw_messages is not None:
        try:
            vars = {k: v for k, v in inputs.items() if k not in RESERVED_KEYS}
            # messages= is already built; only a template needs formatting.
            # Reading `prompt` alone here meant a messages= call traced no
            # media at all.
            messages = (
                list(raw_messages)
                if raw_messages is not None
                else self._build_messages(prompt, vars)
            )
            wrapped = self._wrap_openai_media_blocks(messages)
            inputs = {**inputs, "messages": wrapped}
        except Exception:
            # Tracing must never break execution — swallow formatting errors
            pass
    return inputs, outputs

of ¶

of(
    resource=None,
    *,
    ratios=None,
    fallback=None,
    batch_mode=False,
    seed=None,
    prompt=None,
    messages=None,
    fields=None,
    parser=None,
    validators=None,
    max_retries=0,
    retry_hint=True,
    **kwargs,
) -> LLMOp

Create an LLMOp with flat kwargs.

Simple mode::

llm = LLMOp.of(resource="gpt-4", prompt="Hello {name}", name="Alice")

Conversation — a message list that is data, never formatted::

chat = LLMOp.of(resource="gpt-4", messages=history["messages"])

Structured mode (replaces the removed ask() helper)::

llm = LLMOp.of(
    resource="claude-haiku",
    prompt="Classify: {speech}",
    fields=["result: str"],
    parser="xml",
    validators={"result": ["CONFIRM", "DENY", "@FALLBACK"]},
    max_retries=2,
    speech=PARENT["speech"],
)

Source code in operonx/providers/ops/llm.py

@shorthand
def of(
    cls,
    resource=None,
    *,
    ratios=None,
    fallback=None,
    batch_mode=False,
    seed=None,
    prompt=None,
    messages=None,
    # Structured-output layer (merged from ask()/ParserOp in 1.0.0).
    fields=None,
    parser=None,
    validators=None,
    max_retries=0,
    retry_hint=True,
    **kwargs,
) -> "LLMOp":
    """Create an LLMOp with flat kwargs.

    Simple mode::

        llm = LLMOp.of(resource="gpt-4", prompt="Hello {name}", name="Alice")

    Conversation — a message list that is data, never formatted::

        chat = LLMOp.of(resource="gpt-4", messages=history["messages"])

    Structured mode (replaces the removed ask() helper)::

        llm = LLMOp.of(
            resource="claude-haiku",
            prompt="Classify: {speech}",
            fields=["result: str"],
            parser="xml",
            validators={"result": ["CONFIRM", "DENY", "@FALLBACK"]},
            max_retries=2,
            speech=PARENT["speech"],
        )
    """
    input_mappings, init_kwargs = split_shorthand_kwargs(kwargs)
    if prompt is not None:
        input_mappings["prompt"] = prompt
    if messages is not None:
        input_mappings["messages"] = messages
    return cls(
        resource=resource,
        ratios=ratios,
        fallback=fallback,
        batch_mode=batch_mode,
        seed=seed,
        fields=fields,
        parser=parser,
        validators=validators,
        max_retries=max_retries,
        retry_hint=retry_hint,
        inputs=input_mappings or None,
        **init_kwargs,
    )

serialize ¶

serialize() -> dict

Serialize LLMOp for Rust backend, including backend configs.

Source code in operonx/providers/ops/llm.py

def serialize(self) -> dict:
    """Serialize LLMOp for Rust backend, including backend configs."""
    self._ensure_initialized()
    base = super().serialize()

    base["resource"] = self.resource
    base["ratios"] = self.ratios
    base["fallback"] = self.fallback
    base["batch_mode"] = self.batch_mode

    configs = []
    for llm in self._llms:
        if llm and hasattr(llm, "config"):
            configs.append(llm.config.model_dump(mode="json"))
    if configs:
        base["resource_configs"] = configs

    fallback_configs = []
    for llm in self._fallback_llms:
        if llm and hasattr(llm, "config"):
            fallback_configs.append(llm.config.model_dump(mode="json"))
    if fallback_configs:
        base["fallback_configs"] = fallback_configs

    return base

EmbeddingOp ¶

EmbeddingOp(
    resource: Optional[str] = None,
    inputs: Dict[str, Any] = None,
    outputs: Dict[str, Any] = None,
    **kwargs: Any,
)

Bases: BaseOp

Op that converts texts to vector embeddings via ResourceHub.

Wraps an embedding backend (e.g. BGE-M3, OpenAI, TEI) and returns a list of embedding vectors matching the input order.

Inputs

texts (list[str]): Texts to embed. Required.

Outputs

embeddings (list[list[float]]): Embedding vectors.

Example::

embed = EmbeddingOp.of(resource="bge-m3", texts=PARENT["texts"])

Initialize EmbeddingOp.

Parameters:

Name	Type	Description	Default
`resource`	`Optional[str]`	Resource key for embedding model in ResourceHub (e.g., "bge-m3")	`None`
`inputs`	`Dict[str, Any]`	Input variable mappings	`None`
`outputs`	`Dict[str, Any]`	Output variable mappings	`None`
`**kwargs`	`Any`	Additional keyword arguments for BaseOp	`{}`

Source code in operonx/providers/ops/embedding.py

def __init__(
    self,
    resource: Optional[str] = None,
    inputs: Dict[str, Any] = None,
    outputs: Dict[str, Any] = None,
    **kwargs: Any,
):
    """Initialize EmbeddingOp.

    Args:
        resource: Resource key for embedding model in ResourceHub (e.g., "bge-m3")
        inputs: Input variable mappings
        outputs: Output variable mappings
        **kwargs: Additional keyword arguments for BaseOp
    """
    # Provider ops are I/O-bound by default (HTTP calls to embedding backends)
    kwargs.setdefault("bound", "io")
    super().__init__(**kwargs)

    self.resource = resource

    # Define input/output schema
    input_schema = {
        "texts": Param(type=list, required=True),
    }

    output_schema = {
        "embeddings": Param(type=list, required=True),
    }

    # Merge with user-provided
    self.inputs = self._merge_params(input_schema, inputs)
    self.outputs = self._merge_params(output_schema, outputs)

    # Embedding backend — lazy-initialized on first use to allow
    # graph construction before ResourceHub is set up
    self.backend = None
    self._initialized = False
    self._set_core(self._process)

Attributes¶

specific_metadata `property` ¶

specific_metadata: Dict[str, Any]

Return embedding-specific metadata dictionary.

Methods:¶

warmup ¶

warmup() -> None

Eagerly initialize embedding backend on engine startup.

Source code in operonx/providers/ops/embedding.py

def warmup(self) -> None:
    """Eagerly initialize embedding backend on engine startup."""
    self._ensure_initialized()

of ¶

of(resource=None, **kwargs) -> EmbeddingOp

Create an EmbeddingOp with flat kwargs.

Example::

embed = EmbeddingOp.of(resource="bge-m3", texts=PARENT["texts"], outputs={"*": PARENT})

Source code in operonx/providers/ops/embedding.py

@shorthand
def of(cls, resource=None, **kwargs) -> "EmbeddingOp":
    """Create an EmbeddingOp with flat kwargs.

    Example::

        embed = EmbeddingOp.of(resource="bge-m3", texts=PARENT["texts"], outputs={"*": PARENT})
    """
    input_mappings, init_kwargs = split_shorthand_kwargs(kwargs)
    return cls(resource=resource, inputs=input_mappings or None, **init_kwargs)

serialize ¶

serialize() -> dict

Serialize EmbeddingOp for Rust backend, including backend config.

Source code in operonx/providers/ops/embedding.py

def serialize(self) -> dict:
    """Serialize EmbeddingOp for Rust backend, including backend config."""
    self._ensure_initialized()
    base = super().serialize()
    base["resource"] = self.resource
    if self.backend and hasattr(self.backend, "config"):
        base["resource_config"] = self.backend.config.model_dump(mode="json")
    return base

RerankOp ¶

RerankOp(
    resource: Optional[str] = None,
    inputs: Dict[str, Any] = None,
    outputs: Dict[str, Any] = None,
    **kwargs: Any,
)

Bases: BaseOp

Op that scores and re-orders documents by relevance to a query.

Wraps a reranker backend (e.g. BGE-M3, Pinecone, TEI) accessed via ResourceHub. Returns documents sorted by relevance score.

Inputs

query (str): The query to rank against. Required. documents (list[str]): Documents to rerank. Required. top_k (int): Max results to return. Default: -1 (all). threshold (float): Min score cutoff. Default: 0.0.

Outputs

reranks (list[dict]): Reranked results with index, score, and document fields.

Example::

rerank = RerankOp.of(
    resource="bge-m3",
    query=PARENT["query"],
    documents=PARENT["docs"],
    top_k=5,
)

Initialize RerankOp.

Parameters:

Name	Type	Description	Default
`resource`	`Optional[str]`	Resource key for reranker in ResourceHub (e.g., "bge-m3")	`None`
`inputs`	`Dict[str, Any]`	Input variable mappings	`None`
`outputs`	`Dict[str, Any]`	Output variable mappings	`None`
`**kwargs`	`Any`	Additional keyword arguments for BaseOp	`{}`

Source code in operonx/providers/ops/rerank.py

def __init__(
    self,
    resource: Optional[str] = None,
    inputs: Dict[str, Any] = None,
    outputs: Dict[str, Any] = None,
    **kwargs: Any,
):
    """Initialize RerankOp.

    Args:
        resource: Resource key for reranker in ResourceHub (e.g., "bge-m3")
        inputs: Input variable mappings
        outputs: Output variable mappings
        **kwargs: Additional keyword arguments for BaseOp
    """
    # Provider ops are I/O-bound by default (HTTP calls to reranker backends)
    kwargs.setdefault("bound", "io")
    super().__init__(**kwargs)

    self.resource = resource

    # Define input/output schema
    input_schema = {
        "query": Param(type=str, required=True),
        "documents": Param(type=list, required=True),
        "top_k": Param(type=int, default=-1),
        "threshold": Param(type=float, default=0.0),
    }

    output_schema = {
        "reranks": Param(type=list, required=True),
    }

    # Merge with user-provided
    self.inputs = self._merge_params(input_schema, inputs)
    self.outputs = self._merge_params(output_schema, outputs)

    # Reranker backend — lazy-initialized on first use to allow
    # graph construction before ResourceHub is set up
    self.backend = None
    self._initialized = False
    self._set_core(self._process)

Attributes¶

specific_metadata `property` ¶

specific_metadata: Dict[str, Any]

Return rerank-specific metadata dictionary.

Methods:¶

warmup ¶

warmup() -> None

Eagerly initialize reranker backend on engine startup.

Source code in operonx/providers/ops/rerank.py

def warmup(self) -> None:
    """Eagerly initialize reranker backend on engine startup."""
    self._ensure_initialized()

of ¶

of(resource=None, **kwargs) -> RerankOp

Create a RerankOp with flat kwargs.

Example::

rerank = RerankOp.of(resource="bge-m3", query=PARENT["q"], documents=PARENT["docs"])

Source code in operonx/providers/ops/rerank.py

@shorthand
def of(cls, resource=None, **kwargs) -> "RerankOp":
    """Create a RerankOp with flat kwargs.

    Example::

        rerank = RerankOp.of(resource="bge-m3", query=PARENT["q"], documents=PARENT["docs"])
    """
    input_mappings, init_kwargs = split_shorthand_kwargs(kwargs)
    return cls(resource=resource, inputs=input_mappings or None, **init_kwargs)

serialize ¶

serialize() -> dict

Serialize RerankOp for Rust backend, including backend config.

Source code in operonx/providers/ops/rerank.py

def serialize(self) -> dict:
    """Serialize RerankOp for Rust backend, including backend config."""
    self._ensure_initialized()
    base = super().serialize()
    base["resource"] = self.resource
    if self.backend and hasattr(self.backend, "config"):
        base["resource_config"] = self.backend.config.model_dump(mode="json")
    return base

Retrieval (1.1.0)¶

VectorSearchOp and DocFetchOp are a pair. The vector index is derived data holding vectors, ids, and filterable metadata; document content lives in the store of record. VectorSearchOp answers which documents, DocFetchOp answers what they say — and returns rows in the same order as the ids it was given, so hits stay aligned with their scores.

See the RAG guide for the full pipeline and operonx/providers/vector_stores/README.md for each backend's filter dialect.

VectorSearchOp ¶

VectorSearchOp(
    resource: Optional[str] = None,
    inputs: Dict[str, Any] = None,
    outputs: Dict[str, Any] = None,
    **kwargs: Any,
)

Bases: BaseOp

Op that runs vector similarity search via ResourceHub.

Inputs

query_vector (list[float]): Query embedding. Required. top_k (int): Number of hits. Default 10. filter (dict | str): Backend-native metadata filter — dict for most backends, an expression string for Milvus. Never translated by operonx; each backend validates its own dialect and raises on shapes it doesn't recognise. Default None. collection (str): Collection / table / index to search. Default None, meaning the resource's configured default.

Outputs

ids (list): Hit ids, best match first. scores (list[float]): Similarity per hit, index-aligned with ids. metadata (list[dict]): Indexed filterable fields per hit, index-aligned. {} for backends that store none.

Example::

hits = VectorSearchOp.of(
    resource="docs",
    query_vector=emb["embeddings"][0],
    top_k=20,
    filter={"tenant": "acme"},
)
docs = DocFetchOp.of(resource="main", ids=hits["ids"], collection="docs")

Initialize VectorSearchOp.

Parameters:

Name	Type	Description	Default
`resource`	`Optional[str]`	Resource key for the vector store. A bare name is looked up as `vector_store:{resource}`; a key that already contains `:` is used verbatim.	`None`
`inputs`	`Dict[str, Any]`	Input variable mappings.	`None`
`outputs`	`Dict[str, Any]`	Output variable mappings.	`None`
`**kwargs`	`Any`	Additional keyword arguments for BaseOp.	`{}`

Source code in operonx/providers/ops/vector_search.py

def __init__(
    self,
    resource: Optional[str] = None,
    inputs: Dict[str, Any] = None,
    outputs: Dict[str, Any] = None,
    **kwargs: Any,
):
    """Initialize VectorSearchOp.

    Args:
        resource: Resource key for the vector store. A bare name is
            looked up as ``vector_store:{resource}``; a key that
            already contains ``:`` is used verbatim.
        inputs: Input variable mappings.
        outputs: Output variable mappings.
        **kwargs: Additional keyword arguments for BaseOp.
    """
    # Networked stores dominate, so I/O is the right default. FAISS and
    # other in-process indices override this from their `bound` class
    # attribute once the resource resolves — see _ensure_initialized.
    kwargs.setdefault("bound", "io")
    super().__init__(**kwargs)

    self.resource = resource

    input_schema = {
        "query_vector": Param(type=list, required=True),
        "top_k": Param(type=int, required=False, default=10),
        "filter": Param(type=(dict, str), required=False, default=None),
        "collection": Param(type=str, required=False, default=None),
    }
    output_schema = {
        "ids": Param(type=list, required=True),
        "scores": Param(type=list, required=True),
        "metadata": Param(type=list, required=False),
    }

    self.inputs = self._merge_params(input_schema, inputs)
    self.outputs = self._merge_params(output_schema, outputs)

    self.backend = None
    self._initialized = False
    self._set_core(self._process)

Attributes¶

specific_metadata `property` ¶

specific_metadata: Dict[str, Any]

Return vector-search-specific metadata.

Methods:¶

warmup ¶

warmup() -> None

Eagerly resolve the backend on engine startup.

Source code in operonx/providers/ops/vector_search.py

def warmup(self) -> None:
    """Eagerly resolve the backend on engine startup."""
    self._ensure_initialized()

of ¶

of(resource=None, **kwargs) -> VectorSearchOp

Create a VectorSearchOp with flat kwargs.

Example::

hits = VectorSearchOp.of(resource="docs", query_vector=emb["embeddings"][0])

Source code in operonx/providers/ops/vector_search.py

@shorthand
def of(cls, resource=None, **kwargs) -> "VectorSearchOp":
    """Create a VectorSearchOp with flat kwargs.

    Example::

        hits = VectorSearchOp.of(resource="docs", query_vector=emb["embeddings"][0])
    """
    input_mappings, init_kwargs = split_shorthand_kwargs(kwargs)
    return cls(resource=resource, inputs=input_mappings or None, **init_kwargs)

serialize ¶

serialize() -> dict

Serialize for the Rust backend, including resource config.

Source code in operonx/providers/ops/vector_search.py

def serialize(self) -> dict:
    """Serialize for the Rust backend, including resource config."""
    self._ensure_initialized()
    base = super().serialize()
    base["resource"] = self.resource
    if self.backend and hasattr(self.backend, "config"):
        base["resource_config"] = self.backend.config.model_dump(mode="json")
    return base

DocFetchOp ¶

DocFetchOp(
    resource: Optional[str] = None,
    inputs: Dict[str, Any] = None,
    outputs: Dict[str, Any] = None,
    **kwargs: Any,
)

Bases: BaseOp

Op that hydrates ids into records via ResourceHub.

Inputs

ids (list): Primary keys to fetch. Required — typically VectorSearchOp's ids output. collection (str): Table / collection. Default None, meaning the resource's configured default. fields (list[str]): Column projection. Default None = all. id_field (str): Primary-key field name. Default None, meaning the resource's configured default (itself defaulting to "id").

Outputs

rows (list[dict]): Records in ids order. missing (list): Ids that matched no record.

Two guarantees, and they are why this is an op rather than a snippet:

rows follows ids order. Search returns score-ordered ids; SELECT … WHERE id = ANY(…) returns arbitrary order. Zipping them naively pairs every document with the wrong score — silently.
Missing ids surface in missing instead of quietly shortening rows. An index that has drifted from the store of record is a real condition and should be observable.

Scope is fetch-by-ids with an optional projection. No joins, writes, transactions, or custom SQL — for those, write a bare @op against your own client.

Example::

hits = VectorSearchOp.of(resource="docs", query_vector=emb["embeddings"][0], top_k=20)
docs = DocFetchOp.of(resource="main", ids=hits["ids"],
                     collection="docs", fields=["id", "title", "content"])

Initialize DocFetchOp.

Parameters:

Name	Type	Description	Default
`resource`	`Optional[str]`	Resource key for the document store. A bare name is looked up as `doc_store:{resource}`; a key that already contains `:` is used verbatim.	`None`
`inputs`	`Dict[str, Any]`	Input variable mappings.	`None`
`outputs`	`Dict[str, Any]`	Output variable mappings.	`None`
`**kwargs`	`Any`	Additional keyword arguments for BaseOp.	`{}`

Source code in operonx/providers/ops/doc_fetch.py

def __init__(
    self,
    resource: Optional[str] = None,
    inputs: Dict[str, Any] = None,
    outputs: Dict[str, Any] = None,
    **kwargs: Any,
):
    """Initialize DocFetchOp.

    Args:
        resource: Resource key for the document store. A bare name is
            looked up as ``doc_store:{resource}``; a key that already
            contains ``:`` is used verbatim.
        inputs: Input variable mappings.
        outputs: Output variable mappings.
        **kwargs: Additional keyword arguments for BaseOp.
    """
    kwargs.setdefault("bound", "io")
    super().__init__(**kwargs)

    self.resource = resource

    input_schema = {
        "ids": Param(type=list, required=True),
        "collection": Param(type=str, required=False, default=None),
        "fields": Param(type=list, required=False, default=None),
        "id_field": Param(type=str, required=False, default=None),
    }
    output_schema = {
        "rows": Param(type=list, required=True),
        "missing": Param(type=list, required=False),
    }

    self.inputs = self._merge_params(input_schema, inputs)
    self.outputs = self._merge_params(output_schema, outputs)

    self.backend = None
    self._initialized = False
    self._set_core(self._process)

Attributes¶

specific_metadata `property` ¶

specific_metadata: Dict[str, Any]

Return doc-fetch-specific metadata.

Methods:¶

warmup ¶

warmup() -> None

Eagerly resolve the backend on engine startup.

Source code in operonx/providers/ops/doc_fetch.py

def warmup(self) -> None:
    """Eagerly resolve the backend on engine startup."""
    self._ensure_initialized()

of ¶

of(resource=None, **kwargs) -> DocFetchOp

Create a DocFetchOp with flat kwargs.

Example::

docs = DocFetchOp.of(resource="main", ids=hits["ids"], collection="docs")

Source code in operonx/providers/ops/doc_fetch.py

@shorthand
def of(cls, resource=None, **kwargs) -> "DocFetchOp":
    """Create a DocFetchOp with flat kwargs.

    Example::

        docs = DocFetchOp.of(resource="main", ids=hits["ids"], collection="docs")
    """
    input_mappings, init_kwargs = split_shorthand_kwargs(kwargs)
    return cls(resource=resource, inputs=input_mappings or None, **init_kwargs)

serialize ¶

serialize() -> dict

Serialize for the Rust backend, including resource config.

Source code in operonx/providers/ops/doc_fetch.py

def serialize(self) -> dict:
    """Serialize for the Rust backend, including resource config."""
    self._ensure_initialized()
    base = super().serialize()
    base["resource"] = self.resource
    if self.backend and hasattr(self.backend, "config"):
        base["resource_config"] = self.backend.config.model_dump(mode="json")
    return base

Ordering helpers¶

Vector search returns score-ordered ids; key-based fetches return arbitrary order. DocFetchOp reconciles them internally — these are exported for anyone writing their own fetch op.

reorder_by_ids ¶

reorder_by_ids(
    rows: Sequence[Dict[str, Any]], ids: Sequence[Any], key: str = "id"
) -> List[Dict[str, Any]]

Return rows ordered to match ids, dropping ids with no row.

Parameters:

Name	Type	Description	Default
`rows`	`Sequence[Dict[str, Any]]`	Fetched records, in any order.	required
`ids`	`Sequence[Any]`	Desired order — typically `VectorSearchOp`'s `ids` output.	required
`key`	`str`	Field on each row holding its primary key.	`'id'`

Returns:

Type	Description
`List[Dict[str, Any]]`	Rows in `ids` order. Ids with no matching row are skipped; use
`List[Dict[str, Any]]`	func:`partition_by_ids` when you need to know which.

Note

Ids are matched by value, so an int id from the vector index will not match a str key from the document store. Keep the two stores' key types aligned.

Source code in operonx/providers/doc_stores/_reorder.py

def reorder_by_ids(
    rows: Sequence[Dict[str, Any]],
    ids: Sequence[Any],
    key: str = "id",
) -> List[Dict[str, Any]]:
    """Return ``rows`` ordered to match ``ids``, dropping ids with no row.

    Args:
        rows: Fetched records, in any order.
        ids: Desired order — typically ``VectorSearchOp``'s ``ids`` output.
        key: Field on each row holding its primary key.

    Returns:
        Rows in ``ids`` order. Ids with no matching row are skipped; use
        :func:`partition_by_ids` when you need to know which.

    Note:
        Ids are matched by value, so an ``int`` id from the vector index
        will not match a ``str`` key from the document store. Keep the two
        stores' key types aligned.
    """
    ordered, _ = partition_by_ids(rows, ids, key)
    return ordered

partition_by_ids ¶

partition_by_ids(
    rows: Sequence[Dict[str, Any]], ids: Sequence[Any], key: str = "id"
) -> Tuple[List[Dict[str, Any]], List[Any]]

Split a fetch result into ordered rows and missing ids.

Missing ids are a real condition, not noise: they mean the derived index has drifted from the store of record (deleted document, failed sync). Surfacing them beats silently returning a shorter list.

Parameters:

Name	Type	Description	Default
`rows`	`Sequence[Dict[str, Any]]`	Fetched records, in any order.	required
`ids`	`Sequence[Any]`	Desired order.	required
`key`	`str`	Field on each row holding its primary key.	`'id'`

Returns:

Type	Description
`List[Dict[str, Any]]`	`(ordered_rows, missing_ids)`. Duplicate ids in `ids` each
`List[Any]`	yield the same row; duplicate keys in `rows` keep the first.

Source code in operonx/providers/doc_stores/_reorder.py

def partition_by_ids(
    rows: Sequence[Dict[str, Any]],
    ids: Sequence[Any],
    key: str = "id",
) -> Tuple[List[Dict[str, Any]], List[Any]]:
    """Split a fetch result into ordered rows and missing ids.

    Missing ids are a real condition, not noise: they mean the derived
    index has drifted from the store of record (deleted document, failed
    sync). Surfacing them beats silently returning a shorter list.

    Args:
        rows: Fetched records, in any order.
        ids: Desired order.
        key: Field on each row holding its primary key.

    Returns:
        ``(ordered_rows, missing_ids)``. Duplicate ids in ``ids`` each
        yield the same row; duplicate keys in ``rows`` keep the first.
    """
    by_id: Dict[Any, Dict[str, Any]] = {}
    for row in rows:
        if key in row:
            by_id.setdefault(row[key], row)

    ordered: List[Dict[str, Any]] = []
    missing: List[Any] = []
    for id_ in ids:
        row = by_id.get(id_)
        if row is None:
            missing.append(id_)
        else:
            ordered.append(row)
    return ordered, missing

Structured output (1.0.0)¶

LLMOp gained inline parsing + validators + error-guided semantic retry in 1.0.0 — pass fields=, parser=, validators=, max_retries= directly to LLMOp.of() (see class docs above). The old standalone ask() helper was removed.

For pure text parsing without an LLM call, use the pure functions in operonx.providers.parsing:

parse_and_extract ¶

parse_and_extract(
    text: str,
    parser: ParserFormat,
    fields: List[ExtractField],
    validators: Optional[Dict[str, List[Any]]] = None,
) -> Dict[str, Any]

Parse text, extract fields, and optionally validate.

Always returns a dict shaped as {**field_values, "error": None|str}. Never raises — the error value tells the caller whether to retry. Semantics match the old ParserOp._process exactly so the surface LLMOp exposes is a faithful merge of what ask() provided before.

Source code in operonx/providers/parsing.py

def parse_and_extract(
    text: str,
    parser: ParserFormat,
    fields: List[ExtractField],
    validators: Optional[Dict[str, List[Any]]] = None,
) -> Dict[str, Any]:
    """Parse ``text``, extract ``fields``, and optionally validate.

    Always returns a dict shaped as ``{**field_values, "error": None|str}``.
    Never raises — the ``error`` value tells the caller whether to retry.
    Semantics match the old ``ParserOp._process`` exactly so the surface
    LLMOp exposes is a faithful merge of what ``ask()`` provided before.
    """
    if validators is not None and not isinstance(validators, dict):
        return {
            "error": (f"validators must be a dict, got {type(validators).__name__}: {validators!r}")
        }
    if not text:
        return {"error": "Empty input text"}

    backend = _PARSER_MAP.get(parser)
    if backend is None:
        return {"error": f"Unknown parser format: {parser!r}"}

    try:
        parsed_data = backend(text)
    except Exception as e:
        return {"error": f"Parse error ({parser}): {e}"}

    result: Dict[str, Any] = {}
    missing: List[str] = []
    for field in fields:
        raw = _resolve_field(parsed_data, field.chain_path, parser)
        if raw is MISSING:
            if not field.optional:
                missing.append(".".join(field.chain_path))
            raw = None
        result[field.output_key] = convert_type(raw, field.type_hint)

    if validators:
        err = apply_validators(result, validators)
        if err is not None:
            return {"error": err}
        # A validator's ``@default`` counts as an answer, so a field it
        # filled is no longer missing.
        missing = [p for p in missing if result.get(p.split(".")[-1]) is None]

    if missing:
        # Well-formed output with the wrong keys is a semantic failure, and
        # reporting it as one is what lets ``max_retries`` fire. It used to
        # come back as ``{"result": None, "error": None}`` — indistinguishable
        # from the model answering null on purpose, which still is not an
        # error here.
        return {
            **result,
            "error": (
                f"Missing field(s) in {parser} output: {', '.join(missing)}. "
                f"Parsed keys: {sorted(parsed_data) if isinstance(parsed_data, dict) else '—'}"
            ),
        }

    result["error"] = None
    return result

ExtractField `dataclass` ¶

ExtractField(
    output_key: str,
    chain_path: List[str],
    type_hint: str,
    optional: bool = False,
)

A field to pull out of parsed text.

Attributes:

Name	Type	Description
`output_key`	`str`	Key under which the extracted value is returned.
`chain_path`	`List[str]`	Dot-separated path into the parsed dict.
`type_hint`	`str`	Type name used for coercion (`str` / `int` / `bool` / ...).
`optional`	`bool`	When True, absence is an answer rather than an error.

Methods:¶

from_string `classmethod` ¶

from_string(schema_str: str) -> ExtractField

Parse a schema string like "user.address.city: str".

Missing type hint defaults to Any.

A ? before the colon marks the field optional::

"result: str"        # required — absence is a parse error
"chosen_date?: str"  # optional — absence yields None

Optional matters for a union schema, where one field list covers several response shapes and most entries are expected to be absent on any given call. Without the marker every such call would report missing fields and burn its retries.

Source code in operonx/providers/parsing.py

@classmethod
def from_string(cls, schema_str: str) -> "ExtractField":
    """Parse a schema string like ``"user.address.city: str"``.

    Missing type hint defaults to ``Any``.

    A ``?`` before the colon marks the field optional::

        "result: str"        # required — absence is a parse error
        "chosen_date?: str"  # optional — absence yields None

    Optional matters for a **union schema**, where one field list
    covers several response shapes and most entries are expected to be
    absent on any given call. Without the marker every such call would
    report missing fields and burn its retries.
    """
    if ":" not in schema_str:
        schema_str += ": Any"
    chain_text, type_hint = schema_str.split(":", 1)
    chain_text = chain_text.strip()
    optional = chain_text.endswith("?")
    if optional:
        chain_text = chain_text[:-1].strip()
    chain_path = chain_text.split(".")
    return cls(
        output_key=chain_path[-1],
        chain_path=chain_path,
        type_hint=type_hint.strip(),
        optional=optional,
    )

Resource resolution¶

Backend selection happens by name, not by direct construction. Wire your resources.yaml:

llm:gpt-4o-mini:
  api_type: openai
  api_key: ${OPENAI_API_KEY}
  base_url: https://api.openai.com/v1
  model: gpt-4o-mini

embedding:openai:
  api_type: openai
  api_key: ${OPENAI_API_KEY}
  base_url: https://api.openai.com/v1
  model: text-embedding-3-small
  dimensions: 1536

Then reference by key in your op definitions:

llm = LLMOp.of(resource="gpt-4o-mini", messages=PARENT["msgs"])
embed = EmbeddingOp.of(resource="openai", texts=PARENT["docs"])

Full reference — including the five disambiguated failure branches when a key is missing or unset — is in Resource hub.

Config classes¶

The Pydantic models behind resources.yaml. You rarely construct these directly; the framework loads them from YAML. Listed here for reference.

LLM — LLMConfig, OpenAIConfig, AzureConfig, GeminiConfig, AnthropicConfig, LLMType (in operonx.providers.llms).
Embedding — EmbeddingConfig, EmbeddingType (in operonx.providers.embeddings).
Reranker — RerankingConfig, RerankingType (in operonx.providers.rerankers).
Vector store — VectorStoreConfig, VectorStoreType, VectorStoreMetric (in operonx.providers.vector_stores).
Document store — DocStoreConfig, DocStoreType (in operonx.providers.doc_stores).
Auth — KeycloakTokenConfig (in operonx.providers.auth).

Factory functions¶

Resolve a config to a backend instance. Used internally by ResourceHub — most users don't call these directly.

create_llm ¶

create_llm(config: LLMConfig) -> BaseLLM

Create an LLM backend from config.

Parameters:

Name	Type	Description	Default
`config`	`LLMConfig`	LLMConfig with api_type determining which backend to create.	required

Returns:

Type	Description
`BaseLLM`	BaseLLM instance.

Raises:

Type	Description
`ValueError`	If api_type is unsupported.
`ImportError`	With a helpful pointer to the right `operonx[<extra>]` install when an optional dependency is missing.

Source code in operonx/providers/llms/factory.py

def create_llm(config: LLMConfig) -> BaseLLM:
    """Create an LLM backend from config.

    Args:
        config: LLMConfig with api_type determining which backend to create.

    Returns:
        BaseLLM instance.

    Raises:
        ValueError: If api_type is unsupported.
        ImportError: With a helpful pointer to the right ``operonx[<extra>]``
            install when an optional dependency is missing.
    """
    if config.api_type in [LLMType.VLLM, LLMType.OPENAI]:
        try:
            from .openai import OpenAISDKModel
        except ImportError as e:
            raise ImportError(_missing_extra_message("OpenAISDKModel", "providers", e)) from e
        return OpenAISDKModel(config=config)
    if config.api_type == LLMType.AZURE:
        try:
            from .azure import AzureSDKModel
        except ImportError as e:
            raise ImportError(_missing_extra_message("AzureSDKModel", "providers", e)) from e
        return AzureSDKModel(config=config)
    if config.api_type == LLMType.GEMINI:
        try:
            from .gemini import GeminiOpenAISDKModel
        except ImportError as e:
            raise ImportError(_missing_extra_message("Gemini", "gemini", e)) from e
        return GeminiOpenAISDKModel(config=config)
    if config.api_type == LLMType.ANTHROPIC:
        try:
            from .anthropic import AnthropicModel
        except ImportError as e:
            raise ImportError(_missing_extra_message("AnthropicModel", "anthropic", e)) from e
        return AnthropicModel(config=config)
    raise ValueError(f"Unsupported Model: {config.api_type}")

create_embedding ¶

create_embedding(config: EmbeddingConfig) -> BaseEmbedder

Create an embedding backend from config.

Parameters:

Name	Type	Description	Default
`config`	`EmbeddingConfig`	EmbeddingConfig with api_type determining which backend to create.	required

Returns:

Type	Description
`BaseEmbedder`	BaseEmbedder instance.

Raises:

Type	Description
`ValueError`	If api_type is unsupported.
`ImportError`	With a helpful pointer to the right `operonx[<extra>]` install when an optional dependency is missing.

Source code in operonx/providers/embeddings/factory.py

def create_embedding(config: EmbeddingConfig) -> BaseEmbedder:
    """Create an embedding backend from config.

    Args:
        config: EmbeddingConfig with api_type determining which backend to create.

    Returns:
        BaseEmbedder instance.

    Raises:
        ValueError: If api_type is unsupported.
        ImportError: With a helpful pointer to the right ``operonx[<extra>]``
            install when an optional dependency is missing.
    """
    if config.api_type == EmbeddingType.TEXT_EMBEDDING_INFERENCE:
        try:
            from operonx.providers.embeddings.tei import TEIEmbedding
        except ImportError as e:
            raise ImportError(_missing_extra_message("TEIEmbedding", "providers", e)) from e
        return TEIEmbedding(config)
    if config.api_type in (EmbeddingType.VLLM, EmbeddingType.OPENAI, EmbeddingType.AZURE):
        try:
            from operonx.providers.embeddings.vllm import VLLMEmbedding
        except ImportError as e:
            raise ImportError(_missing_extra_message("VLLMEmbedding", "providers", e)) from e
        return VLLMEmbedding(config)
    if config.api_type == EmbeddingType.HF:
        try:
            from operonx.providers.embeddings.huggingface import HFEmbedding
        except ImportError as e:
            raise ImportError(_missing_extra_message("HFEmbedding", "huggingface", e)) from e
        return HFEmbedding(config)
    if config.api_type == EmbeddingType.ONNX:
        try:
            from operonx.providers.embeddings.onnx import ONNXEmbedding
        except ImportError as e:
            raise ImportError(_missing_extra_message("ONNXEmbedding", "onnx", e)) from e
        return ONNXEmbedding(config)
    raise ValueError(f"Unsupported Model: {config.api_type}")

create_reranking ¶

create_reranking(config: RerankingConfig) -> BaseReranker

Create a reranking backend from config.

Parameters:

Name	Type	Description	Default
`config`	`RerankingConfig`	RerankingConfig with api_type determining which backend to create.	required

Returns:

Type	Description
`BaseReranker`	BaseReranker instance.

Raises:

Type	Description
`ValueError`	If api_type is unsupported.
`ImportError`	With a helpful pointer to the right `operonx[<extra>]` install when an optional dependency is missing.

Source code in operonx/providers/rerankers/factory.py

def create_reranking(config: RerankingConfig) -> BaseReranker:
    """Create a reranking backend from config.

    Args:
        config: RerankingConfig with api_type determining which backend to create.

    Returns:
        BaseReranker instance.

    Raises:
        ValueError: If api_type is unsupported.
        ImportError: With a helpful pointer to the right ``operonx[<extra>]``
            install when an optional dependency is missing.
    """
    if config.api_type == RerankingType.TEXT_EMBEDDING_INFERENCE:
        try:
            from operonx.providers.rerankers.tei import TEIReranker
        except ImportError as e:
            raise ImportError(_missing_extra_message("TEIReranker", "providers", e)) from e
        return TEIReranker(config)
    if config.api_type == RerankingType.VLLM:
        try:
            from operonx.providers.rerankers.vllm import VLLMReranker
        except ImportError as e:
            raise ImportError(_missing_extra_message("VLLMReranker", "providers", e)) from e
        return VLLMReranker(config)
    if config.api_type == RerankingType.PINECONE:
        try:
            from operonx.providers.rerankers.pinecone import PineconeReranker
        except ImportError as e:
            raise ImportError(_missing_extra_message("PineconeReranker", "providers", e)) from e
        return PineconeReranker(config)
    if config.api_type == RerankingType.HF:
        try:
            from operonx.providers.rerankers.huggingface import HFReranker
        except ImportError as e:
            raise ImportError(_missing_extra_message("HFReranker", "huggingface", e)) from e
        return HFReranker(config)
    if config.api_type == RerankingType.ONNX:
        try:
            from operonx.providers.rerankers.onnx import ONNXReranker
        except ImportError as e:
            raise ImportError(_missing_extra_message("ONNXReranker", "onnx", e)) from e
        return ONNXReranker(config)
    raise ValueError(f"Unsupported Model: {config.api_type}")

create_vector_store ¶

create_vector_store(config: VectorStoreConfig) -> BaseVectorStore

Create a vector store backend from config.

Parameters:

Name	Type	Description	Default
`config`	`VectorStoreConfig`	VectorStoreConfig whose api_type selects the backend.	required

Returns:

Type	Description
`BaseVectorStore`	BaseVectorStore instance.

Raises:

Type	Description
`ValueError`	If api_type is unsupported.
`ImportError`	With a pointer to the right `operonx[<extra>]` install when an optional dependency is missing.

Source code in operonx/providers/vector_stores/factory.py

def create_vector_store(config: VectorStoreConfig) -> BaseVectorStore:
    """Create a vector store backend from config.

    Args:
        config: VectorStoreConfig whose api_type selects the backend.

    Returns:
        BaseVectorStore instance.

    Raises:
        ValueError: If api_type is unsupported.
        ImportError: With a pointer to the right ``operonx[<extra>]``
            install when an optional dependency is missing.
    """
    if config.api_type == VectorStoreType.FAISS:
        try:
            from operonx.providers.vector_stores.faiss import FaissVectorStore
        except ImportError as e:
            raise ImportError(_missing_extra_message("FaissVectorStore", "faiss", e)) from e
        return FaissVectorStore(config)
    if config.api_type == VectorStoreType.PGVECTOR:
        try:
            from operonx.providers.vector_stores.pgvector import PgVectorStore
        except ImportError as e:
            raise ImportError(_missing_extra_message("PgVectorStore", "pgvector", e)) from e
        return PgVectorStore(config)
    if config.api_type == VectorStoreType.QDRANT:
        try:
            from operonx.providers.vector_stores.qdrant import QdrantVectorStore
        except ImportError as e:
            raise ImportError(_missing_extra_message("QdrantVectorStore", "qdrant", e)) from e
        return QdrantVectorStore(config)
    raise ValueError(f"Unsupported vector store: {config.api_type}")

create_doc_store ¶

create_doc_store(config: DocStoreConfig) -> BaseDocStore

Create a document store backend from config.

Parameters:

Name	Type	Description	Default
`config`	`DocStoreConfig`	DocStoreConfig whose api_type selects the backend.	required

Returns:

Type	Description
`BaseDocStore`	BaseDocStore instance.

Raises:

Type	Description
`ValueError`	If api_type is unsupported.
`ImportError`	With a pointer to the right `operonx[<extra>]` install when an optional dependency is missing.

Source code in operonx/providers/doc_stores/factory.py

def create_doc_store(config: DocStoreConfig) -> BaseDocStore:
    """Create a document store backend from config.

    Args:
        config: DocStoreConfig whose api_type selects the backend.

    Returns:
        BaseDocStore instance.

    Raises:
        ValueError: If api_type is unsupported.
        ImportError: With a pointer to the right ``operonx[<extra>]``
            install when an optional dependency is missing.
    """
    if config.api_type == DocStoreType.MEMORY:
        # No optional dependency — always importable.
        from operonx.providers.doc_stores.memory import MemoryDocStore

        return MemoryDocStore(config)
    if config.api_type == DocStoreType.POSTGRES:
        try:
            from operonx.providers.doc_stores.postgres import PostgresDocStore
        except ImportError as e:
            raise ImportError(_missing_extra_message("PostgresDocStore", "postgres", e)) from e
        return PostgresDocStore(config)
    # MONGO and REDIS are declared in DocStoreType but no backend module
    # exists yet. Reporting them as a missing *extra* sent users to
    # ``pip install operonx[mongo]`` — an extra that has never existed — so
    # the install appeared to be the fix for something no install provides.
    if config.api_type in (DocStoreType.MONGO, DocStoreType.REDIS):
        raise NotImplementedError(
            f"Doc store backend '{config.api_type.value}' is declared in "
            f"DocStoreType but not implemented. Available today: "
            f"{DocStoreType.MEMORY.value}, {DocStoreType.POSTGRES.value}."
        )
    raise ValueError(f"Unsupported doc store: {config.api_type}")

create_auth ¶

create_auth(config: KeycloakTokenConfig) -> KeycloakTokenProvider

Create a KeycloakTokenProvider from config.

Parameters:

Name	Type	Description	Default
`config`	`KeycloakTokenConfig`	KeycloakTokenConfig instance.	required

Returns:

Type	Description
`KeycloakTokenProvider`	KeycloakTokenProvider instance.

Raises:

Type	Description
`ImportError`	with a pointer to the right `operonx[<extra>]` install if `httpx` (or another keycloak dep) is missing.

Source code in operonx/providers/auth/factory.py

def create_auth(config: KeycloakTokenConfig) -> "KeycloakTokenProvider":  # noqa: F821
    """Create a KeycloakTokenProvider from config.

    Args:
        config: KeycloakTokenConfig instance.

    Returns:
        KeycloakTokenProvider instance.

    Raises:
        ImportError: with a pointer to the right `operonx[<extra>]`
            install if `httpx` (or another keycloak dep) is missing.
    """
    try:
        from .keycloak import KeycloakTokenProvider
    except ImportError as e:
        raise ImportError(
            "KeycloakTokenProvider requires additional packages.\n"
            "  Install with: pip install operonx[providers]\n"
            f"  Original error: {e}"
        ) from e
    return KeycloakTokenProvider(config)

operonx.providers¶

Provider ops¶

LLMOp ¶

Attributes¶

specific_metadata property ¶

Methods:¶

warmup ¶

normalize_trace_io ¶

of ¶

serialize ¶

EmbeddingOp ¶

Attributes¶

specific_metadata property ¶

Methods:¶

warmup ¶

of ¶

serialize ¶

RerankOp ¶

Attributes¶

specific_metadata property ¶

Methods:¶

warmup ¶

of ¶

serialize ¶

Retrieval (1.1.0)¶

VectorSearchOp ¶

Attributes¶

specific_metadata property ¶

Methods:¶

warmup ¶

of ¶

serialize ¶

DocFetchOp ¶

Attributes¶

specific_metadata property ¶

Methods:¶

warmup ¶

of ¶

serialize ¶

Ordering helpers¶

reorder_by_ids ¶

partition_by_ids ¶

Structured output (1.0.0)¶

parse_and_extract ¶

ExtractField dataclass ¶

Methods:¶

from_string classmethod ¶

Resource resolution¶

Config classes¶

Factory functions¶

create_llm ¶

create_embedding ¶

create_reranking ¶

create_vector_store ¶

create_doc_store ¶

create_auth ¶

specific_metadata `property` ¶

specific_metadata `property` ¶

specific_metadata `property` ¶

specific_metadata `property` ¶

specific_metadata `property` ¶

ExtractField `dataclass` ¶

from_string `classmethod` ¶