单个大语言模型可以回答问题,但复杂任务通常不只需要“生成一段文字”:它还涉及资料检索、事实整理、方案撰写和质量检查。CrewAI 的核心思路,是把这些步骤分配给具有不同角色、目标和工具的智能体,再通过明确的任务依赖把它们组织成一条工作流。
这种模式并不是简单地多调用几次模型。真正有价值的部分在于职责拆分:谁收集信息、谁形成结论、谁检查输出,以及每一步必须交付什么。
从一次调用转向角色协作
在 CrewAI 中,一个典型的多智能体应用包含四类对象:
Agent:定义角色、目标、背景和可用工具。Task:描述具体工作、预期输出以及负责人。Crew:把智能体和任务组合起来。Process:决定任务顺序,例如按顺序执行,或采用更复杂的分层协调方式。
例如,要生成一份技术选型简报,可以把流程拆成三个角色:
- 研究员从可信资料中提取事实。
- 架构师基于事实比较方案并给出建议。
- 审校员检查建议是否缺少依据、风险或实施条件。
这里的重点不是给每个智能体起一个拟人化名字,而是让每个角色拥有清晰的输入、工具权限和验收标准。如果三个 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 提供的是一种编排方式,而不是自动保证正确性的机制。先从两到三个角色、顺序执行和只读工具开始,建立可观察、可测试的基线,再根据真实瓶颈增加并行、分层管理或写入能力,通常比一开始设计庞大的智能体组织更容易落地。