LangGraph 완전 정복 — Bloom AI 9시간 마스터클래스 완전 분석

pip install langgraph langchain google-generativeai langchain-google-genai

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # 메시지 누적
    intent: str                               # 분류된 의도
    manual_result: str                        # 매뉴얼 검색 결과
    need_escalation: bool                     # 에스컬레이션 필요 여부

from langchain_google_genai import ChatGoogleGenerativeAI

llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash")

def classify_intent(state: AgentState) -> AgentState:
    """이메일 의도를 분류하는 노드"""
    email = state["messages"][-1].content
    response = llm.invoke(f"다음 이메일의 의도를 분류해: {email}\n반환형식: refund/technical/shipping/general")
    return {"intent": response.content.strip()}

def search_manual(state: AgentState) -> AgentState:
    """매뉴얼에서 관련 내용을 검색하는 노드"""
    # 실제로는 벡터 DB나 키워드 검색 연동
    intent = state["intent"]
    manual = MANUALS.get(intent, "해당 내용 없음")
    return {"manual_result": manual}

def generate_response(state: AgentState) -> AgentState:
    """최종 답변을 생성하는 노드"""
    prompt = f"""
    이메일 의도: {state['intent']}
    매뉴얼 내용: {state['manual_result']}
    고객 이메일을 바탕으로 친절하고 정확한 답변을 작성해주세요.
    """
    response = llm.invoke(prompt)
    return {"messages": [response]}

graph.add_edge("classify_intent", "search_manual")
graph.add_edge("search_manual", "generate_response")

def should_escalate(state: AgentState) -> str:
    """에스컬레이션 필요 여부 판단"""
    if state["need_escalation"]:
        return "human_escalation"
    return "generate_response"

graph.add_conditional_edges(
    "check_complexity",
    should_escalate,
    {
        "human_escalation": "human_escalation",
        "generate_response": "generate_response"
    }
)

from langgraph.graph import StateGraph, END

# 1. StateGraph 생성
graph = StateGraph(AgentState)

# 2. Node 추가
graph.add_node("classify_intent", classify_intent)
graph.add_node("search_manual", search_manual)
graph.add_node("check_complexity", check_complexity)
graph.add_node("generate_response", generate_response)
graph.add_node("human_escalation", human_escalation)

# 3. 시작점 설정
graph.set_entry_point("classify_intent")

# 4. Edge 연결
graph.add_edge("classify_intent", "search_manual")
graph.add_edge("search_manual", "check_complexity")
graph.add_conditional_edges("check_complexity", should_escalate, {...})
graph.add_edge("generate_response", END)
graph.add_edge("human_escalation", END)

# 5. Compile
app = graph.compile()

def generate_topic(state):
    topic = llm.invoke("재미있는 농담 주제를 하나 골라줘")
    return {"topic": topic.content}

def write_joke(state):
    joke = llm.invoke(f"{state['topic']}에 대한 짧은 농담 써줘")
    return {"draft": joke.content}

def refine_joke(state):
    refined = llm.invoke(f"이 농담을 더 재미있게 다듬어줘: {state['draft']}")
    return {"final": refined.content}

# 농담, 시, 소설 첫 문장을 동시에 생성
graph.add_edge("start", "write_joke")
graph.add_edge("start", "write_poem")
graph.add_edge("start", "write_story")

# 세 결과를 합치는 노드
graph.add_edge("write_joke", "combine_results")
graph.add_edge("write_poem", "combine_results")
graph.add_edge("write_story", "combine_results")

def route_customer_query(state) -> str:
    intent = state["intent"]
    routing_map = {
        "payment": "payment_specialist",
        "technical": "tech_specialist",
        "shipping": "shipping_specialist",
    }
    return routing_map.get(intent, "general_support")

graph.add_conditional_edges(
    "classify",
    route_customer_query,
    {
        "payment_specialist": "payment_specialist",
        "tech_specialist": "tech_specialist",
        "shipping_specialist": "shipping_specialist",
        "general_support": "general_support",
    }
)

from langgraph.graph import interrupt

def check_before_send(state):
    if state["is_sensitive"]:
        # 여기서 그래프가 일시 중단되고 사람 확인 대기
        interrupt("민감한 내용이 감지되었습니다. 검토 후 승인해주세요.")
    return state

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
app = graph.compile(checkpointer=memory)

# thread_id로 대화 세션을 구분
config = {"configurable": {"thread_id": "user_123_session_1"}}

# 첫 번째 메시지
result = app.invoke({"messages": [HumanMessage("안녕, 나는 서울에 살아")]}, config)

# 두 번째 메시지 — 이전 맥락을 기억함
result = app.invoke({"messages": [HumanMessage("내 도시가 어디라고?")]}, config)
# → "서울이라고 하셨습니다"

# 현재까지의 모든 체크포인트 조회
history = list(app.get_state_history(config))
for checkpoint in history:
    print(checkpoint.config, checkpoint.values)

# 특정 과거 체크포인트로 이동
past_config = history[3].config  # 4번째 체크포인트
app.invoke(None, past_config)     # 그 시점부터 재실행

# 현재 State 확인
state = app.get_state(config)
print(state.values)  # 현재 State 전체 내용
print(state.next)    # 다음에 실행될 노드

# State 직접 수정 후 재실행 (과거 메시지 수정)
app.update_state(config, {"messages": [HumanMessage("수정된 질문")]})
app.invoke(None, config)  # 수정된 상태에서 재실행

# 컴파일 시 interrupt_before 설정
app = graph.compile(
    checkpointer=memory,
    interrupt_before=["send_email"]  # 이 노드 실행 전 중단
)

# 실행 → "send_email" 직전에 자동으로 멈춤
app.invoke(state, config)

# 사람이 확인 후 승인
print(app.get_state(config).values)  # 현재 State 확인
# → 승인 시
app.invoke(None, config)  # None을 넣으면 멈춘 지점부터 재개

# 체크포인트 A에서 두 가지 접근 방식 비교
branch_config_1 = app.update_state(config, {"strategy": "conservative"})
branch_config_2 = app.update_state(config, {"strategy": "aggressive"})

result_1 = app.invoke(None, branch_config_1)
result_2 = app.invoke(None, branch_config_2)

# 더 좋은 결과를 선택

from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage, SystemMessage

# Supervisor가 어떤 에이전트에게 일을 넘길지 결정
def supervisor_node(state):
    system_prompt = """당신은 팀 슈퍼바이저입니다. 
    다음 팀원 중 적합한 사람에게 작업을 위임하세요:
    - researcher: 정보 조사 및 수집
    - writer: 글 작성 및 편집  
    - analyst: 데이터 분석
    작업이 완료되면 FINISH를 반환하세요."""
    
    response = llm.invoke([SystemMessage(system_prompt)] + state["messages"])
    return {"next": response.content}  # "researcher" | "writer" | "analyst" | "FINISH"

def should_continue(state):
    return state["next"] if state["next"] != "FINISH" else END

print(app.get_graph().draw_mermaid())
# → Mermaid 코드 출력 → https://mermaid.live에서 시각화

for event in app.stream(initial_state, config):
    for key, value in event.items():
        print(f"노드 '{key}' 실행 후 State:", value)