nemoir-runtime 0.8.0

Python runtime core for NemoIR — execute compiled agent workflows as structured state machines with tool orchestration, policy enforcement, model-backed stage execution, and live event streaming.

NemoIR Runtime

Python runtime core for NemoIR — an LLVM-inspired compiler stack for agentic workflows.

Executes compiled agent workflows as structured state machines with tool orchestration, policy enforcement, model-backed stage execution (via LiteLLM), and live event streaming.

Features

• Workflow runtime — state-machine execution with stage ordering, read/write resolution, transition selection, and run limits.
• Tool framework — capability-based tool registration, catalog-driven parameter validation, and policy-gated invocation (fs.read, fs.write, user.confirm, os.shell, user.elicit).
• Policy engine — deny and before-policies with expression evaluation: and/or boolean combinators, eq/starts_with/contains predicates over bound trigger arguments, path containment and equality guards.
• Model integration — ModelStageExecutor with LiteLLM adapter, structured output enforcement, tool-call loop, ModelRouter for per-stage model routing, and optional streaming via ModelStreamingAdapter.
• Deterministic stages — exec: workflow stages run a fixed capability with bound args via DeterministicStageExecutor, with no model call. Tool selection happens at runtime construction (fail-fast on no-match/ambiguous). Tool.output_schema / @tool(returns=…) declare tool return shapes so the runtime can match tools to stage outputs.
• Live event streaming — WorkflowRuntime.stream() / generated Agent.stream() async iterator emitting WorkflowEvent values (run lifecycle, model deltas, tool calls, policy decisions) for UIs, debugging, and observability.
• Compiler backend target — generated workflow-specific Python packages consume this runtime; see nemoir-backend-python in the main NemoIR repo.

Install

pip install nemoir-runtime

Quick start

import asyncio
from pathlib import Path
from nemoir_runtime import WorkflowRuntime, WorkflowManifest, Tool, ToolContext, ToolRegistry

# Define tools — @tool derives input_schema from type hints; output_schema
# is derived from the return annotation (or set explicitly via returns=)
from nemoir_runtime import tool

@tool(capability="fs.read", description="Read a file")
async def read_file(*, path: Path, ctx: ToolContext) -> str:
    return Path(path).read_text()

tools = ToolRegistry([read_file])

# Load a manifest (typically generated by the NemoIR compiler)
manifest = WorkflowManifest(...)

runtime = WorkflowRuntime(manifest=manifest, tools=tools, stage_executor=my_executor)
result = await runtime.run({"task": "analyze code"})
print(result.output)

See the NemoIR project for the full compiler workflow (DSL → IR → generated package).

Policy engine

Deny policies use expression evaluation to gate capability calls. Supported predicates: eq (exact match), starts_with (prefix), contains (substring or path containment). Boolean and/or combinators short-circuit at runtime. in [...] is DSL sugar that lowers to or of eq calls.

from nemoir_runtime import PolicySpec, ExprSpec, TriggerSpec, RefSpec

# deny os.shell(command) if not (
#   command.eq("python run.py")
#   or command.starts_with("git commit -m ")
# )
shell_allowlist = PolicySpec(
    id="shell-allowlist",
    kind="deny",
    trigger=TriggerSpec(capability="os.shell", bind={"command": "command"}),
    condition=ExprSpec(
        kind="not",
        expr=ExprSpec(
            kind="or",
            exprs=(
                ExprSpec(
                    kind="method_call",
                    receiver=ExprSpec(kind="ref", ref=RefSpec(kind="bound", name="command")),
                    method="eq",
                    args=(ExprSpec(kind="literal", type="string", value="python run.py"),),
                ),
                ExprSpec(
                    kind="method_call",
                    receiver=ExprSpec(kind="ref", ref=RefSpec(kind="bound", name="command")),
                    method="starts_with",
                    args=(ExprSpec(kind="literal", type="string", value="git commit -m "),),
                ),
            ),
        ),
    ),
)

# deny fs.write(path) if not path.eq(candidate_path)
write_allowlist = PolicySpec(
    id="write-allowlist",
    kind="deny",
    trigger=TriggerSpec(capability="fs.write", bind={"path": "path"}),
    condition=ExprSpec(
        kind="not",
        expr=ExprSpec(
            kind="method_call",
            receiver=ExprSpec(kind="ref", ref=RefSpec(kind="bound", name="path")),
            method="eq",
            args=(ExprSpec(kind="ref", ref=RefSpec(kind="input", name="candidate_path")),),
        ),
    ),
)

Official tools

nemoir-runtime ships with official, importable Tool implementations for every capability in the catalog. Import exactly the tools you need:

from nemoir_runtime import ToolRegistry
from nemoir_runtime.official_tools import (
    ask_user,
    confirm_user,
    edit_file,
    read_file,
    run_shell,
    write_file,
)

tools = ToolRegistry([read_file, write_file, edit_file, run_shell, ask_user, confirm_user])

Pick a subset if you don't need every capability:

tools = ToolRegistry([read_file, edit_file, run_shell])

Policy boundary

Official tools validate inputs and perform the operation. They do not enforce workflow policy — path containment, write confirmation, shell allowlists, and similar authorization remain owned by NemoIR policies.

The user.elicit and user.confirm tools use the console and will raise on non-interactive environments. Provide your own tool implementations for such deployments.

Reasoning channel

WorkflowEventChannel includes a dedicated "reasoning" value for raw provider chain-of-thought (DeepSeek delta.reasoning_content, Qwen, etc.). It is distinct from "reasoning_summary", which is reserved for future curated public summaries (Anthropic thinking, OpenAI o-series).

Reasoning forwarding is opt-in (default off) to preserve the default posture of not exposing hidden/private chain-of-thought. Enable it via ModelSpec.reasoning or a model config mapping:

agent = Agent(
    model={"name": "openai/deepseek-v4-flash", "reasoning": "raw", ...},
    tools=tools,
)

async for event in agent.stream(inputs):
    if event.kind == "model_delta" and event.channel == "reasoning":
        print(f"[reasoning] {event.text}", end="", flush=True)
    elif event.kind == "model_delta" and event.channel == "assistant":
        print(event.text, end="", flush=True)

Or per-run via RunOptions(reasoning="raw").

Reasoning text is never merged into the final structured-output content; stage output validation is unaffected.

Requirements

• Python ≥ 3.11
• LiteLLM ≥ 1.0.0 (for LiteLLMModelAdapter; custom ModelAdapter implementations can avoid this dependency)