From 4aa737eee5da2820ee651bc55d98e63f8dab126e Mon Sep 17 00:00:00 2001 From: Evan Mattson <35585003+moonbox3@users.noreply.github.com> Date: Fri, 14 Aug 2026 00:31:52 +0000 Subject: [PATCH] Python: [BREAKING] Require building functional workflow instances (#7521) * Harden functional workflow continuation authority Use a versioned opaque single-use token on WorkflowRunResult, validate it before request correlation, consume it immediately before replayed user code, and rotate it on each pause. Carry the same explicit authority through streaming and non-streaming FunctionalWorkflowAgent responses. Files changed: functional workflow/runtime result APIs, functional HITL regression tests, core agent guidance, and the functional HITL sample. Next iteration: enforce pending-state overlap and token-authorized abandonment, then document and test checkpoint authorization boundaries. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Enforce one pending functional continuation Reject fresh messages and checkpoint restores while an in-memory continuation is pending. Add token-authorized abandonment on FunctionalWorkflow and FunctionalWorkflowAgent, and clear retained replay state atomically when authority is consumed while preserving the active message for token rotation and checkpoints. Files changed: functional workflow runtime and agent adapter, functional lifecycle regression tests, and core workflow guidance. Next iteration: preserve and document authorized checkpoint continuation boundaries. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Preserve authorized functional checkpoint continuation Treat checkpoint restore as a host- and storage-authorized path independent of process-local continuation tokens, and issue fresh authority whenever restored execution pauses again. Cover default and per-run storage, deterministic and custom request IDs, token rotation, and checkpoint-plus-response restore. Files changed: functional workflow and checkpoint interface guidance, functional checkpoint lifecycle tests, the functional HITL sample, and core workflow guidance. Next iteration: run the final repository-wide Python validation gates. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Validate Python continuation hardening Run the complete Python workspace checks, aggregate coverage suite, repository hooks, and core package build from the final combined worktree. Keep the validation iteration code-neutral because all gates pass without corrective changes. Files changed: none; this commit records the final validation gate. Blockers: none. Next iteration: no remaining AFK tasks. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Handle functional checkpoint continuation failures Publish retained continuation state only after checkpoint persistence succeeds, and cover reuse after a transient save failure. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: a8f47743-1cdc-4924-8e1b-667d0d790b78 * Address functional continuation review findings Add owner recovery for lost tokens, harden malformed token validation, preserve consistent failure surfaces, and keep agent pending state aligned with resumable workflow state. Document process-local single-use continuation semantics and extend regression coverage across direct, streaming, checkpoint, and agent paths. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: a8f47743-1cdc-4924-8e1b-667d0d790b78 * Handle functional continuation cancellation Release the workflow run guard when cancellation interrupts resumed user code while keeping the single-use continuation token consumed. Replace sample assertions with explicit runtime checks and add cancellation regression coverage. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: a8f47743-1cdc-4924-8e1b-667d0d790b78 * Simplify functional workflow instance isolation Remove continuation-token handling and align functional workflows with the graph workflow ownership model: one stateful instance per logical caller or session. Add create_instance() for independent callers, document the ownership contract, and cover pending-state isolation between instances. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: a8f47743-1cdc-4924-8e1b-667d0d790b78 * Scope functional workflow checkpoint storage Do not inherit checkpoint storage when creating an independent workflow instance. Allow hosts to provide an explicitly caller-scoped storage adapter and document that shared checkpoint access requires host authorization and tenant isolation. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: a8f47743-1cdc-4924-8e1b-667d0d790b78 * Require building functional workflow instances Make @workflow return a stateless FunctionalWorkflowDefinition and require build() before run() or as_agent(). This aligns functional workflows with the graph definition/build lifecycle and prevents module-level decorated definitions from retaining caller state. Move checkpoint configuration to build(), export the definition type, migrate samples, and cover isolated built instances. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: a8f47743-1cdc-4924-8e1b-667d0d790b78 --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: a8f47743-1cdc-4924-8e1b-667d0d790b78 --- python/PACKAGE_STATUS.md | 3 +- python/packages/core/AGENTS.md | 6 + .../packages/core/agent_framework/__init__.py | 2 + .../core/agent_framework/__init__.pyi | 2 + .../agent_framework/_workflows/_functional.py | 95 +++++-- .../workflow/test_functional_workflow.py | 256 +++++++++++------- .../05_functional_workflow_with_agents.py | 3 +- .../06_functional_workflow_basics.py | 3 +- .../functional/agent_integration.py | 7 +- .../03-workflows/functional/basic_pipeline.py | 8 +- .../functional/basic_streaming_pipeline.py | 10 +- .../03-workflows/functional/hitl_review.py | 9 +- .../functional/naive_group_chat.py | 3 +- .../functional/parallel_pipeline.py | 9 +- .../functional/steps_and_checkpointing.py | 13 +- 15 files changed, 291 insertions(+), 138 deletions(-) diff --git a/python/PACKAGE_STATUS.md b/python/PACKAGE_STATUS.md index f1dfb5bdc..d14af5d06 100644 --- a/python/PACKAGE_STATUS.md +++ b/python/PACKAGE_STATUS.md @@ -115,7 +115,8 @@ listed below. - `agent-framework-core`: functional workflow APIs from `agent_framework/_workflows/_functional.py`, including `RunContext`, `step`, - `FunctionalWorkflow`, `workflow`, and `FunctionalWorkflowAgent` + `FunctionalWorkflowDefinition`, `FunctionalWorkflow`, `workflow`, and + `FunctionalWorkflowAgent` #### `HARNESS` diff --git a/python/packages/core/AGENTS.md b/python/packages/core/AGENTS.md index 298537336..593aa1e4b 100644 --- a/python/packages/core/AGENTS.md +++ b/python/packages/core/AGENTS.md @@ -201,6 +201,12 @@ agent_framework/ every output-capable executor not selected by `output_from`. - **`WorkflowRunResult`** - Non-streaming workflow result with Workflow Output `get_outputs()` and Intermediate Output `get_intermediate_outputs()` accessors +- **Functional workflow definition/build lifecycle** - `@workflow` returns a stateless + `FunctionalWorkflowDefinition`. Call `build()` to create a stateful `FunctionalWorkflow` scoped to one logical + caller or session. The definition has no `run()` or `as_agent()` surface, so module-level decorated definitions + cannot accidentally retain caller state. Each built workflow and its `FunctionalWorkflowAgent` must remain scoped + to that caller/session. Pass a caller-scoped checkpoint storage to `build(checkpoint_storage=...)` when needed; + hosts remain responsible for authorizing and tenant-scoping access to any shared checkpoint adapter. - **Orchestrators**: `SequentialOrchestrator`, `ConcurrentOrchestrator`, `GroupChatOrchestrator`, `MagenticOrchestrator`, `HandoffOrchestrator` ## Built-in Providers diff --git a/python/packages/core/agent_framework/__init__.py b/python/packages/core/agent_framework/__init__.py index d7ab91687..46e0d9285 100644 --- a/python/packages/core/agent_framework/__init__.py +++ b/python/packages/core/agent_framework/__init__.py @@ -321,6 +321,7 @@ _LAZY_MODULE_EXPORTS: Final[Mapping[str, tuple[str, ...]]] = { "._workflows._function_executor": ("FunctionExecutor", "executor"), "._workflows._functional": ( "FunctionalWorkflow", + "FunctionalWorkflowDefinition", "FunctionalWorkflowAgent", "RunContext", "StepWrapper", @@ -478,6 +479,7 @@ __all__ = [ "FunctionTool", "FunctionalWorkflow", "FunctionalWorkflowAgent", + "FunctionalWorkflowDefinition", "GeneratedEmbeddings", "GraphConnectivityError", "HistoryProvider", diff --git a/python/packages/core/agent_framework/__init__.pyi b/python/packages/core/agent_framework/__init__.pyi index f72269c02..4f4f10fb0 100644 --- a/python/packages/core/agent_framework/__init__.pyi +++ b/python/packages/core/agent_framework/__init__.pyi @@ -285,6 +285,7 @@ from ._workflows._function_executor import FunctionExecutor, executor from ._workflows._functional import ( FunctionalWorkflow, FunctionalWorkflowAgent, + FunctionalWorkflowDefinition, RunContext, StepWrapper, get_run_context, @@ -442,6 +443,7 @@ __all__ = [ "FunctionTool", "FunctionalWorkflow", "FunctionalWorkflowAgent", + "FunctionalWorkflowDefinition", "GeneratedEmbeddings", "GraphConnectivityError", "HistoryProvider", diff --git a/python/packages/core/agent_framework/_workflows/_functional.py b/python/packages/core/agent_framework/_workflows/_functional.py index 2ffe99807..9e4deb340 100644 --- a/python/packages/core/agent_framework/_workflows/_functional.py +++ b/python/packages/core/agent_framework/_workflows/_functional.py @@ -21,7 +21,10 @@ parameter to access HITL and state APIs directly. Key public symbols: -* :func:`workflow` / :class:`FunctionalWorkflow` — decorator and runtime. +* :func:`workflow` / :class:`FunctionalWorkflowDefinition` — decorator and + stateless definition. +* :class:`FunctionalWorkflow` — stateful runtime created by + :meth:`FunctionalWorkflowDefinition.build`. * :func:`step` / :class:`StepWrapper` — optional step decorator. * :class:`RunContext` — execution context injected into workflow and step functions. @@ -628,6 +631,46 @@ def step( return _decorator +# --------------------------------------------------------------------------- +# FunctionalWorkflowDefinition +# --------------------------------------------------------------------------- + + +@experimental(feature_id=ExperimentalFeature.FUNCTIONAL_WORKFLOWS) +class FunctionalWorkflowDefinition: + """Stateless definition produced by :func:`workflow`. + + Call :meth:`build` to create a stateful :class:`FunctionalWorkflow`. + Each built workflow represents one logical caller or session. + """ + + def __init__( + self, + func: Callable[..., Awaitable[Any]], + *, + name: str | None = None, + description: str | None = None, + ) -> None: + FunctionalWorkflow._classify_signature(func) + self._func = func + self.name = name or func.__name__ + self.description = description + functools.update_wrapper(self, func) # type: ignore[arg-type] + + def build( + self, + *, + checkpoint_storage: CheckpointStorage | None = None, + ) -> FunctionalWorkflow: + """Build a stateful workflow for one logical caller or session.""" + return FunctionalWorkflow( + self._func, + name=self.name, + description=self.description, + checkpoint_storage=checkpoint_storage, + ) + + # --------------------------------------------------------------------------- # FunctionalWorkflow # --------------------------------------------------------------------------- @@ -637,8 +680,8 @@ def step( class FunctionalWorkflow: """A workflow backed by a user-defined async function. - Created by the :func:`workflow` decorator. Exposes the same ``run()`` - interface as graph-based :class:`Workflow` objects, returning a + Built from a :class:`FunctionalWorkflowDefinition`. Exposes the same + ``run()`` interface as graph-based :class:`Workflow` objects, returning a :class:`WorkflowRunResult` (or a :class:`ResponseStream` in streaming mode). @@ -646,6 +689,10 @@ class FunctionalWorkflow: edge wiring is involved. Native Python control flow (``if``/``else``, ``for``, ``asyncio.gather``) is used for branching and parallelism. + Like graph-based :class:`Workflow`, each instance owns mutable execution + state across calls to :meth:`run`. Scope an instance to one logical + caller or session; build separate instances for independent callers. + Args: func: The async function that implements the workflow logic. name: Display name for the workflow. Defaults to ``func.__name__``. @@ -664,7 +711,8 @@ class FunctionalWorkflow: return await to_upper(data) - result = await my_pipeline.run("hello") + pipeline = my_pipeline.build() + result = await pipeline.run("hello") print(result.get_outputs()) # ['HELLO'] """ @@ -933,7 +981,7 @@ class FunctionalWorkflow: if storage is None: raise ValueError( "Cannot restore from checkpoint without checkpoint_storage. " - "Provide checkpoint_storage parameter or set it on the @workflow decorator." + "Provide checkpoint_storage to build() or to this run." ) checkpoint = await storage.load(checkpoint_id) if checkpoint.graph_signature_hash != self.graph_signature_hash: @@ -1258,7 +1306,7 @@ class FunctionalWorkflow: @overload -def workflow(func: Callable[..., Awaitable[Any]]) -> FunctionalWorkflow: ... +def workflow(func: Callable[..., Awaitable[Any]]) -> FunctionalWorkflowDefinition: ... @overload @@ -1266,8 +1314,7 @@ def workflow( *, name: str | None = None, description: str | None = None, - checkpoint_storage: CheckpointStorage | None = None, -) -> Callable[[Callable[..., Awaitable[Any]]], FunctionalWorkflow]: ... +) -> Callable[[Callable[..., Awaitable[Any]]], FunctionalWorkflowDefinition]: ... @experimental(feature_id=ExperimentalFeature.FUNCTIONAL_WORKFLOWS) @@ -1276,29 +1323,26 @@ def workflow( *, name: str | None = None, description: str | None = None, - checkpoint_storage: CheckpointStorage | None = None, -) -> FunctionalWorkflow | Callable[[Callable[..., Awaitable[Any]]], FunctionalWorkflow]: - """Decorator that converts an async function into a :class:`FunctionalWorkflow`. +) -> FunctionalWorkflowDefinition | Callable[[Callable[..., Awaitable[Any]]], FunctionalWorkflowDefinition]: + """Decorator that creates a stateless :class:`FunctionalWorkflowDefinition`. Supports both bare ``@workflow`` and parameterized ``@workflow(name="my_wf")`` forms. The decorated function receives its input as the first positional argument and a :class:`RunContext` instance wherever a parameter is annotated with - that type. The resulting :class:`FunctionalWorkflow` object exposes the - same ``run()`` interface as graph-based workflows. + that type. Call ``build()`` on the resulting definition to create a + stateful :class:`FunctionalWorkflow`. Args: func: The async function to decorate (when using the bare ``@workflow`` form). name: Display name for the workflow. Defaults to ``func.__name__``. description: Optional human-readable description. - checkpoint_storage: Default :class:`CheckpointStorage` for - persisting step results and workflow state. Returns: - A :class:`FunctionalWorkflow` (bare form) or a decorator that - produces one (parameterized form). + A :class:`FunctionalWorkflowDefinition` (bare form) or a decorator + that produces one (parameterized form). Examples: @@ -1311,14 +1355,17 @@ def workflow( # Parameterized form - @workflow(name="my_pipeline", checkpoint_storage=storage) + @workflow(name="my_pipeline") async def pipeline(data: str) -> str: ... + + + instance = pipeline.build(checkpoint_storage=storage) """ if func is not None: - return FunctionalWorkflow(func, name=name, description=description, checkpoint_storage=checkpoint_storage) + return FunctionalWorkflowDefinition(func, name=name, description=description) - def _decorator(fn: Callable[..., Awaitable[Any]]) -> FunctionalWorkflow: - return FunctionalWorkflow(fn, name=name, description=description, checkpoint_storage=checkpoint_storage) + def _decorator(fn: Callable[..., Awaitable[Any]]) -> FunctionalWorkflowDefinition: + return FunctionalWorkflowDefinition(fn, name=name, description=description) return _decorator @@ -1343,6 +1390,12 @@ class FunctionalWorkflowAgent: :class:`WorkflowAgent`), so HITL workflows are callable via this adapter. Callers resume via ``responses=`` / ``checkpoint_id=``. + The wrapped workflow owns mutable execution state. Scope the workflow and + this adapter to one logical caller or session; create separate workflow + instances for independent or mutually untrusted callers. If those + instances use checkpoint storage, the host must also authorize and + tenant-scope access to that external store. + Args: workflow: The :class:`FunctionalWorkflow` to wrap. name: Display name for the agent. Defaults to the workflow name. diff --git a/python/packages/core/tests/workflow/test_functional_workflow.py b/python/packages/core/tests/workflow/test_functional_workflow.py index c4313e4f4..96b7e6edf 100644 --- a/python/packages/core/tests/workflow/test_functional_workflow.py +++ b/python/packages/core/tests/workflow/test_functional_workflow.py @@ -7,17 +7,20 @@ from __future__ import annotations import asyncio import json import logging -from collections.abc import Iterator +from collections.abc import Awaitable, Callable, Iterator from contextlib import contextmanager from dataclasses import dataclass +from typing import Any, overload import pytest from agent_framework import ( AgentResponseUpdate, + CheckpointStorage, ExperimentalFeature, FunctionalWorkflow, FunctionalWorkflowAgent, + FunctionalWorkflowDefinition, InMemoryCheckpointStorage, RunContext, StepWrapper, @@ -37,6 +40,34 @@ from agent_framework._workflows._functional import ( # --------------------------------------------------------------------------- +@overload +def built_workflow(func: Callable[..., Awaitable[Any]]) -> FunctionalWorkflow: ... + + +@overload +def built_workflow( + *, + name: str | None = None, + description: str | None = None, + checkpoint_storage: CheckpointStorage | None = None, +) -> Callable[[Callable[..., Awaitable[Any]]], FunctionalWorkflow]: ... + + +def built_workflow( + func: Callable[..., Awaitable[Any]] | None = None, + *, + name: str | None = None, + description: str | None = None, + checkpoint_storage: CheckpointStorage | None = None, +) -> FunctionalWorkflow | Callable[[Callable[..., Awaitable[Any]]], FunctionalWorkflow]: + """Build a fresh executable workflow for behavior-focused tests.""" + + def decorate(fn: Callable[..., Awaitable[Any]]) -> FunctionalWorkflow: + return workflow(name=name, description=description)(fn).build(checkpoint_storage=checkpoint_storage) + + return decorate(func) if func is not None else decorate + + @step async def add_one(x: int) -> int: return x + 1 @@ -69,7 +100,7 @@ async def failing_step(x: int) -> int: class TestBasicExecution: async def test_simple_sequential_pipeline(self): - @workflow + @built_workflow async def pipeline(x: int) -> int: a = await add_one(x) return await double(a) @@ -80,7 +111,7 @@ class TestBasicExecution: assert outputs == [12] # (5+1)*2 async def test_workflow_with_string_data(self): - @workflow + @built_workflow async def upper_pipeline(text: str) -> str: return await to_upper(text) @@ -88,7 +119,7 @@ class TestBasicExecution: assert result.get_outputs() == ["HELLO"] async def test_workflow_returns_result(self): - @workflow + @built_workflow async def simple(x: int) -> int: return await add_one(x) @@ -96,14 +127,14 @@ class TestBasicExecution: assert result.get_outputs() == [11] async def test_workflow_name_defaults_to_function_name(self): - @workflow + @built_workflow async def my_pipeline(x: int) -> int: return x assert my_pipeline.name == "my_pipeline" async def test_workflow_custom_name(self): - @workflow(name="custom_wf", description="A test workflow") + @built_workflow(name="custom_wf", description="A test workflow") async def wf(x: int) -> int: return x @@ -118,7 +149,7 @@ class TestBasicExecution: class TestEventEmission: async def test_step_events_emitted(self): - @workflow + @built_workflow async def pipeline(x: int) -> int: return await add_one(x) @@ -129,7 +160,7 @@ class TestEventEmission: assert "output" in event_types async def test_step_events_carry_executor_id(self): - @workflow + @built_workflow async def pipeline(x: int) -> int: return await add_one(x) @@ -144,7 +175,7 @@ class TestEventEmission: assert completed_events[0].data == 6 async def test_status_events_in_timeline(self): - @workflow + @built_workflow async def pipeline(x: int) -> int: return x @@ -154,7 +185,7 @@ class TestEventEmission: assert WorkflowRunState.IDLE in states async def test_final_state_is_idle(self): - @workflow + @built_workflow async def pipeline(x: int) -> int: return x @@ -164,7 +195,7 @@ class TestEventEmission: async def test_custom_event(self): from agent_framework import WorkflowEvent - @workflow + @built_workflow async def pipeline(x: int, ctx: RunContext) -> int: await ctx.add_event(WorkflowEvent("intermediate", executor_id="pipeline", data="custom_data")) return x @@ -192,7 +223,7 @@ class TestParallelExecution: await asyncio.sleep(0.01) return x * 2 - @workflow + @built_workflow async def parallel_wf(x: int) -> list[int]: a, b = await asyncio.gather(slow_add(x), slow_double(x)) return [a, b] @@ -210,7 +241,7 @@ class TestParallelExecution: async def task_b(x: int) -> int: return x * 2 - @workflow + @built_workflow async def par_wf(x: int) -> tuple[int, int]: a, b = await asyncio.gather(task_a(x), task_b(x)) return (a, b) @@ -228,8 +259,50 @@ class TestParallelExecution: class TestHITL: - async def test_request_info_interrupts(self): + async def test_workflow_definition_builds_isolated_pending_continuations(self): @workflow + async def review_wf(doc: str, ctx: RunContext) -> str: + feedback = await ctx.request_info(doc, response_type=str) + return f"{doc}:{feedback}" + + assert not hasattr(review_wf, "run") + + caller_a = review_wf.build() + caller_b = review_wf.build() + + caller_a_paused = await caller_a.run("caller-a") + request_id = caller_a_paused.get_request_info_events()[0].request_id + + with pytest.raises(ValueError, match="no pending request_info events"): + await caller_b.run(responses={request_id: "caller-b-response"}) + + caller_a_completed = await caller_a.run(responses={request_id: "caller-a-response"}) + assert caller_a_completed.get_outputs() == ["caller-a:caller-a-response"] + + async def test_build_does_not_inherit_checkpoint_storage_by_default(self): + @workflow + async def review_wf(doc: str) -> str: + return doc + + caller = review_wf.build() + + with pytest.raises(ValueError, match="checkpoint_storage"): + await caller.run(checkpoint_id="missing") + + async def test_build_accepts_tenant_scoped_checkpoint_storage(self): + caller_storage = InMemoryCheckpointStorage() + + @workflow + async def review_wf(doc: str, ctx: RunContext) -> str: + return await ctx.request_info(doc, response_type=str) + + caller = review_wf.build(checkpoint_storage=caller_storage) + await caller.run("caller") + + assert len(await caller_storage.list_checkpoints(workflow_name="review_wf")) == 1 + + async def test_request_info_interrupts(self): + @built_workflow async def review_wf(doc: str, ctx: RunContext) -> str: feedback = await ctx.request_info({"draft": doc}, response_type=str, request_id="req1") return f"Final: {feedback}" @@ -242,7 +315,7 @@ class TestHITL: assert request_events[0].request_id == "req1" async def test_request_info_resume(self): - @workflow + @built_workflow async def review_wf(doc: str, ctx: RunContext) -> str: feedback = await ctx.request_info({"draft": doc}, response_type=str, request_id="req1") return f"Final: {feedback}" @@ -260,7 +333,7 @@ class TestHITL: async def test_fresh_message_while_pending_requests_warns(self, caplog: pytest.LogCaptureFixture) -> None: """A fresh message while request_info events are pending is allowed but logs a warning.""" - @workflow + @built_workflow async def review_wf(doc: str, ctx: RunContext) -> str: feedback = await ctx.request_info({"draft": doc}, response_type=str, request_id="req1") return f"Final: {feedback}" @@ -279,7 +352,7 @@ class TestHITL: async def test_responses_while_pending_requests_does_not_warn(self, caplog: pytest.LogCaptureFixture) -> None: """Delivering responses is the normal completion path and must not warn.""" - @workflow + @built_workflow async def review_wf(doc: str, ctx: RunContext) -> str: feedback = await ctx.request_info({"draft": doc}, response_type=str, request_id="req1") return f"Final: {feedback}" @@ -297,7 +370,7 @@ class TestHITL: async def test_untyped_ctx_parameter(self): """ctx is injected by parameter name even without a RunContext annotation.""" - @workflow # pyright: ignore[reportUnknownArgumentType] + @built_workflow # pyright: ignore[reportUnknownArgumentType] async def review_wf(doc: str, ctx) -> str: # pyright: ignore[reportMissingParameterType, reportUnknownParameterType] feedback: str = await ctx.request_info({"draft": doc}, response_type=str, request_id="req1") # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] return f"Final: {feedback}" @@ -309,7 +382,7 @@ class TestHITL: assert result2.get_outputs() == ["Final: LGTM"] async def test_multiple_sequential_interrupts(self): - @workflow + @built_workflow async def multi_hitl(data: str, ctx: RunContext) -> str: r1 = await ctx.request_info("step1", response_type=str, request_id="r1") r2 = await ctx.request_info("step2", response_type=str, request_id="r2") @@ -330,7 +403,7 @@ class TestHITL: assert result3.get_outputs() == ["A+B"] async def test_request_info_auto_generates_id(self): - @workflow + @built_workflow async def auto_id_wf(x: int, ctx: RunContext) -> None: await ctx.request_info("need data", response_type=str) @@ -347,7 +420,7 @@ class TestHITL: class TestErrorHandling: async def test_step_failure_propagates(self): - @workflow + @built_workflow async def failing_wf(x: int) -> None: await failing_step(x) @@ -355,7 +428,7 @@ class TestErrorHandling: await failing_wf.run(42) async def test_step_failure_emits_executor_failed(self): - @workflow + @built_workflow async def failing_wf(x: int) -> None: await failing_step(x) @@ -371,7 +444,7 @@ class TestErrorHandling: assert failed_events[0].executor_id == "failing_step" async def test_workflow_failure_emits_failed_status(self): - @workflow + @built_workflow async def bad_wf(x: int) -> None: raise RuntimeError("workflow broke") @@ -387,7 +460,7 @@ class TestErrorHandling: assert any(e.state == WorkflowRunState.FAILED for e in status_events) async def test_invalid_params_message_and_responses(self): - @workflow + @built_workflow async def wf(x: int) -> None: pass @@ -395,7 +468,7 @@ class TestErrorHandling: await wf.run("hello", responses={"r1": "val"}) async def test_invalid_params_message_and_checkpoint(self): - @workflow + @built_workflow async def wf(x: int) -> None: pass @@ -403,7 +476,7 @@ class TestErrorHandling: await wf.run("hello", checkpoint_id="abc") async def test_invalid_params_nothing(self): - @workflow + @built_workflow async def wf(x: int) -> None: pass @@ -418,7 +491,7 @@ class TestErrorHandling: class TestStreaming: async def test_streaming_yields_events(self): - @workflow + @built_workflow async def pipeline(x: int) -> int: return await add_one(x) @@ -434,7 +507,7 @@ class TestStreaming: assert "output" in event_types async def test_streaming_final_response(self): - @workflow + @built_workflow async def pipeline(x: int) -> int: return await add_one(x) @@ -446,7 +519,7 @@ class TestStreaming: async def test_streaming_context_reports_streaming(self): streaming_flag = None - @workflow + @built_workflow async def wf(x: int, ctx: RunContext) -> int: nonlocal streaming_flag streaming_flag = ctx.is_streaming() # type: ignore[assignment] @@ -491,7 +564,7 @@ class TestStepPassthrough: class TestStateManagement: async def test_get_set_state(self): - @workflow + @built_workflow async def stateful_wf(x: int, ctx: RunContext) -> int: ctx.set_state("counter", x) return ctx.get_state("counter") @@ -500,7 +573,7 @@ class TestStateManagement: assert result.get_outputs() == [42] async def test_get_state_default(self): - @workflow + @built_workflow async def wf(x: int, ctx: RunContext) -> str: return ctx.get_state("missing", "default_val") @@ -521,7 +594,7 @@ class TestCheckpointing: async def expensive(x: int) -> int: return x * 100 - @workflow(checkpoint_storage=storage) + @built_workflow(checkpoint_storage=storage) async def ckpt_wf(x: int) -> int: return await expensive(x) @@ -539,7 +612,7 @@ class TestCheckpointing: async def compute(x: int) -> int: return x + 1 - @workflow + @built_workflow async def wf(x: int) -> int: return await compute(x) @@ -559,7 +632,7 @@ class TestCheckpointing: call_count += 1 return x + 1 - @workflow(checkpoint_storage=storage) + @built_workflow(checkpoint_storage=storage) async def wf(x: int) -> int: return await counting_task(x) @@ -580,7 +653,7 @@ class TestCheckpointing: async def test_checkpoint_hitl_resume(self): storage = InMemoryCheckpointStorage() - @workflow(checkpoint_storage=storage) + @built_workflow(checkpoint_storage=storage) async def hitl_wf(doc: str, ctx: RunContext) -> str: feedback = await ctx.request_info({"draft": doc}, response_type=str, request_id="req1") return f"Done: {feedback}" @@ -598,7 +671,7 @@ class TestCheckpointing: assert result2.get_outputs() == ["Done: Approved!"] async def test_checkpoint_without_storage_raises(self): - @workflow + @built_workflow async def wf(x: int) -> int: return x @@ -608,7 +681,7 @@ class TestCheckpointing: async def test_checkpoint_preserves_state(self): storage = InMemoryCheckpointStorage() - @workflow(checkpoint_storage=storage) + @built_workflow(checkpoint_storage=storage) async def stateful_wf(x: int, ctx: RunContext) -> str: ctx.set_state("key", "value") feedback = await ctx.request_info("need info", response_type=str, request_id="r1") @@ -648,7 +721,7 @@ class TestCheckpointing: raise RuntimeError("simulated crash") return x * 2 - @workflow(checkpoint_storage=storage) + @built_workflow(checkpoint_storage=storage) async def crash_wf(x: int) -> int: a = await slow_step1(x) return await crashing_step2(a) @@ -687,7 +760,7 @@ class TestCheckpointing: async def s3(x: int) -> int: return x + 3 - @workflow(checkpoint_storage=storage) + @built_workflow(checkpoint_storage=storage) async def multi_step_wf(x: int) -> int: a = await s1(x) b = await s2(a) @@ -708,7 +781,7 @@ class TestCheckpointing: async def compute(x: int) -> int: return x + 1 - @workflow(checkpoint_storage=storage) + @built_workflow(checkpoint_storage=storage) async def wf(x: int) -> int: return await compute(x) @@ -748,7 +821,7 @@ class TestControlFlow: async def quarantine(text: str) -> str: return f"quarantined: {text}" - @workflow + @built_workflow async def email_pipeline(email: str) -> str: cl = await classify(email) if cl.is_spam: @@ -775,7 +848,7 @@ class TestNestedWorkflows: async def step_a(x: int) -> int: return x + 1 - @workflow + @built_workflow async def inner_wf(x: int) -> int: return await step_a(x) @@ -784,7 +857,7 @@ class TestNestedWorkflows: result = await inner_wf.run(x) return result.get_outputs()[0] - @workflow + @built_workflow async def outer_wf(x: int) -> int: return await call_inner(x) @@ -799,7 +872,7 @@ class TestNestedWorkflows: class TestAsAgent: async def test_as_agent_returns_agent(self): - @workflow + @built_workflow async def wf(x: int) -> str: return f"result: {x}" @@ -807,7 +880,7 @@ class TestAsAgent: assert agent.name == "wf" async def test_as_agent_custom_name(self): - @workflow + @built_workflow async def wf(x: int) -> int: return x @@ -815,7 +888,7 @@ class TestAsAgent: assert agent.name == "my_agent" async def test_as_agent_run(self): - @workflow + @built_workflow async def wf(x: int) -> int: return await add_one(x) @@ -824,7 +897,7 @@ class TestAsAgent: assert response.text == "11" async def test_as_agent_run_streaming(self): - @workflow + @built_workflow async def wf(x: int) -> str: return f"result: {x}" @@ -840,7 +913,7 @@ class TestAsAgent: assert len(response.messages) >= 1 async def test_as_agent_has_id_and_description(self): - @workflow(description="A test workflow") + @built_workflow(description="A test workflow") async def wf(x: int) -> int: return x @@ -856,7 +929,7 @@ class TestAsAgent: class TestConcurrencyGuard: async def test_concurrent_run_raises(self): - @workflow + @built_workflow async def slow_wf(x: int) -> int: await asyncio.sleep(0.1) return x @@ -872,7 +945,7 @@ class TestConcurrencyGuard: await stream.get_final_response() async def test_run_after_completion(self): - @workflow + @built_workflow async def wf(x: int) -> int: return x @@ -911,7 +984,7 @@ class TestDecoratorForms: async def my_wf(x: int) -> None: pass - assert isinstance(my_wf, FunctionalWorkflow) + assert isinstance(my_wf, FunctionalWorkflowDefinition) assert my_wf.name == "my_wf" def test_workflow_with_params(self): @@ -919,7 +992,7 @@ class TestDecoratorForms: async def my_wf(x: int) -> None: pass - assert isinstance(my_wf, FunctionalWorkflow) + assert isinstance(my_wf, FunctionalWorkflowDefinition) assert my_wf.name == "custom" assert my_wf.description == "desc" @@ -931,7 +1004,7 @@ class TestDecoratorForms: class TestIncludeStatusEvents: async def test_status_events_excluded_by_default(self): - @workflow + @built_workflow async def wf(x: int) -> int: return x @@ -940,7 +1013,7 @@ class TestIncludeStatusEvents: assert len(status_in_list) == 0 async def test_status_events_included_when_requested(self): - @workflow + @built_workflow async def wf(x: int) -> int: return x @@ -956,7 +1029,7 @@ class TestIncludeStatusEvents: class TestEdgeCases: async def test_workflow_with_no_tasks(self): - @workflow + @built_workflow async def no_tasks(x: int) -> int: return x * 2 @@ -964,7 +1037,7 @@ class TestEdgeCases: assert result.get_outputs() == [10] async def test_workflow_with_no_output(self): - @workflow + @built_workflow async def silent_wf(x: int) -> None: pass # returns None — no output emitted @@ -974,7 +1047,7 @@ class TestEdgeCases: async def test_return_value_auto_yields_output(self): """Returning a non-None value automatically emits it as an output.""" - @workflow + @built_workflow async def wf(x: int) -> int: return x * 3 @@ -982,7 +1055,7 @@ class TestEdgeCases: assert result.get_outputs() == [15] async def test_step_called_multiple_times(self): - @workflow + @built_workflow async def wf(x: int) -> int: a = await add_one(x) b = await add_one(a) @@ -1005,7 +1078,7 @@ class TestEdgeCases: class TestRecoveryAfterErrors: async def test_run_after_failure_is_allowed(self): - @workflow + @built_workflow async def wf(x: int) -> int: if x == 1: raise RuntimeError("boom") @@ -1036,7 +1109,7 @@ class TestWorkflowInterruptedIsBaseException: """User code with ``except Exception`` should not catch WorkflowInterrupted.""" caught = False - @workflow + @built_workflow async def wf(x: int, ctx: RunContext) -> str: nonlocal caught try: @@ -1064,7 +1137,7 @@ class TestCheckpointValidation: storage = InMemoryCheckpointStorage() - @workflow(name="my_wf", checkpoint_storage=storage) + @built_workflow(name="my_wf", checkpoint_storage=storage) async def wf(x: int) -> int: return x @@ -1108,7 +1181,7 @@ class TestExecutorBypassed: call_count += 1 return x + 1 - @workflow(checkpoint_storage=storage) + @built_workflow(checkpoint_storage=storage) async def wf(x: int) -> int: return await tracked(x) @@ -1141,7 +1214,7 @@ class TestExecutorBypassed: async def compute(x: int) -> int: return x * 10 - @workflow(checkpoint_storage=storage) + @built_workflow(checkpoint_storage=storage) async def wf(x: int) -> int: return await compute(x) @@ -1169,7 +1242,7 @@ class TestRequestInfoInStep: feedback = await ctx.request_info({"draft": doc}, response_type=str, request_id="s1") return f"reviewed: {feedback}" - @workflow + @built_workflow async def wf(doc: str) -> str: return await review_step(doc) @@ -1205,7 +1278,7 @@ class TestRequestInfoInStep: ctx.set_state("seen", data) return f"{data}:{ctx.get_state('seen')}" - @workflow + @built_workflow async def wf(data: str) -> str: return await needs_ctx_first(data) @@ -1225,7 +1298,7 @@ class TestRequestInfoInStep: captured_ctx = get_run_context() # type: ignore[assignment] return x - @workflow + @built_workflow async def wf(x: int) -> int: return await capture_ctx(x) @@ -1249,7 +1322,7 @@ class TestNoneResponseHandling: async def test_none_response_logs_warning(self): """Providing None as a response value should log a warning.""" - @workflow + @built_workflow async def wf(doc: str, ctx: RunContext) -> str: val = await ctx.request_info("need input", response_type=str, request_id="r1") return f"got: {val}" @@ -1267,7 +1340,7 @@ class TestNoneResponseHandling: async def test_none_response_is_returned(self): """None is a valid (if discouraged) response value.""" - @workflow + @built_workflow async def wf(x: int, ctx: RunContext) -> str: val = await ctx.request_info("need data", response_type=str, request_id="r1") return f"value={val}" @@ -1322,7 +1395,7 @@ class TestHITLInStepWithCaching: feedback = await ctx.request_info({"val": val}, response_type=str, request_id="r1") return f"{val}:{feedback}" - @workflow + @built_workflow async def wf(x: int) -> str: a = await step_a(x) return await step_b(a) @@ -1353,7 +1426,7 @@ class TestHITLInStepWithCaching: feedback = await ctx.request_info({"val": val}, response_type=str, request_id="rev") return f"reviewed({val}):{feedback}" - @workflow(checkpoint_storage=storage) + @built_workflow(checkpoint_storage=storage) async def wf(x: int) -> str: v = await compute(x) return await review(v) @@ -1384,7 +1457,7 @@ class TestHITLInStepWithCaching: val = await ctx.request_info({"doc": doc}, response_type=str, request_id="r1") return f"got:{val}" - @workflow + @built_workflow async def wf(doc: str) -> str: return await needs_feedback(doc) @@ -1403,7 +1476,7 @@ class TestHITLInStepWithCaching: async def hitl_step(x: int, ctx: RunContext) -> str: return await ctx.request_info("need data", response_type=str, request_id="r1") - @workflow + @built_workflow async def wf(x: int) -> str: return await hitl_step(x) @@ -1422,7 +1495,7 @@ class TestDeterministicAutoRequestId: """Regression for bug_001: auto-generated request_info ids must be stable across replay.""" async def test_auto_request_id_roundtrips_on_resume(self): - @workflow + @built_workflow async def wf(x: int, ctx: RunContext) -> str: # No request_id — framework must generate a deterministic one val = await ctx.request_info("need data", response_type=str) @@ -1441,7 +1514,7 @@ class TestDeterministicAutoRequestId: assert result2.get_outputs() == ["got:hello"] async def test_multiple_auto_ids_are_distinct_and_stable(self): - @workflow + @built_workflow async def wf(x: int, ctx: RunContext) -> str: a = await ctx.request_info("first", response_type=str) b = await ctx.request_info("second", response_type=str) @@ -1468,7 +1541,7 @@ class TestDeterministicAutoRequestId: async def second_review(value: int, ctx: RunContext) -> str: return await ctx.request_info({"step": "second", "value": value}, response_type=str) - @workflow + @built_workflow async def wf(value: int) -> str: first = await first_review(value) second = await second_review(value) @@ -1495,7 +1568,7 @@ class TestPendingRequestsPruned: async def test_final_checkpoint_no_longer_claims_resolved_requests_pending(self): storage = InMemoryCheckpointStorage() - @workflow(checkpoint_storage=storage) + @built_workflow(checkpoint_storage=storage) async def wf(x: int, ctx: RunContext) -> str: a = await ctx.request_info("q1", response_type=str, request_id="r1") b = await ctx.request_info("q2", response_type=str, request_id="r2") @@ -1523,7 +1596,7 @@ class TestArityValidation: return f"{a}+{b}" async def test_ctx_only_workflow_with_message_raises_clear_error(self): - @workflow + @built_workflow async def wf(ctx: RunContext) -> str: return "no message used" @@ -1535,7 +1608,7 @@ class TestArityValidation: # message-receiving parameter. (Running it without a message still # requires providing responses or a checkpoint_id — that's # _validate_run_params's job, not ours.) - @workflow + @built_workflow async def wf(ctx: RunContext) -> str: return "ok" @@ -1546,7 +1619,7 @@ class TestStaleResponsesRejected: """Regression for bug_014: stale responses after clean completion must be rejected.""" async def test_responses_after_clean_completion_raise(self): - @workflow + @built_workflow async def wf(x: int) -> int: return x * 2 @@ -1555,7 +1628,7 @@ class TestStaleResponsesRejected: await wf.run(responses={"stale": "x"}) async def test_responses_mismatched_key_raises(self): - @workflow + @built_workflow async def wf(x: int, ctx: RunContext) -> str: return await ctx.request_info("q", response_type=str, request_id="r1") @@ -1568,7 +1641,7 @@ class TestReservedStateKeys: """Regression for bug_017: set_state must reject underscore-prefixed keys.""" async def test_underscore_key_rejected(self): - @workflow + @built_workflow async def wf(x: int, ctx: RunContext) -> int: ctx.set_state("_private", "user value") return x @@ -1577,7 +1650,7 @@ class TestReservedStateKeys: await wf.run(1) async def test_normal_key_still_works(self): - @workflow + @built_workflow async def wf(x: int, ctx: RunContext) -> int: ctx.set_state("normal_key", "v") assert ctx.get_state("normal_key") == "v" @@ -1597,7 +1670,7 @@ class TestDeepcopyOnCacheHit: async def takes_lock(lock: threading.Lock, n: int) -> int: return n + 1 - @workflow + @built_workflow async def wf(x: int) -> int: lock = threading.Lock() return await takes_lock(lock, x) @@ -1613,11 +1686,11 @@ class TestStepDiscoveryAttributeAccess: """Regression for bug_008: checkpoint hash must differ when function body changes.""" async def test_signature_hash_changes_when_function_body_changes(self): - @workflow + @built_workflow async def wf_a(x: int) -> int: return x + 1 - @workflow(name="wf_b") + @built_workflow(name="wf_b") async def wf_b(x: int) -> int: return x * 100 @@ -1630,7 +1703,7 @@ class TestAsAgentSignatureParity: """Regression for bug_015: as_agent signature must accept description/context_providers.""" async def test_as_agent_accepts_description_override(self): - @workflow(description="workflow level") + @built_workflow(description="workflow level") async def wf(x: str) -> str: return x.upper() @@ -1638,7 +1711,7 @@ class TestAsAgentSignatureParity: assert agent.description == "agent level" async def test_as_agent_accepts_context_providers_kwarg(self): - @workflow + @built_workflow async def wf(x: str) -> str: return x @@ -1647,7 +1720,7 @@ class TestAsAgentSignatureParity: assert list(agent.context_providers or []) == providers async def test_as_agent_description_defaults_to_workflow_description(self): - @workflow(description="from workflow") + @built_workflow(description="from workflow") async def wf(x: str) -> str: return x @@ -1659,7 +1732,7 @@ class TestFunctionalWorkflowAgentHITL: """Regression for bug_013: .as_agent() must surface request_info events.""" async def test_request_info_surfaces_as_function_approval_request(self): - @workflow + @built_workflow async def wf(x: str, ctx: RunContext) -> str: answer = await ctx.request_info({"need": x}, response_type=str, request_id="rid-1") return f"got:{answer}" @@ -1686,7 +1759,7 @@ class TestFunctionalWorkflowAgentHITL: target_agent: str reason: str - @workflow + @built_workflow async def wf(x: str, ctx: RunContext) -> str: answer = await ctx.request_info( HandoffRequest(target_agent=x, reason="overflow"), @@ -1712,7 +1785,7 @@ class TestFunctionalWorkflowAgentHITL: assert json.loads(json.dumps(function_call_arguments)) == function_call_arguments async def test_resume_via_agent_responses_kwarg(self): - @workflow + @built_workflow async def wf(x: str, ctx: RunContext) -> str: answer = await ctx.request_info(x, response_type=str, request_id="rid-1") return f"got:{answer}" @@ -1751,6 +1824,7 @@ class TestFunctionalWorkflowExperimentalStage: StepWrapper, step, FunctionalWorkflow, + FunctionalWorkflowDefinition, workflow, FunctionalWorkflowAgent, ] diff --git a/python/samples/01-get-started/05_functional_workflow_with_agents.py b/python/samples/01-get-started/05_functional_workflow_with_agents.py index 9aff97826..e77c1bc74 100644 --- a/python/samples/01-get-started/05_functional_workflow_with_agents.py +++ b/python/samples/01-get-started/05_functional_workflow_with_agents.py @@ -47,7 +47,8 @@ async def poem_workflow(topic: str) -> str: async def main() -> None: - result = await poem_workflow.run("a cat learning to code") + workflow_instance = poem_workflow.build() + result = await workflow_instance.run("a cat learning to code") print(result.get_outputs()[0]) diff --git a/python/samples/01-get-started/06_functional_workflow_basics.py b/python/samples/01-get-started/06_functional_workflow_basics.py index a679e70a9..ec6fd79f5 100644 --- a/python/samples/01-get-started/06_functional_workflow_basics.py +++ b/python/samples/01-get-started/06_functional_workflow_basics.py @@ -43,7 +43,8 @@ async def text_workflow(text: str) -> str: async def main() -> None: # - result = await text_workflow.run("hello world") + workflow_instance = text_workflow.build() + result = await workflow_instance.run("hello world") print(f"Output: {result.get_outputs()}") print(f"Final state: {result.get_final_state()}") # diff --git a/python/samples/03-workflows/functional/agent_integration.py b/python/samples/03-workflows/functional/agent_integration.py index b5911cb69..23699e271 100644 --- a/python/samples/03-workflows/functional/agent_integration.py +++ b/python/samples/03-workflows/functional/agent_integration.py @@ -94,12 +94,15 @@ async def cached_pipeline(document: str) -> str: async def main(): + simple_workflow = simple_pipeline.build() + cached_workflow = cached_pipeline.build() + # Simple version — agents called inline - result = await simple_pipeline.run("This is a technical document about machine learning...") + result = await simple_workflow.run("This is a technical document about machine learning...") print(result.get_outputs()[0]) # Cached version — same result, but steps won't re-execute on resume - result = await cached_pipeline.run("This is a technical document about machine learning...") + result = await cached_workflow.run("This is a technical document about machine learning...") print(f"\nCached: {result.get_outputs()[0]}") diff --git a/python/samples/03-workflows/functional/basic_pipeline.py b/python/samples/03-workflows/functional/basic_pipeline.py index 81514da53..eb32b54c6 100644 --- a/python/samples/03-workflows/functional/basic_pipeline.py +++ b/python/samples/03-workflows/functional/basic_pipeline.py @@ -23,8 +23,8 @@ async def transform_data(data: dict[str, str | int]) -> str: return f"[{data['status']}] {data['content']}" -# @workflow turns this async function into a FunctionalWorkflow object. -# Without it, this is just a normal async function. With it, you get: +# @workflow turns this async function into a stateless workflow definition. +# Call .build() to create a stateful FunctionalWorkflow with: # - .run() that returns a WorkflowRunResult with events and outputs # - .run(stream=True) for streaming events in real time # - .as_agent() to use this workflow anywhere an agent is expected @@ -48,8 +48,8 @@ async def data_pipeline(url: str) -> str: async def main(): - # .run() is provided by @workflow — a plain async function wouldn't have it - result = await data_pipeline.run("https://example.com/api/data") + workflow_instance = data_pipeline.build() + result = await workflow_instance.run("https://example.com/api/data") print("Output:", result.get_outputs()[0]) print("State:", result.get_final_state()) diff --git a/python/samples/03-workflows/functional/basic_streaming_pipeline.py b/python/samples/03-workflows/functional/basic_streaming_pipeline.py index 4ee61da60..a986b97ee 100644 --- a/python/samples/03-workflows/functional/basic_streaming_pipeline.py +++ b/python/samples/03-workflows/functional/basic_streaming_pipeline.py @@ -26,9 +26,9 @@ async def validate_result(summary: str) -> bool: return len(summary) > 0 and "[200]" in summary -# @workflow enables .run(stream=True), which returns a ResponseStream -# you can iterate over with `async for`. Without @workflow, you'd just -# have a normal async function with no streaming capability. +# @workflow creates a definition. Its built workflow enables +# .run(stream=True), which returns a ResponseStream you can iterate over with +# `async for`. @workflow async def data_pipeline(url: str) -> str: """A simple sequential data pipeline.""" @@ -40,10 +40,12 @@ async def data_pipeline(url: str) -> str: async def main(): + workflow_instance = data_pipeline.build() + # run(stream=True) returns a ResponseStream that yields events as they # are produced. The raw stream includes lifecycle events (started, status) # alongside application events — filter by event.type to find what you need. - stream = data_pipeline.run("https://example.com/api/data", stream=True) + stream = workflow_instance.run("https://example.com/api/data", stream=True) async for event in stream: if event.type == "output": print(f"Output: {event.data}") diff --git a/python/samples/03-workflows/functional/hitl_review.py b/python/samples/03-workflows/functional/hitl_review.py index 39f2dae88..a9343978a 100644 --- a/python/samples/03-workflows/functional/hitl_review.py +++ b/python/samples/03-workflows/functional/hitl_review.py @@ -59,14 +59,17 @@ async def review_pipeline(topic: str, ctx: RunContext) -> str: async def main(): + workflow_instance = review_pipeline.build() + # Phase 1: Run until the workflow pauses for human input print("=== Phase 1: Initial run ===") - result1 = await review_pipeline.run("AI Safety") + result1 = await workflow_instance.run("AI Safety") # If request_info() was reached, the state is IDLE_WITH_PENDING_REQUESTS. # If the workflow completed without hitting request_info(), it would be IDLE. print(f"State: {(final_state := result1.get_final_state())}") - assert final_state == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS + if final_state != WorkflowRunState.IDLE_WITH_PENDING_REQUESTS: + raise RuntimeError(f"Expected pending review input, but workflow entered {final_state}.") requests = result1.get_request_info_events() print(f"Pending request: {requests[0].request_id}") @@ -74,7 +77,7 @@ async def main(): # Phase 2: Resume with the human's response print("\n=== Phase 2: Resume with feedback ===") print("(write_draft should NOT execute again — saved by @step)") - result2 = await review_pipeline.run(responses={"review_request": "Add more details about alignment research"}) + result2 = await workflow_instance.run(responses={"review_request": "Add more details about alignment research"}) print(f"State: {result2.get_final_state()}") print(f"Output: {result2.get_outputs()[0]}") diff --git a/python/samples/03-workflows/functional/naive_group_chat.py b/python/samples/03-workflows/functional/naive_group_chat.py index 45a326604..255b185fb 100644 --- a/python/samples/03-workflows/functional/naive_group_chat.py +++ b/python/samples/03-workflows/functional/naive_group_chat.py @@ -74,7 +74,8 @@ async def group_chat(question: str) -> str: async def main(): - result = await group_chat.run("What's the difference between a list and a tuple in Python?") + workflow_instance = group_chat.build() + result = await workflow_instance.run("What's the difference between a list and a tuple in Python?") print(result.get_outputs()[0]) diff --git a/python/samples/03-workflows/functional/parallel_pipeline.py b/python/samples/03-workflows/functional/parallel_pipeline.py index 21fd8dcea..87a695bf8 100644 --- a/python/samples/03-workflows/functional/parallel_pipeline.py +++ b/python/samples/03-workflows/functional/parallel_pipeline.py @@ -36,9 +36,9 @@ async def synthesize(sources: list[str]) -> str: return "Research Summary:\n" + "\n".join(f" - {s}" for s in sources) -# @workflow wraps the orchestration logic so you get .run(), streaming, -# and events. The functions it calls are plain Python — no decorators -# needed just because they're inside a workflow. +# @workflow defines the orchestration logic. Build the definition to get a +# stateful workflow with .run(), streaming, and events. The functions it calls +# are plain Python — no decorators needed just because they're inside a workflow. @workflow async def research_pipeline(topic: str) -> str: """Fan-out to three research tasks, then synthesize results.""" @@ -58,7 +58,8 @@ async def research_pipeline(topic: str) -> str: async def main(): - result = await research_pipeline.run("AI agents") + workflow_instance = research_pipeline.build() + result = await workflow_instance.run("AI agents") print(result.get_outputs()[0]) diff --git a/python/samples/03-workflows/functional/steps_and_checkpointing.py b/python/samples/03-workflows/functional/steps_and_checkpointing.py index 93a9423df..a41649184 100644 --- a/python/samples/03-workflows/functional/steps_and_checkpointing.py +++ b/python/samples/03-workflows/functional/steps_and_checkpointing.py @@ -55,9 +55,9 @@ async def validate_result(summary: str) -> bool: storage = InMemoryCheckpointStorage() -# checkpoint_storage tells @workflow where to persist step results. +# Build with checkpoint_storage to persist step results. # Each @step saves a checkpoint after it completes. -@workflow(checkpoint_storage=storage) +@workflow async def data_pipeline(url: str) -> str: """Mix of @step functions and plain functions.""" raw = await fetch_data(url) @@ -68,9 +68,11 @@ async def data_pipeline(url: str) -> str: async def main(): + workflow_instance = data_pipeline.build(checkpoint_storage=storage) + # --- Run 1: Everything executes normally --- print("=== Run 1: Fresh execution ===") - result = await data_pipeline.run("https://example.com/api/data") + result = await workflow_instance.run("https://example.com/api/data") print(f"Output: {result.get_outputs()[0]}") print(f"fetch_calls={fetch_calls}, transform_calls={transform_calls}") @@ -85,9 +87,10 @@ async def main(): # Only validate_result() (no @step) actually runs again. print("\n=== Run 2: Restored from checkpoint ===") latest = await storage.get_latest(workflow_name="data_pipeline") - assert latest is not None + if latest is None: + raise RuntimeError("Expected a checkpoint from the first run.") - result2 = await data_pipeline.run(checkpoint_id=latest.checkpoint_id) + result2 = await workflow_instance.run(checkpoint_id=latest.checkpoint_id) print(f"Output: {result2.get_outputs()[0]}") print(f"fetch_calls={fetch_calls}, transform_calls={transform_calls}") print("(call counts unchanged — @step results were restored from checkpoint)")