لماذا تتسبب محادثات الوكلاء بين بعضهم في صدور فاتورة بقيمة 60 مليون وون
TuBrief 편집팀
2026년 9월 13일
0
Computing/Software원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.
커뮤니티의 다른 글
댓글 (0)
Log in to leave a comment
아직 작성된 글이 없습니다
원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.
Log in to leave a comment
아직 작성된 글이 없습니다
في بيئة العمل الداخلية حيث يتم ربط بوت مراجعة الكود الخاص بفريق التطوير مع وكيل استفسارات فريق الدعم، فإن الكارثة الأكثر شيوعاً التي يواجهها المهندسون ليست نقص ذكاء النموذج. بل هي سطر واحد من رسالة فشل التحليل التي تدخل في قائمة الانتظار غير المتزامنة، وحلقة لا نهائية تدور طوال عطلة نهاية الأسبوع بسبب تبادل الأسئلة بين وكيلين.
تعامل الوكلاء كزملاء مستقلين لا ينجح إلا في مختبرات النماذج. بمجرد رفعهم إلى البنية التحتية الداخلية، يصبح الوكلاء مجرد خدمة موزعّة تُخرج مدخلات غير موثوقة. طريقة كتابة "توقف إذا لم تعرف الإجابة" داخل توجيهات اللغة الطبيعية ستنكسر حتماً في مرحلة الإنتاج. سنستعرض فيما يلي خطوط التحكم الفيزيائية التي يجب على مهندسي المنصات وضعها مباشرة على مستوى قائمة الانتظار والبوابة.
إذا تركت الوكلاء يتواصلون باستخدام اللغة الطبيعية أو باكتيكس Markdown المختلطة (json), فستضيع خمس إلى ست ساعات كل أسبوع فقط في تتبع أخطاء التحليل. بسبب مخرجات النموذج غير الحتمية، قد تفقد علامة اقتباس واحدة أو يتغير اسم الحقل قليلاً، مما يؤدي إلى تعطل المستهلك التابع (downstream consumer).
الحل هو فرض مخطط صارم على طبقة النقل، مثل مسودة Google A2A و JSON-RPC 2.0. يجب أن تمر حمولة Kafka المنقولة بين الوكلاء عبر الحقول التالية أثناء التشغيل:
| اسم الحقل | النوع | الإلزامية | الغرض من التحقق |
|---|---|---|---|
message_id |
UUIDv7 | إلزامي | معرف رسالة عام يمكن فرزه زمنياً |
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).
`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. حيث تبادل بوت لتوليد SQL عجز عن حل قيود المفتاح الخارجي وبوت التحقق إرسال الاستفسارات بلا نهاية بتنسيق "بينج-بونج"، ونظراً لأن الإجراء نفسه كان يختلف في كل مرة، لم تنجح عدادات إعادة المحاولة البسيطة ذات "الإجراء المتطابق" البالغة 50 مرة. عمل الاثنان لمدة 11 يوماً (264 ساعة) دون عزل، واستهلكا تكاليف API بقيمة إجمالية قدرها 47,200 دولار (حوالي 63 مليون وون). وعندما قامت IAL-Scan بتحليل 6,549 مستودع وكيل مفتوح المصدر، وُجدت 68 حالة من الحلقات اللانهائية الحرجة في 47 مشروعاً.
لا يجب الاعتماد على شروط الهروب في التوجيهات (Prompts). يجب تعيين مفتاح إيقاف فيزيائي على مستوى بوابة API.
| معيار التحكم | طريقة التحكم | القيمة الموصى بها | قواعد التشغيل |
|---|---|---|---|
| حدد عدد النقزات | تتبع عمق الاستدعاء في الترويسة (Header) | بحد أقصى 5 نقزات | إرجاع 503 وإيقاف التشغيل إذا تم تفويض أكثر من 5 مرات |
| سقف الميزانية | تتبع التكلفة التراكمية لكل معاملة في Redis | 10 دولارات لكل مهمة | حظر وإرجاع 429 عند تجاوز الإنفاق التراكمي 10 دولارات |
| حد إعادة المحاولة | عدد مرات التكرار لنفس معرف المهمة | بحد أقصى 3 مرات | الإنهاء بعد 3 حالات فشل متتالية لنفس المهمة حتى لو تغير المحتوى |
من خلال ربط X-Agent-Hop-Count و Redis في وسيط بوابة API، يمكنك التحكم في الانفجار الهائل للإنفاق عند حدود 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 # الحساب بناءً على 0.015 دولار لكل 1000 رمز
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
`
من الخطورة بمكان منح وكيل تحليل الشفرات الخاص بالشركة مفتاح API رئيسي يمتلك صلاحية الكتابة على مستودع Git الخاص بالشركة بأكملها. يمكن لحقن توجيه واحد أو هلوسة واحدة للنموذج أن تؤدي إلى إرسال استعلام حذف فرع خاطئ.
وفقاً لمبادئ الثقة المعدومة NIST SP 800-207، يجب منع الوكلاء من امتلاك بيانات اعتماد طويلة الأجل. عندما يبدأ الوكيل مهمة ما، يتم جعله يحصل على رمز JWT محدّد النطاق يُحافظ عليه لمدة 300 ثانية فقط (5 دقائق) من خلال تبادل رموز OAuth 2.0 (RFC 8693) في نظام إدارة الهوية (IdP).
من خلال إرفاق سياسة Open Policy Agent (OPA) بالوكيل الجانبي للبوابة (Gateway Sidecar) ونشر سياسة 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 الدلالية، يتم ربط تشغيل كل وكيل في مقطع (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 |
تتبع رموز توليد المخرجات |
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, # 적중 chỉ khi tương đồng cô-sin >= 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، يتم تقليل عدد استدعاءات API الخارجية للنماذج اللغوية الكبيرة بحوالي 35%، ويتم تقصير وقت الاستجابة إلى ثوانٍ معدودة. بدلاً من الاعتماد على صلوات شروط الهروب في التوجيهات، فإن رفع رموز الحظر الفيزيائية على مستوى البوابة وقوائم الانتظار يعتبر أمراً آمناً بكثير.