pyflink.dataframe.ai.llm.LLMAccessor.predict#

LLMAccessor.predict(*input_cols: str, provider: Optional[str] = None, model: Optional[str] = None, output_type: Optional[Union[str, DataType]] = None, cache_table: Optional[str] = None, cache_key: Optional[Union[str, List[str]]] = None, config: Optional[Dict[str, str]] = None, **kwargs) → DataFrame[source]#

Perform prediction using a model.

This is the general-purpose prediction method.

When a provider is configured, a temporary model is created with the given input/output schema. When using a catalog model (no provider), the model’s registered schema is used and output_type is ignored.

Parameters

*input_cols – Column names to use as input.
provider – Registered model provider lookup name. Uses default if not specified. If no provider is configured, model is treated as a catalog model name.
model – Model name (e.g. “qwen3.6-plus”) or catalog model name.
output_type – Output type declaration. None defaults to a single STRING column named output. A SQL type string such as "VARIANT" appends a single column named output with that parsed type. A scalar or container DataType also appends a single column named output. A DataType.struct(...) appends one column per top-level struct field.
cache_table – Optional cache table identifier.
cache_key – Optional cache key column name or list of column names.
config – Optional dict of runtime prediction options.
**kwargs – Per-call overrides for the selected DataFrame model provider’s constructor parameters. Overrides are applied only when the call uses a configured provider.

Returns

A new DataFrame with the model output columns appended.

Examples

Runtime config keys depend on the provider.

Set up a provider and input DataFrame:

::

>>> import pyflink.dataframe as pf
>>> pf.set_model_provider(pf.OpenAICompatProvider(
...     task="chat/completions"))
>>> df = pf.from_dict({
...     "tenant_id": ["tenant-a"],
...     "question": ["Which products need review?"],
...     "image_url": ["https://example.com/product.jpg"],
... })

Default output: appends one output column:

::

>>> df.llm.predict("question", model="qwen3.6-plus")

Multiple input columns with per-column content types. content_type is passed through to pyflink.dataframe.OpenAICompatProvider as a per-call provider option:

::

>>> df.llm.predict("question", "image_url",
...                model="qwen3.6-plus",
...                content_type=["TEXT", "IMAGE_URL"])

Free-form JSON in one VARIANT column named output. response_format is passed through to pyflink.dataframe.OpenAICompatProvider as a per-call provider option:

::

>>> df.llm.predict("question", model="qwen3.6-plus",
...                output_type="VARIANT",
...                response_format="json_object")

Named output column:

::

>>> df.llm.predict("question", model="qwen3.6-plus",
...     output_type=pf.DataType.struct({
...         "answer": pf.DataType.string()
...     }))

Caching prediction results:

::

>>> df.llm.predict(
...     "question", model="qwen3.6-plus",
...     cache_table="`fluss-catalog`.default_database.predict_cache",
...     cache_key=["tenant_id", "question"])

Using runtime config supported by pyflink.dataframe.OpenAICompatProvider:

::

>>> df.llm.predict(
...     "question", model="qwen3.6-plus",
...     config={"max-concurrent-operations": "20"})