Field Log · Entry
Retrieval Service Adapter: ACL Filter·Deadline·Empty Result 구분 (4/10)
오늘의 결론
- Retrieval port는
search(str) -> list[str]가 아닙니다. Query·tenant·ACL·time filter·top-k를 입력으로 받고 source ID·span·score·provenance를 가진Evidence를 반환합니다.- 권한 filter는 검색 후 Python에서 제거하지 않습니다. Backend의 candidate 생성 전에 적용하고, 지원되지 않는 filter는 조용히 버리지 않고 fail closed합니다.
- HTTPX의 connect·read·write·pool timeout은 개별 I/O 제한입니다. 전체 검색 deadline은
asyncio.timeout_at()같은 별도 상한으로 감쌉니다.[]는 정상적으로 근거가 없다는 뜻입니다. Timeout·5xx·schema drift를[]로 바꾸면 Agent가 장애를 “근거 없음”으로 오판합니다.- Cancellation은 failure나 empty result가 아닙니다. Response body와 connection을 정리한 뒤
CancelledError를 다시 전파합니다.
앞 글에서는 전체 deadline 안에서 LLM retry를 제한했습니다. 이제 동일한 시간·권한 계약을 retrieval service까지 전달합니다.
이 글의 대상과 학습 시간
- 대상: Vector DB·검색 API 호출을 RAG application에 안전하게 연결하려는 Python 개발자
- 선수 지식: top-k 검색, metadata filter, async HTTP client의 기본 개념
- 빠르게 읽기: 약 13분
- 코드와 contract test까지 따라 하기: 약 55분
- 산출물: typed
RetrievalPort, shared HTTPX adapter, empty·timeout·unavailable을 구분하는 fake test
1. list[str]가 잃어버리는 정보
다음 interface는 demo에는 충분합니다.
async def search(query: str) -> list[str]: ...Production answer와 citation을 만들려면 더 많은 정보가 필요합니다.
- 어느 tenant와 ACL scope에서 찾았는가?
- 어떤 document version과 index generation인가?
- 원문의 page·character·element span은 어디인가?
- Sparse·dense·fusion·rerank 중 어떤 score인가?
- 검색 시점에 유효한 문서인가?
- 결과가 0건인지 dependency가 실패한 것인지?
- 남은 전체 deadline은 얼마인가?
Retrieval adapter는 검색 알고리즘을 application에 숨기는 동시에, 검색 결과의 의미를 잃지 않게 번역해야 합니다.
2. Search request를 계약으로 만든다
from dataclasses import dataclass
from datetime import datetime
from typing import Protocol
@dataclass(frozen=True)
class PrincipalScope:
tenant_id: str
subject_id: str
allowed_groups: tuple[str, ...]
@dataclass(frozen=True)
class SearchFilter:
document_types: tuple[str, ...] = ()
valid_at: datetime | None = None
language: str | None = None
@dataclass(frozen=True)
class SearchRequest:
request_id: str
query: str
principal: PrincipalScope
filters: SearchFilter
top_k: int = 8
def __post_init__(self) -> None:
if not self.query.strip():
raise ValueError("query must not be empty")
if not self.principal.tenant_id:
raise ValueError("tenant_id is required")
if not 1 <= self.top_k <= 100:
raise ValueError("top_k must be between 1 and 100")PrincipalScope는 user가 request body에 자유롭게 넣은 값을 그대로 쓰면 안 됩니다. 인증 adapter가 검증한 identity·group을 내부 value로 만들고 application이 search request에 전달합니다.
Filter 지원 여부도 계약이다
Backend A는 valid_at을 지원하고 B는 지원하지 않을 수 있습니다. Adapter가 지원하지 않는 filter를 무시하면 잘못된 문서가 검색됩니다.
class UnsupportedSearchFilter(Exception):
def __init__(self, names: set[str]) -> None:
self.names = frozenset(names)
super().__init__(f"unsupported filters: {sorted(names)}")보안·시간 filter는 fallback 없이 fail closed하고, 단순 preference filter만 명시적 policy에 따라 완화할 수 있습니다.
3. Evidence는 citation 가능한 provenance를 가진다
@dataclass(frozen=True)
class SourceSpan:
page: int | None
start_char: int | None
end_char: int | None
element_id: str | None
@dataclass(frozen=True)
class RetrievalScore:
stage: str
value: float
@dataclass(frozen=True)
class Evidence:
evidence_id: str
tenant_id: str
document_id: str
document_version: str
text: str
source_uri: str
span: SourceSpan
scores: tuple[RetrievalScore, ...]
index_generation: strevidence_id는 이번 prompt에서만 쓰는 [S1]과 다릅니다. 검색 corpus에서 안정적으로 document version·span을 가리키는 ID입니다. 85편의 context builder가 이를 짧은 prompt-local citation ID로 바꿉니다.
Score를 하나로 뭉개지 않는다
BM25, cosine similarity, RRF, cross-encoder score는 scale과 의미가 다릅니다. score: float 하나로 덮어쓰면 failure 분석이 어렵습니다. Stage별 score와 rank를 보존하거나 trace artifact로 연결합니다.
4. Port에는 deadline을 명시한다
from collections.abc import Sequence
class RetrievalPort(Protocol):
async def search(
self,
request: SearchRequest,
*,
deadline: "Deadline",
) -> Sequence[Evidence]: ...Deadline은 83편에서 만든 absolute monotonic time입니다. Async code에서는 asyncio.get_running_loop().time()을 clock으로 사용하면 asyncio.timeout_at(deadline.expires_at)과 같은 기준을 공유할 수 있습니다.
전체 deadline과 stage budget
Retrieval이 전체 시간을 다 써 버리면 generation이 시작되지 못합니다. Application은 global deadline보다 이른 stage deadline을 만들 수 있습니다.
def stage_deadline(
overall: Deadline,
*,
max_stage_seconds: float,
clock: Clock,
) -> Deadline:
return Deadline(
expires_at=min(
overall.expires_at,
clock.now() + max_stage_seconds,
)
)이는 retrieval timeout 뒤 전체 요청을 계속 진행하라는 뜻이 아닙니다. 검색 없는 fallback이 허용되는 질문인지 router policy가 별도로 결정해야 합니다.
5. HTTPX timeout만으로 overall deadline이 되지 않는다
HTTPX는 network operation별 timeout을 제공합니다.
import httpx
def transport_timeout(
remaining: float,
*,
settings: "HTTPTimeoutSettings",
) -> httpx.Timeout:
return httpx.Timeout(
connect=min(settings.connect, remaining),
read=min(settings.read, remaining),
write=min(settings.write, remaining),
pool=min(settings.pool, remaining),
)Read timeout은 전체 response 완료 시간이 아니라 다음 network data를 기다리는 제한입니다. Data가 조금씩 계속 도착하면 overall latency가 길어질 수 있습니다. 그래서 전체 adapter call을 absolute deadline으로 한 번 더 감쌉니다.
6. Shared client를 쓰는 HTTP retrieval adapter
import asyncio
import httpx
class HTTPRetrievalAdapter:
def __init__(
self,
*,
client: httpx.AsyncClient,
endpoint: str,
timeout_settings: "HTTPTimeoutSettings",
clock: Clock,
) -> None:
self._client = client
self._endpoint = endpoint
self._timeouts = timeout_settings
self._clock = clock
async def search(
self,
request: SearchRequest,
*,
deadline: Deadline,
) -> tuple[Evidence, ...]:
remaining = deadline.require_remaining(clock=self._clock)
payload = to_search_payload(request)
try:
async with asyncio.timeout_at(deadline.expires_at):
response = await self._client.post(
self._endpoint,
json=payload,
timeout=transport_timeout(
remaining,
settings=self._timeouts,
),
headers={"x-request-id": request.request_id},
)
response.raise_for_status()
return parse_evidence_response(
response.json(),
expected_tenant=request.principal.tenant_id,
expected_limit=request.top_k,
)
except asyncio.CancelledError:
raise
except TimeoutError as exc:
raise RetrievalDeadlineExceeded(
stage="overall",
request_id=request.request_id,
) from exc
except httpx.TimeoutException as exc:
raise RetrievalDeadlineExceeded(
stage=timeout_stage(exc),
request_id=request.request_id,
) from exc
except httpx.HTTPStatusError as exc:
raise map_search_http_error(exc, request.request_id) from exc
except httpx.RequestError as exc:
raise RetrievalUnavailable(
request_id=request.request_id,
reason=type(exc).__name__,
) from exc
except (ValueError, KeyError, TypeError) as exc:
raise InvalidRetrievalResponse(request.request_id) from excAsyncClient는 request마다 만들지 않습니다. Application startup에서 connection limit·TLS·base headers를 검증해 한 번 만들고 shutdown에서 닫습니다.
Catch 순서
asyncio.timeout_at()이 만료되면 context 밖에서 TimeoutError가 발생합니다. HTTPX의 I/O timeout은 httpx.TimeoutException 계열입니다. 두 error code를 구분하면 전체 stage budget이 끝난 것인지 connect/read/pool이 막힌 것인지 운영자가 알 수 있습니다.
CancelledError를 넓은 exception handler가 삼키지 않게 먼저 다시 올립니다. Async context와 HTTPX가 response 자원을 정리한 뒤 상위 API의 취소 흐름이 계속됩니다.
7. ACL은 candidate 생성 전에 적용한다
나쁜 순서는 다음과 같습니다.
전체 tenant에서 top 8 검색
→ Python에서 권한 없는 7개 제거
→ 결과 1개이는 두 문제가 있습니다.
- 허용 문서가 전체 ranking의 9위였다면 결과에서 사라져 recall이 낮아짐
- 권한 없는 document의 ID·score·timing이 application trace에 노출될 수 있음
올바른 의미는 다음입니다.
tenant + ACL + valid_at filter
→ allowed candidate set
→ sparse/dense retrieval
→ fusion/rerank
→ top-kdef to_search_payload(request: SearchRequest) -> dict:
return {
"query": request.query,
"limit": request.top_k,
"filters": {
"tenant_id": {"eq": request.principal.tenant_id},
"acl_groups": {
"overlaps": list(request.principal.allowed_groups),
},
"valid_at": (
request.filters.valid_at.isoformat()
if request.filters.valid_at
else None
),
"document_type": {
"in": list(request.filters.document_types),
},
},
}Payload 문법은 특정 검색 service의 가상 예시입니다. 중요한 점은 adapter contract test가 모든 필수 filter의 전달을 검사하고, backend가 filter capability를 지원하지 않으면 startup 또는 request 전에 실패하는 것입니다.
8. Response를 믿지 말고 다시 검증한다
Backend가 보안 filter를 수행하더라도 application boundary에서 invariant를 확인합니다.
def parse_evidence_response(
raw: dict,
*,
expected_tenant: str,
expected_limit: int,
) -> tuple[Evidence, ...]:
items = raw["results"]
if not isinstance(items, list):
raise TypeError("results must be a list")
if len(items) > expected_limit:
raise ValueError("backend returned more than requested limit")
evidence = tuple(to_evidence(item) for item in items)
if any(item.tenant_id != expected_tenant for item in evidence):
raise AuthorizationBoundaryViolation(expected_tenant)
if len({item.evidence_id for item in evidence}) != len(evidence):
raise ValueError("duplicate evidence_id")
return evidencesource_uri, span, version도 required schema로 검증합니다. Text가 비어 있거나 source를 해석할 수 없는 result는 citation에 쓸 수 없으므로 조용히 drop하지 않고 data quality metric과 명시적 policy를 적용합니다.
9. Empty와 failure를 type으로 구분한다
정상 empty result는 예외가 아닐 수 있습니다.
evidence = await retriever.search(request, deadline=search_deadline)
if not evidence:
return SearchOutcome.no_evidence(
reason="no_authorized_match",
index_generation=current_generation,
)반면 다음은 dependency failure입니다.
RetrievalDeadlineExceeded
RetrievalRateLimited
RetrievalUnavailable
InvalidRetrievalResponse
AuthorizationBoundaryViolation
UnsupportedSearchFilterApplication router가 outcome에 따라 다음 행동을 정합니다.
| Outcome | 가능한 행동 |
|---|---|
| Empty | query rewrite, clarification, insufficient evidence |
| Timeout | budget이 있으면 제한된 retry, 아니면 dependency error |
| Rate limited | Retry-After와 deadline 확인 |
| Unavailable | 승인된 fallback index 또는 명시적 실패 |
| Invalid response | retry보다 contract incident로 격리 |
| ACL violation | 즉시 차단·security alert, 결과 비노출 |
Empty로 모두 합치면 품질 dashboard에는 “문서가 없음”만 쌓이고 실제 search outage가 숨습니다.
10. In-memory fake도 같은 의미를 지킨다
class InMemoryRetrieval:
def __init__(self, items: tuple[Evidence, ...], clock: Clock) -> None:
self._items = items
self._clock = clock
async def search(self, request, *, deadline):
deadline.require_remaining(clock=self._clock)
terms = set(request.query.lower().split())
allowed = [
item for item in self._items
if item.tenant_id == request.principal.tenant_id
and terms.intersection(item.text.lower().split())
]
return tuple(allowed[: request.top_k])Fake의 ranking 품질이 실제 backend와 같을 필요는 없습니다. 그러나 tenant filter, top-k, empty result, deadline과 stable evidence ID 같은 behavioral contract는 같아야 합니다.
11. Contract test를 fake와 adapter에 반복한다
import pytest
@pytest.mark.asyncio
async def test_retriever_never_returns_other_tenant(retriever_factory):
retriever = retriever_factory([
evidence("a-1", tenant="fab-a", text="pump stop 88"),
evidence("b-1", tenant="fab-b", text="pump stop 92"),
])
result = await retriever.search(
search_request(tenant="fab-a", query="pump stop"),
deadline=future_deadline(),
)
assert {item.tenant_id for item in result} <= {"fab-a"}
@pytest.mark.asyncio
async def test_empty_is_not_timeout(retriever_factory):
retriever = retriever_factory([])
result = await retriever.search(
search_request(tenant="fab-a", query="missing"),
deadline=future_deadline(),
)
assert result == ()
@pytest.mark.asyncio
async def test_expired_deadline_raises(retriever_factory):
retriever = retriever_factory([])
with pytest.raises(DeadlineExceeded):
await retriever.search(
search_request(tenant="fab-a", query="pump"),
deadline=expired_deadline(),
)retriever_factory를 in-memory fake와 HTTP mock adapter에 각각 연결합니다. Live search service는 별도 marker에서 소수 smoke contract만 실행합니다.
12. 관측성은 stage와 outcome을 분리한다
Search span에 다음을 기록합니다.
retrieval.system = internal-search-v2
retrieval.index_generation = kb-2026-07-16-03
retrieval.requested_k = 8
retrieval.returned_k = 5
retrieval.outcome = success | empty | timeout | unavailable
retrieval.deadline_ms = 700
retrieval.duration_ms = 184
retrieval.filter_hash = opaque-policy-versionQuery 전문, ACL group 전체, document text는 기본 span attribute에 넣지 않습니다. 고카디널리티와 민감정보를 줄이고 필요하면 접근 제어된 artifact에 redaction 후 저장합니다.
반드시 볼 metric
- Empty rate by intent·tenant tier
- Timeout rate by connect/read/pool/overall
- Returned-k distribution
- ACL violation count는 0이어야 함
- Unsupported filter count
- Index generation별 latency·empty 변화
- Cancellation과 caller deadline count
자주 실패하는 구현 7가지
1. 검색 후 ACL을 적용한다
Recall이 왜곡되고 unauthorized metadata가 내부에 들어옵니다. Candidate 생성 전에 filter합니다.
2. 지원하지 않는 filter를 무시한다
오래된 문서나 다른 tenant 결과가 섞일 수 있습니다. Security·temporal filter는 fail closed합니다.
3. Timeout을 empty list로 바꾼다
Dependency 장애가 evidence 부족으로 위장됩니다. Typed error로 분리합니다.
4. HTTPX read timeout을 전체 deadline으로 믿는다
조금씩 data가 도착하면 전체 시간이 계속 늘 수 있습니다. Adapter 전체를 absolute deadline으로 감쌉니다.
5. 매 request마다 AsyncClient를 만든다
Connection pool·TLS 재사용을 잃고 resource close가 복잡해집니다. Lifecycle 단위 shared client를 사용합니다.
6. Cancellation을 broad exception으로 삼킨다
Client가 떠난 뒤에도 검색과 LLM 비용이 계속 발생합니다. Cleanup 후 다시 전파합니다.
7. Provider score 하나만 저장한다
Fusion·rerank가 추가되면 어느 단계가 바뀌었는지 모릅니다. Stage score·rank·index generation을 보존합니다.
직접 적용 체크리스트
- SearchRequest에 tenant·principal·filter·top-k가 typed field로 있다.
- Identity는 인증 adapter가 검증한 값에서 생성한다.
- ACL·tenant filter가 candidate 생성 전에 적용된다.
- Unsupported security filter는 fail closed한다.
- Evidence가 document version·source span·index generation을 보존한다.
- Stage deadline이 overall deadline보다 늦지 않다.
- HTTP transport timeout과 overall timeout을 함께 둔다.
- Empty·timeout·unavailable·invalid response가 다른 outcome이다.
- Cancellation을 다시 전파한다.
- Fake와 HTTP adapter가 같은 contract suite를 통과한다.
스스로 확인하기
- Top-k 검색 뒤 ACL을 적용하면 authorized recall이 왜 낮아지는가?
- HTTPX read timeout이 전체 검색 시간을 제한하지 못하는 이유는 무엇인가?
- Empty result와 timeout을 같은
[]로 만들면 Agent decision이 어떻게 잘못되는가? - Monotonic deadline을 다른 host에 그대로 보내면 왜 안 되는가?
- Evidence에 document version과 source span이 필요한 이유는 무엇인가?
자주 묻는 질문
Q1. Vector DB client를 직접 쓰면 HTTP adapter가 필요 없나요?
Transport가 HTTP인지 SDK인지와 무관하게 request·result·error mapping boundary는 필요합니다. 공식 async client를 concrete adapter 안에서 쓰고 application은 RetrievalPort만 보게 합니다.
Q2. 빈 결과도 예외로 만들면 더 명확하지 않나요?
가능하지만 정상 domain outcome과 dependency failure를 구분해야 합니다. 이 글은 empty tuple을 정상 반환하고 application이 no_evidence outcome으로 바꾸는 방식을 사용했습니다.
Q3. ACL을 application에서도 다시 확인하면 느리지 않나요?
Backend filter가 실제 보안 경계이고 application 검사는 defense in depth입니다. Tenant ID 비교처럼 싼 invariant를 재검증해 adapter·backend regression을 즉시 차단합니다.
Q4. Search stage budget은 몇 ms가 적당한가요?
고정 정답은 없습니다. 전체 사용자 SLO, query 유형, index latency 분포, generation budget을 함께 보고 정합니다. 글의 숫자는 설명용입니다.
Q5. 검색 장애 때 cached answer를 보여 줘도 되나요?
Cache가 tenant·document version·ACL·freshness를 검증하고 UI가 오래된 답임을 명시하는 경우에만 별도 fallback으로 고려합니다. Empty로 위장하거나 권한 검증을 생략하면 안 됩니다.
마무리
Retrieval adapter의 품질은 검색 API를 호출하는 코드 길이로 결정되지 않습니다. 다음 네 의미를 끝까지 보존해야 합니다.
- 누가 어떤 범위에서 검색했는가?
- 어떤 version과 원문 span을 근거로 받았는가?
- 남은 deadline 안에서 완료됐는가?
- 0건인가, 실패인가, 취소인가?
다음 글에서는 이 Evidence를 context budget 안에 결정적으로 배치하고, model이 만든 citation ID를 server-side evidence map으로 검증하는 Prompt·Context Builder를 구현합니다.
검증 정보
- 글의 성격: Python·HTTPX·gRPC 공식 문서 해설 + 작성자 설계 제안 + 가상 코드 예시
- 마지막 검증일: 2026-07-16
- 검증 환경: Python 3.12
asyncio.timeout_at, HTTPX 0.27 계열 timeout API 개념 - 실행 주의: Search payload·response schema는 가상이며 실제 backend adapter에서 고정 version contract로 구현
- 수치 표기: top-k 8·700ms 등은 설명용 시작값이며 dataset·SLO로 재조정 필요
- 경험 표기: fab·장비·온도·index generation 이름은 실제 운영 정보가 아닌 가상 예시