CrewAI: 51000+ Star 构建多智能体 AI 团队 — 完整设置指南 2026

# 检查 Python 版本
python --version  # 必须是 3.10、3.11、3.12 或 3.13

# 安装 uv（推荐的包管理器）
curl -LsSf https://astral.sh/uv/install.sh | sh

# 安装 CrewAI 核心框架
uv pip install crewai

# 或安装包含可选工具的版本
uv pip install 'crewai[tools]'

# 验证安装
crewai version

# 搭建新的 CrewAI 项目
crewai create crew research_crew

# 进入项目目录
cd research_crew

# 安装项目依赖
crewai install

# .env — 将此文件添加到 .gitignore！
OPENAI_API_KEY=sk-your-openai-key-here
SERPER_API_KEY=your-serper-api-key

# 拉取本地模型
ollama pull llama3.1

# 在智能体配置中使用：ollama/llama3.1

# src/research_crew/config/agents.yaml

researcher:
  role: >
    Senior Research Analyst
  goal: >
    Conduct thorough research on {topic} and gather
    comprehensive, accurate, and up-to-date information.
  backstory: >
    You are a seasoned research analyst with 15 years of
    experience in technology analysis. You find obscure but
    critical data points and synthesize complex information.
  llm: openai/gpt-4o
  max_iter: 15
  verbose: true

writer:
  role: >
    Technical Content Writer
  goal: >
    Transform research findings on {topic} into a
    well-structured, engaging article.
  backstory: >
    You are an award-winning technical writer who excels at
    making complex topics accessible without sacrificing accuracy.
  llm: openai/gpt-4o
  max_iter: 10
  verbose: true

editor:
  role: >
    Senior Content Editor
  goal: >
    Review and polish the article about {topic} to ensure
    factual accuracy, grammar, and readability.
  backstory: >
    You are a meticulous editor with a sharp eye for factual
    errors and logical inconsistencies.
  llm: openai/gpt-4o-mini
  max_iter: 8
  verbose: true

# src/research_crew/config/tasks.yaml

research_task:
  description: >
    Research the topic: {topic}. Gather at least 10 key data points
    from multiple authoritative sources. Include statistics,
    expert opinions, and recent developments.
  expected_output: >
    A structured research brief with 10+ data points, source
    citations, and a summary of key findings.
  agent: researcher

writing_task:
  description: >
    Using the research brief provided, write a comprehensive
    technical article about {topic}. Target 1500 words.
    Use clear headings, examples, and engaging prose.
  expected_output: >
    A markdown-formatted article with introduction, body
    sections, and conclusion. Minimum 1500 words.
  agent: writer
  context: [research_task]

editing_task:
  description: >
    Edit the article for clarity, grammar, factual accuracy,
    and readability. Ensure all claims are supported by the
    research brief.
  expected_output: >
    A polished final article ready for publication.
    Include an editor's note summarizing changes made.
  agent: editor
  context: [writing_task]
  output_file: output/final_article.md

# src/research_crew/crew.py
from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task

@CrewBase
class ResearchCrew:
    """Research crew for producing high-quality articles."""

    agents_config = "config/agents.yaml"
    tasks_config = "config/tasks.yaml"

    @agent
    def researcher(self) -> Agent:
        return Agent(
            config=self.agents_config["researcher"],
            tools=[],
            allow_delegation=False,
        )

    @agent
    def writer(self) -> Agent:
        return Agent(
            config=self.agents_config["writer"],
            tools=[],
            allow_delegation=False,
        )

    @agent
    def editor(self) -> Agent:
        return Agent(
            config=self.agents_config["editor"],
            tools=[],
            allow_delegation=False,
        )

    @task
    def research_task(self) -> Task:
        return Task(config=self.tasks_config["research_task"])

    @task
    def writing_task(self) -> Task:
        return Task(config=self.tasks_config["writing_task"])

    @task
    def editing_task(self) -> Task:
        return Task(
            config=self.tasks_config["editing_task"],
            output_file="output/final_article.md",
        )

    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=self.agents,
            tasks=self.tasks,
            process=Process.sequential,
            verbose=True,
        )

# src/research_crew/main.py
#!/usr/bin/env python
from research_crew.crew import ResearchCrew

def run():
    """Run the research crew."""
    inputs = {
        "topic": "AI coding assistants in 2026"
    }
    result = ResearchCrew().crew().kickoff(inputs=inputs)
    print("\n\n========== FINAL OUTPUT ==========\n")
    print(result.raw)
    print(f"\nToken usage: {result.token_usage}")

if __name__ == "__main__":
    run()

# 通过 CLI 执行
crewai run

# 或通过 Python 直接运行
python -m research_crew.main

# src/research_crew/flow.py
from crewai.flow.flow import Flow, listen, start
from pydantic import BaseModel
from research_crew.crew import ResearchCrew

class ArticleState(BaseModel):
    topic: str = ""
    word_count: int = 0
    final_article: str = ""

class ArticleFlow(Flow[ArticleState]):

    @start()
    def get_topic(self):
        self.state.topic = "Multi-agent AI frameworks in 2026"
        print(f"Starting flow for topic: {self.state.topic}")

    @listen(get_topic)
    def run_research_crew(self):
        result = ResearchCrew().crew().kickoff(
            inputs={"topic": self.state.topic}
        )
        self.state.final_article = result.raw
        self.state.word_count = len(result.raw.split())

    @listen(run_research_crew)
    def validate_output(self):
        if self.state.word_count < 1000:
            print("WARNING: Article too short, triggering revision")
        else:
            print(f"Article validated: {self.state.word_count} words")
            with open("output/article.md", "w") as f:
                f.write(self.state.final_article)

if __name__ == "__main__":
    ArticleFlow().kickoff()

# src/research_crew/tools/custom_tool.py
from crewai.tools import tool
import requests

@tool("Web Search")
def web_search(query: str) -> str:
    """Search the web for information on a given query."""
    response = requests.get(
        "https://serpapi.com/search",
        params={"q": query, "api_key": "${SERPER_API_KEY}"}
    )
    return response.json()["organic_results"][0]["snippet"]

# 在 crew.py 中，导入并附加
from research_crew.tools.custom_tool import web_search

@agent
def researcher(self) -> Agent:
    return Agent(
        config=self.agents_config["researcher"],
        tools=[web_search],  # 附加自定义工具
        allow_delegation=False,
    )

@crew
def crew(self) -> Crew:
    return Crew(
        agents=self.agents,
        tasks=self.tasks,
        process=Process.hierarchical,
        manager_llm="openai/gpt-4o",
        verbose=True,
    )

# api_server.py — 生产部署
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
from research_crew.crew import ResearchCrew
import uuid

app = FastAPI(title="CrewAI Research API")
jobs: dict = {}

class CrewRequest(BaseModel):
    topic: str

@app.post("/research")
async def start_research(request: CrewRequest, background: BackgroundTasks):
    job_id = str(uuid.uuid4())
    jobs[job_id] = {"status": "queued", "topic": request.topic}
    background.add_task(
        lambda: run_crew(job_id, request.topic)
    )
    return {"job_id": job_id, "status": "queued"}

@app.get("/status/{job_id}")
async def get_status(job_id: str):
    return jobs.get(job_id, {"error": "Job not found"})

def run_crew(job_id: str, topic: str):
    jobs[job_id]["status"] = "running"
    result = ResearchCrew().crew().kickoff(inputs={"topic": topic})
    jobs[job_id]["status"] = "completed"
    jobs[job_id]["result"] = result.raw

# 运行：uvicorn api_server:app --host 0.0.0.0 --port 8000

# agents.yaml — 每个智能体的模型选择
researcher:
  role: Research Analyst
  llm: anthropic/claude-sonnet-4-20250514
  # 或: openai/gpt-4o
  # 或: gemini/gemini-2.0-flash

researcher:
  role: Research Analyst
  llm: ollama/llama3.1
  # 需要：ollama pull llama3.1

# 在 CrewAI 中使用 LangChain 工具
from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper
from crewai import Agent

wiki_tool = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())

agent = Agent(
    role="Researcher",
    goal="Research topics thoroughly",
    tools=[wiki_tool],  # LangChain 工具直接可用
    verbose=True,
)

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from crewai.tools import tool

@tool("Document Search")
def document_search(query: str) -> str:
    """Search internal documents for relevant information."""
    documents = SimpleDirectoryReader("./docs").load_data()
    index = VectorStoreIndex.from_documents(documents)
    query_engine = index.as_query_engine()
    return str(query_engine.query(query))

# Dockerfile
FROM python:3.12-slim

WORKDIR /app
COPY pyproject.toml .
COPY src/ ./src/

RUN pip install crewai crewai-tools
RUN crewai install

EXPOSE 8000

CMD ["crewai", "run"]

# docker-compose.yml
version: "3.8"
services:
  crewai:
    build: .
    env_file: .env
    volumes:
      - ./output:/app/output
    ports:
      - "8000:8000"