LangGraph State 설계 가이드

제조 진단 State 설계

from typing import TypedDict, List, Optional, Literal
from langgraph.graph import StateGraph, END

class ManufacturingDiagnosticState(TypedDict):
    # 입력
    equipment_id: str
    symptom: str

    # 진단 결과
    sensor_data: Optional[dict]    # MES 센서 값
    alarm_codes: Optional[List[str]]  # 활성 알람
    maintenance_history: Optional[List[dict]]  # 정비 이력

    # 분석
    probable_causes: Optional[List[str]]  # 추정 원인
    recommended_action: Optional[str]  # 권고 조치
    urgency: Optional[Literal["urgent", "normal", "monitor"]]

    # 최종
    final_report: Optional[str]
    requires_human: bool  # 사람 검토 필요 여부

State 설계 원칙:

• Optional 필드는 각 Node에서 순차적으로 채움
• 불변 데이터(장비 ID)와 가변 데이터(진단 결과) 분리
• requires_human 같은 플래그로 Human-in-the-Loop 구현

조건부 분기 설계

def route_by_urgency(state: ManufacturingDiagnosticState):
    """긴급도에 따라 다음 노드를 결정"""
    if state["requires_human"]:
        return "human_review"  # 사람 검토 필요
    elif state["urgency"] == "urgent":
        return "escalate"      # 즉시 에스컬레이션
    elif state["urgency"] == "normal":
        return "create_work_order"  # 정비 요청 생성
    else:
        return "monitor"           # 모니터링 지속

graph = StateGraph(ManufacturingDiagnosticState)
graph.add_conditional_edges(
    "analyze_fault",  # 이 노드 실행 후
    route_by_urgency,  # 이 함수로 다음 노드 결정
    {
        "human_review": "human_review_node",
        "escalate": "escalate_node",
        "create_work_order": "work_order_node",
        "monitor": "monitor_node"
    }
)

Checkpoint: 워크플로우 이력

from langgraph.checkpoint.memory import MemorySaver

# 메모리 기반 Checkpoint
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# 실행 (thread_id로 대화 이력 관리)
config = {"configurable": {"thread_id": "CNC-M500-incident-001"}}
result = app.invoke({"equipment_id": "CNC-M500", "symptom": "주축 진동"}, config)

# 이후 같은 thread_id로 이어서 대화
result2 = app.invoke({"symptom": "이번엔 온도도 상승"}, config)
# → 이전 진단 결과를 기억하고 계속 분석

실무 가치: 동일 설비의 연속 이상 신호를 하나의 이력으로 관리하면, 누적 데이터 기반의 정확한 진단이 가능하다.