Field Log · Entry

Production RAG Agent 프로젝트 구조: Vertical Slice·Port·Adapter로 시작하기 (1/10)

오늘의 결론

• Production RAG Agent의 첫 파일은 거대한 agent.py가 아니라 바뀌지 않을 application contract와 바뀌기 쉬운 외부 기술을 분리하는 경계입니다.
• Port는 “무엇이 필요한가”를 표현하는 작은 interface이고, adapter는 특정 LLM API·검색엔진·DB·HTTP framework로 그 계약을 구현합니다.
• Layer 폴더만 먼저 늘리지 말고, 질문 하나가 답변 하나로 끝나는 얇은 vertical slice를 실행·테스트한 뒤 기능을 확장합니다.
• Domain·application 코드는 vendor SDK object와 HTTP response를 직접 받지 않습니다. 내부의 typed value와 명시적 error만 주고받습니다.
• 첫 완료 조건은 실제 모델의 멋진 답이 아닙니다. Fake adapter로도 같은 use case가 실행되고, provider를 바꿔도 핵심 정책 테스트가 그대로 통과하는 상태입니다.

앞 글에서는 지식베이스를 data·policy·model·index가 묶인 release로 운영했습니다. 이번 10편은 지금까지의 설계를 하나의 Python 애플리케이션으로 옮깁니다.

이 글의 대상과 학습 시간

• 대상: notebook 또는 framework tutorial 수준의 RAG를 실제 서비스 코드로 옮기려는 초·중급 Python 개발자
• 선수 지식: 첫 RAG 전체 흐름과 Production Agent Harness의 개념
• 빠르게 읽기: 약 12분
• 코드와 테스트까지 따라 하기: 약 50분
• 산출물: 한 개의 answer_question core를 구성하는 typed contract와 fake 기반 unit test

1. 왜 agent.py 하나로 시작하면 곧 막히는가

처음에는 다음 코드도 충분합니다.

def answer(question: str) -> str:
    docs = vector_db.search(question)
    prompt = build_prompt(question, docs)
    return llm.generate(prompt)

문제는 기능이 아니라 변경 이유가 다른 코드가 한 함수 안에 섞였다는 점입니다.

• LLM provider와 model ID가 바뀜
• 검색 backend가 BM25+vector+reranker로 바뀜
• ACL·tenant filter가 추가됨
• prompt와 citation contract가 바뀜
• streaming API가 추가됨
• session 저장 방식이 바뀜
• timeout·retry·rate limit 정책이 바뀜
• 품질 평가와 trace가 추가됨

한 변경을 위해 전체 함수를 건드리면 test 범위와 장애 범위도 커집니다. 반대로 layer를 지나치게 많이 만들면 초반에는 빈 interface와 mapping code만 남습니다. 그래서 한 use case를 세로로 완성하되 외부 경계만 분리하는 출발점이 실용적입니다.

2. Port·Adapter를 RAG Agent 언어로 번역한다

Alistair Cockburn의 원문은 애플리케이션을 UI나 DB 없이도 실행·회귀 테스트할 수 있게 내부와 외부를 port로 나누는 의도를 설명합니다. 육각형 모양 자체가 중요한 것은 아닙니다.

Port

Application이 외부에 요구하는 목적 중심 계약입니다.

from collections.abc import Sequence
from typing import Protocol

class Retriever(Protocol):
    async def search(
        self,
        query: str,
        *,
        tenant_id: str,
        limit: int,
    ) -> Sequence["Evidence"]: ...

QdrantRetriever나 OpenSearchRetriever가 아니라 Retriever라고 부르는 이유는 application이 필요한 대화가 “어느 제품을 호출하라”가 아니라 “권한 범위 안에서 근거를 찾아라”이기 때문입니다.

Python의 typing.Protocol은 명시적 상속 없이도 필요한 method shape를 만족하면 구조적으로 호환되는 interface를 표현합니다.

Adapter

특정 외부 기술을 port에 맞게 번역합니다.

class OpenSearchRetriever:
    def __init__(self, client, index: str) -> None:
        self._client = client
        self._index = index

    async def search(self, query, *, tenant_id, limit):
        raw = await self._client.search(
            index=self._index,
            body={
                "query": {
                    "bool": {
                        "must": [{"match": {"text": query}}],
                        "filter": [{"term": {"tenant_id": tenant_id}}],
                    }
                },
                "size": limit,
            },
        )
        return [to_evidence(hit) for hit in raw["hits"]["hits"]]

Adapter가 provider response를 내부 Evidence로 바꾸므로 이후 code는 OpenSearch JSON shape를 알 필요가 없습니다.

Primary와 secondary adapter

방향	이 프로젝트의 예	누가 호출을 시작하는가
Primary / driving	FastAPI endpoint, CLI, batch eval	사용자·외부 system이 application을 호출
Secondary / driven	LLM, retriever, session store, clock	application이 외부 기능을 호출

3. 먼저 한 개의 Vertical Slice를 고른다

Vertical slice는 화면·service·DB 같은 기술 layer를 각각 완성하는 대신, 사용자 가치 한 조각을 입력부터 결과까지 관통시킵니다.

첫 slice의 scope를 다음처럼 제한합니다.

input
  question + tenant_id + request_id

behavior
  retrieve top-k evidence
  compile one prompt
  ask one model call
  validate cited evidence IDs

output
  answer + citations + usage + trace_id

out of scope for post 81
  streaming · retry · multi-step tools · persistent session · HTTP API

이렇게 범위를 적으면 “Agent”라는 이름 때문에 planning·memory·multi-agent부터 붙이는 일을 피할 수 있습니다.

완료 조건

1. CLI 또는 test에서 use case를 호출할 수 있음
2. Fake retriever와 fake LLM만으로 deterministic test가 통과함
3. 내부 type에 provider SDK class가 없음
4. 검색 결과의 tenant와 source ID가 응답까지 보존됨
5. 실패가 Exception 문자열이 아니라 분류된 application error로 나옴

4. 입력과 출력을 먼저 계약으로 만든다

Pydantic model은 외부 JSON을 검증하는 boundary에 유용합니다. 그러나 모든 내부 object를 Pydantic으로 만들 필요는 없습니다. Domain value는 frozen dataclass로도 충분합니다. 아래 type은 뒤의 use case와 test가 공유하는 최소 계약입니다.

from dataclasses import dataclass
from typing import Protocol
from pydantic import BaseModel, ConfigDict, Field

class AnswerRequest(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)

    request_id: str = Field(min_length=8, max_length=128)
    tenant_id: str = Field(min_length=1, max_length=128)
    question: str = Field(min_length=1, max_length=8_000)

@dataclass(frozen=True)
class Evidence:
    id: str
    tenant_id: str
    text: str
    source_uri: str

@dataclass(frozen=True)
class CompiledPrompt:
    text: str
    evidence_ids: tuple[str, ...]

@dataclass(frozen=True)
class GeneratedAnswer:
    text: str
    cited_ids: tuple[str, ...]
    model: str
    input_tokens: int
    output_tokens: int

class Generator(Protocol):
    async def generate(self, prompt: CompiledPrompt) -> GeneratedAnswer: ...

class PromptBuilder(Protocol):
    def compile(
        self,
        *,
        question: str,
        evidence: tuple[Evidence, ...],
    ) -> CompiledPrompt: ...

@dataclass(frozen=True)
class Citation:
    evidence_id: str
    source_uri: str

@dataclass(frozen=True)
class AnswerResult:
    answer: str
    citations: tuple[Citation, ...]
    model: str
    input_tokens: int
    output_tokens: int
    trace_id: str

Boundary에서 extra="forbid"를 사용하면 client typo가 조용히 무시되는 것을 막을 수 있습니다. Strict mode는 문자열 "3"을 integer 3으로 몰래 바꾸는 식의 coercion을 줄입니다. 실제 버전에 맞는 동작은 Pydantic strict mode 문서로 확인합니다.

계약에 넣지 말아야 할 것

• Provider SDK의 response object
• DB connection이나 HTTP request object
• Prompt 전문
• API key·secret
• 내부 stack trace

Public contract와 내부 diagnostic을 분리해야 provider 교체와 보안 검토가 쉬워집니다.

5. Application use case는 정책의 중심이다

from dataclasses import dataclass

@dataclass(frozen=True)
class AnswerPolicy:
    retrieval_limit: int = 8
    minimum_evidence: int = 1

class AnswerQuestion:
    def __init__(
        self,
        *,
        retriever: Retriever,
        generator: "Generator",
        prompt_builder: "PromptBuilder",
        policy: AnswerPolicy,
    ) -> None:
        self._retriever = retriever
        self._generator = generator
        self._prompt_builder = prompt_builder
        self._policy = policy

    async def execute(self, request: AnswerRequest) -> AnswerResult:
        evidence = await self._retriever.search(
            request.question,
            tenant_id=request.tenant_id,
            limit=self._policy.retrieval_limit,
        )
        if len(evidence) < self._policy.minimum_evidence:
            raise InsufficientEvidence(request.request_id)
        if any(item.tenant_id != request.tenant_id for item in evidence):
            raise AuthorizationBoundaryViolation(request.request_id)

        prompt = self._prompt_builder.compile(
            question=request.question,
            evidence=tuple(evidence),
        )
        generated = await self._generator.generate(prompt)
        return assemble_result(
            generated,
            evidence=tuple(evidence),
            trace_id=request.request_id,
        )

여기에는 HTTP status, vendor request parameter, SQL query가 없습니다. 대신 “근거가 최소 몇 개 필요한가”, “어떤 입력에서 멈추는가” 같은 application policy가 보입니다.

인용 ID는 모델이 만든 URL을 그대로 노출하지 않고 server가 전달한 evidence map에서만 해석합니다.

def assemble_result(
    generated: GeneratedAnswer,
    *,
    evidence: tuple[Evidence, ...],
    trace_id: str,
) -> AnswerResult:
    by_id = {item.id: item for item in evidence}
    unknown = set(generated.cited_ids) - set(by_id)
    if unknown:
        raise InvalidCitationIds(sorted(unknown))

    citations = tuple(
        Citation(evidence_id=item_id, source_uri=by_id[item_id].source_uri)
        for item_id in generated.cited_ids
    )
    return AnswerResult(
        answer=generated.text,
        citations=citations,
        model=generated.model,
        input_tokens=generated.input_tokens,
        output_tokens=generated.output_tokens,
        trace_id=trace_id,
    )

6. 폴더는 의존성 방향을 드러내게 만든다

초기 구조는 작게 유지합니다.

production-rag-agent/
├── pyproject.toml
├── src/rag_agent/
│   ├── domain/
│   │   ├── evidence.py
│   │   └── errors.py
│   ├── application/
│   │   ├── contracts.py
│   │   ├── ports.py
│   │   └── answer_question.py
│   ├── adapters/
│   │   ├── llm/
│   │   ├── retrieval/
│   │   └── persistence/
│   ├── entrypoints/
│   │   ├── cli.py
│   │   └── api.py
│   ├── config.py
│   └── bootstrap.py
└── tests/
    ├── unit/
    ├── contract/
    └── integration/

의존성 규칙

entrypoints ─┐
adapters ────┼──> application ───> domain
bootstrap ───┘

domain/application must not import adapters/entrypoints

이 규칙을 import linter로 자동화할 수도 있지만, 처음에는 code review 체크로도 충분합니다.

utils.py를 피한다

무엇을 위한 utility인지 드러나지 않는 파일은 여러 layer가 공유하며 새 결합을 만듭니다. parse_provider_error.py, citation_validator.py, deadline.py처럼 책임으로 이름을 붙입니다.

7. Composition Root에서 실제 부품을 한 번만 조립한다

Dependency injection framework를 먼저 들이지 않아도 됩니다. Process 시작점 한 곳에서 concrete adapter를 만듭니다.

def build_application(settings: Settings) -> AnswerQuestion:
    http_client = build_http_client(settings.http)

    retriever = OpenSearchRetriever(
        client=build_search_client(settings.search),
        index=settings.search.index,
    )
    generator = ProviderGenerator(
        http_client=http_client,
        api_key=settings.llm.api_key.get_secret_value(),
        model=settings.llm.model,
    )
    prompt_builder = VersionedPromptBuilder(
        version=settings.prompt.version,
    )

    return AnswerQuestion(
        retriever=retriever,
        generator=generator,
        prompt_builder=prompt_builder,
        policy=settings.answer_policy,
    )

bootstrap.py만 concrete class를 알고 application은 port만 압니다. Test에서는 같은 constructor에 fake를 넣습니다.

Configuration도 boundary다

Twelve-Factor App의 config 원칙은 배포마다 달라지는 설정을 code와 분리합니다. 그러나 환경변수를 application 곳곳에서 직접 읽으면 다시 전역 결합이 생깁니다.

environment / secret manager
  → Settings validation at startup
  → immutable settings object
  → composition root
  → only required values passed to each adapter

시작 시 필수 설정을 모두 검증하고, API key는 로그와 repr에서 가립니다. Model·prompt·retrieval policy version은 trace에 남기되 secret은 남기지 않습니다.

8. Fake로 첫 정책 테스트를 만든다

Mock call count만 검사하면 구현 순서가 바뀔 때 test가 쉽게 깨집니다. 작은 in-memory fake는 port 계약을 실제처럼 수행하고 outcome을 검사하게 합니다.

class FakeRetriever:
    def __init__(self, evidence_by_query: dict[str, list[Evidence]]) -> None:
        self._data = evidence_by_query

    async def search(self, query, *, tenant_id, limit):
        allowed = [
            item for item in self._data.get(query, [])
            if item.tenant_id == tenant_id
        ]
        return allowed[:limit]

class PlainPromptBuilder:
    def compile(self, *, question, evidence):
        blocks = [f"[{item.id}] {item.text}" for item in evidence]
        return CompiledPrompt(
            text="QUESTION\n" + question + "\nEVIDENCE\n" + "\n".join(blocks),
            evidence_ids=tuple(item.id for item in evidence),
        )

class RecordingFakeGenerator:
    last_prompt: CompiledPrompt | None = None

    async def generate(self, prompt):
        self.last_prompt = prompt
        return GeneratedAnswer(
            text="정지 기준은 88°C입니다.",
            cited_ids=("doc-7#p4",),
            model="fake-v1",
            input_tokens=120,
            output_tokens=14,
        )

import pytest

def evidence(id: str, *, tenant: str, text: str) -> Evidence:
    return Evidence(
        id=id,
        tenant_id=tenant,
        text=text,
        source_uri=f"https://docs.example/{id}",
    )

@pytest.mark.asyncio
async def test_answer_uses_only_authorized_evidence():
    generator = RecordingFakeGenerator()
    use_case = AnswerQuestion(
        retriever=FakeRetriever({
            "정지 기준": [
                evidence("doc-7#p4", tenant="fab-a", text="88°C"),
                evidence("secret#p1", tenant="fab-b", text="92°C"),
            ]
        }),
        generator=generator,
        prompt_builder=PlainPromptBuilder(),
        policy=AnswerPolicy(),
    )

    result = await use_case.execute(
        AnswerRequest(
            request_id="req-0001",
            tenant_id="fab-a",
            question="정지 기준",
        )
    )

    assert result.answer == "정지 기준은 88°C입니다."
    assert [c.evidence_id for c in result.citations] == ["doc-7#p4"]
    assert generator.last_prompt is not None
    assert "secret#p1" not in generator.last_prompt.text

@pytest.mark.asyncio
async def test_application_rejects_leaky_retriever():
    class LeakyRetriever:
        async def search(self, query, *, tenant_id, limit):
            return [evidence("secret#p1", tenant="fab-b", text="92°C")]

    use_case = AnswerQuestion(
        retriever=LeakyRetriever(),
        generator=RecordingFakeGenerator(),
        prompt_builder=PlainPromptBuilder(),
        policy=AnswerPolicy(),
    )

    with pytest.raises(AuthorizationBoundaryViolation):
        await use_case.execute(
            AnswerRequest(
                request_id="req-0002",
                tenant_id="fab-a",
                question="정지 기준",
            )
        )

숫자·이름·문서는 모두 가상 예시입니다. 첫 test는 허용된 근거만 prompt와 citation에 들어가는지 확인하고, 두 번째 test는 adapter가 계약을 어겨 다른 tenant의 근거를 반환해도 application boundary가 거부하는지 확인합니다.

9. Error도 application contract로 분류한다

모든 예외를 RAGError 하나로 감싸면 client와 운영자가 무엇을 해야 할지 모릅니다.

class ApplicationError(Exception):
    code: str
    retryable: bool

class InvalidRequest(ApplicationError):
    code = "invalid_request"
    retryable = False

class InsufficientEvidence(ApplicationError):
    code = "insufficient_evidence"
    retryable = False

class DependencyUnavailable(ApplicationError):
    code = "dependency_unavailable"
    retryable = True

class DeadlineExceeded(ApplicationError):
    code = "deadline_exceeded"
    retryable = True

class AuthorizationBoundaryViolation(ApplicationError):
    code = "authorization_boundary_violation"
    retryable = False

class InvalidCitationIds(ApplicationError):
    code = "invalid_citation_ids"
    retryable = False

Adapter는 provider-specific exception을 이 taxonomy로 번역하되 원래 status, request ID, retry hint는 내부 diagnostic에 보존합니다. 82~83번에서 stream event와 retry policy까지 확장합니다.

10. 첫 실행은 실제 provider 없이 닫힌다

async def main() -> None:
    app = AnswerQuestion(
        retriever=FakeRetriever(load_fixture_evidence()),
        generator=RecordingFakeGenerator(),
        prompt_builder=PlainPromptBuilder(),
        policy=AnswerPolicy(),
    )
    result = await app.execute(
        AnswerRequest(
            request_id="local-0001",
            tenant_id="demo",
            question="장비 정지 기준은?",
        )
    )
    print(result)

이 slice가 닫혀야 실제 provider adapter를 붙였을 때 문제를 “application policy”와 “외부 API”로 분리할 수 있습니다.

자주 실패하는 설계 7가지

1. Framework object가 domain까지 침투한다

Request, SDK response, ORM model을 use case input으로 직접 넘깁니다. Boundary에서 내부 value로 변환합니다.

2. Port가 vendor API를 그대로 복사한다

create_chat_completion(**kwargs) 같은 port는 provider 교체에 도움이 되지 않습니다. Application이 필요한 generate(prompt, policy) 계약을 만듭니다.

3. 모든 class에 interface를 만든다

변경·테스트 경계가 아닌 내부 helper까지 추상화하면 탐색 비용만 늘어납니다. 외부 I/O와 정책 교체 지점부터 분리합니다.

4. Global client와 global settings를 쓴다

Test 격리, lifecycle, connection close, tenant 안전성이 흐려집니다. Composition root가 소유권을 갖게 합니다.

5. except Exception으로 전부 retry한다

잘못된 schema나 권한 거부도 반복되어 비용과 부하를 늘립니다. 83번에서 retryable 조건을 좁힙니다.

6. 첫 slice부터 streaming과 tools를 넣는다

부분 output·cancel·side effect까지 동시에 생겨 correctness 원인을 찾기 어렵습니다. Non-streaming 한 번 호출로 기준선을 만듭니다.

7. Fake가 production contract와 다르다

Fake도 같은 port contract suite를 통과하게 합니다. 89번에서 fake·provider adapter에 같은 contract test를 실행합니다.

직접 적용 체크리스트

• 첫 vertical slice의 input·behavior·output·out-of-scope를 적었다.
• Domain/application이 vendor SDK·HTTP·DB module을 import하지 않는다.
• LLM·retrieval·session·clock은 목적 중심 port로 분리했다.
• Provider response를 내부 typed value로 변환한다.
• Composition root 한 곳에서 concrete adapter와 lifecycle을 소유한다.
• 설정을 startup에 검증하고 secret을 log에서 제거한다.
• Fake adapter로 실제 network 없이 use case를 실행한다.
• Authorization·citation 같은 invariant를 outcome으로 test한다.
• Error code와 retry 가능 여부를 분리한다.
• 실제 provider 연결 전 기준선 test가 통과한다.

스스로 확인하기

1. Retriever port가 OpenSearchClient를 직접 반환하면 어떤 결합이 생기는가?
2. Primary adapter와 secondary adapter를 현재 RAG Agent에서 각각 두 개씩 찾을 수 있는가?
3. 첫 vertical slice에서 반드시 제외해야 할 기능은 무엇인가?
4. Fake test가 call count 대신 확인해야 할 business invariant는 무엇인가?
5. Composition root 밖에서 concrete adapter를 생성하면 lifecycle과 설정 추적이 왜 어려워지는가?

자주 묻는 질문

Q1. LangChain·LlamaIndex·LangGraph를 쓰면 Port·Adapter가 필요 없나요?

필요합니다. Framework도 adapter 중 하나로 봅니다. Application contract가 framework object에 종속되지 않으면 작은 use case는 직접 구현하고, 필요한 부분만 framework adapter로 교체할 수 있습니다.

Q2. Python은 duck typing인데 Protocol까지 써야 하나요?

작은 script라면 없어도 됩니다. 여러 adapter와 fake가 같은 비동기 계약, keyword argument, return type을 만족해야 할 때 Protocol과 static type check가 drift를 일찍 찾는 데 유용합니다.

Q3. Clean Architecture와 Hexagonal Architecture 중 무엇이 맞나요?

명칭보다 의존성 방향이 중요합니다. Business/application policy가 외부 framework와 provider에 의존하지 않고, 외부 기술이 내부 port를 구현하면 이 글의 목적을 충족합니다.

Q4. Microservice로 나눠야 Production인가요?

아닙니다. 처음에는 modular monolith가 관찰·transaction·debugging에 더 단순할 수 있습니다. 독립 배포·확장·소유권이 실제로 필요할 때 service 경계를 분리합니다.

Q5. Vector DB와 LLM을 모두 fake로 쓰면 현실성이 떨어지지 않나요?

Unit test의 목적은 application policy를 빠르고 결정적으로 검증하는 것입니다. Adapter contract·integration test와 소수의 live smoke test를 별도 계층으로 추가하면 됩니다.

마무리

Production RAG Agent의 첫 설계 목표는 “많은 추상화”가 아니라 다음 세 가지입니다.

1. 사용자 가치 하나를 end-to-end로 닫는다.
2. 외부 기술을 목적 중심 port 뒤로 보낸다.
3. Fake로도 핵심 정책을 검증할 수 있게 만든다.

다음 글에서는 이 Generator port 뒤에 실제 provider를 연결하는 LLM API Client: Message·Streaming·Structured Error를 구현합니다.

검증 정보

• 글의 성격: 공식 문서·architecture 원문 해설 + 작성자 설계 제안 + 가상 코드 예시
• 마지막 검증일: 2026-07-16
• 검증 환경: Python 3.12 문법 기준, Pydantic 2 계열 API 표기
• 실행 주의: 예시는 경계를 설명하기 위한 축약 code이며 provider·DB adapter는 다음 편에서 확장
• 경험 표기: 실제 PHM·반도체 운영 결과를 주장하지 않으며 장비·tenant 이름과 수치는 가상

참고자료

• Alistair Cockburn, Hexagonal Architecture 원문
• Python typing.Protocol
• PEP 544 — Protocols: Structural Subtyping
• Pydantic Models
• Pydantic Strict Mode
• The Twelve-Factor App — Config
• pytest Fixtures