LLM 에이전트가 화면 변경을 놓치지 않게 만드는 상태 동기화

사용자가 화면에서 입주일을 9월 28일에서 9월 30일로 바꿨다고 해보자. 제품 DB와 화면은 즉시 최신값을 보여준다. 그런데 며칠 전에 만든 AI 대화에서 일정을 다시 물으면 9월 28일을 기준으로 답할 수 있다.

이상해 보이지만 자연스러운 결과다. 제품 DB가 바뀌었다는 사실만으로 이미 저장된 대화나 다음 LLM 요청이 함께 바뀌지는 않는다. 더구나 AI도 도구를 호출해 같은 데이터를 수정한다면 변경 경로가 두 개가 된다. 사람이 고친 값과 AI가 고친 값을 모든 대화가 빠짐없이 알아야 한다.

이 문제를 풀면서 두 방식을 비교했다.

• run을 시작할 때 현재 제품 상태를 읽어 system prompt에 넣는 dynamic snapshot
• 변경 이벤트와 대화별 cursor로 마지막 실행 뒤의 변경을 전달하는 event cursor

둘 중 하나를 고르는 문제는 아니었다. Dynamic prompt는 현재 상태를 LLM에 보여주는 전달 장치다. Event와 cursor는 대화별 미처리 변경을 계산하고 실패 시 다시 전달하는 동기화 규칙이다. 필요한 보장 수준에 따라 dynamic snapshot만 쓰거나 두 방식을 결합하면 된다.

DB는 최신인데 LLM 답은 왜 낡았을까

제품 상태, 대화 상태, 모델 입력은 각각 따로 관리된다.

제품 DB에는 입주일, 체크리스트, 현재 단계 같은 업무 데이터가 있다. LangGraph checkpointer는 특정 thread_id의 메시지와 graph state를 저장한다. LLM은 이 두 저장소를 직접 읽지 못한다. 모델이 실제로 확인하는 범위는 호출 시점의 messages, system message, tool schema, 이후에 들어오는 ToolMessage다.

다이어그램 1. 제품 DB가 최신이어도 대화 B의 checkpoint에는 9월 28일 기준 메시지가 남아 있다. DB 조회 결과를 ModelRequest에 넣지 않으면 LLM은 9월 30일을 알 수 없다. Graph state에 값을 저장하는 일과 모델 입력에 값을 넣는 일도 구분해야 한다.

LangGraph persistence 문서에 따르면 checkpointer는 graph state를 thread별 checkpoint로 저장한다. 이 기능은 대화를 이어가거나 중단 지점에서 재개할 때 유용하다. 하지만 checkpointer가 제품 DB의 모든 변경을 알아서 가져오지는 않는다. 여러 thread가 같은 제품을 공유할 때는 제품 상태를 별도로 읽어야 한다.

여기서 자주 생기는 오해가 하나 있다. before_agent에서 값을 state에 넣으면 LLM도 그 값을 안다고 생각하기 쉽다. 모델이 읽는 것은 임의의 state 필드가 아니라 ModelRequest다. State의 값을 system message나 messages로 조립하는 단계가 빠지면 최신값은 백엔드 안에만 남는다.

가장 가벼운 해법은 dynamic snapshot

현재값만 알면 되는 서비스라면 run 시작 시 제품 스냅샷을 한 번 읽고, 모델 호출마다 system prompt에 붙이는 방식으로 충분할 수 있다.

동작은 짧다.

1. abefore_agent에서 현재 사업 상태와 필요한 문서 포인터 조회
2. 조회 결과를 graph state에 저장
3. wrap_model_call 또는 @dynamic_prompt에서 state를 읽어 system message 구성
4. 한 run 안의 모델 호출은 같은 스냅샷을 사용

LangChain custom middleware 문서에 따르면 before_agent는 invocation마다 한 번, wrap_model_call은 모델 호출마다 실행된다. dynamic_prompt는 wrap_model_call로 동적 system prompt를 만드는 편의 기능이다.

개념만 남긴 코드는 이 정도다.

from typing import Any
from typing_extensions import NotRequired
 
from langchain.agents.middleware import AgentMiddleware, AgentState, ModelRequest
from langgraph.runtime import Runtime
 
 
class WorkspaceState(AgentState):
    workspace_snapshot: NotRequired[dict[str, Any]]
 
 
class WorkspaceSnapshotMiddleware(AgentMiddleware):
    state_schema = WorkspaceState
 
    async def abefore_agent(
        self,
        state: WorkspaceState,
        runtime: Runtime,
    ) -> dict[str, Any]:
        snapshot = await load_workspace(runtime.context.workspace_id)
        return {"workspace_snapshot": snapshot}
 
    def wrap_model_call(self, request: ModelRequest, handler):
        header = render_workspace_header(request.state["workspace_snapshot"])
        return handler(append_system_header(request, header))

append_system_header는 기존 system message를 보존하면서 상황 헤더를 덧붙이는 작은 함수라고 보면 된다. 실제 구현에서는 ModelRequest.override()로 새 요청을 만든다.

Dynamic snapshot의 장점은 명확하다. 별도 이벤트 스키마나 cursor 없이 현재값을 모델에 전달할 수 있다. 채팅 한 턴이 매번 새 invocation을 만든다면 턴마다 DB를 다시 읽는 것도 자연스럽다. 모델 호출이 여러 번 이어지는 한 run에서는 조회 결과를 재사용하므로 불필요한 DB 왕복도 줄일 수 있다.

한계도 분명하다.

• 이전 실행 뒤 무엇이 달라졌는지 알 수 없다.
• 사람이 바꿨는지 AI가 바꿨는지 구분하려면 별도 메타데이터가 필요하다.
• 어떤 대화가 특정 변경을 처리했는지 기록하지 않는다.
• 상태 조회 실패를 허용하고 run을 계속하면 최신 상태 없이 답할 수 있다.
• 같은 run 도중 외부에서 값이 바뀌면 다음 invocation 전까지 반영되지 않는다.

현재값 중심 안내에는 잘 맞는다. 오래된 답변을 교정해야 하거나 변경 경위가 답변에 영향을 준다면 정보가 부족하다.

현재값만으로 부족해지는 순간

대화 A에서 AI가 체크리스트를 완료 처리한 뒤 대화 B를 열었다고 해보자. Dynamic snapshot은 대화 B에 현재 체크리스트를 보여줄 수 있다. 하지만 대화 B의 기존 메시지에는 "아직 미완료"라고 적혀 있다.

LLM은 둘이 충돌한다는 사실을 알아차릴 수 있어도 왜 바뀌었는지는 모른다. 사용자가 직접 바꿨는지, 다른 대화의 AI가 바꿨는지, 자동 배치가 수정했는지 알 수 없다. 이전 답변을 어느 범위까지 다시 해석해야 하는지도 불분명하다.

여러 대화가 같은 workspace를 공유하면 필요한 질문이 달라진다.

이 대화가 마지막으로 성공한 뒤 무엇이 바뀌었는가?

현재 스냅샷만으로는 답할 수 없다. 변경 이력과 대화별 처리 위치가 필요하다.

변경은 한 번 기록하고 대화가 실행할 때 가져온다

쓰기 시점에 모든 checkpoint를 갱신하는 방법부터 떠올릴 수 있다. 대화가 N개면 변경 한 번에 N개 checkpoint를 써야 한다. 중간에 실패하면 일부 대화만 최신 상태가 된다. 대화 기록도 제품 상태의 복제본처럼 변한다.

더 단순한 구조는 변경을 한 번 기록하고, 실행한 대화가 자기 몫을 가져가는 방식이다.

제품 row를 바꿀 때 workspace_events에 revision을 하나 남긴다. 각 thread는 마지막으로 처리한 revision을 cursor로 저장한다. 다음 run에서는 cursor < revision <= HEAD인 이벤트를 읽는다. 현재 제품 row도 함께 읽는다. 현재값은 답변의 기준이고 이벤트는 이전 답변이 낡은 이유다.

다이어그램 2. 화면 저장 시 제품 row와 이벤트 r41을 같은 트랜잭션에 기록한다. 대화 B는 cursor r40을 복원한 뒤 r41과 현재값을 읽는다. wrap_model_call에서 모델 요청에 넣은 시점부터 LLM이 변경을 확인한다. 모델 호출이나 검증이 실패하면 cursor는 r40에 남아 다음 run에서 r41을 다시 전달한다.

이 구조는 네 가지 규칙으로 정리된다.

제품 row와 이벤트는 함께 저장

화면 저장과 AI 도구 호출이 같은 쓰기 함수를 사용하도록 모은다. 제품 row만 바뀌고 이벤트가 빠지거나, 이벤트만 남고 실제 값은 바뀌지 않는 상황을 막으려면 둘을 같은 DB 트랜잭션에 넣어야 한다.

이벤트에는 최소한 다음 정보가 필요하다.

workspace_id
revision
entity_type, entity_id
changed_fields
actor_type, actor_id
source_thread_id
created_at

민감한 값 전체를 이벤트에 복제할 필요는 없다. 변경 필드와 식별자만 기록하고, 최종 유효 값은 제품 row에서 읽는 편이 안전하다.

대화마다 cursor를 따로 저장

대화 A가 r45까지 처리했고 대화 B가 r40에서 멈췄다면 두 대화의 미처리 범위는 다르다. Workspace에 cursor 하나만 두면 느린 대화의 위치를 표현할 수 없다. Cursor의 키는 workspace_id + thread_id가 된다.

새 대화는 생성 시점의 HEAD에서 시작하는 편이 보통 낫다. 과거 이벤트 수천 건을 첫 질문에 넣기보다 현재 상태를 읽고, 생성 이후의 변경만 추적한다. 과거 변경 경위가 필요하면 별도 조회 도구로 기간과 대상을 좁힌다.

실제 모델 요청에 넣기

abefore_agent가 현재 row와 미처리 이벤트를 읽어 state에 저장해도 LLM은 아직 모른다. wrap_model_call에서 system message나 messages로 바꿔야 한다.

모델 입력은 세 묶음이면 충분하다.

CURRENT
현재 유효한 제품 상태
 
CHANGES SINCE LAST SUCCESS
r41, user, 입주일 9월 28일 -> 9월 30일
 
HISTORY
checkpoint에서 복원한 기존 대화

현재 상태와 변경 이벤트를 함께 주면 모델은 최신값을 답변 기준으로 삼고, 기존 대화 내용이 왜 낡았는지도 이해할 수 있다.

성공한 범위까지만 cursor 이동

이벤트를 읽자마자 cursor를 옮기면 모델 호출 실패나 출력 검증 실패 뒤에 변경이 사라진다. Cursor는 전달 위치가 아니라 처리 완료 위치로 써야 한다.

run에서 이벤트를 50건만 전달했다면 HEAD가 아니라 실제로 전달한 마지막 revision까지 이동한다. 76건이 쌓였다면 첫 run은 50건, 다음 run은 26건을 처리한다.

이 방식은 at-least-once다. 실패 경계에 따라 같은 이벤트가 다시 들어올 수 있지만 누락되지는 않는다. 중복을 허용하지 않는 exactly-once 처리보다 구현과 운영이 단순하다. 모델 입력을 만드는 소비자는 같은 이벤트를 다시 받아도 안전해야 한다.

LangGraph 환경에서 어느 기능을 언제 쓸까

여기서 before_agent, wrap_model_call, after_agent는 LangGraph의 저수준 node API가 아니라 LangGraph 위에서 실행되는 LangChain agent middleware hook이다. 직접 StateGraph를 만든다면 같은 책임을 별도 node로 구현할 수 있다.

Runtime context에는 workspace_id, 사용자 식별자, 권한처럼 실행 조건을 둔다. 이 값도 모델에 자동 노출되지 않는다. LLM이 알아야 할 값만 명시적으로 prompt에 넣는다. 권한 판정용 식별자는 prompt에 의존하지 말고 도구 실행 경계에서 코드로 강제해야 한다.

Store는 thread를 넘어 유지할 장기 기억에 적합하다. 제품 DB의 정본을 복제하는 장소로 쓰면 두 저장소 중 어느 값이 최신인지 다시 해결해야 한다. 제품 상태는 제품 DB에서 읽고, Store에는 사용자 선호나 에이전트가 따로 축적한 기억처럼 수명이 다른 데이터를 두는 편이 낫다.

미들웨어 구현 골격

아래 코드는 책임 배치를 보여주기 위해 세부 타입과 오류 처리를 줄인 예시다.

from typing import Any
from typing_extensions import NotRequired, TypedDict
 
from langchain.agents.middleware import AgentMiddleware, AgentState, ModelRequest
from langgraph.runtime import Runtime
 
 
class RunContext(TypedDict):
    workspace_id: str
    thread_id: str
 
 
class SyncState(AgentState):
    workspace_cursor: NotRequired[int]
    pending_cursor: NotRequired[int]
    workspace_snapshot: NotRequired[dict[str, Any]]
    unseen_events: NotRequired[list[dict[str, Any]]]
 
 
class WorkspaceSyncMiddleware(AgentMiddleware):
    state_schema = SyncState
 
    async def abefore_agent(
        self,
        state: SyncState,
        runtime: Runtime[RunContext],
    ) -> dict[str, Any]:
        cursor = state.get("workspace_cursor", 0)
        snapshot, events = await read_sync_batch(
            workspace_id=runtime.context["workspace_id"],
            after_revision=cursor,
            limit=50,
        )
        delivered_revision = events[-1]["revision"] if events else cursor
        return {
            "workspace_snapshot": snapshot,
            "unseen_events": events,
            "pending_cursor": delivered_revision,
        }
 
    def wrap_model_call(self, request: ModelRequest, handler):
        header = render_workspace_context(
            current=request.state["workspace_snapshot"],
            changes=request.state.get("unseen_events", []),
        )
        return handler(append_system_header(request, header))
 
    async def aafter_agent(
        self,
        state: SyncState,
        runtime: Runtime[RunContext],
    ) -> dict[str, Any] | None:
        pending = state.get("pending_cursor")
        current = state.get("workspace_cursor", 0)
        if pending is None or pending <= current:
            return None
 
        await commit_thread_cursor(
            workspace_id=runtime.context["workspace_id"],
            thread_id=runtime.context["thread_id"],
            revision=pending,
        )
        return {"workspace_cursor": pending}

실제 코드에서는 read_sync_batch()가 일관된 DB snapshot 안에서 현재 row와 고정된 HEAD를 읽어야 한다. 조회 도중 새 이벤트가 들어와도 이번 run의 처리 범위가 흔들리지 않게 하기 위해서다.

aafter_agent는 agent invocation이 끝났음을 뜻한다. 스트리밍 응답의 마지막 바이트가 사용자 브라우저에 도착했다는 보장은 아니다. Cursor가 사용자의 수신 확인까지 나타내야 한다면 API 계층에 별도 delivery acknowledgment가 필요하다. 대부분의 AI 상태 동기화에서는 agent가 컨텍스트를 처리했다는 의미면 충분하다.

Dynamic snapshot과 event cursor 비교

다이어그램 3. Dynamic snapshot은 현재 상태를 매 run의 system prompt에 넣는다. Event cursor는 대화별 미처리 변경과 처리 완료 위치를 관리한다. 현재 상태와 변경 이유가 모두 필요하면 두 방식을 결합한다.

Dynamic prompt와 event cursor를 대체재로 보면 설계가 꼬인다. Event cursor를 써도 LLM에 정보를 전달하려면 결국 ModelRequest를 바꿔야 한다. 반대로 dynamic prompt만 써도 현재 상태는 전달할 수 있지만 대화별 처리 보장은 생기지 않는다.

실무에서는 아래 조합이 가장 설명하기 쉬웠다.

current snapshot   -> 지금 답변의 기준
unseen events      -> 이전 답변이 낡은 이유
conversation       -> 사용자의 질문과 대화 맥락

abefore_agent가 세 입력을 준비하고 wrap_model_call이 하나의 모델 요청으로 조립한다. aafter_agent는 성공한 revision만 확정한다.

쓰기 충돌과 중복 실행 방지

Cursor는 읽기 누락을 막지만 쓰기 충돌까지 해결하지는 않는다. 세 가지 식별자를 구분하면 문제가 단순해진다.

version은 오래된 값을 읽은 AI가 사용자의 최신 수정을 덮어쓰지 못하게 한다. AI 도구는 expected_version을 보내고, DB는 WHERE version = expected_version 조건이 맞을 때만 갱신한다.

mutation_id는 같은 도구 호출의 재시도를 식별한다. 네트워크 timeout 뒤 같은 요청이 다시 들어와도 제품 row와 이벤트를 한 번만 만든다.

thread_id lease는 같은 대화에서 두 run이 겹치는 일을 막는다. 두 run이 같은 cursor를 읽고 서로 다른 pending cursor를 확정하면 순서가 어긋날 수 있다. 한 thread에는 한 번에 하나의 run만 허용하는 편이 단순하다.

이 세 값은 비슷해 보이지만 막는 실패가 다르다.

적용 기준

현재값만 필요하고 매 턴 최신 상태를 다시 읽을 수 있다면 dynamic snapshot부터 시작하면 된다. 이벤트 테이블과 cursor를 먼저 만들 이유는 없다.

다음 조건 중 하나라도 중요하면 event cursor를 함께 고려할 만하다.

• 사람이 화면에서 수정하고 AI도 도구로 같은 데이터를 수정한다.
• 여러 대화가 하나의 workspace를 공유한다.
• 대화가 며칠이나 몇 주 동안 쉬었다가 다시 열린다.
• 기존 답변이 왜 달라졌는지 설명해야 한다.
• 모델 호출 실패 뒤 변경을 다시 전달해야 한다.
• 변경 주체와 시간을 감사 이력으로 남겨야 한다.

Prompt 크기도 설계 조건이다. 이벤트가 많으면 50건처럼 배치 상한을 두고, 실제로 전달한 마지막 revision까지만 cursor를 옮긴다. 같은 entity가 여러 번 바뀌었다면 현재 row를 기준으로 답하게 하고, 이벤트는 필요한 변경 이유만 압축한다. 원본 이력 전체는 도구로 조회하는 편이 낫다.

마무리

제품 DB, 대화 checkpoint, LLM 입력은 같은 상태가 아니다. 제품 DB가 최신이어도 모델 요청이 낡으면 답변도 낡는다.

Dynamic snapshot은 현재 상태를 LLM에 보여주는 가장 짧은 길이다. Event와 cursor는 대화별 미처리 변경과 실패 복구가 필요할 때 추가한다. 두 방식을 결합하면 LLM은 지금의 정답과 이전 답변이 낡은 이유를 함께 확인할 수 있다.

구현 순서도 이 책임 구분을 따라가면 된다. 쓰기 경로에서 변경을 한 번 기록하고, run 시작 시 현재값과 미처리 변경을 읽고, 모델 호출 직전에 요청에 넣고, agent가 끝난 뒤 처리한 revision을 확정한다.

참고 문서

• LangGraph Persistence
• LangChain Custom Middleware
• LangChain Middleware Reference
• LangChain ModelRequest Reference