CrewAI와 LangGraph로 역할 기반 멀티에이전트 파이프라인 구축하기

import os
from langgraph.graph import StateGraph, END
from typing import TypedDict
from langchain_anthropic import ChatAnthropic
from langchain_community.tools import TavilySearchResults
 
# 환경 변수 설정 (실행 전 아래 키를 발급받아 설정하세요)
# export ANTHROPIC_API_KEY="your-anthropic-api-key"
# export TAVILY_API_KEY="your-tavily-api-key"
 
# 1. 공유 상태 스키마 정의 — 에이전트 간 데이터를 타입 안전하게 전달합니다
class ResearchState(TypedDict):
    query: str
    raw_data: str
    analysis: str
    report: str
 
# 2. LLM 및 도구 초기화
llm = ChatAnthropic(model="claude-sonnet-4-6")
search_tool = TavilySearchResults(max_results=5)
 
def research_agent(state: ResearchState) -> ResearchState:
    """웹 검색을 수행해 원시 데이터를 수집합니다."""
    try:
        results = search_tool.invoke(state["query"])
        raw_data = "\n".join([r["content"] for r in results])
        return {"raw_data": raw_data}
    except Exception as e:
        # 프로덕션에서는 재시도 로직이나 폴백 데이터 소스를 추가하는 것을 권장합니다
        return {"raw_data": f"검색 오류: {str(e)}"}
 
def analysis_agent(state: ResearchState) -> ResearchState:
    """수집된 데이터를 분석해 인사이트를 추출합니다."""
    prompt = f"다음 데이터를 분석하고 핵심 인사이트를 추출하세요:\n{state['raw_data']}"
    analysis = llm.invoke(prompt).content
    return {"analysis": analysis}
 
def report_agent(state: ResearchState) -> ResearchState:
    """분석 결과를 기반으로 임원 보고서를 생성합니다."""
    prompt = f"다음 분석을 바탕으로 임원용 보고서를 작성하세요:\n{state['analysis']}"
    report = llm.invoke(prompt).content
    return {"report": report}
 
# 3. 그래프 구성 및 컴파일
workflow = StateGraph(ResearchState)
workflow.add_node("research", research_agent)
workflow.add_node("analyze", analysis_agent)
workflow.add_node("report", report_agent)
 
workflow.set_entry_point("research")
workflow.add_edge("research", "analyze")
workflow.add_edge("analyze", "report")
workflow.add_edge("report", END)
 
graph = workflow.compile()
 
# 4. 실행
result = graph.invoke({"query": "2025년 멀티에이전트 AI 트렌드"})
print(result["report"])

import os
from crewai import Agent, Crew, Process, Task
from crewai_tools import CodeInterpreterTool
from langchain_anthropic import ChatAnthropic
 
# 환경 변수 설정
# export ANTHROPIC_API_KEY="your-anthropic-api-key"
 
code_interpreter = CodeInterpreterTool()
 
# 1. 에이전트 정의 — 각 역할에 특화된 목표와 배경 지식을 부여합니다
coder = Agent(
    role="시니어 Python 개발자",
    goal="명확하고 효율적인 Python 코드를 작성합니다",
    backstory="10년 경력의 백엔드 엔지니어로, PEP 8과 SOLID 원칙을 철저히 따릅니다",
    tools=[code_interpreter],
    verbose=True,
)
 
reviewer = Agent(
    role="코드 리뷰어",
    goal="코드의 품질, 보안 취약점, 성능 문제를 검토합니다",
    backstory="보안 전문가 출신으로 OWASP Top 10을 숙지하고 있습니다",
    verbose=True,
)
 
tester = Agent(
    role="QA 엔지니어",
    goal="엣지 케이스를 포함한 포괄적인 테스트 케이스를 작성합니다",
    backstory="TDD 방법론을 따르며 pytest를 주로 사용합니다",
    tools=[code_interpreter],
    verbose=True,
)
 
# 2. 태스크 정의 — 각 에이전트가 수행할 구체적인 작업을 명시합니다
code_task = Task(
    description="사용자 인증 모듈을 Python으로 구현하세요. JWT 토큰 발급 및 검증 기능을 포함합니다.",
    expected_output="완성된 Python 코드와 간략한 구현 설명",
    agent=coder,
)
 
review_task = Task(
    description="작성된 코드를 검토하고 보안 취약점, 코드 스타일, 성능 개선점을 리포트하세요.",
    expected_output="검토 보고서 (취약점 목록, 권고 사항 포함)",
    agent=reviewer,
    context=[code_task],  # 이전 태스크 결과를 컨텍스트로 활용합니다
)
 
test_task = Task(
    description="인증 모듈에 대한 pytest 테스트 케이스를 작성하세요. 정상 케이스와 엣지 케이스를 모두 포함합니다.",
    expected_output="pytest 테스트 파일 (커버리지 80% 이상 목표)",
    agent=tester,
    context=[code_task, review_task],
)
 
# 3. 계층형 Crew 구성
# manager_llm은 문자열이 아닌 LLM 객체를 전달해야 합니다
manager_llm = ChatAnthropic(model="claude-opus-4-6")
 
crew = Crew(
    agents=[coder, reviewer, tester],
    tasks=[code_task, review_task, test_task],
    process=Process.hierarchical,
    manager_llm=manager_llm,  # 복잡한 추론이 필요한 오케스트레이터에는 강력한 모델을 사용하는 것을 권장합니다
    verbose=True,
)
 
result = crew.kickoff()
print(result.raw)

import os
from langgraph.graph import StateGraph, END
from typing import TypedDict
from crewai import Agent, Crew, Process, Task
from langchain_anthropic import ChatAnthropic
 
# 환경 변수 설정
# export ANTHROPIC_API_KEY="your-anthropic-api-key"
 
# 전체 파이프라인 상태 스키마
class PipelineState(TypedDict):
    topic: str
    research_result: str
    final_code: str
 
# --- 리서치 크루: 에이전트 및 태스크 정의 ---
researcher = Agent(
    role="리서치 전문가",
    goal="주어진 주제에 대한 심층 기술 조사를 수행합니다",
    backstory="기술 트렌드를 분석하는 5년 경력의 리서처입니다",
    verbose=True,
)
 
analyst = Agent(
    role="데이터 분석가",
    goal="수집된 정보를 구조화하고 핵심 인사이트를 추출합니다",
    backstory="데이터 기반 의사결정을 지원하는 분석 전문가입니다",
    verbose=True,
)
 
research_task = Task(
    description="{topic}에 대한 최신 기술 동향과 구현 사례를 조사하세요.",
    expected_output="기술 동향 요약 보고서",
    agent=researcher,
)
 
analysis_task = Task(
    description="조사된 내용을 바탕으로 구현에 필요한 핵심 기술 스택을 분석하세요.",
    expected_output="기술 스택 분석 및 권고안",
    agent=analyst,
    context=[research_task],
)
 
# --- 코딩 크루: 에이전트 및 태스크 정의 ---
coder = Agent(
    role="시니어 Python 개발자",
    goal="명확하고 효율적인 Python 코드를 작성합니다",
    backstory="10년 경력의 백엔드 엔지니어로, SOLID 원칙을 철저히 따릅니다",
    verbose=True,
)
 
reviewer = Agent(
    role="코드 리뷰어",
    goal="코드 품질, 보안 취약점, 성능 문제를 검토합니다",
    backstory="보안 전문가 출신으로 OWASP Top 10을 숙지하고 있습니다",
    verbose=True,
)
 
tester = Agent(
    role="QA 엔지니어",
    goal="포괄적인 테스트 케이스를 작성합니다",
    backstory="TDD 방법론을 따르며 pytest를 주로 사용합니다",
    verbose=True,
)
 
code_task = Task(
    description="{research} 결과를 바탕으로 데이터 파이프라인 모듈을 Python으로 구현하세요.",
    expected_output="완성된 Python 구현 코드",
    agent=coder,
)
 
review_task = Task(
    description="작성된 코드를 검토하고 보안 취약점과 성능 개선점을 리포트하세요.",
    expected_output="코드 리뷰 보고서",
    agent=reviewer,
    context=[code_task],
)
 
test_task = Task(
    description="구현된 모듈에 대한 pytest 테스트 케이스를 작성하세요.",
    expected_output="pytest 테스트 파일",
    agent=tester,
    context=[code_task, review_task],
)
 
manager_llm = ChatAnthropic(model="claude-opus-4-6")
 
# CrewAI 크루를 LangGraph 노드로 래핑합니다
def run_research_crew(state: PipelineState) -> PipelineState:
    """LangGraph 노드: CrewAI 리서치 크루를 실행합니다."""
    research_crew = Crew(
        agents=[researcher, analyst],
        tasks=[research_task, analysis_task],
        process=Process.sequential,
    )
    result = research_crew.kickoff(inputs={"topic": state["topic"]})
    return {"research_result": result.raw}
 
def run_coding_crew(state: PipelineState) -> PipelineState:
    """LangGraph 노드: CrewAI 코딩 크루를 실행합니다."""
    coding_crew = Crew(
        agents=[coder, reviewer, tester],
        tasks=[code_task, review_task, test_task],
        process=Process.hierarchical,
        manager_llm=manager_llm,
    )
    result = coding_crew.kickoff(inputs={"research": state["research_result"]})
    return {"final_code": result.raw}
 
# LangGraph로 두 크루를 조율합니다
workflow = StateGraph(PipelineState)
workflow.add_node("research_crew", run_research_crew)
workflow.add_node("coding_crew", run_coding_crew)
workflow.set_entry_point("research_crew")
workflow.add_edge("research_crew", "coding_crew")
workflow.add_edge("coding_crew", END)
 
pipeline = workflow.compile()
result = pipeline.invoke({"topic": "실시간 데이터 스트리밍 아키텍처"})
print(result["final_code"])