Written with AI assistance from the source video. The video is the authority.
Log in to leave a comment
No posts yet
डेवलपमेंट टीम के कोड रिव्यू बॉट और सपोर्ट टीम क्वेरी एजेंट को आपस में जोड़ने वाले आंतरिक वातावरण में, इंजीनियरों को सबसे अक्सर जिस आपदा का सामना करना पड़ता है, वह मॉडल की बुद्धिमत्ता की कमी नहीं है। यह एसिंक्रोनस क्यु में आई पार्सिंग विफल संदेश की एक लाइन, और एक अनंत लूप है जहां दो एजेंट पूरे सप्ताहांत तक एक-दूसरे को सवाल भेजते रहते हैं।
एजेंटों को स्वायत्त सहयोगियों के रूप में मानना केवल प्रॉम्प्ट प्रयोगशालाओं में काम करता है। जैसे ही आप इसे आंतरिक बुनियादी ढांचे पर तैनात करते हैं, एजेंट केवल एक वितरित सेवा बन जाते हैं जो अविश्वसनीय इनपुट उगलते हैं। प्राकृतिक भाषा के प्रॉम्प्ट के अंदर "यदि उत्तर नहीं पता है तो रुक जाएं" लिखने का तरीका प्रोडक्शन में निश्चित रूप से विफल हो जाता है। आइए उन भौतिक नियंत्रण रेखाओं पर नजर डालते हैं जिन्हें प्लेटफ़ॉर्म इंजीनियरों को सीधे क्यु और गेटवे स्तर पर लागू करना चाहिए।
यदि आप एजेंटों को प्राकृतिक भाषा या मार्कडाउन बैकटिक्स (````json`) को मिलाकर संवाद करने की अनुमति देते हैं, तो आप केवल पार्सिंग त्रुटियों को ट्रैक करने में सप्ताह में पांच से छह घंटे गंवा देंगे। मॉडल के गैर-निश्चयात्मक आउटपुट के कारण, एक उद्धरण गायब हो सकता है या फ़ील्ड का नाम थोड़ा बदल सकता है, जिससे डाउनस्ट्रीम उपभोक्ता क्रैश हो जाएगा।
इसका समाधान Google A2A ड्राफ्ट और JSON-RPC 2.0 की तरह ट्रांसपोर्ट लेयर पर एक सख्त स्कीमा लागू करना है। एजेंटों के बीच प्रेषित Kafka पेलोड को रनटाइम पर निम्नलिखित फ़ील्ड को पास करना होगा।
| फ़ील्ड नाम | प्रकार | आवश्यक है या नहीं | सत्यापन उद्देश्य |
|---|---|---|---|
message_id |
UUIDv7 | आवश्यक | कालక్రमिक रूप से सॉर्ट करने योग्य वैश्विक संदेश आईडी |
task_id |
UUIDv4 | आवश्यक | एकल व्यावसायिक कार्य ट्रैकिंग इकाई |
context_id |
String | आवश्यक | मूल वार्तालाप सत्र पहचानकर्ता |
sender_id |
String | आवश्यक | प्रेषक नामस्थान (domain:agent_name) |
receiver_id |
String | आवश्यक | प्राप्तकर्ता नामस्थान (domain:agent_name) |
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)
`
इस पैटर्न को सेट करने से पॉइजन पिल की समस्या समाप्त हो जाती है जहाँ उपभोक्ता अनुत्तरदायी हो जाते हैं। पार्सिंग डिबगिंग में खर्च होने वाले इंजीनियरिंग संसाधनों को भी तुरंत वापस पाया जा सकता है।
एजेंट पाइपलाइन में सबसे खतरनाक पल तब होता है जब दो एजेंट एक-दूसरे के आउटपुट को सत्यापित करते हुए अनंत लूप में फंस जाते हैं।
मार्च 2026 की एक पोस्ट-मॉर्टम रिपोर्ट के साथ एक वास्तविक मामला सामने आया है। एक SQL जनरेशन बॉट जो फॉरेन की बाधाओं को हल नहीं कर सका और एक सत्यापन बॉट अंतहीन रूप से अलग-अलग क्वेरी उत्पन्न करते हुए पिंग-पोंग खेल रहे थे, और चूंकि एक्शन हर बार अलग था, इसलिए साधारण 'समान एक्शन' पुनरावृत्ति काउंटर 50 काम नहीं आया। दोनों 11 दिनों (264 घंटे) तक बिना आइसोलेशन के चलते रहे और कुल $47,200 (लगभग 6.3 करोड़ वॉन) की API लागत खर्च कर डाली। जब IAL-Scan ने 6,549 ओपन-सोर्स एजेंट रिपॉजिटरी का विश्लेषण किया, तो 47 प्रोजेक्ट्स में 68 गंभीर अनंत लूप वैसे के वैसे पाए गए।
आपको प्रॉम्प्ट एस्केप शर्तों पर भरोसा नहीं करना चाहिए। आपको API गेटवे स्तर पर भौतिक किल स्विच लगाना होगा।
| नियंत्रण मानदंड | नियंत्रण विधि | अनुशंसित मान | संचालन नियम |
|---|---|---|---|
| हॉप्स की सीमा | हेडर की कॉल गहराई का ट्रैकिंग | अधिकतम 5 हॉप्स | यदि 5 से अधिक बार एक-दूसरे को डेलिगेट करते हैं, तो 503 लौटाएं और बंद करें |
| बजट की ऊपरी सीमा | Redis लेनदेन द्वारा संचयी लागत ट्रैकिंग | प्रति कार्य $10 | संचयी खर्च $10 से अधिक होने पर 429 लौटाएं और ब्लॉक करें |
| पुनरावृत्ति सीमा | समान कार्य आईडी के लक्ष्य पर दोहराव की संख्या | अधिकतम 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 कुंजी देना खतरनाक है। प्रॉम्प्ट इंजेक्शन या मॉडल मतिभ्रम (hallucination) के एक ही बार में गलत शाखा को हटाने वाली क्वेरी चल सकती है।
NIST SP 800-207 ज़ीरो ट्रस्ट सिद्धांतों के अनुसार, आपको एजेंटों को दीर्घकालिक क्रेडेंशियल रखने से रोकना चाहिए। जब कोई एजेंट कोई कार्य शुरू करता है, तो उसे IdP में OAuth 2.0 टोकन एक्सचेंज (RFC 8693) के माध्यम से केवल 300 सेकंड (5 मिनट) के लिए वैध संकीर्ण दायरे वाले Scoped JWT को जारी करवाना चाहिए।
यदि आप गेटवे साइडकार में ओपन पॉलिसी एजेंट (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 का पालन करते हुए प्रत्येक एजेंट निष्पादन को स्पैन में समूहित करें, और उसी चरण के पुनः निष्पादन को शुरू से ही ब्लॉक करने के लिए 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 के साथ DB परिवर्तन घटनाओं का पता लगाकर और तुरंत कैश को साफ़ करके हल किया जा सकता है।
`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% कम हो जाती है और प्रतिक्रिया विलंबता समय को भी कुछ सेकंड तक कम किया जा सकता है। प्रॉम्प्ट में भागने की शर्तों की उम्मीद में प्रार्थना करने के बजाय गेटवे और क्यु स्तरों पर भौतिक ब्लॉकिंग कोड अपलोड करना बहुत सुरक्षित है।