래핑, 미들웨어, 사이드카, 파이프라인, 이벤트 기반 등 AI 시스템 하네스의 5가지 핵심 아키텍처 패턴과 적용 시나리오를 분석합니다.
건축에서 구조가 건물의 수명을 결정하듯, 하네스의 아키텍처는 AI 시스템의 확장성과 유지보수성을 결정합니다. 1장에서 하네스가 무엇인지를 정의했다면, 이번 장에서는 하네스를 어떤 구조로 설계할 것인지를 다룹니다. 프로덕션에서 검증된 5가지 아키텍처 패턴을 살펴보고, 각 패턴이 어떤 상황에 적합한지를 분석하겠습니다.
하네스 아키텍처를 선택하기 전에, 다음 세 가지 질문에 답해야 합니다.
이 질문에 대한 답에 따라 적합한 패턴이 달라집니다.
가장 직관적인 패턴입니다. 모델 호출을 하나의 클래스나 함수로 감싸고, 전처리와 후처리를 같은 레이어에서 수행합니다.
from dataclasses import dataclass
from typing import Any
@dataclass
class HarnessConfig:
model: str = "claude-sonnet-4-20250514"
max_tokens: int = 4096
temperature: float = 0.7
max_retries: int = 3
class ModelWrapper:
"""래핑 패턴: 모델 호출을 단일 클래스로 감싸기"""
def __init__(self, client: Any, config: HarnessConfig):
self.client = client
self.config = config
def invoke(self, prompt: str, context: dict | None = None) -> str:
# 전처리: 컨텍스트 주입 및 프롬프트 구성
enriched_prompt = self._preprocess(prompt, context)
# 모델 호출 (재시도 포함)
raw_response = self._call_with_retry(enriched_prompt)
# 후처리: 파싱, 검증, 변환
return self._postprocess(raw_response)
def _preprocess(self, prompt: str, context: dict | None) -> str:
if context:
context_str = "\n".join(
f"{k}: {v}" for k, v in context.items()
)
return f"Context:\n{context_str}\n\nQuery: {prompt}"
return prompt
def _call_with_retry(self, prompt: str) -> str:
for attempt in range(self.config.max_retries):
try:
response = self.client.messages.create(
model=self.config.model,
max_tokens=self.config.max_tokens,
messages=[{"role": "user", "content": prompt}],
)
return response.content[0].text
except Exception as e:
if attempt == self.config.max_retries - 1:
raise
continue
def _postprocess(self, response: str) -> str:
# 출력 정제: 불필요한 공백 제거, 포맷 검증
return response.strip()래핑 패턴은 "첫 번째 하네스"로 가장 적합합니다. 단일 모델, 단일 용도의 AI 기능을 빠르게 구현할 때 시작점으로 삼으세요. 복잡해지면 다른 패턴으로 전환하면 됩니다.
웹 프레임워크의 미들웨어와 동일한 개념입니다. 요청과 응답이 일련의 미들웨어 체인을 통과하며, 각 미들웨어가 독립적으로 특정 기능을 담당합니다.
type Context = {
prompt: string
response?: string
metadata: Record<string, unknown>
}
type Middleware = (
ctx: Context,
next: () => Promise<void>
) => Promise<void>
class MiddlewareHarness {
private middlewares: Middleware[] = []
use(middleware: Middleware): this {
this.middlewares.push(middleware)
return this
}
async execute(prompt: string): Promise<string> {
const ctx: Context = { prompt, metadata: {} }
// 미들웨어 체인 구성 (역순으로 감싸기)
let index = 0
const dispatch = async (): Promise<void> => {
if (index < this.middlewares.length) {
const mw = this.middlewares[index++]
await mw(ctx, dispatch)
}
}
await dispatch()
return ctx.response ?? ""
}
}
// 미들웨어 정의
const loggingMiddleware: Middleware = async (ctx, next) => {
const start = Date.now()
console.log(`[REQUEST] ${ctx.prompt.slice(0, 50)}...`)
await next()
console.log(`[RESPONSE] ${Date.now() - start}ms`)
}
const guardrailMiddleware: Middleware = async (ctx, next) => {
// 입력 검증
if (ctx.prompt.includes("ignore previous instructions")) {
ctx.response = "요청을 처리할 수 없습니다."
return // next를 호출하지 않아 체인 중단
}
await next()
// 출력 검증
if (ctx.response && ctx.response.length > 10000) {
ctx.response = ctx.response.slice(0, 10000)
}
}
const modelMiddleware: Middleware = async (ctx, next) => {
ctx.response = await callModel(ctx.prompt)
await next()
}
// 조립
const harness = new MiddlewareHarness()
.use(loggingMiddleware)
.use(guardrailMiddleware)
.use(modelMiddleware)NVIDIA의 NeMo Guardrails가 미들웨어 패턴의 대표적인 구현체입니다. 사용자와 LLM 사이에 프로그래밍 가능한 가드레일 미들웨어 계층을 배치하여 입출력을 제어합니다.
사이드카 패턴(Sidecar Pattern)은 마이크로서비스 아키텍처에서 차용한 패턴입니다. 하네스 기능을 모델 서비스와 별도의 프로세스로 분리하되, 같은 네트워크 컨텍스트에서 실행합니다. 모델 서비스를 수정하지 않고도 하네스 기능을 추가할 수 있다는 것이 핵심 장점입니다.
from fastapi import FastAPI, Request
import httpx
app = FastAPI()
MODEL_SERVICE_URL = "http://localhost:8080"
@app.middleware("http")
async def harness_middleware(request: Request, call_next):
"""사이드카 하네스: 모델 서비스를 수정하지 않고 감싸기"""
# 입력 단계: 요청 가로채기 및 전처리
body = await request.json()
enriched_body = await preprocess(body)
# 모델 서비스로 전달
async with httpx.AsyncClient() as client:
response = await client.post(
f"{MODEL_SERVICE_URL}/generate",
json=enriched_body,
)
# 출력 단계: 응답 후처리 및 메트릭 수집
result = response.json()
processed = await postprocess(result)
await collect_metrics(body, processed)
return processed
async def preprocess(body: dict) -> dict:
"""입력 가드레일 및 컨텍스트 보강"""
# 프롬프트 인젝션 검사
# 컨텍스트 주입
# 토큰 제한 확인
return body
async def postprocess(result: dict) -> dict:
"""출력 검증 및 변환"""
# PII 마스킹
# 출력 스키마 검증
# 품질 점수 계산
return result
async def collect_metrics(request: dict, response: dict):
"""메트릭 수집 및 전송"""
# 토큰 사용량, 지연시간, 비용 등
pass파이프라인 패턴(Pipeline Pattern)은 데이터가 정의된 스테이지를 순차적으로 통과하는 구조입니다. 각 스테이지는 입력을 받아 변환하고 다음 스테이지로 전달합니다. ETL 파이프라인이나 컴파일러 파이프라인과 동일한 개념입니다.
from typing import Callable, TypeVar
from dataclasses import dataclass, field
T = TypeVar("T")
@dataclass
class PipelineContext:
"""파이프라인 전체에서 공유되는 컨텍스트"""
raw_input: str
prompt: str = ""
model_output: str = ""
parsed_output: dict = field(default_factory=dict)
final_output: str = ""
metadata: dict = field(default_factory=dict)
# 타입 안전한 스테이지 정의
Stage = Callable[[PipelineContext], PipelineContext]
def validate_input(ctx: PipelineContext) -> PipelineContext:
"""스테이지 1: 입력 유효성 검사"""
if not ctx.raw_input.strip():
raise ValueError("빈 입력은 허용되지 않습니다")
if len(ctx.raw_input) > 100_000:
raise ValueError("입력이 최대 길이를 초과합니다")
return ctx
def enrich_context(ctx: PipelineContext) -> PipelineContext:
"""스테이지 2: 컨텍스트 보강"""
# RAG 검색, 사용자 프로필 조회 등
ctx.metadata["enriched"] = True
return ctx
def compose_prompt(ctx: PipelineContext) -> PipelineContext:
"""스테이지 3: 최종 프롬프트 조립"""
system_prompt = "당신은 도움이 되는 AI 어시스턴트입니다."
ctx.prompt = f"{system_prompt}\n\n{ctx.raw_input}"
return ctx
def call_model(ctx: PipelineContext) -> PipelineContext:
"""스테이지 4: 모델 호출"""
# 실제로는 API 호출
ctx.model_output = model.generate(ctx.prompt)
return ctx
class Pipeline:
def __init__(self):
self.stages: list[Stage] = []
def add_stage(self, stage: Stage) -> "Pipeline":
self.stages.append(stage)
return self
def execute(self, raw_input: str) -> PipelineContext:
ctx = PipelineContext(raw_input=raw_input)
for stage in self.stages:
ctx = stage(ctx)
return ctx
# 파이프라인 조립
pipeline = (
Pipeline()
.add_stage(validate_input)
.add_stage(enrich_context)
.add_stage(compose_prompt)
.add_stage(call_model)
)이벤트 기반 패턴(Event-Driven Pattern)은 하네스의 각 컴포넌트가 이벤트를 발행하고 구독하는 방식으로 통신합니다. 가장 유연하지만 가장 복잡한 패턴이기도 합니다.
type EventType =
| "request.received"
| "input.validated"
| "context.enriched"
| "model.called"
| "output.validated"
| "response.ready"
| "error.occurred"
interface HarnessEvent {
type: EventType
payload: unknown
timestamp: number
traceId: string
}
type EventHandler = (event: HarnessEvent) => Promise<void>
class EventDrivenHarness {
private handlers = new Map<EventType, EventHandler[]>()
on(eventType: EventType, handler: EventHandler): void {
const existing = this.handlers.get(eventType) ?? []
existing.push(handler)
this.handlers.set(eventType, existing)
}
async emit(event: HarnessEvent): Promise<void> {
const handlers = this.handlers.get(event.type) ?? []
// 동일 이벤트의 핸들러들은 병렬 실행
await Promise.all(handlers.map((h) => h(event)))
}
}
// 핸들러 등록
const harness = new EventDrivenHarness()
harness.on("request.received", async (event) => {
// 입력 검증 후 다음 이벤트 발행
const validated = await validateInput(event.payload)
await harness.emit({
type: "input.validated",
payload: validated,
timestamp: Date.now(),
traceId: event.traceId,
})
})
harness.on("model.called", async (event) => {
// 모델 응답 로깅 (비동기, 메인 흐름 차단 없음)
await logToAnalytics(event)
})
harness.on("model.called", async (event) => {
// 출력 검증 (같은 이벤트에 여러 핸들러)
const validated = await validateOutput(event.payload)
await harness.emit({
type: "output.validated",
payload: validated,
timestamp: Date.now(),
traceId: event.traceId,
})
})CrewAI Flows가 이벤트 기반 패턴의 좋은 예시입니다. 에이전트 워크플로우를 이벤트 기반으로 구성하여, 각 단계의 결과에 따라 동적으로 다음 행동을 결정합니다.
| 기준 | 래핑 | 미들웨어 | 사이드카 | 파이프라인 | 이벤트 기반 |
|---|---|---|---|---|---|
| 구현 난이도 | 낮음 | 중간 | 높음 | 중간 | 높음 |
| 확장성 | 낮음 | 높음 | 높음 | 중간 | 매우 높음 |
| 테스트 용이성 | 낮음 | 높음 | 중간 | 높음 | 중간 |
| 결합도 | 높음 | 중간 | 낮음 | 중간 | 낮음 |
| 적합한 규모 | 소규모 | 중규모 | 대규모 | 중규모 | 대규모 |
실전에서는 단일 패턴만 사용하는 경우가 드뭅니다. 대부분의 프로덕션 시스템은 파이프라인 패턴을 골격으로 하되, 각 스테이지 내부에서 미들웨어 패턴을 사용하고, 모니터링은 이벤트 기반으로 처리하는 식의 조합을 채택합니다.
3장에서는 하네스의 가장 기본적이면서도 핵심적인 기능인 AI 모델 래핑과 입출력 제어를 다룹니다. 모델 추상화 계층 설계, 프롬프트 구성과 컨텍스트 주입, 스키마 기반 출력 제어, 폴백 전략 등 실전 코드와 함께 살펴봅니다.
이 글이 도움이 되셨나요?
관련 주제 더 보기
모델 추상화 계층 설계, 프롬프트 구성과 컨텍스트 주입, 스키마 기반 출력 제어, 폴백 전략 등 AI 모델의 입출력을 체계적으로 관리하는 방법을 다룹니다.
AI 에이전트에서 모델을 감싸는 모든 것, 하네스 엔지니어링의 정의와 등장 배경, 그리고 5가지 핵심 역할을 살펴봅니다.
비결정적 출력 테스트, 스냅샷 테스트, 속성 기반 테스트, 회귀 테스트, 에이전트 행동 테스트 등 AI 시스템 테스트의 핵심 기법을 다룹니다.