MINERVA BaseAgent 계약 v2

LangGraph 호환 인터페이스로 묶기 — 그리고 v1 어댑터

Phase C-2 마무리 편. 13~16편의 LangGraph 기초·노드 분해·State 설계·Checkpointing/HITL을 다시 BaseAgent 계약으로 묶는다. v1(ABC + run/stream + Pydantic)에 graph·state_schema· checkpointer 속성을 더해 v2를 정의하고, v1 에이전트를 v2 프레임워크에서 그대로 굴릴 수 있도록 LegacyAgentAdapter 패턴을 제시한다. FastAPI 라우터 변경, A/B 실험 호환, 마이그레이션 전략까지 다룬다.

Agent
저자

Kwangmin Kim

공개

2026년 05월 06일

1 Phase C-2를 묶는 한 가지 계약

13~16편에서 다음을 했다.

  • 13편 — LangGraph가 무엇이고 왜 필요한지 (Node·Edge·State·Conditional Edge)
  • 14편 — _prepare()를 7개 노드로 분해
  • 15편 — TypedDict와 reducer 4 패턴
  • 16편 — Checkpointer와 HITL 사이클

이제 마지막으로, 위 결과물을 하나의 인터페이스 계약으로 묶어야 한다. 묶지 않으면 새 에이전트를 만들 때마다 13~16편의 결정을 다시 반복해야 하고, FastAPI 라우터·A/B 실험·캐싱 같은 공통 기반이 LangGraph로 옮겨가지 못한다.

이번 편은 02편 — BaseAgent 계약 패턴에서 정의한 v1 계약을 LangGraph 시대로 확장하는 v2를 설계한다. 하위 호환을 깨지 않는 것이 설계 제약이다.

선행 학습

2 v1 계약 회상

src/core/contracts.py의 v1 BaseAgent는 다음 형태다.

from abc import ABC, abstractmethod
from typing import Iterator


class BaseAgent(ABC):
    name: str = ""
    domain: Optional[str] = None

    @abstractmethod
    def run(self, query: Query) -> Response:
        ...

    def stream(self, query: Query) -> Iterator[StreamEvent]:
        # 기본 구현은 run()을 한 번 감싸 done 이벤트만 yield
        response = self.run(query)
        yield StreamEvent(type="done", response=response)

v1의 강점

  • 이름·도메인이라는 가벼운 메타 + 두 메서드만 강제 → 에이전트 작성 진입 장벽이 낮다.
  • Query / Response 가 Pydantic이라 FastAPI 직렬화·검증이 자동으로 붙는다.
  • stream의 기본 구현이 있어 SSE가 필요 없는 에이전트는 run만 구현하면 된다.

v1의 한계 (Phase C-2가 드러낸 것)

  • 그래프가 외부에 노출되지 않아 라우터·실험 프레임워크가 노드별 관찰성·HITL을 활용할 수 없다.
  • stream이 토큰 단위 yield라 LangGraph의 astream_events(노드 시작/종료, 도구 호출, LLM 토큰)를 그대로 흘려보낼 수 없다.
  • Checkpointer가 에이전트별 옵션으로 노출되지 않아 운영자가 thread 정책을 설정할 수단이 없다.

3 v2 계약 설계

v1을 그대로 두고 세 가지 능력을 더한다.

from abc import ABC, abstractmethod
from typing import Iterator, AsyncIterator, Any, Optional
from langgraph.graph.state import CompiledStateGraph
from langgraph.checkpoint.base import BaseCheckpointSaver


class BaseAgent(ABC):
    name: str = ""
    domain: Optional[str] = None

    # v2 추가 — LangGraph 기반 에이전트는 모두 채워 둔다
    state_schema: type[Any] = None              # TypedDict 클래스 (예: QnaState)
    graph: Optional[CompiledStateGraph] = None  # compile()된 그래프
    checkpointer: Optional[BaseCheckpointSaver] = None

    # ── v1 계약 (그대로 유지) ──
    @abstractmethod
    def run(self, query: Query) -> Response:
        ...

    def stream(self, query: Query) -> Iterator[StreamEvent]:
        response = self.run(query)
        yield StreamEvent(type="done", response=response)

    # ── v2 신규 ──
    async def astream_events(
        self,
        query: Query,
        config: Optional[dict] = None,
    ) -> AsyncIterator[dict]:
        """LangGraph astream_events 직통.

        그래프가 없으면 stream()을 wrap해 호환 이벤트로 변환한다.
        """
        if self.graph is None:
            for ev in self.stream(query):
                yield {"event": "stream_event", "data": ev.model_dump()}
            return
        async for ev in self.graph.astream_events(
            {"query": query}, config=config, version="v2"
        ):
            yield ev

    def get_state(self, config: dict):
        """HITL 사이클의 'get_state' 진입점."""
        if self.graph is None:
            raise NotImplementedError("v1 에이전트는 get_state를 지원하지 않는다")
        return self.graph.get_state(config)

    def update_state(self, config: dict, values: dict, as_node: Optional[str] = None):
        if self.graph is None:
            raise NotImplementedError("v1 에이전트는 update_state를 지원하지 않는다")
        return self.graph.update_state(config, values, as_node=as_node)

설계 원칙 4가지

  1. 추가만, 제거 없음 — v1의 ABC·메서드는 그대로 유지한다. 기존 에이전트는 한 줄도 수정하지 않아도 된다.
  2. 신규 능력은 옵션 — graph 속성이 None이면 astream_events는 v1 stream을 그대로 흘린다. v1 에이전트가 v2 프레임워크에서 동작한다.
  3. HITL 진입점을 BaseAgent에 노출 — 라우터가 에이전트 종류와 무관하게 get_state/update_state를 호출할 수 있다.
  4. Pydantic 경계는 그대로 — Query/Response/StreamEvent는 변경 없음. State는 그래프 내부에서만 의미 있다.

4 v2 에이전트 구현 — QnaChatbot v2

14편에서 만든 빌더 함수를 BaseAgent v2에 결합한다.

from langgraph.checkpoint.memory import MemorySaver


class QnaChatbotAgentV2(BaseAgent):
    name = "qna_chatbot"
    domain = "standardization"
    state_schema = QnaState

    def __init__(
        self,
        config: RAGConfig,
        documents: dict[str, Path],
        reranker: str = "cosine",
        checkpointer: Optional[BaseCheckpointSaver] = None,
    ):
        self.config = config
        self.documents = documents
        self.reranker = reranker
        self.checkpointer = checkpointer or MemorySaver()
        self._inner = QnaChatbotAgent(documents=documents, config=config)  # v1 인스턴스 재사용
        self.graph = build_qna_graph(self._inner, reranker=reranker).with_config(
            checkpointer=self.checkpointer
        )

    def run(self, query: Query) -> Response:
        config = {"configurable": {"thread_id": query.user_id or "anon"}}
        final_state = self.graph.invoke({"query": query}, config=config)
        return final_state["response"]

    def stream(self, query: Query) -> Iterator[StreamEvent]:
        """v1 호환 stream — astream_events에서 토큰만 추출해 동기 변환."""
        # 단순 구현: 동기 graph.stream()으로 노드 단위 출력 후 done 이벤트 합성
        config = {"configurable": {"thread_id": query.user_id or "anon"}}
        final_state = None
        for chunk in self.graph.stream({"query": query}, config=config):
            for node_name, output in chunk.items():
                if "answer" in output and output["answer"]:
                    yield StreamEvent(type="token", text=output["answer"])
                if "response" in output:
                    final_state = output
        if final_state and final_state.get("response"):
            yield StreamEvent(type="done", response=final_state["response"])

run/stream을 v1 시그니처로 유지하므로 라우터·기존 호출자는 변경되지 않는다. 동시에 graph/get_state/update_state/astream_events가 새로 노출된다.

5 v1 → v2 어댑터 패턴

기존 v1 에이전트(QnaChatbotAgent, DataStandardizerAgent)를 그대로 두면서 v2 인터페이스를 빌려 주는 가벼운 어댑터를 정의한다.

class LegacyAgentAdapter(BaseAgent):
    """v1 에이전트를 BaseAgent v2 형태로 노출.

    graph/state_schema/checkpointer는 None — v2 신규 메서드는 NotImplementedError 또는
    호환 가능한 폴백을 반환한다.
    """

    def __init__(self, inner: BaseAgent):
        self._inner = inner
        self.name = inner.name
        self.domain = inner.domain

    def run(self, query: Query) -> Response:
        return self._inner.run(query)

    def stream(self, query: Query) -> Iterator[StreamEvent]:
        return self._inner.stream(query)

    # graph는 None — astream_events는 BaseAgent 기본 구현이 stream()을 wrap

    def get_state(self, config):
        raise NotImplementedError(f"v1 에이전트 '{self.name}'는 HITL을 지원하지 않는다")

    def update_state(self, config, values, as_node=None):
        raise NotImplementedError(f"v1 에이전트 '{self.name}'는 HITL을 지원하지 않는다")

이 어댑터로 기존 에이전트가 v2 라우터에 그대로 등록된다. 라우터는 어댑터 등 모든 에이전트를 동일 인터페이스로 다루며, HITL 기능 호출 시점에 NotImplementedError로 거른다.

6 FastAPI 라우터 변경

04편 — FastAPI 서빙 레이어에서 본 라우터에 v2 엔드포인트를 더한다. 기존 엔드포인트는 그대로 유지한다.

from fastapi import APIRouter, HTTPException

router = APIRouter()


# v1 — 그대로
@router.post("/agents/{agent_name}/run")
def run(agent_name: str, req: RunRequest) -> RunResponse:
    agent = get_agent_for_user(agent_name, req.user_id)
    response = agent.run(_build_query(req, agent_name))
    return RunResponse(response=response, ...)


# v2 — 신규: HITL 사이클
@router.get("/agents/{agent_name}/state")
def get_state(agent_name: str, thread_id: str):
    agent = registry.get(agent_name)
    config = {"configurable": {"thread_id": thread_id}}
    try:
        snapshot = agent.get_state(config)
    except NotImplementedError as e:
        raise HTTPException(status_code=501, detail=str(e))
    return {"values": snapshot.values, "next": snapshot.next}


@router.post("/agents/{agent_name}/update_state")
def update_state(agent_name: str, req: UpdateStateRequest):
    agent = registry.get(agent_name)
    config = {"configurable": {"thread_id": req.thread_id}}
    try:
        agent.update_state(config, req.values, as_node=req.as_node)
    except NotImplementedError as e:
        raise HTTPException(status_code=501, detail=str(e))
    return {"ok": True}


@router.post("/agents/{agent_name}/resume")
def resume(agent_name: str, req: ResumeRequest):
    """interrupt에서 멈춘 그래프를 재개."""
    agent = registry.get(agent_name)
    config = {"configurable": {"thread_id": req.thread_id}}
    if agent.graph is None:
        raise HTTPException(status_code=501, detail="v1 agent doesn't support resume")
    final_state = agent.graph.invoke(None, config=config)
    return {"response": final_state.get("response")}

핵심 — 기존 /run 엔드포인트는 변경되지 않는다. v1 어댑터를 끼운 v1 에이전트도 같은 라우터에서 동작한다. v2 신규 엔드포인트는 HITL을 지원하는 에이전트에서만 의미 있는 응답을 낸다.

7 A/B 실험 호환

06편 — A/B 실험 프레임워크의 _agent_cache[(experiment, arm)] 패턴은 v2에서도 그대로 쓴다. 단, 캐시되는 객체가 BaseAgent v2 인스턴스라는 점만 다르다.

def get_agent_for_user(agent_name: str, user_id: str) -> BaseAgent:
    config, exp_name, arm_id, overrides = resolve_config_for_user(agent_name, user_id)
    cache_key = (exp_name, arm_id)

    with _cache_lock:
        if cache_key in _agent_cache:
            return _agent_cache[cache_key]

        # arm 별 reranker 결정 → v2 빌드
        reranker = overrides.get("reranker", "cosine")
        agent = QnaChatbotAgentV2(
            config=config,
            documents=DOCUMENTS,
            reranker=reranker,
            checkpointer=PostgresSaver.from_conn_string(POSTGRES_URL),
        )
        _agent_cache[cache_key] = agent
        return agent

A/B arm 간 차이가 reranker 선택뿐이라면 v2 빌더 함수의 reranker 인자만 바뀐다. 같은 캐시 정책, 같은 sticky_hash 결정이 그대로 적용된다.

8 v2의 새 능력 — Subgraph와 위임 준비

15편에서 본 Subgraph 패턴이 v2 계약과 자연스럽게 어울린다.

class RetrievalAgent(BaseAgent):
    name = "retrieval"
    state_schema = RetrievalState

    def __init__(self, config: RAGConfig):
        self.graph = build_retrieval_subgraph(_inner_retriever(config))

    def run(self, query: Query) -> Response:
        result = self.graph.invoke({"query_text": query.text})
        return Response(text="", citations=docs_to_citations(result["parent_docs"]))


class QnaChatbotAgentV2(BaseAgent):
    name = "qna_chatbot"
    state_schema = QnaState

    def __init__(self, retrieval_agent: RetrievalAgent, ...):
        self._retrieval = retrieval_agent
        # 상위 그래프가 retrieval_agent.graph를 sub-node로 호출
        self.graph = build_qna_with_retrieval_subgraph(self._retrieval.graph, ...)

무엇이 가능해지는가

  • 검색 책임이 별도 에이전트로 분리되어 다른 에이전트(Data Standardizer, 미래의 Plan-and-Execute Agent)가 재사용한다.
  • BaseAgent v2 인스턴스가 다른 BaseAgent v2 인스턴스의 graph를 노드처럼 호출할 수 있다 — Phase C-3의 에이전트 위임(C14)이 이 패턴 위에서 작성된다.

9 마이그레이션 전략 — 점진적 전환

v1 코드 베이스를 한 번에 v2로 갈아엎지 않는다. 다음 순서를 권장한다.

단계 작업 영향
1 BaseAgent v2 정의 추가 (v1 메서드 그대로) 기존 에이전트 0줄 변경
2 LegacyAgentAdapter 작성 + 라우터 등록 기존 동작 동일
3 v2 신규 엔드포인트(get_state/update_state/resume) 추가 — v1 에이전트는 501 반환 기존 호출자 영향 없음
4 QnaChatbotAgentV2 구현 (14~16편 결과물 합치기) 신규 클래스, v1과 병행 운영
5 A/B 실험으로 v1 vs v2 비교 (응답 품질·지연·비용) 일부 트래픽으로 안전 검증
6 단계적 트래픽 전환 → v1 에이전트 deprecation 운영 메트릭 기반 결정

각 단계에서 롤백이 가능하다는 점이 중요하다. v1과 v2가 같은 BaseAgent 인터페이스로 라우터에 공존한다.

10 자주 발생하는 오류 패턴

WRONG:

# graph만 있고 checkpointer가 없는데 HITL을 시도
self.graph = build_qna_graph(agent)        # checkpointer 미주입
self.graph.invoke({"query": q}, config={"configurable": {"thread_id": "t1"}})
self.graph.get_state(config)               # 항상 None — 저장된 게 없음

CORRECT:

self.graph = build_qna_graph(agent, checkpointer=PostgresSaver(...))
# 또는 compile 시점에 명시
self.graph = graph.compile(checkpointer=PostgresSaver(...))

HITL은 Checkpointer 없이는 동작하지 않는다. v2 에이전트가 get_state를 지원하려면 그래프 컴파일 시점에 checkpointer를 주입한다.

WRONG:

# v2 에이전트의 stream을 동기 graph.stream로만 구현
def stream(self, query):
    for chunk in self.graph.stream({"query": query}):
        ...           # FastAPI async 환경에서 이벤트 루프 차단

CORRECT:

async def astream(self, query):
    async for ev in self.graph.astream_events(
        {"query": query}, version="v2",
    ):
        yield ev

FastAPI의 SSE 엔드포인트는 async 제너레이터를 기대한다. 그래프의 비동기 메서드(astream/astream_events)를 사용해 이벤트 루프 차단을 피한다.

WRONG:

class MyAgentV2(BaseAgent):
    name = "my_agent"
    # state_schema 명시 안 함 — 자동완성·정적 검사 작동 안 함

CORRECT:

class MyAgentV2(BaseAgent):
    name = "my_agent"
    state_schema = MyAgentState        # TypedDict 클래스

v2 에이전트의 state_schema는 라우터·디버거·문서 생성 도구가 State 타입을 알 수 있게 하는 메타데이터다. 누락해도 동작은 하지만 운영 도구의 가시성이 떨어진다.

11 Phase C-3 예고 — Agentic Mode

Phase C-2(13~16편 + 본 글)는 Chain → Graph 전환을 다뤘다. 본 글은 번호상 02-1로 Phase B의 v1 계약 옆에 놓여 있지만, 내용상 Phase C-2의 마무리이며 13~16편의 결과물을 BaseAgent 계약으로 묶는다. Phase C-3는 한 단계 더 나아가 “에이전트가 스스로 도구를 선택하고 멀티스텝 추론을 수행하는” 자율성을 부여한다.

편 주제 상태
C11 Tool Binding — 정적 파이프라인에서 동적 도구 선택으로 작성 완료
C12 ReAct 루프 — Reasoning + Acting 반복 작성 예정
C13 멀티스텝 플래닝 — Plan-and-Execute 작성 예정
C14 에이전트 위임 — Supervisor → 하위 에이전트 작성 예정

C11~C14는 모두 BaseAgent v2 위에서 작성된다. 본 글에서 만든 계약이 그 토대다.

12 정리

항목 핵심
v2 추가 속성 state_schema(TypedDict), graph(CompiledStateGraph), checkpointer
v2 신규 메서드 astream_events, get_state, update_state
하위 호환 v1 ABC·메서드 그대로 — 0줄 수정
LegacyAgentAdapter v1 에이전트를 v2 인터페이스로 노출, HITL은 501
FastAPI 변경 /run 유지, /state//update_state//resume 추가
A/B 호환 _agent_cache[(experiment, arm)] 그대로, arm별 v2 빌더 호출
마이그레이션 6단계 점진 전환, 각 단계 롤백 가능

BaseAgent v2는 새 인터페이스가 아니다. v1을 깨지 않으면서 LangGraph 시대의 능력을 더한 확장이다. 50명·1000명 단계로 가는 길에서 모든 에이전트가 같은 모양이라는 것은 협업·운영·실험의 전제 조건이다.

13 Phase C-2 마무리

Phase C-2 다섯 편이 끝났다. Phase B에서 만든 Chain 기반 RAG가 LangGraph StateGraph + Checkpointer + HITL을 갖춘 v2 BaseAgent로 진화하는 길을 코드 단위로 그렸다. 다음은 Phase C-3 — 에이전트가 도구를 들고 스스로 추론하는 단계다.

14 관련 주제

선행 학습 (Phase C-2 전체)

Phase B 연결

Phase C-3 예고

  • Tool Binding
  • ReAct 루프 (작성 예정)
  • 멀티스텝 플래닝 (작성 예정)
  • 에이전트 위임 (작성 예정)

다른 카테고리 연결

Subscribe

Enjoy this blog? Get notified of new posts by email: