Field Log · Entry
LLM API Client 설계: Message·Streaming Event·Structured Error (2/10)
오늘의 결론
- Application은 provider SDK의 message·response·exception을 직접 알면 안 됩니다. 내부의
GenerationRequest,GenerationEvent,LLMClientError만 봅니다.- Streaming은 문자열 조각의 반복이 아니라 시작→0개 이상의 delta→정확히 하나의 terminal event를 갖는 protocol입니다.
- 화면에 보낸 partial text와 최종 저장할 answer를 구분하고, usage·finish reason·provider request ID를 terminal result에 보존합니다.
- Provider의 HTTP status와 error payload는 인증·요청·제한·일시 장애·protocol 오류로 번역합니다. Retry는 다음 편의 policy가 결정합니다.
- Cancel은 task 종료만 뜻하지 않습니다. Upstream stream을 닫고 connection을 반환하며 사용자 취소와 dependency 실패를 다른 상태로 남깁니다.
앞 글에서는 application core와 외부 기술을 Port·Adapter로 분리했습니다. 이번에는 Generator port 뒤에 실제 LLM provider를 붙여도 core가 흔들리지 않는 client boundary를 만듭니다.
이 글의 대상과 학습 시간
- 대상: LLM SDK 호출을 service code에 직접 넣었다가 streaming·오류 처리에서 복잡해진 Python 개발자
- 선수 지식: async iterator, dataclass, HTTP status의 기본 개념
- 빠르게 읽기: 약 13분
- 코드와 test까지 따라 하기: 약 55분
- 산출물: provider와 무관한 message·stream event·error contract와 deterministic reducer test
1. SDK 한 줄이 왜 application 전체로 번지는가
처음에는 다음 코드가 자연스럽습니다.
response = await provider.responses.create(
model="some-model",
input=messages,
)
return response.output_text서비스 요구가 늘면 호출 한 줄 주변에 provider 세부사항이 붙습니다.
- System·user·assistant message 표현
- 이미지·파일·tool call 같은 content part
- Streaming event 이름과 delta 조립
- Finish reason과 token usage
- Request ID와 rate-limit hint
- Timeout·연결 오류·HTTP error payload
- Client disconnect와 task cancellation
- Provider별 model name·parameter 차이
이 object들이 endpoint와 application까지 새면 SDK upgrade 한 번에도 넓은 수정이 필요합니다. Client의 일은 API를 얇게 감싸는 것이 아니라 외부 protocol을 내부 의미로 번역하는 것입니다.
application meaning provider protocol
─────────────────── ─────────────────
GenerationRequest ──adapter──> request JSON / SDK object
GenerationEvent <─adapter─── stream event / SSE frame
LLMClientError <─adapter─── HTTP status / exception2. Message를 문자열이 아닌 typed value로 만든다
Prompt를 하나의 긴 문자열로만 넘기면 role 경계와 evidence 출처가 사라집니다. 반대로 provider SDK의 message class를 내부 계약으로 쓰면 vendor 결합이 생깁니다.
from collections.abc import Mapping
from dataclasses import dataclass, field
from enum import StrEnum
from math import isfinite
class Role(StrEnum):
SYSTEM = "system"
USER = "user"
ASSISTANT = "assistant"
@dataclass(frozen=True)
class TextPart:
text: str
@dataclass(frozen=True)
class Message:
role: Role
parts: tuple[TextPart, ...]
def __post_init__(self) -> None:
if not self.parts:
raise ValueError("message must contain at least one part")
if any(not part.text.strip() for part in self.parts):
raise ValueError("message text must not be empty")
@classmethod
def text(cls, role: Role, value: str) -> "Message":
normalized = value.strip()
if not normalized:
raise ValueError("message text must not be empty")
return cls(role=role, parts=(TextPart(normalized),))
@dataclass(frozen=True)
class GenerationPolicy:
model_alias: str
max_output_tokens: int
temperature: float = 0.0
def __post_init__(self) -> None:
if not self.model_alias.strip():
raise ValueError("model alias must not be empty")
if self.max_output_tokens < 1:
raise ValueError("max_output_tokens must be positive")
if not isfinite(self.temperature) or self.temperature < 0:
raise ValueError("temperature must be finite and non-negative")
@dataclass(frozen=True)
class GenerationRequest:
request_id: str
messages: tuple[Message, ...]
policy: GenerationPolicy
metadata: Mapping[str, str] = field(default_factory=dict)
def __post_init__(self) -> None:
if not self.request_id.strip():
raise ValueError("request_id must not be empty")
if not self.messages:
raise ValueError("at least one message is required")
if any(
message.role is Role.SYSTEM
for message in self.messages[1:]
):
raise ValueError("system message is allowed only at index 0")
if self.messages[-1].role is not Role.USER:
raise ValueError("last input message must have user role")model_alias는 provider model ID가 아니라 application policy 이름입니다. 예를 들어 answer-standard를 composition root의 model registry가 실제 model ID로 해석합니다.
Message invariant
초기 version에서는 다음처럼 제한할 수 있습니다.
- Message는 한 개 이상이어야 함
- 첫 message만 system role을 허용
- 마지막 입력은 user role이어야 함
- Empty text와 과도한 입력은 호출 전에 거부
- Secret·raw document 전문은 metadata에 넣지 않음
Multimodal input이 필요해지면 ImagePart, FilePart를 union에 추가합니다. Provider가 지원하지 않는 part는 조용히 버리지 않고 unsupported_content_part 오류를 반환해야 합니다.
3. Streaming을 event state machine으로 정의한다
AsyncIterator[str]만 반환하면 다음 정보를 표현하기 어렵습니다.
- Provider가 요청을 수락했는가?
- 첫 token 전까지 얼마나 걸렸는가?
- 최종 usage와 finish reason은 무엇인가?
- 중간에 실패했다면 이미 보낸 text는 어떻게 처리하는가?
- 정상 완료와 사용자 취소를 어떻게 구분하는가?
내부 event를 명시하면 consumer가 CLI·SSE API·batch runner여도 같은 protocol을 사용합니다.
from dataclasses import dataclass
from typing import AsyncIterator, Protocol, TypeAlias
@dataclass(frozen=True)
class GenerationStarted:
run_id: str
provider_request_id: str | None
@dataclass(frozen=True)
class TextDelta:
sequence: int
text: str
@dataclass(frozen=True)
class UsageSnapshot:
input_tokens: int | None
output_tokens: int | None
@dataclass(frozen=True)
class GenerationCompleted:
text: str
finish_reason: str
usage: UsageSnapshot
model: str
provider_request_id: str | None
@dataclass(frozen=True)
class GenerationFailed:
error: "LLMClientError"
@dataclass(frozen=True)
class GenerationCancelled:
reason: str = "caller_cancelled"
class StreamProtocolError(RuntimeError):
"""Provider event sequence violated the internal stream contract."""
GenerationEvent: TypeAlias = (
GenerationStarted
| TextDelta
| UsageSnapshot
| GenerationCompleted
| GenerationFailed
| GenerationCancelled
)
class LLMClient(Protocol):
def stream(self, request: GenerationRequest) -> AsyncIterator[GenerationEvent]: ...지켜야 할 protocol invariant
Started? → (TextDelta | UsageSnapshot)* → Completed | Failed | Cancelled
terminal count = 1 for a normally consumed stream
delta.sequence = 0, 1, 2, ...
no event after terminal
Completed.text = concatenate(all TextDelta.text)Provider에 따라 Started보다 오류가 먼저 올 수 있으므로 started는 선택적입니다. 그러나 정상적으로 소비한 stream의 terminal은 반드시 하나여야 consumer가 spinner·DB state·HTTP stream을 닫을 수 있습니다. Caller가 Python task 자체를 취소한 경우에는 event를 전달할 기회가 없을 수 있으므로 out-of-band cancellation으로 기록하고 CancelledError를 다시 올립니다.
4. Reducer가 partial output과 final result를 분리한다
UI는 delta를 즉시 보여줄 수 있지만 DB에 answer를 매 delta마다 정답처럼 저장하면 안 됩니다. Reducer는 event를 누적하되 terminal 전에는 완료로 표시하지 않습니다.
from dataclasses import dataclass, replace
@dataclass(frozen=True)
class StreamState:
text: str = ""
next_sequence: int = 0
usage: UsageSnapshot = UsageSnapshot(None, None)
started: bool = False
activity_seen: bool = False
provider_request_id: str | None = None
terminal: str | None = None
def reduce_event(state: StreamState, event: GenerationEvent) -> StreamState:
if state.terminal is not None:
raise StreamProtocolError("event received after terminal")
match event:
case TextDelta(sequence=seq, text=text):
if seq != state.next_sequence:
raise StreamProtocolError(
f"expected delta {state.next_sequence}, got {seq}"
)
return replace(
state,
text=state.text + text,
next_sequence=seq + 1,
activity_seen=True,
)
case UsageSnapshot() as usage:
return replace(state, usage=usage, activity_seen=True)
case GenerationCompleted(text=text):
if text != state.text:
raise StreamProtocolError("final text differs from deltas")
return replace(state, terminal="completed", activity_seen=True)
case GenerationFailed():
return replace(state, terminal="failed", activity_seen=True)
case GenerationCancelled():
return replace(state, terminal="cancelled", activity_seen=True)
case GenerationStarted(provider_request_id=request_id):
if state.started or state.activity_seen:
raise StreamProtocolError("started event is out of order")
return replace(
state,
started=True,
provider_request_id=request_id,
)
raise StreamProtocolError(f"unknown event: {type(event)!r}")Stream을 소비한 최종 answer만 session에 commit합니다. Partial text를 복구용으로 저장한다면 별도의 draft_output 또는 append-only event로 표시해 최종 답과 혼동하지 않습니다.
5. Provider adapter는 세 번 번역한다
Adapter의 핵심 책임은 request, event, error 세 mapping입니다.
5.1 내부 request → provider request
def to_provider_request(
request: GenerationRequest,
*,
model_registry: dict[str, str],
) -> dict:
try:
model = model_registry[request.policy.model_alias]
except KeyError as exc:
raise LLMClientError(
code="unknown_model_alias",
retryable=False,
safe_message="설정된 모델 정책을 찾지 못했습니다.",
) from exc
return {
"model": model,
"input": [
{
"role": message.role.value,
"content": [
{"type": "input_text", "text": part.text}
for part in message.parts
],
}
for message in request.messages
],
"max_output_tokens": request.policy.max_output_tokens,
"temperature": request.policy.temperature,
"stream": True,
}위 JSON key는 Responses 계열 API를 설명하기 위한 예시입니다. 실제 adapter는 고정한 provider API·SDK version의 공식 schema에 맞추고 contract test로 drift를 감지합니다.
5.2 Provider event → 내부 event
IGNORABLE_EVENT_TYPES = {
"response.in_progress",
"response.output_item.added",
"response.output_item.done",
"response.content_part.added",
"response.content_part.done",
"response.output_text.done",
}
def map_raw_event(
raw: dict,
*,
sequence: int,
provider_request_id: str | None,
) -> GenerationEvent | None:
event_type = raw.get("type")
if event_type == "response.created":
response = raw.get("response", {})
return GenerationStarted(
run_id=str(response.get("id", "unknown")),
provider_request_id=provider_request_id,
)
if event_type == "response.output_text.delta":
return TextDelta(sequence=sequence, text=str(raw.get("delta", "")))
if event_type == "response.completed":
response = raw["response"]
return GenerationCompleted(
text=extract_output_text(response),
finish_reason=extract_finish_reason(response),
usage=map_usage(response.get("usage")),
model=str(response.get("model", "unknown")),
provider_request_id=provider_request_id,
)
if event_type in {"response.failed", "error"}:
return GenerationFailed(
error=map_stream_error(
raw,
provider_request_id=provider_request_id,
)
)
if event_type in IGNORABLE_EVENT_TYPES:
return None
raise StreamProtocolError(f"unknown provider event: {event_type!r}")알 수 없는 event를 모두 무시하면 schema drift가 숨어버립니다. Production에서는 다음 두 집합을 분리합니다.
IGNORABLE_EVENT_TYPES: 검토 후 의도적으로 무시- 그 외 unknown event: metric 증가 + sampled payload shape 기록 + protocol error
Payload 전문에는 사용자 입력이 있을 수 있으므로 log에는 event type·key 목록·provider request ID만 남깁니다.
OpenAI 계열의 request ID는 response.created JSON 안의 필드라고 가정하지 않습니다. HTTP 응답의 x-request-id를 transport가 추출하고, 공식 Python SDK를 사용한다면 성공 응답의 _request_id 또는 API status exception의 request_id를 읽어 내부 event·error로 옮깁니다.
5.3 Provider failure → 내부 error
from dataclasses import dataclass
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
@dataclass(frozen=True)
class LLMClientError(Exception):
code: str
retryable: bool
safe_message: str
status_code: int | None = None
provider_request_id: str | None = None
retry_after_seconds: float | None = None
def parse_retry_after(
value: str | None,
*,
now: datetime | None = None,
) -> float | None:
if value is None:
return None
candidate = value.strip()
if candidate.isdigit():
return float(candidate)
try:
target = parsedate_to_datetime(candidate)
except (TypeError, ValueError, OverflowError):
return None
if target.tzinfo is None:
target = target.replace(tzinfo=timezone.utc)
current = now or datetime.now(timezone.utc)
return max(0.0, (target - current).total_seconds())
def classify_http_error(response) -> LLMClientError:
status = response.status_code
request_id = response.headers.get("x-request-id")
retry_after = parse_retry_after(response.headers.get("retry-after"))
if status == 400:
code, retryable = "invalid_provider_request", False
elif status == 401:
code, retryable = "provider_authentication_failed", False
elif status == 403:
code, retryable = "provider_permission_denied", False
elif status == 404:
code, retryable = "provider_resource_not_found", False
elif status == 408:
code, retryable = "provider_request_timeout", True
elif status == 429:
code, retryable = "provider_rate_limited", True
elif 500 <= status:
code, retryable = "provider_unavailable", True
else:
code, retryable = "unexpected_provider_response", False
return LLMClientError(
code=code,
retryable=retryable,
safe_message="LLM provider 요청을 완료하지 못했습니다.",
status_code=status,
provider_request_id=request_id,
retry_after_seconds=retry_after,
)safe_message는 client에 보여도 되는 문장입니다. Provider body와 stack trace는 접근 제어된 diagnostic에만 남깁니다. retryable=True라고 해서 무조건 재시도하지 않는 이유는 83편에서 다룹니다.
6. 실제 stream adapter의 뼈대
Provider SDK를 직접 쓰든 httpx.AsyncClient를 쓰든 lifecycle ownership이 보여야 합니다.
import asyncio
from collections.abc import AsyncIterator
class ProviderLLMClient:
def __init__(self, transport, model_registry: dict[str, str]) -> None:
self._transport = transport
self._models = model_registry
async def stream(
self,
request: GenerationRequest,
) -> AsyncIterator[GenerationEvent]:
sequence = 0
terminal_emitted = False
try:
payload = to_provider_request(
request,
model_registry=self._models,
)
async with self._transport.stream_json(payload) as raw_stream:
# transport가 HTTP response header에서 추출한 값이다.
provider_request_id = raw_stream.headers.get("x-request-id")
async for raw in raw_stream:
event = map_raw_event(
raw,
sequence=sequence,
provider_request_id=provider_request_id,
)
if event is None:
continue
if isinstance(event, TextDelta):
sequence += 1
yield event
if isinstance(
event,
(GenerationCompleted, GenerationFailed),
):
terminal_emitted = True
return
terminal_emitted = True
yield GenerationFailed(
LLMClientError(
code="stream_ended_without_terminal",
retryable=False,
safe_message="응답 stream이 비정상적으로 종료됐습니다.",
provider_request_id=provider_request_id,
)
)
except asyncio.CancelledError:
self._record_cancelled(request.request_id)
raise
except LLMClientError as exc:
if terminal_emitted:
self._record_cleanup_error(request.request_id, exc)
return
terminal_emitted = True
yield GenerationFailed(exc)
except TransportTimeout as exc:
if terminal_emitted:
self._record_cleanup_error(request.request_id, exc)
return
terminal_emitted = True
yield GenerationFailed(map_timeout(exc))
except TransportHTTPError as exc:
if terminal_emitted:
self._record_cleanup_error(request.request_id, exc)
return
terminal_emitted = True
yield GenerationFailed(classify_http_error(exc.response))
except (KeyError, TypeError, ValueError, StreamProtocolError) as exc:
if terminal_emitted:
self._record_cleanup_error(request.request_id, exc)
return
terminal_emitted = True
yield GenerationFailed(map_protocol_error(exc))raw_stream.headers는 예시 transport 계약이며 HTTPX object를 그대로 뜻하지 않습니다. Transport가 response header와 async JSON event iterator를 한 context object로 묶어 제공한다고 가정합니다. CancelledError는 일반 failure로 바꾸지 않습니다. Async context manager가 upstream body와 connection을 닫은 뒤 취소를 다시 전파하고, span status와 cancel counter는 별도로 갱신합니다. 명시적 취소 명령을 event로 전달하는 상위 runtime은 GenerationCancelled를 만들 수 있지만 task 취소 중인 generator가 반드시 그 event를 yield한다고 가정해서는 안 됩니다.
extract_output_text, map_usage, map_stream_error, map_timeout, map_protocol_error와 Transport* 타입은 고정한 provider·transport에 맞춰 구현할 adapter helper이므로 이 뼈대에서는 이름만 표시했습니다. 이 글의 실행 가능한 최소 산출물은 그 아래의 provider-independent request·event·reducer와 fake test입니다.
7. Non-streaming API도 같은 event를 소비한다
Streaming과 non-streaming을 별도 구현하면 mapping과 오류 정책이 갈라집니다. generate()는 stream()을 끝까지 소비하는 convenience method로 둡니다.
async def generate(
client: LLMClient,
request: GenerationRequest,
) -> GenerationCompleted:
state = StreamState()
completed: GenerationCompleted | None = None
async for event in client.stream(request):
state = reduce_event(state, event)
if isinstance(event, GenerationCompleted):
completed = event
elif isinstance(event, GenerationFailed):
raise event.error
elif isinstance(event, GenerationCancelled):
raise asyncio.CancelledError
if completed is None:
raise StreamProtocolError("stream closed without completed event")
return completed이 구조는 CLI가 전체 답만 원할 때와 FastAPI가 delta를 즉시 전송할 때 동일한 adapter contract를 공유하게 합니다.
8. Usage와 식별자는 부가 정보가 아니다
최종 결과에는 최소한 다음을 남깁니다.
| 필드 | 쓰임 |
|---|---|
| internal request ID | 사용자 요청·session·trace 연결 |
| provider request ID | provider 지원 문의와 장애 상관관계 |
| model·policy version | 회귀와 비용 변화 귀속 |
| input/output token | 비용·context 폭증 감지 |
| finish reason | 길이 제한·안전 거부·정상 종료 구분 |
| time to first delta | 사용자가 느끼는 초기 응답성 |
| total latency | 전체 처리량과 deadline 분석 |
Request ID는 prompt나 사용자 ID를 hash해 만들지 않습니다. 충돌과 개인정보 추론을 피하도록 무작위·opaque ID를 사용합니다.
로그에 남기지 않을 것
- API key와 authorization header
- Prompt·evidence·answer 전문
- Provider raw error body 전문
- 사용자 이메일·사번 같은 직접 식별자
- Tool credential과 signed URL
내용을 조사해야 할 때는 opt-in sample, redaction, 짧은 retention, 접근 감사를 함께 설계합니다.
9. Fake stream으로 protocol을 검증한다
Network 없이 event ordering과 terminal rule을 test할 수 있어야 합니다.
from types import SimpleNamespace
import pytest
class FakeLLMClient:
def __init__(self, events: list[GenerationEvent]) -> None:
self._events = events
async def stream(self, request):
for event in self._events:
yield event
def sample_request() -> GenerationRequest:
return GenerationRequest(
request_id="req-0001",
messages=(Message.text(Role.USER, "정지 기준은?"),),
policy=GenerationPolicy(
model_alias="answer-standard",
max_output_tokens=256,
),
)
def fake_response(status: int):
return SimpleNamespace(status_code=status, headers={})
@pytest.mark.asyncio
async def test_stream_reduces_to_one_completed_answer():
client = FakeLLMClient([
GenerationStarted("run-1", "provider-7"),
TextDelta(0, "정지 "),
TextDelta(1, "기준은 88°C입니다."),
UsageSnapshot(240, 18),
GenerationCompleted(
text="정지 기준은 88°C입니다.",
finish_reason="stop",
usage=UsageSnapshot(240, 18),
model="model-a",
provider_request_id="provider-7",
),
])
result = await generate(client, sample_request())
assert result.text == "정지 기준은 88°C입니다."
assert result.usage.output_tokens == 18
assert result.provider_request_id == "provider-7"pytest-asyncio를 test dependency에 추가하고 marker를 사용합니다. Failure path도 각각 test합니다.
def test_reducer_rejects_sequence_gap_and_late_started():
with pytest.raises(StreamProtocolError, match="expected delta 0"):
reduce_event(StreamState(), TextDelta(1, "잘못된 순서"))
state = reduce_event(StreamState(), TextDelta(0, "일부"))
with pytest.raises(StreamProtocolError, match="out of order"):
reduce_event(state, GenerationStarted("late", "provider-1"))
def test_reducer_rejects_mismatched_final_and_second_terminal():
state = reduce_event(StreamState(), TextDelta(0, "실제 delta"))
mismatched = GenerationCompleted(
text="다른 최종문",
finish_reason="stop",
usage=UsageSnapshot(10, 3),
model="model-a",
provider_request_id="provider-1",
)
with pytest.raises(StreamProtocolError, match="differs"):
reduce_event(state, mismatched)
cancelled = reduce_event(StreamState(), GenerationCancelled())
with pytest.raises(StreamProtocolError, match="after terminal"):
reduce_event(
cancelled,
GenerationFailed(
LLMClientError(
code="cancel-test",
retryable=False,
safe_message="test",
)
),
)
@pytest.mark.asyncio
async def test_generate_rejects_stream_without_terminal():
client = FakeLLMClient([
GenerationStarted("run-1", "provider-7"),
TextDelta(0, "끝나지 않은 답"),
])
with pytest.raises(StreamProtocolError, match="without completed"):
await generate(client, sample_request())
@pytest.mark.parametrize(
("status", "code", "retryable"),
[
(400, "invalid_provider_request", False),
(401, "provider_authentication_failed", False),
(429, "provider_rate_limited", True),
(503, "provider_unavailable", True),
],
)
def test_http_error_mapping(status, code, retryable):
error = classify_http_error(fake_response(status))
assert error.code == code
assert error.retryable is retryable
def test_retry_after_delay_seconds():
assert parse_retry_after("7") == 7.0
assert parse_retry_after(None) is NoneProvider adapter에는 recorded fixture를 쓰되 실제 prompt와 secret을 제거합니다. Live test는 비용과 변동성이 있으므로 소수 smoke test로 분리합니다.
자주 실패하는 구현 7가지
1. SDK response를 endpoint까지 전달한다
API response schema가 provider upgrade에 끌려갑니다. Adapter에서 내부 completed result로 변환합니다.
2. Delta를 그대로 DB의 final answer에 append한다
중간 실패 답이 정상 답처럼 남습니다. Draft event와 committed answer를 분리합니다.
3. Unknown stream event를 모두 무시한다
Provider schema drift를 놓칩니다. 검토된 ignore allowlist 밖의 event는 protocol metric과 error로 다룹니다.
4. 모든 exception을 같은 500으로 만든다
잘못된 request와 일시 장애를 구분할 수 없습니다. Stable error code·retryability·safe message로 번역합니다.
5. Client를 요청마다 새로 만든다
Connection pool과 TLS handshake 이점을 잃습니다. Application lifecycle 동안 공유하고 startup/shutdown에서 엽니다.
6. 사용자 disconnect를 provider 장애로 기록한다
가용성 지표가 왜곡됩니다. Cancelled를 독립 terminal outcome으로 둡니다.
7. Provider model ID를 business code에 흩뿌린다
Model migration과 A/B route를 추적하기 어렵습니다. Model alias와 versioned registry를 사용합니다.
직접 적용 체크리스트
- Application message가 provider SDK class를 import하지 않는다.
- Request·event·error mapping 함수가 각각 분리돼 있다.
- Stream은 정확히 하나의 terminal outcome으로 끝난다.
- Delta sequence와 final text 일치 여부를 검증한다.
- Partial output과 committed answer 저장소가 구분된다.
- Usage·finish reason·model·provider request ID를 보존한다.
- Cancel 시 upstream body와 connection을 닫는다.
- Unknown event와 malformed payload를 관측한다.
- Error log에서 secret과 본문을 redaction한다.
- Fake stream으로 success·failure·cancel protocol을 test한다.
스스로 확인하기
AsyncIterator[str]로는 어떤 lifecycle 정보를 잃는가?GenerationCompleted.text와 delta 결합문이 다르면 무엇을 의심해야 하는가?- HTTP 400과 429를 같은 retry policy로 처리하면 어떤 문제가 생기는가?
- Provider request ID와 내부 request ID를 왜 둘 다 보존해야 하는가?
- Cancellation을 exception log만 남기면 운영 지표가 어떻게 왜곡되는가?
자주 묻는 질문
Q1. Provider SDK의 streaming helper를 쓰면 event contract가 필요 없나요?
SDK helper는 wire protocol 처리를 줄여 줍니다. 그러나 application과 API consumer가 의존할 안정적인 event 이름, terminal rule, error code는 여전히 내부에서 정의해야 합니다.
Q2. 한 provider만 쓸 예정인데 adapter가 과한가요?
Provider 교체보다 SDK version 변경, model migration, streaming 추가, test 격리를 위해 먼저 가치가 생깁니다. Provider를 완전히 추상화하려고 모든 parameter의 최소 공통분모만 남기지는 마세요. Application에 필요한 capability를 계약으로 둡니다.
Q3. Usage event가 stream 중간에 없으면 어떻게 하나요?
Terminal response에서만 채워도 됩니다. None을 허용하고 “0 token”과 “측정 불가”를 구분합니다. 추정값을 provider 실측값처럼 저장하지 않습니다.
Q4. Partial text를 사용자에게 보낸 뒤 실패하면 어떻게 표시하나요?
SSE consumer에 별도의 error terminal event를 보내고 UI가 “완료되지 않은 답변”임을 표시하게 합니다. Session의 final answer로 commit하지 않습니다.
Q5. Tool call event도 이 client에 넣나요?
Provider event mapping에는 포함할 수 있지만 실행은 하지 않습니다. Typed tool proposal로 변환한 뒤 86편의 registry·validation·policy를 통과시킵니다.
마무리
좋은 LLM client는 provider 기능을 숨기는 얇은 wrapper가 아닙니다. 다음 세 계약을 안정화하는 anti-corruption layer입니다.
- Application이 의도한 typed request
- 순서와 terminal이 명확한 streaming event
- 운영 행동으로 연결되는 structured error
다음 글에서는 이 client에 Timeout·Retry·Rate Limit·Circuit Breaker를 붙이되, 재시도가 오히려 장애를 증폭하지 않도록 하나의 deadline과 retry budget으로 통제합니다.
검증 정보
- 글의 성격: 공식 protocol·Python 문서 해설 + 작성자 설계 제안 + 가상 코드 예시
- 마지막 검증일: 2026-07-16
- 검증 환경: Python 3.12 문법, HTTPX 0.27 계열 개념, provider API key는 공식 문서 재확인 필요
- 실행 주의: Raw event 이름과 payload는 provider·API version에 따라 달라지므로 adapter fixture와 contract test로 고정
- 경험 표기: 장비명·온도·request ID·token 수는 실제 운영 결과가 아닌 설명용 가상 값