Merge pull request #285 from PrefectHQ/planning

Update planning module

docs/concepts/tasks/running-tasks.mdx

@@ -56,11 +56,31 @@ Crafting worlds and shaping dreams.
 
 </CodeGroup>
 
-Note that this example is functionally equivalent to the previous one.
+Note that this example is functionally equivalent to the previous one. There is also an async equivalent, `cf.run_async()`.
 
 This operation is so common that you'll see `cf.run()` used throughout the ControlFlow documentation.
 
+## `cf.run_tasks()`
 
+ControlFlow also provides a `run_tasks` function, which orchestrates one or more preexisting tasks to completion. Note that you need to access the task's `result` to see the output.
+
+```python
+import controlflow as cf
+
+task_1 = cf.Task('Write a poem about AI')
+task_2 = cf.Task('Critique the poem', depends_on=[task_1])
+
+cf.run_tasks([task_1, task_2])
+
+print(task_1.result)
+print(task_2.result)
+```
+
+There is also an equivalent async function, `cf.run_tasks_async`.
+
+<Tip>
+When you run tasks as a batch, they share context because they are automatically run in a single flow.
+</Tip>
 
 ## `@task`
 

docs/patterns/planning.mdx

@@ -1,57 +1,45 @@
 ---
-title: Planning
+title: AI Planning
+description: Use AI to generate new tasks.
+icon: compass
 ---
 
-<Tip>
-Automatically generate subtasks to break complex tasks into manageable steps.
-</Tip>
+The `plan()` function in ControlFlow extends the capabilities of AI workflows by allowing dynamic generation of tasks. This feature allows you to leverage AI for creating structured, goal-oriented task sequences programmatically.
 
-ControlFlow has many features that help you structure your AI workflows into small, well-defined tasks. However, sometimes it will be impractical or even impossible to identify all the necessary tasks upfront. In these cases, you can use ControlFlow's planning capabilities to automatically generate subtasks based on the main task's objective.
+## Purpose of AI planning
 
-By calling the `generate_subtasks()` method on a task, you can instruct an AI agent to come up with a plan for achieving the main task. The agent will generate a series of subtasks, each representing a step in the plan, and set up the necessary dependencies between them.
+While ControlFlow allows manual creation of tasks for AI workflows, there are scenarios where automatically generating tasks can be beneficial:
 
+1. **Dynamic Task Generation**: When the specific steps to achieve a goal aren't known in advance.
+2. **Complex Problem Decomposition**: For objectives that require breaking down into subtasks based on context or intermediate results.
+3. **Adaptive Workflows**: In processes that need to adjust based on changing conditions or new information.
 
 
+## The `plan()` function
+
+The `plan()` function takes a high-level objective and generates a structured sequence of tasks to achieve that goal. Here's a basic example:
+
 ```python
 import controlflow as cf
 
-task = cf.Task(
-    objective="Compare the height of the tallest building "
-    "in North America to the tallest building in Europe",
+tasks = cf.plan(
+    objective="Analyze customer feedback data",
+    n_tasks=3  # Optionally specify the number of tasks
 )
 
-task.generate_subtasks()
-
-print([f'{i+1}: {t.objective}' for i, t in enumerate(task.subtasks)])
+# Execute the generated plan
+cf.run_tasks(tasks)
 ```
 
-Running the above code will print something like:
-
-```python
-[
-    "1: Identify the Tallest Building in North America",
-    "2: Identify the Tallest Building in Europe",
-    "3: Obtain Height of the Tallest Building in North America",
-    "4: Obtain Height of the Tallest Building in Europe",
-    "5: Compare the Heights",
-]
- ```
-If you investigate more closely, you'll see that the subtasks have proper dependencies. In the above example, #3 depends on #1, #4 depends on #2, and #5 depends on #3 and #4. And of course, the parent task depends on all of them. 
-
-ControlFlow's orchestration engine will not allow the parent task to be considered complete until all of its subtasks have been successfully executed, which is why this is an effective way to structure complex workflows.
-
-<Tip>
-Subtask generation isn't magic: it's a ControlFlow flow!
-</Tip>
-
-## Customizing subtask generation
-You can influence subtask generation in a few ways.
+In this example, `plan()` will generate a list of 3 tasks that, when completed, should result in an analysis of customer feedback data. These tasks might include steps like "Load data", "Preprocess text", "Perform sentiment analysis", etc.
 
-### Planning agent
+## Advanced usage
 
-By default, subtasks are generated by the first agent assigned to the parent task. You can customize this by passing an `agent` argument to `generate_subtasks()`.
+The `plan` function can do more than just generate tasks that achieve an objective. 
 
+### Dependencies
 
-### Instructions
+If appropriate, `plan` can generate tasks that depend on each other or have parent/child relationships. You can influence this behavior by providing `instructions`. 
 
-You can provide natural language `instructions` to help the agent generate subtasks. This is especially useful when the task is ambiguous or requires domain-specific knowledge.
+### Agents and tools
+You can pass a list of agents or tools to the `plan` function. It will take these into account when generating tasks and assign agents or tools to tasks as needed.

docs/quickstart.mdx

@@ -1,6 +1,6 @@
 ---
 title: "Quickstart"
-description: Build your first agentic workflow in less than 30 seconds.
+description: Build your first agentic workflow in less than a minute.
 icon: rocket
 ---
 
@@ -21,29 +21,30 @@ Install ControlFlow with pip:
 pip install controlflow
 ```
 
-Configure your preferred LLM provider. By default, ControlFlow uses OpenAI, so you'll need to set your API key:
+Next, set up your LLM provider. By default, ControlFlow uses OpenAI, so you'll need to configure an OpenAI API key:
 
 ```bash
 export OPENAI_API_KEY="your-api-key"
 ```
 
-To use another provider, see the docs on [configuring LLMs](/guides/llms). Note that one of the agents in this quickstart is configured with OpenAI's GPT-4o-mini model; you can change the model name to whatever you prefer.
+To use another provider, see the docs on [configuring LLMs](/guides/llms).
 
-## Quickstart setup
+## Create some data
 
 
 In this quickstart, we're going to build an email processing pipelines, so let's create some sample data to work with. Execute this code in your Python interpreter to set up the (very simple) example emails we'll use throughout the quickstart:
 
+<Tip>
+Try changing the emails to your own content to see how ControlFlow works with different inputs.
+</Tip>
+
 ```python
 emails = [
     "Hello, I need an update on the project status.",
     "Subject: Exclusive offer just for you!",
     "Urgent: Project deadline moved up by one week.",
 ]
 ```
-<Tip>
-Try changing the emails to your own content to see how ControlFlow works with different inputs.
-</Tip>
 ## Running a single task
 
 Let's start with the basics. We're going to create a task that generates a reply to an email.
@@ -105,8 +106,6 @@ You may have noticed that in the last example, we didn't assign an agent to the
 
 Agents are sort of like portable configurations for how to perform tasks, which could include specific LLMs, tools, instructions, and more. For our spam classifier, we'll create a new agent that uses a smaller, faster LLM and specialized instructions.
 
-
-
 In addition, note that the `result_type` of this task is a list of labels, indicating that the agent must choose one of the provided options. This is the simplest way to create a classification task, but you can require more complex [output formats](/concepts/tasks/task-results) as well.
 
 

src/controlflow/__init__.py

@@ -15,7 +15,7 @@
 from .instructions import instructions
 from .decorators import flow, task
 from .tools import tool
-from .run import run, run_async, run_tasks, run_tasks_async
+from .fns import run, run_async, run_tasks, run_tasks_async, plan
 
 
 # --- Version ---

src/controlflow/agents/agent.py

@@ -34,6 +34,7 @@
 
 if TYPE_CHECKING:
     from controlflow.orchestration.turn_strategies import TurnStrategy
+    from controlflow.tasks import Task
     from controlflow.tools.tools import Tool
 logger = logging.getLogger(__name__)
 
@@ -190,36 +191,61 @@ def __exit__(self, *exc_info):
     def run(
         self,
         objective: str,
-        *task_args,
-        agents: list["Agent"] = None,
+        *,
         turn_strategy: "TurnStrategy" = None,
         **task_kwargs,
     ):
-        agents = agents or [] + [self]
-        task = controlflow.Task(
+        return controlflow.run(
             objective=objective,
-            agents=agents,
-            *task_args,
+            agents=[self],
+            turn_strategy=turn_strategy,
             **task_kwargs,
         )
-        return task.run(turn_strategy=turn_strategy)
 
     async def run_async(
         self,
         objective: str,
-        *task_args,
-        agents: list["Agent"] = None,
+        *,
         turn_strategy: "TurnStrategy" = None,
         **task_kwargs,
     ):
-        agents = agents or [] + [self]
-        task = controlflow.Task(
+        return await controlflow.run_async(
             objective=objective,
-            agents=agents,
-            *task_args,
+            agents=[self],
+            turn_strategy=turn_strategy,
             **task_kwargs,
         )
-        return await task.run_async(turn_strategy=turn_strategy)
+
+    def plan(
+        self,
+        objective: str,
+        instructions: Optional[str] = None,
+        agents: Optional[list["Agent"]] = None,
+        tools: Optional[list["Tool"]] = None,
+        context: Optional[dict] = None,
+    ) -> list["Task"]:
+        """
+        Generate a list of tasks that represent a structured plan for achieving
+        the objective.
+
+        Args:
+            objective (str): The objective to plan for.
+            instructions (Optional[str]): Optional instructions for the planner.
+            agents (Optional[list[Agent]]): Optional list of agents to include in the plan. If None, this agent is used.
+            tools (Optional[list[Tool]]): Optional list of tools to include in the plan. If None, this agent's tools are used.
+            context (Optional[dict]): Optional context to include in the plan.
+
+        Returns:
+            list[Task]: A list of tasks that represent a structured plan for achieving the objective.
+        """
+        return controlflow.tasks.plan(
+            objective=objective,
+            instructions=instructions,
+            agent=self,
+            agents=agents or [self],
+            tools=tools or [self.tools],
+            context=context,
+        )
 
     def _run_model(
         self,

src/controlflow/fns/__init__.py

@@ -0,0 +1,2 @@
+from .run import run, run_async, run_tasks, run_tasks_async
+from .plan import plan