State Machine 설계

from typing import TypedDict, Annotated, Literal
from operator import add
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class WorkflowState(TypedDict):
    # add_messages: 새 메시지를 리스트에 추가 (덮어쓰지 않음)
    messages: Annotated[list[BaseMessage], add_messages]

    # add: 각 노드 결과를 누적 (병렬 노드 안전)
    search_results: Annotated[list[str], add]
    errors: Annotated[list[str], add]

    # 단순 덮어쓰기 (최신값만 유지)
    current_plan: list[str]
    status: Literal["planning", "executing", "reviewing", "done"]
    iteration: int
    human_approved: bool

# Private 상태 (서브그래프 내부에서만 사용)
class SearchSubState(TypedDict):
    query: str
    raw_results: list        # 외부에 노출 안 됨
    filtered_results: list   # 외부에 노출 안 됨
    final_summary: str       # 이것만 WorkflowState.search_results에 병합

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, END

checkpointer = MemorySaver()

# 브레이크포인트: human_review 노드 실행 전 중단
graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["human_review"],    # 노드 실행 전 중단
    interrupt_after=["dangerous_action"]  # 노드 실행 후 중단
)

config = {"configurable": {"thread_id": "workflow-001"}}

# 1단계: 브레이크포인트까지 실행
state = await graph.ainvoke(
    {"messages": [{"role": "user", "content": "데이터베이스 마이그레이션 실행"}]},
    config=config
)
# → human_review 노드 직전에 중단됨

# 2단계: 현재 상태 확인
current = await graph.aget_state(config)
print(f"중단 위치: {current.next}")  # ('human_review',)
print(f"현재 계획: {current.values['current_plan']}")

# 3단계: 사람이 검토 후 승인 — 상태 업데이트 후 재개
await graph.aupdate_state(
    config,
    {"human_approved": True},
    as_node="human_review"  # 어느 노드가 업데이트했는지 명시
)

# 4단계: 중단점 이후부터 재개 (None 입력으로 계속)
result = await graph.ainvoke(None, config=config)

# 거부 시: 다른 상태로 업데이트 후 재개
await graph.aupdate_state(
    config,
    {"human_approved": False, "status": "done"},
    as_node="human_review"
)

from langgraph.types import interrupt, Command

async def approval_node(state: WorkflowState) -> dict:
    """위험한 작업 전 사람의 승인을 요청"""
    plan_summary = "\n".join(
        f"{i+1}. {step}" for i, step in enumerate(state["current_plan"])
    )

    # 실행 중단 + 사람에게 데이터 전달
    human_decision = interrupt({
        "type": "approval_required",
        "message": f"다음 계획을 실행합니다:\n{plan_summary}\n\n승인하시겠습니까?",
        "plan": state["current_plan"]
    })

    # interrupt() 반환값: 사람이 aupdate_state()로 전달한 값
    if human_decision == "approve":
        return {"human_approved": True, "status": "executing"}
    elif human_decision == "reject":
        return {"human_approved": False, "status": "done"}
    else:
        # 수정된 계획 받기
        return {
            "current_plan": human_decision["revised_plan"],
            "status": "planning"
        }

# API에서 사람의 응답 처리
async def handle_human_response(thread_id: str, decision: str):
    config = {"configurable": {"thread_id": thread_id}}
    # interrupt()의 반환값으로 decision을 전달
    await graph.aupdate_state(config, Command(resume=decision))
    # 재개
    return await graph.ainvoke(None, config=config)

# ─── 서브그래프 정의 ─────────────────────────────────
search_builder = StateGraph(SearchSubState)
search_builder.add_node("fetch",   fetch_node)
search_builder.add_node("filter",  filter_node)
search_builder.add_node("summarize", summarize_node)
search_builder.set_entry_point("fetch")
search_builder.add_edge("fetch", "filter")
search_builder.add_edge("filter", "summarize")
search_builder.add_edge("summarize", END)
search_subgraph = search_builder.compile()

# ─── 메인 그래프에서 서브그래프를 노드로 사용 ────────
async def search_adapter(state: WorkflowState) -> dict:
    """WorkflowState → SearchSubState 변환 후 서브그래프 실행"""
    sub_result = await search_subgraph.ainvoke({
        "query": state["messages"][-1].content
    })
    # 서브그래프 출력 → 메인 상태로 병합
    return {"search_results": [sub_result["final_summary"]]}

main_builder.add_node("search", search_adapter)