Brandon/cre 130 pipeline project structure (#1066)

* WIP. Procedure appears to be working well. Working on mocking properly for tests

* All tests are passing now

* rshift working

* Add back in Gui's tool_usage fix

* WIP

* Going to start refactoring for pipeline_output

* Update terminology

* new pipeline flow with traces and usage metrics working. need to add more tests and make sure PipelineOutput behaves likew CrewOutput

* Fix pipelineoutput to look more like crewoutput and taskoutput

* Implemented additional tests for pipeline. One test is failing. Need team support

* Update docs for pipeline

* Update pipeline to properly process input and ouput dictionary

* Update Pipeline docs

* Add back in commentary at top of pipeline file

* Starting to work on router

* Drop router for now. will add in separately

* In the middle of fixing router. A ton of circular dependencies. Moving over to a new design.

* WIP.

* Fix circular dependencies and updated PipelineRouter

* Add in Eduardo feedback. Still need to add in more commentary describing the design decisions for pipeline

* Add developer notes to explain what is going on in pipelines.

* Add doc strings

* Fix missing rag datatype

* WIP. Converting usage metrics from a dict to an object

* Fix tests that were checking usage metrics

* Drop todo

* Fix 1 type error in pipeline

* Update pipeline to use UsageMetric

* Add missing doc string

* WIP.

* Change names

* Rename variables based on joaos feedback

* Fix critical circular dependency issues. Now needing to fix trace issue.

* Tests working now!

* Add more tests which showed underlying issue with traces

* Fix tests

* Remove overly complicated test

* Add router example to docs

* Clean up end of docs

* Clean up docs

* Working on creating Crew templates and pipeline templates

* WIP.

* WIP

* Fix poetry install from templates

* WIP

* Restructure

* changes for lorenze

* more todos

* WIP: create pipelines cli working

* wrapped up router

* ignore mypy src on templates

* ignored signature of copy

* fix all verbose

* rm print statements

* brought back correct folders

* fixes missing folders and then rm print statements

* fixed tests

* fixed broken test

* fixed type checker

* fixed type ignore

* ignore types for templates

* needed

* revert

* exclude only required

* rm type errors on templates

* rm excluding type checks for template files on github action

* fixed missing quotes

---------

Co-authored-by: Brandon Hancock <brandon@brandonhancock.io>

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

@@ -1,14 +1,4 @@
-def memoize(func):
-    cache = {}
-
-    def memoized_func(*args, **kwargs):
-        key = (args, tuple(kwargs.items()))
-        if key not in cache:
-            cache[key] = func(*args, **kwargs)
-        return cache[key]
-
-    memoized_func.__dict__.update(func.__dict__)
-    return memoized_func
+from crewai.project.utils import memoize
 
 
 def task(func):
@@ -61,6 +51,21 @@ def cache_handler(func):
     return memoize(func)
 
 
+def stage(func):
+    func.is_stage = True
+    return memoize(func)
+
+
+def router(func):
+    func.is_router = True
+    return memoize(func)
+
+
+def pipeline(func):
+    func.is_pipeline = True
+    return memoize(func)
+
+
 def crew(func):
     def wrapper(self, *args, **kwargs):
         instantiated_tasks = []

src/crewai/project/crew_base.py

@@ -24,6 +24,7 @@ def CrewBase(cls):
         original_agents_config_path = getattr(
             cls, "agents_config", "config/agents.yaml"
         )
+
         original_tasks_config_path = getattr(cls, "tasks_config", "config/tasks.yaml")
 
         def __init__(self, *args, **kwargs):
@@ -37,9 +38,11 @@ def CrewBase(cls):
             self.agents_config = self.load_yaml(
                 os.path.join(self.base_directory, self.original_agents_config_path)
             )
+
             self.tasks_config = self.load_yaml(
                 os.path.join(self.base_directory, self.original_tasks_config_path)
             )
+
             self.map_all_agent_variables()
             self.map_all_task_variables()
 

src/crewai/project/pipeline_base.py

@@ -0,0 +1,58 @@
+from typing import Callable, Dict
+
+from pydantic import ConfigDict
+
+from crewai.crew import Crew
+from crewai.pipeline.pipeline import Pipeline
+from crewai.routers.router import Router
+
+
+# TODO: Could potentially remove. Need to check with @joao and @gui if this is needed for CrewAI+
+def PipelineBase(cls):
+    class WrappedClass(cls):
+        model_config = ConfigDict(arbitrary_types_allowed=True)
+        is_pipeline_class: bool = True
+
+        def __init__(self, *args, **kwargs):
+            super().__init__(*args, **kwargs)
+            self.stages = []
+            self._map_pipeline_components()
+
+        def _get_all_functions(self):
+            return {
+                name: getattr(self, name)
+                for name in dir(self)
+                if callable(getattr(self, name))
+            }
+
+        def _filter_functions(
+            self, functions: Dict[str, Callable], attribute: str
+        ) -> Dict[str, Callable]:
+            return {
+                name: func
+                for name, func in functions.items()
+                if hasattr(func, attribute)
+            }
+
+        def _map_pipeline_components(self):
+            all_functions = self._get_all_functions()
+            crew_functions = self._filter_functions(all_functions, "is_crew")
+            router_functions = self._filter_functions(all_functions, "is_router")
+
+            for stage_attr in dir(self):
+                stage = getattr(self, stage_attr)
+                if isinstance(stage, (Crew, Router)):
+                    self.stages.append(stage)
+                elif callable(stage) and hasattr(stage, "is_crew"):
+                    self.stages.append(crew_functions[stage_attr]())
+                elif callable(stage) and hasattr(stage, "is_router"):
+                    self.stages.append(router_functions[stage_attr]())
+                elif isinstance(stage, list) and all(
+                    isinstance(item, Crew) for item in stage
+                ):
+                    self.stages.append(stage)
+
+        def build_pipeline(self) -> Pipeline:
+            return Pipeline(stages=self.stages)
+
+    return WrappedClass

src/crewai/project/utils.py

@@ -0,0 +1,11 @@
+def memoize(func):
+    cache = {}
+
+    def memoized_func(*args, **kwargs):
+        key = (args, tuple(kwargs.items()))
+        if key not in cache:
+            cache[key] = func(*args, **kwargs)
+        return cache[key]
+
+    memoized_func.__dict__.update(func.__dict__)
+    return memoized_func