CrewAI: 51K+ 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"