异步启动团队 - CrewAI 框架

简介

CrewAI 提供了异步启动团队的能力，允许您以非阻塞方式开始团队执行。当您想要同时运行多个团队，或者在团队执行期间需要执行其他任务时，此功能特别有用。 CrewAI 提供了两种异步执行方案：

方法	类型	描述
`akickoff()`	原生异步	整个执行链中真正的异步/等待 (async/await)
`kickoff_async()`	基于线程	将同步执行封装在 `asyncio.to_thread` 中

对于高并发工作负载，推荐使用 akickoff()，因为它在任务执行、内存操作和知识检索中使用了原生异步。

使用 `akickoff()` 进行原生异步执行

akickoff() 方法提供了真正的原生异步执行，在整个执行链（包括任务执行、内存操作和知识查询）中使用 async/await。

方法签名

代码

async def akickoff(self, inputs: dict) -> CrewOutput:

参数

inputs (dict)：包含任务所需输入数据的字典。

返回值

CrewOutput：表示团队执行结果的对象。

示例：原生异步团队执行

代码

import asyncio
from crewai import Crew, Agent, Task

# Create an agent
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
)

# 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
analysis_crew = Crew(
    agents=[coding_agent],
    tasks=[data_analysis_task]
)

# Native async execution
async def main():
    result = await analysis_crew.akickoff(inputs={"ages": [25, 30, 35, 40, 45]})
    print("Crew Result:", result)

asyncio.run(main())

示例：多个原生异步团队

使用 asyncio.gather() 和原生异步同时运行多个团队

代码

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())

示例：针对多输入的原生异步

使用 akickoff_for_each() 通过原生异步同时针对多个输入执行团队

代码

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())

使用 `kickoff_async()` 进行基于线程的异步执行

kickoff_async() 方法通过将同步 kickoff() 封装在线程中来提供异步执行。这对于更简单的异步集成或向后兼容非常有用。

方法签名

代码

async def kickoff_async(self, inputs: dict) -> CrewOutput:

参数

inputs (dict)：包含任务所需输入数据的字典。

返回值

CrewOutput：表示团队执行结果的对象。

示例：基于线程的异步执行

代码

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())

示例：多个基于线程的异步团队

代码

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 of participants. 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 of participants. 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 async_multiple_crews():
    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]})

    results = await asyncio.gather(result_1, result_2)

    for i, result in enumerate(results, 1):
        print(f"Crew {i} Result:", result)

asyncio.run(async_multiple_crews())

异步流式传输

当团队设置了 stream=True 时，两种异步方法均支持流式传输

代码

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())

潜在用例

并行内容生成：异步启动多个独立的团队，每个团队负责生成不同主题的内容。例如，一个团队可能在研究并起草关于人工智能趋势的文章，而另一个团队则生成关于新产品发布会的社交媒体帖子。
并发市场调研任务：异步启动多个团队并行进行市场调研。一个团队可以分析行业趋势，另一个检查竞争对手策略，第三个评估消费者情绪。
独立旅行规划模块：执行单独的团队来独立规划旅行的不同方面。一个团队可以处理航班选项，另一个处理住宿，第三个规划活动。

在 `akickoff()` 和 `kickoff_async()` 之间进行选择

特性	`akickoff()`	`kickoff_async()`
执行模型	原生 async/await	基于线程的包装器
任务执行	使用 `aexecute_sync()` 的异步	线程池中的同步
内存操作	异步	线程池中的同步
知识检索	异步	线程池中的同步
最适合	高并发、I/O 密集型工作负载	简单的异步集成
流式传输支持	是	是

文档索引

​简介

​使用 akickoff() 进行原生异步执行

​方法签名

​参数

​返回值

​示例：原生异步团队执行

​示例：多个原生异步团队

​示例：针对多输入的原生异步

​使用 kickoff_async() 进行基于线程的异步执行

​方法签名

​参数

​返回值

​示例：基于线程的异步执行

​示例：多个基于线程的异步团队

​异步流式传输

​潜在用例

​在 akickoff() 和 kickoff_async() 之间进行选择

简介

使用 `akickoff()` 进行原生异步执行

方法签名

参数

返回值

示例：原生异步团队执行

示例：多个原生异步团队

示例：针对多输入的原生异步

使用 `kickoff_async()` 进行基于线程的异步执行

方法签名

参数

返回值

示例：基于线程的异步执行

示例：多个基于线程的异步团队

异步流式传输

潜在用例

在 `akickoff()` 和 `kickoff_async()` 之间进行选择