teff.graph.execution

teff.graph.execution

The graph execution engine.

The heavy lifting of a :class:~teff.graph.Graph run lives here as a standalone :func:execute function so the :class:~teff.graph.Graph class stays a thin facade. It walks the graph from the entry point, executes each node, routes along edges whose conditions match the current state, and (optionally) persists checkpoints so a crashed or interrupted run can be resumed.

Functions:

Name	Description
execute	Run graph with run/session correlation ids in the log context.

execute async

execute(
    graph,
    state,
    tools=None,
    registry=None,
    reducers=None,
    hooks=None,
    node_timeout=None,
    max_iterations=None,
    checkpointer=None,
    checkpoint_id=None,
    owner=DEFAULT_OWNER,
    resume=None,
    tracer=None,
    state_schema=None,
    emit=None,
    providers=None,
    default_provider=None,
    default_model=None,
    on_llm_payload=None,
)

Run graph with run/session correlation ids in the log context.

Source code in teff/graph/execution.py

async def execute(
    graph,
    state: dict | State,
    tools: list[Tool] | None = None,
    registry: NodeRegistry | None = None,
    reducers: dict[str, Reducer] | None = None,
    hooks: dict[str, Callable] | None = None,
    node_timeout: float | None = None,
    max_iterations: int | None = None,
    checkpointer: Checkpointer | None = None,
    checkpoint_id: str | None = None,
    owner: str = DEFAULT_OWNER,
    resume: dict | None = None,
    tracer: RunTracer | None = None,
    state_schema: dict | None = None,
    emit: "Callable[[StreamEvent], Awaitable[None]] | None" = None,
    providers: "dict | ProviderRegistry | None" = None,
    default_provider: str | None = None,
    default_model: str | None = None,
    on_llm_payload: "Callable[..., Awaitable[None]] | None" = None,
) -> dict | State:
    """Run *graph* with run/session correlation ids in the log context."""
    run = new_run_id()
    session = checkpoint_id or ""
    with run_id_ctx(run_id=run, session_id=session):
        log.info("run_start checkpoint=%s", session or "-")
        try:
            result = await _execute_impl(
                graph,
                state,
                tools=tools,
                registry=registry,
                reducers=reducers,
                hooks=hooks,
                node_timeout=node_timeout,
                max_iterations=max_iterations,
                checkpointer=checkpointer,
                checkpoint_id=checkpoint_id,
                owner=owner,
                resume=resume,
                tracer=tracer,
                state_schema=state_schema,
                emit=emit,
                providers=providers,
                default_provider=default_provider,
                default_model=default_model,
                on_llm_payload=on_llm_payload,
            )
        except GraphInterrupt:
            raise
        except Exception as exc:
            log.error("run_end status=error error=%r", str(exc))
            raise
        log.info("run_end status=ok")
        return result