ractogateway.pipelines.list_classifier.pipeline

ListClassifierPipeline — map a natural-language query to items from a list.

Two classes are exported:

ListClassifierPipeline — run() is sync, arun() is async.
AsyncListClassifierPipeline — run() is async only.

The pipeline takes a Python list[str] of candidate options and a user query, then asks the configured LLM to pick the best matching option(s). Internally it builds a dynamic Python enum.Enum from the options list and validates every LLM response against it — hallucinated or paraphrased answers are automatically caught, fuzzy-corrected if possible, or retried.

Provider / model selection

Pass any RactoGateway kit directly:

from ractogateway.openai_developer_kit import Chat as OpenAIChat
from ractogateway.anthropic_developer_kit import Chat as ClaudeChat
from ractogateway.google_developer_kit import Chat as GeminiChat
from ractogateway.ollama_developer_kit import Chat as OllamaChat
from ractogateway.huggingface_developer_kit import Chat as HFChat

pipeline = ListClassifierPipeline(kit=OpenAIChat(model="gpt-4o-mini"), ...)

Or use the convenience factory — no separate import needed:

pipeline = ListClassifierPipeline.from_provider(
    provider="anthropic",
    model="claude-haiku-4-5-20251001",
    options=["Billing", "Support", "Sales"],
)

# Supported providers: "openai", "anthropic", "google", "ollama", "huggingface"

Selection modes

"single" — LLM must return exactly one option (default).
"multiple" — LLM may return one or more options from the list.

Output formats

"pydantic" — returns a ClassifierResult (default).
"string" — returns a comma-joined string (e.g. "Billing, Account").
"dict" — returns {"selected": [...], "confidences": [...], ...}.

Production features

Dynamic Enum validation — options → Python Enum per-call; invalid LLM responses auto-retried (up to max_retries, default 2).
Fuzzy fallback — stdlib difflib fuzzy-matches near-misses before consuming a retry (fuzzy_fallback=True, default).
Option descriptions — per-option natural-language descriptions help the LLM distinguish similar-sounding categories.
Score-all mode — score_all=True asks the LLM for a confidence score for every option, not just the selected ones; stored in result.all_scores.
Uncertain label — uncertain_label="Other" injects a catch-all option so the LLM has somewhere to fall when nothing matches.
Batch classification — batch_run(queries) / abatch_run(queries) run multiple queries; async version runs them concurrently.
Runtime option management — add_option(), remove_option(), set_options(), get_options() mutate the pipeline’s default list.
Safe mode — exceptions → result.error field instead of raising.
Telemetry — tracer=RactoTracer(...) (OTEL) + metrics=GatewayMetricsMiddleware(...).
Rate limiting — duck-typed rate_limiter=.
Conversation memory — duck-typed memory= + per-call session_id.
Case-insensitive matching (default).

class ractogateway.pipelines.list_classifier.pipeline.ListClassifierPipeline(kit, *, options=None, selection_mode='single', output_format='pydantic', prompt=None, temperature=0.0, max_tokens=512, max_retries=2, include_confidence=True, include_reasoning=False, score_all=False, option_descriptions=None, fuzzy_fallback=True, uncertain_label=None, confidence_threshold=None, case_sensitive=False, safe_mode=False, exact_cache=None, semantic_cache=None, audit_logger=None, tracer=None, metrics=None, rate_limiter=None, memory=None, user_id=None)[source]

Bases: object

Map a natural-language query to one or more items from a candidate list.

Supports every RactoGateway provider via the kit parameter or the from_provider() class factory. Internally builds a dynamic Python enum.Enum from the options list and validates every LLM response against it — hallucinations and paraphrased answers are caught, fuzzy- corrected if close enough, and retried otherwise.

Two variants

ListClassifierPipeline — run() sync, arun() async.
AsyncListClassifierPipeline — run() is async only.

type kit:: Any
param kit:: Any RactoGateway developer kit (OpenAI, Anthropic, Google, Ollama, HuggingFace). Must expose .chat(ChatConfig) and .achat(ChatConfig) methods. Use from_provider() instead of constructing kits manually when you only need provider + model.
type options:: list[str] | None
param options:: Default candidate strings. Can be overridden per-call. Must be non-empty and duplicate-free when provided.
type selection_mode:: Literal['single', 'multiple']
param selection_mode:: "single" (default) — exactly one option. "multiple" — one or more options. Overridable per-call.
type output_format:: Literal['string', 'dict', 'pydantic']
param output_format:: "pydantic" (default) — ClassifierResult. "string" — comma-joined string. "dict" — plain dict. Overridable per-call.
type prompt:: RactoPrompt | None
param prompt:: Custom RactoPrompt to replace the built-in system prompt.
type temperature:: float
param temperature:: LLM temperature. Default 0.0 for deterministic output.
type max_tokens:: int
param max_tokens:: Response token budget. Default 512.
type max_retries:: int
param max_retries:: Retry attempts when LLM returns invalid JSON / unknown option. Default 2.
type include_confidence:: bool
param include_confidence:: Ask LLM for per-selection confidence scores [0.0–1.0]. Default True.
type include_reasoning:: bool
param include_reasoning:: Ask LLM for a one-sentence explanation. Default False.
type score_all:: bool
param score_all:: Ask LLM for a score for every option (not just selected ones). Stored in result.all_scores. Default False.
type option_descriptions:: dict[str, str] | None
param option_descriptions:: {option: description} — shown inline next to each option in the prompt to help the LLM distinguish similar categories.
type fuzzy_fallback:: bool
param fuzzy_fallback:: Use stdlib difflib to correct near-miss LLM responses before consuming a retry. Default True.
type uncertain_label:: str | None
param uncertain_label:: When set, this string is appended as an extra option that the LLM can pick when nothing matches (e.g. "Other / None of the above"). result.uncertain is True when this label is selected.
type confidence_threshold:: float | None
param confidence_threshold:: Drop selections below this score. Keeps highest-confidence match as fallback. Default None (no filtering).
type case_sensitive:: bool
param case_sensitive:: Whether option matching is case-sensitive. Default False.
type safe_mode:: bool
param safe_mode:: Return ClassifierResult(error=...) instead of raising. Default False.
type tracer:: Any | None
param tracer:: Optional RactoTracer.
type metrics:: Any | None
param metrics:: Optional GatewayMetricsMiddleware.
type rate_limiter:: Any | None
param rate_limiter:: Duck-typed — check_and_consume(user_id, tokens) -> bool + get_remaining(user_id) -> int.
type memory:: Any | None
param memory:: Duck-typed — get_history(session_id) -> list[dict] + append(session_id, role, content).
type user_id:: str | None
param user_id:: Default user ID for rate limiting. Overridable per-call.

Example

# Via kit directly
from ractogateway.openai_developer_kit import Chat
from ractogateway.pipelines import ListClassifierPipeline

pipeline = ListClassifierPipeline(
    kit=Chat(model="gpt-4o-mini"),
    options=["Billing", "Technical Support", "Sales"],
    include_confidence=True,
    include_reasoning=True,
)
result = pipeline.run("My invoice is wrong")
print(result.first, result.top_confidence)

# Via from_provider() — no manual kit import needed
pipeline = ListClassifierPipeline.from_provider(
    "anthropic", "claude-haiku-4-5-20251001",
    options=["Billing", "Technical Support", "Sales"],
)

classmethod from_provider(provider, model, *, api_key=None, base_url=None, options=None, **kwargs)[source]

Create a pipeline by specifying provider + model — no kit import needed.

Parameters:

provider (str) – One of "openai", "anthropic", "google", "ollama", "huggingface".
model (str) –
Model identifier string, e.g.:
- OpenAI: "gpt-4o-mini", "gpt-4o"
- Anthropic: "claude-haiku-4-5-20251001", "claude-sonnet-4-6"
- Google: "gemini-2.0-flash", "gemini-1.5-pro"
- Ollama: "llama3.2", "mistral"
- HuggingFace: "meta-llama/Llama-3.2-3B-Instruct"
api_key (str | None) – Provider API key. Falls back to the standard env var for each provider (e.g. OPENAI_API_KEY, ANTHROPIC_API_KEY).
base_url (str | None) – Custom endpoint — used for Ollama (http://localhost:11434) or OpenAI-compatible proxies.
options (list[str] | None) – Default candidate options list.
**kwargs (Any) – Any other ListClassifierPipeline constructor parameters (selection_mode, include_confidence, safe_mode, etc.).

Return type:

ListClassifierPipeline

Returns:

ListClassifierPipeline

Example

pipeline = ListClassifierPipeline.from_provider(
    "anthropic",
    "claude-haiku-4-5-20251001",
    options=["Billing", "Support", "Sales"],
    include_reasoning=True,
    safe_mode=True,
)

static make_enum(options, name='OptionsEnum')[source]

Build a standalone dynamic enum.Enum from an options list.

Useful when you want enum-typed values outside the pipeline.

Parameters:

options (list[str]) – List of option strings.
name (str) – Enum class name. Default "OptionsEnum".

Return type:

type[Enum]

Returns:

type[Enum]

Example

E = ListClassifierPipeline.make_enum(["Red", "Green", "Blue"])
E["Red"].value   # "Red"

get_options()[source]

Return the pipeline-level options list, or None if not set.

Return type:: list[str] | None

set_options(options)[source]

Replace the entire pipeline-level options list.

Thread-safe — safe to call while the pipeline is in use.

Parameters:: options (list[str]) – New options list. Must be non-empty and duplicate-free.
Return type:: None

add_option(option, description=None)[source]

Append a new option to the pipeline-level list.

Parameters:

option (str) – The option string to add.
description (str | None) – Optional inline description for the option.

Return type:

None

remove_option(option)[source]

Remove an option from the pipeline-level list.

Parameters:: option (str) – The option string to remove. Raises ValueError if not found.
Return type:: None

run(user_query, *, options=<object object>, selection_mode=None, output_format=None, temperature=None, max_tokens=None, confidence_threshold=<object object>, session_id=None, user_id=None)[source]

Classify user_query synchronously.

Parameters:

user_query (str) – Natural-language query to classify.
options (list[str] | None) – Per-call override for the candidate list. Omit to use the pipeline-level list. Pass [] to get a ValueError.
selection_mode (Literal['single', 'multiple'] | None) – Per-call override — "single" or "multiple".
output_format (Literal['string', 'dict', 'pydantic'] | None) – Per-call override — "pydantic", "string", or "dict".
max_tokens (int | None) – Per-call LLM setting overrides.
confidence_threshold (float | None) – Per-call override. Pass None explicitly to disable filtering for this call even if a pipeline-level threshold is set.
session_id (str | None) – Conversation session ID for memory retrieval/storage.
user_id (str | None) – Per-call user ID for rate limiting and audit.

Return type:

ClassifierResult | str | dict[str, Any]

Returns:

ClassifierResult | str | dict – Type depends on output_format.

batch_run(queries, *, options=<object object>, selection_mode=None, output_format=None, temperature=None, max_tokens=None, confidence_threshold=<object object>, session_id=None, user_id=None)[source]

Classify multiple queries synchronously, one after another.

Shares all per-call overrides across every query in the batch. Use abatch_run() to run them concurrently in async contexts.

Parameters:: queries (list[str]) – List of natural-language queries to classify.
Return type:: list[ClassifierResult | str | dict[str, Any]]
Returns:: list – One result per query, in the same order.

async arun(user_query, *, options=<object object>, selection_mode=None, output_format=None, temperature=None, max_tokens=None, confidence_threshold=<object object>, session_id=None, user_id=None)[source]

Async variant of run() — identical parameters.

Return type:: ClassifierResult | str | dict[str, Any]

async abatch_run(queries, *, options=<object object>, selection_mode=None, output_format=None, temperature=None, max_tokens=None, confidence_threshold=<object object>, session_id=None, user_id=None, max_concurrency=None)[source]

Classify multiple queries concurrently with asyncio.gather.

Parameters:

queries (list[str]) – List of natural-language queries.
max_concurrency (int | None) – Cap the number of simultaneous LLM calls. None (default) runs all queries in parallel. Set to e.g. 5 to avoid rate-limit errors on large batches.

Return type:

list[ClassifierResult | str | dict[str, Any]]

Returns:

list – Results in the same order as queries.

class ractogateway.pipelines.list_classifier.pipeline.AsyncListClassifierPipeline(kit, *, options=None, selection_mode='single', output_format='pydantic', prompt=None, temperature=0.0, max_tokens=512, max_retries=2, include_confidence=True, include_reasoning=False, score_all=False, option_descriptions=None, fuzzy_fallback=True, uncertain_label=None, confidence_threshold=None, case_sensitive=False, safe_mode=False, exact_cache=None, semantic_cache=None, audit_logger=None, tracer=None, metrics=None, rate_limiter=None, memory=None, user_id=None)[source]

Bases: ListClassifierPipeline

Async-first variant of ListClassifierPipeline.

run() is a coroutine — await pipeline.run(...) directly. Designed for FastAPI, aiohttp, Starlette, and other async frameworks.

Constructor and all run() parameters are identical to ListClassifierPipeline.

Example

pipeline = AsyncListClassifierPipeline.from_provider(
    "openai", "gpt-4o-mini",
    options=["Billing", "Support", "Sales"],
    safe_mode=True,
)

# FastAPI handler:
@app.post("/classify")
async def classify(query: str):
    result = await pipeline.run(query)
    return result.as_dict()

async run(user_query, **kwargs)[source]

Async run() — delegates to ListClassifierPipeline.arun().

Return type:: ClassifierResult | str | dict[str, Any]