AI 에이전트 상태 관리 아키텍처 — LangGraph 체크포인팅으로 프로덕션 신뢰성 확보하기

from typing import Annotated, TypedDict
from operator import add
from langchain_core.messages import BaseMessage
 
class AgentState(TypedDict):
    # 누적 필드: 체크포인트 재개 시 이전 목록에 새 항목을 append
    messages: Annotated[list[BaseMessage], add]
    collected_docs: Annotated[list[str], add]
 
    # 덮어쓰기 필드: 새 값으로 그냥 교체
    current_task: str
    status: str
    error_count: int

from typing import Annotated, TypedDict
from operator import add
from langchain_core.messages import BaseMessage
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.graph import StateGraph
 
async def run_agent():
    conn_string = "postgresql://agent_user:secret@localhost:5432/agent_db"
 
    with PostgresSaver.from_conn_string(conn_string) as checkpointer:
        graph = StateGraph(AgentState)
 
        graph.add_node("research", research_node)
        graph.add_node("analyze", analyze_node)
        graph.add_edge("research", "analyze")
 
        app = graph.compile(checkpointer=checkpointer)
 
        # thread_id로 세션을 식별 — 재개할 때 동일한 thread_id를 씁니다
        config = {"configurable": {"thread_id": "task-2026-001"}}
        result = await app.ainvoke({"current_task": "analyze_reports"}, config)

# 3번째 단계에서 강제 종료 (Ctrl+C)
# → PostgreSQL에 step=2까지의 체크포인트 저장됨
 
# 동일 thread_id로 재개
>>> result = await app.ainvoke(None, config)  # None = 저장된 상태에서 재개
# {'status': 'completed', 'collected_docs': ['doc_0.json', 'doc_1.json', ...], ...}
# → 0, 1, 2번 단계는 재처리 없이 3번 단계부터 이어서 실행됨

import json
from typing import Annotated, TypedDict
from operator import add
import boto3
 
s3_client = boto3.client("s3")
 
class DocumentState(TypedDict):
    # 포인터만 저장 — 실제 내용은 S3에
    processed_doc_keys: Annotated[list[str], add]
    summary_keys: Annotated[list[str], add]
 
    # 경량 메타데이터만 상태에 보존
    current_index: int
    total_docs: int
    failed_indices: Annotated[list[int], add]
 
def process_document_node(state: DocumentState) -> dict:
    idx = state["current_index"]
    doc_content = fetch_document(idx)
 
    analysis = run_llm_analysis(doc_content)
 
    # 분석 결과는 S3에 저장, 키만 상태에 기록
    key = f"analysis/{idx}/result.json"
    s3_client.put_object(
        Bucket="agent-results",
        Key=key,
        Body=json.dumps(analysis)
    )
 
    return {
        "processed_doc_keys": [key],
        "current_index": idx + 1
    }

from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import List
import uuid
 
class EventType(Enum):
    TASK_STARTED = "task_started"
    DECISION_MADE = "decision_made"
    TOOL_CALLED = "tool_called"
    TOOL_RESULT_RECEIVED = "tool_result_received"
    TASK_COMPLETED = "task_completed"
    ERROR_OCCURRED = "error_occurred"
 
@dataclass
class AgentEvent:
    event_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    event_type: EventType = EventType.TASK_STARTED
    agent_id: str = ""
    timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    payload: dict = field(default_factory=dict)
    # 이벤트 로그는 append-only — 수정·삭제 없음
 
class EventStore:
    def __init__(self):
        self._events: List[AgentEvent] = []
 
    def append(self, event: AgentEvent) -> None:
        self._events.append(event)
 
    def project_current_state(self, task_id: str) -> dict:
        """이벤트 투영으로 현재 상태 도출"""
        state = {"status": "pending", "decisions": [], "tool_results": {}}
 
        for event in self._events:
            if event.payload.get("task_id") != task_id:
                continue
            if event.event_type == EventType.TASK_STARTED:
                state["status"] = "running"
                state["started_at"] = event.timestamp
            elif event.event_type == EventType.DECISION_MADE:
                state["decisions"].append(event.payload)
            elif event.event_type == EventType.TOOL_RESULT_RECEIVED:
                state["tool_results"][event.payload["tool"]] = event.payload["result"]
            elif event.event_type == EventType.TASK_COMPLETED:
                state["status"] = "completed"
 
        return state
 
    def replay_from(self, task_id: str, event_index: int) -> dict:
        """특정 시점까지의 이벤트만으로 상태 재구성 — 디버깅·감사 추적에 활용"""
        state = {"status": "pending", "decisions": [], "tool_results": {}}
 
        for event in self._events[:event_index]:
            if event.payload.get("task_id") != task_id:
                continue
            if event.event_type == EventType.TASK_STARTED:
                state["status"] = "running"
                state["started_at"] = event.timestamp
            elif event.event_type == EventType.DECISION_MADE:
                state["decisions"].append(event.payload)
            elif event.event_type == EventType.TOOL_RESULT_RECEIVED:
                state["tool_results"][event.payload["tool"]] = event.payload["result"]
            elif event.event_type == EventType.TASK_COMPLETED:
                state["status"] = "completed"
 
        return state

from typing import Annotated, TypedDict
from operator import add
from langgraph.types import interrupt, Command
from langgraph.graph import StateGraph
from langgraph.checkpoint.postgres import PostgresSaver
 
class AgentState(TypedDict):
    messages: Annotated[list, add]
    risk_score: float
    approved: bool
 
async def risk_assessment_node(state: AgentState) -> dict:
    """위험도 계산 후 임계값 초과 시 human review로 자동 전환"""
    risk = calculate_risk(state)
 
    if risk > 0.8:
        # interrupt()가 호출되면 현재 상태가 체크포인트에 저장되고 실행 중단
        # 재개 시 여기서 반환한 값이 decision 변수에 담겨 들어옴
        decision = interrupt({
            "message": "위험도가 높습니다. 승인하시겠습니까?",
            "risk_score": risk,
            "context": state["messages"][-3:]
        })
        return {"risk_score": risk, "approved": decision.get("approved", False)}
 
    return {"risk_score": risk, "approved": True}
 
# --- 외부 웹훅이나 슬랙 봇이 응답을 받은 뒤 재개하는 코드 ---
 
async def resume_with_approval(app, config: dict, approved: bool):
    # Command(resume=...)로 interrupt에 응답을 전달하면서 재개
    result = await app.ainvoke(
        Command(resume={"approved": approved}),
        config  # 동일한 thread_id — 저장된 체크포인트에서 이어서 실행
    )
    return result

SELECT pg_column_size(checkpoint) AS checkpoint_bytes
FROM checkpoints
ORDER BY created_at DESC
LIMIT 10;