fastllm

fastllm provides a common async interface for Anthropic, OpenAI Responses and Chat, Gemini, and OpenAI-compatible providers such as Kimi. Define messages and tools in a shared format and call acomplete. Select the model and, where needed, its provider without rewriting the conversation or tool loop. Supported features vary by provider.

AsyncResponses provides a streaming Responses API interface over the same providers. It translates input items and events and keeps a canonical ResponseState between turns. For transports without provider continuation ids, it replays that state automatically.

Install

Run pip install python-fastllm or clone from github and install locally.

Setup

from aidialog.msg_parts import Msg, Part, Text, Thinking, ToolUse, InputImage, mk_tool_res_msg, Completion
from fastllm.acomplete import acomplete
from fastllm.types import sample_img_url, sample_doms, sonn, sonn46
import asyncio, json

def user(text): return Msg(role='user', content=[Text(text)])

async def stream(msgs, model, max_think=0, **kw):
    "Print streamed parts and return the final Completion; limit printed thinking deltas with max_think."
    cnt = 0
    async for o in await acomplete(msgs, model, stream=True, **kw):
        if not isinstance(o, Part): continue
        if isinstance(o, Thinking):
            cnt += 1
            if cnt > max_think: continue
        print(o.formatted, end='', flush=True)
    print()
    return o

mtok = 1024

Chat across providers

Use the same acomplete interface with Claude, GPT, Gemini, or Kimi. Set model and any provider-specific routing arguments:

models = [(sonn, {}), ('gpt-4o-mini', {}), ('models/gemini-3-flash-preview', {}),
    ('accounts/fireworks/models/kimi-k3', dict(vendor_name='fireworks_ai'))]
for name, kw in models:
    r = await acomplete([user("Translate 'hello' into French. Return only the translation.")],
        model=name, max_tokens=mtok, **kw)
    text = ''.join(p.text for p in r.message.content if isinstance(p, Text))
    print(f"{name}: {text.strip()}")

claude-sonnet-5: Bonjour
gpt-4o-mini: Bonjour
models/gemini-3-flash-preview: Bonjour
accounts/fireworks/models/kimi-k3: Bonjour

System prompts

Pass system to supply a system prompt. fastllm translates it to the provider’s field: Anthropic system, OpenAI Responses instructions, or Gemini system_instruction.

sys = "You are a pirate chef. Always respond in pirate speak and mention food. Use one short sentence."

print("Claude: ", end='')
r = await stream([user("What should I do today?")], model=sonn, system=sys, max_tokens=mtok)

print("Gemini: ", end='')
r = await stream([user("What should I do today?")], model='models/gemini-3-flash-preview', system=sys, max_tokens=mtok)

Claude: Arrr, chart a course fer the galley and cook up a hearty stew, matey!
Gemini: Sharpen yer cutlass and feast on spicy shark stew, ye scurvy dog!

Tool calling

Define tools in the shared schema. fastllm translates it to the provider’s tool format. This example handles a tool request, adds the result to the history, and continues the conversation:

tools = [{"type": "function", "function": dict(name="get_weather", description="Get current weather for a city",
    parameters=dict(type="object", properties={"city": {"type": "string"}}, required=["city"]))}]

msgs = [user("What's the weather in Paris?")]
r1 = await stream(msgs, model=sonn, tools=tools, max_tokens=mtok)
print("Tool calls:", r1.tool_calls)

- ⏳ `get_weather(city="Paris")` ⏳

Tool calls: [ToolUse(raw={'caller': {'type': 'direct'}}, cache_control=None, id='toolu_01XacBpJHnsjtPWJWr55nA2m', name='get_weather', arguments={'city': 'Paris'}, server=False, text=None)]

msgs += [r1.message, mk_tool_res_msg(r1.tool_calls, ['22°C, sunny with light clouds']),
    user("Should I bring a jacket? Answer in one short sentence.")]
r2 = await stream(msgs, model=sonn, tools=tools, max_tokens=mtok)

No, a jacket isn't necessary — it's a mild, sunny 22°C day.

Tool Choice

Control whether the model must use tools, can’t use tools, or decides on its own:

r = await acomplete([user('Hello there!')], model=sonn, tools=tools, tool_choice='required', max_tokens=mtok)
print("Required:", [tc.name for tc in r.tool_calls])

r = await acomplete([user("What's the weather?")], model=sonn, tools=tools, tool_choice='none', max_tokens=mtok)
print("None:", r.tool_calls)

Required: ['get_weather']
None: []

Thinking and extended reasoning

Set reasoning_effort to low, medium, or high for supported models. The adapters translate these values to provider-specific settings. Reasoning output uses Thinking parts; this example prints a short excerpt alongside the answer.

print("Claude: ", end='')
r = await stream([user("What is 127 × 849? Return just the number.")], model=sonn46,
    reasoning_effort='low', max_tokens=8192)
for p in r.message.content:
    if isinstance(p, Thinking): print(f"Thinking excerpt: {p.text[:150]}...")

print("Kimi: ", end='')
r = await stream([user('What is 127 × 849? Return just the number.')], model='accounts/fireworks/models/kimi-k3',
    vendor_name='fireworks_ai', reasoning_effort='low', max_tokens=8192)
for p in r.message.content:
    if isinstance(p, Thinking): print(f"Thinking excerpt: {p.text[:150]}...")

Claude: 107823
Thinking excerpt: 127 × 849 = 127 × 800 + 127 × 49 = 101600 + 6223 = 107823...
Kimi: 107823
Thinking excerpt: 127 × 849 = 107823. Let me verify: 127*800=101600; 127*49=6223; total 107823. Return just number....

Web Search (Server Tools)

OpenAI’s Responses API supports server-side web search. Server tool calls are normalized alongside regular tool calls:

ws_tools = [{"type": "web_search_preview"}]
r = await acomplete([user("What is the latest Python release? Reply with the version number only.")],
    model='gpt-4o-mini', tools=ws_tools, max_tokens=512)
print("Server tools used:", [tc.name for tc in r.tool_calls if tc.server])

Server tools used: ['web_search']

Caching with Anthropic

Set a part’s cache_control to request prompt caching. Later requests can reuse eligible cached content. The example prints usage for two calls with the same system prompt:

long_ctx = "You are an expert on the solar system. " * 200
system = Text(long_ctx, cache_control={'type': 'ephemeral'})

r1 = await acomplete([user("What is Jupiter's mass? Answer in one sentence.")],
    model=sonn, system=system, max_tokens=mtok)
print("First call: cache creation tokens", r1.usage.cache_creation_tokens)

r2 = await acomplete([user("What is Saturn's mass? Answer in one sentence.")],
    model=sonn, system=system, max_tokens=mtok)
print("Second call: cache read tokens", r2.usage.cached_tokens)

First call: cache creation tokens 2204
Second call: cache read tokens 2204

Media inputs

Use InputImage with a model that supports images. The example sends the same image message to three providers:

img_msg = Msg(role='user', content=[InputImage(sample_img_url), Text('List three visible objects, using nouns only.')])

for name, kw in [(sonn, {}), ('gpt-4o-mini', {}), ('models/gemini-3-flash-preview', {})]:
    print(f"{name}: ", end='')
    r = await stream([img_msg], model=name, max_tokens=mtok, **kw)

claude-sonnet-5: Mountains, lake, trees
gpt-4o-mini: 1. Mountains  
2. Trees  
3. Lake
models/gemini-3-flash-preview: planks, lake, mountains

The media adapters support the following part types:

Media parts accept a URL or base64 data URL as their text. An unsupported combination raises ValueError.