Polar Llama

import polars as pl
from polar_llama import typesafe_eval, noul, choice, score

df = pl.DataFrame({"message": [
    "Help! My payouts have been failing for 3 days.",
    "Hi, just wondering what your enterprise pricing looks like.",
]})

out = df.with_columns(
    ts=typesafe_eval(
        pl.col("message"),
        questions={
            "is_urgent": noul(
                "Does this convey urgency?",
                true="Explicitly time-sensitive",
                false="No urgency expressed",
            ),
            "department": choice(
                "Which team should handle this?",
                {
                    "billing": "Payments, invoicing, refunds",
                    "technical": "Bugs, outages, integrations",
                    "sales": "Pricing, upgrades, new accounts",
                },
            ),
            "frustration": score(
                "How frustrated is the customer?",
                ["Calm", "Frustrated", "Very angry"],
            ),
        },
    )
).unnest("ts")

# message              is_urgent  department  department_confidence  frustration  _error
# "Help! My payouts…"  0.95       billing     0.79                   1.04         null
# "Hi, just wonderi…"  0.06       sales       1.0                    0.0          null

# Confidence is the lever: act automatically, or route to a human.
auto   = out.filter(pl.col("department_confidence") > 0.9)
review = out.filter(pl.col("department_confidence").is_between(0.5, 0.9))
human  = out.filter(pl.col("department_confidence") < 0.5)

typesafe_eval(
    pl.col("message"), pl.col("order_id"), pl.col("charge_count"),
    questions={"duplicate": noul("Do the records show a duplicate charge?")},
)
# state = {"message": "...", "order_id": "A-104", "charge_count": 2}

typesafe_eval(
    pl.col("message"),
    pl.lit(refund_policy).alias("policy"),   # broadcast to every row
    questions={"refund_ok": noul("Does the policy support a refund here?")},
)

from typing import Literal
from pydantic import BaseModel, Field
from polar_llama import typesafe_eval, score_field

class ClauseFeatures(BaseModel):
    is_payment: bool = Field(description="Does this clause create a payment obligation?")
    category: Literal["fees", "term", "liability", "other"] = Field(
        description="What kind of clause is this?")
    severity: float = score_field(
        "How onerous is this clause for the Customer?",
        ["Benign", "Notable", "Onerous"])

df.with_columns(ts=typesafe_eval(pl.col("clause"), contract=ClauseFeatures)).unnest("ts")

from polar_llama import typesafe_models

[m["name"] for m in typesafe_models()]   # e.g. ['jev-latest', 'jev-preview']

from polar_llama import (
    mcp_tools, tools_to_response_model, execute_tool_calls,
    tool_results_to_message, combine_messages, inference_messages, Provider,
)

tools = mcp_tools("http://localhost:8811/mcp")     # tools/list introspection
ToolCalls = tools_to_response_model(tools)          # emission schema

df = (
    df
    # 1. Emit: the LLM parameterizes N calls per row (structured output)
    .with_columns(calls=pl.col("meal").llama.inference_async(
        provider=Provider.OPENAI, model="gpt-4o-mini", response_model=ToolCalls))
    # 2. Execute: all calls across all rows, in parallel, errors as data
    .with_columns(results=execute_tool_calls(
        pl.col("calls"), transport="http://localhost:8811/mcp", tools=tools))
    # 3. Synthesize: fold results back through a second inference pass
    .with_columns(answer=inference_messages(
        combine_messages(
            pl.col("meal").llama.to_message(role="user"),
            tool_results_to_message(pl.col("results")),
        ),
        provider=Provider.OPENAI, model="gpt-4o-mini"))
)

import polars as pl
from polar_llama import Predict, Signature, BootstrapFewShot, InstructionOptimizer, evaluate

# 1. Declare the task
module = Predict(
    Signature("question -> answer", instructions="Answer concisely."),
    provider="openai",
    model="gpt-4o-mini",
)

# 2. Labeled training data
trainset = pl.DataFrame({
    "question": ["What is 2+2?", "Capital of France?", "Largest planet?"],
    "answer": ["4", "Paris", "Jupiter"],
})

# 3. A metric: (gold row, prediction) -> bool | float
def exact_match(example, prediction):
    return example["answer"].strip().lower() == (prediction["answer"] or "").strip().lower()

# 4a. Bootstrap few-shot demos from rows the model already gets right
compiled = BootstrapFewShot(metric=exact_match, max_demos=4).compile(module, trainset)

# 4b. Or search for better instructions (COPRO-style)
optimizer = InstructionOptimizer(metric=exact_match, n_candidates=4)
compiled = optimizer.compile(module, trainset)
print(optimizer.history)  # [(instructions, score), ...]

# 5. Run the optimized module on new data; outputs land in pred_* columns
result = compiled(pl.DataFrame({"question": ["What is 3+3?"]}))
print(result["pred_answer"])

# Score any module against a labeled set
print(evaluate(compiled, trainset, exact_match).score)

import polars as pl
from polar_llama import (
    inference_async, Provider, inference_stream, ResponseCache, DedupeStats, build_manifest, with_manifest_id,
)

# Streaming (0.6.0+): token deltas via a callback, Struct{text, finished} per row
df = df.with_columns(
    streamed=inference_stream(pl.col("prompt"), provider=Provider.OPENAI,
                              model="gpt-4o-mini", on_token=lambda i, d: print(d, end=""))
)

# Checkpointing (0.6.1+): re-run after a crash and only unfinished rows hit the API
df = df.with_columns(
    answer=inference_async(pl.col("prompt"), provider=Provider.OPENAI,
                           model="gpt-4o-mini", checkpoint="run1.ckpt")
)

# Usage & cost (0.6.2+): Struct{response, usage{input_tokens, ..., cost_usd}}
df = df.with_columns(
    r=inference_async(pl.col("prompt"), provider=Provider.OPENAI,
                      model="gpt-4o-mini", usage=True)
)
spend = df.select(pl.col("r").struct.field("usage").struct.field("cost_usd").sum())

# Dedupe + persistent response cache (0.6.3+): pay once per unique request
stats = DedupeStats()
df = df.with_columns(
    cached=inference_async(pl.col("prompt"), model="gpt-4o-mini",
                           response_cache=ResponseCache("responses.cache", ttl="24h"),
                           dedupe_stats=stats)
)
print(stats.hit_rate, stats.calls_made)

# Run manifests (0.8.3+): a deterministic, integrity-checked record of the run
manifest = build_manifest(df, symbol="inference_async", provider="openai",
                          model="gpt-4o-mini")
manifest.save("runs/nightly.manifest.json")
df = with_manifest_id(df, manifest)

import polars as pl
from polar_llama import (
    Provider, induce_codebook, apply_codebook, cohens_kappa, krippendorffs_alpha, QualityConfig, quality_report, export_review_sample, import_corrections,
)

# Codebook induction (0.7.0+): discover themes, then apply them as multi-label codes
result = induce_codebook(responses, "text", provider=Provider.OPENAI, model="gpt-4o-mini")
coded = result.df.with_columns(
    labels=apply_codebook(pl.col("text"), result.codebook, provider=Provider.OPENAI)
)

# Agreement (0.7.1+): LLM vs. human and across a rater panel, per group
labels.group_by("batch").agg(
    kappa=cohens_kappa("llm_label", "human_label"),
    alpha=krippendorffs_alpha(["rater_1", "rater_2", "rater_3"]),
)

# Survey quality flags (0.7.2+): graded scores, never drops a row
report = quality_report(survey_df, QualityConfig(
    id_column="respondent_id", grid_columns=["g1", "g2", "g3"],
    text_columns=["oe1"], duration_column="duration_s",
))
report.summary

# Human review loop (0.7.3+): sample for review, then score the corrections
export_review_sample(coded_df, n=100, strata="code", path="review_batch.csv")
review = import_corrections(coded_df, reviewed_df, code_column="code")
print(review.kappa, review.n_changed)

import polars as pl
import polar_llama  # registers the .llama namespace

df = pl.DataFrame({"ticket": [
    "My laptop won't turn on even when it's plugged in.",
    "I forgot my password and I'm locked out of my account.",
]})

tagged = df.with_columns(
    tag=pl.col("ticket").llama.inference_local(
        model="mlx-community/gemma-3n-E4B-it-lm-4bit",
        system="Classify the ticket as exactly one of: Hardware, Account, Billing.",
        engine="in_process",   # on-device mlx-lm, batched over the column
        max_tokens=16,
        temperature=0.0,
    )
)

import polars as pl
from polar_llama import string_to_message, inference_async, Provider

# Load customer feedback data
feedback_df = pl.DataFrame({
    'customer_id': [101, 102, 103, 104, 105],
    'feedback': [
        'The product is amazing but shipping was slow',
        'Great quality, highly recommend!',
        'Disappointed with customer service',
        'Perfect for my needs, will buy again',
        'Product arrived damaged, requesting refund'
    ]
})

# Create sentiment analysis prompts
sentiment_prompt = """Analyze the sentiment of this customer feedback
and classify it as Positive, Negative, or Neutral.
Also provide a brief reason.

Feedback: {feedback}"""

df = feedback_df.with_columns(
    prompt=pl.format(sentiment_prompt, pl.col('feedback'))
)

# Convert to messages and run inference
df = df.with_columns(
    message=string_to_message("prompt", message_type='user')
)

df = df.with_columns(
    sentiment_analysis=inference_async('message',
                                      provider=Provider.OPENAI,
                                      model='gpt-4o-mini')
)

# Extract key insights
print(df.select(['customer_id', 'feedback', 'sentiment_analysis']))

import polars as pl
from polar_llama import HnswIndex, embedding_async, Provider

corpus = pl.DataFrame({
    "doc_id": ["doc-1", "doc-2", "doc-3", "doc-4"],
    "text": ["AI research", "cooking tips", "machine learning", "recipes"],
}).with_columns(
    embedding=embedding_async(pl.col("text"), provider=Provider.OPENAI)
)
# Or fully offline on Apple Silicon (0.8.1+), same List[Float64] output:
#   embedding=embedding_local(pl.col("text"))
# Build a persistent index once, then query it as often as you like
index = HnswIndex.build(corpus, id_col="doc_id", embedding_col="embedding")

queries = pl.DataFrame({"q": ["artificial intelligence"]}).with_columns(
    embedding=embedding_async(pl.col("q"), provider=Provider.OPENAI)
)
print(index.query(queries, embedding_col="embedding", k=2))
# query_id | neighbor_id | distance | rank  ->  doc-1, doc-3

index.save("corpus.hnsw")        # reload later with HnswIndex.load(...)

import polars as pl
from polar_llama import tag_taxonomy, Provider

# Define your taxonomy
taxonomy = {
    "sentiment": {
        "description": "The emotional tone of the text",
        "values": {
            "positive": "Text expresses positive emotions or favorable opinions",
            "negative": "Text expresses negative emotions or unfavorable opinions",
            "neutral": "Text is factual and objective without clear emotional content"
        }
    },
    "urgency": {
        "description": "How urgent the content is",
        "values": {
            "high": "Requires immediate attention",
            "medium": "Should be addressed soon",
            "low": "Can be addressed at any time"
        }
    }
}

df = pl.DataFrame({
    "id": [1, 2],
    "message": [
        "URGENT: Server is down!",
        "Thanks for your help yesterday."
    ]
})

result = df.with_columns(
    tags=tag_taxonomy(
        pl.col("message"),
        taxonomy,
        provider=Provider.GROQ,
        model="llama-3.3-70b-versatile"
    )
)

# Extract specific values
result.select([
    "message",
    pl.col("tags").struct.field("sentiment").struct.field("value").alias("sentiment"),
    pl.col("tags").struct.field("sentiment").struct.field("confidence").alias("confidence"),
    pl.col("tags").struct.field("urgency").struct.field("value").alias("urgency")
])

# Per-candidate reasoning: thinking is a list of {value, reasoning} (0.5.3+)
result.select(
    "id",
    pl.col("tags").struct.field("sentiment").struct.field("thinking").alias("thinking"),
).explode("thinking").unnest("thinking")