LangGraph vs CrewAI: 에이전트가 3개를 넘어서면 상태 머신 오케스트레이션으로 대응하라

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
import operator
 
# 1. 공유 상태 스키마 — 모든 에이전트가 이 타입을 공유한다
class ReviewState(TypedDict):
    code: str
    review_result: str
    test_result: str
    security_result: str
    errors: Annotated[list[str], operator.add]  # reducer: 덮어쓰기 대신 누적
    approved: bool
    iteration: int
 
# 2. 에이전트 노드 — 순수 함수, 상태 diff만 반환
def code_review_agent(state: ReviewState) -> dict:
    # 실제 구현: llm.invoke(f"다음 코드를 리뷰하세요: {state['code']}")
    return {
        "review_result": "구조 개선 필요: 의존성 주입 패턴 미적용",
        "iteration": state["iteration"] + 1
    }
 
def test_agent(state: ReviewState) -> dict:
    return {"test_result": "커버리지 78% — 경계값 테스트 누락"}
 
def security_scan_agent(state: ReviewState) -> dict:
    return {"security_result": "SQL Injection 취약점 감지 (line 42)"}
 
# 3. 조건부 엣지 함수 — 다음 노드를 결정
def approval_gate(
    state: ReviewState,
) -> Literal["approve", "reject", "reject_final"]:
    has_vulnerability = "취약점" in state["security_result"]
    iteration_limit_reached = state["iteration"] >= 3
 
    if has_vulnerability and iteration_limit_reached:
        # 최대 반복 도달 시 취약점 있는 코드를 승인하지 않고 실패 종료
        # 프로덕션에서는 알림 발송 + 수동 검토 요청으로 연결
        return "reject_final"
    if has_vulnerability:
        return "reject"
    return "approve"
 
# 4. 그래프 조립
builder = StateGraph(ReviewState)
builder.add_node("code_review", code_review_agent)
builder.add_node("test", test_agent)
builder.add_node("security_scan", security_scan_agent)
 
builder.set_entry_point("code_review")
builder.add_edge("code_review", "test")
builder.add_edge("test", "security_scan")
builder.add_conditional_edges(
    "security_scan",
    approval_gate,
    {
        "approve": END,
        "reject": "code_review",    # 실패 시 첫 단계로 루프백
        "reject_final": END          # 반복 초과 시 실패 종료 (취약점 있는 상태로 승인 금지)
    }
)
 
# 5. 체크포인트 — 중간 실패 시 해당 노드에서 재시작 가능
# Human-in-the-loop 활성화 예시:
# graph = builder.compile(checkpointer=checkpointer, interrupt_before=["security_scan"])
# → 보안 스캔 전에 실행이 멈추고, 사람이 직접 검토 후 graph.invoke()로 재개 가능
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)
 
# 실행
config = {"configurable": {"thread_id": "review-001"}}
result = graph.invoke(
    {"code": "...", "errors": [], "iteration": 0, "approved": False},
    config=config
)

from crewai import Agent, Task, Crew, Process
 
# 1. 역할 기반 에이전트 정의
researcher = Agent(
    role="기술 리서처",
    goal="멀티에이전트 오케스트레이션 최신 트렌드와 실제 사례를 수집한다",
    backstory=(
        "10년 경력의 기술 저널리스트. arXiv, GitHub, 공식 문서만 신뢰하며, "
        "마케팅 블로그 내용을 그대로 인용하지 않는다."
    ),
    allow_delegation=False,   # 켜두면 에이전트가 작업을 재위임해 예상치 못한 태스크 체인이 생긴다
    verbose=True
)
 
fact_checker = Agent(
    role="팩트 체커",
    goal="리서치 결과의 정확성을 검증하고 수치·인용 오류를 제거한다",
    backstory="전직 학술 연구원. 출처 없는 주장은 모두 의심하고 검증을 요구한다.",
    allow_delegation=False
)
 
writer = Agent(
    role="기술 작가",
    goal="검증된 정보를 개발자 친화적인 마크다운 블로그 글로 변환한다",
    backstory="오픈소스 커뮤니티에서 활동하는 5년 경력의 기술 블로거.",
    allow_delegation=False
)
 
editor = Agent(
    role="편집자",
    goal="초안의 가독성, 논리 흐름, 문체를 개선해 최종 출판 품질로 다듬는다",
    backstory="기술 미디어 10년 경력의 시니어 에디터. 독자 친화성을 최우선으로 본다.",
    allow_delegation=False
)
 
# 2. 태스크 정의 — context가 핸드오프를 처리한다
research_task = Task(
    description="LangGraph와 CrewAI의 멀티에이전트 오케스트레이션 패턴 조사",
    expected_output="핵심 개념, 실제 사례, 장단점, 참고 자료 목록",
    agent=researcher
)
 
fact_check_task = Task(
    description="리서치 결과의 수치와 사실 관계 검증. 오류 발견 시 수정사항 명시",
    expected_output="검증된 사실 목록과 수정된 오류 내역",
    agent=fact_checker,
    context=[research_task]      # 리서처 결과를 컨텍스트로 자동 주입
)
 
write_task = Task(
    description="검증된 정보로 2000자 이상의 기술 블로그 초안 작성 (마크다운)",
    expected_output="제목, 도입, 핵심 개념, 실전 예시, 결론을 포함한 마크다운 글",
    agent=writer,
    context=[research_task, fact_check_task]   # 두 태스크 결과 모두 주입
)
 
edit_task = Task(
    description="작성된 초안을 검토하여 가독성, 논리 흐름, 표현을 최종 수준으로 개선",
    expected_output="편집 완료된 최종 마크다운 글",
    agent=editor,
    context=[write_task]
)
 
# 3. Crew 조립 및 실행
crew = Crew(
    agents=[researcher, fact_checker, writer, editor],
    tasks=[research_task, fact_check_task, write_task, edit_task],
    process=Process.sequential,  # 순차 실행
    verbose=True
)
 
result = crew.kickoff()
print(result.raw)

def triage_agent(state: dict) -> dict:
    intent = classify_intent(state["query"])  # "payment" | "shipping" | "general"
    return {"intent": intent}
 
def route_by_intent(state: dict) -> str:
    return state["intent"]   # 반환값이 다음 노드 이름
 
builder.add_node("triage", triage_agent)
builder.add_node("payment_agent", payment_agent)
builder.add_node("shipping_agent", shipping_agent)
builder.add_node("general_agent", general_agent)
 
builder.set_entry_point("triage")
builder.add_conditional_edges(
    "triage",
    route_by_intent,
    {
        "payment": "payment_agent",
        "shipping": "shipping_agent",
        "general": "general_agent"
    }
)

from crewai import Agent, Task, Crew, Process
 
payment_agent = Agent(
    role="결제 전문가",
    goal="결제 관련 고객 문의를 처리한다",
    backstory="결제 시스템 5년 경력. 환불, 결제 오류, 카드 문제를 전문으로 한다.",
    allow_delegation=False
)
 
shipping_agent = Agent(
    role="배송 전문가",
    goal="배송 추적과 배송 관련 문의를 처리한다",
    backstory="물류 시스템 전문가. 배송 지연, 분실, 주소 변경을 담당한다.",
    allow_delegation=False
)
 
manager = Agent(
    role="고객 서비스 매니저",
    goal="고객 문의를 분석하고 적절한 전문가에게 위임한다",
    backstory="고객 서비스 총괄. 모든 문의를 접수하고 전문팀에 배분한다.",
)
 
handle_query_task = Task(
    description="고객 문의를 처리하고 완전한 답변을 제공한다",
    expected_output="고객 문의에 대한 완전하고 정확한 답변",
    agent=manager
)
 
crew = Crew(
    agents=[payment_agent, shipping_agent],
    tasks=[handle_query_task],
    process=Process.hierarchical,   # Manager LLM이 자동으로 적절한 에이전트에 위임
    manager_agent=manager,
    verbose=True
)