WIP

src/crewai/__init__.py

@@ -1,6 +1,7 @@
 from crewai.agent import Agent
 from crewai.crew import Crew
+from crewai.pipeline import Pipeline
 from crewai.process import Process
 from crewai.task import Task
 
-__all__ = ["Agent", "Crew", "Process", "Task"]
+__all__ = ["Agent", "Crew", "Process", "Task", "Pipeline"]

src/crewai/cli/cli.py

@@ -21,19 +21,15 @@ def crewai():
 @crewai.command()
 @click.argument("type", type=click.Choice(["crew", "pipeline"]))
 @click.argument("name")
-@click.argument("crew_names", nargs=-1)
-def create(type, name, crew_names):
+@click.option(
+    "--router", is_flag=True, help="Create a pipeline with router functionality"
+)
+def create(type, name, router):
     """Create a new crew or pipeline."""
     if type == "crew":
         create_crew(name)
     elif type == "pipeline":
-        if not crew_names:
-            click.secho(
-                "Error: At least one crew name must be provided for a pipeline.",
-                fg="red",
-            )
-            return
-        create_pipeline(name, crew_names)
+        create_pipeline(name, router)
     else:
         click.secho("Error: Invalid type. Must be 'crew' or 'pipeline'.", fg="red")
 

src/crewai/cli/create_pipeline.py

@@ -1,119 +1,52 @@
 import os
+import shutil
 from pathlib import Path
 
 import click
 
-from crewai.cli.create_crew import create_crew
-from crewai.cli.utils import copy_template
 
-
-def create_pipeline(pipeline_name, crew_names):
-    """Create a new pipeline with multiple crews."""
-    folder_name = pipeline_name.replace(" ", "_").replace("-", "_").lower()
-    class_name = (
-        pipeline_name.replace("_", " ").replace("-", " ").title().replace(" ", "")
-    )
+def create_pipeline(name, router=False):
+    """Create a new pipeline project."""
+    folder_name = name.replace(" ", "_").replace("-", "_").lower()
+    class_name = name.replace("_", " ").replace("-", " ").title().replace(" ", "")
 
     click.secho(f"Creating pipeline {folder_name}...", fg="green", bold=True)
 
-    # Create the main project structure
     project_root = Path(folder_name)
-    project_root.mkdir(exist_ok=True)
-    (project_root / "src" / folder_name).mkdir(parents=True, exist_ok=True)
-    (project_root / "src" / folder_name / "crews").mkdir(parents=True, exist_ok=True)
-    (project_root / "src" / folder_name / "config").mkdir(parents=True, exist_ok=True)
-    (project_root / "src" / folder_name / "tools").mkdir(parents=True, exist_ok=True)
-    (project_root / "tests").mkdir(exist_ok=True)
-
-    # Create .env file
-    with open(project_root / ".env", "w") as file:
-        file.write("OPENAI_API_KEY=YOUR_API_KEY")
+    if project_root.exists():
+        click.secho(f"Error: Folder {folder_name} already exists.", fg="red")
+        return
 
     package_dir = Path(__file__).parent
-    templates_dir = package_dir / "templates" / "pipeline"
+    template_folder = "pipeline_router" if router else "pipeline"
+    templates_dir = package_dir / "templates" / template_folder
 
-    # Process main.py template
-    with open(templates_dir / "main.py", "r") as f:
-        main_content = f.read()
+    # Copy the entire template directory structure
+    shutil.copytree(templates_dir, project_root)
 
-    # Replace variables in main.py
-    main_content = main_content.replace("{{folder_name}}", folder_name)
-    main_content = main_content.replace("{{pipeline_name}}", class_name)
-
-    # Write updated main.py
-    with open(project_root / "src" / folder_name / "main.py", "w") as f:
-        f.write(main_content)
-
-    # Process pipeline.py template
-    with open(templates_dir / "pipeline.py", "r") as f:
-        pipeline_content = f.read()
-
-    # Replace pipeline name
-    pipeline_content = pipeline_content.replace("{{pipeline_name}}", class_name)
-
-    # Generate crew initialization lines
-    crew_init_lines = []
-    crew_stage_lines = []
-    for crew_name in crew_names:
-        crew_class_name = (
-            crew_name.replace("_", " ").replace("-", " ").title().replace(" ", "")
-        )
-        crew_init_lines.append(
-            f"        self.{crew_name.lower()}_crew = {crew_class_name}Crew().crew()"
-        )
-        crew_stage_lines.append(f"                self.{crew_name.lower()}_crew,")
-
-    # Replace crew initialization placeholder
-    pipeline_content = pipeline_content.replace(
-        "        {% for crew_name in crew_names %}\n"
-        "        self.{{crew_name.lower()}}_crew = {{crew_name}}Crew().crew()\n"
-        "        {% endfor %}",
-        "\n".join(crew_init_lines),
-    )
-
-    # Replace crew stages placeholder
-    pipeline_content = pipeline_content.replace(
-        "                {% for crew_name in crew_names %}\n"
-        "                self.{{crew_name.lower()}}_crew,\n"
-        "                {% endfor %}",
-        "\n".join(crew_stage_lines),
-    )
-
-    # Update imports with correct package structure
-    crew_imports = [
-        f"from {folder_name}.src.{folder_name}.crews.{name.lower()}.crew import {name.replace('_', ' ').replace('-', ' ').title().replace(' ', '')}Crew"
-        for name in crew_names
-    ]
-    pipeline_content = pipeline_content.replace(
-        "from crews.crew import *", "\n".join(crew_imports)
-    )
-
-    with open(project_root / "src" / folder_name / "pipeline.py", "w") as f:
-        f.write(pipeline_content)
-
-    # Copy and process other template files
-    template_files = [
-        (".gitignore", project_root),
-        ("pyproject.toml", project_root),
-        ("README.md", project_root),
-        ("__init__.py", project_root / "src" / folder_name),
-        ("tools/custom_tool.py", project_root / "src" / folder_name / "tools"),
-        ("tools/__init__.py", project_root / "src" / folder_name / "tools"),
-        ("config/agents.yaml", project_root / "src" / folder_name / "config"),
-        ("config/tasks.yaml", project_root / "src" / folder_name / "config"),
+    # Process and replace placeholders in specific files
+    files_to_process = [
+        "README.md",
+        "pyproject.toml",
+        "main.py",
+        "pipeline.py",
+        "crews/research_crew.py",
+        "crews/write_x_crew.py",
+        "crews/write_linkedin_crew.py",
     ]
 
-    for template_file, destination in template_files:
-        src_file = templates_dir / template_file
-        dst_file = destination / os.path.basename(template_file)
-        copy_template(src_file, dst_file, pipeline_name, class_name, folder_name)
+    for file_path in files_to_process:
+        full_path = project_root / file_path
+        if full_path.exists():
+            with open(full_path, "r") as file:
+                content = file.read()
 
-    # Create crew files
-    for crew_name in crew_names:
-        create_crew(crew_name, project_root / "src" / folder_name / "crews")
+            content = content.replace("{{name}}", name)
+            content = content.replace("{{crew_name}}", class_name)
+            content = content.replace("{{folder_name}}", folder_name)
+            content = content.replace("{{pipeline_name}}", class_name)  # Add this line
 
-    click.secho(
-        f"Pipeline {pipeline_name} created successfully with crews: {', '.join(crew_names)}!",
-        fg="green",
-        bold=True,
-    )
+            with open(full_path, "w") as file:
+                file.write(content)
+
+    click.secho(f"Pipeline {name} created successfully!", fg="green", bold=True)

src/crewai/cli/templates/pipeline_router/.gitignore

@@ -0,0 +1,2 @@
+.env
+__pycache__/

src/crewai/cli/templates/pipeline_router/README.md

@@ -0,0 +1,57 @@
+# {{crew_name}} Crew
+
+Welcome to the {{crew_name}} Crew project, powered by [crewAI](https://crewai.com). This template is designed to help you set up a multi-agent AI system with ease, leveraging the powerful and flexible framework provided by crewAI. Our goal is to enable your agents to collaborate effectively on complex tasks, maximizing their collective intelligence and capabilities.
+
+## Installation
+
+Ensure you have Python >=3.10 <=3.13 installed on your system. This project uses [Poetry](https://python-poetry.org/) for dependency management and package handling, offering a seamless setup and execution experience.
+
+First, if you haven't already, install Poetry:
+
+```bash
+pip install poetry
+```
+
+Next, navigate to your project directory and install the dependencies:
+
+1. First lock the dependencies and then install them:
+```bash
+poetry lock
+```
+```bash
+poetry install
+```
+### Customizing
+
+**Add your `OPENAI_API_KEY` into the `.env` file**
+
+- Modify `src/{{folder_name}}/config/agents.yaml` to define your agents
+- Modify `src/{{folder_name}}/config/tasks.yaml` to define your tasks
+- Modify `src/{{folder_name}}/crew.py` to add your own logic, tools and specific args
+- Modify `src/{{folder_name}}/main.py` to add custom inputs for your agents and tasks
+
+## Running the Project
+
+To kickstart your crew of AI agents and begin task execution, run this from the root folder of your project:
+
+```bash
+poetry run {{folder_name}}
+```
+
+This command initializes the {{name}} Crew, assembling the agents and assigning them tasks as defined in your configuration.
+
+This example, unmodified, will run the create a `report.md` file with the output of a research on LLMs in the root folder.
+
+## Understanding Your Crew
+
+The {{name}} Crew is composed of multiple AI agents, each with unique roles, goals, and tools. These agents collaborate on a series of tasks, defined in `config/tasks.yaml`, leveraging their collective skills to achieve complex objectives. The `config/agents.yaml` file outlines the capabilities and configurations of each agent in your crew.
+
+## Support
+
+For support, questions, or feedback regarding the {{crew_name}} Crew or crewAI.
+- Visit our [documentation](https://docs.crewai.com)
+- Reach out to us through our [GitHub repository](https://github.com/joaomdmoura/crewai)
+- [Join our Discord](https://discord.com/invite/X4JWnZnxPb)
+- [Chat with our docs](https://chatg.pt/DWjSBZn)
+
+Let's create wonders together with the power and simplicity of crewAI.

src/crewai/cli/templates/pipeline_router/config/agents.yaml

@@ -0,0 +1,19 @@
+researcher:
+  role: >
+    {topic} Senior Data Researcher
+  goal: >
+    Uncover cutting-edge developments in {topic}
+  backstory: >
+    You're a seasoned researcher with a knack for uncovering the latest
+    developments in {topic}. Known for your ability to find the most relevant
+    information and present it in a clear and concise manner.
+
+reporting_analyst:
+  role: >
+    {topic} Reporting Analyst
+  goal: >
+    Create detailed reports based on {topic} data analysis and research findings
+  backstory: >
+    You're a meticulous analyst with a keen eye for detail. You're known for
+    your ability to turn complex data into clear and concise reports, making
+    it easy for others to understand and act on the information you provide.

src/crewai/cli/templates/pipeline_router/config/tasks.yaml

@@ -0,0 +1,17 @@
+research_task:
+  description: >
+    Conduct a thorough research about {topic}
+    Make sure you find any interesting and relevant information given
+    the current year is 2024.
+  expected_output: >
+    A list with 10 bullet points of the most relevant information about {topic}
+  agent: researcher
+
+reporting_task:
+  description: >
+    Review the context you got and expand each topic into a full section for a report.
+    Make sure the report is detailed and contains any and all relevant information.
+  expected_output: >
+    A fully fledge reports with the mains topics, each with a full section of information.
+    Formatted as markdown without '```'
+  agent: reporting_analyst

src/crewai/cli/templates/pipeline_router/crews/crew.py

@@ -0,0 +1,53 @@
+from crewai import Agent, Crew, Process, Task
+from crewai.project import CrewBase, agent, crew, task
+
+# Uncomment the following line to use an example of a custom tool
+# from {{folder_name}}.tools.custom_tool import MyCustomTool
+
+# Check our tools documentations for more information on how to use them
+# from crewai_tools import SerperDevTool
+
+@CrewBase
+class {{crew_name}}Crew():
+	"""{{crew_name}} crew"""
+	agents_config = '../config/agents.yaml'
+	tasks_config = '../config/tasks.yaml'
+
+	@agent
+	def researcher(self) -> Agent:
+		return Agent(
+			config=self.agents_config['researcher'],
+			# tools=[MyCustomTool()], # Example of custom tool, loaded on the beginning of file
+			verbose=True
+		)
+
+	@agent
+	def reporting_analyst(self) -> Agent:
+		return Agent(
+			config=self.agents_config['reporting_analyst'],
+			verbose=True
+		)
+
+	@task
+	def research_task(self) -> Task:
+		return Task(
+			config=self.tasks_config['research_task'],
+		)
+
+	@task
+	def reporting_task(self) -> Task:
+		return Task(
+			config=self.tasks_config['reporting_task'],
+			output_file='report.md'
+		)
+
+	@crew
+	def crew(self) -> Crew:
+		"""Creates the {{crew_name}} crew"""
+		return Crew(
+			agents=self.agents, # Automatically created by the @agent decorator
+			tasks=self.tasks, # Automatically created by the @task decorator
+			process=Process.sequential,
+			verbose=2,
+			# process=Process.hierarchical, # In case you wanna use that instead https://docs.crewai.com/how-to/Hierarchical/
+		)

src/crewai/cli/templates/pipeline_router/main.py

@@ -0,0 +1,17 @@
+#!/usr/bin/env python
+import asyncio
+from {{folder_name}}.src.{{folder_name}}.pipeline import {{pipeline_name}}Pipeline
+
+
+async def run():
+    """
+    Run the pipeline.
+    """
+    inputs = [
+        {"topic": "AI LLMs"},
+    ]
+    await {{pipeline_name}}Pipeline().pipeline().kickoff(inputs=inputs)
+
+
+if __name__ == "__main__":
+    asyncio.run(run())

src/crewai/cli/templates/pipeline_router/pipeline.py

@@ -0,0 +1,26 @@
+from crewai import Pipeline
+from crewai.project import PipelineBase
+from crews.crew import *
+
+@PipelineBase
+class {{pipeline_name}}Pipeline:
+    def __init__(self):
+        # Initialize crews
+        {% for crew_name in crew_names %}
+        self.{{crew_name.lower()}}_crew = {{crew_name}}Crew().crew()
+        {% endfor %}
+
+    @pipeline
+    def create_pipeline(self):
+        return Pipeline(
+            stages=[
+                {% for crew_name in crew_names %}
+                self.{{crew_name.lower()}}_crew,
+                {% endfor %}
+            ]
+        )
+
+    async def run(self, inputs):
+        pipeline = self.create_pipeline()
+        results = await pipeline.kickoff(inputs)
+        return results

src/crewai/cli/templates/pipeline_router/pyproject.toml

@@ -0,0 +1,19 @@
+[tool.poetry]
+name = "{{folder_name}}"
+version = "0.1.0"
+description = "{{name}} using crewAI"
+authors = ["Your Name <you@example.com>"]
+
+[tool.poetry.dependencies]
+python = ">=3.10,<=3.13"
+crewai = { extras = ["tools"], version = "^0.46.0" }
+
+[tool.poetry.scripts]
+{{folder_name}} = "{{folder_name}}.main:run"
+train = "{{folder_name}}.main:train"
+replay = "{{folder_name}}.main:replay"
+test = "{{folder_name}}.main:test"
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

src/crewai/cli/templates/pipeline_router/tools/custom_tool.py

@@ -0,0 +1,12 @@
+from crewai_tools import BaseTool
+
+
+class MyCustomTool(BaseTool):
+    name: str = "Name of my tool"
+    description: str = (
+        "Clear description for what this tool is useful for, you agent will need this information to use it."
+    )
+
+    def _run(self, argument: str) -> str:
+        # Implementation goes here
+        return "this is an example of a tool output, ignore it and move along."

src/crewai/project/__init__.py

@@ -1,15 +1,17 @@
 from .annotations import (
     agent,
+    cache_handler,
+    callback,
     crew,
-    task,
+    llm,
     output_json,
     output_pydantic,
+    pipeline,
+    task,
     tool,
-    callback,
-    llm,
-    cache_handler,
 )
 from .crew_base import CrewBase
+from .pipeline_base import PipelineBase
 
 __all__ = [
     "agent",
@@ -20,6 +22,8 @@ __all__ = [
     "tool",
     "callback",
     "CrewBase",
+    "PipelineBase",
     "llm",
     "cache_handler",
+    "pipeline",
 ]

src/crewai/project/annotations.py

@@ -61,6 +61,11 @@ def router(func):
     return memoize(func)
 
 
+def pipeline(func):
+    func.is_pipeline = True
+    return memoize(func)
+
+
 def crew(func):
     def wrapper(self, *args, **kwargs):
         instantiated_tasks = []