Dagic

A minimal workflow DAG (Directed Acyclic Graph) definition language and an asynchronous execution engine, implemented in Python.

Dagic lets you describe a computation graph using a tiny, deliberately limited language. The graph's nodes are functions supplied by your host program; the edges are the function arguments. At compile time the graph is type-checked, and at run time it is executed concurrently across its independent branches.

Why

Dagic is aimed at bringing piping capabilities to LLMs.

In traditional tool calling, the agent is itself responsible for shuttling state between tools. Every intermediate result has to round-trip through the model — as an argument (or part of it) — which wastes turns: the agent spends tokens describing values it already produced instead of deciding what to do next.

The common fix is a code execution tool, which lets the model chain calls with ordinary variable assignments. That is trivial in a local harness like a coding CLI, but complicated and expensive for server-side harnesses, which must sandbox arbitrary code, manage runtimes, and defend a large attack surface.

Dagic sits in the gap between simple tool calls and full code execution: a workflow orchestration layer that is more expressive than a single tool call (arbitrary chains, parallelism, type checking) yet far safer and simpler to host than arbitrary code. The language is stripped down to just the two operations you need to define a DAG — assignment and function calls — so the model composes existing tools into controlled, typed, verifiable workflows instead of running free-form code.

Installation

Requires Python 3.10+.

pip install dagic

Syntax

A Dagic program is a sequence of statements. Each statement is one of:

• an assignment: name = <expression>;
• a function call: func(<expression>, ...);

result = add(create("1"), create("2"));
store(result);

joined = join(["Hello", "World!"], " ");
print(joined);

• The functions are the nodes of the DAG.
• The edges are the function arguments.
• Every edge has a strict type. Passing a value of the wrong type to a function is a compile-time error.
• Function definitions and types are provided by your host program (see below).

Data types

The only built-in types are:

• strings — "Hello, World!"
• arrays — ["Hello", "World!"] (items must share the exact same type, and the array's type is inferred from them)

Every other type (numbers, objects, custom types, ...) is defined by the host program. An array's type edges come from functions registered for it, so passing a List[float] where a List[int] is expected is rejected at compile time.

Execution model

1. A Dagic program compiles into a DAG.
2. Execution starts from the terminal nodes — top-level function calls that return nothing (like print or store) — and runs backwards, resolving every dependency.
3. Independent branches execute concurrently.
4. A program without at least one terminal node is invalid. It would have no exit point, so it is rejected at compile time.
5. Similarly, every named subgraph must be referenced at least once; unused ("orphan") subgraphs are rejected.

Defining functions (the host side)

Functions and their types are defined in Python and given to Dagic when it runs. Register them on a Module with the @register decorator. A registered function must:

• have type annotations on every parameter and the return value;
• have no *args, **kwargs, default parameters, or keyword-only parameters.

from dagic import Module

math = Module(name="math", desc="Basic arithmetic.")


@math.register
def create(value: str) -> float:
    """Create a float from a string."""
    return float(value)


@math.register
def add(a: float, b: float) -> float:
    """Add two numbers."""
    return a + b

Functions whose -> None return type are terminals. Everything else produces a value that must be consumed by another call.

Using the engine

The package ships a small builtin module, float_math (float arithmetic: add, subtract, multiply, divide, power, modulus, floor_divide, absolute, negate, and a create from-string constructor). It is a Module instance exported as float_math.float_math. Pass any host modules you need to Dagic:

import asyncio
from dagic import Dagic, Module
from dagic.builtins import float_math

sink = []

io = Module(name="io", desc="I/O helpers.")


@io.register
def store(value: float) -> None:
    sink.append(value)


async def main():
    dagic = Dagic([float_math.float_math, io])
    await dagic.run('result = add(create("1"), create("2")); store(result);')
    print(sink)  # [3.0]


asyncio.run(main())

Dagic.run is async: it compiles the source against the registered modules, builds the graph, and executes it concurrently.

Development

make test    # run the test suite (pytest)
make format  # format with ruff

License

MIT.