에이전트끼리 대화하다 6천만 원 청구서가 나오는 이유
TuBrief 편집팀
2026년 9월 13일
0
컴퓨터/소프트웨어원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.
커뮤니티의 다른 글
댓글 (0)
Log in to leave a comment
아직 작성된 글이 없습니다
원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.
Log in to leave a comment
아직 작성된 글이 없습니다
개발팀의 코드 리뷰 봇과 지원팀 질의 에이전트를 연동해 둔 사내 환경에서, 엔지니어가 가장 자주 겪는 참사는 모델의 지능 부족이 아닙니다. 비동기 큐에 들어간 파싱 실패 메시지 한 줄, 그리고 에이전트 둘이 질문을 주고받다 주말 내내 돌아버린 무한 루프입니다.
에이전트를 자율적인 동료로 대우하는 건 프롬프트 실험실에서나 통합니다. 사내 인프라에 올리는 순간 에이전트는 신뢰할 수 없는 입력을 뱉는 분산 서비스일 뿐입니다. 자연어 프롬프트 안에 "답을 모르면 중단해"라고 적어두는 방식은 프로덕션에서 무조건 깨집니다. 플랫폼 엔지니어가 큐와 게이트웨이 레벨에 직접 걸어야 하는 물리적 통제선들을 짚어봅니다.
에이전트끼리 자연어나 마크다운 백틱(````json`)을 섞어 통신하게 두면, 파싱 에러 추적에만 일주일에 대여섯 시간을 뺏깁니다. 모델의 비결정적 출력 때문에 따옴표 하나가 빠지거나 필드명이 미세하게 바뀌어 다운스트림 컨슈머가 뻗어버립니다.
해결책은 Google A2A 초안과 JSON-RPC 2.0처럼 전송 계층에 엄격한 스키마를 강제하는 것입니다. 에이전트 간 전달되는 Kafka 페이로드는 런타임에 다음 필드를 반드시 통과해야 합니다.
| 필드명 | 타입 | 필수 여부 | 검증 목적 |
|---|---|---|---|
message_id |
UUIDv7 | 필수 | 시간순 정렬이 가능한 전역 메시지 ID |
task_id |
UUIDv4 | 필수 | 단일 비즈니스 작업 추적 단위 |
context_id |
String | 필수 | 상위 대화 세션 식별자 |
sender_id |
String | 필수 | 발신 네임스페이스 (도메인:에이전트명) |
receiver_id |
String | 필수 | 수신 네임스페이스 (도메인:에이전트명) |
hop_count |
Integer | 필수 | 에이전트 간 전달 누적 횟수 (시작값: 0) |
max_hops |
Integer | 필수 | 허용 최대 전달 횟수 (권장값: 5) |
constraints |
Object | 선택 | 타임아웃, 토큰 예산 한도 |
data |
Object | 필수 | 정형화된 비즈니스 데이터 |
컨슈머 프로세스가 깨지는 사고를 막으려면 유입 지점에 Pydantic 검증 인터셉터를 두고, 규격을 어긴 메시지는 즉시 데드 레터 큐(DLQ)로 밀어내야 합니다.
import uuid
from typing import Any, Dict
from pydantic import BaseModel, Field, ValidationError
from confluent_kafka import Consumer, Producer, KafkaError
class TaskConstraints(BaseModel):
timeout_ms: int = Field(default=30000, ge=1000, le=300000)
token_budget: int = Field(default=8000, ge=500, le=128000)
allow_delegation: bool = Field(default=True)
class A2AMessagePayload(BaseModel):
message_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
task_id: str = Field(..., description="비즈니스 작업 고유 ID")
context_id: str = Field(..., description="트랜잭션 세션 식별자")
sender_id: str = Field(..., pattern=r"^[a-z0-9_-]+:[a-z0-9_-]+$")
receiver_id: str = Field(..., pattern=r"^[a-z0-9_-]+:[a-z0-9_-]+$")
hop_count: int = Field(default=0, ge=0)
max_hops: int = Field(default=5, ge=1, le=10)
constraints: TaskConstraints = Field(default_factory=TaskConstraints)
data: Dict[str, Any] = Field(..., description="비즈니스 페이로드")
class ResilientAgentConsumer:
def __init__(self, kafka_conf: dict, main_topic: str, dlq_topic: str):
self.consumer = Consumer(kafka_conf)
self.producer = Producer({"bootstrap.servers": kafka_conf["bootstrap.servers"]})
self.main_topic = main_topic
self.dlq_topic = dlq_topic
self.consumer.subscribe([self.main_topic])
def route_to_dlq(self, raw_bytes: bytes, reason: str):
headers = [("dlq_error", reason.encode("utf-8")), ("origin_topic", self.main_topic.encode("utf-8"))]
self.producer.produce(topic=self.dlq_topic, value=raw_bytes, headers=headers)
self.producer.flush()
def process_events(self, dispatch_fn):
msg = self.consumer.poll(timeout=1.0)
if msg is None:
return
if msg.error():
if msg.error().code() != KafkaError._PARTITION_EOF:
self.route_to_dlq(msg.value() or b"", str(msg.error()))
return
try:
validated = A2AMessagePayload.model_validate_json(msg.value().decode("utf-8"))
except (ValidationError, UnicodeDecodeError) as err:
self.route_to_dlq(msg.value(), f"SCHEMA_VALIDATION_ERROR: {str(err)}")
self.consumer.commit(msg)
return
try:
dispatch_fn(validated)
self.consumer.commit(msg)
except Exception as exec_err:
self.route_to_dlq(msg.value(), f"EXECUTION_ERROR: {str(exec_err)}")
self.consumer.commit(msg)
이 패턴을 걸어두면 컨슈머가 먹통이 되는 포이즌 필 현상이 사라집니다. 파싱 디버깅에 쏟던 엔지니어링 리소스도 즉시 회수할 수 있습니다.
에이전트 파이프라인에서 가장 위험한 순간은 두 에이전트가 상대의 출력을 검증하며 무한 루프를 돌 때입니다.
2026년 3월 사후 분석 보고서가 나온 실제 사례가 있습니다. 외래 키 제약을 풀지 못한 SQL 생성 봇과 검증 봇이 서로 다른 쿼리를 끝없이 생성하며 핑퐁을 쳤고, 액션 자체가 매번 달랐던 탓에 단순 '동일 액션' 재시도 카운터 50회가 먹히지 않았습니다. 둘은 11일(264시간) 동안 격리 없이 가동되었고 총 47,200달러(약 6,300만 원)의 API 비용을 태웠습니다. IAL-Scan이 오픈소스 에이전트 저장소 6,549개를 분석했을 때도 47개 프로젝트에서 68건의 치명적 무한 루프가 그대로 발견되었습니다.
프롬프트 탈출 조건을 믿으면 안 됩니다. API 게이트웨이 레벨에서 물리적인 킬 스위치를 걸어야 합니다.
| 통제 기준 | 제어 방식 | 권장 값 | 동작 규칙 |
|---|---|---|---|
| 홉 수 제한 | 헤더의 호출 깊이 추적 | 최대 5홉 | 5회 넘게 서로를 위임하면 503 반환 후 중단 |
| 예산 상한 | Redis 트랜잭션별 누적 비용 추적 | 태스크당 $10 | 누적 지출 10달러 초과 시 429 반환 및 차단 |
| 재시도 한도 | 동일 작업 ID 대상 반복 횟수 | 최대 3회 | 내용이 바뀌어도 동일 작업 3회 실패 시 종료 |
게이트웨이 미들웨어에서 X-Agent-Hop-Count와 Redis를 물려두면 지출 폭증을 $10 선에서 통제할 수 있습니다.
from fastapi import FastAPI, Request, Response, status
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
import redis.asyncio as redis
import logging
logger = logging.getLogger("AgentCircuitBreaker")
class AgentGovernanceMiddleware(BaseHTTPMiddleware):
def __init__(self, app: FastAPI, redis_pool: redis.Redis, max_hops: int = 5, cost_limit_usd: float = 10.0):
super().__init__(app)
self.redis = redis_pool
self.max_hops = max_hops
self.cost_limit_usd = cost_limit_usd
self.token_cost_ratio = 0.000015 # 1,000토큰당 $0.015 기준 계산
async def dispatch(self, request: Request, call_next) -> Response:
trace_id = request.headers.get("X-Trace-ID") or request.headers.get("traceparent", "trace-root")
current_hops = int(request.headers.get("X-Agent-Hop-Count", "0"))
if current_hops >= self.max_hops:
logger.error(f"홉 한도 초과 차단: trace_id={trace_id}, hops={current_hops}")
return JSONResponse(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
content={"error": "CIRCUIT_BREAKER_HOP_LIMIT_EXCEEDED", "trace_id": trace_id}
)
cost_key = f"governance:cost:{trace_id}"
spent_cost_raw = await self.redis.get(cost_key)
accumulated_cost = float(spent_cost_raw.decode("utf-8")) if spent_cost_raw else 0.0
if accumulated_cost >= self.cost_limit_usd:
logger.error(f"예산 초과 차단: trace_id={trace_id}, spent=${accumulated_cost}")
return JSONResponse(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
content={"error": "CIRCUIT_BREAKER_BUDGET_EXHAUSTED", "trace_id": trace_id}
)
custom_headers = dict(request.scope["headers"])
custom_headers[b"x-agent-hop-count"] = str(current_hops + 1).encode("utf-8")
request.scope["headers"] = list(custom_headers.items())
response = await call_next(request)
consumed_tokens_hdr = response.headers.get("X-LLM-Tokens-Consumed")
if consumed_tokens_hdr:
incremental_cost = int(consumed_tokens_hdr) * self.token_cost_ratio
await self.redis.incrbyfloat(cost_key, incremental_cost)
await self.redis.expire(cost_key, 3600)
return response
사내 코드 분석 에이전트에게 전사 Git 저장소 쓰기 권한이 있는 마스터 API 키를 넘겨주는 것은 위험합니다. 프롬프트 주입이나 모델 환각 한 번에 잘못된 브랜치 삭제 쿼리가 날아갈 수 있습니다.
NIST SP 800-207 제로 트러스트 원칙대로 에이전트는 장기 자격 증명을 가지지 못하게 막아야 합니다. 에이전트가 작업을 시작할 때 IdP에서 OAuth 2.0 토큰 교환(RFC 8693)으로 딱 300초(5분) 동안 유지되는 좁은 범위의 Scoped JWT를 발급받게 만듭니다.
게이트웨이 사이드카에 Open Policy Agent(OPA)를 붙이고 다음 Rego 정책을 배포하면, 읽기 전용 분석 에이전트가 악의적인 변경 쿼리를 쏘더라도 인프라가 403으로 쳐냅니다.
package agent.authz
import future.keywords.in
default allow = false
required_perm_map := {
"GET": "read",
"HEAD": "read",
"POST": "write",
"PUT": "write",
"PATCH": "write",
"DELETE": "admin"
}
allow {
input.token.payload.exp > time.now_ns() / 1000000000
startswith(input.token.payload.sub, "agent:")
input.token.payload.aud == "enterprise-internal-api"
required_perm := required_perm_map[input.http_method]
expected_scope := sprintf("%s:%s", [input.resource_type, required_perm])
expected_scope in input.token.payload.scopes
not is_forbidden_mutation(input.token.payload.role, input.path)
}
is_forbidden_mutation(role, path) {
role == "readonly_sweeper"
regex.match("^/.*/(mutate|delete|drop|update|write)$", path)
}
에이전트가 여럿 엮인 파이프라인에서 하류 에이전트가 뻗으면 큐가 전체 메시지를 재시도합니다. 이때 상류 에이전트가 똑같은 비싼 프롬프트를 다시 호출하지 않도록 막아야 합니다.
OpenTelemetry GenAI Semantic Conventions를 준수해 각 에이전트 실행을 스팬으로 묶고, Redis 분산 락을 걸어 동일 단계 재실행을 원천 차단합니다.
| 시맨틱 속성 키 | 타입 | 예시 값 | 관측 목적 |
|---|---|---|---|
gen_ai.operation.name |
String | invoke_agent |
에이전트 작업 유형 구분 |
gen_ai.provider.name |
String | openai |
공급자별 레이턴시 및 에러율 |
gen_ai.request.model |
String | gpt-4o |
사용 모델 식별 |
gen_ai.usage.input_tokens |
Integer | 2048 |
단계별 입력 비용 정산 |
gen_ai.usage.output_tokens |
Integer | 512 |
완성 생성 토큰 추적 |
gen_ai.conversation.id |
String | task-session-9821 |
전체 에이전트 세션 추적 |
agent.prompt.hash |
String | sha256:7f83b165... |
입력 프롬프트 버전 관리 |
import hashlib
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode
import redis.asyncio as redis
tracer = trace.get_tracer("agent.pipeline.worker", "1.0.0")
async def execute_agent_step_idempotent(task_payload: dict, redis_conn: redis.Redis, llm_gateway_client) -> dict:
task_id = task_payload["task_id"]
step_id = task_payload["step_id"]
prompt = task_payload["prompt"]
idempotency_key = f"step:result:{task_id}:{step_id}"
lock_key = f"lock:step:{task_id}:{step_id}"
acquired = await redis_conn.set(lock_key, "processing", nx=True, ex=120)
if not acquired:
raise RuntimeError(f"Step {step_id} for Task {task_id} is already in progress.")
try:
cached_result = await redis_conn.get(idempotency_key)
if cached_result:
return {"status": "CACHED", "output": cached_result.decode("utf-8")}
prompt_hash = hashlib.sha256(prompt.encode("utf-8")).hexdigest()
with tracer.start_as_current_span(f"step_{step_id}") as span:
span.set_attribute("gen_ai.operation.name", "invoke_agent")
span.set_attribute("gen_ai.provider.name", "openai")
span.set_attribute("gen_ai.request.model", "gpt-4o")
span.set_attribute("gen_ai.conversation.id", task_payload["context_id"])
span.set_attribute("agent.prompt.hash", prompt_hash)
try:
inference_resp = await llm_gateway_client.generate(prompt)
span.set_attribute("gen_ai.usage.input_tokens", inference_resp.prompt_tokens)
span.set_attribute("gen_ai.usage.output_tokens", inference_resp.completion_tokens)
span.set_status(Status(StatusCode.OK))
await redis_conn.set(idempotency_key, inference_resp.content, ex=86400)
return {"status": "SUCCESS", "output": inference_resp.content}
except Exception as exc:
span.record_exception(exc)
span.set_status(Status(StatusCode.ERROR, str(exc)))
raise exc
finally:
await redis_conn.delete(lock_key)
사내 코드 리뷰 봇과 배포 질의 봇은 표현만 살짝 바뀐 동일한 기술 표준 질의를 계속 던집니다. 단순 문자열 일치 캐시는 어순이 바뀌면 무용지물이라 외부 API 호출이 그대로 발생합니다.
RedisVL 기반 시맨틱 캐시를 앞단에 붙이고, 사내 기술 핸드북 데이터의 왜곡을 방지하기 위해 코사인 거리를 0.1(유사도 0.95 이상)로 타이트하게 묶어야 합니다.
| 코사인 유사도 | 코사인 거리 | 캐시 적중률 | 의미 왜곡 위험 | 권장 용도 |
|---|---|---|---|---|
| 0.95 이상 | 0.1 이하 | 30% ~ 40% | 0.1% 미만 | 사내 규정, API 명세, 코드 어시스턴트 |
| 0.85 ~ 0.94 | 0.1 ~ 0.2 | 50% ~ 70% | 중간 수준 | 일반 안내 및 사내 편의 질의 |
| 0.80 미만 | 0.2 초과 | 75% 이상 | 매우 높음 | 운영 환경 사용 불가 |
원본 문서가 수정되었을 때 잘못된 캐시를 뱉는 문제는 Debezium CDC로 DB 변경 이벤트를 감지해 캐시를 즉시 지우면 해결됩니다.
from fastapi import FastAPI, BackgroundTasks
from redisvl.extensions.cache.llm import SemanticCache
from redisvl.utils.vectorize import OpenAITextVectorizer
app = FastAPI()
vectorizer = OpenAITextVectorizer(model="text-embedding-3-small")
semantic_cache = SemanticCache(
redis_url="redis://localhost:6379",
distance_threshold=0.1, # 코사인 유사도 0.95 이상만 적중
vectorizer=vectorizer,
ttl=86400
)
@app.post("/v1/agent/query")
async def execute_agent_query(payload: dict):
query_text = payload["query"]
hit = semantic_cache.check(prompt=query_text)
if hit:
return {
"source": "SEMANTIC_CACHE",
"distance": hit[0].get("vector_distance"),
"response": hit[0]["response"]
}
llm_result = await call_upstream_llm(query_text)
semantic_cache.store(
prompt=query_text,
response=llm_result,
metadata={"domain": payload.get("domain", "general")}
)
return {"source": "LLM_GENERATED", "response": llm_result}
@app.post("/v1/cache/invalidate")
async def handle_cdc_invalidation(event: dict, background_tasks: BackgroundTasks):
table = event.get("source", {}).get("table")
op = event.get("op")
if table == "engineering_handbook" and op in ["u", "d"]:
background_tasks.add_task(purge_cache_index)
return {"status": "INVALIDATION_TRIGGERED"}
async def purge_cache_index():
semantic_cache.clear()
async def call_upstream_llm(prompt: str) -> str:
return "LLM Inference Result"
거리 임계값 0.1 기준의 시맨틱 캐시 레이어를 두면 외부 LLM API 호출 횟수를 약 35% 줄이고 응답 지연 시간도 수 초 단위로 단축됩니다. 프롬프트에 탈출 조건을 빌며 기도하는 대신 게이트웨이와 큐 레벨에 물리적인 차단 코드를 올리는 편이 훨씬 안전합니다.