为什么智能体之间对话会产生 6 千万韩元账单
TuBrief 편집팀
2026년 9월 13일
0
Computing/Software원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.
커뮤니티의 다른 글
댓글 (0)
Log in to leave a comment
아직 작성된 글이 없습니다
원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.
Log in to leave a comment
아직 작성된 글이 없습니다
在将开发团队的代码审查机器人与支持团队的问答智能体进行联动的企业内部环境中，工程师最常经历的灾难并不是模型智能的不足。而是进入异步队列的一行解析失败消息，以及两个智能体互相问答、在整个周末疯狂打转的死循环。
把智能体当作自主同事对待的做法只在提示词实验室里行得通。一旦部署到企业内部基础设施中，智能体就只是一个会输出不可信输入的分布式服务。在自然语言提示词中写下 "如果不知道答案就停止" 的方法在生产环境中注定会失效。本文将探讨平台工程师必须直接在队列和网关层面设立的物理控制线。
如果任由智能体之间使用自然语言或混合 Markdown 反引号（`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 | 可选 | 超时、Token 预算限制 |
data |
Object | 必填 | 规范化的业务数据 |
为了防止消费者进程崩溃的事故，必须在入口处设置 Pydantic 验证拦截器，并将违反规范的消息立即推送到死信队列（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)
`
设置好这个模式后，导致消费者死机的毒丸（Poison Pill）现象就会消失。原本投入到解析调试中的工程资源也能立即收回。
在智能体管道中最危险的时刻，是两个智能体在互相验证对方的输出时陷入无限循环。
有一个在 2026 年 3 月发布事后分析报告的真实案例。未能解除外键约束的 SQL 生成机器人与验证机器人互相不断生成不同的查询并进行乒乓交互，由于每次的操作本身都不同，简单的“相同操作” 50 次重试计数器失去了作用。两者在没有隔离的情况下运行了 11 天（264 小时），消耗了总计 47,200 美元（约 6,300 万韩元）的 API 费用。当 IAL-Scan 分析了 6,549 个开源智能体仓库时，也在 47 个项目中直接发现了 68 起致命的无限循环。
不能信任提示词逃逸条件。必须在 API 网关层面设置物理急停开关（Kill Switch）。
| 控制标准 | 控制方式 | 推荐值 | 动作规则 |
|---|---|---|---|
| 跳数限制 | 追踪头部调用深度 | 最大 5 跳 | 互相委托超过 5 次则返回 503 并中断 |
| 预算上限 | 追踪每个 Redis 事务的累计成本 | 每个任务 $10 | 累计支出超过 10 美元时返回 429 并拦截 |
| 重试限制 | 针对相同任务 ID 的重复次数 | 最大 3 次 | 即使内容改变，相同任务失败 3 次也终止 |
在网关中间件中挂载 X-Agent-Hop-Count 和 Redis，就可以将支出暴增控制在 $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
`
将拥有全公司 Git 仓库写权限的主 API 密钥交给内部代码分析智能体是非常危险的。只要发生一次提示词注入或模型幻觉，就可能导致错误的删除分支查询被执行。
按照 NIST SP 800-207 零信任原则，必须阻止智能体拥有长期凭证。当智能体开始工作时，通过 IdP 的 OAuth 2.0 令牌交换（RFC 8693）颁发一个仅维持 300 秒（5分钟）的、权限范围狭窄的 Scoped JWT。
如果在网关边车（Sidecar）中附加 Open Policy Agent (OPA) 并部署以下 Rego 策略，那么即使只读分析智能体发送了恶意的修改查询，基础设施也会通过 403 将其拒绝。
`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)
}
`
在多个智能体交织的管道中，如果下游智能体崩溃，队列会重试整个消息。此时必须防止上游智能体再次调用相同的昂贵提示词。
遵循 OpenTelemetry GenAI Semantic Conventions 将每个智能体的执行绑定为 Span，并通过 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 |
追踪生成完成的 Token |
gen_ai.conversation.id |
String | task-session-9821 |
追踪整体智能体会话 |
agent.prompt.hash |
String | sha256:7f83b165... |
输入提示词版本管理 |
`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)
`
公司内部的代码审查机器人和部署查询机器人会不断抛出仅更改了表述的相同技术标准查询。简单的字符串匹配缓存在语序改变时形同虚设，因此外部 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 检测数据库变更事件并立即清除缓存来解决。
`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"
设置以距离阈值 0.1 为标准的语义缓存层，可以将外部 LLM API 调用次数减少约 35%语义，响应延迟也能缩短至数秒单位。与其在提示词中祈祷逃逸条件生效，不如在网关和队列层面上架设物理拦截代码要安全得多。