from llm_gent.flow import Role, Factory, planner, grader
analyst = Role("analyst", "anthropic", "claude-opus-5-5")
@planner(role=analyst)
async def plan(ctx, task): ...
@grader(role=analyst)
async def grade(ctx, result): ...
flow = Factory(lg, saia_factory=factory).create()
await flow.call(plan).call(grade).run()
Multi-role agentic flows
- Mix models & providers per role
- Fluent primitives: loops, fan-out, panels, branches
- Pause/resume & crash resilience
- Integrated cost tracking & boundaries
- Runtime for agent swarms
pip install llm-gent
from llm_saia import SAIA
saia = SAIA.builder().backend(backend).build()
check = await saia.verify(code, "adds two numbers")
if not check.value.passed:
crit = await saia.critique(check.value.reason)
code = await saia.refine(
code, crit.value.counter_argument)
Verb vocabulary for AI agents
- ask, verify, critique, refine, extract, classify
- Full inner loop with tool calling
- Guard system with retries
- Tracing system for transparency
- Zero-dependency core
pip install llm-saia
tesla = kelt.kg.entities.create(
scope_key="global",
canonical_name="Tesla",
entity_type="company",
aliases=["TSLA", "Tesla Motors"])
elon, _ = kelt.kg.entities.find_or_create(
"global", "Elon Musk", "person")
kelt.kg.relationships.add(elon, tesla.id,
"founded", scope_key="global")
Knowledge · Embedding · Learning · Training
- Atomic facts model & knowledge graph
- Contexts, compaction & RAG
- SFT & DPO fine-tuning with provenance (CAS)
- Embedding quantization: f16, int8, int4
pip install llm-kelt
from llm_infer.client import Factory, FallbackClient
router = Factory(lg).from_config(config["llm"])
client = FallbackClient(lg, router,
fallbacks={"qwen3.8-27b": "claude-opus-5-5"})
response = client.chat(messages, model="qwen3.8-27b")
Multi-provider/model inference routing
- Role-based routing with fallback
- Sync, async, streaming
- Local models via Ollama or vLLM
- Rate limits with backoff
pip install "llm-infer[client]"
from appinfra.app import AppBuilder
app = AppBuilder("mytool").build()
@app.tool(name="sync")
@app.argument("--limit", type=int)
def sync_tool(self):
self.lg.info(f"limit={self.args.limit}")
Infrastructure for Python services
- Fluent CLI builder
- Enhanced YAML config with hot-reload
- Multi-target structured logging
- Service lifecycle with restart policies
pip install appinfra