Building an External Control Plane for LLM Agents

dio.""" def init(self, languages: list[str] = None) -> None: self._analyzer = AnalyzerEngine() self._anonymizer = AnonymizerEngine() self._languages = languages or ["en"] self._target_entities = [ "PERSON", "EMAIL_ADDRESS", "PHONE_NUMBER", "CREDIT_CARD", "IBAN_CODE", "IP_ADDRESS" ] self._replacement_map = { "PERSON": "[REDACTED_USER]", "EMAIL_ADDRESS": "[REDACTED_EMAIL]", "PHONE_NUMBER": "[REDACTED_PHONE]", "CREDIT_CARD": "[REDACTED_CARD]", "IBAN_CODE": "[REDACTED_IBAN]", "IP_ADDRESS": "[REDACTED_IP]", }

def sanitize(self, text: str) -> str:
    all_findings = []
    for lang in self._languages:
        findings = self._analyzer.analyze(
            text=text,
            language=lang,
            entities=self._target_entities,
            score_threshold=0.65,
        )
        all_findings.extend(findings)

    if not all_findings:
        return text

    for finding in all_findings:
        PII_REDACTED.labels(entity_category=finding.entity_type).inc()
        logger.debug("PII detected: type=%s confidence=%.2f", finding.entity_type, finding.score)

    sanitized = self._anonymizer.anonymize(
        text=text,
        analyzer_results=all_findings,
        operators={
            entity: OperatorConfig("replace", {"new_value": replacement})
            for entity, replacement in self._replacement_map.items()
        },
    )
    return sanitized.text

class SafetyInterceptor(BaseHTTPMiddleware): """External control plane middleware for LLM agent responses."""

def __init__(self, app, languages: list[str] = None) -> None:
    super().__init__(app)
    self._compliance = ComplianceEngine(FORBIDDEN_PATTERNS)
    self._redactor = DataRedactor(languages)

async def dispatch(self, request: Request, call_next) -> Response:
    start = time.perf_counter()
    response = await call_next(request)

    # Reassemble streamed body with memory guard
    body_chunks = []
    total_size = 0
    async for chunk in response.body_iterator:
        total_size += len(chunk)
        if total_size > MAX_RESPONSE_BYTES:
            logger.warning("Response exceeds size limit, terminating buffer")
            return JSONResponse(
                status_code=413,
                content={"error": "payload_too_large", "limit_bytes": MAX_RESPONSE_BYTES}
            )
        body_chunks.append(chunk)
    
    raw_payload = b"".join(body_chunks).decode("utf-8", errors="replace")

    # Run CPU-bound NLP analysis off the event loop
    loop = asyncio.get_event_loop()
    violation = await loop.run_in_executor(None, self._compliance.evaluate, raw_payload)
    
    if violation:
        POLICY_BLOCKED.labels(rule_type=violation).inc()
        logger.info("Policy violation intercepted: %s", violation)
        return JSONResponse(
            status_code=403,
            content={"error": "compliance_block", "reason": violation}
        )

    sanitized_payload = await loop.run_in_executor(None, self._redactor.sanitize, raw_payload)

    latency = time.perf_counter() - start
    LATENCY_HISTOGRAM.observe(latency)
    logger.info("Agent response validated: %.3fs | %d chars", latency, len(sanitized_payload))

    return Response(
        content=sanitized_payload.encode("utf-8"),
        status_code=response.status_code,
        headers=dict(response.headers),
        media_type=response.media_type,
    )

### Architecture Decisions & Rationale

1. **External Middleware over Agent Logic**: Embedding safety checks inside agent loops creates coupling. If you swap models or refactor orchestration, guardrails break. Middleware enforces policy at the transport boundary, guaranteeing execution regardless of internal routing.
2. **Thread Pool for NLP Analysis**: Presidio's spaCy models perform CPU-intensive tokenization and entity recognition. Running them directly in an async handler blocks the event loop, degrading throughput for all concurrent requests. `run_in_executor` isolates CPU work while preserving async I/O performance.
3. **Hard Payload Limit**: Streaming responses can theoretically grow indefinitely. A 2MB buffer limit prevents memory exhaustion attacks and forces agents to paginate or truncate outputs, which aligns with production API design standards.
4. **Metric-Driven Observability**: Histograms capture latency distribution, enabling p95/p99 alerting. Counters track violation trends. Both integrate natively with Prometheus/Grafana stacks, providing audit-ready telemetry without custom logging pipelines.
5. **Header & Status Preservation**: Reconstructing the response without copying metadata strips `Content-Type`, `Cache-Control`, and error codes. Explicitly forwarding these fields maintains HTTP contract integrity for downstream clients.

## Pitfall Guide

### 1. Unbounded Response Buffering
**Explanation**: Collecting streamed chunks without a size limit allows malicious or malfunctioning agents to allocate gigabytes of RAM, triggering OOM kills.
**Fix**: Implement a strict byte counter during chunk iteration. Terminate and return `413 Payload Too Large` when the threshold is crossed.

### 2. Blocking the Async Event Loop
**Explanation**: Running NLP models or heavy regex operations directly in `dispatch()` stalls the asyncio loop, causing cascading timeouts across all endpoints.
**Fix**: Offload CPU-bound analysis to `asyncio.to_thread()` or `loop.run_in_executor()`. Keep I/O operations strictly async.

### 3. Middleware Registration Order Blindness
**Explanation**: FastAPI executes middleware in reverse registration order (LIFO). Placing safety checks before authentication wastes compute on unauthorized requests.
**Fix**: Register authentication/authorization middleware last so it executes first. Follow with safety interceptors, then logging/telemetry.

### 4. Logging Raw Sensitive Values
**Explanation**: Accidentally logging PII during debugging or error handling violates GDPR/CCPA and creates audit liabilities.
**Fix**: Log only entity types, confidence scores, and redaction counts. Never log the original payload or matched substrings. Use structured logging with explicit PII filters.

### 5. Regex-Only PII Detection
**Explanation**: Regular expressions cannot distinguish between a credit card number and a product serial number, generating high false-positive rates.
**Fix**: Use context-aware NLP engines like Presidio that leverage linguistic patterns, surrounding tokens, and validation checksums (Luhn, IBAN) to reduce false positives.

### 6. Dropping Original Response Metadata
**Explanation**: Returning a new `Response` object without copying headers and status codes strips `Content-Type`, `ETag`, and error states, breaking client expectations.
**Fix**: Explicitly pass `status_code=response.status_code`, `headers=dict(response.headers)`, and `media_type=response.media_type` when reconstructing the response.

### 7. Ignoring Streaming/SSE Responses
**Explanation**: Server-Sent Events and WebSocket upgrades cannot be buffered synchronously. Applying byte-level reconstruction breaks real-time feeds.
**Fix**: Detect `media_type` or `upgrade` headers early. Bypass buffering for streaming endpoints and apply chunk-level validation or skip sanitization for real-time UI updates.

## Production Bundle

### Action Checklist
- [ ] Define policy blocklist and PII entity targets in a version-controlled configuration file
- [ ] Implement payload size limits and early termination logic in the interceptor
- [ ] Offload NLP analysis to thread pools to prevent event loop blocking
- [ ] Register authentication middleware last to ensure LIFO execution order
- [ ] Configure Prometheus histograms for latency distribution and counters for violation tracking
- [ ] Add OpenTelemetry instrumentation to trace agent tool calls and middleware spans
- [ ] Validate header and status code preservation during response reconstruction
- [ ] Establish alerting thresholds for p95 latency spikes and policy violation rate increases

### Decision Matrix

| Scenario | Recommended Approach | Why | Cost Impact |
|----------|---------------------|-----|-------------|
| Single agent, low compliance requirements | Prompt-embedded safety + lightweight regex | Simpler stack, lower latency overhead | Minimal infrastructure cost |
| Multi-agent platform, strict regulatory compliance | External middleware control plane + Presidio | Centralized policy, audit-ready telemetry, consistent enforcement | Moderate compute overhead (~15–30ms/request) |
| High-throughput public API | Async interceptor with streaming bypass + size limits | Prevents OOM, maintains real-time UX, scales horizontally | Higher memory allocation for buffer management |
| Internal enterprise tooling | Full sanitization + dependency-injected sandboxing | Limits blast radius, enforces least-privilege tool access | Increased DevOps complexity for DI container management |

### Configuration Template

```python
# app/main.py
from fastapi import FastAPI
from control_plane.safety_interceptor import SafetyInterceptor
from control_plane.telemetry import setup_opentelemetry, setup_prometheus
from control_plane.sandbox import ToolAccessDependency

app = FastAPI(title="Agent Control Plane", version="1.0.0")

# Telemetry initialization
setup_prometheus()
setup_opentelemetry(app)

# Dependency injection for sandboxed tool access
app.dependency_overrides[ToolAccessDependency] = ToolAccessDependency

# Middleware stack (LIFO execution)
app.add_middleware(SafetyInterceptor, languages=["en", "es"])
app.add_middleware(AuthenticationGuard)  # Registered last → executes first

@app.get("/agent/query")
async def query_agent(request: dict):
    # Agent logic executes here
    # Middleware intercepts response automatically
    return {"status": "processed"}

Quick Start Guide

1. Install dependencies: pip install fastapi presidio-analyzer presidio-anonymizer prometheus-client opentelemetry-api opentelemetry-sdk
2. Create the interceptor module: Copy the SafetyInterceptor, ComplianceEngine, and DataRedactor classes into control_plane/safety_interceptor.py
3. Register middleware: Add app.add_middleware(SafetyInterceptor) to your FastAPI application after route definitions
4. Configure telemetry: Initialize Prometheus metrics and OpenTelemetry exporters before starting the server
5. Deploy and validate: Send a test request containing a known PII pattern and a policy trigger. Verify that the response is redacted, the violation is logged, and Prometheus metrics increment correctly.