Models

A model adapter is the object an agent uses to talk to an LLM provider. You usually don’t build one yourself: pass a model name as a string and loomflow picks the adapter from its prefix. Construct an adapter directly when you need to configure the client, and write your own when your provider isn’t covered (any class that satisfies the Model protocol works).

from loomflow import Agent
 
agent = Agent("...", model="claude-opus-4-7")     # → AnthropicModel
agent = Agent("...", model="gpt-4o")              # → OpenAIModel
agent = Agent("...", model="mistral-large")       # → LiteLLMModel
agent = Agent("...", model="bedrock/claude-3...")  # → LiteLLMModel
agent = Agent("...", model="ollama/llama3")       # → LiteLLMModel (local)
agent = Agent("...", model="echo")                # → EchoModel (zero-key)

Five shipped adapters

When to construct an instance directly

• Custom client config: Pass your own AsyncOpenAI / AsyncAnthropic:

  from openai import AsyncOpenAI
  from loomflow import Agent
  from loomflow.model.openai import OpenAIModel
   
  client = AsyncOpenAI(timeout=30.0, max_retries=0)
  agent = Agent("...", model=OpenAIModel("gpt-4o", client=client))

• Tuning per-adapter options (output token cap, request timeout, pricing override):

  from loomflow import Agent
  from loomflow.model.anthropic import AnthropicModel
   
  agent = Agent("...", model=AnthropicModel(
      "claude-opus-4-7",
      api_key="...",
      max_tokens=8192,
      request_timeout_s=60,
  ))

  Sampling temperature is not a constructor option. It is a per-call argument of the adapter’s complete() and stream() methods, and the agent loop doesn’t set it, so agents run at the provider’s default. Pass temperature= yourself only when you call the adapter directly.

Read more

Custom Model adapter

The Model protocol requires a name and an async stream() method. complete() is optional: if your adapter has one, agent.run() uses it for a single non-streaming call, and otherwise the loop consumes the stream. The loop passes the keyword arguments below on every call, so accept them all even if your provider ignores some.

from collections.abc import AsyncIterator
from typing import Any
 
from loomflow import Agent
from loomflow.core.types import Message, ModelChunk, Role, ToolDef, Usage
 
 
class ShoutModel:
    name = "shout"
 
    async def stream(
        self,
        messages: list[Message],
        *,
        tools: list[ToolDef] | None = None,
        temperature: float = 1.0,
        max_tokens: int | None = None,
        output_schema: Any | None = None,
        effort: str | None = None,
        strict_effort: bool = False,
        prompt_caching: Any = None,
    ) -> AsyncIterator[ModelChunk]:
        last = next(m for m in reversed(messages) if m.role == Role.USER)
        yield ModelChunk(kind="text", text=last.content.upper())
        yield ModelChunk(
            kind="finish",
            finish_reason="stop",
            usage=Usage(input_tokens=1, output_tokens=1),
        )
 
 
agent = Agent("Shout back.", model=ShoutModel())
result = await agent.run("hello")
print(result.output)  # HELLO

stream() is an async generator yielding ModelChunks, discriminated by kind:

• ModelChunk(kind="text", text="...") for streamed output text.
• ModelChunk(kind="thinking", text="...") for reasoning text, on models that expose it.
• ModelChunk(kind="tool_call", tool_call=...) for each tool call the model makes.
• ModelChunk(kind="finish", finish_reason=..., usage=...) once at the end, with token counts.

The agent loop aggregates chunks; it doesn’t care about chunk boundaries.