MINERVA 데이터 흐름 추적

질문에서 답변까지 — 타입과 변환의 여정

MINERVA에서 사용자 질문이 최종 답변이 되기까지 거치는 모든 변환 단계를 추적한다. RunRequest 파싱 → 실험 라우팅 → Query 변환 → RAG(검색·리랭크·Parent 매핑) → LCEL 체인 → Response 조립 → JSON → React 렌더링의 각 경계에서 데이터 타입과 shape가 어떻게 바뀌는지, SSE 스트리밍 경로는 어떻게 다른지 실제 코드로 정리한다.

Agent
저자

Kwangmin Kim

공개

2026년 05월 05일

1 왜 데이터 흐름을 추적하는가

Phase B에서 MINERVA의 각 레이어를 구현하고 배포했다. 그런데 실제 운영에서 문제가 생겼을 때 “어느 단계에서 데이터가 어떻게 생겼는지”를 몰라 디버깅하는 데 시간이 걸렸다.

데이터 흐름 추적은 세 가지 질문에 답한다.

  1. 타입 경계: 각 레이어 사이에서 데이터 타입이 무엇으로 바뀌는가
  2. shape 변환: 하나의 질문이 검색 결과 여러 개로, 다시 하나의 답변으로 압축되는 과정
  3. 분기점: 동기 호출(JSON 응답)과 스트리밍 호출(SSE)이 어디서 갈라지는가

이 추적이 명확해야 Phase C-2에서 각 단계를 LangGraph Node로 분해할 수 있다.

선행 학습

2 전체 데이터 흐름 지도

브라우저 (React)
│  POST /agents/qna_chatbot/run
│  body: { "text": "...", "history": [...], "user_id": "..." }  # JSON
│
▼
FastAPI (routers/qna_chatbot.py)
│  RunRequest { text, history, agent_params, user_id, session_id }  # Pydantic
│
▼  resolve_config_for_user()  ← A/B 실험 arm 결정
│  (config, exp_name, arm_id, overrides)
│
▼  _agent_cache[(exp_name, arm_id)]  ← thread-safe 캐시
│  QnaChatbotAgent 인스턴스
│
▼  _build_query()
│  Query { text, history, agent_params, user_id, session_id, experiment_id, arm_id }
│
▼  agent.run(query)  /  agent.stream(query)
│       │
│       ▼  agent._prepare()  (run·stream 공유)
│       │
│       │  _ensure_initialized()  ← prompt·llm lazy 로드
│       │
│       │  _retrieve_docs(query.text, document_id)
│       │      str → list[Document]  (child chunks, k*2~k*3개)
│       │          │ hybrid_search / similarity_search / mmr
│       │          ▼ _rerank_children()  또는  _compute_cosine_scores()
│       │      list[Document] (child, 재정렬 + score 부여)
│       │          ▼ _map_to_parents_with_child_score()
│       │      list[Document] (parent chunks, k개)
│       │
│       │  _build_chain_inputs(query, docs)
│       │      list[Document] → str (context)
│       │      history       → str (chat_history)
│       │      → dict { context, chat_history, question }
│       │
│       ▼  LCEL chain: prompt | llm | StrOutputParser()
│           chain.invoke(chain_inputs)  → str (answer)
│           [or chain.stream() for streaming]
│
▼  _build_response(text, docs, query)
│  Response { text, citations, run_id, model, latency_ms, ttft_ms, arm_id, ... }
│
▼
RunResponse { response, experiment_id, arm_id }  → JSON
│
▼
브라우저 (React): JSON 역직렬화 → state 업데이트 → 렌더링

3 단계별 상세 추적

3.1 단계 1: HTTP 진입 — RunRequest 파싱

POST /agents/qna_chatbot/run으로 들어온 JSON 본문을 FastAPI가 RunRequest로 파싱한다.

# src/services/api/schemas.py
class RunRequest(BaseModel):
    text: str                                    # 사용자 질문 (NOT "query")
    history: list[ConversationTurn] = Field(default_factory=list)
    agent_params: dict[str, Any] = Field(default_factory=dict)
    user_id: Optional[str] = None               # A/B 실험 sticky hash 기준
    session_id: Optional[str] = None
# src/core/contracts.py
class ConversationTurn(BaseModel):
    role: str       # "user" | "assistant"
    content: str

파싱 후 데이터 형태:

속성 타입 설명
req.text str 사용자 질문
req.history list[ConversationTurn] 이전 대화 이력
req.agent_params dict 에이전트별 옵션 (예: document_id, source_filter, force_arm)
req.user_id str \| None sticky hash용 사용자 식별자

3.2 단계 2: 실험 라우팅 — A/B arm 결정

요청이 들어오면 가장 먼저 실험 설정을 결정한다. 사용자 ID 기반으로 어떤 RAG 설정(arm)을 사용할지 결정하고, 그 설정으로 만들어진 에이전트를 캐시에서 찾는다.

# src/services/api/routers/qna_chatbot.py
@router.post("/run", response_model=RunResponse)
def run(req: RunRequest) -> RunResponse:
    force_arm = req.agent_params.get("force_arm")
    agent, exp_name, arm_id, overrides = _build_agent(
        user_id=req.user_id,
        force_arm=force_arm,
    )
    ...
def _build_agent(user_id=None, force_arm=None):
    # 1. 실험 설정 결정
    config, exp_name, arm_id, overrides = resolve_config_for_user(
        "qna_chatbot", user_id=user_id, force_arm=force_arm,
    )

    # 2. (exp_name, arm_id) 조합으로 캐시 조회
    key = (exp_name, arm_id)
    agent = _agent_cache.get(key)

    if agent is None:
        with _agent_cache_lock:             # thread-safe double-checked locking
            if _agent_cache.get(key) is None:
                agent = QnaChatbotAgent(
                    documents=DEFAULT_DOCUMENTS,
                    documents_metadata=DEFAULT_DOCUMENTS_METADATA,
                    default_document=DEFAULT_DOCUMENT_ID,
                    config=config,          # arm별 RAGConfig
                )
                _agent_cache[key] = agent

    return agent, exp_name, arm_id, overrides

캐시 키가 (exp_name, arm_id) 튜플인 이유: 동일한 실험 arm에서는 같은 RAG 설정을 사용하므로 에이전트를 재사용할 수 있다. 에이전트를 새로 만들면 retriever, prompt, LLM의 lazy cache가 초기화되어 문서 재로딩이 발생한다.

3.3 단계 3: 도메인 모델 변환 — RunRequest → Query

서빙 레이어의 RunRequest를 에이전트 계약의 Query로 변환한다.

# src/services/api/routers/qna_chatbot.py
def _build_query(req: RunRequest, exp_name, arm_id) -> Query:
    return Query(
        text=req.text,
        history=req.history,
        agent_params=req.agent_params,
        user_id=req.user_id,
        session_id=req.session_id,
        experiment_id=exp_name,   # 서버가 결정한 실험 ID
        arm_id=arm_id,            # 서버가 결정한 arm ID
    )
# src/core/contracts.py
class Query(BaseModel):
    text: str
    history: list[ConversationTurn] = Field(default_factory=list)
    agent_params: dict[str, Any] = Field(default_factory=dict)
    user_id: Optional[str] = None
    session_id: Optional[str] = None
    experiment_id: Optional[str] = None    # 클라이언트는 모름 — 서버가 채움
    arm_id: Optional[str] = None           # 클라이언트는 모름 — 서버가 채움

RunRequest와 Query를 분리하는 이유: experiment_id, arm_id는 클라이언트가 지정하는 것이 아니라 서버의 실험 시스템이 결정한다. 서빙 스키마와 도메인 계약을 분리하면 REST 외 다른 진입점(gRPC, CLI, 테스트)에서도 동일한 Query 계약을 사용할 수 있다.

3.4 단계 4: agent._prepare() — RAG 파이프라인

run()과 stream() 모두 _prepare()를 먼저 실행한다. 이 메서드가 검색부터 chain 조립까지를 담당한다.

# src/agents/qna_chatbot/agent.py
def _prepare(self, query: Query):
    t0 = time.perf_counter()

    # 1. Lazy 초기화 (prompt·llm — 최초 1회만)
    self._ensure_initialized()

    # 2. 문서·필터 선택
    document_id = query.agent_params.get("document_id", self._default_document_id)
    source_filter = query.agent_params.get("source_filter")

    # 3. 검색 (str → list[Document])
    docs = self._retrieve_docs(query.text, document_id, source_filter=source_filter)

    # 4. Chain inputs 조립
    chain_inputs = self._build_chain_inputs(query, docs)

    # 5. LLM + LCEL chain 조립
    llm, model_used = self._get_llm_for_query(query)
    chain = self._prompt | llm | StrOutputParser()

    return docs, chain, chain_inputs, model_used

3.4.1 4-1: 검색 — str → list[Document]

def _retrieve_docs(self, question: str, document_id: str, source_filter=None):
    retriever = self._get_retriever(document_id)    # lazy 캐시
    search_type = self.config.retrieval.search_type
    k = self.config.retrieval.k
    fetch_k = k * 3 if source_filter and source_filter != "all" else k

    # search_type에 따라 검색 방식 분기
    if search_type == "hybrid" and provider == "azure":
        docs = retriever.hybrid_search(query=question, k=fetch_k)
    elif search_type == "mmr":
        docs = retriever.max_marginal_relevance_search(query=question, k=fetch_k, ...)
    else:
        docs = retriever.similarity_search(query=question, k=fetch_k)

    # source_filter 후처리 (메타데이터 기반 필터링)
    if source_filter and source_filter != "all":
        docs = self._filter_docs_by_source(docs, source_filter)[:k]

    return docs

retriever는 ParentChunkRetriever이다. 내부에서 child chunk로 검색하고, rerank 후 parent chunk로 매핑한다.

# src/core/rag/retriever.py
def hybrid_search(self, query: str, k: int = 4) -> list[Document]:
    # 1. Azure Search hybrid (BM25 + Vector) — child chunk k*2개
    child_docs, query_vec = self._hybrid_search_with_vectors(query=query, k=k * 2)

    # 2. Reranker 또는 Cosine 유사도 계산
    if reranker_type in ("flashrank", "cross_encoder"):
        child_docs = self._rerank_children(child_docs, query)
    elif embeddings is not None:
        child_docs = self._compute_cosine_scores(child_docs, query, query_vec=query_vec)

    # 3. child → parent 매핑 (child의 relevance_score를 parent에 전달)
    parents = self._map_to_parents_with_child_score(child_docs, k)
    return parents

데이터 shape 변환:

단계 입력 출력
검색 str (질문 1개) list[Document] (child, k2~k3개)
Rerank/Cosine list[Document] (child, 순서·점수 미정) list[Document] (child, relevance_score 부여)
Parent 매핑 list[Document] (child, n개) list[Document] (parent, k개)

Parent 매핑 로직: child의 metadata["parent_id"]로 parent_store를 조회한다. 같은 parent에 여러 child가 매칭되면 가장 높은 relevance_score를 parent에 유지하고, matched_child_contents 리스트에 모두 누적한다.

Citation.metadata에 외부 메타가 합류하는 지점

검색이 반환한 Document.metadata는 단순 {parent_id, source, ...} 수준이지만, _build_response()에서 Citation을 만들 때 core/rag/metadata_loader.py가 외부 yml(docs/metadata/sources.yml 등)에서 읽어둔 참고문헌·표준 메타(source_type, source_name, domain, authority_level)를 setdefault로 병합한다. 그래서 프론트엔드 참조 패널이 “표준화 도메인 / 1차 자료 / DAMA-DMBOK”처럼 출처 등급·도메인 라벨을 렌더링할 수 있다 — 이 메타는 인덱스에는 없고 yml에만 있다.

3.4.2 4-1-a: Azure 저수준 검색 — _hybrid_search_with_vectors()

LangChain AzureSearch.hybrid_search()는 응답에서 content_vector 필드를 제거한다(용량 이슈로 의도적). 그러나 cosine 재계산 시 child별 벡터를 재사용하면 embed_documents API 호출을 절약할 수 있다.

_hybrid_search_with_vectors()는 LangChain 래퍼를 우회해 Azure SDK의 client.search()를 직접 호출하고, raw 응답에서 content_vector를 꺼내 Document.metadata[_CHILD_VECTOR_KEY]에 임시로 싣는다.

# src/core/rag/retriever.py
_CHILD_VECTOR_KEY = "_child_vector_cache"

def _hybrid_search_with_vectors(self, query, k):
    embedding = vs.embed_query(query)
    query_vec = np.array(embedding)

    results = vs.client.search(
        search_text=query,
        vector_queries=[VectorizedQuery(
            vector=embedding,
            k_nearest_neighbors=k,
            fields=FIELDS_CONTENT_VECTOR,
        )],
        top=k,
    )

    docs = []
    for result in results:
        metadata = ...  # FIELDS_METADATA 파싱
        vector = result.get(FIELDS_CONTENT_VECTOR)
        if vector is not None:
            metadata[_CHILD_VECTOR_KEY] = vector  # 벡터 캐시
        docs.append(Document(page_content=result[FIELDS_CONTENT], metadata=metadata))

    return docs, query_vec  # query_vec도 재사용 위해 함께 반환

이 경로가 활성화되면 _compute_cosine_scores()는 embed_query와 embed_documents를 모두 생략한다.

def _compute_cosine_scores(self, child_docs, query, query_vec=None):
    if query_vec is None:
        query_vec = np.array(self._embeddings.embed_query(query))  # 생략 가능

    cached = [doc.metadata.get(_CHILD_VECTOR_KEY) for doc in child_docs]
    if all(v is not None for v in cached):
        chunk_vecs = np.array(cached)              # 캐시 사용
        doc_embed_source = "cached"
    else:
        chunk_texts = [doc.page_content for doc in child_docs]
        chunk_vecs = np.array(self._embeddings.embed_documents(chunk_texts))  # API 호출
        doc_embed_source = "api"

분기 매트릭스:

reranker_type embeddings 사용 경로 효과
flashrank/cross_encoder - 표준 래퍼 + reranker 벡터 불필요
(없음) 있음 low-level + cosine 캐시 API 호출 0회 (인덱스가 retrievable=True인 경우)
azure_semantic - semantic_hybrid_search_with_score rerankerScore 0~4 → 0~1 정규화

3.4.3 4-2: Context 조립 — list[Document] → str

def _build_context_string(self, docs) -> str:
    parts = []
    for i, doc in enumerate(docs, start=1):
        meta = getattr(doc, "metadata", {}) or {}
        source_name = meta.get("source_name")
        attribution = f" (출처: {source_name})" if source_name else ""
        parts.append(f"[Document {i}]{attribution}\n{doc.page_content}")
    return "\n\n".join(parts)

list[Document](k개) → str. LLM 프롬프트에 전달될 컨텍스트 블록이다. [Document 1] (출처: DAMA-DMBOK) 형식으로 출처를 prefix에 붙여 LLM이 답변에서 [1], [2] 인용 마커를 자연스럽게 사용하도록 유도한다.

3.4.4 4-3: Chain inputs 조립

def _build_chain_inputs(self, query: Query, docs) -> dict:
    context = self._build_context_string(docs)

    # 최근 N턴 history만 사용 (토큰 절약)
    keep_history = query.agent_params.get("keep_chat_history", True)
    history_turns = self.config.conversation.history_turns
    recent_history = query.history[-(history_turns * 2):]
    chat_history = self._format_history(recent_history, keep_history=keep_history)

    return {
        "context": context,
        "chat_history": chat_history,   # assistant 답변은 lead+tail 축약
        "question": query.text,
    }

LCEL chain은 chain_inputs dict를 프롬프트 템플릿 변수로 사용한다.

3.4.5 4-4: LCEL chain 실행 — str → str

chain = self._prompt | llm | StrOutputParser()

# run()의 경우
answer: str = chain.invoke(chain_inputs)

# stream()의 경우
for chunk in chain.stream(chain_inputs):
    yield StreamEvent(type="token", text=chunk)

LCEL 파이프라인 단계별 타입:

단계 입력 출력
prompt dict (chain_inputs) list[BaseMessage] (시스템+사용자 메시지)
llm list[BaseMessage] AIMessage
StrOutputParser() AIMessage str (content만 추출)

3.5 단계 5: Response 조립

def _build_response(self, text, docs, query, latency_ms=None, ttft_ms=None, model_used=None) -> Response:
    # 1. LLM이 말미에 추가하는 "참고:" 섹션 제거
    cleaned = self._clean_response_text(text)

    # 2. 검색 docs → Citation 리스트
    all_citations = self._docs_to_citations(docs)

    # 3. 실제 답변에서 [N] 마커로 인용된 것만 필터
    filtered_citations = self._filter_citations(cleaned, all_citations)

    return Response(
        text=cleaned,
        citations=filtered_citations,
        model=model_used,
        latency_ms=latency_ms,
        ttft_ms=ttft_ms,
        arm_id=query.arm_id,
        # run_id는 uuid.uuid4().hex로 자동 생성
    )
# src/core/contracts.py
class Citation(BaseModel):
    index: int                          # 1-based (답변의 [1], [2] 마커와 대응)
    content: str                        # parent chunk 텍스트
    metadata: dict[str, Any]            # source, section, page 등
    score: Optional[float] = None       # 원래 cosine 유사도 (0~1)
    display_score: Optional[float] = None  # UI 표시용 보정 점수
    section: Optional[str] = None       # 예: "§7.2.4"

class Response(BaseModel):
    text: str                           # 최종 답변 (참고: 섹션 제거 후)
    citations: list[Citation]
    run_id: str                         # 사용자 피드백 매칭 키
    model: Optional[str] = None
    latency_ms: Optional[int] = None   # 전체 완료 시간
    ttft_ms: Optional[int] = None      # Time to First Token
    input_tokens: Optional[int] = None
    output_tokens: Optional[int] = None
    arm_id: Optional[str] = None
    agent_data: dict[str, Any] = {}
    timestamp: datetime

Citation 인덱스 필터링 로직: 답변 텍스트에서 [1], [2, §7.2.4] 형식의 마커를 정규식으로 추출하여 실제로 인용된 Citation만 반환한다. 마커가 없으면 전체 Citation을 그대로 반환한다.

운영 관측성 — 별편으로 분리

단계 5 직후 운영 관점에서 일어나는 부가 작업(perf_counter 기반 timing 로그, runs.jsonl 메트릭 append, feedback.jsonl 피드백 사이드 채널, pricing.py로 사후 비용 계산)은 동기 흐름의 본질이 아니므로 08-1편 스트리밍·관측성으로 분리했다. 본 글은 동기 호출 한 호흡(HTTP 진입 → Response 조립 → HTTP 응답)에 집중한다.

3.6 단계 6: HTTP 응답 직렬화

# src/services/api/routers/qna_chatbot.py
def run(req: RunRequest) -> RunResponse:
    ...
    response = agent.run(query)
    log_run(record_from_response("qna_chatbot", query, response, extras=extras))
    return RunResponse(response=response, experiment_id=exp_name, arm_id=arm_id)
# src/services/api/schemas.py
class RunResponse(BaseModel):
    response: Response          # 에이전트 응답 전체
    experiment_id: Optional[str] = None   # 어떤 실험이 적용됐는지
    arm_id: Optional[str] = None          # 어떤 arm이 선택됐는지

FastAPI는 RunResponse를 JSON으로 자동 직렬화한다. 클라이언트는 response.text(답변)와 response.citations(인용) 외에 experiment_id, arm_id를 받아 어떤 실험 설정으로 응답됐는지 알 수 있다.

4 SSE 스트리밍 경로 — 별편 참조

동기 경로(/run)와 갈라지는 분기점부터 done 이벤트의 SSE 전송 형식까지는 08-1편 스트리밍·관측성에서 다룬다. 본 글은 동기 호출 한 호흡에 집중하므로 별편의 다음 항목을 참조하면 된다:

  • 분기점: chain.invoke() vs chain.stream()
  • 스트리밍 이벤트 타입 (token/done/error)
  • 스트리밍 구현 (perf_counter TTFT 측정)
  • SSE 전송 형식 (data: {...}\n\n)
  • 동기 vs 스트리밍 비교표

5 데이터 타입 요약표

전체 흐름에서 데이터가 어떤 타입으로 변환되는지 한 눈에 정리한다.

경계 변환 전 변환 후 담당
브라우저 → HTTP TypeScript 객체 JSON string (body) fetch
HTTP → FastAPI JSON string RunRequest FastAPI Pydantic
실험 라우팅 RunRequest (config, exp_name, arm_id) resolve_config_for_user()
FastAPI → Agent RunRequest Query _build_query()
검색 str (질문) list[Document] (child, k2~k3개) ParentChunkRetriever
Rerank/Cosine list[Document] (점수 없음) list[Document] (relevance_score 부여) reranker 또는 cosine
Parent 매핑 list[Document] (child) list[Document] (parent, k개) _map_to_parents_with_child_score()
Context 조립 list[Document] str (context 블록) _build_context_string()
LCEL prompt dict (chain_inputs) list[BaseMessage] PromptTemplate
LCEL llm list[BaseMessage] AIMessage LLM
LCEL parser AIMessage str StrOutputParser()
Response 조립 str + list[Document] Response _build_response()
HTTP 응답 Response RunResponse → JSON FastAPI

6 Worked Example — “데이터 표준화 원칙은?” 추적

질문 “데이터 표준화의 핵심 원칙 5가지를 알려주세요”가 어떻게 변환되는지 단계별로 추적한다.

6.1 입력

POST /agents/qna_chatbot/run
Content-Type: application/json

{
  "text": "데이터 표준화의 핵심 원칙 5가지를 알려주세요",
  "history": [],
  "agent_params": { "document_id": "표준화", "keep_chat_history": true },
  "user_id": "anon-9d3f2a"
}

6.2 단계별 데이터 형태

[1] req: RunRequest
    text="데이터 표준화의 핵심 원칙 5가지를 알려주세요"
    history=[], agent_params={"document_id": "표준화", ...}
    user_id="anon-9d3f2a"

[2] resolve_config_for_user("qna_chatbot", user_id="anon-9d3f2a")
    md5("anon-9d3f2a:reranker_ab_test")[:8] → 0x73a2f1c8 / 0xFFFFFFFF = 0.4517
    cumulative: control(0.5) → 0.4517 ≤ 0.5 → arm_id="control"
    overrides={} → config는 base 그대로
    → (config, "reranker_ab_test", "control", {})

[3] _agent_cache.get(("reranker_ab_test", "control")) → cached agent

[4] query: Query
    text="데이터 표준화의 핵심 원칙 5가지를 알려주세요"
    experiment_id="reranker_ab_test", arm_id="control"

[5] _prepare(query):
    [5a] _ensure_initialized() → no-op (warmup에서 이미 로드)
    [5b] _retrieve_docs("데이터 표준화의...", "표준화"):
         hybrid_search(query, k=6, fetch_k=6)
           → azure: 12개 child Document 반환 (k*2)
           → cosine: 12개 score 부여 (max=0.87, min=0.41)
           → map_parents: parent_id별 dedup → 6개 parent
         → docs: list[Document] (6개, parent chunks ~1500자 각)
    [5c] _build_chain_inputs:
         context = "[Document 1] (출처: 표준화)\n...\n\n[Document 2]..."
                   (총 6개 블록, 약 9000자)
         chat_history = "(이전 대화 없음)"  # history=[]
         → {"context": ..., "chat_history": ..., "question": "데이터 표준화의..."}
    [5d] chain = prompt | llm | StrOutputParser()

[6] chain.invoke(chain_inputs):
    LLM call → AIMessage(content="데이터 표준화의 핵심 원칙은 다음과 같다.\n\n1. 명확성 [1]...")
    → str: "데이터 표준화의 핵심 원칙은 다음과 같다.\n\n1. 명확성 [1]..."

[7] _build_response:
    cleaned = strip(text, "참고:" 섹션)
    all_citations = [Citation(index=1, content=docs[0].page_content, ...), ...] (6개)
    extracted_indices = {1, 2, 3, 4, 5}  # 본문 [1]~[5] 마커
    filtered_citations = all_citations[:5]
    → Response(text=cleaned, citations=[...], run_id="a3f2b1c4...",
               model="gpt-4.1", latency_ms=3240, ttft_ms=3240, arm_id="control")

[8] log_run(record_from_response(...)):
    data/runtime/runs.jsonl에 1줄 append
    {"timestamp": "2026-05-05T10:23:45", "run_id": "a3f2b1c4...",
     "agent_name": "qna_chatbot", "query": "데이터 표준화의...",
     "answer": "데이터 표준화의 핵심...", "total_time_ms": 3240,
     "ttft_ms": 3240, "success": true, "citation_count": 5, ...}

[9] RunResponse:
    {"response": {...}, "experiment_id": "reranker_ab_test", "arm_id": "control"}
    → JSON 직렬화 → HTTP 200

6.3 메모리 사용량 추정

단계 객체 크기
[5b] child docs (12개) list[Document] ~12 × 400자 + 메타 ≈ 6KB
[5b] parent docs (6개) list[Document] ~6 × 1500자 + 메타 ≈ 12KB
[5c] context string str ~9KB
[6] LLM prompt list[BaseMessage] ~12KB (시스템 프롬프트 포함)
[6] LLM response AIMessage ~3KB (답변 평균 ~2000자)
[7] Response Pydantic ~15KB (citations 포함)

요청당 피크 메모리: ~50KB (RAGConfig·prompt·LLM 캐시 제외).

7 후속 — 별편으로 이어가기

동기 호출 한 호흡 위에 얹히는 운영 계층(SSE 스트리밍, timing 로그, 메트릭 JSONL, 피드백 사이드 채널), 흐름에서 발견된 설계 취약점, Data Standardizer Supervisor의 다른 흐름, Phase C-2 LangGraph 분해 예고는 08-1편 스트리밍·관측성에서 이어진다.

8 관련 주제

선행 학습

바로 이어 읽을 글

Phase C-1 후속 포스트

Phase C-2 — LangGraph 전환

Subscribe

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