Written with AI assistance from the source video. The video is the authority.
Log in to leave a comment
No posts yet
In an internal environment where a development team's code review bot and a support team's query agent are integrated, the disaster that engineers experience most frequently is not a lack of model intelligence. It is a single line of a parsing failure message entering an asynchronous queue, followed by an infinite loop where two agents ping-pong questions back and forth all weekend long.
Treating agents as autonomous colleagues only works in a prompt lab. The moment you deploy them to internal infrastructure, agents are merely distributed services that emit untrusted inputs. Writing "stop if you don't know the answer" inside a natural language prompt will inevitably fail in production. Let's look at the physical control lines that platform engineers must directly implement at the queue and gateway levels.
If you let agents communicate using mixed natural language or Markdown backticks (`json), you will waste five or six hours a week just tracking down parsing errors. Due to the model's non-deterministic output, a missing quotation mark or a slightly altered field name will crash downstream consumers.
The solution is to enforce a strict schema at the transport layer, similar to the Google A2A draft and JSON-RPC 2.0. Kafka payloads transmitted between agents must pass the following fields at runtime:
| Field Name | Type | Required? | Validation Purpose |
|---|---|---|---|
message_id |
UUIDv7 | Required | Globally unique message ID orderable by time |
task_id |
UUIDv4 | Required | Single business task tracking unit |
context_id |
String | Required | Parent conversation session identifier |
sender_id |
String | Required | Sender namespace (domain:agent_name) |
receiver_id |
String | Required | Receiver namespace (domain:agent_name) |
hop_count |
Integer | Required | Cumulative transmission count between agents (Start: 0) |
max_hops |
Integer | Required | Maximum allowed transmission count (Recommended: 5) |
constraints |
Object | Optional | Timeout, token budget limits |
data |
Object | Required | Structured business data |
To prevent consumer crashes, place a Pydantic validation interceptor at the ingestion point, and immediately push non-compliant messages to a 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="Unique business task ID")
context_id: str = Field(..., description="Transaction session identifier")
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="Business payload")
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)
`
By applying this pattern, poison pill phenomena where consumers freeze disappear. Engineering resources spent on parsing debugging can also be instantly recovered.
The most dangerous moment in an agent pipeline is when two agents enter an infinite loop while validating each other's outputs.
There is a real-world case analyzed in a post-mortem report from March 2026. An SQL generation bot that failed to resolve a foreign key constraint and a validation bot ping-ponged by endlessly generating different queries. Because the actions themselves differed each time, a simple 'same-action' retry counter of 50 failed to trigger. The two ran unisolated for 11 days (264 hours) and burned a total of $47,200 (approx. 63 million KRW) in API costs. When IAL-Scan analyzed 6,549 open-source agent repositories, 68 critical infinite loops were discovered as-is across 47 projects.
You cannot rely on prompt escape conditions. A physical kill switch must be implemented at the API gateway level.
| Control Criterion | Control Mechanism | Recommended Value | Operating Rule |
|---|---|---|---|
| Hop Limit | Header-based call depth tracking | Max 5 hops | Returns 503 and stops if delegated across each other more than 5 times |
| Budget Ceiling | Cumulative cost tracking per Redis transaction | $10 per task | Returns 429 and blocks when cumulative spending exceeds $10 |
| Retry Limit | Repetition count targeting the same task ID | Max 3 times | Terminates after 3 failures for the same task even if content changes |
Wiring X-Agent-Hop-Count and Redis into the gateway middleware allows you to control spending surges right around the $10 mark.
`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 # Calculated based on $0.015 per 1,000 tokens
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"Hop limit exceeded block: 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"Budget exceeded block: 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
`
Handing over a master API key with write permissions for company-wide Git repositories to an internal code analysis bot is dangerous. A single prompt injection or model hallucination can trigger an erroneous branch deletion query.
In accordance with the NIST SP 800-207 Zero Trust principles, agents must be prevented from holding long-term credentials. When an agent starts a task, have it obtain a narrowly scoped JWT that lasts for exactly 300 seconds (5 minutes) through OAuth 2.0 Token Exchange (RFC 8693) from the IdP.
By attaching Open Policy Agent (OPA) to the gateway sidecar and deploying the following Rego policy, even if a read-only analysis agent fires a malicious mutation query, the infrastructure will reject it with a 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)
}
`
In a multi-agent pipeline, if a downstream agent crashes, the queue retries the entire message. In this case, you must prevent upstream agents from re-invoking the exact same expensive prompt.
Complying with OpenTelemetry GenAI Semantic Conventions, group each agent execution into a span, and place a Redis distributed lock to fundamentally block re-execution of the same stage.
| Semantic Attribute Key | Type | Example Value | Observability Purpose |
|---|---|---|---|
gen_ai.operation.name |
String | invoke_agent |
Differentiate agent task types |
gen_ai.provider.name |
String | openai |
Latency and error rate by provider |
gen_ai.request.model |
String | gpt-4o |
Identify the model used |
gen_ai.usage.input_tokens |
Integer | 2048 |
Step-by-step input cost settlement |
gen_ai.usage.output_tokens |
Integer | 512 |
Track generated completion tokens |
gen_ai.conversation.id |
String | task-session-9821 |
Overall agent session tracking |
agent.prompt.hash |
String | sha256:7f83b165... |
Input prompt version management |
`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)
`
Internal code review bots and deployment query bots continuously fire identical technical standard queries with only slightly altered wording. Simple string-matching caches are useless if word order changes, leading to the same external API calls being generated.
A RedisVL-based semantic cache should be placed in front, and the cosine distance must be tied tightly to 0.1 (similarity of 0.95 or higher) to prevent distortion of internal technical handbook data.
| Cosine Similarity | Cosine Distance | Cache Hit Rate | Semantic Distortion Risk | Recommended Use Case |
|---|---|---|---|---|
| 0.95 or higher | 0.1 or lower | 30% ~ 40% | Less than 0.1% | Internal policies, API specs, code assistants |
| 0.85 ~ 0.94 | 0.1 ~ 0.2 | 50% ~ 70% | Medium level | General guides and internal convenience queries |
| Below 0.80 | Above 0.2 | 75% or higher | Very high | Unusable in production environments |
The issue of spitting out stale cache when the original document is modified can be resolved by detecting DB change events via Debezium CDC and immediately purging the cache.
`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, # Hits only for cosine similarity >= 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"
Placing a semantic cache layer based on a 0.1 distance threshold reduces external LLM API calls by about 35% and also shortens response latencies to the scale of seconds. Rather than praying for escape conditions in prompts, putting physical blocking code at the gateway and queue levels is much safer.