nemoir-runtime 0.4.1

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 (e.g., path containment guards).
• Model integration — ModelStageExecutor with LiteLLM adapter, structured output enforcement, tool-call loop, ModelRouter for per-stage model routing, and optional streaming via ModelStreamingAdapter.
• 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
async def read_file(*, path: Path, ctx: ToolContext) -> str:
    return Path(path).read_text()

tools = ToolRegistry([
    Tool(name="read_file", capability="fs.read", description="Read a file",
         input_schema={"path": Path}, handler=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).

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)