BytesRouter

主题

CrewAI 文档版本 跟随上游 源快照 334d2da1 更新于 2026-10-18

异步启动 Crew

异步启动 Crew

引言

CrewAI 支持异步启动 crew，让你以非阻塞方式启动 crew 执行。 当你希望并发运行多个 crew，或需要在 crew 执行期间执行其他任务时，该功能尤其有用。

CrewAI 提供两种异步执行方式：

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

NOTE

对于高并发工作负载，建议使用 akickoff()，因为它在任务执行、记忆操作和知识检索中都使用原生异步。

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

akickoff() 方法提供真正的原生异步执行，在整个执行链中贯穿 async/await，包括任务执行、记忆操作和知识查询。

方法签名

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

参数

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

返回值

  • CrewOutput：表示 crew 执行结果的对象。

示例：原生异步 Crew 执行

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

示例：多个原生异步 Crew

使用 asyncio.gather() 配合原生异步并发运行多个 crew：

python
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() 以原生异步方式针对多个输入并发执行你的 crew：

python
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() 包装在某个线程中来提供异步执行。这适用于较简单的异步集成或向后兼容。

方法签名

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

参数

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

返回值

  • CrewOutput：表示 crew 执行结果的对象。

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

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

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

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

异步流式传输

当在 crew 上设置 stream=True 时，两种异步方法都支持流式传输：

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

潜在用例

  • 并行内容生成：异步启动多个相互独立的 crew，每个负责生成不同主题的内容。例如，一个 crew 可能研究并起草一篇关于 AI 趋势的文章，而另一个 crew 生成关于新产品发布的社交媒体帖子。

  • 并发市场调研任务：异步启动多个 crew 并行开展市场调研。一个 crew 可能分析行业趋势，另一个考察竞争对手策略，还有一个评估消费者情绪。

  • 独立的旅行规划模块：执行独立的 crew，分别规划旅行的不同方面。一个 crew 可能处理航班选项，另一个处理住宿，第三个规划活动。

在 akickoff() 与 kickoff_async() 之间选择

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

友情链接

  • Agents.md面向 AI 编码助手的仓库说明文件规范
  • LangChainLangChain 是一个开放的Agent平台
  • LangSmithAgent 工程平台：可观测与评估、部署、Fleet、Engine 等
  • EmbabelEmbabel 是一个JVM平台Agent框架
  • SpringAISpringAI 是一个基于 Spring 的 AI 框架
  • MCPModel Context Protocol，连接 AI 应用与外部数据/工具的开放标准
  • LangChain4jJava 生态的 LLM 应用开发框架
  • LlamaIndex把数据接入 LLM 的框架与文档平台
  • LlamaParse智能体式 OCR 文档解析：把 PDF、扫描件变成 LLM 可用的文本与结构化数据
  • LiteParse本地优先、轻量的文档解析库，不需要 API key
  • CrewAI多智能体编排框架：以 Crew/Flow 组织自主协作的 AI Agent
  • MAFMicrosoft Agent Framework，企业级 .NET/Python 智能体框架

BytesRouter 是社区翻译站点，所有译文仅供参考，一切以官方英文原文为准。

© 2026 BytesRouter · 译文采用 CC BY 4.0 许可