用 CrewAI 在 Python 中编排多智能体团队:角色、工具与任务流实战

2026-07-29 29 预计阅读时间: 1 分钟
来源: realpython.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:11 分钟

单个大语言模型可以回答问题,但复杂任务通常不只需要“生成一段文字”:它还涉及资料检索、事实整理、方案撰写和质量检查。CrewAI 的核心思路,是把这些步骤分配给具有不同角色、目标和工具的智能体,再通过明确的任务依赖把它们组织成一条工作流。

这种模式并不是简单地多调用几次模型。真正有价值的部分在于职责拆分:谁收集信息、谁形成结论、谁检查输出,以及每一步必须交付什么。

从一次调用转向角色协作

在 CrewAI 中,一个典型的多智能体应用包含四类对象:

  • Agent:定义角色、目标、背景和可用工具。
  • Task:描述具体工作、预期输出以及负责人。
  • Crew:把智能体和任务组合起来。
  • Process:决定任务顺序,例如按顺序执行,或采用更复杂的分层协调方式。

例如,要生成一份技术选型简报,可以把流程拆成三个角色:

  1. 研究员从可信资料中提取事实。
  2. 架构师基于事实比较方案并给出建议。
  3. 审校员检查建议是否缺少依据、风险或实施条件。

这里的重点不是给每个智能体起一个拟人化名字,而是让每个角色拥有清晰的输入、工具权限和验收标准。如果三个 Agent 都收到“分析这个主题”的宽泛指令,团队只会产生更多重复内容。

一个可运行的顺序协作示例

下面构建一个最小项目:智能体读取本地知识库,生成技术调研报告,再由审校角色整理最终版本。示例假设使用支持 CrewAI 的 OpenAI 模型;不同 CrewAI 版本对模型名称和参数的支持可能略有差异,实际项目应锁定依赖版本。

创建项目并安装依赖:

mkdir crewai-team-demo
cd crewai-team-demo
python -m venv .venv
source .venv/bin/activate
pip install "crewai>=0.80" python-dotenv

Windows PowerShell 激活命令为:

.venv\Scripts\Activate.ps1

配置环境变量:

export OPENAI_API_KEY="your-api-key"

创建 knowledge_base.md,放入团队已经确认的内部资料:

# 服务背景

现有服务使用 Python,部署在 Kubernetes 中。
团队需要定时执行数据处理任务,并保留失败重试记录。
当前规模约为每天 5000 个任务,单个任务耗时 10 秒到 5 分钟。

# 约束

- 运维团队熟悉 PostgreSQL 和 Kubernetes。
- 新组件必须支持可观测性和失败重试。
- 首期不希望维护过多基础设施。

再创建 main.py

from pathlib import Path

from crewai import Agent, Crew, Process, Task
from crewai.tools import tool


@tool("read_internal_knowledge")
def read_internal_knowledge() -> str:
    """Read the approved local knowledge base for the architecture review."""
    return Path("knowledge_base.md").read_text(encoding="utf-8")


researcher = Agent(
    role="Technical Researcher",
    goal="Extract verified requirements, constraints, and unknowns from the internal material",
    backstory=(
        "You are a careful researcher. You distinguish documented facts "
        "from assumptions and never invent missing requirements."
    ),
    tools=[read_internal_knowledge],
    llm="openai/gpt-4o-mini",
    verbose=True,
)

architect = Agent(
    role="Software Architect",
    goal="Compare practical implementation options and recommend one with explicit tradeoffs",
    backstory=(
        "You design production Python systems and prefer operationally simple "
        "solutions that satisfy stated constraints."
    ),
    llm="openai/gpt-4o-mini",
    verbose=True,
)

reviewer = Agent(
    role="Technical Reviewer",
    goal="Produce a concise final report whose claims are supported by the research",
    backstory=(
        "You review architecture proposals for unsupported claims, omitted risks, "
        "and vague implementation steps."
    ),
    llm="openai/gpt-4o-mini",
    verbose=True,
)

research_task = Task(
    description=(
        "Read the internal knowledge base and analyze the requirements for {topic}. "
        "Return documented facts, constraints, open questions, and evaluation criteria. "
        "Label every assumption explicitly."
    ),
    expected_output=(
        "A Markdown research note with sections for facts, constraints, "
        "open questions, assumptions, and evaluation criteria."
    ),
    agent=researcher,
)

architecture_task = Task(
    description=(
        "Using the research note, compare 2-3 implementation options for {topic}. "
        "Evaluate reliability, operational cost, observability, retry behavior, "
        "and migration effort. Recommend one option without inventing requirements."
    ),
    expected_output=(
        "A Markdown decision proposal containing an option table, recommendation, "
        "tradeoffs, and a phased implementation plan."
    ),
    agent=architect,
    context=[research_task],
)

review_task = Task(
    description=(
        "Review the research and architecture proposal. Remove unsupported claims, "
        "preserve explicitly labeled assumptions, and add missing risks or validation steps."
    ),
    expected_output=(
        "A final Markdown architecture brief with requirements, options, decision, "
        "risks, unresolved questions, and next actions."
    ),
    agent=reviewer,
    context=[research_task, architecture_task],
    output_file="architecture_brief.md",
)

crew = Crew(
    agents=[researcher, architect, reviewer],
    tasks=[research_task, architecture_task, review_task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff(
    inputs={"topic": "a reliable scheduler for Python background jobs"}
)

print(result)

运行工作流:

python main.py

执行完成后,终端会显示最终结果,审校任务还会把报告写入 architecture_brief.md。由于示例只允许研究员读取本地文件,模型无法自行验证知识库之外的事实;如果要比较具体产品,应接入经过审核的搜索、数据库或内部文档工具。

任务边界比角色描述更重要

多智能体流程的质量很大程度上取决于 Task 是否可验收。下面两种任务描述看似相近,实际效果差别很大:

较弱:研究 Python 后台任务方案并给出建议。

较强:比较 2-3 个方案,按可靠性、运维成本、可观测性、
失败重试和迁移成本进行评估;区分事实与假设,并输出决策表。

第二种描述规定了比较范围、评价维度和输出格式。下游智能体因此能够消费结构相对稳定的结果,审校员也有具体标准判断内容是否完整。

生产项目中还应控制上下文规模。不要默认把所有历史消息、工具结果和中间草稿传给每个 Agent。更稳妥的做法是让上游任务输出结构化摘要,只把与当前决策有关的内容交给下游。

工具权限决定系统边界

工具让 Agent 能够读取文件、查询数据库、调用 HTTP API,甚至执行变更操作。工具越强,风险越高。实践中可以按能力分级:

  • 只读工具:文档检索、指标查询、工单读取。
  • 受限写入工具:创建草稿、提交待审批记录。
  • 高风险工具:修改生产配置、删除资源、发送外部通知。

高风险动作不应仅依赖提示词约束。应在工具函数中加入参数校验、权限检查、超时、重试、审计日志和人工审批。对于“研究后自动部署”一类工作流,研究与部署最好拆成不同执行阶段,并在中间设置可验证的审批条件。

还要考虑提示注入:外部网页或文档中的文字可能伪装成系统指令。工具返回的内容应被视为不可信数据,不能让文档自行扩大 Agent 权限或覆盖原始任务。

什么时候值得使用多智能体

多智能体并不适合所有问题。简单摘要、格式转换和单次分类通常使用一个模型调用更便宜、更快,也更容易测试。以下情况更适合拆分团队:

  • 任务包含明显不同的专业职责。
  • 中间结果需要独立检查或修订。
  • 不同步骤必须访问不同工具或权限。
  • 工作流需要保留决策记录和阶段性交付物。
  • 单个提示已经变得过长,且输入输出边界难以维护。

上线前可以检查以下事项:

  • 每个 Agent 是否只有一个清晰职责。
  • 每个 Task 是否定义了可验证的 expected_output
  • 下游任务是否只接收必要上下文。
  • 外部数据是否按不可信输入处理。
  • 写操作是否经过代码级授权和人工审批。
  • 是否记录模型、提示、工具调用、耗时和成本。
  • 是否准备固定测试样本,用来比较流程修改前后的质量。

CrewAI 提供的是一种编排方式,而不是自动保证正确性的机制。先从两到三个角色、顺序执行和只读工具开始,建立可观察、可测试的基线,再根据真实瓶颈增加并行、分层管理或写入能力,通常比一开始设计庞大的智能体组织更容易落地。


相关推荐