사용자 요청이 에이전트에서 처리되기까지

앞선 글들은 시스템을 일부러 나누어 보았다. Conversation은 Run이 아니다에서는 대화와 실행을 분리했고, Worker가 죽으면 어떻게 되는가에서는 전달과 완료를 분리했다.

이번에는 사용자가 보낸 한 건의 요청을 끝까지 따라가 본다.

NVIDIA의 최근 실적과 valuation을 분석해줘.

이 글을 읽고 세 가지만 답할 수 있으면 충분하다.

1. 요청이 중간에 사라지지 않으려면 무엇을 남겨야 하는가?
2. 각 순간 실행 책임은 누구에게 있는가?
3. 브라우저 연결이 끊겨도 사용자는 결과를 어떻게 다시 찾는가?

1. 1

   접수브라우저 → API

   인증·권한·입력 검증

   이 단계에서는 아직 실행하지 않는다.
   ↓
2. 2

   기록API → PostgreSQL

   run과 outbox를 함께 기록

   서버가 다시 시작돼도 요청을 찾을 수 있다.
   ↓
3. 3

   전달릴레이 → Redis Stream

   실행에 필요한 식별자만 발행

   전달에 실패해도 다시 시도할 수 있다.
   ↓
4. 4

   실행워커 attempt → LangGraph

   이번 시도의 실행 권한 확인

   중복 전달돼도 바로 다시 실행하지 않는다.
   ↓
5. 5

   확정PostgreSQL → Redis → XACK

   결과를 먼저 확정하고 ack는 마지막

   장애 뒤에도 최종 상태를 다시 찾을 수 있다.

1. 요청을 받아 안전하게 남긴다

API의 첫 책임은 실행이 아니라 검증이다

브라우저는 하나의 대화 세션 안에서 새 실행(run)을 요청한다. API는 이 시점에 LLM이나 도구를 실행하지 않는다. 먼저 인증된 사용자와 세션의 소유 관계, 그리고 요청 내용을 확인한다.

@app.post("/sessions/{session_id}/runs", status_code=202)
async def create_run(
    session_id: UUID,
    request: AnalysisRequest,
    identity: Annotated[Identity, Depends(current_identity)],
):
    session = await load_owned_session(
        tenant_id=identity.tenant_id,
        user_id=identity.user_id,
        session_id=session_id,
    )
    if session is None:
        raise HTTPException(404, "session not found")

SSO나 ingress가 인증을 마쳤더라도, 클라이언트가 보낸 X-User-ID 같은 임의 헤더를 그대로 믿으면 안 된다. 이 확인을 마친 뒤에야 Redis 키, checkpoint, 실행 기록에 접근할 수 있다.

202를 반환하기 전에 남겨야 할 것

API 프로세스가 응답 직후 멈춰도 요청을 다시 찾아낼 수 있으려면, run_id와 입력이 프로세스 밖에 남아 있어야 한다. 이 글에서는 run 레코드와 outbox 레코드를 같은 PostgreSQL 트랜잭션으로 기록하는 방식을 기준으로 삼는다.

BEGIN;

INSERT INTO runs (run_id, tenant_id, user_id, session_id, status, input_json)
VALUES (:run_id, :tenant_id, :user_id, :session_id, 'submitting', :input_json);

INSERT INTO run_outbox (event_id, run_id, topic, payload_json)
VALUES (:event_id, :run_id, 'agent:jobs', :payload_json);

COMMIT;

submitting은 “사용자 요청은 안전하게 받았지만, 아직 큐 전달이 끝났는지는 확인하지 못했다”는 뜻이다. 곧바로 queued라고 기록하면 DB 레코드만 생기고 큐 전달이 실패한 경우를 표현하기 어렵다.

간단한 시스템이라면 DB에 기록한 뒤 XADD를 바로 호출하고, submitting 상태의 run을 주기적으로 다시 발행하는 reconciler를 둘 수 있다. 어떤 방식을 택하든 원칙은 같다. 202 Accepted는 큐가 이미 실행을 시작했다는 뜻이 아니라, 요청을 복구 가능한 형태로 받았다는 뜻이다.

2. 실행 책임을 워커에게 넘긴다

run과 outbox가 기록된 뒤, 릴레이 프로세스가 Redis Stream에 작은 실행 메시지를 발행한다. 긴 프롬프트나 대화 전체를 복제하지 않고, 워커가 영속 저장소에서 다시 읽을 수 있는 식별자만 보낸다.

await redis.xadd(
    "agent:jobs",
    {
        "run_id": str(run_id),
        "session_id": str(session_id),
        "outbox_event_id": str(event_id),
    },
)

릴레이 프로세스가 Redis에 발행한 뒤 완료 표시를 남기기 전에 멈추면 같은 메시지가 다시 발행될 수 있다. 그래서 outbox는 정확히 한 번만 처리됨(exactly-once)을 약속하는 장치가 아니다. 중복 전달은 허용하되, 워커가 같은 run을 다시 실행하지 않게 만드는 장치다.

1. 01접수 담당API사용자 식별 · 권한 확인 · 입력 검증

   run과 outbox를 남기면 API의 일은 끝난다.

   책임이 다음 주체에게 넘어간다 ↓
2. 02전달 담당Queue미전달 → pending

   아직 확인되지 않은 전달을 기억한다.

   책임이 다음 주체에게 넘어간다 ↓
3. 03실행 담당Worker attempt그래프 · 도구 · checkpoint

   lease와 fence 아래에서 한 번의 시도를 수행한다.

   책임이 다음 주체에게 넘어간다 ↓
4. 04최종 결과 담당PostgreSQL종료 상태 · 결과 · 감사 기록

   재연결과 재전달을 판단하는 최종 기준이다.

큐의 entry와 워커의 attempt는 다르다

워커 인스턴스 하나가 consumer group에서 entry를 읽으면, 그 entry는 PEL에 남는다. 하지만 워커는 payload만 믿고 도구를 호출하지 않는다. PostgreSQL에서 run을 다시 읽어 이미 끝났는지, 취소됐는지, 이번 attempt가 실행 권한을 갖는지 확인한다.

run = await run_store.get(run_id)

if run.status in {"succeeded", "failed", "cancelled"}:
    await queue.ack(entry_id)
    return

attempt = await run_store.start_attempt(
    run_id=run.id,
    worker_id=worker_id,
)

attempt_id와 fencing token은 “이번 전달이 가진 실행 권한”을 구분한다. 워커가 멈추고 다른 워커가 reclaim하면 같은 run에 새 attempt가 생길 수 있다. 이때 이전 워커의 늦은 결과가 최종 결과를 덮지 못하도록, terminal update는 attempt와 fence를 조건으로 원자적으로 처리해야 한다.

3. 실행 결과를 화면에 필요한 형태로 바꾼다

session, run, attempt는 같은 ID가 아니다

워커는 실행 전에 쓰임이 서로 다른 정보를 읽는다.

• session: 이전 대화와 이어지는 맥락
• run: 사용자가 이번에 요청한 처리 단위
• attempt: 특정 워커가 수행하는 한 번의 시도
• checkpoint: LangGraph가 재개를 위해 보관하는 실행 상태

LangGraph의 thread_id는 세션의 맥락을 잇는 실행 키가 될 수 있지만, 이번 요청의 run_id와 같지는 않다. 이 구분이 있어야 “대화는 이어지되, 이전 실행의 실패가 다음 실행으로 넘어가지 않는다”는 원칙을 지킬 수 있다.

그래프 내부 상태를 그대로 브라우저에 보내지 않는다

graph.astream()에서 나오는 값은 그래프가 내부적으로 쓰는 값이다. 브라우저에 필요한 것은 그 전체가 아니라, 현재 화면을 복원할 작은 상태 정보와 시간순으로 보여 줄 이벤트다.

async for part in graph.astream(
    build_graph_input(run),
    config=config,
    stream_mode=["updates", "custom"],
    version="v2",
):
    await project_stream_part(run, attempt, part)

updates는 화면에 필요한 현재 상태로 줄여 Redis Hash에 저장하고, custom은 진행 상황·승인 요청처럼 남겨야 할 이벤트로 정규화해 Redis Stream에 append한다. 실제 이벤트 형식은 사용 중인 LangGraph 버전에 맞춰 한 곳에서 변환하는 편이 안전하다.

RUNTIME실행 중 생성된 값

그래프 내부 상태를 그대로 브라우저에 보내지 않는다.

↓ 애플리케이션 형식으로 변환

화면 상태UI snapshotstatus · progress · current_step

Redis Hash에 저장해 화면을 빠르게 복원한다.

이벤트 시간순 기록Event timelinestep_started · approval · completed

Redis Stream에 추가해 SSE를 다시 연결할 때 쓴다.

↓

FastAPI SSE → Browser

브라우저는 실행을 맡지 않고, snapshot과 event를 받아 화면에 반영한다.

브라우저는 실행자가 아니라 관측자다

브라우저는 202을 받으면 먼저 run의 현재 상태, 최근 이벤트, 이어 읽을 cursor를 가져온다. 그다음 SSE를 연결한다. 처음부터 SSE만 기다리면 연결되기 전에 생긴 상태를 놓칠 수 있다.

연결이 끊겨도 워커의 run은 계속된다. 다시 연결한 브라우저는 마지막으로 적용한 event_id 이후의 이벤트를 받고, 이벤트 보존 기간을 넘겼다면 snapshot을 다시 읽는다. SSE endpoint도 session·run의 소유권을 확인해야 한다. streaming 연결이라고 해서 권한 검사가 사라지는 것은 아니다.

4. 완료를 확정한다

그래프가 결과를 냈다고 해서 큐 작업까지 끝난 것은 아니다. 완료에는 지켜야 할 순서가 있다.

1. 최종 결과와 종료 상태를 PostgreSQL에 확정한다.
2. Redis의 화면용 상태를 completed로 갱신한다.
3. 완료 이벤트를 append한다.
4. 마지막에 큐 entry를 XACK 한다.

result = await execute_graph(run, attempt)

await run_store.complete(
    run_id=run.id,
    attempt_id=attempt.id,
    fence=attempt.fence,
    result=result,
)

await redis.hset(f"run:{run.id}:state", mapping={"status": "succeeded"})
await append_completion_event(run.id)
await redis.xack("agent:jobs", "agent-workers", entry_id)

Redis에 만든 화면용 상태나 완료 이벤트 기록이 실패해도 PostgreSQL의 종료 레코드가 최종 기준이다. 반대로 결과를 commit한 뒤 XACK 전에 워커가 멈추면 entry는 다시 전달된다. 새 워커는 종료 상태를 확인하고 그래프를 다시 실행하지 않은 채 XACK만 한다.

이 순서가 중요한 이유는 간단하다. ack는 “다시는 이 일을 볼 필요가 없다”는 신호이므로, 그 전에 결과가 안전하게 남아 있어야 한다.

5. 장애가 나도 다시 찾아 처리한다

정상 흐름만 따라가면 설계가 그럴듯해 보인다. 실제로는 각 경계에서 프로세스가 멈췄을 때 무엇이 남는지가 더 중요하다.

1. 1
   멈춘 지점기록 전
   남는 기준

   아무 run도 남지 않는다

   복구

   요청이 실패했다는 응답을 보내고, 새 요청을 받는다.

2. 2
   멈춘 지점기록 후 · 전달 전
   남는 기준

   run과 outbox가 남는다

   복구

   릴레이 또는 reconciler가 큐에 다시 발행한다.

3. 3
   멈춘 지점결과 확정 후 · ack 전
   남는 기준

   최종 결과가 남는다

   복구

   새 워커는 다시 실행하지 않고 XACK만 한다.

4. 4
   멈춘 지점SSE 연결 종료
   남는 기준

   실행과 이벤트가 남는다

   복구

   cursor로 놓친 이벤트를 다시 읽고, 범위를 넘으면 snapshot을 다시 읽는다.

운영 로그에는 서로 연결되는 ID를 남겨야 한다

run_id 하나만으로는 부족하다.

• session_id: 어느 대화에 속하는가
• run_id: 사용자가 요청한 처리 단위는 무엇인가
• attempt_id: 어떤 워커의 시도였는가
• queue_entry_id: 어떤 전달이 pending 또는 ack 되었는가
• event_id: 브라우저가 어디까지 받아 봤는가
• tool_call_id: 외부 부수 효과를 중복 없이 식별할 수 있는가

이 연결이 있어야 “사용자는 결과를 못 봤는데 도구는 실행됐는가?” 같은 운영 질문에 답할 수 있다.

테스트도 경계 전체를 끊어 봐야 한다

최소한의 정상 흐름 테스트는 실제 API, 큐, 워커, 저장소를 연결한다. 하지만 더 중요한 것은 각 경계에서 일부러 멈추게 하는 테스트다.

• run을 commit한 뒤, 큐에 publish하기 전
• XREADGROUP 뒤, attempt 생성 전
• 도구의 부수 효과 뒤, execution ledger를 commit하기 전
• 결과를 commit한 뒤, XACK 하기 전
• SSE 이벤트를 받은 뒤, UI에 적용하기 전

각 테스트에서 확인할 것은 “에러가 발생했다”가 아니다. run이 유실되지 않고, 부수 효과가 중복되지 않으며, 사용자가 끝내 최종 결과를 다시 읽을 수 있는가다.

이 설계에서 섞으면 안 되는 네 가지

• session과 run
• 큐의 delivery와 워커의 attempt
• LangGraph 내부 상태와 화면용 상태
• 결과 확정과 큐 ack

사용자 요청 한 건을 끝까지 따라가면 각 경계가 선명해진다. 다음 질문도 자연스럽게 나온다. 사용자가 실행 중 취소를 누르면 누가 멈추고, 이미 시작한 도구는 어떻게 다뤄야 할까?

다음 글에서는 Cancellation은 단순한 상태값이 아니라 하나의 protocol이라는 관점에서, 취소 요청·워커의 cooperative check·도구 경계·종료 상태 경쟁을 다룬다.