Flow updates: prevent args from becoming part of context

docs/concepts/flows.mdx

@@ -86,9 +86,14 @@ There are two ways to create and use a flow in ControlFlow: using the `Flow` obj
 
 In both cases, the goal is to instantiate a flow that provides a shared context for all tasks and agents within the flow. The @flow decorator is the most portable and flexible way to create a flow, as it encapsulates the entire flow within a function that gains additional capabilities as a result, because it becomes a Prefect flow as well. However, the `Flow` context manager can be used to quickly create flows for ad-hoc purposes.
 
+<Tip>
+#### Decorator or context manager?
+
+In general, you should use the `@flow` decorator for most flows, as it is more capable, flexible, and portable. You should use the `Flow` context manager primarily for nested or ad-hoc flows, when your primary goal is to create a shared thread for a few tasks. 
+</Tip>
 ### The `@flow` decorator
 
-To create a flow using the `@flow` decorator, apply `@cf.flow` to any function. Any tasks run inside the decorated function will execute within the context of the same flow. Flow functions can 
+To create a flow using a decorator, apply `@cf.flow` to any function. Any tasks run inside the decorated function will execute within the context of the same flow. 
 
 <CodeGroup>
 ```python Code
@@ -118,6 +123,16 @@ The following flow properties are inferred from the decorated function:
 
 Additional properties can be set by passing keyword arguments directly to the `@flow` decorator or to the `flow_kwargs` parameter when calling the decorated function.
 
+<Tip>
+You may not want the arguments to your flow function to be used as context. In that case, you can set `args_as_context=False` when decorating or calling the function:
+
+```python
+@cf.flow(args_as_context=False)
+def my_flow(secret_var: str):
+    ...
+```
+</Tip>
+
 ### The `Flow` object and context manager
 
 For more precise control over a flow, you can instantiate a `Flow` object directly. Most commonly, you'll use the flow as a context manager to create a new thread for one or more tasks. 
@@ -158,6 +173,7 @@ The flow's description is shown to all participating agents to help them underst
 If you provide a list of tools to the flow, they will be available to all agents on all tasks within the flow. This is useful if you have a tool that you want to be universally available.
 
 ### Agent
+
 You can provide a default agent that will be used in place of ControlFlow's global default agent for any tasks that don't explicitly specify their own agents.
 
 ### Context

docs/examples/language-tutor.mdx

@@ -35,7 +35,7 @@ def language_learning_session(language: str) -> None:
         """
     )
 
-    @cf.flow(agent=tutor)
+    @cf.flow(default_agent=tutor)
     def learning_flow():
         cf.run(
             f"Greet the user, learn their name,and introduce the {language} learning session",
@@ -99,7 +99,7 @@ This implementation showcases several important ControlFlow features and concept
 2. **Flow-level Agent Assignment**: We assign the tutor agent to the entire flow, eliminating the need to specify it for each task.
 
    ```python
-   @cf.flow(agent=tutor)
+   @cf.flow(default_agent=tutor)
    def learning_flow():
        ...
    ```

docs/guides/default-agent.mdx

@@ -26,22 +26,20 @@ task = cf.Task('What is 2 + 2?')
 task.run()  # Result: 42
 ```
 
-## Changing a Flow's Default Agents
+## Changing a Flow's Default Agent
 
-You can also set a default agent (or agents) for a specific flow. This allows you to use different default agents for different parts of your application without changing the global default.
-
-To set a default agent for a flow, use the `agents` parameter when decorating your flow function:
+You can also set a default agent for a specific flow. by using its `default_agent` parameter when decorating your flow function or creating the flow object.
 
 ```python
 import controlflow as cf
 
 researcher = cf.Agent('Researcher', instructions='Conduct thorough research')
 writer = cf.Agent('Writer', instructions='Write clear, concise content')
 
-@cf.flow(agents=[researcher, writer])
+@cf.flow(default_agent=writer)
 def research_flow():
-    research_task = cf.Task("Research the topic")
-    writing_task = cf.Task("Write a report")
+    research_task = cf.Task("Research the topic", agents=[researcher])
+    writing_task = cf.Task("Write a report") # will use the writer agent by default
     return writing_task
 
 result = research_flow()
@@ -54,7 +52,7 @@ In this example, both the `research_task` and `writing_task` will use the `resea
 When ControlFlow needs to assign an agent to a task, it follows this precedence:
 
 1. Agents specified directly on the task (`task.agents`)
-2. Agents specified for the flow (`@flow(agents=[...])`)
+2. The agent specified by the flow (`@flow(default_agent=...)`)
 3. The global default agent (`controlflow.defaults.agent`)
 
 This means you can always override the default agent by specifying agents directly on a task, regardless of what default agents are set at the flow or global level.
@@ -64,23 +62,23 @@ import controlflow as cf
 
 global_agent = cf.Agent('Global', instructions='I am the global default')
 cf.defaults.agent = global_agent
-
 flow_agent = cf.Agent('Flow', instructions='I am the flow default')
-
 task_agent = cf.Agent('Task', instructions='I am specified for this task')
 
-@cf.flow(agents=[flow_agent])
+@cf.flow(default_agent=flow_agent)
 def example_flow():
-    task1 = cf.Task("Task with flow default")
-    task2 = cf.Task("Task with specific agent", agents=[task_agent])
-    return task1, task2
+    task1 = cf.run("Task with flow default")
+    task2 = cf.run("Task with specific agent", agents=[task_agent])
+
+task3 = cf.run("Task with global default")
+
 
 results = example_flow()
 ```
 
 In this example:
 - `task1` will use the `flow_agent`
 - `task2` will use the `task_agent`
-- If we created a task outside of `example_flow`, it would use the `global_agent`
+- `task3` will use the `global_agent`
 
 By understanding and utilizing these different levels of agent configuration, you can create more flexible and customized workflows in ControlFlow.

src/controlflow/decorators.py

@@ -20,7 +20,7 @@ def flow(
     thread: Optional[str] = None,
     instructions: Optional[str] = None,
     tools: Optional[list[Callable[..., Any]]] = None,
-    agents: Optional[list[Agent]] = None,
+    default_agent: Optional[Agent] = None,  # Changed from 'agents'
     retries: Optional[int] = None,
     retry_delay_seconds: Optional[Union[float, int]] = None,
     timeout_seconds: Optional[Union[float, int]] = None,
@@ -44,7 +44,7 @@ def flow(
         thread (str, optional): The thread to execute the flow on. Defaults to None.
         instructions (str, optional): Instructions for the flow. Defaults to None.
         tools (list[Callable], 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.
+        default_agent (Agent, optional): The default agent to be used in the flow. Defaults to None.
         args_as_context (bool, optional): Whether to pass the arguments as context to the flow. Defaults to True.
     Returns:
         callable: The wrapped function or a new flow decorator if `fn` is not provided.
@@ -57,7 +57,7 @@ def flow(
             thread=thread,
             instructions=instructions,
             tools=tools,
-            agents=agents,
+            default_agent=default_agent,  # Changed from 'agents'
             retries=retries,
             retry_delay_seconds=retry_delay_seconds,
             timeout_seconds=timeout_seconds,
@@ -90,8 +90,10 @@ def wrapper(
             flow_kwargs.setdefault("thread_id", thread)
         if tools is not None:
             flow_kwargs.setdefault("tools", tools)
-        if agents is not None:
-            flow_kwargs.setdefault("agents", agents)
+        if default_agent is not None:  # Changed from 'agents'
+            flow_kwargs.setdefault(
+                "default_agent", default_agent
+            )  # Changed from 'agents'
 
         context = bound.arguments if args_as_context else {}
 

src/controlflow/flows/flow.py

@@ -31,7 +31,7 @@ class Flow(ControlFlowModel):
         default_factory=list,
         description="Tools that will be available to every agent in the flow",
     )
-    agent: Optional[Agent] = Field(
+    default_agent: Optional[Agent] = Field(
         None,
         description="The default agent for the flow. This agent will be used "
         "for any task that does not specify an agent.",

src/controlflow/tasks/task.py

@@ -439,8 +439,8 @@ def get_agents(self) -> list[Agent]:
                 flow = get_flow()
             except ValueError:
                 flow = None
-            if flow and flow.agent:
-                return [flow.agent]
+            if flow and flow.default_agent:
+                return [flow.default_agent]
             else:
                 return [controlflow.defaults.agent]
 

tests/flows/test_flows.py

@@ -10,7 +10,7 @@ def test_flow_initialization(self):
         flow = Flow()
         assert flow.thread_id is not None
         assert len(flow.tools) == 0
-        assert flow.agent is None
+        assert flow.default_agent is None
         assert flow.context == {}
 
     def test_flow_with_custom_tools(self):
@@ -179,16 +179,16 @@ def test_flow_sets_thread_id_for_history(self, tmpdir):
 class TestFlowCreatesDefaults:
     def test_flow_with_custom_agents(self):
         agent1 = Agent()
-        flow = Flow(agent=agent1)
-        assert flow.agent == agent1
+        flow = Flow(default_agent=agent1)  # Changed from 'agent'
+        assert flow.default_agent == agent1  # Changed from 'agent'
 
     def test_flow_agent_becomes_task_default(self):
         agent = Agent()
         t1 = Task("t1")
         assert agent not in t1.get_agents()
         assert len(t1.get_agents()) == 1
 
-        with Flow(agent=agent):
+        with Flow(default_agent=agent):  # Changed from 'agent'
             t2 = Task("t2")
             assert agent in t2.get_agents()
             assert len(t2.get_agents()) == 1

tests/tasks/test_tasks.py

@@ -120,7 +120,7 @@ def test_task_loads_agent_from_parent():
 def test_task_loads_agent_from_flow():
     def_agent = controlflow.defaults.agent
     agent = Agent()
-    with Flow(agent=agent):
+    with Flow(default_agent=agent):
         task = SimpleTask()
 
         assert task.agents is None
@@ -141,7 +141,7 @@ def test_task_loads_agent_from_default_if_none_otherwise():
 def test_task_loads_agent_from_parent_before_flow():
     agent1 = Agent()
     agent2 = Agent()
-    with Flow(agent=agent1):
+    with Flow(default_agent=agent1):
         with SimpleTask(agents=[agent2]):
             child = SimpleTask()