FunctAIFunc¶

FunctAIFunc(
    fn,
    *,
    tools=None,
    template=None,
    messages=None,
    module_kwargs=None,
    examples=None,
    requires=None,
    **cfg,
)

A typed Python function whose body is a model call. Build with @ai.

Calling it runs the model and gives the answer, typed as the function's return type; predict gives every output; acall and apredict are the same in async code (an async def AI function is awaited directly).

Attributes¶

Methods¶

acall¶

FunctAIFunc.acall(*args, **kwargs)

await fn.acall(...): the answer, in async code. The call runs in a worker thread, so the event loop is free while the model answers.

apredict¶

FunctAIFunc.apredict(*args, **kwargs)

await fn.apredict(...): predict in async code (every output, the usage and the call's id), run in a worker thread so the event loop is free while the model answers.

bake¶

FunctAIFunc.bake(data, **options)

Train weights that answer this function; returns the baked model. fast = fn.using(lm=baked) runs the function on them. See functai.bake.bake for the options (student, teacher, labels, test, ...).

conversation¶

FunctAIFunc.conversation(
    id=None,
    *,
    store=None,
    context=None,
    earlier_without=(),
    sends='queue',
    **settings,
)

A conversation with this function: each call a turn that sees the earlier ones, kept in store; the function itself is unchanged.

Parameters¶

Returns¶

Examples¶

import functai
from functai import *

@ai
def tutor(message: str) -> str:
    """Tutor a student in fractions, one small step at a time."""
    ...

chat = tutor.conversation("alex")
chat("Hi, I'm Alex.")
chat("What is 1/2 + 1/3?")
[t.inputs["message"] for t in chat.turns[-1].saw]

functai: no model chosen, so using gpt-4.1-mini (environment ($OPENAI_API_KEY)). Choose one with functai.configure(lm=...).
["Hi, I'm Alex."]

explain¶

FunctAIFunc.explain()

How calls are laid out for the current model: adapter, reader, transports, formats.

freeze¶

FunctAIFunc.freeze()

Stop automatic instruction refinement for this function, now.

Only matters when refinement was turned on (@ai(instruction_autorefine_calls=n)): the first n calls then ask a model to rewrite the instruction from what it saw. freeze() keeps the instruction as it is from here on, which you want before evaluating, saving or comparing versions. It changes the function in place and returns it.

load¶

FunctAIFunc.load(path)

Read an instruction and demos written by fn.save(path) and use them, in place. Returns the function. (To save a whole program with everything it depends on: functai.save.)

load_state¶

FunctAIFunc.load_state(state)

Use this instruction and these demos (a ProgramState, or its dict from state().to_dict()), in place. Returns the function.

map¶

FunctAIFunc.map(data, *, threads=None, num_threads=None, progress=None)

Run on every row of a table, and return the run table.

evaluate without the scoring: the rows, the predictions (pred_<output>), and each row's error, seconds, tokens and model. A row that fails keeps its error; the others go on. Needs pip install "functai[data]".

For long runs, keep replies on disk (functai.configure( cache_replies="disk")): running map again after an interruption, or to retry the rows that failed, sends only what has no kept reply.

Parameters¶

Returns¶

See Also¶

• FunctAIFunc.vectorize: the function as a column expression.

Examples¶

@ai
def capital(country: str) -> str:
    """The country's capital city."""
    ...

capital.map([{"country": "Norway"}, {"country": "Ghana"}], threads=2, progress=False)

opt¶

FunctAIFunc.opt(data=None, *, optimizer=None, metric=None, valset=None, **opts)

An improved copy: its instruction and worked examples chosen from rows with known answers. This function is unchanged.

Only what the function sends besides its inputs changes: the instruction and the demos. Code, types and layout are never touched. functai.labeled_few_shot, functai.bootstrap_few_shot and functai.gepa are the common cases, by name.

Parameters¶

Returns¶

See Also¶

• evaluate: measure before and after.

Examples¶

from typing import Literal

@ai
def category(message: str) -> Literal["shipping", "billing", "product"]:
    """The support category of the message."""
    ...

train = [
    {"message": "The vase came smashed.", "result": "shipping"},
    {"message": "Money back please, the chair wobbles.", "result": "billing"},
    {"message": "The handle came off after two uses.", "result": "product"},
]
taught = category.opt(train)
[d.inputs["message"] for d in taught.demos]

optimization_runs¶

FunctAIFunc.optimization_runs()

How this function was improved, oldest first: the optimizer, the examples, what changed (and, for a search, its trials).

plan¶

FunctAIFunc.plan()

The lmcc plan for the current model: .explain(), .describe(), .render(...).

predict¶

FunctAIFunc.predict(*args, **kwargs)

The call, with everything it produced: every output (p.result, p.reasoning...), the tokens, the model's replies, the call's id.

Examples¶

@ai
def solve(question: str) -> float:
    """Solve the word problem."""
    reasoning: str = _ai     # step by step
    return _ai

p = solve.predict("3 pencils cost $1.20. How much do 10 cost?")
p.result, p.reasoning

render¶

FunctAIFunc.render(*args, **kwargs)

The exact request the next call would send, without sending it.

Parameters¶

Returns¶

See Also¶

• phistory: what was actually sent.

Examples¶

@ai
def capital(country: str) -> str:
    """The country's capital city."""
    ...

request = capital.render("Chile")
print(request.system)
print(request.messages[0].parts[0].text)

save¶

FunctAIFunc.save(path)

Write the instruction and demos to a JSON file (load reads it back).

state¶

FunctAIFunc.state()

The instruction and demos in use.

stream¶

FunctAIFunc.stream(*args, **kwargs)

Call the function and watch the answer being written.

The call starts at once, in the background, and is the same call as fn(...): the same retries, tools and call log line, the same value in the end. Iterate the stream for the answer's text as it arrives.

Parameters¶

Returns¶

See Also¶

• Stream: what this returns.

Examples¶

@ai
def haiku(topic: str) -> str:
    """A haiku about the topic."""
    ...

for piece in haiku.stream("the first snow"):
    print(piece, end="", flush=True)

Everything the model writes, reasoning first:

@ai
def solve(problem: str) -> float:
    """Solve the word problem."""
    reasoning: str = _ai["Step by step."]
    return _ai

s = solve.stream("3 pencils cost $1.20. How much do 10 cost?")
for event in s.events():
    if event.kind == "text":
        print(event.text, end="", flush=True)
s.result

to_dspy¶

FunctAIFunc.to_dspy(deepcopy=False)

Removed: FunctAI no longer runs on DSPy, so there is no DSPy program to give.

Always raises NotImplementedError. What an optimizer found is fn.state() (the instruction and the worked examples), which fn.save(path) writes as JSON; a whole program, with everything it depends on, is saved with functai.save.

unpack¶

FunctAIFunc.unpack(*args, threads=None, errors='raise', prefix='', **kwargs)

One column per field of the answer, to spread into a table.

For a function whose answer is a record (a dataclass, a pydantic model, a TypedDict): table.mutate(**fn.unpack(col.text)) adds one column per field. Each distinct row still costs one model call.

Parameters¶

Returns¶

See Also¶

• FunctAIFunc.vectorize: the whole answer as one column.

Examples¶

from dataclasses import dataclass
from dpyr import read, col

@dataclass
class Contact:
    name: str
    city: str | None   # None when the text does not say

@ai
def contact(text: str) -> Contact:
    """The person the text is about."""
    ...

people = read([{"text": "Ada Lovelace wrote to us from London."},
               {"text": "Grace Hopper called."}])
people.mutate(**contact.unpack(col.text))

using¶

FunctAIFunc.using(template=_KEEP, **settings)

A copy of this function with other settings or another layout.

The copy starts with the same instruction and demos; the original is untouched. A setting given as None is no longer set by the copy: it comes from configure or the defaults. An adapter replaces the template, and a template replaces the adapter.

Parameters¶

Returns¶

Examples¶

@ai
def capital(country: str) -> str:
    """The country's capital city."""
    ...

capital.using(lm="gpt-4.1-nano")("Chile")

vectorize¶

FunctAIFunc.vectorize(dtype=None, threads=None, errors='raise')

This function as a column expression, with options.

Calling an AI function on a dpyr column (fn(col.text)) is the same with the defaults. Each distinct input is sent once; answers are remembered for the session; the prompt in use now is the one the column is computed with.

Parameters¶

Returns¶

Examples¶

from dpyr import read, col

@ai
def capital(country: str) -> str:
    """The country's capital city."""
    ...

read([{"country": "Norway"}, {"country": "Ghana"}]).mutate(
    capital=capital.vectorize(threads=2)(col.country))