LangGraph vs CrewAI vs AutoGen — 2026년 AI 에이전트 프레임워크, 실전에서 어떤 걸 선택해야 할까

import os
from langgraph.graph import StateGraph, END
from typing import TypedDict
 
class ResearchState(TypedDict):
    query: str
    search_results: list[str]
    analysis: str
    final_report: str
 
def search_node(state: ResearchState) -> ResearchState:
    results = web_search(state["query"])
    return {"search_results": results}
 
def analyze_node(state: ResearchState) -> ResearchState:
    analysis = llm_analyze(state["search_results"])
    return {"analysis": analysis}
 
def report_node(state: ResearchState) -> ResearchState:
    report = llm_generate_report(state["analysis"])
    return {"final_report": report}
 
def should_retry(state: ResearchState) -> str:
    # 분석이 불충분하면 재검색, 충분하면 report 노드로
    if len(state["analysis"]) < 100:
        return "search"
    return "report"
 
workflow = StateGraph(ResearchState)
workflow.add_node("search", search_node)
workflow.add_node("analyze", analyze_node)
workflow.add_node("report", report_node)
 
workflow.set_entry_point("search")
workflow.add_edge("search", "analyze")
# should_retry가 "search" 또는 "report" 문자열을 반환 → 해당 노드 이름으로 전환
workflow.add_conditional_edges(
    "analyze",
    should_retry,
    {"search": "search", "report": "report"}  # 반환값 → 노드 이름 매핑
)
workflow.add_edge("report", END)
 
app = workflow.compile(checkpointer=memory_saver)

import os
from crewai import Agent, Task, Crew, Process
 
researcher = Agent(
    role="Senior Research Analyst",
    goal="Find accurate and up-to-date information on given topics",
    backstory="You are an expert researcher with 10 years of experience...",
    tools=[web_search_tool, scraping_tool],
    verbose=True
)
 
writer = Agent(
    role="Content Writer",
    goal="Write engaging technical blog posts",
    backstory="You are a skilled writer who transforms research into compelling content...",
    tools=[file_write_tool]
)
 
research_task = Task(
    description="Research the latest trends in {topic}",
    expected_output="A comprehensive summary with key findings",
    agent=researcher
)
 
writing_task = Task(
    description="Write a blog post based on the research",
    expected_output="A 1000-word blog post in markdown format",
    agent=writer,
    context=[research_task]  # 이전 태스크 결과를 자동으로 컨텍스트에 포함
)
 
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential
)
 
result = crew.kickoff(inputs={"topic": "멀티에이전트 프레임워크"})
print(result.raw)  # 최종 결과 출력

import os
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
 
llm_config = {
    "config_list": [{
        "model": "gpt-4o",
        "api_key": os.getenv("OPENAI_API_KEY")
    }]
}
 
coder = AssistantAgent(
    name="Coder",
    system_message="You are a senior software engineer. Write clean, efficient code.",
    llm_config=llm_config
)
 
reviewer = AssistantAgent(
    name="CodeReviewer",
    system_message="You are a code reviewer. Find bugs and suggest improvements.",
    llm_config=llm_config
)
 
user_proxy = UserProxyAgent(
    name="UserProxy",
    human_input_mode="NEVER",
    # 주의: work_dir 설정 시 로컬 파일 시스템에 실제 파일이 생성됩니다
    code_execution_config={"work_dir": "coding"}
)
 
groupchat = GroupChat(
    agents=[user_proxy, coder, reviewer],
    messages=[],
    max_round=10
)
 
manager = GroupChatManager(groupchat=groupchat, llm_config=llm_config)
 
user_proxy.initiate_chat(
    manager,
    message="Python으로 퀵소트 알고리즘을 구현하고 코드 리뷰까지 완료해줘"
)

import os
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.redis import RedisSaver
from typing import TypedDict
 
class CreditRiskState(TypedDict):
    customer_id: str
    financial_data: dict
    risk_score: float
    risk_level: str  # "LOW", "MEDIUM", "HIGH"
    human_review_required: bool
    human_decision: str | None
    final_decision: str
 
def assess_risk(state: CreditRiskState) -> CreditRiskState:
    score = calculate_risk_score(state["financial_data"])
    level = "HIGH" if score > 0.7 else "MEDIUM" if score > 0.4 else "LOW"
    return {
        "risk_score": score,
        "risk_level": level,
        "human_review_required": level == "HIGH"
    }
 
def route_by_risk(state: CreditRiskState) -> str:
    if state["human_review_required"]:
        return "human_review"
    return "auto_decision"
 
def auto_decision(state: CreditRiskState) -> CreditRiskState:
    decision = "APPROVE" if state["risk_level"] == "LOW" else "REJECT"
    return {"final_decision": decision}
 
def await_human_review(state: CreditRiskState) -> CreditRiskState:
    # interrupt_before 설정으로 이 노드 직전에 실행이 멈추고 사람 입력을 기다림
    return {"final_decision": state.get("human_decision", "PENDING")}
 
memory = RedisSaver.from_conn_string("redis://localhost:6379")
 
workflow = StateGraph(CreditRiskState)
workflow.add_node("assess", assess_risk)
workflow.add_node("auto_decision", auto_decision)
workflow.add_node("human_review", await_human_review)
 
workflow.set_entry_point("assess")
workflow.add_conditional_edges(
    "assess",
    route_by_risk,
    {"human_review": "human_review", "auto_decision": "auto_decision"}
)
workflow.add_edge("auto_decision", END)
workflow.add_edge("human_review", END)
 
app = workflow.compile(checkpointer=memory, interrupt_before=["human_review"])

import os
from crewai import Agent, Task, Crew, Process
from crewai_tools import CSVSearchTool, WebsiteSearchTool
 
csv_tool = CSVSearchTool(csv="leads.csv")
web_tool = WebsiteSearchTool()
 
data_validator = Agent(
    role="Data Quality Specialist",
    goal="Validate and clean CRM lead data for accuracy and completeness",
    backstory="""You specialize in B2B sales data quality. 
    You know common data issues like duplicate entries, 
    missing fields, and inconsistent formatting.""",
    tools=[csv_tool],
    llm="gpt-4o"
)
 
enrichment_agent = Agent(
    role="Lead Intelligence Analyst",
    goal="Enrich lead profiles with current company information",
    backstory="""You research companies and contacts to add valuable 
    context to sales leads, including recent news and funding rounds.""",
    tools=[web_tool],
    llm="gpt-4o"
)
 
scoring_agent = Agent(
    role="Sales Prioritization Expert",
    # ICP fit: 이상적 고객 프로파일(Ideal Customer Profile) 부합도
    # buying signals: 구매 의향을 나타내는 행동 지표 (최근 채용, 자금 조달 등)
    goal="Score and prioritize leads based on ICP fit and buying signals",
    backstory="""You analyze enriched lead data and assign priority scores 
    based on ideal customer profile criteria and engagement signals.""",
    llm="gpt-4o"
)
 
validation_task = Task(
    description="Analyze leads.csv and identify data quality issues. Flag duplicates and missing required fields.",
    expected_output="JSON report with quality issues and cleaned dataset",
    agent=data_validator
)
 
enrichment_task = Task(
    # firmographic data: 기업 규모, 업종, 소재지 등 기업 특성 정보
    description="For each validated lead, research current company info and add firmographic data.",
    expected_output="Enriched lead dataset with company size, funding, recent news",
    agent=enrichment_agent,
    context=[validation_task]
)
 
scoring_task = Task(
    description="Score leads 1-100 based on ICP fit. Output prioritized list with reasoning.",
    expected_output="CSV with lead scores and priority tier (HOT/WARM/COLD)",
    agent=scoring_agent,
    context=[enrichment_task],
    output_file="prioritized_leads.csv"
)
 
crew = Crew(
    agents=[data_validator, enrichment_agent, scoring_agent],
    tasks=[validation_task, enrichment_task, scoring_task],
    process=Process.sequential,
    verbose=True
)
 
result = crew.kickoff()
print(result.raw)  # CrewOutput 객체의 .raw로 최종 텍스트 결과 접근

import os
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
 
llm_config = {
    "config_list": [{
        "model": "gpt-4o",
        "api_key": os.getenv("OPENAI_API_KEY")
    }]
}
 
security_reviewer = AssistantAgent(
    name="SecurityExpert",
    system_message="""You are a security expert. Review code for:
    - SQL injection, XSS, authentication vulnerabilities
    - Insecure dependencies
    Always start your response with 'SECURITY REVIEW:'""",
    llm_config=llm_config
)
 
performance_reviewer = AssistantAgent(
    name="PerformanceExpert",
    system_message="""You are a performance optimization expert. Review for:
    - N+1 queries, memory leaks, inefficient algorithms
    - Caching opportunities
    Always start with 'PERFORMANCE REVIEW:'""",
    llm_config=llm_config
)
 
architect = AssistantAgent(
    name="SoftwareArchitect",
    system_message="""You synthesize all reviews and provide final recommendations.
    Prioritize issues by severity and provide actionable improvements.
    Always start with 'ARCHITECTURE SUMMARY:'""",
    llm_config=llm_config
)
 
user_proxy = UserProxyAgent(
    name="Developer",
    human_input_mode="TERMINATE",
    code_execution_config=False,
    is_termination_msg=lambda x: "LGTM" in x.get("content", "")
)
 
groupchat = GroupChat(
    agents=[user_proxy, security_reviewer, performance_reviewer, architect],
    messages=[],
    max_round=8,
    speaker_selection_method="round_robin"
)
 
manager = GroupChatManager(groupchat=groupchat, llm_config=llm_config)
 
code_to_review = """
def get_user(user_id):
    query = f"SELECT * FROM users WHERE id = {user_id}"  # 위험한 코드
    return db.execute(query)
"""
 
user_proxy.initiate_chat(
    manager,
    message=f"다음 코드를 리뷰해주세요:\n```python\n{code_to_review}\n```"
)