모델 추상화 계층 설계, 프롬프트 구성과 컨텍스트 주입, 스키마 기반 출력 제어, 폴백 전략 등 AI 모델의 입출력을 체계적으로 관리하는 방법을 다룹니다.
프로덕션 AI 시스템에서 모델 API를 직접 호출하는 코드는 놀라울 정도로 적습니다. 전체 코드베이스의 대부분은 "모델 호출 전에 무엇을 준비하고, 모델 응답 후에 무엇을 처리할 것인가"에 할애됩니다. 이것이 바로 모델 래핑의 영역이며, 하네스 엔지니어링에서 가장 기본적이면서도 가장 빈번하게 작업하게 되는 부분입니다.
왜 모델을 추상화해야 할까요? 실무에서는 다음과 같은 상황이 빈번하게 발생합니다.
이런 변경이 발생할 때마다 애플리케이션 코드 전체를 수정해야 한다면, 유지보수 비용이 급격히 증가합니다. 모델 추상화 계층은 이 문제를 해결합니다.
from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
class ModelProvider(Enum):
ANTHROPIC = "anthropic"
OPENAI = "openai"
LOCAL = "local"
@dataclass
class ModelRequest:
"""공급자에 독립적인 요청 형식"""
system_prompt: str
messages: list[dict[str, str]]
max_tokens: int = 4096
temperature: float = 0.7
stop_sequences: list[str] | None = None
@dataclass
class ModelResponse:
"""공급자에 독립적인 응답 형식"""
content: str
model: str
usage: dict[str, int] # input_tokens, output_tokens
latency_ms: float
finish_reason: str
class ModelAdapter(ABC):
"""모델 추상화 인터페이스"""
@abstractmethod
async def generate(self, request: ModelRequest) -> ModelResponse:
...
@abstractmethod
def supports_streaming(self) -> bool:
...
class AnthropicAdapter(ModelAdapter):
def __init__(self, client, model_id: str = "claude-sonnet-4-20250514"):
self.client = client
self.model_id = model_id
async def generate(self, request: ModelRequest) -> ModelResponse:
import time
start = time.monotonic()
response = await self.client.messages.create(
model=self.model_id,
system=request.system_prompt,
messages=request.messages,
max_tokens=request.max_tokens,
temperature=request.temperature,
)
elapsed = (time.monotonic() - start) * 1000
return ModelResponse(
content=response.content[0].text,
model=self.model_id,
usage={
"input_tokens": response.usage.input_tokens,
"output_tokens": response.usage.output_tokens,
},
latency_ms=elapsed,
finish_reason=response.stop_reason,
)
def supports_streaming(self) -> bool:
return True
class OpenAIAdapter(ModelAdapter):
def __init__(self, client, model_id: str = "gpt-4o"):
self.client = client
self.model_id = model_id
async def generate(self, request: ModelRequest) -> ModelResponse:
import time
start = time.monotonic()
messages = [{"role": "system", "content": request.system_prompt}]
messages.extend(request.messages)
response = await self.client.chat.completions.create(
model=self.model_id,
messages=messages,
max_tokens=request.max_tokens,
temperature=request.temperature,
)
elapsed = (time.monotonic() - start) * 1000
return ModelResponse(
content=response.choices[0].message.content,
model=self.model_id,
usage={
"input_tokens": response.usage.prompt_tokens,
"output_tokens": response.usage.completion_tokens,
},
latency_ms=elapsed,
finish_reason=response.choices[0].finish_reason,
)
def supports_streaming(self) -> bool:
return True모델 추상화 계층을 설계할 때 가장 중요한 원칙은 "가장 단순한 공통 인터페이스"를 찾는 것입니다. 모든 공급자의 모든 기능을 추상화하려 하면 오히려 복잡해집니다. 핵심 기능(생성, 스트리밍)만 추상화하고, 공급자 특화 기능은 어댑터 레벨에서 처리하세요.
모델에 전달되는 입력은 사용자의 원본 질문 그대로가 아닙니다. 하네스는 원본 입력에 다양한 컨텍스트를 주입하여 최종 프롬프트를 구성합니다.
from dataclasses import dataclass
@dataclass
class PromptTemplate:
"""구조화된 프롬프트 템플릿"""
system: str
context_prefix: str = "참고 자료:"
few_shot_examples: list[dict[str, str]] | None = None
def compose(
self,
user_input: str,
context: list[str] | None = None,
) -> list[dict[str, str]]:
messages = []
# 컨텍스트가 있는 경우 (RAG 등)
if context:
context_block = f"{self.context_prefix}\n"
for i, doc in enumerate(context, 1):
context_block += f"\n[{i}] {doc}\n"
messages.append({
"role": "user",
"content": context_block,
})
messages.append({
"role": "assistant",
"content": "참고 자료를 확인했습니다. 질문해 주세요.",
})
# 퓨샷 예제
if self.few_shot_examples:
for example in self.few_shot_examples:
messages.append({
"role": "user",
"content": example["input"],
})
messages.append({
"role": "assistant",
"content": example["output"],
})
# 사용자 입력
messages.append({"role": "user", "content": user_input})
return messages
# 사용 예시
code_review_prompt = PromptTemplate(
system=(
"당신은 시니어 소프트웨어 엔지니어입니다. "
"코드 리뷰를 수행하고 개선 사항을 제안합니다. "
"반드시 한국어로 응답하세요."
),
few_shot_examples=[
{
"input": "def add(a,b): return a+b",
"output": "타입 힌트가 누락되었습니다. "
"`def add(a: int, b: int) -> int:`로 수정을 권장합니다.",
}
],
)컨텍스트 주입은 모델이 더 정확한 응답을 생성하도록 돕는 핵심 기법입니다. 주요 전략은 다음과 같습니다.
컨텍스트 윈도우 크기에는 한계가 있습니다. 무조건 많은 컨텍스트를 주입하는 것이 아니라, 관련성이 높은 컨텍스트를 선별하여 주입하는 것이 중요합니다. 컨텍스트가 너무 많으면 오히려 "바늘 찾기(needle-in-a-haystack)" 문제로 성능이 저하될 수 있습니다.
모델의 출력은 자유 형식의 텍스트입니다. 이를 애플리케이션이 사용할 수 있는 구조화된 데이터로 변환하는 것이 출력 후처리의 역할입니다.
import json
import re
from typing import TypeVar, Type
from pydantic import BaseModel, ValidationError
T = TypeVar("T", bound=BaseModel)
class OutputParser:
"""모델 출력을 구조화된 데이터로 변환"""
@staticmethod
def parse_json(text: str) -> dict:
"""JSON 블록 추출 및 파싱"""
# 마크다운 코드 블록 내 JSON 추출
json_match = re.search(
r"```(?:json)?\s*\n(.*?)\n```",
text,
re.DOTALL,
)
if json_match:
return json.loads(json_match.group(1))
# 직접 JSON 파싱 시도
try:
return json.loads(text)
except json.JSONDecodeError:
pass
# 중괄호 범위 추출
brace_match = re.search(r"\{.*\}", text, re.DOTALL)
if brace_match:
return json.loads(brace_match.group())
raise ValueError(f"JSON을 추출할 수 없습니다: {text[:100]}...")
@staticmethod
def parse_to_model(text: str, model_class: Type[T]) -> T:
"""Pydantic 모델로 파싱 및 검증"""
data = OutputParser.parse_json(text)
return model_class.model_validate(data)
# 사용 예시
class CodeReviewResult(BaseModel):
severity: str # "critical" | "warning" | "info"
line: int
message: str
suggestion: str
class CodeReviewResponse(BaseModel):
summary: str
issues: list[CodeReviewResult]
overall_score: float
# 모델 출력을 타입 안전한 객체로 변환
raw_output = model.generate(prompt)
review = OutputParser.parse_to_model(raw_output, CodeReviewResponse)
# review.issues[0].severity -> 타입 체크 가능최신 LLM API들은 구조화된 출력(Structured Output)을 네이티브로 지원하기 시작했습니다. 모델에게 출력 스키마를 사전에 알려주고, 해당 스키마에 맞는 응답만 생성하도록 강제하는 방식입니다.
from pydantic import BaseModel, Field
class TaskAnalysis(BaseModel):
"""작업 분석 결과 스키마"""
task_type: str = Field(
description="작업 유형: 'code_review', 'bug_fix', 'feature'"
)
complexity: int = Field(
ge=1, le=10,
description="복잡도 (1-10)"
)
estimated_hours: float = Field(
gt=0,
description="예상 소요 시간"
)
required_skills: list[str] = Field(
description="필요한 기술 스택"
)
risks: list[str] = Field(
default_factory=list,
description="잠재적 위험 요소"
)
# Anthropic의 도구 기반 구조화된 출력
async def analyze_task(description: str) -> TaskAnalysis:
response = await client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
tools=[{
"name": "submit_analysis",
"description": "작업 분석 결과를 제출합니다",
"input_schema": TaskAnalysis.model_json_schema(),
}],
tool_choice={"type": "tool", "name": "submit_analysis"},
messages=[{
"role": "user",
"content": f"다음 작업을 분석해주세요:\n{description}",
}],
)
tool_use = next(
b for b in response.content
if b.type == "tool_use"
)
return TaskAnalysis.model_validate(tool_use.input)스키마 기반 출력 제어는 Guardrails AI 프레임워크의 핵심 아이디어이기도 합니다. Pydantic 모델로 출력 스키마를 정의하고, 모델 출력이 스키마를 위반하면 자동으로 재시도하는 방식으로 동작합니다. 이에 대해서는 6장에서 더 자세히 다룹니다.
프로덕션 환경에서 모델 호출은 다양한 이유로 실패할 수 있습니다. 네트워크 오류, 레이트 리밋, 모델 서비스 장애, 출력 형식 오류 등이 발생할 수 있으며, 하네스는 이런 상황에 대비한 폴백 전략을 갖추어야 합니다.
import asyncio
from typing import Callable, Awaitable
class FallbackChain:
"""다단계 폴백 체인"""
def __init__(self):
self.strategies: list[
tuple[str, Callable[..., Awaitable]]
] = []
def add(
self, name: str, strategy: Callable[..., Awaitable]
) -> "FallbackChain":
self.strategies.append((name, strategy))
return self
async def execute(self, *args, **kwargs):
last_error = None
for name, strategy in self.strategies:
try:
result = await strategy(*args, **kwargs)
if name != self.strategies[0][0]:
print(f"[FALLBACK] {name} 전략으로 성공")
return result
except Exception as e:
print(f"[FALLBACK] {name} 실패: {e}")
last_error = e
continue
raise RuntimeError(
f"모든 폴백 전략 실패. 마지막 에러: {last_error}"
)
# 폴백 체인 구성
fallback = (
FallbackChain()
.add("primary", lambda prompt: call_claude_opus(prompt))
.add("secondary", lambda prompt: call_claude_sonnet(prompt))
.add("cache", lambda prompt: get_cached_response(prompt))
.add("static", lambda prompt: get_static_fallback(prompt))
)
# 실행: primary 실패 시 secondary, secondary 실패 시 cache...
result = await fallback.execute(user_prompt)import asyncio
import random
async def retry_with_backoff(
func,
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 60.0,
):
"""지수 백오프를 적용한 재시도"""
for attempt in range(max_retries):
try:
return await func()
except Exception as e:
if attempt == max_retries - 1:
raise
# 지수 백오프 + 지터
delay = min(
base_delay * (2 ** attempt) + random.uniform(0, 1),
max_delay,
)
print(
f"[RETRY] 시도 {attempt + 1}/{max_retries} "
f"실패, {delay:.1f}초 후 재시도: {e}"
)
await asyncio.sleep(delay)모든 재시도와 폴백이 실패했을 때, 시스템이 완전히 중단되는 것이 아니라 제한된 기능이라도 제공하는 것을 그레이스풀 디그레이데이션(Graceful Degradation)이라 합니다.
async def handle_request(query: str) -> dict:
"""그레이스풀 디그레이데이션 적용"""
try:
# 1순위: 풀 기능 응답
return await generate_full_response(query)
except ModelUnavailableError:
try:
# 2순위: 캐시된 유사 응답
return await find_similar_cached(query)
except CacheMissError:
# 3순위: 정적 안내 메시지
return {
"content": "현재 AI 응답 서비스에 일시적인 장애가 있습니다. "
"잠시 후 다시 시도해 주세요.",
"fallback": True,
"degraded": True,
}폴백 전략을 설계할 때 주의할 점이 있습니다. 폴백 응답의 품질이 기대치보다 크게 낮다면, 차라리 명시적으로 실패를 알리는 것이 나을 수 있습니다. 저품질 응답을 마치 정상 응답인 것처럼 제공하면 사용자의 신뢰를 더 크게 훼손할 수 있습니다.
4장에서는 AI 시스템의 품질을 보증하는 테스트 하네스를 다룹니다. 비결정적 출력을 어떻게 테스트할 것인가라는 근본적인 질문에서 시작하여, 스냅샷 테스트, 속성 기반 테스트, 회귀 테스트, 에이전트 행동 테스트 등의 구체적인 기법을 살펴봅니다.
이 글이 도움이 되셨나요?
관련 주제 더 보기
비결정적 출력 테스트, 스냅샷 테스트, 속성 기반 테스트, 회귀 테스트, 에이전트 행동 테스트 등 AI 시스템 테스트의 핵심 기법을 다룹니다.
래핑, 미들웨어, 사이드카, 파이프라인, 이벤트 기반 등 AI 시스템 하네스의 5가지 핵심 아키텍처 패턴과 적용 시나리오를 분석합니다.
lm-evaluation-harness, Inspect AI, HELM 프레임워크 분석과 커스텀 평가 하네스 설계, 벤치마크 스위트 구성, 자동화된 모델 비교 방법을 다룹니다.