Install the package from PyPI:
pip install sno-llmixThe package name is sno-llmix. The import path is llmix.
For Redis-backed response cache:
pip install "sno-llmix[redis]"import asyncio
import os
from llmix import (
CallInput,
CallPipeline,
KeyPool,
PipelineConfig,
TwoTierCache,
openai_dispatch,
)
async def main() -> None:
pipeline = CallPipeline(
PipelineConfig(
dispatch=openai_dispatch(),
response_cache=TwoTierCache("memory"),
)
)
pipeline.set_key_pool("openai", KeyPool([os.environ["OPENAI_API_KEY"]]))
response = await pipeline.call(
CallInput(
config={
"provider": "openai",
"model": "gpt-4o-mini",
"common": {"temperature": 0.7, "max_output_tokens": 256},
"caching": {"strategy": "memory"},
},
messages=[
{"role": "user", "content": "In one sentence, what is LLMix?"}
],
)
)
print(response.content)
print(f"cache_hit={response.cache_hit} usage={response.usage}")
await pipeline.close()
asyncio.run(main())Run it:
OPENAI_API_KEY=sk-... python quickstart.pyUse redis-or-memory when you want Redis in deployed services but still want a
local fallback when Redis is unavailable.
import os
from llmix import PipelineConfig, TwoTierCache, openai_dispatch
config = PipelineConfig(
dispatch=openai_dispatch(),
response_cache=TwoTierCache(
"redis-or-memory",
redis_url=os.environ.get("REDIS_URL"),
max_items=2048,
ttl_seconds=3600,
),
)For strict Redis mode, use TwoTierCache("redis", redis_url=...). It raises if
no Redis URL is configured.
Register one pool per provider:
from llmix import KeyPool, load_keys_from_env
pipeline.set_key_pool("openai", load_keys_from_env("openai"))
# Or explicitly:
pipeline.set_key_pool("openai", KeyPool(["sk-live-1", "sk-live-2"]))load_keys_from_env("openai") checks OPENAI_KEYS first, then
OPENAI_API_KEY. OPENAI_KEYS is comma-separated.
If you build a Python dispatch helper with a prebuilt client=..., do not
register a key pool for that provider. The client already owns the API key.
LLMix will reject that combination to avoid marking the wrong key dead.
Use the registry when presets are part of a running service:
from llmix import ConfigRegistryManager, ConfigRegistryPublisher, resolve_config_dir
root = resolve_config_dir().config_dir
ConfigRegistryPublisher(root).publish()
manager = ConfigRegistryManager.open(root)
config = manager.get_preset("search", "summary")
print(manager.available_presets())For signed production bundles, verify MDA source files while publishing and open the registry through a signed root:
from llmix import (
ConfigRegistryManager,
ConfigRegistryOpenOptions,
ConfigRegistryPublisher,
ConfigRegistryPublishOptions,
RegistryRootSigningOptions,
RegistryRootVerificationOptions,
)
ConfigRegistryPublisher(root).publish(
options=ConfigRegistryPublishOptions(
trusted_runtime=True,
trust_policy=trust_policy,
did_web_verifier=did_web_verifier,
registry_root=RegistryRootSigningOptions(signer=registry_root_signer),
)
)
manager = ConfigRegistryManager.open(
root,
ConfigRegistryOpenOptions(
signed_root=RegistryRootVerificationOptions(
trust_policy=registry_root_trust_policy,
did_web_verifier=did_web_verifier,
expected_root_digest=expected_root_digest,
)
),
)Then pass the resolved config into the pipeline:
response = await pipeline.call(
CallInput(
config=config,
messages=[{"role": "user", "content": "Summarize this."}],
)
)Direct loaders are useful for tests, authoring tools, and migrations:
from llmix import load_mda_config, load_mda_config_preset
config = load_mda_config("./config/llm/search/summary.mda")
preset = load_mda_config_preset("summary", "./config/llm/search")For production runtime code, prefer ConfigRegistryManager.
The dispatch function is just the provider call. That is the design center.
from llmix import LLMUsage, ProviderResult
async def my_dispatch(ctx):
result = await my_client.chat(
model=ctx.model,
messages=ctx.messages,
api_key=ctx.api_key,
**ctx.kwargs,
)
return ProviderResult(
content=result.text,
model=ctx.model,
usage=LLMUsage(input_tokens=0, output_tokens=0, total_tokens=0),
)Wire it into the pipeline:
pipeline = CallPipeline(PipelineConfig(dispatch=my_dispatch))Treat timeout.total_time in MDA config as a runtime budget, not as automatic
transport cancellation. The Python pipeline does not wrap dispatch in
asyncio.wait_for, and it does not force-cancel provider requests before retrying.
The built-in provider dispatchers use provider SDK or HTTP client defaults, but
those defaults are not derived from config["timeout"]["total_time"]. If your
service needs a hard timeout, put it at the provider transport layer:
import httpx
from llmix import LLMUsage, ProviderResult
async def my_dispatch(ctx):
total_time = ctx.config.get("timeout", {}).get("total_time", 120)
timeout = httpx.Timeout(total_time, connect=min(10.0, total_time))
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.post(
"https://api.example.com/v1/chat/completions",
headers={"Authorization": f"Bearer {ctx.api_key}"},
json={
"model": ctx.model,
"messages": ctx.messages,
**ctx.kwargs,
},
)
response.raise_for_status()
data = response.json()
usage = data.get("usage", {})
return ProviderResult(
content=data["choices"][0]["message"]["content"],
model=data.get("model", ctx.model),
usage=LLMUsage(
input_tokens=usage.get("prompt_tokens", 0),
output_tokens=usage.get("completion_tokens", 0),
total_tokens=usage.get("total_tokens", 0),
),
)Do not rely on caller-side timeout wrappers alone unless cancellation reaches the real network request. Before retrying a timed-out attempt, make sure the previous provider request has been cancelled, closed, or allowed to finish; otherwise the service can create duplicate in-flight generations.
PipelineConfig(
dispatch=openai_dispatch(),
max_retries=3,
retry_base_ms=1000,
retry_max_delay_ms=30000,
circuit_breaker_threshold=3,
circuit_breaker_cooldown_seconds=30.0,
semaphore_initial=32,
semaphore_min=4,
response_cache=TwoTierCache("memory"),
)Most services should start with defaults. Tune only after real traffic shows a specific pressure point.