LangGraph란

LangGraph = LangChain 팀이 만든 Agent 오케스트레이션 프레임워크

핵심 아이디어:
  Agent의 실행 흐름을 "유향 그래프(Directed Graph)"로 정의한다.

  - Node = 작업 단위 (함수)
  - Edge = 작업 간 연결 (흐름)
  - State = 공유 상태 (메모리)

---

1. State: 공유 메모리

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class ManufacturingState(TypedDict):
    """제조 Agent 상태"""
    # 메시지 기록 (자동 누적)
    messages: Annotated[list, add_messages]

    # 설비 정보
    equipment_id: str
    equipment_status: dict

    # 진단 결과
    diagnosis: str
    severity: str  # "normal", "warning", "critical"

    # 조치 이력
    actions_taken: list[str]

    # 종료 여부
    should_end: bool

State는 모든 Node가 읽고 쓸 수 있는 공유 칠판이다. Node A가 equipment_status에 값을 쓰면, Node B가 그걸 읽을 수 있다.

---

2. Node: 작업 단위

def check_sensors(state: ManufacturingState) -> dict:
    """센서 데이터를 확인하는 Node"""
    equipment_id = state["equipment_id"]

    # 센서 데이터 조회 (실제로는 MCP 호출)
    sensor_data = get_sensor_data(equipment_id)

    # State 업데이트 (변경된 부분만 반환)
    return {
        "equipment_status": sensor_data,
        "messages": [{"role": "system", "content": f"센서 확인 완료: {equipment_id}"}],
    }


def diagnose(state: ManufacturingState) -> dict:
    """진단을 수행하는 Node"""
    status = state["equipment_status"]
    temp = status.get("temperature", 0)
    vib = status.get("vibration", 0)

    if temp >= 80 or vib >= 3.0:
        return {"diagnosis": "긴급 조치 필요", "severity": "critical"}
    elif temp >= 65 or vib >= 2.0:
        return {"diagnosis": "주의 필요", "severity": "warning"}
    else:
        return {"diagnosis": "정상", "severity": "normal"}

Node는 State를 입력받고, 변경사항을 반환하는 함수다. 반환하지 않은 필드는 그대로 유지된다.

---

3. Edge: 흐름 제어

일반 Edge (항상 같은 경로)

graph.add_edge("check_sensors", "diagnose")
# check_sensors 완료 후 항상 diagnose로 이동

Conditional Edge (조건부 분기)

def route_by_severity(state: ManufacturingState) -> str:
    """심각도에 따라 다른 경로로 분기"""
    severity = state["severity"]
    if severity == "critical":
        return "emergency_stop"     # 긴급 정지
    elif severity == "warning":
        return "send_alert"         # 경고 알림
    else:
        return "generate_report"    # 정상 보고

graph.add_conditional_edges(
    "diagnose",              # 출발 Node
    route_by_severity,       # 분기 함수
    {                        # 분기 매핑
        "emergency_stop": "emergency_stop",
        "send_alert": "send_alert",
        "generate_report": "generate_report",
    }
)

---

전체 그래프 조립

from langgraph.graph import StateGraph, END

# 1. 그래프 생성
graph = StateGraph(ManufacturingState)

# 2. Node 추가
graph.add_node("check_sensors", check_sensors)
graph.add_node("diagnose", diagnose)
graph.add_node("emergency_stop", emergency_stop)
graph.add_node("send_alert", send_alert)
graph.add_node("generate_report", generate_report)

# 3. Edge 연결
graph.set_entry_point("check_sensors")
graph.add_edge("check_sensors", "diagnose")
graph.add_conditional_edges("diagnose", route_by_severity, {
    "emergency_stop": "emergency_stop",
    "send_alert": "send_alert",
    "generate_report": "generate_report",
})
graph.add_edge("emergency_stop", END)
graph.add_edge("send_alert", END)
graph.add_edge("generate_report", END)

# 4. 컴파일
app = graph.compile()

# 5. 실행
result = app.invoke({
    "equipment_id": "PRESS-001",
    "messages": [],
    "actions_taken": [],
    "should_end": False,
})

---

LangGraph vs ReAct vs A2A 비교

비교	ReAct	LangGraph	A2A
구조	자유 루프	정의된 그래프	Agent 간 통신
상태 관리	없음	State 객체	Task/Message
분기	Tool 선택만	Conditional Edge	Agent Card 기반
복잡도	낮음	중간	높음
적합한 경우	단순 Tool 사용	복잡한 워크플로우	Multi-Agent 시스템