Kwangmin Kim - MINERVA State 설계

1 분해 후에 만나는 진짜 문제

14편에서 _prepare()를 7개 노드로 분해했다. 분해는 끝이 아니라 시작이다 — 노드 사이를 흘러다닐 자료 구조(State)가 잘 설계되어 있어야 분해의 실효가 나타난다.

State가 잘못 설계되면 다음 문제가 생긴다.

누적되어야 할 필드가 덮어써진다 — 첫 노드의 timing이 두 번째 노드 호출 시 사라진다.
덮어써야 할 필드가 누적된다 — answer가 재시도 루프에서 이전 답변에 append되어 버린다.
Pydantic 객체와 dict가 섞인다 — Query는 Pydantic, chain_inputs는 dict — 어떤 필드는 .text, 어떤 필드는 ["context"] — 일관성이 깨진다.
State가 부풀어 오른다 — 50개 필드짜리 State는 사실상 글로벌 변수다. 노드의 책임 경계가 흐려진다.

이번 편은 위 네 가지 함정을 피하는 State 설계 패턴을 정리한다.

선행 학습

MINERVA LangGraph 기초 — Node·Edge·State 개념
MINERVA RAG Chain 분해 — 7 노드 분해와 최소 State
MINERVA 상태 관리 해부 — 4계층 상태 분포 (React·localStorage·_agent_cache·JSONL)

2 State의 두 가지 합치기 규칙

LangGraph는 노드가 반환한 dict를 이전 State에 “합쳐서” 다음 노드로 넘긴다. 합치기 규칙은 필드별로 정의되며, Annotated 타입의 두 번째 인자로 표현한다.

from typing import Annotated, TypedDict
from operator import add


class AgentState(TypedDict):
    answer: str                              # reducer 미지정 → 덮어쓰기
    log_messages: Annotated[list[str], add]  # reducer = add → 누적

규칙

reducer 미지정 → 새 값으로 덮어쓰기
reducer 지정 → reducer(prev, new)의 반환값으로 합치기

add는 operator.add 그대로다 — list에 적용하면 concat, int에 적용하면 합산.

3 4가지 핵심 reducer 패턴

3.1 덮어쓰기 — 최신 값만 의미 있는 필드

class State(TypedDict):
    answer: str           # 답변은 매번 새로 생성됨
    response: Response    # 최종 응답도 한 번만 의미
    chain_inputs: dict    # 새 노드 호출마다 재조립

가장 단순한 패턴. 재시도 루프에서도 “최신 답변 = 최종 답변”이라면 이 형태가 맞다.

3.2 누적 — 이력이 의미 있는 필드

from operator import add

class State(TypedDict):
    log_messages: Annotated[list[str], add]
    timings: Annotated[list[dict], add]
    tool_calls: Annotated[list[dict], add]

add는 [a, b] + [c]를 [a, b, c]로 만든다. 모든 노드의 로그·timing·도구 호출 이력이 자동으로 쌓인다.

def node_a(state):
    return {"log_messages": ["a finished"]}        # 이전 [] + ["a finished"] → ["a finished"]

def node_b(state):
    return {"log_messages": ["b finished"]}        # 이전 ["a finished"] + ["b finished"]

3.3 병합 — dict 합치기

from operator import or_      # Python 3.9+ dict union

class State(TypedDict):
    metadata: Annotated[dict, or_]

여러 노드가 같은 dict의 다른 키를 채울 때 쓴다. {"a": 1} | {"b": 2} → {"a": 1, "b": 2}. 같은 키가 충돌하면 오른쪽(새 값)이 이긴다.

3.4 커스텀 reducer — 도메인 규칙이 있을 때

reducer는 임의 함수다. 시그니처는 (prev, new) -> merged.

def merge_unique_docs(prev: list[Document], new: list[Document]) -> list[Document]:
    """parent_id 기준 중복 제거하며 누적."""
    seen = {d.metadata.get("parent_id") for d in prev}
    return prev + [d for d in new if d.metadata.get("parent_id") not in seen]


class State(TypedDict):
    parent_docs: Annotated[list[Document], merge_unique_docs]

검색 실패 폴백 후 결과를 합칠 때 같은 문서가 두 경로에서 들어오는 것을 방지한다. 이런 도메인 규칙은 14편에서 본 web_search와 map_to_parents 결과 합치기에 쓸 수 있다.

reducer 선택 결정 기준

한 번만 의미 있는 결과? → 덮어쓰기 (default)
노드별로 이어붙이고 싶은 이력? → add (list)
키별로 한 노드씩 채우는 dict? → or_
도메인 규칙이 있는 합치기? → 커스텀 함수

4 MINERVA QnaState — 실전 설계

14편에서는 분해에 필요한 최소 State만 두었다. 운영 단계에서 필요한 필드를 더해 본격 설계로 확장한다.

from typing import Annotated, TypedDict, Optional
from operator import add
from langchain_core.documents import Document
from core.contracts import Query, Response


class QnaState(TypedDict):
    # 입력 — 한 번만 채워짐
    query: Query

    # 검색 단계 — 최신 결과만 의미 있음
    child_docs: list[Document]
    parent_docs: list[Document]

    # 생성 단계 — 재시도 시 덮어씀
    chain_inputs: dict
    answer: str

    # 평가/제어 — 재시도 루프 카운터, 점수
    citation_count: int
    relevance_score: float
    retry_count: int

    # 출력 — 최종 한 번만
    response: Optional[Response]

    # 관찰성 — 누적
    timings: Annotated[list[dict], add]
    log_messages: Annotated[list[str], add]

    # 에러 — 누적 (모든 단계의 예외를 모음)
    errors: Annotated[list[dict], add]

이 State에는 네 가지 설계 결정이 들어 있다.

4.1 결정 1 — Pydantic 객체를 State 필드로 그대로 보관

Query와 Response는 Pydantic이다. State에 그대로 두면 두 가지 이점이 생긴다.

타입 안전성: 노드 안에서 state["query"].text로 접근하며 IDE 자동완성·정적 검사가 작동한다.
직렬화 일관성: Checkpointer가 State를 JSON으로 직렬화할 때 Pydantic이 알아서 처리한다.

dict로 풀어 두는 안티패턴은 피한다.

# 안티패턴 — 필드를 풀어서 dict로
class BadState(TypedDict):
    query_text: str
    query_history: list[dict]
    query_user_id: str
    query_session_id: str
    # ... Query의 모든 필드를 풀어 놓음

# 권장 — Pydantic 객체 그대로
class GoodState(TypedDict):
    query: Query

Query 스키마가 바뀔 때 전자는 State 정의도 함께 수정해야 한다. 후자는 Pydantic 정의만 바꾸면 된다.

4.2 결정 2 — `chain_inputs`는 덮어쓰기

chain_inputs는 매 노드 호출마다 재조립되는 단명 자료다. 누적할 이유가 없다. 재시도 루프에서 새 컨텍스트로 다시 만들어진다.

4.3 결정 3 — `errors`는 누적

여러 단계에서 발생한 예외를 한 곳에 모아 두면 후처리가 단순해진다. build_response_node가 마지막에 state["errors"]를 검사해 일부 실패에도 polite한 응답을 조립할 수 있다.

def safe_node(node_name, fn):
    def wrapped(state):
        try:
            return fn(state)
        except Exception as e:
            return {"errors": [{
                "node": node_name,
                "type": type(e).__name__,
                "message": str(e),
            }]}
    return wrapped

이 데코레이터를 노드에 감싸면 한 단계 실패가 전체 그래프를 멈추지 않는다. 자세한 폴백 흐름은 16편 — Checkpointing과 HITL에서 다룬다.

4.4 결정 4 — 도구 이력은 messages 패턴 검토

QnaChatbot은 도구 호출이 단순하지만, Data Standardizer나 추후 Agentic Mode에서는 도구 호출 이력이 중요해진다. LangGraph가 제공하는 MessagesState 패턴이 이때 맞다.

from langgraph.graph import MessagesState
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage


class AgenticState(MessagesState):
    """MessagesState는 messages: Annotated[list[BaseMessage], add_messages]를 미리 갖고 있다.

    add_messages reducer는 단순 add보다 똑똑해서 같은 id의 메시지가 들어오면 덮어쓴다
    (재생성·수정 시나리오 지원).
    """
    parent_docs: list[Document]
    retry_count: int

add_messages는 단순 add와 다르다 — id 기반 dedup·교체가 가능해 ReAct 루프에서 도구 호출의 수정 시나리오를 자연스럽게 다룬다. ReAct·도구 호출 패턴은 Phase C-3(C11~C14)에서 본격적으로 다룬다.

5 대화 히스토리는 어디에 두는가

Query.history는 입력으로 받은 이전 대화 턴이다. State에 두는 방식은 두 갈래다.

5.1 방식 A — Query 안에 그대로 보관 (현재)

class QnaState(TypedDict):
    query: Query          # query.history 안에 보관

장점: 단순. 노드는 state["query"].history로만 접근.
단점: 새 턴이 추가될 때 Query 객체를 재생성해야 함. 노드가 입력을 변경하는 효과.

5.2 방식 B — State에 별도 필드로 분리

from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage

class QnaState(TypedDict):
    query: Query                                       # 텍스트·메타만
    messages: Annotated[list[BaseMessage], add_messages]

장점: 새 턴을 노드가 자연스럽게 append할 수 있음. 멀티턴 대화에 강함.
단점: query.history와 state["messages"] 두 군데에 정보가 흩어질 위험.

현재 MINERVA(QnaChatbot)는 단일 요청-응답 패턴이라 방식 A로 충분하다. Phase C-3에서 Plan-and-Execute나 ReAct로 가면 방식 B가 자연스러워진다. 두 방식의 전환 비용은 State 정의 변경 + 노드 몇 개의 history 접근 경로 변경으로 제한된다.

6 State가 부풀어 오를 때 — Subgraph로 분리

State 필드가 30개를 넘기 시작하면 다음 신호다.

일부 필드는 일부 노드만 사용하고, 다른 노드는 무시한다.
“이 필드는 어느 단계에서 채워지는가”를 한눈에 알기 어려워진다.
새 기능을 추가할 때 State 변경의 파급이 크다.

해결은 Subgraph로 책임을 분리하는 것이다.

# 검색 책임만 가진 Subgraph
class RetrievalState(TypedDict):
    query_text: str
    child_docs: list[Document]
    parent_docs: list[Document]
    relevance_score: float


def build_retrieval_subgraph(agent):
    g = StateGraph(RetrievalState)
    g.add_node("retrieve_children", make_retrieve_children_node(agent))
    g.add_node("rerank", make_rerank_cosine_node(agent))
    g.add_node("map_to_parents", make_map_to_parents_node(agent))
    g.add_edge(START, "retrieve_children")
    g.add_edge("retrieve_children", "rerank")
    g.add_edge("rerank", "map_to_parents")
    g.add_edge("map_to_parents", END)
    return g.compile()


# 상위 그래프
class QnaState(TypedDict):
    query: Query
    parent_docs: list[Document]
    chain_inputs: dict
    answer: str
    response: Optional[Response]


def call_retrieval_subgraph(retrieval_app):
    def node(state: QnaState) -> dict:
        sub_input = {"query_text": state["query"].text}
        sub_result = retrieval_app.invoke(sub_input)
        return {"parent_docs": sub_result["parent_docs"]}
    return node

상위 State는 검색 내부의 child·rerank·score를 알 필요가 없다. 검색 책임은 RetrievalState에 캡슐화된다.

Subgraph 분리 기준 — 다음 중 둘 이상이면 분리를 고려한다.

그 영역의 필드가 6개를 넘는다.
그 영역의 노드가 3개를 넘는다.
그 영역만 독립 테스트하고 싶다 (다른 영역의 mock 없이).
그 영역을 재사용하는 다른 에이전트가 있다.

Data Standardizer State는 분리 기준을 가장 잘 만족한다

QnA Chatbot이 평탄한 단일 그래프에 적합한 반면, Data Standardizer는 이미 4개 sub_agent(RagRecommender·DomainAuditor·post_processing/{tables,code})가 직렬·조건부로 호출되는 구조다(02·08편 참조). State 설계 시 다음 영역들이 셋 다 분리 기준을 만족한다:

추천 (Recommendation): query_text, mode, child_docs, parent_docs, recommendation_md, cited_docs (필드 6) — RetrievalSubgraph로 분리
후처리 (Post-Processing): recommendation_md, domain_cache, physical_names, processed_table_md, mode(분기) (필드 5) — 도메인 캐시는 인스턴스 외부 영속화 후보
감사 (Audit): processed_table_md, priority_yml, corrections, audited_md, audit_skip_reason (필드 5) — DomainAuditor만 캡슐화하면 LLM 호출이 선택적으로 격리됨

세 sub-graph를 각각 분리하면 부모 State는 query/final_text/response 같은 인터페이스 필드만 남아 6개 이내로 유지된다. 이 패턴이 21편 — 에이전트 위임에서 다룰 supervisor 패턴의 자료 구조 출발점이다.

7 State와 Pydantic의 역할 분담

질문 — “왜 State를 Pydantic이 아니라 TypedDict로 정의하는가?”

관점	TypedDict	Pydantic BaseModel
역할	노드 간 자료 흐름 (런타임)	외부 경계 검증 (HTTP, 디스크)
검증	없음 (정적 검사만)	런타임 type coercion + validation
reducer 지원	`Annotated[..., reducer]` 일급	LangGraph가 직접 지원 안 함
직렬화	수동	자동 (`.model_dump()`)
비용	가벼움	검증·생성 오버헤드

원칙

외부 경계(HTTP, 디스크, 사용자 입력)는 Pydantic으로 검증한다 — Query, Response, RAGConfig.
그래프 내부의 노드 간 자료 흐름은 TypedDict로 정의한다 — QnaState, RetrievalState.
State 필드 안에 Pydantic 객체를 보관해도 된다 — 외부에서 검증된 값을 노드들이 그대로 사용한다.

이 분담은 11편 — Config 의존성 추적에서 본 RAGConfig의 Pydantic 사용과 자연스럽게 어울린다.

8 자주 발생하는 오류 패턴

WRONG:

class State(TypedDict):
    answer: Annotated[str, add]    # str + str = "이전답변새답변" (의도와 다름)

def generate(state):
    return {"answer": "최종 답변"}

CORRECT:

class State(TypedDict):
    answer: str                     # 덮어쓰기

def generate(state):
    return {"answer": "최종 답변"}

str·int 같은 단일 값에 add를 reducer로 두면 누적이 일어나 의도와 다르게 동작한다. 누적이 정말 의미 있을 때만 reducer를 둔다.

WRONG:

def append_log(state):
    state["log_messages"].append("done")    # 입력 list 직접 변경
    return {"log_messages": state["log_messages"]}

CORRECT:

def append_log(state):
    return {"log_messages": ["done"]}        # 새 list로 반환 — reducer가 누적

Annotated[list, add] reducer가 prev + new를 수행한다. 노드는 “이번에 추가할 항목”만 list로 반환하면 된다. 입력 list를 직접 변경하면 reducer가 중복 누적할 수 있다.

WRONG:

from pydantic import BaseModel

class State(BaseModel):                # LangGraph는 BaseModel을 State로 직접 받지 않음
    query: Query
    answer: str

CORRECT:

class State(TypedDict):                 # TypedDict 사용
    query: Query                        # 필드는 Pydantic이어도 됨
    answer: str

LangGraph의 StateGraph는 TypedDict 기반이다. State 컨테이너 자체는 TypedDict로 두고, 필드 안에 Pydantic 객체를 보관한다.

9 다음 편 예고 — Checkpointing과 HITL

State가 잘 정의되면 다음 단계는 “그 State를 어떻게 영속화하고, 사람이 끼어들 지점을 어디에 둘 것인가”이다. 다음 편 16편에서 다룬다.

Checkpointer로 State를 SQLite·Postgres에 저장
interrupt_before/interrupt_after로 노드 직전·직후 멈추기
멈춘 그래프를 사람의 승인·수정 후 재개하기
A/B 실험 분기를 그래프 차원에서 처리

10 정리

항목	핵심
합치기 규칙	reducer 미지정=덮어쓰기, 지정=`reducer(prev, new)`
4 패턴	덮어쓰기 / `add` 누적 / `or_` dict 병합 / 커스텀 함수
Pydantic 보관	State 컨테이너는 TypedDict, 필드 안에 Query·Response 등 보관
히스토리	단일 요청은 `Query.history`, 멀티턴/Agentic은 `MessagesState`
Subgraph 분리	필드 6+, 노드 3+, 독립 테스트, 재사용 — 둘 이상이면 분리
State vs Pydantic	외부 경계는 Pydantic, 내부 흐름은 TypedDict

State는 분해된 노드를 잇는 “계약”이다. 계약이 명확하면 분해의 비용이 회수되고, 부풀면 그래프는 다시 거대 함수와 다를 바 없어진다. MINERVA의 RAG는 7 노드 + 15 필드 안쪽에서 운영 가능하며, Agentic 모드로 갈 때 MessagesState와 Subgraph로 분리하면 된다.

11 관련 주제

선행 학습

MINERVA LangGraph 기초
MINERVA RAG Chain 분해
MINERVA 상태 관리 해부 — 4계층 상태 분포
MINERVA Config 의존성 추적 — Pydantic 기반 외부 경계 검증

후속 주제

다른 카테고리 연결

1 분해 후에 만나는 진짜 문제

2 State의 두 가지 합치기 규칙

3 4가지 핵심 reducer 패턴

3.1 덮어쓰기 — 최신 값만 의미 있는 필드

3.2 누적 — 이력이 의미 있는 필드

3.3 병합 — dict 합치기

3.4 커스텀 reducer — 도메인 규칙이 있을 때

4 MINERVA QnaState — 실전 설계

4.1 결정 1 — Pydantic 객체를 State 필드로 그대로 보관

4.2 결정 2 — chain_inputs는 덮어쓰기

4.3 결정 3 — errors는 누적

4.4 결정 4 — 도구 이력은 messages 패턴 검토

5 대화 히스토리는 어디에 두는가

5.1 방식 A — Query 안에 그대로 보관 (현재)

5.2 방식 B — State에 별도 필드로 분리

6 State가 부풀어 오를 때 — Subgraph로 분리

7 State와 Pydantic의 역할 분담

8 자주 발생하는 오류 패턴

9 다음 편 예고 — Checkpointing과 HITL

10 정리

11 관련 주제

4.2 결정 2 — `chain_inputs`는 덮어쓰기

4.3 결정 3 — `errors`는 누적