[AI] 🤝 CrewAI: 멀티 에이전트 AI 협업 프레임워크 완벽 가이드

from crewai_tools import SerperDevTool, ScrapeWebsiteTool, WebsiteSearchTool

search_tool = SerperDevTool()      # Google 검색
scrape_tool = ScrapeWebsiteTool()  # 웹 페이지 스크래핑
rag_tool    = WebsiteSearchTool()  # 웹 콘텐츠 RAG 검색

import os
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, ScrapeWebsiteTool

load_dotenv()
os.environ["OPENAI_MODEL_NAME"] = "gpt-4o"

# ── Tools ────────────────────────────────────────
search_tool = SerperDevTool()
scrape_tool = ScrapeWebsiteTool()

# ── Agents ───────────────────────────────────────
researcher = Agent(
    role="시니어 리서처",
    goal="주어진 주제에 대해 웹에서 최신 정보를 수집하고 핵심 인사이트를 정리한다",
    backstory="다양한 기술 문서와 블로그를 분석하여 최고의 리서치 결과를 만드는 전문가",
    tools=[search_tool, scrape_tool],
    allow_delegation=False,
    verbose=True,
)

editor = Agent(
    role="IT 블로거 / 에디터",
    goal="리서치 결과를 바탕으로 읽기 쉽고 실용적인 블로그 글을 작성한다",
    backstory="개발자 독자를 위해 명확하고 유익한 기술 콘텐츠를 작성하는 전문 블로거",
    verbose=True,
)

# ── Tasks ────────────────────────────────────────
research_task = Task(
    description="""'{topic}'에 관한 최신 정보를 웹에서 검색하고 분석하세요.
    핵심 개념, 주요 기능, 실제 활용 사례를 정리하세요.""",
    agent=researcher,
    expected_output="주제의 핵심 내용, 주요 기능, 활용 사례를 포함한 리서치 보고서",
)

writing_task = Task(
    description="""리서치 결과를 바탕으로 '{topic}'에 대한 블로그 포스트를 작성하세요.
    도입부, 주요 3개 섹션, 실용적 예시, 결론으로 구성하세요.""",
    agent=editor,
    expected_output="마크다운 형식의 블로그 포스트 (도입부, 3개 섹션, 결론 포함)",
    context=[research_task],       # 리서치 결과를 컨텍스트로 전달
    output_file="output_blog.md",
)

# ── Crew ─────────────────────────────────────────
crew = Crew(
    agents=[researcher, editor],
    tasks=[research_task, writing_task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff(inputs={"topic": "CrewAI 멀티 에이전트 프레임워크"})
print(result.raw)

crew = Crew(
    agents=[researcher, editor, reviewer],
    tasks=[research_task, writing_task, review_task],
    process=Process.sequential,
)

from langchain_openai import ChatOpenAI

crew = Crew(
    agents=[researcher, editor, reviewer],
    tasks=[complex_task],
    process=Process.hierarchical,
    manager_llm=ChatOpenAI(model="gpt-4o"),  # 매니저 LLM 지정
)

researcher = Agent(
    role="리서처",
    goal="정보 수집",
    backstory="...",
    memory=True,           # 메모리 활성화
)

crew = Crew(
    agents=[researcher],
    tasks=[task],
    memory=True,           # Crew 레벨 메모리 활성화
)

from crewai.tools import tool

@tool('naver_news_search')
def search_naver_news(query: str) -> str:
    """네이버 뉴스 API에서 최신 기사 URL을 검색합니다."""
    import requests
    # 실제 API 호출 구현
    response = requests.get(
        "https://openapi.naver.com/v1/search/news.json",
        params={"query": query, "display": 5},
        headers={
            "X-Naver-Client-Id": os.getenv("NAVER_CLIENT_ID"),
            "X-Naver-Client-Secret": os.getenv("NAVER_CLIENT_SECRET"),
        }
    )
    items = response.json().get("items", [])
    return "\n".join(item["link"] for item in items)

from crewai import Agent, Task, Crew, Process
from crewai.project import CrewBase, agent, task, crew

@CrewBase
class BlogCrew:
    agents_config = 'config/agents.yaml'
    tasks_config  = 'config/tasks.yaml'

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

    @agent
    def editor(self) -> Agent:
        return Agent(config=self.agents_config['editor'])

    @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'])

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

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

@tool('fetch_article')
def fetch_article(url: str) -> str:
    """주어진 URL의 기사 본문을 추출합니다."""
    from newspaper import Article
    article = Article(url)
    article.download()
    article.parse()
    return article.text

# Agents
searcher = Agent(
    role="뉴스 검색 전문가",
    goal="사용자 질문에 맞는 최신 뉴스 URL을 수집한다",
    backstory="다양한 뉴스 소스에서 정확한 기사를 찾는 리서치 전문가",
    tools=[search_tool],
)

analyzer = Agent(
    role="콘텐츠 분석가",
    goal="기사 URL에서 본문을 추출하고 핵심 내용을 분석한다",
    backstory="복잡한 기사를 빠르게 파악하고 핵심을 추출하는 분석가",
    tools=[fetch_article],
)

answerman = Agent(
    role="최종 응답 작성자",
    goal="분석된 정보를 바탕으로 사용자에게 명확한 답변을 제공한다",
    backstory="복잡한 정보를 쉽고 정확하게 전달하는 커뮤니케이션 전문가",
)

# Tasks
search_task = Task(
    description="'{query}'에 관한 최신 뉴스 기사 URL 5개를 수집하세요.",
    agent=searcher,
    expected_output="관련 뉴스 기사 URL 목록 (최소 5개)",
)

analysis_task = Task(
    description="수집된 URL에서 기사 본문을 추출하고 핵심 내용을 정리하세요.",
    agent=analyzer,
    expected_output="각 기사의 핵심 내용 요약",
    context=[search_task],
)

answer_task = Task(
    description="분석 결과를 바탕으로 '{query}'에 대한 종합 답변을 작성하세요.",
    agent=answerman,
    expected_output="출처와 근거를 포함한 명확한 답변",
    context=[analysis_task],
)

# Crew 실행
crew = Crew(
    agents=[searcher, analyzer, answerman],
    tasks=[search_task, analysis_task, answer_task],
    process=Process.sequential,
)

result = crew.kickoff(inputs={"query": "2026년 AI Agent 트렌드"})
print(result.raw)

from langchain_anthropic import ChatAnthropic

agent = Agent(
    role="분석가",
    goal="...",
    backstory="...",
    llm=ChatAnthropic(model="claude-sonnet-4-6"),
)