Models
The gateway: five backends, one spec grammar, retries, spend metering.
How do I get a model?
get_model(spec) is the only entry point you need; a spec is backend/model, and describe() tells you how one splits without building anything. There are five backends: claude-cli (a Claude subscription, no API key), openrouter (OPENROUTER_API_KEY, one key for most vendors with per-call cost in the response), openai (the OpenAI API directly, or any endpoint via OPENAI_BASE_URL), ollama (no credential, a local server) and mock (a scripted test double). Extra keyword arguments go straight to the backend's constructor โ temperature=0, max_tokens=512, and the gateway-wide retry_policy= / cost_ceiling_usd= / price_per_million= / spend= all arrive this way. The registry imports each adapter lazily inside the branch that needs it, so a missing optional dependency fails only for the backend that wanted it. openrouter, openai and ollama all speak the OpenAI wire format and share one base class, so they behave identically on everything except money and routing: same bind_tools, same with_structured_output, same streaming and async, same retry policy, same usage envelope.
from grapharc.gateway import describe, get_model for spec in ( "claude-cli/claude-sonnet-5", "openrouter/anthropic/claude-haiku-4.5", "openai/gpt-4o-mini", "ollama/llama3.1", "mock/anything", ): print(describe(spec)) # The mock backend needs no credential and never leaves the process. model = get_model("mock/anything", responses=["hello from a scripted model"]) print(model.invoke("say hi").content)
{'spec': 'claude-cli/claude-sonnet-5', 'backend': 'claude-cli', 'model': 'claude-sonnet-5'}
{'spec': 'openrouter/anthropic/claude-haiku-4.5', 'backend': 'openrouter', 'model': 'anthropic/claude-haiku-4.5'}
{'spec': 'openai/gpt-4o-mini', 'backend': 'openai', 'model': 'gpt-4o-mini'}
{'spec': 'ollama/llama3.1', 'backend': 'ollama', 'model': 'llama3.1'}
{'spec': 'mock/anything', 'backend': 'mock', 'model': 'anything'}
hello from a scripted modelHow do I write a spec, and what happens if I typo one?
Only the first segment of a spec is a backend, because OpenRouter model ids are themselves author/slug and must survive intact; :floor and :nitro suffixes are part of the model id and pass through untouched. A bare name gets the default backend, claude-cli. Unrecognised heads are rejected immediately with UnknownBackendError rather than folded into a model name โ that rejection is the case this rule exists to catch.
from grapharc.gateway import split_spec from grapharc.gateway.registry import UnknownBackendError print(split_spec("claude-sonnet-5")) # bare -> default backend print(split_spec("openrouter/openai/gpt-4o-mini:floor")) # author/slug survives print(split_spec("anthropic/claude-haiku-4.5")) # a known author, NOT openrouter print(split_spec("openai/gpt-4o-mini")) # a backend that is also an author try: split_spec("opnerouter/openai/gpt-4o-mini") except UnknownBackendError as exc: print(f"UnknownBackendError: {exc}")
('claude-cli', 'claude-sonnet-5')
('openrouter', 'openai/gpt-4o-mini:floor')
('claude-cli', 'anthropic/claude-haiku-4.5')
('openai', 'gpt-4o-mini')
UnknownBackendError: unknown backend 'opnerouter' in spec 'opnerouter/openai/gpt-4o-mini'; expected one of: claude-cli, openrouter, openai, ollama, mock โ or a bare model name for the claude-cli defaultHow do I see what my machine can actually reach?
grapharc models prints the backends and a few example specs and contacts nothing; giving it a spec resolves just that one. --check probes credentials, optional dependencies and PATH, and its output depends on your machine. <unset> is what you see with no key configured; with one, that line shows a redacted fingerprint, never the key. The Ollama line is an address rather than a credential, so it is printed whole โ it is where a request would go, not evidence that anything is listening. --check exits non-zero when no real provider is usable; mock being always-available does not count.
$ grapharc models backends: claude-cli, openrouter, openai, ollama, mock openrouter key: <unset> openai key: <unset> ollama url: http://localhost:11434/v1 examples: claude-cli/claude-sonnet-5 subscription, no API key openrouter/anthropic/claude-haiku-4.5 many providers, one key openrouter/openai/gpt-4o-mini:floor cheapest provider for that model openai/gpt-4o-mini the OpenAI API directly, your key ollama/llama3.1 a local server, no key and no bill grapharc models --check probes which of these this machine can use
$ grapharc models --check claude-cli usable 'claude' on PATH at ~/.local/bin/claude credential: claude subscription login (no API key) openrouter unusable no API key (set OPENROUTER_API_KEY, or add one to .env) credential: <unset> openai unusable no API key (set OPENAI_API_KEY, or add one to .env) credential: <unset> ollama usable local server at http://localhost:11434/v1 credential: none needed (local server) mock usable scripted test double; never reaches a provider local probe only โ no provider was contacted, so a configured key is not a validated one.
How do I run on a Claude subscription with no API key?
The claude-cli backend drives claude -p, authenticating with your existing Claude Code login so there is no API key anywhere. The tradeoffs are real: no tool-calling (bind_tools raises), no structured output (with_structured_output raises), no real streaming (only _generate is implemented, so .stream() yields one chunk โ the whole finished message โ and .ainvoke() runs the blocking subprocess in an executor thread), and no caching you control (every call is a fresh claude -p with --no-session-persistence and an empty working directory). The first two are enforced, not documented-and-hoped. It works this way because claude -p is a full agent with its own tools, settings, hooks and CLAUDE.md pickup, none of which GraphARC's permission engine can see or veto, so the adapter invokes it as a pure inference endpoint and gives up the agent features on purpose.
from grapharc.gateway import get_model model = get_model("claude-cli/claude-sonnet-5", timeout_seconds=120) print(model.invoke("Reply with exactly one word: pong").content) print(model.last_usage)
['claude', '-p', '--output-format', 'json', '--model', 'claude-sonnet-5', '--setting-sources', '', '--no-session-persistence', '--disallowedTools', '*', 'Task', 'Bash', 'BashOutput', 'KillShell', 'Read', 'Write', 'Edit', 'MultiEdit', 'NotebookEdit', 'Glob', 'Grep', 'WebFetch', 'WebSearch', 'TodoWrite', 'SlashCommand', 'Skill', 'ExitPlanMode', '--system-prompt', 'be terse']
How do I get tool-calling, structured output, streaming and async?
Use OpenRouter: one key reaches models from most vendors, and because the backend subclasses ChatOpenAI you get the whole LangChain chat-model surface โ bind_tools, with_structured_output, .stream(), .ainvoke(). Install the extra (uv sync --extra openrouter) and set OPENROUTER_API_KEY (or put it in a .env file) first. Two constructor defaults are worth knowing: max_tokens defaults to 4096 rather than the model's ceiling, and max_retries (the OpenAI SDK's own retry layer) defaults to 0.
model = get_model("openrouter/openai/gpt-4o-mini", api_key="sk-or-not-a-real-key") print(model._llm_type, "|", model.model_name) print("max_tokens:", model.max_tokens, "| sdk max_retries:", model.max_retries) # Neither of these raises here, and both raise NotImplementedError on claude-cli. model.bind_tools([get_weather]) model.with_structured_output(Verdict) print("bind_tools + with_structured_output: available") print("ainvoke:", callable(model.ainvoke), "| stream:", callable(model.stream))
grapharc-openrouter | openai/gpt-4o-mini max_tokens: 4096 | sdk max_retries: 0 bind_tools + with_structured_output: available ainvoke: True | stream: True
How do I use my own OpenAI key?
openai/โฆ goes straight to api.openai.com with no broker in between, which is what a contract naming who may see the prompt tends to require. The gateway accepts OPENAI_API_KEY, OPENAI_KEY, or openai-api-key in a file; langchain-openai would read the environment variable itself, so going through the gateway is what adds the file, the spellings, and an error naming the variable instead of an SDK exception four frames down. max_tokens is None here where OpenRouter defaults to 4096, because that default exists to dodge OpenRouter's credit reservation and OpenAI reserves nothing. OPENAI_BASE_URL (or the older OPENAI_API_BASE) is honoured, making this the backend for any OpenAI-compatible endpoint that is not Ollama โ a corporate gateway, a proxy, a self-hosted vLLM.
from grapharc.gateway import get_model priced = get_model( "openai/gpt-4o-mini", api_key="sk-not-a-real-key", price_per_million={"input": 0.15, "cached_input": 0.075, "output": 0.60}, ) unpriced = get_model("openai/gpt-4o-mini", api_key="sk-not-a-real-key")
with a rate card cost_usd=0.18 unpriced_calls=0 without one cost_usd=None unpriced_calls=1
How do I run models on my own machine?
ollama/โฆ talks to a local Ollama server over its OpenAI-compatible endpoint โ no key, no bill, no network egress โ and, unlike claude-cli, full tool-calling, so it is the cheapest way to exercise an agent node. The api key sent: ollama value is a placeholder: Ollama ignores the Authorization header and the OpenAI client refuses to send an empty one, so a constant with nothing to protect is sent; set OLLAMA_API_KEY when the address points at an authenticating proxy. OLLAMA_HOST โ the variable the ollama CLI itself reads โ is accepted in the shorthand forms people actually write, and the port is filled in only for the bare form, since a value with a scheme is a URL and is left to URL rules. Cost is charged 0.0 and does not land in unpriced_calls, because that means "the meter missed a bill" and here there is none to miss.
from grapharc.gateway.config import normalize_ollama_base_url for raw in ("127.0.0.1:11434", "gpu-box", "http://gpu-box:11434", "https://ollama.internal/v1/"): print(f"{raw:<28} -> {normalize_ollama_base_url(raw)}")
127.0.0.1:11434 -> http://127.0.0.1:11434/v1 gpu-box -> http://gpu-box:11434/v1 http://gpu-box:11434 -> http://gpu-box:11434/v1 https://ollama.internal/v1/ -> https://ollama.internal/v1
How do I give a reviewer a genuinely different provider?
A verifier that grades its author's own model family is correlated evidence, and different_providers() answers that question by comparing the vendor rather than the object. Two specs from the same author are False even though they are different models โ that is the point โ and the same vendor reached two different ways (claude-cli/claude-sonnet-5 vs openrouter/anthropic/โฆ, openai/gpt-4o-mini vs openrouter/openai/gpt-4o-mini) is False too. Both used to read True, because the check short-circuited whenever the backends differed, and adding a direct openai backend made that failure trivial to hit. The CLI wires this into live runs: grapharc demo stage5 --model โฆ --reviewer-model โฆ warns when the pair is correlated and proceeds anyway, because a stated weakness beats a silent one.
from grapharc.gateway import different_providers pairs = [ ("openrouter/anthropic/claude-haiku-4.5", "openrouter/openai/gpt-4o-mini"), ("openrouter/anthropic/claude-opus-4.5", "openrouter/anthropic/claude-haiku-4.5"), ("claude-cli/claude-sonnet-5", "openrouter/openai/gpt-4o-mini"), ("claude-cli/claude-sonnet-5", "claude-cli/claude-haiku-4.5"), ("claude-cli/claude-sonnet-5", "openrouter/anthropic/claude-haiku-4.5"), ("openai/gpt-4o-mini", "openrouter/openai/gpt-4o-mini"), ("ollama/llama3.1", "openai/gpt-4o-mini"), ] for author, reviewer in pairs: print(f"{different_providers(author, reviewer)!s:<5} {author} vs {reviewer}")
How do I add fallbacks and steer provider routing?
OpenRouter has two independent failover layers and GraphARC exposes both as constructor arguments: fallback_models is model-level (try a different model), while provider_order and friends are provider-level (try a different host serving the same model). The primary model is always first in models, and an openrouter/ prefix on a fallback is stripped. sort takes "price", "throughput" or "latency"; max_price_per_million caps the prompt price; require_parameters=True filters out providers that would silently drop a parameter you sent. Set none of them and the whole provider block is omitted rather than sent empty. None of these are OpenAI parameters, so they ride in extra_body, which langchain-openai merges into the JSON body verbatim.
model = get_model(
"openrouter/openai/gpt-4o-mini",
api_key="sk-or-not-a-real-key",
fallback_models=["openrouter/anthropic/claude-haiku-4.5", "google/gemini-2.5-flash"],
provider_order=["openai", "azure"],
allow_provider_fallbacks=False,
sort="price",
max_price_per_million=2.5,
require_parameters=True,
)
body = model._get_request_payload([HumanMessage(content="hi")])["extra_body"]
print(json.dumps(body, indent=2)){
"usage": { "include": true },
"models": ["openai/gpt-4o-mini", "anthropic/claude-haiku-4.5", "google/gemini-2.5-flash"],
"provider": {
"order": ["openai", "azure"],
"allow_fallbacks": false,
"sort": "price",
"max_price": { "prompt": 2.5 },
"require_parameters": true
}
}How do I know what a call cost?
Every backend fills in the same last_usage envelope after every non-streamed call, so a meter reads one shape whichever backend ran the turn. Cached input is still input: the envelope folds cache_creation and cache_read into input_tokens and keeps the breakdown in input_token_details, with the raw uncached figure preserved as uncached_input_tokens. In the worked example input_tokens is 1812, not the provider's 12. cost_usd is None when nobody could price the call.
model = get_model("claude-cli/claude-sonnet-5") with patch.object(subprocess, "run", lambda *a, **k: Completed()): print(model.invoke("ping").content) for key, value in model.last_usage.items(): print(f" {key}: {value}")
pong
input_tokens: 1812
output_tokens: 5
total_tokens: 1817
input_token_details: {'cache_creation': 300, 'cache_read': 1500}
uncached_input_tokens: 12
cost_usd: 0.0123
model: claude-sonnet-5
retries: 0
cumulative_cost_usd: 0.0123How do I stop a graph spending more than a fixed amount?
A SpendMeter accumulates cost_usd and refuses to go past a ceiling, enforcing at two points: after a call, so overspend is bounded by the single call that crossed the line, and before the next one, so an exhausted budget never reaches the provider at all. The two messages differ โ "exceeded" is the call that crossed, "reached before this call" is every call after it โ and the crossing call is charged before the raise, so a run catching CostCeilingExceeded still knows what it actually spent. You rarely build the meter by hand: cost_ceiling_usd= seeds one per model, and spend= shares one across several, which is what you want for a whole run. This is separate from Budget(max_tokens=โฆ), which the runtime meters per run โ the spend meter is per model object (or shared group) and counts dollars; the budget is per run and counts tokens, iterations and seconds.
from grapharc.gateway import CostCeilingExceeded, SpendMeter meter = SpendMeter(ceiling_usd=0.10) meter.charge(0.04, model="claude-sonnet-5") meter.charge(0.05, model="claude-sonnet-5") try: meter.charge(0.03, model="gpt-4o-mini") except CostCeilingExceeded as exc: print(exc) print("spent after the raise:", exc.spent_usd) # The next call is refused before it reaches a provider. try: meter.ensure_headroom(model="gpt-4o-mini") except CostCeilingExceeded as exc: print(exc) print(meter.snapshot())
cost ceiling exceeded: $0.120000 spent of $0.100000 after 3 call(s); this call cost $0.030000 โ model 'gpt-4o-mini'
spent after the raise: 0.12
cost ceiling reached before this call: $0.120000 spent of $0.100000 over 3 call(s) โ model 'gpt-4o-mini'
{'spent_usd': 0.12, 'ceiling_usd': 0.1, 'calls': 3, 'unpriced_calls': 0, 'per_model_usd': {'claude-sonnet-5': 0.09, 'gpt-4o-mini': 0.03}}What happens to cost accounting when I stream?
This is the one place where the accounting is knowingly incomplete, so it gets its own recipe rather than a footnote. A streamed call is checked against the ceiling before it starts and lands in unpriced_calls afterwards. LangChain routes streamed calls through _stream, never _generate, and OpenRouter reports the per-call cost in the final SSE chunk, which langchain-openai does not surface โ so last_usage is cleared to None rather than left holding the previous call's numbers, and the meter records that it missed one instead of implying it saw the whole bill.
meter = SpendMeter(ceiling_usd=0.01)
model = get_model(
"openrouter/openai/gpt-4o-mini",
api_key="sk-or-not-a-real-key",
spend=meter,
streaming=True,
)
with patch.object(ChatOpenAI, "_stream", lambda self, *a, **k: iter(CHUNKS)):
print("".join(c.content for c in model.stream("ping")))
print("last_usage:", model.last_usage)
print("meter:", meter.snapshot())pong
last_usage: None
meter: {'spent_usd': 0.0, 'ceiling_usd': 0.01, 'calls': 1, 'unpriced_calls': 1, 'per_model_usd': {}}What gets retried, and what does not?
A model call fails in two ways and treating them alike is expensive in both directions. is_transient is the whole policy, and it is closed by default: an exception has to present evidence of transience to be retried. Retried: 408, 409, 425, 429, any 5xx (529 included), connection and timeout errors that never reached a verdict, and anything raised as TransientGatewayError. Not retried: 400, 401, 402, 403, 404, 422, a content refusal, CostCeilingExceeded (spending more cannot fix having spent too much), and โ importantly โ anything unrecognised.
from grapharc.gateway import GatewayError, TransientGatewayError, is_transient cases = [ HTTPError(429), HTTPError(503), HTTPError(529), HTTPError(400), HTTPError(401), HTTPError(402), TimeoutError("read timed out"), ConnectionError("reset by peer"), TransientGatewayError("overloaded"), GatewayError("not logged in"), ValueError("I cannot help with that"), # a refusal is a verdict ] for exc in cases: print(f"{is_transient(exc)!s:<5} {type(exc).__name__}: {exc}")
How long does it wait between attempts?
Delay before attempt n+1 is initial * multiplier**(n-1), capped at max_backoff_seconds, then multiplied by a random factor in [1 - jitter, 1]. The defaults are 3 attempts, 0.5s initial, 20s cap, ร2, 25% jitter. Jitter shrinks the delay rather than centring it, so every draw is strictly larger than the previous attempt's, not merely larger on average โ a burst of unlucky short waits hammering a provider that just said 429 is exactly what that avoids. A provider's Retry-After header raises the wait but never lowers it, and is itself capped by max_backoff_seconds. Wire a policy into a model with retry_policy=; NO_RETRY is the one-attempt policy. The CLI has no status codes, so its failures are classified from their own text, and the deterministic markers are checked first and win.
policy = RetryPolicy(max_attempts=4, initial_backoff_seconds=0.5, jitter=0.25) print("un-jittered backoff:", [policy.base_delay(n) for n in range(1, 5)]) # `sleep=` is the seam that keeps this fast: nothing actually waits. print(call_with_retry(flaky, policy=policy, sleep=waited.append)) print("attempts:", len(attempts), "waits:", [round(w, 3) for w in waited])
un-jittered backoff: [0.5, 1.0, 2.0, 4.0] ok attempts: 3 waits: [0.486, 0.976] HTTP 400 -> attempts: 1 waits: []
pong after 3 attempts retries recorded in the envelope: 2 claude -p exited 1: Invalid API key ยท Please run /login -> 1 attempt(s)
How do I test all of this without spending anything?
ScriptedChatModel replays a fixed list of responses and records what it was asked; this is how the entire GraphARC test suite runs โ the gate tests verify orchestration mechanics (routing, budgets, traces, replay) and scripted responses exercise those deterministically. Running off the end of the script raises by default, because a graph that made more model calls than you scripted has changed behaviour; on_exhausted="repeat" is there for loops whose iteration count is not the thing under test. Dropped into a graph it is the same object the real backends are, so swapping the backend is a one-line change. The runtime meters the model call itself โ the node never touches the budget โ because a LangChain callback is installed for the duration of every node, so any chat model invoked on that thread reports usage to the run's meter, including calls buried in library code the node merely called. The ceiling is enforced inside on_llm_end, at the call that crossed the line, rather than at the node boundary after everything has already been paid for.
from grapharc.testing import ScriptedChatModel model = ScriptedChatModel(responses=["first answer", "second answer"]) print(model.invoke("q1").content) reply = model.invoke("q2") print(reply.content, reply.usage_metadata) print("calls:", model.call_count, "| first prompt:", model.calls[0][0].content) try: model.invoke("q3") except RuntimeError as exc: print("exhausted:", exc) repeating = ScriptedChatModel(responses=["same"], on_exhausted="repeat") print([repeating.invoke(str(n)).content for n in range(3)])
grapharc demo stage1 --model openrouter/anthropic/claude-haiku-4.5
grapharc demo stage5 --model claude-cli/claude-sonnet-5 \
--reviewer-model openrouter/openai/gpt-4o-mini