혼자 일하는 AI는 이제 옛날 — LangGraph로 멀티 에이전트 시스템 만들기 [2026 실전 튜토리얼]

[단일 AI]
사용자 → AI (혼자 다 함) → 결과
           └ 조사도, 작성도, 검증도 전부 혼자
[멀티 에이전트]
사용자 → 슈퍼바이저 (팀장)
              ├→ 리서처 에이전트 (조사 전문)
              ├→ 라이터 에이전트  (작성 전문)
              └→ 팩트체커 에이전트 (검증 전문)
                        ↓
                    최종 결과물

[사용자 입력: "Python 2026 트렌드"]
        ↓
[슈퍼바이저 에이전트] ← 팀장. 전체 흐름 결정
        ↓
[리서처 에이전트]     ← 웹 검색, 정보 수집
        ↓
[라이터 에이전트]     ← 수집된 정보로 블로그 초안 작성
        ↓
[팩트체커 에이전트]   ← 주요 주장을 검증, 수정 요청 or 승인
        ↓
    [최종 결과물] ← 검증된 블로그 초안

pip install langgraph langchain langchain-anthropic \
            langchain-community python-dotenv

ANTHROPIC_API_KEY=your-api-key-here

multi-agent/
├── .env
├── main.py          # 실행 진입점
├── state.py         # 공유 상태 정의
├── agents.py        # 에이전트 3개
├── supervisor.py    # 슈퍼바이저 로직
└── graph.py         # 그래프 조립

# 팀 공유 메모장
state = {
    "topic": "Python 2026 트렌드",     # 입력
    "research": "",                     # 리서처가 채움
    "draft": "",                        # 라이터가 채움
    "fact_check": "",                   # 팩트체커가 채움
    "approved": False,                  # 최종 승인 여부
    "iteration": 0                      # 수정 횟수
}

def researcher_node(state):
    # state에서 topic 읽기
    # 리서치 수행
    # state의 research 필드 업데이트
    return {"research": "...리서치 결과..."}

# 팩트체크 결과에 따라 분기
def should_continue(state):
    if state["approved"]:
        return "END"          # 승인됐으면 종료
    elif state["iteration"] >= 3:
        return "END"          # 3번 이상 수정됐으면 그냥 종료
    else:
        return "writer"       # 승인 안 됐으면 라이터에게 다시

# state.py
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
class ResearchState(TypedDict):
    """팀 전체가 공유하는 상태"""
    topic: str              # 리서치 주제
    research_notes: str     # 리서처의 조사 결과
    draft: str              # 라이터의 초안
    fact_check_result: str  # 팩트체커의 검토 의견
    approved: bool          # 최종 승인 여부
    revision_notes: str     # 수정 요청 사항
    iteration: int          # 현재 수정 횟수 (무한루프 방지)
    final_output: str       # 최종 결과물

# agents.py
import os
from dotenv import load_dotenv
from langchain_anthropic import ChatAnthropic
from langchain.prompts import ChatPromptTemplate
from state import ResearchState
load_dotenv()
# 공통 LLM (모든 에이전트가 같은 모델 사용, 역할은 프롬프트로 구분)
llm = ChatAnthropic(
    model="claude-sonnet-4-20250514",
    api_key=os.getenv("ANTHROPIC_API_KEY"),
    temperature=0.3,
    max_tokens=2000
)
# ────────────────────────────────────────
# 에이전트 1: 리서처
# ────────────────────────────────────────
def researcher_agent(state: ResearchState) -> dict:
    """
    주제를 받아 핵심 정보를 조사합니다.
    실제 프로덕션에서는 Tavily, SerpAPI 등 검색 툴을 연결하세요.
    여기서는 LLM의 사전 지식을 활용합니다.
    """
    print("🔍 [리서처] 조사 시작...")
    
    prompt = ChatPromptTemplate.from_template("""
당신은 전문 리서처입니다. 주어진 주제에 대해 핵심 정보를 조사하세요.
주제: {topic}
다음 형식으로 조사 결과를 작성하세요:
## 핵심 사실 (5~7가지)
- 사실 1: ...
- 사실 2: ...
## 주요 통계 & 데이터
- ...
## 최신 트렌드
- ...
## 참고할 만한 사례
- ...
정확하고 구체적인 정보만 포함하세요. 불확실한 정보는 "(추정)" 표시를 붙이세요.
""")
    
    chain = prompt | llm
    result = chain.invoke({"topic": state["topic"]})
    
    print(f"✅ [리서처] 조사 완료 — {len(result.content)}자")
    return {"research_notes": result.content}
# ────────────────────────────────────────
# 에이전트 2: 라이터
# ────────────────────────────────────────
def writer_agent(state: ResearchState) -> dict:
    """
    리서처의 조사 결과를 바탕으로 블로그 초안을 작성합니다.
    팩트체커의 수정 요청이 있으면 그것도 반영합니다.
    """
    print("✍️ [라이터] 초안 작성 시작...")
    
    # 첫 번째 작성인지, 수정 요청이 온 건지 구분
    revision_context = ""
    if state.get("revision_notes"):
        revision_context = f"""
⚠️ 팩트체커의 수정 요청:
{state['revision_notes']}
위 피드백을 반드시 반영해서 수정해주세요.
"""
    
    prompt = ChatPromptTemplate.from_template("""
당신은 기술 블로그 전문 작가입니다. 
리서처의 조사 자료를 바탕으로 독자가 읽기 쉬운 블로그 글을 작성하세요.
주제: {topic}
리서처 조사 자료:
{research_notes}
{revision_context}
작성 기준:
- 제목 포함
- 소개 → 본문 (3~4 섹션) → 마무리 구조
- 전문 용어는 쉽게 설명
- 구체적인 수치와 예시 활용
- 약 800~1000자 분량
""")
    
    chain = prompt | llm
    result = chain.invoke({
        "topic": state["topic"],
        "research_notes": state["research_notes"],
        "revision_context": revision_context
    })
    
    print(f"✅ [라이터] 초안 완성 — {len(result.content)}자")
    return {
        "draft": result.content,
        "revision_notes": ""  # 수정 후 노트 초기화
    }
# ────────────────────────────────────────
# 에이전트 3: 팩트체커
# ────────────────────────────────────────
def fact_checker_agent(state: ResearchState) -> dict:
    """
    라이터의 초안을 검토합니다.
    조사 자료와 대조해 사실 오류나 과장을 찾아냅니다.
    통과 or 수정 요청 둘 중 하나를 결정합니다.
    """
    print("🔎 [팩트체커] 검토 시작...")
    
    prompt = ChatPromptTemplate.from_template("""
당신은 엄격한 팩트체커입니다. 블로그 초안을 검토하세요.
원본 리서치 자료:
{research_notes}
검토할 초안:
{draft}
다음 기준으로 검토하세요:
1. 리서치 자료에 없는 내용을 지어냈는가?
2. 통계/수치가 정확한가?
3. 과장된 표현이 있는가?
4. 논리적 모순이 있는가?
반드시 아래 형식으로만 응답하세요:
판정: [APPROVED 또는 NEEDS_REVISION]
검토 의견:
(승인 시: 품질이 양호한 이유 1~2줄)
(수정 요청 시: 구체적인 수정 항목과 이유)
""")
    
    chain = prompt | llm
    result = chain.invoke({
        "research_notes": state["research_notes"],
        "draft": state["draft"]
    })
    
    content = result.content
    approved = "APPROVED" in content.upper()
    
    if approved:
        print("✅ [팩트체커] 승인!")
        return {
            "fact_check_result": content,
            "approved": True,
            "final_output": state["draft"],  # 승인된 초안이 최종 결과물
            "iteration": state.get("iteration", 0) + 1
        }
    else:
        print(f"⚠️ [팩트체커] 수정 요청 (시도 {state.get('iteration', 0) + 1}회)")
        return {
            "fact_check_result": content,
            "approved": False,
            "revision_notes": content,
            "iteration": state.get("iteration", 0) + 1
        }

# supervisor.py
from state import ResearchState
def should_continue(state: ResearchState) -> str:
    """
    팩트체크 결과를 보고 다음 단계를 결정하는 팀장 로직.
    
    반환값:
    - "writer"  → 라이터에게 수정 요청
    - "end"     → 작업 완료
    """
    MAX_ITERATIONS = 3  # 무한루프 방지: 최대 3번까지만 수정
    
    current_iter = state.get("iteration", 0)
    
    if state.get("approved", False):
        print(f"\n🎉 [슈퍼바이저] 최종 승인! (총 {current_iter}회 시도)")
        return "end"
    
    if current_iter >= MAX_ITERATIONS:
        print(f"\n⏱️ [슈퍼바이저] 최대 수정 횟수 도달. 마지막 버전으로 마무리.")
        return "end"
    
    print(f"\n🔄 [슈퍼바이저] 수정 요청 → 라이터에게 전달 (시도 {current_iter + 1}/{MAX_ITERATIONS})")
    return "writer"

# graph.py
from langgraph.graph import StateGraph, END
from state import ResearchState
from agents import researcher_agent, writer_agent, fact_checker_agent
from supervisor import should_continue
def build_research_graph():
    """
    3인조 에이전트 팀을 그래프로 조립합니다.
    
    흐름:
    researcher → writer → fact_checker
                   ↑            ↓
                   └── (수정 요청 시 루프) ──┘
                                ↓ (승인 or 최대 횟수)
                               END
    """
    
    # 1. 그래프 초기화
    workflow = StateGraph(ResearchState)
    
    # 2. 노드 등록 (에이전트 연결)
    workflow.add_node("researcher", researcher_agent)
    workflow.add_node("writer", writer_agent)
    workflow.add_node("fact_checker", fact_checker_agent)
    
    # 3. 시작점 설정
    workflow.set_entry_point("researcher")
    
    # 4. 일반 엣지 (순서대로 진행)
    workflow.add_edge("researcher", "writer")   # 리서처 → 라이터
    workflow.add_edge("writer", "fact_checker") # 라이터 → 팩트체커
    
    # 5. 조건부 엣지 (팩트체커 결과에 따라 분기)
    workflow.add_conditional_edges(
        "fact_checker",     # 이 노드의 출력을 기준으로
        should_continue,    # 이 함수가 다음 노드를 결정
        {
            "writer": "writer",  # "writer" 반환 시 → 라이터로
            "end": END           # "end" 반환 시 → 종료
        }
    )
    
    # 6. 컴파일 (실행 가능한 앱으로 변환)
    return workflow.compile()
# 그래프 시각화 (선택사항, graphviz 설치 필요)
# pip install grandalf
def print_graph_structure(graph):
    try:
        print(graph.get_graph().draw_ascii())
    except Exception:
        print("그래프 시각화 라이브러리가 없습니다. pip install grandalf")

# main.py
from dotenv import load_dotenv
from graph import build_research_graph, print_graph_structure
load_dotenv()
def run_research_team(topic: str) -> dict:
    """
    3인조 에이전트 팀을 실행합니다.
    """
    print(f"\n{'='*60}")
    print(f"🚀 리서치 팀 가동 — 주제: '{topic}'")
    print(f"{'='*60}\n")
    
    # 그래프 생성
    app = build_research_graph()
    
    # 초기 상태
    initial_state = {
        "topic": topic,
        "research_notes": "",
        "draft": "",
        "fact_check_result": "",
        "approved": False,
        "revision_notes": "",
        "iteration": 0,
        "final_output": ""
    }
    
    # 실행
    result = app.invoke(initial_state)
    
    print(f"\n{'='*60}")
    print("📄 최종 결과물")
    print(f"{'='*60}")
    
    final = result.get("final_output") or result.get("draft", "결과물 없음")
    print(final)
    
    print(f"\n{'='*60}")
    print(f"📊 실행 요약")
    print(f"  - 총 수정 횟수: {result.get('iteration', 0)}회")
    print(f"  - 최종 승인: {'✅ 승인' if result.get('approved') else '⏱️ 시간 초과'}")
    print(f"{'='*60}\n")
    
    return result
if __name__ == "__main__":
    # 테스트 실행
    result = run_research_team("Python과 AI 개발 2026년 트렌드")

============================================================
🚀 리서치 팀 가동 — 주제: 'Python과 AI 개발 2026년 트렌드'
============================================================
🔍 [리서처] 조사 시작...
✅ [리서처] 조사 완료 — 1243자
✍️ [라이터] 초안 작성 시작...
✅ [라이터] 초안 완성 — 987자
🔎 [팩트체커] 검토 시작...
⚠️ [팩트체커] 수정 요청 (시도 1회)
🔄 [슈퍼바이저] 수정 요청 → 라이터에게 전달 (시도 1/3)
✍️ [라이터] 초안 작성 시작...
✅ [라이터] 초안 완성 — 1054자
🔎 [팩트체커] 검토 시작...
✅ [팩트체커] 승인!
🎉 [슈퍼바이저] 최종 승인! (총 2회 시도)
============================================================

# 리서처가 여러 소스를 동시에 조사
from langgraph.graph import StateGraph
# 여러 리서처를 병렬로 실행하고 결과를 합치는 패턴
workflow.add_node("researcher_news", news_researcher)
workflow.add_node("researcher_data", data_researcher)
workflow.add_node("aggregator", combine_results)
# 두 리서처가 동시에 실행되고 aggregator에서 합쳐짐
workflow.add_edge("researcher_news", "aggregator")
workflow.add_edge("researcher_data", "aggregator")

# 민감한 결정 전에 사람의 승인을 기다리는 패턴
from langgraph.types import interrupt
def sensitive_action_node(state):
    # 실행 전 사람에게 확인 요청
    approval = interrupt({
        "message": "이 내용을 발행하시겠습니까?",
        "draft": state["draft"]
    })
    
    if approval.get("approved"):
        return {"final_output": state["draft"]}
    else:
        return {"revision_notes": approval.get("feedback", "")}