Skip to content

Task Module

openaivec.task

Prebuilt task factories and registry helpers.

Use function-based task factories from domain modules:

  • openaivec.task.nlp
  • openaivec.task.customer_support
  • openaivec.task.table

Registry helpers provide discoverable task lookup by key.

Migration note (breaking change): - Constant-style task names were removed. - Example: nlp.SENTIMENT_ANALYSIS -> nlp.sentiment_analysis() - Example: customer_support.INTENT_ANALYSIS -> customer_support.intent_analysis()

Classes

TaskSpec dataclass 

TaskSpec(
    key: str,
    domain: str,
    summary: str,
    factory: TaskFactory,
    response_format: type[Any],
)

Specification for a registered task factory.

Functions

list_tasks 

list_tasks(domain: str | None = None) -> list[str]

Return registered task keys, optionally filtered by domain prefix.

Source code in src/openaivec/task/__init__.py 
35
36
37
def list_tasks(domain: str | None = None) -> list[str]:
    """Return registered task keys, optionally filtered by domain prefix."""
    return [spec.key for spec in list_task_specs(domain=domain)]

get_task_spec 

get_task_spec(key: str) -> TaskSpec

Return the task spec for a registry key.

Source code in src/openaivec/task/__init__.py 
40
41
42
def get_task_spec(key: str) -> TaskSpec:
    """Return the task spec for a registry key."""
    return get_task_spec_or_raise(key)

get_task 

get_task(key: str, **kwargs: Any) -> PreparedTask[Any]

Create a prepared task instance from a registry key.

Source code in src/openaivec/task/__init__.py 
45
46
47
48
def get_task(key: str, **kwargs: Any) -> PreparedTask[Any]:
    """Create a prepared task instance from a registry key."""
    spec = get_task_spec_or_raise(key)
    return spec.factory(**kwargs)

Modules

customer_support

Customer support task factories.

Modules
customer_sentiment

Customer support sentiment task definition.

Classes
CustomerSentiment

Bases: BaseModel

Customer sentiment analysis output.

Functions
customer_sentiment 
customer_sentiment(
    business_context: str = "general customer support",
) -> PreparedTask[CustomerSentiment]

Create a customer support sentiment analysis task.

Source code in src/openaivec/task/customer_support/customer_sentiment.py 
63
64
65
66
67
68
def customer_sentiment(business_context: str = "general customer support") -> PreparedTask[CustomerSentiment]:
    """Create a customer support sentiment analysis task."""
    return PreparedTask(
        instructions=_build_instructions(business_context=business_context),
        response_format=CustomerSentiment,
    )
inquiry_classification

Customer inquiry-classification task definition.

Classes
InquiryClassification

Bases: BaseModel

Inquiry classification output.

Functions
inquiry_classification 
inquiry_classification(
    categories: Mapping[str, list[str]] | None = None,
    routing_rules: Mapping[str, str] | None = None,
    priority_rules: Mapping[str, str] | None = None,
    business_context: str = "general customer support",
    custom_keywords: Mapping[str, list[str]] | None = None,
) -> PreparedTask[InquiryClassification]

Create an inquiry classification task.

Source code in src/openaivec/task/customer_support/inquiry_classification.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def inquiry_classification(
    categories: Mapping[str, list[str]] | None = None,
    routing_rules: Mapping[str, str] | None = None,
    priority_rules: Mapping[str, str] | None = None,
    business_context: str = "general customer support",
    custom_keywords: Mapping[str, list[str]] | None = None,
) -> PreparedTask[InquiryClassification]:
    """Create an inquiry classification task."""
    resolved_categories = dict(categories or _default_categories())
    resolved_routing_rules = dict(routing_rules or _default_routing_rules())
    resolved_priority_rules = dict(priority_rules or _default_priority_rules())
    return PreparedTask(
        instructions=_build_instructions(
            categories=resolved_categories,
            routing_rules=resolved_routing_rules,
            priority_rules=resolved_priority_rules,
            business_context=business_context,
            custom_keywords=custom_keywords,
        ),
        response_format=InquiryClassification,
    )
inquiry_summary

Customer inquiry-summary task definition.

Classes
InquirySummary

Bases: BaseModel

Inquiry summary output.

Functions
inquiry_summary 
inquiry_summary(
    summary_length: str = "concise",
    business_context: str = "general customer support",
) -> PreparedTask[InquirySummary]

Create an inquiry summary task.

Source code in src/openaivec/task/customer_support/inquiry_summary.py 
54
55
56
57
58
59
60
61
62
def inquiry_summary(
    summary_length: str = "concise",
    business_context: str = "general customer support",
) -> PreparedTask[InquirySummary]:
    """Create an inquiry summary task."""
    return PreparedTask(
        instructions=_build_instructions(summary_length=summary_length, business_context=business_context),
        response_format=InquirySummary,
    )
intent_analysis

Customer inquiry intent-analysis task definition.

Classes
IntentAnalysis

Bases: BaseModel

Intent analysis output.

Functions
intent_analysis 
intent_analysis(
    business_context: str = "general customer support",
) -> PreparedTask[IntentAnalysis]

Create a customer intent analysis task.

Source code in src/openaivec/task/customer_support/intent_analysis.py 
66
67
68
69
70
71
def intent_analysis(business_context: str = "general customer support") -> PreparedTask[IntentAnalysis]:
    """Create a customer intent analysis task."""
    return PreparedTask(
        instructions=_build_instructions(business_context=business_context),
        response_format=IntentAnalysis,
    )
response_suggestion

Customer response-suggestion task definition.

Classes
ResponseSuggestion

Bases: BaseModel

Response suggestion output.

Functions
response_suggestion 
response_suggestion(
    response_style: str = "professional",
    company_name: str = "our company",
    business_context: str = "general customer support",
) -> PreparedTask[ResponseSuggestion]

Create a response suggestion task.

Source code in src/openaivec/task/customer_support/response_suggestion.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def response_suggestion(
    response_style: str = "professional",
    company_name: str = "our company",
    business_context: str = "general customer support",
) -> PreparedTask[ResponseSuggestion]:
    """Create a response suggestion task."""
    return PreparedTask(
        instructions=_build_instructions(
            response_style=response_style,
            company_name=company_name,
            business_context=business_context,
        ),
        response_format=ResponseSuggestion,
    )
urgency_analysis

Customer inquiry urgency-analysis task definition.

Classes
UrgencyAnalysis

Bases: BaseModel

Urgency analysis output.

Functions
urgency_analysis 
urgency_analysis(
    urgency_levels: Mapping[str, str] | None = None,
    response_times: Mapping[str, str] | None = None,
    customer_tiers: Mapping[str, str] | None = None,
    escalation_rules: Mapping[str, str] | None = None,
    urgency_keywords: Mapping[str, list[str]] | None = None,
    business_context: str = "general customer support",
    business_hours: str = "24/7 support",
    sla_rules: Mapping[str, str] | None = None,
) -> PreparedTask[UrgencyAnalysis]

Create an urgency analysis task.

Source code in src/openaivec/task/customer_support/urgency_analysis.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def urgency_analysis(
    urgency_levels: Mapping[str, str] | None = None,
    response_times: Mapping[str, str] | None = None,
    customer_tiers: Mapping[str, str] | None = None,
    escalation_rules: Mapping[str, str] | None = None,
    urgency_keywords: Mapping[str, list[str]] | None = None,
    business_context: str = "general customer support",
    business_hours: str = "24/7 support",
    sla_rules: Mapping[str, str] | None = None,
) -> PreparedTask[UrgencyAnalysis]:
    """Create an urgency analysis task."""
    resolved_urgency_levels = dict(urgency_levels or _default_urgency_levels())
    resolved_response_times = dict(response_times or _default_response_times())
    resolved_customer_tiers = dict(customer_tiers or _default_customer_tiers())
    resolved_escalation_rules = dict(escalation_rules or _default_escalation_rules())
    resolved_urgency_keywords = dict(urgency_keywords or _default_urgency_keywords())
    resolved_sla_rules = dict(sla_rules or _default_sla_rules())
    return PreparedTask(
        instructions=_build_instructions(
            urgency_levels=resolved_urgency_levels,
            response_times=resolved_response_times,
            customer_tiers=resolved_customer_tiers,
            escalation_rules=resolved_escalation_rules,
            urgency_keywords=resolved_urgency_keywords,
            business_context=business_context,
            business_hours=business_hours,
            sla_rules=resolved_sla_rules,
        ),
        response_format=UrgencyAnalysis,
    )

nlp

Natural language processing task factories.

Functions
multilingual_translation 
multilingual_translation() -> PreparedTask[
    TranslatedString
]

Create a multilingual translation task.

Source code in src/openaivec/task/nlp/translation.py 
 99
100
101
102
103
104
def multilingual_translation() -> PreparedTask[TranslatedString]:
    """Create a multilingual translation task."""
    return PreparedTask(
        instructions=_build_instructions(),
        response_format=TranslatedString,
    )
Modules
dependency_parsing

Dependency parsing task definition.

Classes
DependencyRelation

Bases: BaseModel

Single dependency edge.

DependencyParsing

Bases: BaseModel

Dependency parsing output.

Functions
dependency_parsing 
dependency_parsing() -> PreparedTask[DependencyParsing]

Create a dependency parsing task.

Source code in src/openaivec/task/nlp/dependency_parsing.py 
43
44
45
46
47
48
def dependency_parsing() -> PreparedTask[DependencyParsing]:
    """Create a dependency parsing task."""
    return PreparedTask(
        instructions=_build_instructions(),
        response_format=DependencyParsing,
    )
keyword_extraction

Keyword extraction task definition.

Classes
Keyword

Bases: BaseModel

Single keyword or keyphrase entry.

KeywordExtraction

Bases: BaseModel

Keyword extraction output.

Functions
keyword_extraction 
keyword_extraction() -> PreparedTask[KeywordExtraction]

Create a keyword extraction task.

Source code in src/openaivec/task/nlp/keyword_extraction.py 
42
43
44
45
46
47
def keyword_extraction() -> PreparedTask[KeywordExtraction]:
    """Create a keyword extraction task."""
    return PreparedTask(
        instructions=_build_instructions(),
        response_format=KeywordExtraction,
    )
morphological_analysis

Morphological analysis task definition.

Classes
MorphologicalAnalysis

Bases: BaseModel

Morphological analysis output.

Functions
morphological_analysis 
morphological_analysis() -> PreparedTask[
    MorphologicalAnalysis
]

Create a morphological analysis task.

Source code in src/openaivec/task/nlp/morphological_analysis.py 
33
34
35
36
37
38
def morphological_analysis() -> PreparedTask[MorphologicalAnalysis]:
    """Create a morphological analysis task."""
    return PreparedTask(
        instructions=_build_instructions(),
        response_format=MorphologicalAnalysis,
    )
named_entity_recognition

Named entity recognition task definition.

Classes
NamedEntity

Bases: BaseModel

Single named-entity span.

NamedEntityRecognition

Bases: BaseModel

Named entity recognition output.

Functions
named_entity_recognition 
named_entity_recognition() -> PreparedTask[
    NamedEntityRecognition
]

Create a named entity recognition task.

Source code in src/openaivec/task/nlp/named_entity_recognition.py 
47
48
49
50
51
52
def named_entity_recognition() -> PreparedTask[NamedEntityRecognition]:
    """Create a named entity recognition task."""
    return PreparedTask(
        instructions=_build_instructions(),
        response_format=NamedEntityRecognition,
    )
sentiment_analysis

Sentiment analysis task definition.

Classes
SentimentAnalysis

Bases: BaseModel

Sentiment analysis output.

Functions
sentiment_analysis 
sentiment_analysis() -> PreparedTask[SentimentAnalysis]

Create a sentiment analysis task.

Source code in src/openaivec/task/nlp/sentiment_analysis.py 
42
43
44
45
46
47
def sentiment_analysis() -> PreparedTask[SentimentAnalysis]:
    """Create a sentiment analysis task."""
    return PreparedTask(
        instructions=_build_instructions(),
        response_format=SentimentAnalysis,
    )
translation

Multilingual translation task definition.

Classes
TranslatedString

Bases: BaseModel

Translations for a fixed set of language-code fields.

Functions
multilingual_translation 
multilingual_translation() -> PreparedTask[
    TranslatedString
]

Create a multilingual translation task.

Source code in src/openaivec/task/nlp/translation.py 
 99
100
101
102
103
104
def multilingual_translation() -> PreparedTask[TranslatedString]:
    """Create a multilingual translation task."""
    return PreparedTask(
        instructions=_build_instructions(),
        response_format=TranslatedString,
    )

table

Classes
FillNaResponse

Bases: BaseModel

Response model for missing value imputation results.

Contains the row index and the imputed value for a specific missing entry in the target column.

Modules
fillna

Missing value imputation task for DataFrame columns.

This module provides functionality to intelligently fill missing values in DataFrame columns using AI-powered analysis. The task analyzes existing data patterns to generate contextually appropriate values for missing entries.

Example

Basic usage with pandas DataFrame:

import pandas as pd
from openaivec import pandas_ext  # Required for .ai accessor
from openaivec.task.table import fillna

# Create DataFrame with missing values
df = pd.DataFrame({
    "name": ["Alice", "Bob", None, "David"],
    "age": [25, 30, 35, None],
    "city": ["New York", "London", "Tokyo", "Paris"],
    "salary": [50000, 60000, 70000, None]
})

# Fill missing values in the 'salary' column
task = fillna(df, "salary")
filled_salaries = df[df["salary"].isna()].ai.task(task)

# Apply filled values back to DataFrame
for result in filled_salaries:
    df.loc[result.index, "salary"] = result.output

With BatchResponses for more control:

from openai import OpenAI
from openaivec import BatchResponses
from openaivec.task.table import fillna

client = OpenAI()
df = pd.DataFrame({...})  # Your DataFrame with missing values

# Create fillna task for target column
task = fillna(df, "target_column")

# Get rows with missing values in target column
missing_rows = df[df["target_column"].isna()]

# Process with BatchResponses
filler = BatchResponses.of_task(
    client=client,
    model_name="gpt-4.1-mini",
    task=task
)

# Generate inputs for missing rows
inputs = []
for idx, row in missing_rows.iterrows():
    inputs.append({
        "index": idx,
        "input": {k: v for k, v in row.items() if k != "target_column"}
    })

filled_values = filler.parse(inputs)
Classes
FillNaResponse

Bases: BaseModel

Response model for missing value imputation results.

Contains the row index and the imputed value for a specific missing entry in the target column.

Functions
fillna 
fillna(
    df: DataFrame,
    target_column_name: str,
    max_examples: int = 500,
) -> PreparedTask[FillNaResponse]

Create a prepared task for filling missing values in a DataFrame column.

Analyzes the provided DataFrame to understand data patterns and creates a configured task that can intelligently fill missing values in the specified target column. The task uses few-shot learning with examples extracted from non-null rows in the DataFrame.

Parameters:

Name Type Description Default
df DataFrame

Source DataFrame containing the data with missing values.

 required
target_column_name str

Name of the column to fill missing values for. This column should exist in the DataFrame and contain some non-null values to serve as training examples.

 required
max_examples int

Maximum number of example rows to use for few-shot learning. Defaults to 500. Higher values provide more context but increase token usage and processing time.

 500

Returns:

Type Description
PreparedTask[FillNaResponse]

PreparedTask configured for missing value imputation with:
PreparedTask[FillNaResponse]
  • Instructions based on DataFrame patterns
PreparedTask[FillNaResponse]
  • FillNaResponse format for structured output
PreparedTask[FillNaResponse]
  • No embedded API parameter defaults

Raises:

Type Description
ValueError

If target_column_name doesn't exist in DataFrame, contains no non-null values for training examples, DataFrame is empty, or max_examples is not a positive integer.
Example 
import pandas as pd
from openaivec.task.table import fillna

df = pd.DataFrame({
    "product": ["laptop", "phone", "tablet", "laptop"],
    "brand": ["Apple", "Samsung", None, "Dell"],
    "price": [1200, 800, 600, 1000]
})

# Create task to fill missing brand values
task = fillna(df, "brand")

# Use with pandas AI accessor
missing_brands = df[df["brand"].isna()].ai.task(task)
Source code in src/openaivec/task/table/fillna.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def fillna(df: pd.DataFrame, target_column_name: str, max_examples: int = 500) -> PreparedTask[FillNaResponse]:
    """Create a prepared task for filling missing values in a DataFrame column.

    Analyzes the provided DataFrame to understand data patterns and creates
    a configured task that can intelligently fill missing values in the
    specified target column. The task uses few-shot learning with examples
    extracted from non-null rows in the DataFrame.

    Args:
        df (pd.DataFrame): Source DataFrame containing the data with missing values.
        target_column_name (str): Name of the column to fill missing values for.
            This column should exist in the DataFrame and contain some
            non-null values to serve as training examples.
        max_examples (int): Maximum number of example rows to use for few-shot
            learning. Defaults to 500. Higher values provide more context
            but increase token usage and processing time.

    Returns:
        PreparedTask configured for missing value imputation with:
        - Instructions based on DataFrame patterns
        - FillNaResponse format for structured output
        - No embedded API parameter defaults

    Raises:
        ValueError: If target_column_name doesn't exist in DataFrame,
            contains no non-null values for training examples, DataFrame is empty,
            or max_examples is not a positive integer.

    Example:
        ```python
        import pandas as pd
        from openaivec.task.table import fillna

        df = pd.DataFrame({
            "product": ["laptop", "phone", "tablet", "laptop"],
            "brand": ["Apple", "Samsung", None, "Dell"],
            "price": [1200, 800, 600, 1000]
        })

        # Create task to fill missing brand values
        task = fillna(df, "brand")

        # Use with pandas AI accessor
        missing_brands = df[df["brand"].isna()].ai.task(task)
        ```
    """
    if df.empty:
        raise ValueError("DataFrame is empty.")
    if not isinstance(max_examples, int) or max_examples <= 0:
        raise ValueError("max_examples must be a positive integer.")
    if target_column_name not in df.columns:
        raise ValueError(f"Column '{target_column_name}' does not exist in the DataFrame.")
    if df[target_column_name].notna().sum() == 0:
        raise ValueError(f"Column '{target_column_name}' contains no non-null values for training examples.")
    instructions = _build_instructions(df, target_column_name, max_examples)
    return PreparedTask(instructions=instructions, response_format=FillNaResponse)