Alasan Mengapa Tagihan 60 Juta Won Muncul Saat Antar-Agen Saling Berbicara
TuBrief 편집팀
2026년 9월 13일
0
Computing/Software원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.
커뮤니티의 다른 글
댓글 (0)
Log in to leave a comment
아직 작성된 글이 없습니다
원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.
Log in to leave a comment
아직 작성된 글이 없습니다
Dalam lingkungan internal tempat bot code review tim pengembangan dan agen kueri tim dukungan terhubung, bencana yang paling sering dialami oleh para insinyur bukanlah kurangnya kecerdasan model. Melainkan satu baris pesan kegagalan parsing yang masuk ke antrean asinkron (async queue), serta infinite loop yang berputar sepanjang akhir pekan karena kedua agen saling melempar pertanyaan.
Memperlakukan agen sebagai rekan kerja yang otonom hanya berhasil di lab prompt. Begitu dimasukkan ke infrastruktur internal, agen hanyalah layanan terdistribusi yang mengeluarkan input tidak dapat dipercaya. Metode menulis "Hentikan jika Anda tidak tahu jawabannya" di dalam prompt bahasa alami pasti akan gagal di produksi. Mari kita bahas garis kontrol fisik yang harus dipasang oleh platform engineer langsung pada level antrean (queue) dan gateway.
Jika Anda membiarkan agen berkomunikasi menggunakan campuran bahasa alami atau markdown backtick (json), Anda akan membuang lima hingga enam jam seminggu hanya untuk melacak error parsing. Karena output model yang non-determinis, tanda kutip yang hilang atau sedikit perubahan pada nama field dapat menyebabkan downstream consumer mengalami crash.
Solusinya adalah dengan menegakkan skema yang ketat pada lapisan transmisi, seperti draf Google A2A dan JSON-RPC 2.0. Payload Kafka yang ditransmisikan antar-agen harus melewati field berikut pada saat runtime.
| Nama Field | Tipe | Wajib/Tidak | Tujuan Validasi |
|---|---|---|---|
message_id |
UUIDv7 | Wajib | ID pesan global yang dapat diurutkan berdasarkan waktu |
task_id |
UUIDv4 | Wajib | Unit pelacakan tugas bisnis tunggal |
context_id |
String | Wajib | Pengenal sesi percakapan induk |
sender_id |
String | Wajib | Namespace pengirim (domain:nama_agen) |
receiver_id |
String | Wajib | Namespace penerima (domain:nama_agen) |
hop_count |
Integer | Wajib | Jumlah kumulatif penerusan antar-agen (nilai awal: 0) |
max_hops |
Integer | Wajib | Maksimum jumlah penerusan yang diizinkan (nilai rekomendasi: 5) |
constraints |
Object | Opsional | Batas timeout, anggaran token |
data |
Object | Wajib | Data bisnis terstruktur |
Untuk mencegah insiden di mana proses consumer rusak, interceptor validasi Pydantic harus ditempatkan di titik masuk, dan pesan yang melanggar spesifikasi harus segera didorong ke Dead Letter Queue (DLQ).
`python
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_-]+")
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)
`
Dengan menerapkan pola ini, fenomena poison pill yang membuat consumer tidak responsif akan hilang. Sumber daya teknik yang dihabiskan untuk debugging parsing juga dapat segera dipulihkan.
Momen paling berbahaya dalam pipeline agen adalah ketika dua agen berputar dalam infinite loop sambil memvalidasi output satu sama lain.
Ada kasus nyata yang laporan post-mortemnya dirilis pada Maret 2026. Bot pembuat SQL yang gagal melepaskan batasan foreign key dan bot verifikasi saling melempar kueri tanpa akhir, dan karena aksi itu sendiri berbeda setiap saat, penghitung coba ulang "aksi yang sama" sebanyak 50 kali tidak berfungsi. Keduanya beroperasi tanpa isolasi selama 11 hari (264 jam) dan menghabiskan total biaya API sebesar $47.200 (sekitar 63 juta won). Ketika IAL-Scan menganalisis 6,549 repositori agen open-source, 68 kasus infinite loop yang fatal ditemukan begitu saja di 47 proyek.
Kita tidak boleh mengandalkan kondisi keluar pada prompt. Kill switch fisik harus dipasang pada level API gateway.
| Kriteria Kontrol | Metode Kontrol | Nilai Rekomendasi | Aturan Operasi |
|---|---|---|---|
| Batasan Jumlah Hop | Pelacakan kedalaman panggilan pada header | Maksimal 5 hop | Jika saling mendelegasikan lebih dari 5 kali, kembalikan 503 lalu hentikan |
| Batas Atas Anggaran | Pelacakan biaya kumulatif per transaksi Redis | $10 per tugas | Jika pengeluaran kumulatif melebihi $10, kembalikan 429 dan blokir |
| Batas Coba Ulang | Jumlah pengulangan untuk ID tugas yang sama | Maksimal 3 kali | Meskipun konten berubah, akhiri jika tugas yang sama gagal 3 kali |
Jika Anda menghubungkan X-Agent-Hop-Count dan Redis di middleware gateway, lonjakan pengeluaran dapat dikendalikan di kisaran $10.
`python
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
`
Sangat berbahaya untuk menyerahkan master API key yang memiliki hak tulis repositori Git perusahaan kepada agen analisis kode internal. Hanya dengan satu injeksi prompt atau halusinasi model, kueri penghapusan cabang yang salah bisa terkirim.
Sesuai dengan prinsip Zero Trust NIST SP 800-207, agen harus dicegah agar tidak memiliki kredensial jangka panjang. Saat agen memulai tugas, buat mereka menerbitkan Scoped JWT dengan cakupan sempit yang hanya berlaku selama tepat 300 detik (5 menit) melalui pertukaran token OAuth 2.0 (RFC 8693) dari IdP.
Jika Anda melampirkan Open Policy Agent (OPA) ke sidecar gateway dan mendebarkan kebijakan Rego berikut, infrastruktur akan menolaknya dengan status 403 meskipun agen analisis hanya-baca mengirimkan kueri modifikasi yang berbahaya.
`rego
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)
}
`
Dalam pipeline di mana banyak agen terhubung, jika agen hilir mengalami crash, antrean akan mencoba ulang seluruh pesan. Pada saat ini, agen hulu harus dicegah agar tidak memanggil kembali prompt mahal yang sama.
Patuhi OpenTelemetry GenAI Semantic Conventions untuk mengelompokkan setiap eksekusi agen ke dalam span, dan pasang Redis distributed lock untuk mencegah eksekusi ulang pada tahap yang sama dari akarnya.
| Kunci Atribut Semantik | Tipe | Nilai Contoh | Tujuan Observasi |
|---|---|---|---|
gen_ai.operation.name |
String | invoke_agent |
Membedakan jenis pekerjaan agen |
gen_ai.provider.name |
String | openai |
Latensi dan tingkat error per penyedia |
gen_ai.request.model |
String | gpt-4o |
Identifikasi model yang digunakan |
gen_ai.usage.input_tokens |
Integer | 2048 |
Penyelesaian biaya input per tahap |
gen_ai.usage.output_tokens |
Integer | 512 |
Pelacakan token pembuatan komplesi |
gen_ai.conversation.id |
String | task-session-9821 |
Pelacakan sesi agen secara keseluruhan |
agent.prompt.hash |
String | sha256:7f83b165... |
Manajemen versi prompt input |
`python
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)
`
Bot code review internal dan bot kueri deployment terus menerbitkan kueri standar teknis yang sama dengan sedikit perubahan pada ekspresinya. Cache kecocokan string sederhana tidak berguna jika urutan kata berubah, sehingga panggilan API eksternal tetap terjadi.
Semantic cache berbasis RedisVL harus dipasang di bagian depan, dan jarak cosine harus diikat secara ketat ke 0.1 (kesamaan 0.95 atau lebih tinggi) untuk mencegah distorsi data buku panduan teknis internal.
| Kesamaan Cosine | Jarak Cosine | Hit Rate Cache | Risiko Distorsi Makna | Penggunaan yang Disarankan |
|---|---|---|---|---|
| 0.95 atau lebih | 0.1 atau kurang | 30% ~ 40% | Kurang dari 0.1% | Peraturan internal, spesifikasi API, asisten kode |
| 0.85 ~ 0.94 | 0.1 ~ 0.2 | 50% ~ 70% | Tingkat menengah | Panduan umum dan kueri kenyamanan internal |
| Di bawah 0.80 | Lebih dari 0.2 | 75% atau lebih | Sangat tinggi | Tidak dapat digunakan di lingkungan produksi |
Masalah pengeluaran cache yang salah saat dokumen asli dimodifikasi dapat diselesaikan dengan mendeteksi event perubahan DB menggunakan Debezium CDC dan segera menghapus cache tersebut.
`python
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"
Menempatkan lapisan semantic cache berdasarkan ambang batas jarak 0.1 mengurangi jumlah panggilan API LLM eksternal sekitar 35% dan mempersingkat waktu latensi respons hingga hitungan detik. Alih-alih berdoa sambil mengharapkan kondisi keluar pada prompt, jauh lebih aman untuk mengunggah kode pemblokiran fisik pada tingkat gateway dan antrean.