pyflink.dataframe.ai.llm.LLMAccessor.ai_classify#
- LLMAccessor.ai_classify(input_col: Union[str, Expression], labels: List[str], *, provider: Optional[str] = None, model: Optional[str] = None, cache_table: Optional[str] = None, cache_key: Optional[Union[str, List[str]]] = None, config: Optional[Dict[str, str]] = None, **kwargs) DataFrame[source]#
Classify text into one of the provided labels.
- Parameters
input_col – Column name (str) or Expression for the input text.
labels – List of label strings.
provider – Registered model provider lookup name. Uses default if not specified. If no provider is configured,
modelis treated as a catalog model name.model – Model name.
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.
Note: This function uses its own task-specific prompt. Provider configurations for the system prompt, such as
system_promptonOpenAICompatProviderandDashScopeProvider, do not take effect for this call.
- Returns
A new DataFrame with columns appended –
category(STRING): the predicted label.confidence(DOUBLE): confidence score.
Examples
Simple invocation:
- ::
>>> import pyflink.dataframe as pf >>> pf.set_model_provider(pf.OpenAICompatProvider( ... task="chat/completions")) >>> df = pf.from_dict({"text": ["The delivery was fast."]}) >>> df.llm.ai_classify("text", ["positive", "negative"], ... model="qwen3.6-plus")
With cache:
- ::
>>> df.llm.ai_classify( ... "text", ["positive", "negative"], model="qwen3.6-plus", ... cache_table="`fluss-catalog`.default_database.classify_cache", ... cache_key="text")
Using runtime config supported by
pyflink.dataframe.OpenAICompatProvider:- ::
>>> df.llm.ai_classify( ... "text", ["positive", "negative"], model="qwen3.6-plus", ... config={"max-concurrent-operations": "20"})