Skip to content

Compiled graph

The runnable graph returned by Graph.compile(). See Running a graph.

agentflow.compiled.CompiledGraph

Source code in agentflow/compiled.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
class CompiledGraph:
    def __init__(
        self,
        *,
        schema: type,
        channels: Mapping[str, Channel],
        nodes: Mapping[str, Any],
        edges: Mapping[Any, list[Any]],
        branches: Mapping[str, Any],
        checkpointer: Any = None,
        step_limit: int = 100,
        hooks: Any = None,
        max_node_concurrency: int | None = None,
        isolate_state: str = "fanout",
    ) -> None:
        from agentflow.observability import Hooks

        self.schema = schema
        self.checkpointer = checkpointer
        self.hooks = hooks or Hooks()
        self._graph = _Graph(
            channels=channels,
            nodes=nodes,
            edges=edges,
            branches=branches,
            step_limit=step_limit,
            max_node_concurrency=max_node_concurrency,
            isolate_state=isolate_state,
        )

    # ------------------------------------------------------------------
    # Running
    # ------------------------------------------------------------------

    async def invoke(
        self,
        input: Mapping[str, Any] | None = None,
        *,
        thread: str = "default",
        timeout: float | None = None,
    ) -> dict[str, Any]:
        """Run from ``input`` to completion and return the final state.

        If the run suspends on an interrupt, the interrupted checkpoint's
        state is returned; inspect ``get_state(thread)`` to see it is not done
        and call :meth:`resume`.

        ``timeout`` bounds the whole run. On expiry the in-flight super-step is
        cancelled and :class:`~agentflow.errors.RunTimeout` is raised; the last
        completed step's checkpoint is intact, so a checkpointed run resumes
        from there.
        """
        data = self._validate_input(input)
        coro = run(
            self._graph,
            data,
            thread=thread,
            start_frontier=None,
            start_step=0,
            checkpointer=self.checkpointer,
            hooks=self.hooks,
        )
        cp = await self._run_with_timeout(coro, thread, timeout)
        return dict(cp.state)

    async def stream(
        self, input: Mapping[str, Any] | None = None, *, thread: str = "default"
    ) -> AsyncIterator[StreamEvent]:
        """Run from ``input``, yielding :class:`StreamEvent`s as they occur."""
        async for ev in self._stream_run(
            thread=thread,
            initial_state=self._validate_input(input),
            start_frontier=None,
            start_step=0,
        ):
            yield ev

    # ------------------------------------------------------------------
    # Resume (human-in-the-loop)
    # ------------------------------------------------------------------

    async def resume(
        self, thread: str, value: Any = None, *, timeout: float | None = None
    ) -> dict[str, Any]:
        """Continue a suspended run, feeding ``value`` to the interrupted node."""
        cp = await self._load_for_resume(thread)
        resume_values = {cp.interrupt_node: value} if cp.interrupt_node else {}
        coro = run(
            self._graph,
            dict(cp.state),
            thread=thread,
            start_frontier=list(cp.next),
            start_step=cp.step,
            checkpointer=self.checkpointer,
            resume_values=resume_values,
            hooks=self.hooks,
        )
        result = await self._run_with_timeout(coro, thread, timeout)
        return dict(result.state)

    async def stream_resume(self, thread: str, value: Any = None) -> AsyncIterator[StreamEvent]:
        """Streaming form of :meth:`resume`."""
        cp = await self._load_for_resume(thread)
        resume_values = {cp.interrupt_node: value} if cp.interrupt_node else {}
        async for ev in self._stream_run(
            thread=thread,
            initial_state=dict(cp.state),
            start_frontier=list(cp.next),
            start_step=cp.step,
            resume_values=resume_values,
        ):
            yield ev

    async def run_until_done(
        self,
        input: Mapping[str, Any] | None = None,
        *,
        thread: str = "default",
        max_sleep: float | None = None,
    ) -> dict[str, Any]:
        """Run inline, driving ``ctx.wait`` timer suspends to completion.

        Unlike :meth:`invoke`, which returns as soon as the run suspends, this
        keeps the process alive across timer waits: when the run parks on a
        ``ctx.wait``, it sleeps until the checkpoint's ``wake_at`` and resumes,
        looping until the run finishes. A *human* interrupt (``ctx.interrupt``)
        still returns control — a person must resolve that, not the clock.

        This is the inline counterpart to the control plane's parked execution.
        It is honest about its one limit: the process must stay alive for the
        whole wait. For true parked, cross-restart waiting, drive the graph
        through the control plane instead. Requires a checkpointer.

        ``max_sleep`` caps each individual sleep (useful in tests) — the wait
        still resolves only once the real ``wake_at`` has passed.
        """
        if self.checkpointer is None:
            raise CheckpointError("run_until_done requires a checkpointer")
        state = await self.invoke(input, thread=thread)
        while True:
            cp = await self.checkpointer.get(thread)
            if cp is None or cp.suspend_kind != "timer":
                return state  # finished, or parked on a human interrupt
            await self._sleep_until(cp.wake_at, max_sleep)
            state = await self.resume(thread)

    @staticmethod
    async def _sleep_until(wake_at: str | None, max_sleep: float | None) -> None:
        if not wake_at:
            return
        from datetime import UTC, datetime

        delay = (datetime.fromisoformat(wake_at) - datetime.now(UTC)).total_seconds()
        if delay <= 0:
            return
        if max_sleep is not None:
            delay = min(delay, max_sleep)
        await asyncio.sleep(delay)

    # ------------------------------------------------------------------
    # Synchronous facade (convenience for non-async callers)
    # ------------------------------------------------------------------

    def invoke_sync(
        self,
        input: Mapping[str, Any] | None = None,
        *,
        thread: str = "default",
        timeout: float | None = None,
    ) -> dict[str, Any]:
        """Blocking wrapper over :meth:`invoke` via ``asyncio.run``.

        For synchronous callers only. Raises :class:`RuntimeError` if called
        from within a running event loop — use ``await invoke(...)`` there.
        """
        return _run_sync(self.invoke(input, thread=thread, timeout=timeout))

    def resume_sync(
        self, thread: str, value: Any = None, *, timeout: float | None = None
    ) -> dict[str, Any]:
        """Blocking wrapper over :meth:`resume`."""
        return _run_sync(self.resume(thread, value, timeout=timeout))

    def stream_sync(
        self, input: Mapping[str, Any] | None = None, *, thread: str = "default"
    ) -> list[StreamEvent]:
        """Blocking wrapper that drains :meth:`stream` into a list.

        Returns all events at once (not incremental) — a convenience for sync
        scripts; use the async ``stream`` for live consumption.
        """

        async def _collect() -> list[StreamEvent]:
            return [ev async for ev in self.stream(input, thread=thread)]

        return _run_sync(_collect())

    # ------------------------------------------------------------------
    # Inspection / time-travel
    # ------------------------------------------------------------------

    async def get_state(self, thread: str, step: int | None = None) -> Checkpoint | None:
        if self.checkpointer is None:
            raise CheckpointError("no checkpointer configured")
        return await self.checkpointer.get(thread, step)

    def history(self, thread: str) -> AsyncIterator[Checkpoint]:
        if self.checkpointer is None:
            raise CheckpointError("no checkpointer configured")
        return self.checkpointer.history(thread)

    # ------------------------------------------------------------------
    # Internals
    # ------------------------------------------------------------------

    def _validate_input(self, input: Mapping[str, Any] | None) -> dict[str, Any]:
        """Reject input keys that are not declared channels, before running.

        A typo in an input key would otherwise sit silently in the state until
        a node happened to read it. Fail fast with a clear message instead.
        """
        data = dict(input or {})
        if not data:
            return data
        declared = set(self._graph.channels)
        unknown = [
            k
            for k in data
            if k not in declared or k.startswith(RESERVED_PREFIX) or k in RESERVED_NAMES
        ]
        if unknown:
            raise GraphError(
                f"input has undeclared channel(s) {sorted(unknown)}; "
                f"declared channels: {sorted(declared)}"
            )
        return data

    async def _run_with_timeout(self, coro, thread: str, timeout: float | None):
        if timeout is None:
            return await coro
        try:
            return await asyncio.wait_for(coro, timeout=timeout)
        except TimeoutError:
            # The in-flight super-step was cancelled; the last completed step
            # is already checkpointed. Report the step reached, if known.
            step = 0
            if self.checkpointer is not None:
                last = await self.checkpointer.get(thread)
                if last is not None:
                    step = last.step
            raise RunTimeout(thread, timeout, step) from None

    async def _load_for_resume(self, thread: str) -> Checkpoint:
        if self.checkpointer is None:
            raise CheckpointError("resume requires a checkpointer")
        cp = await self.checkpointer.get(thread)
        if cp is None:
            raise GraphError(f"no checkpoint for thread {thread!r}")
        if cp.done:
            raise GraphError(f"run {thread!r} already finished; nothing to resume")
        return cp

    async def _stream_run(
        self,
        *,
        thread: str,
        initial_state: Mapping[str, Any],
        start_frontier: list[str] | None,
        start_step: int,
        resume_values: Mapping[str, Any] | None = None,
    ) -> AsyncIterator[StreamEvent]:
        """Run in a background task, draining the emit queue as a stream."""
        queue: asyncio.Queue[Any] = asyncio.Queue()
        done = object()

        async def _driver() -> None:
            try:
                await run(
                    self._graph,
                    dict(initial_state),
                    thread=thread,
                    start_frontier=start_frontier,
                    start_step=start_step,
                    checkpointer=self.checkpointer,
                    resume_values=resume_values,
                    emit_queue=queue,
                    hooks=self.hooks,
                )
            finally:
                queue.put_nowait(done)

        task = asyncio.create_task(_driver())
        try:
            while True:
                item = await queue.get()
                if item is done:
                    break
                yield item
        finally:
            # Surface a driver exception rather than swallowing it.
            await task

invoke async 

invoke(input: Mapping[str, Any] | None = None, *, thread: str = 'default', timeout: float | None = None) -> dict[str, Any]

Run from input to completion and return the final state.

If the run suspends on an interrupt, the interrupted checkpoint's state is returned; inspect get_state(thread) to see it is not done and call :meth:resume.

timeout bounds the whole run. On expiry the in-flight super-step is cancelled and :class:~agentflow.errors.RunTimeout is raised; the last completed step's checkpoint is intact, so a checkpointed run resumes from there.

Source code in agentflow/compiled.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
async def invoke(
    self,
    input: Mapping[str, Any] | None = None,
    *,
    thread: str = "default",
    timeout: float | None = None,
) -> dict[str, Any]:
    """Run from ``input`` to completion and return the final state.

    If the run suspends on an interrupt, the interrupted checkpoint's
    state is returned; inspect ``get_state(thread)`` to see it is not done
    and call :meth:`resume`.

    ``timeout`` bounds the whole run. On expiry the in-flight super-step is
    cancelled and :class:`~agentflow.errors.RunTimeout` is raised; the last
    completed step's checkpoint is intact, so a checkpointed run resumes
    from there.
    """
    data = self._validate_input(input)
    coro = run(
        self._graph,
        data,
        thread=thread,
        start_frontier=None,
        start_step=0,
        checkpointer=self.checkpointer,
        hooks=self.hooks,
    )
    cp = await self._run_with_timeout(coro, thread, timeout)
    return dict(cp.state)

stream async 

stream(input: Mapping[str, Any] | None = None, *, thread: str = 'default') -> AsyncIterator[StreamEvent]

Run from input, yielding :class:StreamEvents as they occur.

Source code in agentflow/compiled.py 
105
106
107
108
109
110
111
112
113
114
115
async def stream(
    self, input: Mapping[str, Any] | None = None, *, thread: str = "default"
) -> AsyncIterator[StreamEvent]:
    """Run from ``input``, yielding :class:`StreamEvent`s as they occur."""
    async for ev in self._stream_run(
        thread=thread,
        initial_state=self._validate_input(input),
        start_frontier=None,
        start_step=0,
    ):
        yield ev

resume async 

resume(thread: str, value: Any = None, *, timeout: float | None = None) -> dict[str, Any]

Continue a suspended run, feeding value to the interrupted node.

Source code in agentflow/compiled.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
async def resume(
    self, thread: str, value: Any = None, *, timeout: float | None = None
) -> dict[str, Any]:
    """Continue a suspended run, feeding ``value`` to the interrupted node."""
    cp = await self._load_for_resume(thread)
    resume_values = {cp.interrupt_node: value} if cp.interrupt_node else {}
    coro = run(
        self._graph,
        dict(cp.state),
        thread=thread,
        start_frontier=list(cp.next),
        start_step=cp.step,
        checkpointer=self.checkpointer,
        resume_values=resume_values,
        hooks=self.hooks,
    )
    result = await self._run_with_timeout(coro, thread, timeout)
    return dict(result.state)

stream_resume async 

stream_resume(thread: str, value: Any = None) -> AsyncIterator[StreamEvent]

Streaming form of :meth:resume.

Source code in agentflow/compiled.py 
140
141
142
143
144
145
146
147
148
149
150
151
async def stream_resume(self, thread: str, value: Any = None) -> AsyncIterator[StreamEvent]:
    """Streaming form of :meth:`resume`."""
    cp = await self._load_for_resume(thread)
    resume_values = {cp.interrupt_node: value} if cp.interrupt_node else {}
    async for ev in self._stream_run(
        thread=thread,
        initial_state=dict(cp.state),
        start_frontier=list(cp.next),
        start_step=cp.step,
        resume_values=resume_values,
    ):
        yield ev

run_until_done async 

run_until_done(input: Mapping[str, Any] | None = None, *, thread: str = 'default', max_sleep: float | None = None) -> dict[str, Any]

Run inline, driving ctx.wait timer suspends to completion.

Unlike :meth:invoke, which returns as soon as the run suspends, this keeps the process alive across timer waits: when the run parks on a ctx.wait, it sleeps until the checkpoint's wake_at and resumes, looping until the run finishes. A human interrupt (ctx.interrupt) still returns control — a person must resolve that, not the clock.

This is the inline counterpart to the control plane's parked execution. It is honest about its one limit: the process must stay alive for the whole wait. For true parked, cross-restart waiting, drive the graph through the control plane instead. Requires a checkpointer.

max_sleep caps each individual sleep (useful in tests) — the wait still resolves only once the real wake_at has passed.

Source code in agentflow/compiled.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
async def run_until_done(
    self,
    input: Mapping[str, Any] | None = None,
    *,
    thread: str = "default",
    max_sleep: float | None = None,
) -> dict[str, Any]:
    """Run inline, driving ``ctx.wait`` timer suspends to completion.

    Unlike :meth:`invoke`, which returns as soon as the run suspends, this
    keeps the process alive across timer waits: when the run parks on a
    ``ctx.wait``, it sleeps until the checkpoint's ``wake_at`` and resumes,
    looping until the run finishes. A *human* interrupt (``ctx.interrupt``)
    still returns control — a person must resolve that, not the clock.

    This is the inline counterpart to the control plane's parked execution.
    It is honest about its one limit: the process must stay alive for the
    whole wait. For true parked, cross-restart waiting, drive the graph
    through the control plane instead. Requires a checkpointer.

    ``max_sleep`` caps each individual sleep (useful in tests) — the wait
    still resolves only once the real ``wake_at`` has passed.
    """
    if self.checkpointer is None:
        raise CheckpointError("run_until_done requires a checkpointer")
    state = await self.invoke(input, thread=thread)
    while True:
        cp = await self.checkpointer.get(thread)
        if cp is None or cp.suspend_kind != "timer":
            return state  # finished, or parked on a human interrupt
        await self._sleep_until(cp.wake_at, max_sleep)
        state = await self.resume(thread)

invoke_sync 

invoke_sync(input: Mapping[str, Any] | None = None, *, thread: str = 'default', timeout: float | None = None) -> dict[str, Any]

Blocking wrapper over :meth:invoke via asyncio.run.

For synchronous callers only. Raises :class:RuntimeError if called from within a running event loop — use await invoke(...) there.

Source code in agentflow/compiled.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
def invoke_sync(
    self,
    input: Mapping[str, Any] | None = None,
    *,
    thread: str = "default",
    timeout: float | None = None,
) -> dict[str, Any]:
    """Blocking wrapper over :meth:`invoke` via ``asyncio.run``.

    For synchronous callers only. Raises :class:`RuntimeError` if called
    from within a running event loop — use ``await invoke(...)`` there.
    """
    return _run_sync(self.invoke(input, thread=thread, timeout=timeout))

resume_sync 

resume_sync(thread: str, value: Any = None, *, timeout: float | None = None) -> dict[str, Any]

Blocking wrapper over :meth:resume.

Source code in agentflow/compiled.py 
217
218
219
220
221
def resume_sync(
    self, thread: str, value: Any = None, *, timeout: float | None = None
) -> dict[str, Any]:
    """Blocking wrapper over :meth:`resume`."""
    return _run_sync(self.resume(thread, value, timeout=timeout))

stream_sync 

stream_sync(input: Mapping[str, Any] | None = None, *, thread: str = 'default') -> list[StreamEvent]

Blocking wrapper that drains :meth:stream into a list.

Returns all events at once (not incremental) — a convenience for sync scripts; use the async stream for live consumption.

Source code in agentflow/compiled.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
def stream_sync(
    self, input: Mapping[str, Any] | None = None, *, thread: str = "default"
) -> list[StreamEvent]:
    """Blocking wrapper that drains :meth:`stream` into a list.

    Returns all events at once (not incremental) — a convenience for sync
    scripts; use the async ``stream`` for live consumption.
    """

    async def _collect() -> list[StreamEvent]:
        return [ev async for ev in self.stream(input, thread=thread)]

    return _run_sync(_collect())