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[Validators] = None,
    max_retries: int = 0,
    retry_hint: bool = True,
    on_failure: str = "raise",
    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): One {"id", "name", "args"} dict per call (empty list when absent), whichever provider answered. args is a dict, or the model's text when it is not a JSON object. Sent back in an assistant message, the backends turn it into their own wire form. usage (dict): Flat token-cost metrics. extras (dict): Bag of uncommon fields (thinking_content, refusal, logprobs, citations — a provider's native citations as spans of content; Anthropic only for now, None elsewhere).

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`. With `stream=True` only until the first delta is yielded: after that a failure propagates, because a fallback restarts the answer and would contradict the deltas already sent.	`None`
`batch_mode`	`bool`	Use OpenAI Batch API (50% cheaper). `fields` / `validators` / `max_retries` apply to the batch answer as to a live one (a semantic retry is another batch submission). Takes one resource and no `fallback`: a list of several resources (load balancing) or a fallback list raises `ValueError`.	`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, named after the path's last segment or an `as` alias (`"user.id as user_id: str"`). Two fields with one output name, or one named like an output of this op (`content`, `error`, ...), raise `ValueError`.	`None`
`parser`	`Optional[str]`	Parser format when `fields` is set: `"xml"`, `"json"`, or `"yaml"`. Defaults to `"xml"` when `fields` is provided.	`None`
`validators`	`Optional[Validators]`	Optional validation applied after extraction. Either a per-field allow-list — `{"field": [allowed, ...]}`, where an `"@default"` entry stands in for an unrecognised value (`"@@x"` is the literal `"@x"`; an absent `?` field is not checked) — or a `Callable[[dict], bool]` over the whole parsed dict, for structural checks an allow-list cannot state. A callable that raises counts as a rejection, not a crash.	`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 failures are handled separately by `_call_with_retry`, driven by the resource's own `max_retries`, and are 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`
`on_failure`	`str`	What a hard failure does once retries and `fallback` are spent — a timeout, a transport error, a refusal. `"raise"` (default) fails the op. `"error"` returns what a parse failure returns: every field `None` and `error` set — for a step whose output is optional, so a downstream op can carry on without it. Requires `fields`.	`'raise'`
`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["Validators"] = None,
    max_retries: int = 0,
    retry_hint: bool = True,
    on_failure: str = "raise",
    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``.
            With ``stream=True`` only until the first delta is yielded:
            after that a failure propagates, because a fallback restarts
            the answer and would contradict the deltas already sent.
        batch_mode: Use OpenAI Batch API (50% cheaper). ``fields`` /
            ``validators`` / ``max_retries`` apply to the batch answer as
            to a live one (a semantic retry is another batch submission).
            Takes one resource and no ``fallback``: a list of several
            resources (load balancing) or a fallback list raises
            ``ValueError``.
        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, named after the path's last segment or an
            ``as`` alias (``"user.id as user_id: str"``). Two fields with
            one output name, or one named like an output of this op
            (``content``, ``error``, ...), raise ``ValueError``.
        parser: Parser format when ``fields`` is set: ``"xml"``, ``"json"``,
            or ``"yaml"``. Defaults to ``"xml"`` when ``fields`` is provided.
        validators: Optional validation applied after extraction.
            Either a per-field allow-list — ``{"field": [allowed, ...]}``,
            where an ``"@default"`` entry stands in for an unrecognised
            value (``"@@x"`` is the literal ``"@x"``; an absent ``?``
            field is not checked) — or a ``Callable[[dict], bool]`` over
            the whole parsed dict, for structural checks an allow-list
            cannot state. A callable that raises counts as a rejection,
            not a crash.
        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 failures are
            handled separately by ``_call_with_retry``, driven by the
            resource's own ``max_retries``, and are 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.
        on_failure: What a **hard** failure does once retries and
            ``fallback`` are spent — a timeout, a transport error, a
            refusal. ``"raise"`` (default) fails the op. ``"error"``
            returns what a parse failure returns: every field ``None``
            and ``error`` set — for a step whose output is optional, so
            a downstream op can carry on without it. Requires ``fields``.
        inputs: Input variable mappings.
        outputs: Output variable mappings.
        **kwargs: Additional keyword arguments for BaseOp.
    """
    kwargs.setdefault("bound", "io")
    super().__init__(**kwargs)

    # ``resource=`` given as a graph input: chosen per call (_runtime_resource).
    resource, inputs, self._dynamic = take_resource(resource, inputs)
    self._bound = {}
    if self._dynamic and (ratios is not None or batch_mode):
        raise ValueError(
            "LLMOp: resource= as a graph input takes one key per call; ratios= and "
            "batch_mode= need their resources fixed when the graph is built"
        )
    self.batch_mode = batch_mode
    self.contain_generation = True
    self.fallback = fallback
    self._rng = random.Random(seed)

    # Structured-output config (merged from ParserOp).
    _check_validators(validators)
    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}")
    if batch_mode and fallback:
        # The batch coordinator is bound to the primary resource. A
        # fallback would have to be a live call at full price, or another
        # batch round on a resource nobody configured for batching —
        # neither is what the caller chose, and ignoring the list is how
        # this used to go wrong.
        raise ValueError(
            "LLMOp(batch_mode=True) does not take fallback=[...]: a batch "
            "request is answered by the primary resource's batch job, and "
            "falling back would turn it into a live, full-price call. Drop "
            "fallback=, or run the op live (batch_mode=False)."
        )
    if batch_mode and isinstance(resource, list) and len(resource) > 1:
        # A list is load balancing, but the batch path builds one
        # coordinator for resource[0] and sends every request there:
        # the rest of the list and ``ratios`` did nothing. Balancing
        # batch traffic would take a coordinator and batch job per
        # resource, and only some backends can batch at all (the
        # OpenAI one implements generate_batch; a mixed list would fail
        # one flush interval later). Refusing costs nothing that worked.
        raise ValueError(
            f"LLMOp(batch_mode=True) takes one resource, got {resource!r}: "
            f"a batch job belongs to one resource, so every request would go "
            f"to {resource[0]!r} and the rest of the list and ratios= would be "
            f"ignored. Pass resource={resource[0]!r}, or one batch op per resource."
        )
    if on_failure not in ("raise", "error"):
        raise ValueError(f"on_failure is 'raise' or 'error', got {on_failure!r}")
    if on_failure == "error" and not fields:
        raise TypeError(
            "LLMOp(on_failure='error') requires fields=[...] — the failure is "
            "reported in the `error` output, which only structured mode has."
        )
    self.on_failure = on_failure
    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
    # Show keys: with extraction the parsed fields are the answer, not
    # the raw content — unless the caller declared their own.
    if kwargs.get("show_keys") is None and self._extract_fields:
        self.show_keys = tuple(f.output_key for f in self._extract_fields)

    # 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),
    }
    if self._dynamic:
        input_schema["resource"] = RESOURCE_PARAM

    output_schema = {
        "role": Param(type=str, default="assistant"),
        # What this frame adds: a token delta while streaming, ""
        # on the closing frame, the whole answer for a batch call.
        "content": Param(type=str, required=True),
        # The whole answer. Set on the closing frame of a stream and on
        # a batch call; None on a delta. See ``_stream_final``.
        "full_content": Param(type=str, default=None),
        # Streaming only: False on a token delta, True on the closing
        # frame. Batch calls are always 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={}),
        "cost_usd": Param(type=float, default=None),
        "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:
        # Each field is one output key. Two sharing a key, or one
        # shadowing the op's own outputs, would lose a value at merge
        # time with nothing reported — so it is refused here.
        collision = check_output_keys(self._extract_fields, reserved=(*output_schema, "error"))
        if collision is not None:
            raise ValueError(collision)
        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
    core = "_stream_core" if self.stream else "_generate_core"
    self._set_core(per_call(self, core) if self._dynamic else getattr(self, 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."""
    if self._dynamic:  # the key arrives with each call
        return
    self._ensure_initialized()

normalize_trace_io ¶

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

Record the request this call sent, as messages, beside its variables.

A template (prompt=) is rendered with the recorded variables by the op's own rule, so the trace shows the conversation the model received rather than a template a reader has to fill in. A conversation passed as messages= is already that, and is kept as is. Multimodal blocks (image_url, input_audio) are wrapped as Media for the consumers to offload. Real op state is untouched.

inputs is the trace's filtered copy, so exclude=/include= hold: messages itself can be hidden, and a template whose variable is hidden is not rendered — the rendering would put the hidden value back. A template that does not render failed the op with the same PromptError; that failure is the record.

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

def normalize_trace_io(self, inputs: Dict[str, Any], outputs: Dict[str, Any]) -> tuple:
    """Record the request this call sent, as ``messages``, beside its variables.

    A template (``prompt=``) is rendered with the recorded variables by
    the op's own rule, so the trace shows the conversation the model
    received rather than a template a reader has to fill in. A
    conversation passed as ``messages=`` is already that, and is kept
    as is. Multimodal blocks (``image_url``, ``input_audio``) are
    wrapped as ``Media`` for the consumers to offload. Real op state is
    untouched.

    *inputs* is the trace's filtered copy, so ``exclude=``/``include=``
    hold: ``messages`` itself can be hidden, and a template whose
    variable is hidden is not rendered — the rendering would put the
    hidden value back. A template that does not render failed the op
    with the same ``PromptError``; that failure is the record.
    """
    if not should_emit_for_channel(self, "messages", "trace"):
        return inputs, outputs
    prompt = inputs.get("prompt")
    raw_messages = inputs.get("messages")
    if raw_messages is not None:
        messages = list(raw_messages)
    elif prompt is not None:
        vars = {k: v for k, v in inputs.items() if k not in RESERVED_KEYS}
        try:
            messages = self._build_messages(prompt, vars)
        except PromptError:
            return inputs, outputs
    else:
        return inputs, outputs
    return {**inputs, "messages": self._wrap_openai_media_blocks(messages)}, outputs

of ¶

of(
    resource=None,
    *,
    ratios=None,
    fallback=None,
    batch_mode=False,
    seed=None,
    prompt=None,
    messages=None,
    fields=None,
    parser=None,
    validators: Optional[Validators] = None,
    max_retries=0,
    retry_hint=True,
    on_failure="raise",
    **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: Optional["Validators"] = None,
    max_retries=0,
    retry_hint=True,
    on_failure="raise",
    **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,
        on_failure=on_failure,
        inputs=input_mappings or None,
        **init_kwargs,
    )

serialize ¶

serialize() -> dict

Serialize LLMOp, including the resolved backend configs.

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

def serialize(self) -> dict:
    """Serialize LLMOp, including the resolved backend configs."""
    if not self._dynamic:  # a runtime resource has no backend to describe
        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)

    resource, inputs, self._dynamic = take_resource(resource, inputs)
    self._bound = {}
    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
    if self._dynamic:
        input_schema = {**input_schema, "resource": RESOURCE_PARAM}
    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(per_call(self, "_process") if self._dynamic else 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."""
    if self._dynamic:  # the key arrives with each call
        return
    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, including the resolved backend config.

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

def serialize(self) -> dict:
    """Serialize EmbeddingOp, including the resolved backend config."""
    if not self._dynamic:  # a runtime resource has no backend to describe
        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)

    resource, inputs, self._dynamic = take_resource(resource, inputs)
    self._bound = {}
    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
    if self._dynamic:
        input_schema = {**input_schema, "resource": RESOURCE_PARAM}
    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(per_call(self, "_process") if self._dynamic else 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."""
    if self._dynamic:  # the key arrives with each call
        return
    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, including the resolved backend config.

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

def serialize(self) -> dict:
    """Serialize RerankOp, including the resolved backend config."""
    if not self._dynamic:  # a runtime resource has no backend to describe
        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: VectorStoreOpBase

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. empty_index (bool): True when an unfiltered search found nothing, which a nearest-neighbour index does only when it holds no vectors. A WARNING is logged too: an index nobody populated otherwise answers every question with "no documents", and the pipeline downstream reads that as a real answer. Under a filter, no hits is a legitimate answer and is not flagged.

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")

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 the op.

    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)

    resource, inputs, self._dynamic = take_resource(resource, inputs)
    self._bound = {}
    self.resource = resource

    input_schema, output_schema = self._ports()
    if self._dynamic:
        input_schema = {**input_schema, "resource": RESOURCE_PARAM}
    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(per_call(self, "_process") if self._dynamic else self._process)

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)

    resource, inputs, self._dynamic = take_resource(resource, inputs)
    self._bound = {}
    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),
    }

    if self._dynamic:
        input_schema = {**input_schema, "resource": RESOURCE_PARAM}
    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(per_call(self, "_process") if self._dynamic else 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."""
    if self._dynamic:  # the key arrives with each call
        return
    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, including the resolved resource config.

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

def serialize(self) -> dict:
    """Serialize, including the resolved resource config."""
    if not self._dynamic:  # a runtime resource has no backend to describe
        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

VectorUpsertOp and VectorDeleteOp are the write half: they keep the index in step with the store of record as documents arrive and go.

VectorUpsertOp ¶

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

Bases: VectorStoreOpBase

Op that inserts or replaces vectors by id via ResourceHub.

Inputs

ids (list): Primary keys, one per vector. Required. Writing an id that is already there replaces its vector and metadata. vectors (list[list[float]]): Embeddings, index-aligned with ids. Required — typically EmbeddingOp's embeddings. metadata (list[dict]): Filterable fields, one dict per vector. Default None. Keep document content out: it belongs in the store of record. FAISS stores no metadata and ignores it; pgvector and Qdrant raise on a key outside metadata_columns. collection (str): Collection / table / index. Default None, meaning the resource's configured default.

Outputs

upserted (int): How many vectors were written.

Example::

emb = EmbeddingOp.of(resource="bge-m3", texts=chunks["texts"])
write = VectorUpsertOp.of(
    resource="docs",
    ids=chunks["ids"],
    vectors=emb["embeddings"],
    metadata=chunks["tags"],
)

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 the op.

    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)

    resource, inputs, self._dynamic = take_resource(resource, inputs)
    self._bound = {}
    self.resource = resource

    input_schema, output_schema = self._ports()
    if self._dynamic:
        input_schema = {**input_schema, "resource": RESOURCE_PARAM}
    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(per_call(self, "_process") if self._dynamic else self._process)

VectorDeleteOp ¶

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

Bases: VectorStoreOpBase

Op that removes vectors by id, or by a filter, via ResourceHub.

Inputs

ids (list): Primary keys to remove. Default None. filter (dict | str): Backend-native filter, the dialect VectorSearchOp takes; removes every vector it matches. Default None. FAISS cannot filter and raises. collection (str): Collection / table / index. Default None, meaning the resource's configured default.

Exactly one of ids and filter: neither, both, or an empty filter raise, because each could otherwise read as "delete everything". ids=[] removes nothing, and ids the index does not hold are not an error, so a cleanup pass can run again after a crash.

Outputs

deleted (int | None): How many vectors were removed; None when the backend does not report it (Qdrant).

Example::

gone = VectorDeleteOp.of(resource="docs", ids=diff["removed"])
purge = VectorDeleteOp.of(resource="docs", filter={"document_id": doc_id})

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 the op.

    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)

    resource, inputs, self._dynamic = take_resource(resource, inputs)
    self._bound = {}
    self.resource = resource

    input_schema, output_schema = self._ports()
    if self._dynamic:
        input_schema = {**input_schema, "resource": RESOURCE_PARAM}
    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(per_call(self, "_process") if self._dynamic else self._process)

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.

The structured layer applies to batch_mode=True too: the batch answer is parsed and validated like a live one, and a semantic retry is one more batch submission. batch_mode=True takes one resource: with fallback= or a list of several resources it raises ValueError at construction — the batch job belongs to one resource, a fallback would be a live, full-price call, and a load-balanced list used to send everything to its first entry.

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[Validators] = 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["Validators"] = 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) or callable(validators)):
        return {
            "error": (
                f"validators must be a dict or a callable, got "
                f"{type(validators).__name__}: {validators!r}"
            )
        }
    collision = check_output_keys(fields, reserved=("error",))
    if collision is not None:
        return {"error": collision}
    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[ExtractField] = []
    # Optional fields the payload left out: None, and not validated.
    absent: List[str] = []
    # output_key -> the structure found where a single value was declared.
    misshapen: Dict[str, Any] = {}
    # output_key -> (the value, why it is not the declared type).
    uncoerced: Dict[str, tuple] = {}
    for field in fields:
        raw = _resolve_field(parsed_data, field.chain_path, parser, field.type_hint)
        if raw is MISSING:
            if field.optional:
                absent.append(field.output_key)
            else:
                missing.append(field)
            raw = None
        elif _wants_scalar(field.type_hint) and _is_subtree(raw):
            # Not coerced: ``str()`` of a dict is a Python repr, ``int()``
            # leaves it a dict and ``bool()`` makes it True — each a
            # plausible answer with ``error: None``. It stays raw so a
            # validator's ``@default`` can stand in for it like for any
            # unrecognised value, and is an error below if nothing did.
            misshapen[field.output_key] = raw
            result[field.output_key] = raw
            continue
        try:
            result[field.output_key] = convert_type(raw, field.type_hint)
        except ValueError as e:
            # Kept raw, like a misshapen value, so a validator's
            # ``@default`` can stand in for it; an error below otherwise.
            uncoerced[field.output_key] = (raw, str(e))
            result[field.output_key] = raw

    if validators:
        err = apply_validators(result, validators, absent=absent)
        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. Looked up by output key: the path's
        # last segment is not the key once a field is aliased.
        missing = [f for f in missing if result.get(f.output_key) is None]

    # Identity, not equality: a default that replaced the structure is an
    # answer; the structure itself still sitting there is not.
    wrong_shape = [key for key, raw in misshapen.items() if result.get(key) is raw]
    if wrong_shape:
        for key in wrong_shape:
            result[key] = None
        hints = {f.output_key: f for f in fields}
        described = ", ".join(
            f"'{key}' is declared {hints[key].type_hint} but holds "
            f"{_describe_structure(misshapen[key])}"
            for key in wrong_shape
        )
        return {
            **result,
            "error": (
                f"Structure where a single value was expected in {parser} output: "
                f"{described}. Ask for a path inside it (e.g. "
                f"'{'.'.join(hints[wrong_shape[0]].chain_path)}.<child>: str') "
                f"or declare it ': dict'."
            ),
        }

    wrong_type = [key for key, (raw, _) in uncoerced.items() if result.get(key) is raw]
    if wrong_type:
        for key in wrong_type:
            result[key] = None
        hints = {f.output_key: f for f in fields}
        described = "; ".join(
            f"'{key}' is declared {hints[key].type_hint}, but {uncoerced[key][1]}"
            for key in wrong_type
        )
        return {**result, "error": f"Wrong type in {parser} output: {described}."}

    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: "
                f"{', '.join('.'.join(f.chain_path) for f in 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.

The output key is the path's last segment. as names it instead, as import a.b as c does — needed when two paths end in the same leaf, which would otherwise write one key::

"user.id as user_id: str"
"order.id as order_id?: str"   # optional, aliased

Raises:

Type	Description
`ValueError`	an `as` with no name, or a name that is not an identifier.

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.

    The output key is the path's last segment. ``as`` names it instead,
    as ``import a.b as c`` does — needed when two paths end in the same
    leaf, which would otherwise write one key::

        "user.id as user_id: str"
        "order.id as order_id?: str"   # optional, aliased

    Raises:
        ValueError: an ``as`` with no name, or a name that is not an
            identifier.
    """
    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()
    alias = None
    aliased = _ALIAS.fullmatch(chain_text)
    if aliased:
        chain_text, alias = aliased.group(1), (aliased.group(2) or "").strip()
        if not alias.isidentifier():
            raise ValueError(
                f"Field {schema_str!r}: 'as' must be followed by an output "
                f"name (an identifier), got {alias!r}."
            )
        # "user.id? as uid" reads as naturally as "user.id as uid?".
        if chain_text.endswith("?"):
            optional = True
            chain_text = chain_text[:-1].strip()
    chain_path = chain_text.split(".")
    return cls(
        output_key=alias or 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

An embedding resource's base_url depends on its api_type:

`api_type`	`base_url`	e.g.
`openai`	the API root, as for `llm:*` — the client adds `/embeddings`	`https://api.openai.com/v1`
`azure`	the resource endpoint; `model` is the deployment, `api_version` is required	`https://acme.openai.azure.com`
`vllm`, `tei`	the exact endpoint URL	`http://gpu-01:8000/v1/embeddings`
`triton`	a bare `host:port` (TLS is `ssl: true`)	`triton.internal:8001`

With dimensions set, an openai/azure resource sends it (models that can shorten their vectors do) and raises on a reply of any other width.

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 == LLMType.FAKE:
        from .fake import FakeLLM

        return FakeLLM(config=config)
    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.DB_ANTHROPIC:
        try:
            from .databricks import DatabricksAnthropic
        except ImportError as e:
            raise ImportError(_missing_extra_message("DatabricksAnthropic", "providers", e)) from e
        return DatabricksAnthropic(config=config)
    if config.api_type == LLMType.DB_GEMINI:
        try:
            from .databricks import DatabricksGemini
        except ImportError as e:
            raise ImportError(_missing_extra_message("DatabricksGemini", "providers", e)) from e
        return DatabricksGemini(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.OPENAI, EmbeddingType.AZURE):
        try:
            from operonx.providers.embeddings.openai import AzureOpenAIEmbedding, OpenAIEmbedding
        except ImportError as e:
            raise ImportError(_missing_extra_message("OpenAIEmbedding", "openai", e)) from e
        if config.api_type == EmbeddingType.AZURE:
            return AzureOpenAIEmbedding(config)
        return OpenAIEmbedding(config)
    if config.api_type == EmbeddingType.VLLM:
        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.TRITON:
        try:
            from operonx.providers.embeddings.triton import TritonEmbedding
        except ImportError as e:
            raise ImportError(_missing_extra_message("TritonEmbedding", "triton", e)) from e
        return TritonEmbedding(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)
    supported = sorted(t.value for t in EmbeddingType if t is not EmbeddingType.GEMINI)
    raise ValueError(
        f"Unsupported embedding api_type {config.api_type.value!r}; supported: {supported}"
    )

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 ¶

DocFetchOp ¶

Attributes¶

specific_metadata property ¶

Methods:¶

warmup ¶

of ¶

serialize ¶

VectorUpsertOp ¶

VectorDeleteOp ¶

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` ¶

ExtractField `dataclass` ¶

from_string `classmethod` ¶