Smelt AI

LLM-powered structured data transformation. Feed in rows of data, get back strictly typed Pydantic models or free-text responses — batched, concurrent, and validated.

from smelt import Model, Job
from pydantic import BaseModel

class Classification(BaseModel):
    sector: str
    sub_sector: str
    is_public: bool

model = Model(provider="openai", name="gpt-4.1-mini")
job = Job(
    prompt="Classify each company by industry sector and whether it's publicly traded.",
    output_model=Classification,
)

result = job.run(model, data=[
    {"name": "Apple", "desc": "Consumer electronics and software"},
    {"name": "Stripe", "desc": "Payment processing platform"},
    {"name": "Mayo Clinic", "desc": "Nonprofit medical center"},
])

for row in result.data:
    print(row)  # Classification(sector='Technology', sub_sector='Consumer Electronics', is_public=True)

Free-text mode — skip the schema, get plain text back:

job = Job(prompt="Write a one-paragraph summary for each company")

result = job.run(model, data=[
    {"name": "Apple", "desc": "Consumer electronics and software"},
    {"name": "Stripe", "desc": "Payment processing platform"},
])

for text in result.data:
    print(text)  # "Apple is a multinational technology company..."

Install

pip install smelt-ai[openai]      # OpenAI models
pip install smelt-ai[anthropic]   # Anthropic models
pip install smelt-ai[google]      # Google Gemini models
pip install smelt-ai[vision]      # Image support (Pillow)

Combine extras: pip install smelt-ai[openai,vision]

Requires Python 3.10+.

How It Works

list[dict] → Tag with row_id → Split into batches → Concurrent LLM calls → Validate → Reorder → SmeltResult[T]

1. Each input row gets a row_id for tracking
2. Rows are split into batches of configurable size
3. Batches run concurrently through the LLM with structured output
4. Each response is validated (schema, row IDs, count)
5. Results are reordered to match original input order
6. Everything is returned as a typed SmeltResult with metrics

Vision (Image Support)

Pass PIL images directly in your data dicts — smelt auto-detects them, base64-encodes, and sends multimodal content blocks to vision-capable LLMs.

pip install smelt-ai[anthropic,vision]

from PIL import Image
from pydantic import BaseModel, Field
from smelt import Model, Job

class ECGAnalysis(BaseModel):
    heart_rhythm: str = Field(description="Detected heart rhythm")
    heart_rate_bpm: int = Field(description="Estimated heart rate in bpm")
    abnormalities: list[str] = Field(description="List of detected abnormalities")

model = Model(provider="anthropic", name="claude-sonnet-4-6")
job = Job(
    prompt="Analyze the ECG image and provide a structured cardiac assessment.",
    output_model=ECGAnalysis,
    batch_size=1,
)

result = job.run(model, data=[
    {"patient_id": "P001", "ecg": Image.open("ecg_1.jpeg")},
])

print(result.data[0])
# ECGAnalysis(heart_rhythm='Sinus Tachycardia', heart_rate_bpm=120, abnormalities=[...])

Works with any vision-capable model — OpenAI GPT-4o, Anthropic Claude, Google Gemini, etc. Use batch_size=1 for image-heavy payloads.

Aggregate (Many-to-One)

Reduce an entire dataset to a single output. Two strategies:

from pydantic import BaseModel, Field
from smelt import AggregateJob, Model

class PortfolioSummary(BaseModel):
    total_companies: int
    sectors: list[str]
    total_revenue_millions: float
    top_5_by_revenue: list[str]

model = Model(provider="anthropic", name="claude-sonnet-4-6")

# Tree (default) — parallel map then pairwise merge, faster
job = AggregateJob(
    prompt="Analyze this portfolio. Count totals, list all sectors, sum revenues.",
    output_model=PortfolioSummary,
    strategy="tree",
    batch_size=15,
)

# Sequential — each step sees accumulated output + next batch, higher quality
job = AggregateJob(
    prompt="Analyze this portfolio. Count totals, list all sectors, sum revenues.",
    output_model=PortfolioSummary,
    strategy="sequential",
    batch_size=15,
)

result = job.run(model, data=companies)  # 60 companies → 1 summary
print(result.data[0])
# PortfolioSummary(total_companies=60, sectors=['Technology', 'Finance', ...], ...)

Design your schema with additive fields (lists, counts) for best results.

With stop_on_exhaustion=False, aggregation degrades gracefully: batches that fail after all retries are skipped and the aggregate is computed from the surviving rows. Check result.errors for the excluded batches and result.metrics.failed_rows for the excluded row count.

API

Model

Wraps a LangChain chat model provider. Any LangChain-supported provider works.

model = Model(
    provider="openai",          # LangChain provider name
    name="gpt-4.1-mini",       # Model identifier
    api_key="sk-...",           # Optional — falls back to env var (e.g. OPENAI_API_KEY)
    params={"temperature": 0},  # Forwarded to the chat model constructor
)

Job

Defines what transformation to run and how to batch it.

job = Job(
    prompt="Your transformation instructions here",
    output_model=MyPydanticModel,  # Schema for each output row (None for free-text)
    batch_size=10,                 # Rows per LLM request (default: 10)
    concurrency=3,                 # Max concurrent requests (default: 3)
    max_retries=3,                 # Retries per failed batch (default: 3)
    shuffle=False,                 # Shuffle rows before batching (default: False)
    stop_on_exhaustion=True,       # Raise on failure vs collect errors (default: True)
)

Run:

result = job.run(model, data=rows)          # Sync
result = await job.arun(model, data=rows)   # Async

Test with a single row first:

result = job.test(model, data=rows)         # Sync — runs only the first row
result = await job.atest(model, data=rows)  # Async

AggregateJob

Defines a many-to-one aggregation.

job = AggregateJob(
    prompt="Your aggregation instructions here",
    output_model=MySummaryModel,       # Schema for the aggregate output (None for free-text)
    strategy="tree",                   # "tree" (parallel) or "sequential" (fold)
    batch_size=50,                     # Rows per step (default: 50)
    concurrency=3,                     # Max concurrent steps, tree only (default: 3)
    max_retries=3,                     # Retries per failed step (default: 3)
    stop_on_exhaustion=True,           # Raise on failure vs collect errors (default: True)
)

result = job.run(model, data=rows)          # Sync
result = await job.arun(model, data=rows)   # Async
result.data[0]                              # Single aggregate output

SmeltResult[T]

result.data       # list[T] — transformed rows in original order
result.errors     # list[BatchError] — failed batches
result.metrics    # SmeltMetrics — tokens, timing, retries
result.success    # bool — True if no errors

Error Handling

All exceptions inherit from SmeltError.

from smelt.errors import SmeltExhaustionError

try:
    result = job.run(model, data=rows)
except SmeltExhaustionError as e:
    print(f"Partial: {len(e.partial_result.data)} rows succeeded")

Or collect errors without raising:

job = Job(prompt="...", output_model=MyModel, stop_on_exhaustion=False)
result = job.run(model, data=rows)

if not result.success:
    for err in result.errors:
        print(f"Batch {err.batch_index} failed: {err.message}")

Supported Providers

Links

• Documentation
• PyPI
• Issue Tracker

License

MIT