chore: add docs on native async

docs/en/concepts/crews.mdx

@@ -307,12 +307,27 @@ print(result)
 
 ### Different Ways to Kick Off a Crew
 
-Once your crew is assembled, initiate the workflow with the appropriate kickoff method. CrewAI provides several methods for better control over the kickoff process: `kickoff()`, `kickoff_for_each()`, `kickoff_async()`, and `kickoff_for_each_async()`.
+Once your crew is assembled, initiate the workflow with the appropriate kickoff method. CrewAI provides several methods for better control over the kickoff process.
+
+#### Synchronous Methods
 
 - `kickoff()`: Starts the execution process according to the defined process flow.
 - `kickoff_for_each()`: Executes tasks sequentially for each provided input event or item in the collection.
-- `kickoff_async()`: Initiates the workflow asynchronously.
-- `kickoff_for_each_async()`: Executes tasks concurrently for each provided input event or item, leveraging asynchronous processing.
+
+#### Asynchronous Methods
+
+CrewAI offers two approaches for async execution:
+
+| Method | Type | Description |
+|--------|------|-------------|
+| `akickoff()` | Native async | True async/await throughout the entire execution chain |
+| `akickoff_for_each()` | Native async | Native async execution for each input in a list |
+| `kickoff_async()` | Thread-based | Wraps synchronous execution in `asyncio.to_thread` |
+| `kickoff_for_each_async()` | Thread-based | Thread-based async for each input in a list |
+
+<Note>
+For high-concurrency workloads, `akickoff()` and `akickoff_for_each()` are recommended as they use native async for task execution, memory operations, and knowledge retrieval.
+</Note>
 
 ```python Code
 # Start the crew's task execution
@@ -325,19 +340,30 @@ results = my_crew.kickoff_for_each(inputs=inputs_array)
 for result in results:
     print(result)
 
-# Example of using kickoff_async
+# Example of using native async with akickoff
+inputs = {'topic': 'AI in healthcare'}
+async_result = await my_crew.akickoff(inputs=inputs)
+print(async_result)
+
+# Example of using native async with akickoff_for_each
+inputs_array = [{'topic': 'AI in healthcare'}, {'topic': 'AI in finance'}]
+async_results = await my_crew.akickoff_for_each(inputs=inputs_array)
+for async_result in async_results:
+    print(async_result)
+
+# Example of using thread-based kickoff_async
 inputs = {'topic': 'AI in healthcare'}
 async_result = await my_crew.kickoff_async(inputs=inputs)
 print(async_result)
 
-# Example of using kickoff_for_each_async
+# Example of using thread-based kickoff_for_each_async
 inputs_array = [{'topic': 'AI in healthcare'}, {'topic': 'AI in finance'}]
 async_results = await my_crew.kickoff_for_each_async(inputs=inputs_array)
 for async_result in async_results:
     print(async_result)
 ```
 
-These methods provide flexibility in how you manage and execute tasks within your crew, allowing for both synchronous and asynchronous workflows tailored to your needs.
+These methods provide flexibility in how you manage and execute tasks within your crew, allowing for both synchronous and asynchronous workflows tailored to your needs. For detailed async examples, see the [Kickoff Crew Asynchronously](/en/learn/kickoff-async) guide.
 
 ### Streaming Crew Execution
 

docs/en/learn/kickoff-async.mdx

@@ -7,17 +7,28 @@ mode: "wide"
 
 ## Introduction
 
-CrewAI provides the ability to kickoff a crew asynchronously, allowing you to start the crew execution in a non-blocking manner. 
+CrewAI provides the ability to kickoff a crew asynchronously, allowing you to start the crew execution in a non-blocking manner.
 This feature is particularly useful when you want to run multiple crews concurrently or when you need to perform other tasks while the crew is executing.
 
-## Asynchronous Crew Execution
+CrewAI offers two approaches for async execution:
 
-To kickoff a crew asynchronously, use the `kickoff_async()` method. This method initiates the crew execution in a separate thread, allowing the main thread to continue executing other tasks.
+| Method | Type | Description |
+|--------|------|-------------|
+| `akickoff()` | Native async | True async/await throughout the entire execution chain |
+| `kickoff_async()` | Thread-based | Wraps synchronous execution in `asyncio.to_thread` |
+
+<Note>
+For high-concurrency workloads, `akickoff()` is recommended as it uses native async for task execution, memory operations, and knowledge retrieval.
+</Note>
+
+## Native Async Execution with `akickoff()`
+
+The `akickoff()` method provides true native async execution, using async/await throughout the entire execution chain including task execution, memory operations, and knowledge queries.
 
 ### Method Signature
 
 ```python Code
-def kickoff_async(self, inputs: dict) -> CrewOutput:
+async def akickoff(self, inputs: dict) -> CrewOutput:
 ```
 
 ### Parameters
@@ -28,23 +39,13 @@ def kickoff_async(self, inputs: dict) -> CrewOutput:
 
 - `CrewOutput`: An object representing the result of the crew execution.
 
-## Potential Use Cases
-
-- **Parallel Content Generation**: Kickoff multiple independent crews asynchronously, each responsible for generating content on different topics. For example, one crew might research and draft an article on AI trends, while another crew generates social media posts about a new product launch. Each crew operates independently, allowing content production to scale efficiently.
-
-- **Concurrent Market Research Tasks**: Launch multiple crews asynchronously to conduct market research in parallel. One crew might analyze industry trends, while another examines competitor strategies, and yet another evaluates consumer sentiment. Each crew independently completes its task, enabling faster and more comprehensive insights.
-
-- **Independent Travel Planning Modules**: Execute separate crews to independently plan different aspects of a trip. One crew might handle flight options, another handles accommodation, and a third plans activities. Each crew works asynchronously, allowing various components of the trip to be planned simultaneously and independently for faster results.
-
-## Example: Single Asynchronous Crew Execution
-
-Here's an example of how to kickoff a crew asynchronously using asyncio and awaiting the result:
+### Example: Native Async Crew Execution
 
 ```python Code
 import asyncio
 from crewai import Crew, Agent, Task
 
-# Create an agent with code execution enabled
+# Create an agent
 coding_agent = Agent(
     role="Python Data Analyst",
     goal="Analyze data and provide insights using Python",
@@ -52,37 +53,165 @@ coding_agent = Agent(
     allow_code_execution=True
 )
 
-# Create a task that requires code execution
+# Create a task
 data_analysis_task = Task(
     description="Analyze the given dataset and calculate the average age of participants. Ages: {ages}",
     agent=coding_agent,
     expected_output="The average age of the participants."
 )
 
-# Create a crew and add the task
+# Create a crew
 analysis_crew = Crew(
     agents=[coding_agent],
     tasks=[data_analysis_task]
 )
 
-# Async function to kickoff the crew asynchronously
-async def async_crew_execution():
-    result = await analysis_crew.kickoff_async(inputs={"ages": [25, 30, 35, 40, 45]})
+# Native async execution
+async def main():
+    result = await analysis_crew.akickoff(inputs={"ages": [25, 30, 35, 40, 45]})
     print("Crew Result:", result)
 
-# Run the async function
-asyncio.run(async_crew_execution())
+asyncio.run(main())
 ```
 
-## Example: Multiple Asynchronous Crew Executions
+### Example: Multiple Native Async Crews
 
-In this example, we'll show how to kickoff multiple crews asynchronously and wait for all of them to complete using `asyncio.gather()`:
+Run multiple crews concurrently using `asyncio.gather()` with native async:
+
+```python Code
+import asyncio
+from crewai import Crew, Agent, Task
+
+coding_agent = Agent(
+    role="Python Data Analyst",
+    goal="Analyze data and provide insights using Python",
+    backstory="You are an experienced data analyst with strong Python skills.",
+    allow_code_execution=True
+)
+
+task_1 = Task(
+    description="Analyze the first dataset and calculate the average age. Ages: {ages}",
+    agent=coding_agent,
+    expected_output="The average age of the participants."
+)
+
+task_2 = Task(
+    description="Analyze the second dataset and calculate the average age. Ages: {ages}",
+    agent=coding_agent,
+    expected_output="The average age of the participants."
+)
+
+crew_1 = Crew(agents=[coding_agent], tasks=[task_1])
+crew_2 = Crew(agents=[coding_agent], tasks=[task_2])
+
+async def main():
+    results = await asyncio.gather(
+        crew_1.akickoff(inputs={"ages": [25, 30, 35, 40, 45]}),
+        crew_2.akickoff(inputs={"ages": [20, 22, 24, 28, 30]})
+    )
+
+    for i, result in enumerate(results, 1):
+        print(f"Crew {i} Result:", result)
+
+asyncio.run(main())
+```
+
+### Example: Native Async for Multiple Inputs
+
+Use `akickoff_for_each()` to execute your crew against multiple inputs concurrently with native async:
+
+```python Code
+import asyncio
+from crewai import Crew, Agent, Task
+
+coding_agent = Agent(
+    role="Python Data Analyst",
+    goal="Analyze data and provide insights using Python",
+    backstory="You are an experienced data analyst with strong Python skills.",
+    allow_code_execution=True
+)
+
+data_analysis_task = Task(
+    description="Analyze the dataset and calculate the average age. Ages: {ages}",
+    agent=coding_agent,
+    expected_output="The average age of the participants."
+)
+
+analysis_crew = Crew(
+    agents=[coding_agent],
+    tasks=[data_analysis_task]
+)
+
+async def main():
+    datasets = [
+        {"ages": [25, 30, 35, 40, 45]},
+        {"ages": [20, 22, 24, 28, 30]},
+        {"ages": [30, 35, 40, 45, 50]}
+    ]
+
+    results = await analysis_crew.akickoff_for_each(datasets)
+
+    for i, result in enumerate(results, 1):
+        print(f"Dataset {i} Result:", result)
+
+asyncio.run(main())
+```
+
+## Thread-Based Async with `kickoff_async()`
+
+The `kickoff_async()` method provides async execution by wrapping the synchronous `kickoff()` in a thread. This is useful for simpler async integration or backward compatibility.
+
+### Method Signature
+
+```python Code
+async def kickoff_async(self, inputs: dict) -> CrewOutput:
+```
+
+### Parameters
+
+- `inputs` (dict): A dictionary containing the input data required for the tasks.
+
+### Returns
+
+- `CrewOutput`: An object representing the result of the crew execution.
+
+### Example: Thread-Based Async Execution
+
+```python Code
+import asyncio
+from crewai import Crew, Agent, Task
+
+coding_agent = Agent(
+    role="Python Data Analyst",
+    goal="Analyze data and provide insights using Python",
+    backstory="You are an experienced data analyst with strong Python skills.",
+    allow_code_execution=True
+)
+
+data_analysis_task = Task(
+    description="Analyze the given dataset and calculate the average age of participants. Ages: {ages}",
+    agent=coding_agent,
+    expected_output="The average age of the participants."
+)
+
+analysis_crew = Crew(
+    agents=[coding_agent],
+    tasks=[data_analysis_task]
+)
+
+async def async_crew_execution():
+    result = await analysis_crew.kickoff_async(inputs={"ages": [25, 30, 35, 40, 45]})
+    print("Crew Result:", result)
+
+asyncio.run(async_crew_execution())
+```
+
+### Example: Multiple Thread-Based Async Crews
 
 ```python Code
 import asyncio
 from crewai import Crew, Agent, Task
 
-# Create an agent with code execution enabled
 coding_agent = Agent(
     role="Python Data Analyst",
     goal="Analyze data and provide insights using Python",
@@ -90,7 +219,6 @@ coding_agent = Agent(
     allow_code_execution=True
 )
 
-# Create tasks that require code execution
 task_1 = Task(
     description="Analyze the first dataset and calculate the average age of participants. Ages: {ages}",
     agent=coding_agent,
@@ -103,22 +231,76 @@ task_2 = Task(
     expected_output="The average age of the participants."
 )
 
-# Create two crews and add tasks
 crew_1 = Crew(agents=[coding_agent], tasks=[task_1])
 crew_2 = Crew(agents=[coding_agent], tasks=[task_2])
 
-# Async function to kickoff multiple crews asynchronously and wait for all to finish
 async def async_multiple_crews():
-    # Create coroutines for concurrent execution
     result_1 = crew_1.kickoff_async(inputs={"ages": [25, 30, 35, 40, 45]})
     result_2 = crew_2.kickoff_async(inputs={"ages": [20, 22, 24, 28, 30]})
 
-    # Wait for both crews to finish
     results = await asyncio.gather(result_1, result_2)
 
     for i, result in enumerate(results, 1):
         print(f"Crew {i} Result:", result)
 
-# Run the async function
 asyncio.run(async_multiple_crews())
 ```
+
+## Async Streaming
+
+Both async methods support streaming when `stream=True` is set on the crew:
+
+```python Code
+import asyncio
+from crewai import Crew, Agent, Task
+
+agent = Agent(
+    role="Researcher",
+    goal="Research and summarize topics",
+    backstory="You are an expert researcher."
+)
+
+task = Task(
+    description="Research the topic: {topic}",
+    agent=agent,
+    expected_output="A comprehensive summary of the topic."
+)
+
+crew = Crew(
+    agents=[agent],
+    tasks=[task],
+    stream=True  # Enable streaming
+)
+
+async def main():
+    streaming_output = await crew.akickoff(inputs={"topic": "AI trends in 2024"})
+
+    # Async iteration over streaming chunks
+    async for chunk in streaming_output:
+        print(f"Chunk: {chunk.content}")
+
+    # Access final result after streaming completes
+    result = streaming_output.result
+    print(f"Final result: {result.raw}")
+
+asyncio.run(main())
+```
+
+## Potential Use Cases
+
+- **Parallel Content Generation**: Kickoff multiple independent crews asynchronously, each responsible for generating content on different topics. For example, one crew might research and draft an article on AI trends, while another crew generates social media posts about a new product launch.
+
+- **Concurrent Market Research Tasks**: Launch multiple crews asynchronously to conduct market research in parallel. One crew might analyze industry trends, while another examines competitor strategies, and yet another evaluates consumer sentiment.
+
+- **Independent Travel Planning Modules**: Execute separate crews to independently plan different aspects of a trip. One crew might handle flight options, another handles accommodation, and a third plans activities.
+
+## Choosing Between `akickoff()` and `kickoff_async()`
+
+| Feature | `akickoff()` | `kickoff_async()` |
+|---------|--------------|-------------------|
+| Execution model | Native async/await | Thread-based wrapper |
+| Task execution | Async with `aexecute_sync()` | Sync in thread pool |
+| Memory operations | Async | Sync in thread pool |
+| Knowledge retrieval | Async | Sync in thread pool |
+| Best for | High-concurrency, I/O-bound workloads | Simple async integration |
+| Streaming support | Yes | Yes |

docs/en/learn/streaming-crew-execution.mdx

@@ -95,7 +95,11 @@ print(f"Final result: {streaming.result.raw}")
 
 ## Asynchronous Streaming
 
-For async applications, use `kickoff_async()` with async iteration:
+For async applications, you can use either `akickoff()` (native async) or `kickoff_async()` (thread-based) with async iteration:
+
+### Native Async with `akickoff()`
+
+The `akickoff()` method provides true native async execution throughout the entire chain:
 
 ```python Code
 import asyncio
@@ -107,7 +111,35 @@ async def stream_crew():
         stream=True
     )
 
-    # Start async streaming
+    # Start native async streaming
+    streaming = await crew.akickoff(inputs={"topic": "AI"})
+
+    # Async iteration over chunks
+    async for chunk in streaming:
+        print(chunk.content, end="", flush=True)
+
+    # Access final result
+    result = streaming.result
+    print(f"\n\nFinal output: {result.raw}")
+
+asyncio.run(stream_crew())
+```
+
+### Thread-Based Async with `kickoff_async()`
+
+For simpler async integration or backward compatibility:
+
+```python Code
+import asyncio
+
+async def stream_crew():
+    crew = Crew(
+        agents=[researcher],
+        tasks=[task],
+        stream=True
+    )
+
+    # Start thread-based async streaming
     streaming = await crew.kickoff_async(inputs={"topic": "AI"})
 
     # Async iteration over chunks
@@ -121,6 +153,10 @@ async def stream_crew():
 asyncio.run(stream_crew())
 ```
 
+<Note>
+For high-concurrency workloads, `akickoff()` is recommended as it uses native async for task execution, memory operations, and knowledge retrieval. See the [Kickoff Crew Asynchronously](/en/learn/kickoff-async) guide for more details.
+</Note>
+
 ## Streaming with kickoff_for_each
 
 When executing a crew for multiple inputs with `kickoff_for_each()`, streaming works differently depending on whether you use sync or async: