add configurable eager mode

Author: jlowin

requirements-dev.lock

@@ -107,12 +107,8 @@ distlib==0.3.8
     # via virtualenv
 distro==1.9.0
     # via openai
-dnspython==2.6.1
-    # via email-validator
 docker==6.1.3
     # via prefect
-email-validator==2.1.1
-    # via pydantic
 envier==0.5.1
     # via ddtrace
 execnet==2.1.1
@@ -160,7 +156,6 @@ identify==2.5.35
     # via pre-commit
 idna==3.6
     # via anyio
-    # via email-validator
     # via httpx
     # via requests
 importlib-metadata==7.0.0
@@ -220,7 +215,7 @@ markupsafe==2.1.5
     # via mkdocs-autorefs
     # via mkdocstrings
     # via werkzeug
-marvin @ git+https://github.com/prefecthq/marvin@aedfb9573e5f0844a1dc1b45b59d279524b0d363
+marvin @ git+https://github.com/prefecthq/marvin@fbfa2e6d2f8f65d611f3519941966e7fc382a880
     # via controlflow
 matplotlib-inline==0.1.6
     # via ipython
@@ -301,7 +296,7 @@ pluggy==1.4.0
     # via pytest
 pre-commit==3.7.0
     # via prefect
-prefect @ git+https://github.com/prefecthq/prefect@8454a1c8938bdd1204eefbdb3606129e37501653
+prefect @ git+https://github.com/prefecthq/prefect@cae0efd9d667ca8b2003e061c5041619a4c5a881
     # via controlflow
 prompt-toolkit==3.0.43
     # via ipython
@@ -449,6 +444,7 @@ s3transfer==0.10.1
 setuptools==69.2.0
     # via ddtrace
     # via nodeenv
+    # via prefect
     # via readchar
 shellingham==1.5.4
     # via typer-slim

requirements.lock

@@ -107,12 +107,8 @@ distlib==0.3.8
     # via virtualenv
 distro==1.9.0
     # via openai
-dnspython==2.6.1
-    # via email-validator
 docker==6.1.3
     # via prefect
-email-validator==2.1.1
-    # via pydantic
 envier==0.5.1
     # via ddtrace
 execnet==2.1.1
@@ -160,7 +156,6 @@ identify==2.5.36
     # via pre-commit
 idna==3.6
     # via anyio
-    # via email-validator
     # via httpx
     # via requests
 importlib-metadata==7.0.0
@@ -220,7 +215,7 @@ markupsafe==2.1.5
     # via mkdocs-autorefs
     # via mkdocstrings
     # via werkzeug
-marvin @ git+https://github.com/prefecthq/marvin@aedfb9573e5f0844a1dc1b45b59d279524b0d363
+marvin @ git+https://github.com/prefecthq/marvin@fbfa2e6d2f8f65d611f3519941966e7fc382a880
     # via controlflow
 matplotlib-inline==0.1.7
     # via ipython
@@ -301,7 +296,7 @@ pluggy==1.5.0
     # via pytest
 pre-commit==3.7.1
     # via prefect
-prefect @ git+https://github.com/prefecthq/prefect@8454a1c8938bdd1204eefbdb3606129e37501653
+prefect @ git+https://github.com/prefecthq/prefect@cae0efd9d667ca8b2003e061c5041619a4c5a881
     # via controlflow
 prompt-toolkit==3.0.43
     # via ipython
@@ -449,6 +444,7 @@ s3transfer==0.10.1
 setuptools==69.2.0
     # via ddtrace
     # via nodeenv
+    # via prefect
     # via readchar
 shellingham==1.5.4
     # via typer-slim

src/controlflow/decorators.py

@@ -24,7 +24,7 @@ def flow(
     instructions: str = None,
     tools: list[ToolType] = None,
     agents: list["Agent"] = None,
-    resolve_results: bool = None,
+    eager: bool = None,
 ):
     """
     A decorator that wraps a function as a ControlFlow flow.
@@ -43,7 +43,7 @@ def flow(
         instructions (str, optional): Instructions for the flow. Defaults to None.
         tools (list[ToolType], optional): List of tools to be used in the flow. Defaults to None.
         agents (list[Agent], optional): List of agents to be used in the flow. Defaults to None.
-        resolve_results (bool, optional): Whether to resolve the results of tasks. Defaults to True.
+        eager (bool, optional): Whether the flow should be run eagerly. Defaults to the global setting (True)
 
     Returns:
         callable: The wrapped function or a new flow decorator if `fn` is not provided.
@@ -57,11 +57,12 @@ def flow(
             instructions=instructions,
             tools=tools,
             agents=agents,
-            resolve_results=resolve_results,
+            eager=eager,
         )
 
-    if resolve_results is None:
-        resolve_results = True
+    if eager is None:
+        eager = controlflow.settings.eager_mode
+
     sig = inspect.signature(fn)
 
     @functools.wraps(fn)
@@ -92,20 +93,22 @@ def wrapper(
 
         # create a function to wrap as a Prefect flow
         @prefect.flow
-        def wrapped_flow(*args, **kwargs):
+        def wrapped_flow(*args, eager_=None, **kwargs):
             with flow_obj, patch_marvin():
                 with controlflow.instructions(instructions):
                     result = fn(*args, **kwargs)
 
-                    if resolve_results:
+                    # allow runtime override of eager mode
+                    eager_mode = eager_ if eager_ is not None else eager
+                    if eager_mode:
                         # resolve any returned tasks; this will raise on failure
                         result = resolve_tasks(result)
 
-                    # run all tasks in the flow to completion
-                    Controller(
-                        flow=flow_obj,
-                        tasks=list(flow_obj._tasks.values()),
-                    ).run()
+                        # run all tasks in the flow to completion
+                        Controller(
+                            flow=flow_obj,
+                            tasks=list(flow_obj._tasks.values()),
+                        ).run()
 
                 return result
 
@@ -126,13 +129,16 @@ def task(
     agents: list["Agent"] = None,
     tools: list[ToolType] = None,
     user_access: bool = None,
+    eager: bool = None,
 ):
     """
     A decorator that turns a Python function into a Task. The Task objective is
     set to the function name, and the instructions are set to the function
-    docstring. When the function is called, the arguments are provided to the
-    task as context, and the task is run to completion. If successful, the task
-    result is returned; if failed, an error is raised.
+    docstring. When the function is called in eager mode, the arguments are
+    provided to the task as context, and the task is run to completion. If
+    successful, the task result is returned; if failed, an error is raised. When
+    the function is called with eager mode set to False, a Task object is
+    returned which can be run later.
 
     Args:
         fn (callable, optional): The function to be wrapped as a task. If not provided,
@@ -145,6 +151,7 @@ def task(
         tools (list[ToolType], optional): List of tools to be used in the task. Defaults to None.
         user_access (bool, optional): Whether the task requires user access. Defaults to None,
             in which case it is set to False.
+        eager (bool, optional): Whether the task should be run eagerly. Defaults to the global setting (True)
 
     Returns:
         callable: The wrapped function or a new task decorator if `fn` is not provided.
@@ -158,6 +165,7 @@ def task(
             agents=agents,
             tools=tools,
             user_access=user_access,
+            eager=eager,
         )
 
     sig = inspect.signature(fn)
@@ -168,8 +176,13 @@ def task(
     if instructions is None:
         instructions = fn.__doc__
 
+    if eager is None:
+        eager = controlflow.settings.eager_mode
+
+    result_type = fn.__annotations__.get("return")
+
     @functools.wraps(fn)
-    def wrapper(*args, **kwargs):
+    def wrapper(*args, eager_: bool = None, **kwargs):
         # first process callargs
         bound = sig.bind(*args, **kwargs)
         bound.apply_defaults()
@@ -179,12 +192,20 @@ def wrapper(*args, **kwargs):
             instructions=instructions,
             agents=agents,
             context=bound.arguments,
-            result_type=fn.__annotations__.get("return"),
+            result_type=result_type,
             user_access=user_access or False,
             tools=tools or [],
         )
 
-        task.run()
-        return task.result
+        # allow runtime override of eager mode
+        eager_mode = eager_ if eager_ is not None else eager
+        if eager_mode:
+            task.run()
+            return task.result
+        else:
+            return task
+
+    if eager is False:
+        wrapper.__annotations__["return"] = Task
 
     return wrapper

src/controlflow/settings.py

@@ -60,6 +60,11 @@ class Settings(ControlFlowSettings):
         description="If True, a global flow is created for convenience, so users don't have to wrap every invocation in a flow function. Disable to avoid accidentally sharing context between agents.",
     )
     openai_api_key: Optional[str] = Field(None, validate_assignment=True)
+    eager_mode: bool = Field(
+        True,
+        description="If True, @task- and @flow-decorated functions are run immediately. "
+        "This can be set on a per-task or per-flow basis using the `eager` argument.",
+    )
 
     def __init__(self, **data):
         super().__init__(**data)

tests/test_decorators.py

@@ -1,6 +1,8 @@
+import controlflow
 import pytest
 from controlflow import Task
-from controlflow.decorators import flow
+from controlflow.decorators import flow, task
+from controlflow.settings import temporary_settings
 
 
 @pytest.mark.usefixtures("mock_controller")
@@ -52,3 +54,109 @@ def test_flow():
 
         result = test_flow()
         assert result == "hello"
+
+
+class TestTaskDecorator:
+    pass
+
+
+@pytest.mark.usefixtures("mock_controller")
+class TestTaskEagerMode:
+    def test_eager_mode_enabled_by_default(self):
+        assert controlflow.settings.eager_mode is True
+
+    def test_task_eager_mode(self, mock_controller_run_agent):
+        @task
+        def return_42() -> int:
+            """Return the number 42"""
+            pass
+
+        return_42()
+        assert mock_controller_run_agent.call_count == 1
+
+    def test_task_eager_mode_off(self, mock_controller_run_agent):
+        @task(eager=False)
+        def return_42() -> int:
+            """Return the number 42"""
+            pass
+
+        result = return_42()
+        assert mock_controller_run_agent.call_count == 0
+        assert isinstance(result, Task)
+        assert result.objective == "return_42"
+        assert result.result_type == int
+        assert result.instructions == "Return the number 42"
+
+    def test_task_eager_mode_loads_default(self, mock_controller_run_agent):
+        with temporary_settings(eager_mode=False):
+
+            @task
+            def return_42() -> int:
+                """Return the number 42"""
+                pass
+
+        result = return_42()
+        assert mock_controller_run_agent.call_count == 0
+        assert isinstance(result, Task)
+        assert result.objective == "return_42"
+        assert result.result_type == int
+        assert result.instructions == "Return the number 42"
+
+    @pytest.mark.parametrize("eager_mode", [True, False])
+    def test_override_eager_mode_at_call_time(
+        self, mock_controller_run_agent, eager_mode
+    ):
+        with temporary_settings(eager_mode=eager_mode):
+
+            @task
+            def return_42() -> int:
+                """Return the number 42"""
+                pass
+
+        return_42(eager_=not eager_mode)
+        if eager_mode:
+            assert mock_controller_run_agent.call_count == 0
+        else:
+            assert mock_controller_run_agent.call_count == 1
+
+
+@pytest.mark.usefixtures("mock_controller")
+class TestFlowEagerMode:
+    def test_flow_eager_mode(self, mock_controller_run_agent):
+        @flow
+        def test_flow():
+            task = Task("say hello", result="hello")
+            return task
+
+        result = test_flow()
+        assert mock_controller_run_agent.call_count == 1
+        assert result == "hello"
+
+    def test_flow_eager_mode_off(self, mock_controller_run_agent):
+        @flow(eager=False)
+        def test_flow():
+            task = Task("say hello", result="hello")
+            return task
+
+        result = test_flow()
+        assert mock_controller_run_agent.call_count == 0
+        assert isinstance(result, Task)
+        assert result.objective == "say hello"
+        assert result.result == "hello"
+
+    def test_flow_eager_mode_off_doesnt_affect_tasks_with_eager_mode_on(
+        self, mock_controller_run_agent
+    ):
+        @task(eager=True)
+        def return_42() -> int:
+            """Return the number 42"""
+            pass
+
+        @flow(eager=False)
+        def test_flow():
+            result = return_42()
+            return result
+
+        result = test_flow()
+        assert mock_controller_run_agent.call_count == 1
+        assert not isinstance(result, Task)