Clean up end of docs

docs/core-concepts/Pipeline.md

@@ -0,0 +1,196 @@
+---
+title: crewAI Pipelines
+description: Understanding and utilizing pipelines in the crewAI framework for efficient multi-stage task processing.
+---
+
+## What is a Pipeline?
+
+A pipeline in crewAI represents a structured workflow that allows for the sequential or parallel execution of multiple crews. It provides a way to organize complex processes involving multiple stages, where the output of one stage can serve as input for subsequent stages.
+
+## Key Terminology
+
+Understanding the following terms is crucial for working effectively with pipelines:
+
+- **Stage**: A distinct part of the pipeline, which can be either sequential (a single crew) or parallel (multiple crews executing concurrently).
+- **Run**: A specific execution of the pipeline for a given set of inputs, representing a single instance of processing through the pipeline.
+- **Branch**: Parallel executions within a stage (e.g., concurrent crew operations).
+- **Trace**: The journey of an individual input through the entire pipeline, capturing the path and transformations it undergoes.
+
+Example pipeline structure:
+
+```
+crew1 >> [crew2, crew3] >> crew4
+```
+
+This represents a pipeline with three stages:
+
+1. A sequential stage (crew1)
+2. A parallel stage with two branches (crew2 and crew3 executing concurrently)
+3. Another sequential stage (crew4)
+
+Each input creates its own run, flowing through all stages of the pipeline. Multiple runs can be processed concurrently, each following the defined pipeline structure.
+
+## Pipeline Attributes
+
+| Attribute  | Parameters | Description                                                                           |
+| :--------- | :--------- | :------------------------------------------------------------------------------------ |
+| **Stages** | `stages`   | A list of crews or lists of crews representing the stages to be executed in sequence. |
+
+## Creating a Pipeline
+
+When creating a pipeline, you define a series of stages, each consisting of either a single crew or a list of crews for parallel execution. The pipeline ensures that each stage is executed in order, with the output of one stage feeding into the next.
+
+### Example: Assembling a Pipeline
+
+```python
+from crewai import Crew, Agent, Task, Pipeline
+
+# Define your crews
+research_crew = Crew(
+    agents=[researcher],
+    tasks=[research_task],
+    process=Process.sequential
+)
+
+analysis_crew = Crew(
+    agents=[analyst],
+    tasks=[analysis_task],
+    process=Process.sequential
+)
+
+writing_crew = Crew(
+    agents=[writer],
+    tasks=[writing_task],
+    process=Process.sequential
+)
+
+# Assemble the pipeline
+my_pipeline = Pipeline(
+    stages=[research_crew, analysis_crew, writing_crew]
+)
+```
+
+## Pipeline Methods
+
+| Method           | Description                                                                                                                                                                    |
+| :--------------- | :----------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| **process_runs** | Executes the pipeline, processing all stages and returning the results. This method initiates one or more runs through the pipeline, handling the flow of data between stages. |
+
+## Pipeline Output
+
+!!! note "Understanding Pipeline Outputs"
+The output of a pipeline in the crewAI framework is encapsulated within two main classes: `PipelineOutput` and `PipelineRunResult`. These classes provide a structured way to access the results of the pipeline's execution, including various formats such as raw strings, JSON, and Pydantic models.
+
+### Pipeline Output Attributes
+
+| Attribute       | Parameters    | Type                      | Description                                                                                               |
+| :-------------- | :------------ | :------------------------ | :-------------------------------------------------------------------------------------------------------- |
+| **ID**          | `id`          | `UUID4`                   | A unique identifier for the pipeline output.                                                              |
+| **Run Results** | `run_results` | `List[PipelineRunResult]` | A list of `PipelineRunResult` objects, each representing the output of a single run through the pipeline. |
+
+### Pipeline Output Methods
+
+| Method/Property    | Description                                            |
+| :----------------- | :----------------------------------------------------- |
+| **add_run_result** | Adds a `PipelineRunResult` to the list of run results. |
+
+### Pipeline Run Result Attributes
+
+| Attribute         | Parameters      | Type                       | Description                                                                                   |
+| :---------------- | :-------------- | :------------------------- | :-------------------------------------------------------------------------------------------- |
+| **ID**            | `id`            | `UUID4`                    | A unique identifier for the run result.                                                       |
+| **Raw**           | `raw`           | `str`                      | The raw output of the final stage in the pipeline run.                                        |
+| **Pydantic**      | `pydantic`      | `Optional[BaseModel]`      | A Pydantic model object representing the structured output of the final stage, if applicable. |
+| **JSON Dict**     | `json_dict`     | `Optional[Dict[str, Any]]` | A dictionary representing the JSON output of the final stage, if applicable.                  |
+| **Token Usage**   | `token_usage`   | `Dict[str, Any]`           | A summary of token usage across all stages of the pipeline run.                               |
+| **Trace**         | `trace`         | `List[Any]`                | A trace of the journey of inputs through the pipeline run.                                    |
+| **Crews Outputs** | `crews_outputs` | `List[CrewOutput]`         | A list of `CrewOutput` objects, representing the outputs from each crew in the pipeline run.  |
+
+### Pipeline Run Result Methods and Properties
+
+| Method/Property | Description                                                                                              |
+| :-------------- | :------------------------------------------------------------------------------------------------------- |
+| **json**        | Returns the JSON string representation of the run result if the output format of the final task is JSON. |
+| **to_dict**     | Converts the JSON and Pydantic outputs to a dictionary.                                                  |
+| \***\*str\*\*** | Returns the string representation of the run result, prioritizing Pydantic, then JSON, then raw.         |
+
+### Accessing Pipeline Outputs
+
+Once a pipeline has been executed, its output can be accessed through the `PipelineOutput` object returned by the `process_runs` method. The `PipelineOutput` class provides access to individual `PipelineRunResult` objects, each representing a single run through the pipeline.
+
+#### Example
+
+```python
+# Define input data for the pipeline
+input_data = [{"initial_query": "Latest advancements in AI"}, {"initial_query": "Future of robotics"}]
+
+# Execute the pipeline
+pipeline_output = await my_pipeline.process_runs(input_data)
+
+# Access the results
+for run_result in pipeline_output.run_results:
+    print(f"Run ID: {run_result.id}")
+    print(f"Final Raw Output: {run_result.raw}")
+    if run_result.json_dict:
+        print(f"JSON Output: {json.dumps(run_result.json_dict, indent=2)}")
+    if run_result.pydantic:
+        print(f"Pydantic Output: {run_result.pydantic}")
+    print(f"Token Usage: {run_result.token_usage}")
+    print(f"Trace: {run_result.trace}")
+    print("Crew Outputs:")
+    for crew_output in run_result.crews_outputs:
+        print(f"  Crew: {crew_output.raw}")
+    print("\n")
+```
+
+This example demonstrates how to access and work with the pipeline output, including individual run results and their associated data.
+
+## Using Pipelines
+
+Pipelines are particularly useful for complex workflows that involve multiple stages of processing, analysis, or content generation. They allow you to:
+
+1. **Sequence Operations**: Execute crews in a specific order, ensuring that the output of one crew is available as input to the next.
+2. **Parallel Processing**: Run multiple crews concurrently within a stage for increased efficiency.
+3. **Manage Complex Workflows**: Break down large tasks into smaller, manageable steps executed by specialized crews.
+
+### Example: Running a Pipeline
+
+```python
+# Define input data for the pipeline
+input_data = [{"initial_query": "Latest advancements in AI"}]
+
+# Execute the pipeline, initiating a run for each input
+results = await my_pipeline.process_runs(input_data)
+
+# Access the results
+for result in results:
+    print(f"Final Output: {result.raw}")
+    print(f"Token Usage: {result.token_usage}")
+    print(f"Trace: {result.trace}")  # Shows the path of the input through all stages
+```
+
+## Advanced Features
+
+### Parallel Execution within Stages
+
+You can define parallel execution within a stage by providing a list of crews, creating multiple branches:
+
+```python
+parallel_analysis_crew = Crew(agents=[financial_analyst], tasks=[financial_analysis_task])
+market_analysis_crew = Crew(agents=[market_analyst], tasks=[market_analysis_task])
+
+my_pipeline = Pipeline(
+    stages=[
+        research_crew,
+        [parallel_analysis_crew, market_analysis_crew],  # Parallel execution (branching)
+        writing_crew
+    ]
+)
+```
+
+### Error Handling and Validation
+
+The Pipeline class includes validation mechanisms to ensure the robustness of the pipeline structure:
+
+- Validates that stages contain only Crew instances or lists of Crew instances.
+- Prevents double nesting of stages to maintain a clear structure.

docs/index.md

@@ -46,6 +46,11 @@ Cutting-edge framework for orchestrating role-playing, autonomous AI agents. By
                     Crews
                 </a>
             </li>
+            <li>
+                <a href="./core-concepts/Pipeline">
+                    Pipeline
+                </a>
+            </li>
             <li>
                 <a href="./core-concepts/Training-Crew">
                     Training

poetry.lock

@@ -1,14 +1,14 @@
-# This file is automatically @generated by Poetry 1.8.3 and should not be changed by hand.
+# This file is automatically @generated by Poetry 1.7.1 and should not be changed by hand.
 
 [[package]]
 name = "agentops"
-version = "0.3.2"
+version = "0.3.4"
 description = "Python SDK for developing AI agent evals and observability"
 optional = true
 python-versions = ">=3.7"
 files = [
-    {file = "agentops-0.3.2-py3-none-any.whl", hash = "sha256:b35988e04378624204572bb3d7a454094f879ea573f05b57d4e75ab0bfbb82af"},
-    {file = "agentops-0.3.2.tar.gz", hash = "sha256:55559ac4a43634831dfa8937c2597c28e332809dc7c6bb3bc3c8b233442e224c"},
+    {file = "agentops-0.3.4-py3-none-any.whl", hash = "sha256:126f7aed4ba43c1399b5488d67a03d10cb4c531e619c650776f826ca00c1aa24"},
+    {file = "agentops-0.3.4.tar.gz", hash = "sha256:a92c9cb7c511197f0ecb8cb5aca15d35022c15a3d2fd2aaaa34cd7e5dc59393f"},
 ]
 
 [package.dependencies]
@@ -355,17 +355,17 @@ lxml = ["lxml"]
 
 [[package]]
 name = "boto3"
-version = "1.34.146"
+version = "1.34.149"
 description = "The AWS SDK for Python"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "boto3-1.34.146-py3-none-any.whl", hash = "sha256:7ec568fb19bce82a70be51f08fddac1ef927ca3fb0896cbb34303a012ba228d8"},
-    {file = "boto3-1.34.146.tar.gz", hash = "sha256:5686fe2a6d1aa1de8a88e9589cdcc33361640d3d7a13da718a30717248886124"},
+    {file = "boto3-1.34.149-py3-none-any.whl", hash = "sha256:11edeeacdd517bda3b7615b754d8440820cdc9ddd66794cc995a9693ddeaa3be"},
+    {file = "boto3-1.34.149.tar.gz", hash = "sha256:f4e6489ba9dc7fb37d53e0e82dbc97f2cb0a4969ef3970e2c88b8f94023ae81a"},
 ]
 
 [package.dependencies]
-botocore = ">=1.34.146,<1.35.0"
+botocore = ">=1.34.149,<1.35.0"
 jmespath = ">=0.7.1,<2.0.0"
 s3transfer = ">=0.10.0,<0.11.0"
 
@@ -374,13 +374,13 @@ crt = ["botocore[crt] (>=1.21.0,<2.0a0)"]
 
 [[package]]
 name = "botocore"
-version = "1.34.146"
+version = "1.34.149"
 description = "Low-level, data-driven core of boto 3."
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "botocore-1.34.146-py3-none-any.whl", hash = "sha256:3fd4782362bd29c192704ebf859c5c8c5189ad05719e391eefe23088434427ae"},
-    {file = "botocore-1.34.146.tar.gz", hash = "sha256:849cb8e54e042443aeabcd7822b5f2b76cb5cfe33fe3a71f91c7c069748a869c"},
+    {file = "botocore-1.34.149-py3-none-any.whl", hash = "sha256:ae6c4be52eeee96f68c116b27d252bab069cd046d61a17cfe8e9da411cf22906"},
+    {file = "botocore-1.34.149.tar.gz", hash = "sha256:2e1eb5ef40102a3d796bb3dd05f2ac5e8fb43fe1ff114b4f6d33153437f5a372"},
 ]
 
 [package.dependencies]
@@ -1012,13 +1012,13 @@ idna = ">=2.0.0"
 
 [[package]]
 name = "embedchain"
-version = "0.1.118"
+version = "0.1.119"
 description = "Simplest open source retrieval (RAG) framework"
 optional = false
 python-versions = "<=3.13,>=3.9"
 files = [
-    {file = "embedchain-0.1.118-py3-none-any.whl", hash = "sha256:38ead471df9d9234bf42e6f7a32cab26431d50d6f2f894f18a6cabc0b02bf31a"},
-    {file = "embedchain-0.1.118.tar.gz", hash = "sha256:1fa1e799882a1dc4e63af344595b043f1c1f30fbd59461b6660b1934b85a1e4b"},
+    {file = "embedchain-0.1.119-py3-none-any.whl", hash = "sha256:8ec3e7f139939fa1dc8fda898f8d8d9d31a5abfe08e184b607e38733d863d606"},
+    {file = "embedchain-0.1.119.tar.gz", hash = "sha256:0f4f45e092b7f3192ea6fe82575726532573b1231d7af6c22edc695b701b4223"},
 ]
 
 [package.dependencies]
@@ -1032,7 +1032,7 @@ langchain = ">0.2,<=0.3"
 langchain-cohere = ">=0.1.4,<0.2.0"
 langchain-community = ">=0.2.6,<0.3.0"
 langchain-openai = ">=0.1.7,<0.2.0"
-mem0ai = ">=0.0.5,<0.0.6"
+mem0ai = ">=0.0.9,<0.0.10"
 openai = ">=1.1.1"
 posthog = ">=3.0.2,<4.0.0"
 pypdf = ">=4.0.1,<5.0.0"
@@ -1061,20 +1061,6 @@ together = ["together (>=1.2.1,<2.0.0)"]
 vertexai = ["langchain-google-vertexai (>=1.0.6,<2.0.0)"]
 weaviate = ["weaviate-client (>=3.24.1,<4.0.0)"]
 
-[[package]]
-name = "eval-type-backport"
-version = "0.2.0"
-description = "Like `typing._eval_type`, but lets older Python versions use newer typing features."
-optional = false
-python-versions = ">=3.8"
-files = [
-    {file = "eval_type_backport-0.2.0-py3-none-any.whl", hash = "sha256:ac2f73d30d40c5a30a80b8739a789d6bb5e49fdffa66d7912667e2015d9c9933"},
-    {file = "eval_type_backport-0.2.0.tar.gz", hash = "sha256:68796cfbc7371ebf923f03bdf7bef415f3ec098aeced24e054b253a0e78f7b37"},
-]
-
-[package.extras]
-tests = ["pytest"]
-
 [[package]]
 name = "exceptiongroup"
 version = "1.2.2"
@@ -1402,13 +1388,13 @@ requests = ["requests (>=2.20.0,<3.0.0.dev0)"]
 
 [[package]]
 name = "google-cloud-aiplatform"
-version = "1.59.0"
+version = "1.60.0"
 description = "Vertex AI API client library"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "google-cloud-aiplatform-1.59.0.tar.gz", hash = "sha256:2bebb59c0ba3e3b4b568305418ca1b021977988adbee8691a5bed09b037e7e63"},
-    {file = "google_cloud_aiplatform-1.59.0-py2.py3-none-any.whl", hash = "sha256:549e6eb1844b0f853043309138ebe2db00de4bbd8197b3bde26804ac163ef52a"},
+    {file = "google-cloud-aiplatform-1.60.0.tar.gz", hash = "sha256:782c7f1ec0e77a7c7daabef3b65bfd506ed2b4b1dc2186753c43cd6faf8dd04e"},
+    {file = "google_cloud_aiplatform-1.60.0-py2.py3-none-any.whl", hash = "sha256:5f14159c9575f4b46335027e3ceb8fa57bd5eaa76a07f858105b8c6c034ec0d6"},
 ]
 
 [package.dependencies]
@@ -1430,8 +1416,8 @@ cloud-profiler = ["tensorboard-plugin-profile (>=2.4.0,<3.0.0dev)", "tensorflow
 datasets = ["pyarrow (>=10.0.1)", "pyarrow (>=14.0.0)", "pyarrow (>=3.0.0,<8.0dev)"]
 endpoint = ["requests (>=2.28.1)"]
 full = ["cloudpickle (<3.0)", "docker (>=5.0.3)", "explainable-ai-sdk (>=1.0.0)", "fastapi (>=0.71.0,<=0.109.1)", "google-cloud-bigquery", "google-cloud-bigquery-storage", "google-cloud-logging (<4.0)", "google-vizier (>=0.1.6)", "httpx (>=0.23.0,<0.25.0)", "immutabledict", "lit-nlp (==0.4.0)", "mlflow (>=1.27.0,<=2.1.1)", "numpy (>=1.15.0)", "pandas (>=1.0.0)", "pandas (>=1.0.0,<2.2.0)", "pyarrow (>=10.0.1)", "pyarrow (>=14.0.0)", "pyarrow (>=3.0.0,<8.0dev)", "pyarrow (>=6.0.1)", "pydantic (<2)", "pyyaml (>=5.3.1,<7)", "ray[default] (>=2.4,<2.5.dev0 || >2.9.0,!=2.9.1,!=2.9.2,<=2.9.3)", "ray[default] (>=2.5,<=2.9.3)", "requests (>=2.28.1)", "setuptools (<70.0.0)", "starlette (>=0.17.1)", "tensorboard-plugin-profile (>=2.4.0,<3.0.0dev)", "tensorflow (>=2.3.0,<3.0.0dev)", "tensorflow (>=2.3.0,<3.0.0dev)", "tensorflow (>=2.4.0,<3.0.0dev)", "tqdm (>=4.23.0)", "urllib3 (>=1.21.1,<1.27)", "uvicorn[standard] (>=0.16.0)", "werkzeug (>=2.0.0,<2.1.0dev)"]
-langchain = ["langchain (>=0.1.16,<0.3)", "langchain-core (<0.2)", "langchain-google-vertexai (<2)", "openinference-instrumentation-langchain (>=0.1.19,<0.2)", "tenacity (<=8.3)"]
-langchain-testing = ["absl-py", "cloudpickle (>=3.0,<4.0)", "langchain (>=0.1.16,<0.3)", "langchain-core (<0.2)", "langchain-google-vertexai (<2)", "openinference-instrumentation-langchain (>=0.1.19,<0.2)", "opentelemetry-exporter-gcp-trace (<2)", "opentelemetry-sdk (<2)", "pydantic (>=2.6.3,<3)", "pytest-xdist", "tenacity (<=8.3)"]
+langchain = ["langchain (>=0.1.16,<0.3)", "langchain-core (<0.3)", "langchain-google-vertexai (<2)", "openinference-instrumentation-langchain (>=0.1.19,<0.2)", "tenacity (<=8.3)"]
+langchain-testing = ["absl-py", "cloudpickle (>=3.0,<4.0)", "google-cloud-trace (<2)", "langchain (>=0.1.16,<0.3)", "langchain-core (<0.3)", "langchain-google-vertexai (<2)", "openinference-instrumentation-langchain (>=0.1.19,<0.2)", "opentelemetry-exporter-gcp-trace (<2)", "opentelemetry-sdk (<2)", "pydantic (>=2.6.3,<3)", "pytest-xdist", "tenacity (<=8.3)"]
 lit = ["explainable-ai-sdk (>=1.0.0)", "lit-nlp (==0.4.0)", "pandas (>=1.0.0)", "tensorflow (>=2.3.0,<3.0.0dev)"]
 metadata = ["numpy (>=1.15.0)", "pandas (>=1.0.0)"]
 pipelines = ["pyyaml (>=5.3.1,<7)"]
@@ -1441,7 +1427,7 @@ private-endpoints = ["requests (>=2.28.1)", "urllib3 (>=1.21.1,<1.27)"]
 rapid-evaluation = ["pandas (>=1.0.0,<2.2.0)", "tqdm (>=4.23.0)"]
 ray = ["google-cloud-bigquery", "google-cloud-bigquery-storage", "immutabledict", "pandas (>=1.0.0,<2.2.0)", "pyarrow (>=6.0.1)", "pydantic (<2)", "ray[default] (>=2.4,<2.5.dev0 || >2.9.0,!=2.9.1,!=2.9.2,<=2.9.3)", "ray[default] (>=2.5,<=2.9.3)", "setuptools (<70.0.0)"]
 ray-testing = ["google-cloud-bigquery", "google-cloud-bigquery-storage", "immutabledict", "pandas (>=1.0.0,<2.2.0)", "pyarrow (>=6.0.1)", "pydantic (<2)", "pytest-xdist", "ray[default] (>=2.4,<2.5.dev0 || >2.9.0,!=2.9.1,!=2.9.2,<=2.9.3)", "ray[default] (>=2.5,<=2.9.3)", "ray[train] (==2.9.3)", "scikit-learn", "setuptools (<70.0.0)", "tensorflow", "torch (>=2.0.0,<2.1.0)", "xgboost", "xgboost-ray"]
-reasoningengine = ["cloudpickle (>=3.0,<4.0)", "opentelemetry-exporter-gcp-trace (<2)", "opentelemetry-sdk (<2)", "pydantic (>=2.6.3,<3)"]
+reasoningengine = ["cloudpickle (>=3.0,<4.0)", "google-cloud-trace (<2)", "opentelemetry-exporter-gcp-trace (<2)", "opentelemetry-sdk (<2)", "pydantic (>=2.6.3,<3)"]
 tensorboard = ["tensorboard-plugin-profile (>=2.4.0,<3.0.0dev)", "tensorflow (>=2.3.0,<3.0.0dev)", "tensorflow (>=2.4.0,<3.0.0dev)", "werkzeug (>=2.0.0,<2.1.0dev)"]
 testing = ["bigframes", "cloudpickle (<3.0)", "docker (>=5.0.3)", "explainable-ai-sdk (>=1.0.0)", "fastapi (>=0.71.0,<=0.109.1)", "google-api-core (>=2.11,<3.0.0)", "google-cloud-bigquery", "google-cloud-bigquery-storage", "google-cloud-logging (<4.0)", "google-vizier (>=0.1.6)", "grpcio-testing", "httpx (>=0.23.0,<0.25.0)", "immutabledict", "ipython", "kfp (>=2.6.0,<3.0.0)", "lit-nlp (==0.4.0)", "mlflow (>=1.27.0,<=2.1.1)", "nltk", "numpy (>=1.15.0)", "pandas (>=1.0.0)", "pandas (>=1.0.0,<2.2.0)", "pyarrow (>=10.0.1)", "pyarrow (>=14.0.0)", "pyarrow (>=3.0.0,<8.0dev)", "pyarrow (>=6.0.1)", "pydantic (<2)", "pyfakefs", "pytest-asyncio", "pytest-xdist", "pyyaml (>=5.3.1,<7)", "ray[default] (>=2.4,<2.5.dev0 || >2.9.0,!=2.9.1,!=2.9.2,<=2.9.3)", "ray[default] (>=2.5,<=2.9.3)", "requests (>=2.28.1)", "requests-toolbelt (<1.0.0)", "scikit-learn", "sentencepiece (>=0.2.0)", "setuptools (<70.0.0)", "starlette (>=0.17.1)", "tensorboard-plugin-profile (>=2.4.0,<3.0.0dev)", "tensorflow (==2.13.0)", "tensorflow (==2.16.1)", "tensorflow (>=2.3.0,<3.0.0dev)", "tensorflow (>=2.3.0,<3.0.0dev)", "tensorflow (>=2.4.0,<3.0.0dev)", "torch (>=2.0.0,<2.1.0)", "torch (>=2.2.0)", "tqdm (>=4.23.0)", "urllib3 (>=1.21.1,<1.27)", "uvicorn[standard] (>=0.16.0)", "werkzeug (>=2.0.0,<2.1.0dev)", "xgboost"]
 tokenization = ["sentencepiece (>=0.2.0)"]
@@ -1756,25 +1742,6 @@ files = [
 backports-strenum = {version = ">=1.3", markers = "python_version < \"3.11\""}
 colorama = ">=0.4"
 
-[[package]]
-name = "groq"
-version = "0.9.0"
-description = "The official Python library for the groq API"
-optional = false
-python-versions = ">=3.7"
-files = [
-    {file = "groq-0.9.0-py3-none-any.whl", hash = "sha256:d0e46f4ad645504672bb09c8100af3ced3a7db0d5119dc13e4aca535fc455874"},
-    {file = "groq-0.9.0.tar.gz", hash = "sha256:130ed5e35d3acfaab46b9e7a078eeaebf91052f4a9d71f86f87fb319b5fec332"},
-]
-
-[package.dependencies]
-anyio = ">=3.5.0,<5"
-distro = ">=1.7.0,<2"
-httpx = ">=0.23.0,<1"
-pydantic = ">=1.9.0,<3"
-sniffio = "*"
-typing-extensions = ">=4.7,<5"
-
 [[package]]
 name = "grpc-google-iam-v1"
 version = "0.13.1"
@@ -2077,13 +2044,13 @@ files = [
 
 [[package]]
 name = "huggingface-hub"
-version = "0.24.0"
+version = "0.24.3"
 description = "Client library to download and publish models, datasets and other repos on the huggingface.co hub"
 optional = false
 python-versions = ">=3.8.0"
 files = [
-    {file = "huggingface_hub-0.24.0-py3-none-any.whl", hash = "sha256:7ad92edefb93d8145c061f6df8d99df2ff85f8379ba5fac8a95aca0642afa5d7"},
-    {file = "huggingface_hub-0.24.0.tar.gz", hash = "sha256:6c7092736b577d89d57b3cdfea026f1b0dc2234ae783fa0d59caf1bf7d52dfa7"},
+    {file = "huggingface_hub-0.24.3-py3-none-any.whl", hash = "sha256:69ecce486dd6cdad69937ba76779e893c224a670a9d947636c1d5cbd049e44d8"},
+    {file = "huggingface_hub-0.24.3.tar.gz", hash = "sha256:bfdc05cc9b64a0e24e8614a44222698799183268f6b68be209aa2df70cff2cde"},
 ]
 
 [package.dependencies]
@@ -2161,22 +2128,22 @@ files = [
 
 [[package]]
 name = "importlib-metadata"
-version = "7.1.0"
+version = "8.0.0"
 description = "Read metadata from Python packages"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "importlib_metadata-7.1.0-py3-none-any.whl", hash = "sha256:30962b96c0c223483ed6cc7280e7f0199feb01a0e40cfae4d4450fc6fab1f570"},
-    {file = "importlib_metadata-7.1.0.tar.gz", hash = "sha256:b78938b926ee8d5f020fc4772d487045805a55ddbad2ecf21c6d60938dc7fcd2"},
+    {file = "importlib_metadata-8.0.0-py3-none-any.whl", hash = "sha256:15584cf2b1bf449d98ff8a6ff1abef57bf20f3ac6454f431736cd3e660921b2f"},
+    {file = "importlib_metadata-8.0.0.tar.gz", hash = "sha256:188bd24e4c346d3f0a933f275c2fec67050326a856b9a359881d7c2a697e8812"},
 ]
 
 [package.dependencies]
 zipp = ">=0.5"
 
 [package.extras]
-docs = ["furo", "jaraco.packaging (>=9.3)", "jaraco.tidelift (>=1.4)", "rst.linker (>=1.9)", "sphinx (>=3.5)", "sphinx-lint"]
+doc = ["furo", "jaraco.packaging (>=9.3)", "jaraco.tidelift (>=1.4)", "rst.linker (>=1.9)", "sphinx (>=3.5)", "sphinx-lint"]
 perf = ["ipython"]
-testing = ["flufl.flake8", "importlib-resources (>=1.3)", "jaraco.test (>=5.4)", "packaging", "pyfakefs", "pytest (>=6)", "pytest-checkdocs (>=2.4)", "pytest-cov", "pytest-enabler (>=2.2)", "pytest-mypy", "pytest-perf (>=0.9.2)", "pytest-ruff (>=0.2.1)"]
+test = ["flufl.flake8", "importlib-resources (>=1.3)", "jaraco.test (>=5.4)", "packaging", "pyfakefs", "pytest (>=6,!=8.1.*)", "pytest-checkdocs (>=2.4)", "pytest-cov", "pytest-enabler (>=2.2)", "pytest-mypy", "pytest-perf (>=0.9.2)", "pytest-ruff (>=0.2.1)"]
 
 [[package]]
 name = "importlib-resources"
@@ -2456,19 +2423,19 @@ tests = ["aiohttp", "duckdb", "pandas (>=1.4)", "polars (>=0.19)", "pytest", "py
 
 [[package]]
 name = "langchain"
-version = "0.2.10"
+version = "0.2.11"
 description = "Building applications with LLMs through composability"
 optional = false
 python-versions = "<4.0,>=3.8.1"
 files = [
-    {file = "langchain-0.2.10-py3-none-any.whl", hash = "sha256:b4fb58c7faf4f4999cfe3325474979a7121a1737dd101655a723a1d957ef0617"},
-    {file = "langchain-0.2.10.tar.gz", hash = "sha256:1f861c1b59ac9c91b02bb0fa58d3adad1c1d0686636872b5b357bbce3ce41d06"},
+    {file = "langchain-0.2.11-py3-none-any.whl", hash = "sha256:5a7a8b4918f3d3bebce9b4f23b92d050699e6f7fb97591e8941177cf07a260a2"},
+    {file = "langchain-0.2.11.tar.gz", hash = "sha256:d7a9e4165f02dca0bd78addbc2319d5b9286b5d37c51d784124102b57e9fd297"},
 ]
 
 [package.dependencies]
 aiohttp = ">=3.8.3,<4.0.0"
 async-timeout = {version = ">=4.0.0,<5.0.0", markers = "python_version < \"3.11\""}
-langchain-core = ">=0.2.22,<0.3.0"
+langchain-core = ">=0.2.23,<0.3.0"
 langchain-text-splitters = ">=0.2.0,<0.3.0"
 langsmith = ">=0.1.17,<0.2.0"
 numpy = [
@@ -2504,20 +2471,20 @@ langchain-community = ["langchain-community (>=0.2.4)"]
 
 [[package]]
 name = "langchain-community"
-version = "0.2.9"
+version = "0.2.10"
 description = "Community contributed LangChain integrations."
 optional = false
 python-versions = "<4.0,>=3.8.1"
 files = [
-    {file = "langchain_community-0.2.9-py3-none-any.whl", hash = "sha256:b51d3adf9346a1161c1098917585b9e303cf24e2f5c71f5d232a0504edada5f2"},
-    {file = "langchain_community-0.2.9.tar.gz", hash = "sha256:1e7c180232916cbe35fe00509680dd1f805e32d7c87b5e80b3a9ec8754ecae37"},
+    {file = "langchain_community-0.2.10-py3-none-any.whl", hash = "sha256:9f4d1b5ab7f0b0a704f538e26e50fce45a461da6d2bf6b7b636d24f22fbc088a"},
+    {file = "langchain_community-0.2.10.tar.gz", hash = "sha256:3a0404bad4bd07d6f86affdb62fb3d080a456c66191754d586a409d9d6024d62"},
 ]
 
 [package.dependencies]
 aiohttp = ">=3.8.3,<4.0.0"
 dataclasses-json = ">=0.5.7,<0.7"
 langchain = ">=0.2.9,<0.3.0"
-langchain-core = ">=0.2.22,<0.3.0"
+langchain-core = ">=0.2.23,<0.3.0"
 langsmith = ">=0.1.0,<0.2.0"
 numpy = [
     {version = ">=1,<2", markers = "python_version < \"3.12\""},
@@ -2530,13 +2497,13 @@ tenacity = ">=8.1.0,<8.4.0 || >8.4.0,<9.0.0"
 
 [[package]]
 name = "langchain-core"
-version = "0.2.22"
+version = "0.2.24"
 description = "Building applications with LLMs through composability"
 optional = false
 python-versions = "<4.0,>=3.8.1"
 files = [
-    {file = "langchain_core-0.2.22-py3-none-any.whl", hash = "sha256:7731a86440c0958b3186c003fb9b26b2d5a682a6344bda7bfb9174e2898f8b43"},
-    {file = "langchain_core-0.2.22.tar.gz", hash = "sha256:582d6f929a43b830139444e4124123cd415331ad62f25757b1406252958cdcac"},
+    {file = "langchain_core-0.2.24-py3-none-any.whl", hash = "sha256:9444fc082d21ef075d925590a684a73fe1f9688a3d90087580ec929751be55e7"},
+    {file = "langchain_core-0.2.24.tar.gz", hash = "sha256:f2e3fa200b124e8c45d270da9bf836bed9c09532612c96ff3225e59b9a232f5a"},
 ]
 
 [package.dependencies]
@@ -2552,13 +2519,13 @@ tenacity = ">=8.1.0,<8.4.0 || >8.4.0,<9.0.0"
 
 [[package]]
 name = "langchain-experimental"
-version = "0.0.62"
+version = "0.0.63"
 description = "Building applications with LLMs through composability"
 optional = false
 python-versions = "<4.0,>=3.8.1"
 files = [
-    {file = "langchain_experimental-0.0.62-py3-none-any.whl", hash = "sha256:9240f9e3490e819976f20a37863970036e7baacb7104b9eb6833d19ab6d518c9"},
-    {file = "langchain_experimental-0.0.62.tar.gz", hash = "sha256:9737fbc8429d24457ea4d368e3c9ba9ed1cace0564fb5f1a96a3027a588bd0ac"},
+    {file = "langchain_experimental-0.0.63-py3-none-any.whl", hash = "sha256:cb4ae7a685bb3c077d138b4533ed02e8df1f5f784333c3e52dcae8c80f031ca2"},
+    {file = "langchain_experimental-0.0.63.tar.gz", hash = "sha256:fc894599bfac43445004a9ff60d9a28751426b2fea1979e4b2fa453c847850c4"},
 ]
 
 [package.dependencies]
@@ -2567,17 +2534,17 @@ langchain-core = ">=0.2.10,<0.3.0"
 
 [[package]]
 name = "langchain-openai"
-version = "0.1.17"
+version = "0.1.19"
 description = "An integration package connecting OpenAI and LangChain"
 optional = false
 python-versions = "<4.0,>=3.8.1"
 files = [
-    {file = "langchain_openai-0.1.17-py3-none-any.whl", hash = "sha256:30bef5574ecbbbb91b8025b2dc5a1bd81fd62157d3ad1a35d820141f31c5b443"},
-    {file = "langchain_openai-0.1.17.tar.gz", hash = "sha256:c5d70ddecdcb93e146f376bdbadbb6ec69de9ac0f402cd5b83de50b655ba85ee"},
+    {file = "langchain_openai-0.1.19-py3-none-any.whl", hash = "sha256:a7a739f1469d54cd988865420e7fc21b50fb93727b2e6da5ad30273fc61ecf19"},
+    {file = "langchain_openai-0.1.19.tar.gz", hash = "sha256:3bf342bb302d1444f4abafdf01c467dbd9b248497e1133808c4bae70396c79b3"},
 ]
 
 [package.dependencies]
-langchain-core = ">=0.2.20,<0.3.0"
+langchain-core = ">=0.2.24,<0.3.0"
 openai = ">=1.32.0,<2.0.0"
 tiktoken = ">=0.7,<1"
 
@@ -2773,23 +2740,20 @@ files = [
 
 [[package]]
 name = "mem0ai"
-version = "0.0.5"
+version = "0.0.9"
 description = "Long-term memory for AI Agents"
 optional = false
 python-versions = "<4.0,>=3.8"
 files = [
-    {file = "mem0ai-0.0.5-py3-none-any.whl", hash = "sha256:6f6e5356fd522adf0510322cd581476ea456fd7ccefca11b5ac050e9a6f00f36"},
-    {file = "mem0ai-0.0.5.tar.gz", hash = "sha256:f2ac35d15e4e620becb8d06b8ebeb1ffa85fac0b7cb2d3138056babec48dd5dd"},
+    {file = "mem0ai-0.0.9-py3-none-any.whl", hash = "sha256:d4de435729af4fd3d597d022ffb2af89a0630d6c3b4769792bbe27d2ce816858"},
+    {file = "mem0ai-0.0.9.tar.gz", hash = "sha256:e4374d5d04aa3f543cd3325f700e4b62f5358ae1c6fa5c44b2ff790c10c4e5f1"},
 ]
 
 [package.dependencies]
-boto3 = ">=1.34.144,<2.0.0"
-groq = ">=0.9.0,<0.10.0"
 openai = ">=1.33.0,<2.0.0"
 posthog = ">=3.5.0,<4.0.0"
 pydantic = ">=2.7.3,<3.0.0"
 qdrant-client = ">=1.9.1,<2.0.0"
-together = ">=1.2.1,<2.0.0"
 
 [[package]]
 name = "mergedeep"
@@ -3338,13 +3302,13 @@ sympy = "*"
 
 [[package]]
 name = "openai"
-version = "1.37.0"
+version = "1.37.1"
 description = "The official Python library for the openai API"
 optional = false
 python-versions = ">=3.7.1"
 files = [
-    {file = "openai-1.37.0-py3-none-any.whl", hash = "sha256:a903245c0ecf622f2830024acdaa78683c70abb8e9d37a497b851670864c9f73"},
-    {file = "openai-1.37.0.tar.gz", hash = "sha256:dc8197fc40ab9d431777b6620d962cc49f4544ffc3011f03ce0a805e6eb54adb"},
+    {file = "openai-1.37.1-py3-none-any.whl", hash = "sha256:9a6adda0d6ae8fce02d235c5671c399cfa40d6a281b3628914c7ebf244888ee3"},
+    {file = "openai-1.37.1.tar.gz", hash = "sha256:faf87206785a6b5d9e34555d6a3242482a6852bc802e453e2a891f68ee04ce55"},
 ]
 
 [package.dependencies]
@@ -3361,42 +3325,42 @@ datalib = ["numpy (>=1)", "pandas (>=1.2.3)", "pandas-stubs (>=1.1.0.11)"]
 
 [[package]]
 name = "opentelemetry-api"
-version = "1.25.0"
+version = "1.26.0"
 description = "OpenTelemetry Python API"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "opentelemetry_api-1.25.0-py3-none-any.whl", hash = "sha256:757fa1aa020a0f8fa139f8959e53dec2051cc26b832e76fa839a6d76ecefd737"},
-    {file = "opentelemetry_api-1.25.0.tar.gz", hash = "sha256:77c4985f62f2614e42ce77ee4c9da5fa5f0bc1e1821085e9a47533a9323ae869"},
+    {file = "opentelemetry_api-1.26.0-py3-none-any.whl", hash = "sha256:7d7ea33adf2ceda2dd680b18b1677e4152000b37ca76e679da71ff103b943064"},
+    {file = "opentelemetry_api-1.26.0.tar.gz", hash = "sha256:2bd639e4bed5b18486fef0b5a520aaffde5a18fc225e808a1ac4df363f43a1ce"},
 ]
 
 [package.dependencies]
 deprecated = ">=1.2.6"
-importlib-metadata = ">=6.0,<=7.1"
+importlib-metadata = ">=6.0,<=8.0.0"
 
 [[package]]
 name = "opentelemetry-exporter-otlp-proto-common"
-version = "1.25.0"
+version = "1.26.0"
 description = "OpenTelemetry Protobuf encoding"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "opentelemetry_exporter_otlp_proto_common-1.25.0-py3-none-any.whl", hash = "sha256:15637b7d580c2675f70246563363775b4e6de947871e01d0f4e3881d1848d693"},
-    {file = "opentelemetry_exporter_otlp_proto_common-1.25.0.tar.gz", hash = "sha256:c93f4e30da4eee02bacd1e004eb82ce4da143a2f8e15b987a9f603e0a85407d3"},
+    {file = "opentelemetry_exporter_otlp_proto_common-1.26.0-py3-none-any.whl", hash = "sha256:ee4d8f8891a1b9c372abf8d109409e5b81947cf66423fd998e56880057afbc71"},
+    {file = "opentelemetry_exporter_otlp_proto_common-1.26.0.tar.gz", hash = "sha256:bdbe50e2e22a1c71acaa0c8ba6efaadd58882e5a5978737a44a4c4b10d304c92"},
 ]
 
 [package.dependencies]
-opentelemetry-proto = "1.25.0"
+opentelemetry-proto = "1.26.0"
 
 [[package]]
 name = "opentelemetry-exporter-otlp-proto-grpc"
-version = "1.25.0"
+version = "1.26.0"
 description = "OpenTelemetry Collector Protobuf over gRPC Exporter"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "opentelemetry_exporter_otlp_proto_grpc-1.25.0-py3-none-any.whl", hash = "sha256:3131028f0c0a155a64c430ca600fd658e8e37043cb13209f0109db5c1a3e4eb4"},
-    {file = "opentelemetry_exporter_otlp_proto_grpc-1.25.0.tar.gz", hash = "sha256:c0b1661415acec5af87625587efa1ccab68b873745ca0ee96b69bb1042087eac"},
+    {file = "opentelemetry_exporter_otlp_proto_grpc-1.26.0-py3-none-any.whl", hash = "sha256:e2be5eff72ebcb010675b818e8d7c2e7d61ec451755b8de67a140bc49b9b0280"},
+    {file = "opentelemetry_exporter_otlp_proto_grpc-1.26.0.tar.gz", hash = "sha256:a65b67a9a6b06ba1ec406114568e21afe88c1cdb29c464f2507d529eb906d8ae"},
 ]
 
 [package.dependencies]
@@ -3404,39 +3368,39 @@ deprecated = ">=1.2.6"
 googleapis-common-protos = ">=1.52,<2.0"
 grpcio = ">=1.0.0,<2.0.0"
 opentelemetry-api = ">=1.15,<2.0"
-opentelemetry-exporter-otlp-proto-common = "1.25.0"
-opentelemetry-proto = "1.25.0"
-opentelemetry-sdk = ">=1.25.0,<1.26.0"
+opentelemetry-exporter-otlp-proto-common = "1.26.0"
+opentelemetry-proto = "1.26.0"
+opentelemetry-sdk = ">=1.26.0,<1.27.0"
 
 [[package]]
 name = "opentelemetry-exporter-otlp-proto-http"
-version = "1.25.0"
+version = "1.26.0"
 description = "OpenTelemetry Collector Protobuf over HTTP Exporter"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "opentelemetry_exporter_otlp_proto_http-1.25.0-py3-none-any.whl", hash = "sha256:2eca686ee11b27acd28198b3ea5e5863a53d1266b91cda47c839d95d5e0541a6"},
-    {file = "opentelemetry_exporter_otlp_proto_http-1.25.0.tar.gz", hash = "sha256:9f8723859e37c75183ea7afa73a3542f01d0fd274a5b97487ea24cb683d7d684"},
+    {file = "opentelemetry_exporter_otlp_proto_http-1.26.0-py3-none-any.whl", hash = "sha256:ee72a87c48ec977421b02f16c52ea8d884122470e0be573905237b540f4ee562"},
+    {file = "opentelemetry_exporter_otlp_proto_http-1.26.0.tar.gz", hash = "sha256:5801ebbcf7b527377883e6cbbdda35ee712dc55114fff1e93dfee210be56c908"},
 ]
 
 [package.dependencies]
 deprecated = ">=1.2.6"
 googleapis-common-protos = ">=1.52,<2.0"
 opentelemetry-api = ">=1.15,<2.0"
-opentelemetry-exporter-otlp-proto-common = "1.25.0"
-opentelemetry-proto = "1.25.0"
-opentelemetry-sdk = ">=1.25.0,<1.26.0"
+opentelemetry-exporter-otlp-proto-common = "1.26.0"
+opentelemetry-proto = "1.26.0"
+opentelemetry-sdk = ">=1.26.0,<1.27.0"
 requests = ">=2.7,<3.0"
 
 [[package]]
 name = "opentelemetry-instrumentation"
-version = "0.46b0"
+version = "0.47b0"
 description = "Instrumentation Tools & Auto Instrumentation for OpenTelemetry Python"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "opentelemetry_instrumentation-0.46b0-py3-none-any.whl", hash = "sha256:89cd721b9c18c014ca848ccd11181e6b3fd3f6c7669e35d59c48dc527408c18b"},
-    {file = "opentelemetry_instrumentation-0.46b0.tar.gz", hash = "sha256:974e0888fb2a1e01c38fbacc9483d024bb1132aad92d6d24e2e5543887a7adda"},
+    {file = "opentelemetry_instrumentation-0.47b0-py3-none-any.whl", hash = "sha256:88974ee52b1db08fc298334b51c19d47e53099c33740e48c4f084bd1afd052d5"},
+    {file = "opentelemetry_instrumentation-0.47b0.tar.gz", hash = "sha256:96f9885e450c35e3f16a4f33145f2ebf620aea910c9fd74a392bbc0f807a350f"},
 ]
 
 [package.dependencies]
@@ -3446,55 +3410,55 @@ wrapt = ">=1.0.0,<2.0.0"
 
 [[package]]
 name = "opentelemetry-instrumentation-asgi"
-version = "0.46b0"
+version = "0.47b0"
 description = "ASGI instrumentation for OpenTelemetry"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "opentelemetry_instrumentation_asgi-0.46b0-py3-none-any.whl", hash = "sha256:f13c55c852689573057837a9500aeeffc010c4ba59933c322e8f866573374759"},
-    {file = "opentelemetry_instrumentation_asgi-0.46b0.tar.gz", hash = "sha256:02559f30cf4b7e2a737ab17eb52aa0779bcf4cc06573064f3e2cb4dcc7d3040a"},
+    {file = "opentelemetry_instrumentation_asgi-0.47b0-py3-none-any.whl", hash = "sha256:b798dc4957b3edc9dfecb47a4c05809036a4b762234c5071212fda39ead80ade"},
+    {file = "opentelemetry_instrumentation_asgi-0.47b0.tar.gz", hash = "sha256:e78b7822c1bca0511e5e9610ec484b8994a81670375e570c76f06f69af7c506a"},
 ]
 
 [package.dependencies]
 asgiref = ">=3.0,<4.0"
 opentelemetry-api = ">=1.12,<2.0"
-opentelemetry-instrumentation = "0.46b0"
-opentelemetry-semantic-conventions = "0.46b0"
-opentelemetry-util-http = "0.46b0"
+opentelemetry-instrumentation = "0.47b0"
+opentelemetry-semantic-conventions = "0.47b0"
+opentelemetry-util-http = "0.47b0"
 
 [package.extras]
 instruments = ["asgiref (>=3.0,<4.0)"]
 
 [[package]]
 name = "opentelemetry-instrumentation-fastapi"
-version = "0.46b0"
+version = "0.47b0"
 description = "OpenTelemetry FastAPI Instrumentation"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "opentelemetry_instrumentation_fastapi-0.46b0-py3-none-any.whl", hash = "sha256:e0f5d150c6c36833dd011f0e6ef5ede6d7406c1aed0c7c98b2d3b38a018d1b33"},
-    {file = "opentelemetry_instrumentation_fastapi-0.46b0.tar.gz", hash = "sha256:928a883a36fc89f9702f15edce43d1a7104da93d740281e32d50ffd03dbb4365"},
+    {file = "opentelemetry_instrumentation_fastapi-0.47b0-py3-none-any.whl", hash = "sha256:5ac28dd401160b02e4f544a85a9e4f61a8cbe5b077ea0379d411615376a2bd21"},
+    {file = "opentelemetry_instrumentation_fastapi-0.47b0.tar.gz", hash = "sha256:0c7c10b5d971e99a420678ffd16c5b1ea4f0db3b31b62faf305fbb03b4ebee36"},
 ]
 
 [package.dependencies]
 opentelemetry-api = ">=1.12,<2.0"
-opentelemetry-instrumentation = "0.46b0"
-opentelemetry-instrumentation-asgi = "0.46b0"
-opentelemetry-semantic-conventions = "0.46b0"
-opentelemetry-util-http = "0.46b0"
+opentelemetry-instrumentation = "0.47b0"
+opentelemetry-instrumentation-asgi = "0.47b0"
+opentelemetry-semantic-conventions = "0.47b0"
+opentelemetry-util-http = "0.47b0"
 
 [package.extras]
-instruments = ["fastapi (>=0.58,<1.0)"]
+instruments = ["fastapi (>=0.58,<1.0)", "fastapi-slim (>=0.111.0,<0.112.0)"]
 
 [[package]]
 name = "opentelemetry-proto"
-version = "1.25.0"
+version = "1.26.0"
 description = "OpenTelemetry Python Proto"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "opentelemetry_proto-1.25.0-py3-none-any.whl", hash = "sha256:f07e3341c78d835d9b86665903b199893befa5e98866f63d22b00d0b7ca4972f"},
-    {file = "opentelemetry_proto-1.25.0.tar.gz", hash = "sha256:35b6ef9dc4a9f7853ecc5006738ad40443701e52c26099e197895cbda8b815a3"},
+    {file = "opentelemetry_proto-1.26.0-py3-none-any.whl", hash = "sha256:6c4d7b4d4d9c88543bcf8c28ae3f8f0448a753dc291c18c5390444c90b76a725"},
+    {file = "opentelemetry_proto-1.26.0.tar.gz", hash = "sha256:c5c18796c0cab3751fc3b98dee53855835e90c0422924b484432ac852d93dc1e"},
 ]
 
 [package.dependencies]
@@ -3502,43 +3466,44 @@ protobuf = ">=3.19,<5.0"
 
 [[package]]
 name = "opentelemetry-sdk"
-version = "1.25.0"
+version = "1.26.0"
 description = "OpenTelemetry Python SDK"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "opentelemetry_sdk-1.25.0-py3-none-any.whl", hash = "sha256:d97ff7ec4b351692e9d5a15af570c693b8715ad78b8aafbec5c7100fe966b4c9"},
-    {file = "opentelemetry_sdk-1.25.0.tar.gz", hash = "sha256:ce7fc319c57707ef5bf8b74fb9f8ebdb8bfafbe11898410e0d2a761d08a98ec7"},
+    {file = "opentelemetry_sdk-1.26.0-py3-none-any.whl", hash = "sha256:feb5056a84a88670c041ea0ded9921fca559efec03905dddeb3885525e0af897"},
+    {file = "opentelemetry_sdk-1.26.0.tar.gz", hash = "sha256:c90d2868f8805619535c05562d699e2f4fb1f00dbd55a86dcefca4da6fa02f85"},
 ]
 
 [package.dependencies]
-opentelemetry-api = "1.25.0"
-opentelemetry-semantic-conventions = "0.46b0"
+opentelemetry-api = "1.26.0"
+opentelemetry-semantic-conventions = "0.47b0"
 typing-extensions = ">=3.7.4"
 
 [[package]]
 name = "opentelemetry-semantic-conventions"
-version = "0.46b0"
+version = "0.47b0"
 description = "OpenTelemetry Semantic Conventions"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "opentelemetry_semantic_conventions-0.46b0-py3-none-any.whl", hash = "sha256:6daef4ef9fa51d51855d9f8e0ccd3a1bd59e0e545abe99ac6203804e36ab3e07"},
-    {file = "opentelemetry_semantic_conventions-0.46b0.tar.gz", hash = "sha256:fbc982ecbb6a6e90869b15c1673be90bd18c8a56ff1cffc0864e38e2edffaefa"},
+    {file = "opentelemetry_semantic_conventions-0.47b0-py3-none-any.whl", hash = "sha256:4ff9d595b85a59c1c1413f02bba320ce7ea6bf9e2ead2b0913c4395c7bbc1063"},
+    {file = "opentelemetry_semantic_conventions-0.47b0.tar.gz", hash = "sha256:a8d57999bbe3495ffd4d510de26a97dadc1dace53e0275001b2c1b2f67992a7e"},
 ]
 
 [package.dependencies]
-opentelemetry-api = "1.25.0"
+deprecated = ">=1.2.6"
+opentelemetry-api = "1.26.0"
 
 [[package]]
 name = "opentelemetry-util-http"
-version = "0.46b0"
+version = "0.47b0"
 description = "Web util for OpenTelemetry"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "opentelemetry_util_http-0.46b0-py3-none-any.whl", hash = "sha256:8dc1949ce63caef08db84ae977fdc1848fe6dc38e6bbaad0ae3e6ecd0d451629"},
-    {file = "opentelemetry_util_http-0.46b0.tar.gz", hash = "sha256:03b6e222642f9c7eae58d9132343e045b50aca9761fcb53709bd2b663571fdf6"},
+    {file = "opentelemetry_util_http-0.47b0-py3-none-any.whl", hash = "sha256:3d3215e09c4a723b12da6d0233a31395aeb2bb33a64d7b15a1500690ba250f19"},
+    {file = "opentelemetry_util_http-0.47b0.tar.gz", hash = "sha256:352a07664c18eef827eb8ddcbd64c64a7284a39dd1655e2f16f577eb046ccb32"},
 ]
 
 [[package]]
@@ -3917,13 +3882,13 @@ test = ["coverage", "flake8", "freezegun (==0.3.15)", "mock (>=2.0.0)", "pylint"
 
 [[package]]
 name = "pre-commit"
-version = "3.7.1"
+version = "3.8.0"
 description = "A framework for managing and maintaining multi-language pre-commit hooks."
 optional = false
 python-versions = ">=3.9"
 files = [
-    {file = "pre_commit-3.7.1-py2.py3-none-any.whl", hash = "sha256:fae36fd1d7ad7d6a5a1c0b0d5adb2ed1a3bda5a21bf6c3e5372073d7a11cd4c5"},
-    {file = "pre_commit-3.7.1.tar.gz", hash = "sha256:8ca3ad567bc78a4972a3f1a477e94a79d4597e8140a6e0b651c5e33899c3654a"},
+    {file = "pre_commit-3.8.0-py2.py3-none-any.whl", hash = "sha256:9a90a53bf82fdd8778d58085faf8d83df56e40dfe18f45b19446e26bf1b3a63f"},
+    {file = "pre_commit-3.8.0.tar.gz", hash = "sha256:8bb6494d4a20423842e198980c9ecf9f96607a07ea29549e180eef9ae80fe7af"},
 ]
 
 [package.dependencies]
@@ -3952,22 +3917,22 @@ testing = ["google-api-core (>=1.31.5)"]
 
 [[package]]
 name = "protobuf"
-version = "4.25.3"
+version = "4.25.4"
 description = ""
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "protobuf-4.25.3-cp310-abi3-win32.whl", hash = "sha256:d4198877797a83cbfe9bffa3803602bbe1625dc30d8a097365dbc762e5790faa"},
-    {file = "protobuf-4.25.3-cp310-abi3-win_amd64.whl", hash = "sha256:209ba4cc916bab46f64e56b85b090607a676f66b473e6b762e6f1d9d591eb2e8"},
-    {file = "protobuf-4.25.3-cp37-abi3-macosx_10_9_universal2.whl", hash = "sha256:f1279ab38ecbfae7e456a108c5c0681e4956d5b1090027c1de0f934dfdb4b35c"},
-    {file = "protobuf-4.25.3-cp37-abi3-manylinux2014_aarch64.whl", hash = "sha256:e7cb0ae90dd83727f0c0718634ed56837bfeeee29a5f82a7514c03ee1364c019"},
-    {file = "protobuf-4.25.3-cp37-abi3-manylinux2014_x86_64.whl", hash = "sha256:7c8daa26095f82482307bc717364e7c13f4f1c99659be82890dcfc215194554d"},
-    {file = "protobuf-4.25.3-cp38-cp38-win32.whl", hash = "sha256:f4f118245c4a087776e0a8408be33cf09f6c547442c00395fbfb116fac2f8ac2"},
-    {file = "protobuf-4.25.3-cp38-cp38-win_amd64.whl", hash = "sha256:c053062984e61144385022e53678fbded7aea14ebb3e0305ae3592fb219ccfa4"},
-    {file = "protobuf-4.25.3-cp39-cp39-win32.whl", hash = "sha256:19b270aeaa0099f16d3ca02628546b8baefe2955bbe23224aaf856134eccf1e4"},
-    {file = "protobuf-4.25.3-cp39-cp39-win_amd64.whl", hash = "sha256:e3c97a1555fd6388f857770ff8b9703083de6bf1f9274a002a332d65fbb56c8c"},
-    {file = "protobuf-4.25.3-py3-none-any.whl", hash = "sha256:f0700d54bcf45424477e46a9f0944155b46fb0639d69728739c0e47bab83f2b9"},
-    {file = "protobuf-4.25.3.tar.gz", hash = "sha256:25b5d0b42fd000320bd7830b349e3b696435f3b329810427a6bcce6a5492cc5c"},
+    {file = "protobuf-4.25.4-cp310-abi3-win32.whl", hash = "sha256:db9fd45183e1a67722cafa5c1da3e85c6492a5383f127c86c4c4aa4845867dc4"},
+    {file = "protobuf-4.25.4-cp310-abi3-win_amd64.whl", hash = "sha256:ba3d8504116a921af46499471c63a85260c1a5fc23333154a427a310e015d26d"},
+    {file = "protobuf-4.25.4-cp37-abi3-macosx_10_9_universal2.whl", hash = "sha256:eecd41bfc0e4b1bd3fa7909ed93dd14dd5567b98c941d6c1ad08fdcab3d6884b"},
+    {file = "protobuf-4.25.4-cp37-abi3-manylinux2014_aarch64.whl", hash = "sha256:4c8a70fdcb995dcf6c8966cfa3a29101916f7225e9afe3ced4395359955d3835"},
+    {file = "protobuf-4.25.4-cp37-abi3-manylinux2014_x86_64.whl", hash = "sha256:3319e073562e2515c6ddc643eb92ce20809f5d8f10fead3332f71c63be6a7040"},
+    {file = "protobuf-4.25.4-cp38-cp38-win32.whl", hash = "sha256:7e372cbbda66a63ebca18f8ffaa6948455dfecc4e9c1029312f6c2edcd86c4e1"},
+    {file = "protobuf-4.25.4-cp38-cp38-win_amd64.whl", hash = "sha256:051e97ce9fa6067a4546e75cb14f90cf0232dcb3e3d508c448b8d0e4265b61c1"},
+    {file = "protobuf-4.25.4-cp39-cp39-win32.whl", hash = "sha256:90bf6fd378494eb698805bbbe7afe6c5d12c8e17fca817a646cd6a1818c696ca"},
+    {file = "protobuf-4.25.4-cp39-cp39-win_amd64.whl", hash = "sha256:ac79a48d6b99dfed2729ccccee547b34a1d3d63289c71cef056653a846a2240f"},
+    {file = "protobuf-4.25.4-py3-none-any.whl", hash = "sha256:bfbebc1c8e4793cfd58589acfb8a1026be0003e852b9da7db5a4285bde996978"},
+    {file = "protobuf-4.25.4.tar.gz", hash = "sha256:0dc4a62cc4052a036ee2204d26fe4d835c62827c855c8a03f29fe6da146b380d"},
 ]
 
 [[package]]
@@ -4317,13 +4282,13 @@ torch = ["torch"]
 
 [[package]]
 name = "pymdown-extensions"
-version = "10.8.1"
+version = "10.9"
 description = "Extension pack for Python Markdown."
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "pymdown_extensions-10.8.1-py3-none-any.whl", hash = "sha256:f938326115884f48c6059c67377c46cf631c733ef3629b6eed1349989d1b30cb"},
-    {file = "pymdown_extensions-10.8.1.tar.gz", hash = "sha256:3ab1db5c9e21728dabf75192d71471f8e50f216627e9a1fa9535ecb0231b9940"},
+    {file = "pymdown_extensions-10.9-py3-none-any.whl", hash = "sha256:d323f7e90d83c86113ee78f3fe62fc9dee5f56b54d912660703ea1816fed5626"},
+    {file = "pymdown_extensions-10.9.tar.gz", hash = "sha256:6ff740bcd99ec4172a938970d42b96128bdc9d4b9bcad72494f29921dc69b753"},
 ]
 
 [package.dependencies]
@@ -4388,13 +4353,13 @@ files = [
 
 [[package]]
 name = "pyright"
-version = "1.1.372"
+version = "1.1.373"
 description = "Command line wrapper for pyright"
 optional = false
 python-versions = ">=3.7"
 files = [
-    {file = "pyright-1.1.372-py3-none-any.whl", hash = "sha256:25b15fb8967740f0949fd35b963777187f0a0404c0bd753cc966ec139f3eaa0b"},
-    {file = "pyright-1.1.372.tar.gz", hash = "sha256:a9f5e0daa955daaa17e3d1ef76d3623e75f8afd5e37b437d3ff84d5b38c15420"},
+    {file = "pyright-1.1.373-py3-none-any.whl", hash = "sha256:b805413227f2c209f27b14b55da27fe5e9fb84129c9f1eb27708a5d12f6f000e"},
+    {file = "pyright-1.1.373.tar.gz", hash = "sha256:f41bcfc8b9d1802b09921a394d6ae1ce19694957b628bc657629688daf8a83ff"},
 ]
 
 [package.dependencies]
@@ -4428,13 +4393,13 @@ files = [
 
 [[package]]
 name = "pytest"
-version = "8.3.1"
+version = "8.3.2"
 description = "pytest: simple powerful testing with Python"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "pytest-8.3.1-py3-none-any.whl", hash = "sha256:e9600ccf4f563976e2c99fa02c7624ab938296551f280835ee6516df8bc4ae8c"},
-    {file = "pytest-8.3.1.tar.gz", hash = "sha256:7e8e5c5abd6e93cb1cc151f23e57adc31fcf8cfd2a3ff2da63e23f732de35db6"},
+    {file = "pytest-8.3.2-py3-none-any.whl", hash = "sha256:4ba08f9ae7dcf84ded419494d229b48d0903ea6407b030eaec46df5e6a73bba5"},
+    {file = "pytest-8.3.2.tar.gz", hash = "sha256:c132345d12ce551242c87269de812483f5bcc87cdbb4722e48487ba194f9fdce"},
 ]
 
 [package.dependencies]
@@ -4448,6 +4413,24 @@ tomli = {version = ">=1", markers = "python_version < \"3.11\""}
 [package.extras]
 dev = ["argcomplete", "attrs (>=19.2)", "hypothesis (>=3.56)", "mock", "pygments (>=2.7.2)", "requests", "setuptools", "xmlschema"]
 
+[[package]]
+name = "pytest-asyncio"
+version = "0.23.8"
+description = "Pytest support for asyncio"
+optional = false
+python-versions = ">=3.8"
+files = [
+    {file = "pytest_asyncio-0.23.8-py3-none-any.whl", hash = "sha256:50265d892689a5faefb84df80819d1ecef566eb3549cf915dfb33569359d1ce2"},
+    {file = "pytest_asyncio-0.23.8.tar.gz", hash = "sha256:759b10b33a6dc61cce40a8bd5205e302978bbbcc00e279a8b61d9a6a3c82e4d3"},
+]
+
+[package.dependencies]
+pytest = ">=7.0.0,<9"
+
+[package.extras]
+docs = ["sphinx (>=5.3)", "sphinx-rtd-theme (>=1.0)"]
+testing = ["coverage (>=6.2)", "hypothesis (>=5.7.1)"]
+
 [[package]]
 name = "pytest-vcr"
 version = "1.0.2"
@@ -4883,22 +4866,22 @@ files = [
 
 [[package]]
 name = "selenium"
-version = "4.22.0"
+version = "4.23.1"
 description = "Official Python bindings for Selenium WebDriver"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "selenium-4.22.0-py3-none-any.whl", hash = "sha256:e424991196e9857e19bf04fe5c1c0a4aac076794ff5e74615b1124e729d93104"},
-    {file = "selenium-4.22.0.tar.gz", hash = "sha256:903c8c9d61b3eea6fcc9809dc7d9377e04e2ac87709876542cc8f863e482c4ce"},
+    {file = "selenium-4.23.1-py3-none-any.whl", hash = "sha256:3a8d9f23dc636bd3840dd56f00c2739e32ec0c1e34a821dd553e15babef24477"},
+    {file = "selenium-4.23.1.tar.gz", hash = "sha256:128d099e66284437e7128d2279176ec7a06e6ec7426e167f5d34987166bd8f46"},
 ]
 
 [package.dependencies]
 certifi = ">=2021.10.8"
 trio = ">=0.17,<1.0"
 trio-websocket = ">=0.9,<1.0"
-typing_extensions = ">=4.9.0"
+typing_extensions = ">=4.9,<5.0"
 urllib3 = {version = ">=1.26,<3", extras = ["socks"]}
-websocket-client = ">=1.8.0"
+websocket-client = ">=1.8,<2.0"
 
 [[package]]
 name = "semver"
@@ -4913,13 +4896,13 @@ files = [
 
 [[package]]
 name = "setuptools"
-version = "71.1.0"
+version = "72.1.0"
 description = "Easily download, build, install, upgrade, and uninstall Python packages"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "setuptools-71.1.0-py3-none-any.whl", hash = "sha256:33874fdc59b3188304b2e7c80d9029097ea31627180896fb549c578ceb8a0855"},
-    {file = "setuptools-71.1.0.tar.gz", hash = "sha256:032d42ee9fb536e33087fb66cac5f840eb9391ed05637b3f2a76a7c8fb477936"},
+    {file = "setuptools-72.1.0-py3-none-any.whl", hash = "sha256:5a03e1860cf56bb6ef48ce186b0e557fdba433237481a9a625176c2831be15d1"},
+    {file = "setuptools-72.1.0.tar.gz", hash = "sha256:8d243eff56d095e5817f796ede6ae32941278f542e0f941867cc05ae52b162ec"},
 ]
 
 [package.extras]
@@ -5268,34 +5251,6 @@ webencodings = ">=0.4"
 doc = ["sphinx", "sphinx_rtd_theme"]
 test = ["pytest", "ruff"]
 
-[[package]]
-name = "together"
-version = "1.2.3"
-description = "Python client for Together's Cloud Platform!"
-optional = false
-python-versions = "<4.0,>=3.8"
-files = [
-    {file = "together-1.2.3-py3-none-any.whl", hash = "sha256:bbafb4b8340e0f7e0ddb11ad447eb3467c591090910d0291cfbf74b47af045c1"},
-    {file = "together-1.2.3.tar.gz", hash = "sha256:4ea7626a9581d16fbf293e3eaf91557c43dea044627cf6dbe458bbf43408a6b2"},
-]
-
-[package.dependencies]
-aiohttp = ">=3.9.3,<4.0.0"
-click = ">=8.1.7,<9.0.0"
-eval-type-backport = ">=0.1.3,<0.3.0"
-filelock = ">=3.13.1,<4.0.0"
-numpy = [
-    {version = ">=1.23.5", markers = "python_version < \"3.12\""},
-    {version = ">=1.26.0", markers = "python_version >= \"3.12\""},
-]
-pillow = ">=10.3.0,<11.0.0"
-pyarrow = ">=10.0.1"
-pydantic = ">=2.6.3,<3.0.0"
-requests = ">=2.31.0,<3.0.0"
-tabulate = ">=0.9.0,<0.10.0"
-tqdm = ">=4.66.2,<5.0.0"
-typer = ">=0.9,<0.13"
-
 [[package]]
 name = "tokenizers"
 version = "0.19.1"
@@ -6144,4 +6099,4 @@ tools = ["crewai-tools"]
 [metadata]
 lock-version = "2.0"
 python-versions = ">=3.10,<=3.13"
-content-hash = "f5ad9babb3c57c405e39232020e8cbfaaeb5c315c2e7c5bb8fdf66792f260343"
+content-hash = "8df022f5ec0997c0a0f5710476139d9117c1057889c158e958f2c8efd22a4756"

pyproject.toml

@@ -52,6 +52,7 @@ crewai-tools = "^0.4.26"
 pytest = "^8.0.0"
 pytest-vcr = "^1.0.2"
 python-dotenv = "1.0.0"
+pytest-asyncio = "^0.23.7"
 
 [tool.poetry.scripts]
 crewai = "crewai.cli.cli:crewai"

src/crewai/agents/agent_builder/utilities/base_token_process.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict
+from crewai.types.usage_metrics import UsageMetrics
 
 
 class TokenProcess:
@@ -18,10 +18,10 @@ class TokenProcess:
     def sum_successful_requests(self, requests: int):
         self.successful_requests = self.successful_requests + requests
 
-    def get_summary(self) -> Dict[str, Any]:
-        return {
-            "total_tokens": self.total_tokens,
-            "prompt_tokens": self.prompt_tokens,
-            "completion_tokens": self.completion_tokens,
-            "successful_requests": self.successful_requests,
-        }
+    def get_summary(self) -> UsageMetrics:
+        return UsageMetrics(
+            total_tokens=self.total_tokens,
+            prompt_tokens=self.prompt_tokens,
+            completion_tokens=self.completion_tokens,
+            successful_requests=self.successful_requests,
+        )

src/crewai/agents/executor.py

@@ -56,7 +56,7 @@ class CrewAgentExecutor(AgentExecutor, CrewAgentExecutorMixin):
         )
         intermediate_steps: List[Tuple[AgentAction, str]] = []
         # Allowing human input given task setting
-        if self.task.human_input:
+        if self.task and self.task.human_input:
             self.should_ask_for_human_input = True
 
         # Let's start tracking the number of iterations and time elapsed

src/crewai/crew.py

@@ -3,7 +3,7 @@ import json
 import uuid
 from concurrent.futures import Future
 from hashlib import md5
-from typing import Any, Dict, List, Optional, Tuple, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union
 
 from langchain_core.callbacks import BaseCallbackHandler
 from pydantic import (
@@ -32,11 +32,9 @@ from crewai.tasks.conditional_task import ConditionalTask
 from crewai.tasks.task_output import TaskOutput
 from crewai.telemetry import Telemetry
 from crewai.tools.agent_tools import AgentTools
+from crewai.types.usage_metrics import UsageMetrics
 from crewai.utilities import I18N, FileHandler, Logger, RPMController
-from crewai.utilities.constants import (
-    TRAINED_AGENTS_DATA_FILE,
-    TRAINING_DATA_FILE,
-)
+from crewai.utilities.constants import TRAINED_AGENTS_DATA_FILE, TRAINING_DATA_FILE
 from crewai.utilities.evaluators.crew_evaluator_handler import CrewEvaluator
 from crewai.utilities.evaluators.task_evaluator import TaskEvaluator
 from crewai.utilities.formatter import (
@@ -52,6 +50,9 @@ try:
 except ImportError:
     agentops = None
 
+if TYPE_CHECKING:
+    from crewai.pipeline.pipeline import Pipeline
+
 
 class Crew(BaseModel):
     """
@@ -97,12 +98,13 @@ class Crew(BaseModel):
         default_factory=TaskOutputStorageHandler
     )
 
+    name: Optional[str] = Field(default=None)
     cache: bool = Field(default=True)
     model_config = ConfigDict(arbitrary_types_allowed=True)
     tasks: List[Task] = Field(default_factory=list)
     agents: List[BaseAgent] = Field(default_factory=list)
     process: Process = Field(default=Process.sequential)
-    verbose: Union[int, bool] = Field(default=0)
+    verbose: int = Field(default=0)
     memory: bool = Field(
         default=False,
         description="Whether the crew should use memory to store memories of it's execution",
@@ -111,7 +113,7 @@ class Crew(BaseModel):
         default={"provider": "openai"},
         description="Configuration for the embedder to be used for the crew.",
     )
-    usage_metrics: Optional[dict] = Field(
+    usage_metrics: Optional[UsageMetrics] = Field(
         default=None,
         description="Metrics for the LLM usage during all tasks execution.",
     )
@@ -147,8 +149,8 @@ class Crew(BaseModel):
         default=None,
         description="Path to the prompt json file to be used for the crew.",
     )
-    output_log_file: Optional[Union[bool, str]] = Field(
-        default=False,
+    output_log_file: Optional[str] = Field(
+        default=None,
         description="output_log_file",
     )
     planning: Optional[bool] = Field(
@@ -453,7 +455,7 @@ class Crew(BaseModel):
         if self.planning:
             self._handle_crew_planning()
 
-        metrics = []
+        metrics: List[UsageMetrics] = []
 
         if self.process == Process.sequential:
             result = self._run_sequential_process()
@@ -463,11 +465,12 @@ class Crew(BaseModel):
             raise NotImplementedError(
                 f"The process '{self.process}' is not implemented yet."
             )
+
         metrics += [agent._token_process.get_summary() for agent in self.agents]
 
-        self.usage_metrics = {
-            key: sum([m[key] for m in metrics if m is not None]) for key in metrics[0]
-        }
+        self.usage_metrics = UsageMetrics()
+        for metric in metrics:
+            self.usage_metrics.add_usage_metrics(metric)
 
         return result
 
@@ -476,12 +479,7 @@ class Crew(BaseModel):
         results: List[CrewOutput] = []
 
         # Initialize the parent crew's usage metrics
-        total_usage_metrics = {
-            "total_tokens": 0,
-            "prompt_tokens": 0,
-            "completion_tokens": 0,
-            "successful_requests": 0,
-        }
+        total_usage_metrics = UsageMetrics()
 
         for input_data in inputs:
             crew = self.copy()
@@ -489,8 +487,7 @@ class Crew(BaseModel):
             output = crew.kickoff(inputs=input_data)
 
             if crew.usage_metrics:
-                for key in total_usage_metrics:
-                    total_usage_metrics[key] += crew.usage_metrics.get(key, 0)
+                total_usage_metrics.add_usage_metrics(crew.usage_metrics)
 
             results.append(output)
 
@@ -519,29 +516,10 @@ class Crew(BaseModel):
 
         results = await asyncio.gather(*tasks)
 
-        total_usage_metrics = {
-            "total_tokens": 0,
-            "prompt_tokens": 0,
-            "completion_tokens": 0,
-            "successful_requests": 0,
-        }
+        total_usage_metrics = UsageMetrics()
         for crew in crew_copies:
             if crew.usage_metrics:
-                for key in total_usage_metrics:
-                    total_usage_metrics[key] += crew.usage_metrics.get(key, 0)
-
-        self.usage_metrics = total_usage_metrics
-
-        total_usage_metrics = {
-            "total_tokens": 0,
-            "prompt_tokens": 0,
-            "completion_tokens": 0,
-            "successful_requests": 0,
-        }
-        for crew in crew_copies:
-            if crew.usage_metrics:
-                for key in total_usage_metrics:
-                    total_usage_metrics[key] += crew.usage_metrics.get(key, 0)
+                total_usage_metrics.add_usage_metrics(crew.usage_metrics)
 
         self.usage_metrics = total_usage_metrics
         self._task_output_handler.reset()
@@ -932,25 +910,18 @@ class Crew(BaseModel):
             )
         self._telemetry.end_crew(self, final_string_output)
 
-    def calculate_usage_metrics(self) -> Dict[str, int]:
+    def calculate_usage_metrics(self) -> UsageMetrics:
         """Calculates and returns the usage metrics."""
-        total_usage_metrics = {
-            "total_tokens": 0,
-            "prompt_tokens": 0,
-            "completion_tokens": 0,
-            "successful_requests": 0,
-        }
+        total_usage_metrics = UsageMetrics()
 
         for agent in self.agents:
             if hasattr(agent, "_token_process"):
                 token_sum = agent._token_process.get_summary()
-                for key in total_usage_metrics:
-                    total_usage_metrics[key] += token_sum.get(key, 0)
+                total_usage_metrics.add_usage_metrics(token_sum)
 
         if self.manager_agent and hasattr(self.manager_agent, "_token_process"):
             token_sum = self.manager_agent._token_process.get_summary()
-            for key in total_usage_metrics:
-                total_usage_metrics[key] += token_sum.get(key, 0)
+            total_usage_metrics.add_usage_metrics(token_sum)
 
         return total_usage_metrics
 
@@ -969,5 +940,17 @@ class Crew(BaseModel):
 
         evaluator.print_crew_evaluation_result()
 
+    def __rshift__(self, other: "Crew") -> "Pipeline":
+        """
+        Implements the >> operator to add another Crew to an existing Pipeline.
+        """
+        from crewai.pipeline.pipeline import Pipeline
+
+        if not isinstance(other, Crew):
+            raise TypeError(
+                f"Unsupported operand type for >>: '{type(self).__name__}' and '{type(other).__name__}'"
+            )
+        return Pipeline(stages=[self, other])
+
     def __repr__(self):
         return f"Crew(id={self.id}, process={self.process}, number_of_agents={len(self.agents)}, number_of_tasks={len(self.tasks)})"

src/crewai/crews/crew_output.py

@@ -5,6 +5,7 @@ from pydantic import BaseModel, Field
 
 from crewai.tasks.output_format import OutputFormat
 from crewai.tasks.task_output import TaskOutput
+from crewai.types.usage_metrics import UsageMetrics
 
 
 class CrewOutput(BaseModel):
@@ -20,9 +21,7 @@ class CrewOutput(BaseModel):
     tasks_output: list[TaskOutput] = Field(
         description="Output of each task", default=[]
     )
-    token_usage: Dict[str, Any] = Field(
-        description="Processed token summary", default={}
-    )
+    token_usage: UsageMetrics = Field(description="Processed token summary", default={})
 
     @property
     def json(self) -> Optional[str]:

src/crewai/pipeline/__init__.py

@@ -0,0 +1,3 @@
+from crewai.pipeline.pipeline import Pipeline
+
+__all__ = ["Pipeline"]

src/crewai/pipeline/pipeline.py

@@ -0,0 +1,371 @@
+import asyncio
+import copy
+from typing import Any, Dict, List, Tuple, Union
+
+from pydantic import BaseModel, Field, model_validator
+
+from crewai.crew import Crew
+from crewai.crews.crew_output import CrewOutput
+from crewai.pipeline.pipeline_kickoff_result import PipelineKickoffResult
+from crewai.types.usage_metrics import UsageMetrics
+
+Trace = Union[Union[str, Dict[str, Any]], List[Union[str, Dict[str, Any]]]]
+
+
+"""
+Developer Notes:
+
+This module defines a Pipeline class that represents a sequence of operations (stages)
+to process inputs. Each stage can be either sequential or parallel, and the pipeline
+can process multiple kickoffs concurrently.
+
+Core Loop Explanation:
+1. The `process_kickoffs` method processes multiple kickoffs in parallel, each going through
+   all pipeline stages.
+2. The `process_single_kickoff` method handles the processing of a single kickouff through
+   all stages, updating metrics and input data along the way.
+3. The `_process_stage` method determines whether a stage is sequential or parallel
+   and processes it accordingly.
+4. The `_process_single_crew` and `_process_parallel_crews` methods handle the
+   execution of single and parallel crew stages.
+5. The `_update_metrics_and_input` method updates usage metrics and the current input
+   with the outputs from a stage.
+6. The `_build_pipeline_kickoff_results` method constructs the final results of the
+   pipeline kickoff, including traces and outputs.
+
+Handling Traces and Crew Outputs:
+- During the processing of stages, we handle the results (traces and crew outputs)
+  for all stages except the last one differently from the final stage.
+- For intermediate stages, the primary focus is on passing the input data between stages.
+  This involves merging the output dictionaries from all crews in a stage into a single
+  dictionary and passing it to the next stage. This merged dictionary allows for smooth
+  data flow between stages.
+- For the final stage, in addition to passing the input data, we also need to prepare
+  the final outputs and traces to be returned as the overall result of the pipeline kickoff.
+  In this case, we do not merge the results, as each result needs to be included
+  separately in its own pipeline kickoff result.
+
+Pipeline Terminology:
+- Pipeline: The overall structure that defines a sequence of operations.
+- Stage: A distinct part of the pipeline, which can be either sequential or parallel.
+- Kickoff: A specific execution of the pipeline for a given set of inputs, representing a single instance of processing through the pipeline.
+- Branch: Parallel executions within a stage (e.g., concurrent crew operations).
+- Trace: The journey of an individual input through the entire pipeline.
+
+Example pipeline structure:
+crew1 >> crew2 >> crew3 
+
+This represents a pipeline with three sequential stages:
+1. crew1 is the first stage, which processes the input and passes its output to crew2.
+2. crew2 is the second stage, which takes the output from crew1 as its input, processes it, and passes its output to crew3.
+3. crew3 is the final stage, which takes the output from crew2 as its input and produces the final output of the pipeline.
+
+Each input creates its own kickoff, flowing through all stages of the pipeline.
+Multiple kickoffss can be processed concurrently, each following the defined pipeline structure.
+
+Another example pipeline structure:
+crew1 >> [crew2, crew3] >> crew4
+
+This represents a pipeline with three stages:
+1. A sequential stage (crew1)
+2. A parallel stage with two branches (crew2 and crew3 executing concurrently)
+3. Another sequential stage (crew4)
+
+Each input creates its own kickoff, flowing through all stages of the pipeline.
+Multiple kickoffs can be processed concurrently, each following the defined pipeline structure.
+"""
+
+
+class Pipeline(BaseModel):
+    stages: List[Union[Crew, List[Crew]]] = Field(
+        ..., description="List of crews representing stages to be executed in sequence"
+    )
+
+    @model_validator(mode="before")
+    @classmethod
+    def validate_stages(cls, values):
+        """
+        Validates the stages to ensure correct nesting and types.
+
+        Args:
+            values (dict): Dictionary containing the pipeline stages.
+
+        Returns:
+            dict: Validated stages.
+        """
+        stages = values.get("stages", [])
+
+        def check_nesting_and_type(item, depth=0):
+            if depth > 1:
+                raise ValueError("Double nesting is not allowed in pipeline stages")
+            if isinstance(item, list):
+                for sub_item in item:
+                    check_nesting_and_type(sub_item, depth + 1)
+            elif not isinstance(item, Crew):
+                raise ValueError(
+                    f"Expected Crew instance or list of Crews, got {type(item)}"
+                )
+
+        for stage in stages:
+            check_nesting_and_type(stage)
+        return values
+
+    async def kickoff(
+        self, inputs: List[Dict[str, Any]]
+    ) -> List[PipelineKickoffResult]:
+        """
+        Processes multiple runs in parallel, each going through all pipeline stages.
+
+        Args:
+            inputs (List[Dict[str, Any]]): List of inputs for each run.
+
+        Returns:
+            List[PipelineKickoffResult]: List of results from each run.
+        """
+        pipeline_results: List[PipelineKickoffResult] = []
+
+        # Process all runs in parallel
+        all_run_results = await asyncio.gather(
+            *(self.process_single_kickoff(input_data) for input_data in inputs)
+        )
+
+        # Flatten the list of lists into a single list of results
+        pipeline_results.extend(
+            result for run_result in all_run_results for result in run_result
+        )
+
+        return pipeline_results
+
+    async def process_single_kickoff(
+        self, kickoff_input: Dict[str, Any]
+    ) -> List[PipelineKickoffResult]:
+        """
+        Processes a single run through all pipeline stages.
+
+        Args:
+            input (Dict[str, Any]): The input for the run.
+
+        Returns:
+            List[PipelineKickoffResult]: The results of processing the run.
+        """
+        initial_input = copy.deepcopy(kickoff_input)
+        current_input = copy.deepcopy(kickoff_input)
+        pipeline_usage_metrics: Dict[str, UsageMetrics] = {}
+        all_stage_outputs: List[List[CrewOutput]] = []
+        traces: List[List[Union[str, Dict[str, Any]]]] = [[initial_input]]
+
+        for stage in self.stages:
+            stage_input = copy.deepcopy(current_input)
+            stage_outputs, stage_trace = await self._process_stage(stage, stage_input)
+
+            self._update_metrics_and_input(
+                pipeline_usage_metrics, current_input, stage, stage_outputs
+            )
+            traces.append(stage_trace)
+            all_stage_outputs.append(stage_outputs)
+
+        return self._build_pipeline_kickoff_results(
+            all_stage_outputs, traces, pipeline_usage_metrics
+        )
+
+    async def _process_stage(
+        self, stage: Union[Crew, List[Crew]], current_input: Dict[str, Any]
+    ) -> Tuple[List[CrewOutput], List[Union[str, Dict[str, Any]]]]:
+        """
+        Processes a single stage of the pipeline, which can be either sequential or parallel.
+
+        Args:
+            stage (Union[Crew, List[Crew]]): The stage to process.
+            current_input (Dict[str, Any]): The input for the stage.
+
+        Returns:
+            Tuple[List[CrewOutput], List[Union[str, Dict[str, Any]]]]: The outputs and trace of the stage.
+        """
+        if isinstance(stage, Crew):
+            return await self._process_single_crew(stage, current_input)
+        else:
+            return await self._process_parallel_crews(stage, current_input)
+
+    async def _process_single_crew(
+        self, crew: Crew, current_input: Dict[str, Any]
+    ) -> Tuple[List[CrewOutput], List[Union[str, Dict[str, Any]]]]:
+        """
+        Processes a single crew.
+
+        Args:
+            crew (Crew): The crew to process.
+            current_input (Dict[str, Any]): The input for the crew.
+
+        Returns:
+            Tuple[List[CrewOutput], List[Union[str, Dict[str, Any]]]]: The output and trace of the crew.
+        """
+        output = await crew.kickoff_async(inputs=current_input)
+        return [output], [crew.name or str(crew.id)]
+
+    async def _process_parallel_crews(
+        self, crews: List[Crew], current_input: Dict[str, Any]
+    ) -> Tuple[List[CrewOutput], List[Union[str, Dict[str, Any]]]]:
+        """
+        Processes multiple crews in parallel.
+
+        Args:
+            crews (List[Crew]): The list of crews to process in parallel.
+            current_input (Dict[str, Any]): The input for the crews.
+
+        Returns:
+            Tuple[List[CrewOutput], List[Union[str, Dict[str, Any]]]]: The outputs and traces of the crews.
+        """
+        parallel_outputs = await asyncio.gather(
+            *[crew.kickoff_async(inputs=current_input) for crew in crews]
+        )
+        return parallel_outputs, [crew.name or str(crew.id) for crew in crews]
+
+    def _update_metrics_and_input(
+        self,
+        usage_metrics: Dict[str, UsageMetrics],
+        current_input: Dict[str, Any],
+        stage: Union[Crew, List[Crew]],
+        outputs: List[CrewOutput],
+    ) -> None:
+        """
+        Updates metrics and current input with the outputs of a stage.
+
+        Args:
+            usage_metrics (Dict[str, Any]): The usage metrics to update.
+            current_input (Dict[str, Any]): The current input to update.
+            stage (Union[Crew, List[Crew]]): The stage that was processed.
+            outputs (List[CrewOutput]): The outputs of the stage.
+        """
+        for crew, output in zip([stage] if isinstance(stage, Crew) else stage, outputs):
+            usage_metrics[crew.name or str(crew.id)] = output.token_usage
+            current_input.update(output.to_dict())
+
+    def _build_pipeline_kickoff_results(
+        self,
+        all_stage_outputs: List[List[CrewOutput]],
+        traces: List[List[Union[str, Dict[str, Any]]]],
+        token_usage: Dict[str, UsageMetrics],
+    ) -> List[PipelineKickoffResult]:
+        """
+        Builds the results of a pipeline run.
+
+        Args:
+            all_stage_outputs (List[List[CrewOutput]]): All stage outputs.
+            traces (List[List[Union[str, Dict[str, Any]]]]): All traces.
+            token_usage (Dict[str, Any]): Token usage metrics.
+
+        Returns:
+            List[PipelineKickoffResult]: The results of the pipeline run.
+        """
+        formatted_traces = self._format_traces(traces)
+        formatted_crew_outputs = self._format_crew_outputs(all_stage_outputs)
+
+        return [
+            PipelineKickoffResult(
+                token_usage=token_usage,
+                trace=formatted_trace,
+                raw=crews_outputs[-1].raw,
+                pydantic=crews_outputs[-1].pydantic,
+                json_dict=crews_outputs[-1].json_dict,
+                crews_outputs=crews_outputs,
+            )
+            for crews_outputs, formatted_trace in zip(
+                formatted_crew_outputs, formatted_traces
+            )
+        ]
+
+    def _format_traces(
+        self, traces: List[List[Union[str, Dict[str, Any]]]]
+    ) -> List[List[Trace]]:
+        """
+        Formats the traces of a pipeline run.
+
+        Args:
+            traces (List[List[Union[str, Dict[str, Any]]]]): The traces to format.
+
+        Returns:
+            List[List[Trace]]: The formatted traces.
+        """
+        formatted_traces: List[Trace] = self._format_single_trace(traces[:-1])
+        return self._format_multiple_traces(formatted_traces, traces[-1])
+
+    def _format_single_trace(
+        self, traces: List[List[Union[str, Dict[str, Any]]]]
+    ) -> List[Trace]:
+        """
+        Formats single traces.
+
+        Args:
+            traces (List[List[Union[str, Dict[str, Any]]]]): The traces to format.
+
+        Returns:
+            List[Trace]: The formatted single traces.
+        """
+        formatted_traces: List[Trace] = []
+        for trace in traces:
+            formatted_traces.append(trace[0] if len(trace) == 1 else trace)
+        return formatted_traces
+
+    def _format_multiple_traces(
+        self,
+        formatted_traces: List[Trace],
+        final_trace: List[Union[str, Dict[str, Any]]],
+    ) -> List[List[Trace]]:
+        """
+        Formats multiple traces.
+
+        Args:
+            formatted_traces (List[Trace]): The formatted single traces.
+            final_trace (List[Union[str, Dict[str, Any]]]): The final trace to format.
+
+        Returns:
+            List[List[Trace]]: The formatted multiple traces.
+        """
+        traces_to_return: List[List[Trace]] = []
+        if len(final_trace) == 1:
+            formatted_traces.append(final_trace[0])
+            traces_to_return.append(formatted_traces)
+        else:
+            for trace in final_trace:
+                copied_traces = formatted_traces.copy()
+                copied_traces.append(trace)
+                traces_to_return.append(copied_traces)
+        return traces_to_return
+
+    def _format_crew_outputs(
+        self, all_stage_outputs: List[List[CrewOutput]]
+    ) -> List[List[CrewOutput]]:
+        """
+        Formats the outputs of all stages into a list of crew outputs.
+
+        Args:
+            all_stage_outputs (List[List[CrewOutput]]): All stage outputs.
+
+        Returns:
+            List[List[CrewOutput]]: Formatted crew outputs.
+        """
+        crew_outputs: List[CrewOutput] = [
+            output
+            for stage_outputs in all_stage_outputs[:-1]
+            for output in stage_outputs
+        ]
+        return [crew_outputs + [output] for output in all_stage_outputs[-1]]
+
+    def __rshift__(self, other: Any) -> "Pipeline":
+        """
+        Implements the >> operator to add another Stage (Crew or List[Crew]) to an existing Pipeline.
+
+        Args:
+            other (Any): The stage to add.
+
+        Returns:
+            Pipeline: A new pipeline with the added stage.
+        """
+        if isinstance(other, Crew):
+            return type(self)(stages=self.stages + [other])
+        elif isinstance(other, list) and all(isinstance(crew, Crew) for crew in other):
+            return type(self)(stages=self.stages + [other])
+        else:
+            raise TypeError(
+                f"Unsupported operand type for >>: '{type(self).__name__}' and '{type(other).__name__}'"
+            )

src/crewai/pipeline/pipeline_kickoff_result.py

@@ -0,0 +1,61 @@
+import json
+import uuid
+from typing import Any, Dict, List, Optional, Union
+
+from pydantic import UUID4, BaseModel, Field
+
+from crewai.crews.crew_output import CrewOutput
+from crewai.types.usage_metrics import UsageMetrics
+
+
+class PipelineKickoffResult(BaseModel):
+    """Class that represents the result of a pipeline run."""
+
+    id: UUID4 = Field(
+        default_factory=uuid.uuid4,
+        frozen=True,
+        description="Unique identifier for the object, not set by user.",
+    )
+    raw: str = Field(description="Raw output of the pipeline run", default="")
+    pydantic: Any = Field(
+        description="Pydantic output of the pipeline run", default=None
+    )
+    json_dict: Union[Dict[str, Any], None] = Field(
+        description="JSON dict output of the pipeline run", default={}
+    )
+
+    token_usage: Dict[str, UsageMetrics] = Field(
+        description="Token usage for each crew in the run"
+    )
+    trace: List[Any] = Field(
+        description="Trace of the journey of inputs through the run"
+    )
+    crews_outputs: List[CrewOutput] = Field(
+        description="Output from each crew in the run",
+        default=[],
+    )
+
+    @property
+    def json(self) -> Optional[str]:
+        if self.crews_outputs[-1].tasks_output[-1].output_format != "json":
+            raise ValueError(
+                "No JSON output found in the final task of the final crew. Please make sure to set the output_json property in the final task in your crew."
+            )
+
+        return json.dumps(self.json_dict)
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert json_output and pydantic_output to a dictionary."""
+        output_dict = {}
+        if self.json_dict:
+            output_dict.update(self.json_dict)
+        elif self.pydantic:
+            output_dict.update(self.pydantic.model_dump())
+        return output_dict
+
+    def __str__(self):
+        if self.pydantic:
+            return str(self.pydantic)
+        if self.json_dict:
+            return str(self.json_dict)
+        return self.raw

src/crewai/pipeline/pipeline_output.py

@@ -0,0 +1,20 @@
+import uuid
+from typing import List
+
+from pydantic import UUID4, BaseModel, Field
+
+from crewai.pipeline.pipeline_kickoff_result import PipelineKickoffResult
+
+
+class PipelineOutput(BaseModel):
+    id: UUID4 = Field(
+        default_factory=uuid.uuid4,
+        frozen=True,
+        description="Unique identifier for the object, not set by user.",
+    )
+    run_results: List[PipelineKickoffResult] = Field(
+        description="List of results for each run through the pipeline", default=[]
+    )
+
+    def add_run_result(self, result: PipelineKickoffResult):
+        self.run_results.append(result)

src/crewai/task.py

@@ -47,6 +47,7 @@ class Task(BaseModel):
     tools_errors: int = 0
     delegations: int = 0
     i18n: I18N = I18N()
+    name: Optional[str] = Field(default=None)
     prompt_context: Optional[str] = None
     description: str = Field(description="Description of the actual task.")
     expected_output: str = Field(

src/crewai/types/usage_metrics.py

@@ -0,0 +1,36 @@
+from pydantic import BaseModel, Field
+
+
+class UsageMetrics(BaseModel):
+    """
+    Model to track usage metrics for the crew's execution.
+
+    Attributes:
+        total_tokens: Total number of tokens used.
+        prompt_tokens: Number of tokens used in prompts.
+        completion_tokens: Number of tokens used in completions.
+        successful_requests: Number of successful requests made.
+    """
+
+    total_tokens: int = Field(default=0, description="Total number of tokens used.")
+    prompt_tokens: int = Field(
+        default=0, description="Number of tokens used in prompts."
+    )
+    completion_tokens: int = Field(
+        default=0, description="Number of tokens used in completions."
+    )
+    successful_requests: int = Field(
+        default=0, description="Number of successful requests made."
+    )
+
+    def add_usage_metrics(self, usage_metrics: "UsageMetrics"):
+        """
+        Add the usage metrics from another UsageMetrics object.
+
+        Args:
+            usage_metrics (UsageMetrics): The usage metrics to add.
+        """
+        self.total_tokens += usage_metrics.total_tokens
+        self.prompt_tokens += usage_metrics.prompt_tokens
+        self.completion_tokens += usage_metrics.completion_tokens
+        self.successful_requests += usage_metrics.successful_requests

tests/crew_test.py

@@ -18,6 +18,7 @@ from crewai.task import Task
 from crewai.tasks.conditional_task import ConditionalTask
 from crewai.tasks.output_format import OutputFormat
 from crewai.tasks.task_output import TaskOutput
+from crewai.types.usage_metrics import UsageMetrics
 from crewai.utilities import Logger, RPMController
 from crewai.utilities.task_output_storage_handler import TaskOutputStorageHandler
 
@@ -597,14 +598,10 @@ def test_crew_kickoff_usage_metrics():
     assert len(results) == len(inputs)
     for result in results:
         # Assert that all required keys are in usage_metrics and their values are not None
-        for key in [
-            "total_tokens",
-            "prompt_tokens",
-            "completion_tokens",
-            "successful_requests",
-        ]:
-            assert key in result.token_usage
-            assert result.token_usage[key] > 0
+        assert result.token_usage.total_tokens > 0
+        assert result.token_usage.prompt_tokens > 0
+        assert result.token_usage.completion_tokens > 0
+        assert result.token_usage.successful_requests > 0
 
 
 def test_agents_rpm_is_never_set_if_crew_max_RPM_is_not_set():
@@ -1315,12 +1312,12 @@ def test_agent_usage_metrics_are_captured_for_hierarchical_process():
 
     print(crew.usage_metrics)
 
-    assert crew.usage_metrics == {
-        "total_tokens": 219,
-        "prompt_tokens": 201,
-        "completion_tokens": 18,
-        "successful_requests": 1,
-    }
+    assert crew.usage_metrics == UsageMetrics(
+        total_tokens=219,
+        prompt_tokens=201,
+        completion_tokens=18,
+        successful_requests=1,
+    )
 
 
 @pytest.mark.vcr(filter_headers=["authorization"])

tests/pipeline/test_pipeline.py

@@ -0,0 +1,468 @@
+import json
+from unittest.mock import MagicMock
+
+import pytest
+from crewai.agent import Agent
+from crewai.crew import Crew
+from crewai.crews.crew_output import CrewOutput
+from crewai.pipeline.pipeline import Pipeline
+from crewai.pipeline.pipeline_kickoff_result import PipelineKickoffResult
+from crewai.process import Process
+from crewai.task import Task
+from crewai.tasks.task_output import TaskOutput
+from crewai.types.usage_metrics import UsageMetrics
+from pydantic import BaseModel, ValidationError
+
+DEFAULT_TOKEN_USAGE = UsageMetrics(
+    total_tokens=100, prompt_tokens=50, completion_tokens=50, successful_requests=3
+)
+
+
+@pytest.fixture
+def mock_crew_factory():
+    def _create_mock_crew(name: str, output_json_dict=None, pydantic_output=None):
+        crew = MagicMock(spec=Crew)
+        task_output = TaskOutput(
+            description="Test task", raw="Task output", agent="Test Agent"
+        )
+        crew_output = CrewOutput(
+            raw="Test output",
+            tasks_output=[task_output],
+            token_usage=DEFAULT_TOKEN_USAGE,
+            json_dict=output_json_dict if output_json_dict else None,
+            pydantic=pydantic_output,
+        )
+
+        async def async_kickoff(inputs=None):
+            print("inputs in async_kickoff", inputs)
+            return crew_output
+
+        crew.kickoff_async.side_effect = async_kickoff
+
+        # Add more attributes that Procedure might be expecting
+        crew.verbose = False
+        crew.output_log_file = None
+        crew.max_rpm = None
+        crew.memory = False
+        crew.process = Process.sequential
+        crew.config = None
+        crew.cache = True
+        crew.name = name
+
+        # Add non-empty agents and tasks
+        mock_agent = MagicMock(spec=Agent)
+        mock_task = MagicMock(spec=Task)
+        mock_task.agent = mock_agent
+        mock_task.async_execution = False
+        mock_task.context = None
+
+        crew.agents = [mock_agent]
+        crew.tasks = [mock_task]
+
+        return crew
+
+    return _create_mock_crew
+
+
+def test_pipeline_initialization(mock_crew_factory):
+    """
+    Test that a Pipeline is correctly initialized with the given stages.
+    """
+    crew1 = mock_crew_factory(name="Crew 1")
+    crew2 = mock_crew_factory(name="Crew 2")
+
+    pipeline = Pipeline(stages=[crew1, crew2])
+    assert len(pipeline.stages) == 2
+    assert pipeline.stages[0] == crew1
+    assert pipeline.stages[1] == crew2
+
+
+@pytest.mark.asyncio
+async def test_pipeline_with_empty_input(mock_crew_factory):
+    """
+    Ensure the pipeline handles an empty input list correctly.
+    """
+    crew = mock_crew_factory(name="Test Crew")
+    pipeline = Pipeline(stages=[crew])
+
+    input_data = []
+    pipeline_results = await pipeline.kickoff(input_data)
+
+    assert (
+        len(pipeline_results) == 0
+    ), "Pipeline should return empty results for empty input"
+
+
+@pytest.mark.asyncio
+async def test_pipeline_process_streams_single_input(mock_crew_factory):
+    """
+    Test that Pipeline.process_streams() correctly processes a single input
+    and returns the expected CrewOutput.
+    """
+    crew_name = "Test Crew"
+    mock_crew = mock_crew_factory(name="Test Crew")
+    pipeline = Pipeline(stages=[mock_crew])
+    input_data = [{"key": "value"}]
+    pipeline_results = await pipeline.kickoff(input_data)
+
+    mock_crew.kickoff_async.assert_called_once_with(inputs={"key": "value"})
+
+    for pipeline_result in pipeline_results:
+        assert isinstance(pipeline_result, PipelineKickoffResult)
+        assert pipeline_result.raw == "Test output"
+        assert len(pipeline_result.crews_outputs) == 1
+        print("pipeline_result.token_usage", pipeline_result.token_usage)
+        assert pipeline_result.token_usage == {crew_name: DEFAULT_TOKEN_USAGE}
+        assert pipeline_result.trace == [input_data[0], "Test Crew"]
+
+
+@pytest.mark.asyncio
+async def test_pipeline_result_ordering(mock_crew_factory):
+    """
+    Ensure that results are returned in the same order as the inputs, especially with parallel processing.
+    """
+    crew1 = mock_crew_factory(name="Crew 1", output_json_dict={"output": "crew1"})
+    crew2 = mock_crew_factory(name="Crew 2", output_json_dict={"output": "crew2"})
+    crew3 = mock_crew_factory(name="Crew 3", output_json_dict={"output": "crew3"})
+
+    pipeline = Pipeline(
+        stages=[crew1, [crew2, crew3]]
+    )  # Parallel stage to test ordering
+
+    input_data = [{"id": 1}, {"id": 2}, {"id": 3}]
+    pipeline_results = await pipeline.kickoff(input_data)
+
+    assert (
+        len(pipeline_results) == 6
+    ), "Should have 2 results for each input due to the parallel final stage"
+
+    # Group results by their original input id
+    grouped_results = {}
+    for result in pipeline_results:
+        input_id = result.trace[0]["id"]
+        if input_id not in grouped_results:
+            grouped_results[input_id] = []
+        grouped_results[input_id].append(result)
+
+    # Check that we have the correct number of groups and results per group
+    assert len(grouped_results) == 3, "Should have results for each of the 3 inputs"
+    for input_id, results in grouped_results.items():
+        assert (
+            len(results) == 2
+        ), f"Each input should have 2 results, but input {input_id} has {len(results)}"
+
+    # Check the ordering and content of the results
+    for input_id in range(1, 4):
+        group = grouped_results[input_id]
+        assert group[0].trace == [
+            {"id": input_id},
+            "Crew 1",
+            "Crew 2",
+        ], f"Unexpected trace for first result of input {input_id}"
+        assert group[1].trace == [
+            {"id": input_id},
+            "Crew 1",
+            "Crew 3",
+        ], f"Unexpected trace for second result of input {input_id}"
+        assert (
+            group[0].json_dict["output"] == "crew2"
+        ), f"Unexpected output for first result of input {input_id}"
+        assert (
+            group[1].json_dict["output"] == "crew3"
+        ), f"Unexpected output for second result of input {input_id}"
+
+
+class TestPydanticOutput(BaseModel):
+    key: str
+    value: int
+
+
+@pytest.mark.asyncio
+async def test_pipeline_process_streams_single_input_pydantic_output(mock_crew_factory):
+    crew_name = "Test Crew"
+    mock_crew = mock_crew_factory(
+        name=crew_name,
+        output_json_dict=None,
+        pydantic_output=TestPydanticOutput(key="test", value=42),
+    )
+    pipeline = Pipeline(stages=[mock_crew])
+    input_data = [{"key": "value"}]
+    pipeline_results = await pipeline.kickoff(input_data)
+
+    assert len(pipeline_results) == 1
+    pipeline_result = pipeline_results[0]
+
+    print("pipeline_result.trace", pipeline_result.trace)
+
+    assert isinstance(pipeline_result, PipelineKickoffResult)
+    assert pipeline_result.raw == "Test output"
+    assert len(pipeline_result.crews_outputs) == 1
+    assert pipeline_result.token_usage == {crew_name: DEFAULT_TOKEN_USAGE}
+    print("INPUT DATA POST PROCESS", input_data)
+    assert pipeline_result.trace == [input_data[0], "Test Crew"]
+
+    assert isinstance(pipeline_result.pydantic, TestPydanticOutput)
+    assert pipeline_result.pydantic.key == "test"
+    assert pipeline_result.pydantic.value == 42
+    assert pipeline_result.json_dict is None
+
+
+@pytest.mark.asyncio
+async def test_pipeline_preserves_original_input(mock_crew_factory):
+    crew_name = "Test Crew"
+    mock_crew = mock_crew_factory(
+        name=crew_name,
+        output_json_dict={"new_key": "new_value"},
+    )
+    pipeline = Pipeline(stages=[mock_crew])
+
+    # Create a deep copy of the input data to ensure we're not comparing references
+    original_input_data = [{"key": "value", "nested": {"a": 1}}]
+    input_data = json.loads(json.dumps(original_input_data))
+
+    await pipeline.kickoff(input_data)
+
+    # Assert that the original input hasn't been modified
+    assert (
+        input_data == original_input_data
+    ), "The original input data should not be modified"
+
+    # Ensure that even nested structures haven't been modified
+    assert (
+        input_data[0]["nested"] == original_input_data[0]["nested"]
+    ), "Nested structures should not be modified"
+
+    # Verify that adding new keys to the crew output doesn't affect the original input
+    assert (
+        "new_key" not in input_data[0]
+    ), "New keys from crew output should not be added to the original input"
+
+
+@pytest.mark.asyncio
+async def test_pipeline_process_streams_multiple_inputs(mock_crew_factory):
+    """
+    Test that Pipeline.process_streams() correctly processes multiple inputs
+    and returns the expected CrewOutputs.
+    """
+    mock_crew = mock_crew_factory(name="Test Crew")
+    pipeline = Pipeline(stages=[mock_crew])
+    input_data = [{"key1": "value1"}, {"key2": "value2"}]
+    pipeline_results = await pipeline.kickoff(input_data)
+
+    assert mock_crew.kickoff_async.call_count == 2
+    assert len(pipeline_results) == 2
+    for pipeline_result in pipeline_results:
+        print("pipeline_result,", pipeline_result)
+        assert all(
+            isinstance(crew_output, CrewOutput)
+            for crew_output in pipeline_result.crews_outputs
+        )
+
+
+@pytest.mark.asyncio
+async def test_pipeline_with_parallel_stages(mock_crew_factory):
+    """
+    Test that Pipeline correctly handles parallel stages.
+    """
+    crew1 = mock_crew_factory(name="Crew 1")
+    crew2 = mock_crew_factory(name="Crew 2")
+    crew3 = mock_crew_factory(name="Crew 3")
+
+    pipeline = Pipeline(stages=[crew1, [crew2, crew3]])
+    input_data = [{"initial": "data"}]
+
+    pipeline_result = await pipeline.kickoff(input_data)
+
+    crew1.kickoff_async.assert_called_once_with(inputs={"initial": "data"})
+
+    assert len(pipeline_result) == 2
+    pipeline_result_1, pipeline_result_2 = pipeline_result
+
+    pipeline_result_1.trace = [
+        "Crew 1",
+        "Crew 2",
+    ]
+    pipeline_result_2.trace = [
+        "Crew 1",
+        "Crew 3",
+    ]
+
+    expected_token_usage = {
+        "Crew 1": DEFAULT_TOKEN_USAGE,
+        "Crew 2": DEFAULT_TOKEN_USAGE,
+        "Crew 3": DEFAULT_TOKEN_USAGE,
+    }
+
+    assert pipeline_result_1.token_usage == expected_token_usage
+    assert pipeline_result_2.token_usage == expected_token_usage
+
+
+@pytest.mark.asyncio
+async def test_pipeline_with_parallel_stages_end_in_single_stage(mock_crew_factory):
+    """
+    Test that Pipeline correctly handles parallel stages.
+    """
+    crew1 = mock_crew_factory(name="Crew 1")
+    crew2 = mock_crew_factory(name="Crew 2")
+    crew3 = mock_crew_factory(name="Crew 3")
+    crew4 = mock_crew_factory(name="Crew 4")
+
+    pipeline = Pipeline(stages=[crew1, [crew2, crew3], crew4])
+    input_data = [{"initial": "data"}]
+
+    pipeline_result = await pipeline.kickoff(input_data)
+
+    crew1.kickoff_async.assert_called_once_with(inputs={"initial": "data"})
+
+    assert len(pipeline_result) == 1
+    pipeline_result_1 = pipeline_result[0]
+
+    pipeline_result_1.trace = [
+        input_data[0],
+        "Crew 1",
+        ["Crew 2", "Crew 3"],
+        "Crew 4",
+    ]
+
+    expected_token_usage = {
+        "Crew 1": DEFAULT_TOKEN_USAGE,
+        "Crew 2": DEFAULT_TOKEN_USAGE,
+        "Crew 3": DEFAULT_TOKEN_USAGE,
+        "Crew 4": DEFAULT_TOKEN_USAGE,
+    }
+
+    assert pipeline_result_1.token_usage == expected_token_usage
+
+
+def test_pipeline_rshift_operator(mock_crew_factory):
+    """
+    Test that the >> operator correctly creates a Pipeline from Crews and lists of Crews.
+    """
+    crew1 = mock_crew_factory(name="Crew 1")
+    crew2 = mock_crew_factory(name="Crew 2")
+    crew3 = mock_crew_factory(name="Crew 3")
+
+    # Test single crew addition
+    pipeline = Pipeline(stages=[]) >> crew1
+    assert len(pipeline.stages) == 1
+    assert pipeline.stages[0] == crew1
+
+    # Test adding a list of crews
+    pipeline = Pipeline(stages=[crew1])
+    pipeline = pipeline >> [crew2, crew3]
+    print("pipeline.stages:", pipeline.stages)
+    assert len(pipeline.stages) == 2
+    assert pipeline.stages[1] == [crew2, crew3]
+
+    # Test error case: trying to shift with non-Crew object
+    with pytest.raises(TypeError):
+        pipeline >> "not a crew"
+
+
+@pytest.mark.asyncio
+async def test_pipeline_parallel_crews_to_parallel_crews(mock_crew_factory):
+    """
+    Test that feeding parallel crews to parallel crews works correctly.
+    """
+    crew1 = mock_crew_factory(name="Crew 1", output_json_dict={"output1": "crew1"})
+    crew2 = mock_crew_factory(name="Crew 2", output_json_dict={"output2": "crew2"})
+    crew3 = mock_crew_factory(name="Crew 3", output_json_dict={"output3": "crew3"})
+    crew4 = mock_crew_factory(name="Crew 4", output_json_dict={"output4": "crew4"})
+
+    pipeline = Pipeline(stages=[[crew1, crew2], [crew3, crew4]])
+
+    input_data = [{"input": "test"}]
+    pipeline_results = await pipeline.kickoff(input_data)
+
+    assert len(pipeline_results) == 2, "Should have 2 results for final parallel stage"
+
+    pipeline_result_1, pipeline_result_2 = pipeline_results
+
+    # Check the outputs
+    assert pipeline_result_1.json_dict == {"output3": "crew3"}
+    assert pipeline_result_2.json_dict == {"output4": "crew4"}
+
+    # Check the traces
+    expected_traces = [
+        [{"input": "test"}, ["Crew 1", "Crew 2"], "Crew 3"],
+        [{"input": "test"}, ["Crew 1", "Crew 2"], "Crew 4"],
+    ]
+
+    for result, expected_trace in zip(pipeline_results, expected_traces):
+        assert result.trace == expected_trace, f"Unexpected trace: {result.trace}"
+
+
+def test_pipeline_double_nesting_not_allowed(mock_crew_factory):
+    """
+    Test that double nesting in pipeline stages is not allowed.
+    """
+    crew1 = mock_crew_factory(name="Crew 1")
+    crew2 = mock_crew_factory(name="Crew 2")
+    crew3 = mock_crew_factory(name="Crew 3")
+    crew4 = mock_crew_factory(name="Crew 4")
+
+    with pytest.raises(ValidationError) as exc_info:
+        Pipeline(stages=[crew1, [[crew2, crew3], crew4]])
+
+    error_msg = str(exc_info.value)
+    print(f"Full error message: {error_msg}")  # For debugging
+    assert (
+        "Double nesting is not allowed in pipeline stages" in error_msg
+    ), f"Unexpected error message: {error_msg}"
+
+
+def test_pipeline_invalid_crew(mock_crew_factory):
+    """
+    Test that non-Crew objects are not allowed in pipeline stages.
+    """
+    crew1 = mock_crew_factory(name="Crew 1")
+    not_a_crew = "This is not a crew"
+
+    with pytest.raises(ValidationError) as exc_info:
+        Pipeline(stages=[crew1, not_a_crew])
+
+    error_msg = str(exc_info.value)
+    print(f"Full error message: {error_msg}")  # For debugging
+    assert (
+        "Expected Crew instance or list of Crews, got <class 'str'>" in error_msg
+    ), f"Unexpected error message: {error_msg}"
+
+
+"""
+TODO: Figure out what is the proper output for a pipeline with multiple stages
+
+Options:
+- Should the final output only include the last stage's output?
+- Should the final output include the accumulation of previous stages' outputs?
+"""
+
+
+@pytest.mark.asyncio
+async def test_pipeline_data_accumulation(mock_crew_factory):
+    crew1 = mock_crew_factory(name="Crew 1", output_json_dict={"key1": "value1"})
+    crew2 = mock_crew_factory(name="Crew 2", output_json_dict={"key2": "value2"})
+
+    pipeline = Pipeline(stages=[crew1, crew2])
+    input_data = [{"initial": "data"}]
+    results = await pipeline.kickoff(input_data)
+
+    # Check that crew1 was called with only the initial input
+    crew1.kickoff_async.assert_called_once_with(inputs={"initial": "data"})
+
+    # Check that crew2 was called with the combined input from the initial data and crew1's output
+    crew2.kickoff_async.assert_called_once_with(
+        inputs={"initial": "data", "key1": "value1"}
+    )
+
+    # Check the final output
+    assert len(results) == 1
+    final_result = results[0]
+    assert final_result.json_dict == {"key2": "value2"}
+
+    # Check that the trace includes all stages
+    assert final_result.trace == [{"initial": "data"}, "Crew 1", "Crew 2"]
+
+    # Check that crews_outputs contain the correct information
+    assert len(final_result.crews_outputs) == 2
+    assert final_result.crews_outputs[0].json_dict == {"key1": "value1"}
+    assert final_result.crews_outputs[1].json_dict == {"key2": "value2"}