DEV Community: Ashutosh Rana

EU AI Act Goes Live in 90 Days: What Developers Building AI Agents Actually Need to Do

Ashutosh Rana — Fri, 01 May 2026 05:17:10 +0000

If you're building AI agents for anything enterprise — education platforms, HR tools, healthcare apps, financial services — August 2, 2026 is a date worth knowing.

That's when the EU AI Act's Annex III obligations kick in for high-risk AI systems. Not "start planning" — actual legal enforcement, with fines up to €30 million or 6% of global annual turnover for non-compliance.

Most developer guides on the EU AI Act read like they were written by lawyers for other lawyers. This one is for people who write code.

What Actually Applies to You

The first question is always: does this apply to me?

The EU AI Act uses a four-tier risk classification:

Prohibited (Article 5): Manipulation, social scoring, real-time biometric surveillance in public spaces. Most developers aren't building this.
High-risk (Article 6 + Annex III): This is where most enterprise AI agents land.
Limited-risk: Chatbots, AI-generated content — transparency obligations apply.
Minimal-risk: Spam filters, recommendation engines, game AI. Essentially unregulated.

You're in the high-risk category if your system is either a safety component in a regulated product (medical device, vehicle) or operates in one of the sectors listed in Annex III:

Education: AI that determines access to educational institutions, or evaluates students' learning outcomes
Employment: Recruitment tools, CV screening, performance monitoring, task allocation
Essential services: Credit scoring, insurance risk assessment, utility access
Law enforcement and migration/asylum management
Critical infrastructure management

If you're building an admissions AI, an HR screening tool, a financial risk model, or a medical triage agent — you're in Annex III territory.

The multi-agent problem

When you chain agents together, the compliance question compounds. Each agent in a pipeline that makes a decision affecting a person in a covered sector needs to comply. There's no "the LLM just made a suggestion" defense if its output directly influences a consequential decision.

Frameworks like Google ADK, CrewAI, LangGraph, and AutoGen are neutral infrastructure. They don't know whether you're building a compliance-sensitive admissions system or a low-risk content assistant. That means the compliance layer is entirely your responsibility to add.

The Five Things You Actually Have to Build

1. Audit Logging (Article 12)

Every action your agent takes on behalf of someone covered by Annex III needs to be logged with enough detail to reconstruct the decision after the fact. This isn't optional debugging output — it's a legal record that must be retained and producible for auditors.

A useful audit log event for an agent action looks like this:

import json
from datetime import datetime, timezone
from dataclasses import dataclass, asdict

@dataclass
class AgentAuditEvent:
    timestamp: str
    session_id: str
    agent_id: str
    action: str
    inputs: dict
    outputs: dict
    confidence_score: float
    decision_rationale: str
    human_override_available: bool

def log_agent_action(event: AgentAuditEvent):
    record = asdict(event)
    record["timestamp"] = datetime.now(timezone.utc).isoformat()
    # Write to your SIEM, append-only database, or structured log store
    print(json.dumps(record))

Design this layer to be immutable and queryable from day one. Retrofitting audit logging into an existing agent pipeline is painful.

2. Human Oversight (Article 14)

Article 14 is the one that requires the most architectural thought. High-risk AI systems must be designed so that humans can:

Monitor the system during operation
Understand outputs well enough to exercise appropriate judgment
Override, interrupt, or stop the system at any point

That last requirement is the hard one for agentic systems. When you have a multi-agent pipeline running autonomously, you need a technical mechanism — not just a documented policy — that allows a human to halt execution.

Confidence-gated escalation is one pattern that satisfies Article 14 structurally. The agent monitors its own uncertainty and routes to a human when confidence drops below a defined threshold, rather than proceeding with an unreliable answer:

from confidence_escalation import ConfidenceEscalationMiddleware, ThresholdPolicy

policy = ThresholdPolicy(
    low_confidence_threshold=0.6,  # route to human review below this
    critical_threshold=0.3,        # hard stop below this
)

middleware = ConfidenceEscalationMiddleware(
    policy=policy,
    on_escalate=lambda ctx: human_review_queue.enqueue(ctx),
    on_critical=lambda ctx: session_halt(ctx),
)

The confidence-escalation package implements this pattern across LangChain, CrewAI, AutoGen, and Google ADK. But the pattern itself doesn't require any specific library. The key is that your agent has a defined behavior when it's uncertain, and that behavior routes to a human rather than guessing.

3. Transparency (Article 13)

Users interacting with a high-risk AI system must be told:

That they're interacting with an AI
What the system's capabilities and limitations are
How to contact a human if they need to

For voice and chat interfaces, this means disclosure at the start of every session, not buried in terms of service. For backend decision systems — like a loan scoring model — it means the person affected by the decision receives a plain-language explanation.

Build disclosure into session initialization as a first-class feature, not as a one-time consent screen that users click past once.

4. Accuracy and Robustness (Article 15)

Your system must minimize errors, resist adversarial inputs, and degrade gracefully. For LLM-based agents, this maps to:

Hallucination mitigation: Don't let uncertain outputs reach consequential decisions without a confidence check
Adversarial input handling: The OWASP Top 10 for LLM Applications covers prompt injection, data poisoning, and the Agentic AI Top 10 in detail — worth reading directly
Graceful degradation: If the AI can't answer reliably, define the fallback path explicitly. "I'm not confident enough to answer this" is a valid agent output. A hallucinated answer is not.

5. Risk Management System (Article 9)

Article 9 requires an ongoing risk management process, not a one-time compliance review. For engineering teams, this means:

A documented process for identifying new risks when you update the model or change the agent's tool set
Regular testing against your accuracy and robustness baselines
An incident log when the system behaves unexpectedly

This doesn't have to be heavyweight. A written process, a structured incident log, and a quarterly review cadence is a defensible starting point.

Building the Compliance Stack for Multi-Agent Systems

Here's the architecture challenge: all the major agent frameworks are compliance-neutral. They don't know which sector your agent operates in. This means you need to add a policy enforcement layer that runs before each agent action.

The pattern that works is a pre-execution compliance gate — a check that validates any planned action against your regulatory rules before it executes. For Google ADK, this maps cleanly to a BeforeToolInvocationCallback:

from regulated_ai_governance.adapters.google_adk_adapter import create_compliant_agent
from regulated_ai_governance import PolicyStack, FERPAPolicy, EUAIActPolicy

policy_stack = PolicyStack([
    EUAIActPolicy(
        risk_tier="high_risk",
        human_oversight_required=True,
        transparency_required=True,
    ),
    FERPAPolicy(
        authorized_user_types=["student", "registrar"],
    ),
])

agent = create_compliant_agent(
    base_agent=my_adk_agent,
    policy_stack=policy_stack,
    audit_logger=my_audit_logger,
)

The regulated-ai-governance package implements this gate across Google ADK, CrewAI, LangChain, AutoGen, and Semantic Kernel. The same architectural pattern applies regardless of which framework you're using — policy evaluation before the action, not after.

For RAG systems in regulated sectors, the compliance layer needs to operate at the retrieval layer too, not just at the agent action layer. A FERPA-covered education AI should filter documents before they enter the context window, not after the LLM has already processed unauthorized content. The enterprise-rag-patterns library handles this with pre-retrieval filtering that enforces access control based on user identity and regulatory scope.

The August 2 Deadline: What's Actually Enforceable When

A quick timeline clarification:

Date	What Applies
February 2, 2025	Prohibited practices (Article 5) — already in force
August 2, 2025	GPAI model obligations (Article 51-56)
August 2, 2026	High-risk AI systems in Annex III — education, employment, essential services
August 2, 2027	Annex I safety component systems

Fines for Annex III non-compliance: up to €20 million or 4% of global annual turnover, whichever is higher. The full regulation text is publicly available on EUR-Lex — the recitals are worth reading because they explain the legislative intent behind specific articles in plain language.

The EU AI Act doesn't require pre-registration the way GDPR's Data Protection Impact Assessment does. But if an incident occurs or a regulator audits you, you need to demonstrate that Articles 9-15 were implemented. The burden of proof is on the system operator.

Where to Start

If you're building AI agents in any Annex III sector, here's a practical starting checklist:

Classify your system honestly. "Educational AI" that influences student outcomes = high-risk. Don't minimize it.
Add structured audit logging now. Every agent action, every tool call, every confidence score. Retrofit is painful.
Design in a human override path. At minimum: a review queue where a human can halt any agent decision before it becomes final.
Document your risk management process. A one-page document describing how you identify and address new risks is better than nothing — and it's evidence.
Build AI disclosure into session init. Not a checkbox. An actual first-message disclosure at the start of every user session.
Test for adversarial inputs. At least run prompt injection and data poisoning test cases against your agent before August.

The technical implementations here — audit logging, confidence checks, human escalation, policy gates — are engineering best practices that happen to also be legally required. The systems that handle these well tend to work better anyway: fewer silent failures, clearer failure modes, more trustworthy outputs.

The deadline is real. Three months is enough time to build this right.

If you want to dig into the implementation patterns, the repos I reference in this article all have working examples: regulated-ai-governance, confidence-escalation, enterprise-rag-patterns. The EU AI Act official consolidated text is on EUR-Lex.

MACF: The 6-Component Framework Every Enterprise Multi-Agent AI System Needs

Ashutosh Rana — Wed, 29 Apr 2026 14:57:37 +0000

The Problem Nobody Talks About

You can wire together a multi-agent system in an afternoon with LangChain, CrewAI, or AutoGen. You'll have agents calling tools, passing messages, and producing outputs. It works.

Then someone asks: "Which agent decided to access that patient record? Can you show us the full decision trail for the compliance audit?"

Or: "The orchestrator passed a context payload to the subagent — was PII scrubbed before that transfer?"

Or simply: "How do we know the right agent handled this request? Is there a capability check before invocation?"

Most frameworks don't answer these questions. They give you plumbing — not governance. For internal prototypes that's fine. For enterprise deployments in regulated industries (healthcare, education, financial services), it isn't.

This article introduces MACF — the Multi-Agent Collaborative Framework — a six-component reference architecture that adds the layer most frameworks skip: compliance-enforced, auditable, privacy-aware agent coordination.

Multi-Agent Systems in 90 Seconds

Before diving into MACF, a brief grounding for readers new to multi-agent architectures.

A multi-agent system is a collection of AI agents — each with a defined capability scope — that collaborate to complete tasks neither could do alone. A common pattern: an orchestrator agent receives a user request, decomposes it, delegates subtasks to specialist agents (a retrieval agent, a summarization agent, a compliance-check agent), and assembles the final response.

This is different from a single LLM with tool calls. Here, each agent may run a different model, hold its own context window, call its own tools, and run in parallel with other agents. Communication between agents is structured — each message has a sender, receiver, and payload.

The value is clear: parallelism, specialization, and modularity. The challenge is equally clear: when agents are coordinated across a pipeline, failures — and compliance violations — can cascade. A privacy leak in one agent propagates silently to every downstream agent unless something stops it.

MACF is that something.

What Current Frameworks Provide (and What They Skip)

LangChain, CrewAI, AutoGen, and Google ADK are all excellent at defining agent topology — which agents exist, how they connect, and what tools they can call.

What none of them include out of the box:

Concern	LangChain	CrewAI	AutoGen	Google ADK
Capability-gated agent invocation	❌	❌	❌	Partial
Regulatory context propagation across agent hops	❌	❌	❌	❌
PII/PHI scrubbing before context transfer	❌	❌	❌	❌
Pre-response compliance enforcement (HIPAA, TCPA, GDPR)	❌	❌	❌	❌
Tamper-evident immutable audit trail	❌	❌	❌	❌

MACF doesn't replace these frameworks. It runs as an infrastructure layer that wraps them — the same way an API gateway sits in front of microservices without replacing the services themselves.

The MACF Architecture

MACF defines six components. Each has a typed interface, a defined responsibility, and a compliance guarantee.

flowchart TB
    User([User Request]) --> Orch

    subgraph Agents ["Your Agent Framework  (LangChain / CrewAI / AutoGen / ADK)"]
        Orch[Orchestrator Agent]
        SA1[Specialist Agent A]
        SA2[Specialist Agent B]
    end

    subgraph MACF ["MACF Infrastructure Layer"]
        AR["AgentRegistry\nauthorize() · capability check"]
        MB["MessageBus\nroute() · regulatory_context propagation"]
        CS["ContextStore\nget() / set() · TTL · audit flag"]
        PF["PrivacyFilter\nscrub() · PHI/PII redaction"]
        CG["ComplianceGate\nenforce() · HIPAA · TCPA · GDPR · EU AI Act"]
        AT["AuditTrail\nappend() · verify() · hash chain"]
    end

    Orch -->|1 authorize| AR
    AR -->|✅ capability check| Orch
    Orch -->|2 route + scrub| MB
    MB --> PF
    PF --> MB
    MB -->|log| AT
    MB --> SA1
    MB --> SA2
    SA1 -->|read context| CS
    CS -->|log read| AT
    SA1 -->|response| Orch
    Orch -->|3 enforce| CG
    CG -->|log gate result| AT
    CG -->|✅ gated response| Response([Final Response])

Let's walk through each component.

1. AgentRegistry

The AgentRegistry is the single source of truth for what agents exist and what they're allowed to do. Before any agent is invoked, the registry performs a capability authorization check — validating that the requesting agent has permission to call the target agent in the current regulatory context.

from dataclasses import dataclass, field
from typing import Set, Optional, Callable

@dataclass
class AgentCapability:
    agent_id: str
    name: str
    allowed_callers: Set[str]          # which agents can invoke this one
    required_context_keys: Set[str]    # context fields that must be present
    regulatory_scope: Set[str]         # e.g. {"hipaa", "ferpa", "eu_ai_act"}
    handler: Callable

class AgentRegistry:
    def __init__(self):
        self._agents: dict[str, AgentCapability] = {}

    def register(self, capability: AgentCapability) -> None:
        self._agents[capability.agent_id] = capability

    def authorize(
        self,
        caller_id: str,
        target_id: str,
        context: dict,
    ) -> bool:
        cap = self._agents.get(target_id)
        if not cap:
            return False
        if caller_id not in cap.allowed_callers:
            return False
        missing = cap.required_context_keys - set(context.keys())
        if missing:
            raise ValueError(f"Missing required context keys: {missing}")
        return True

    def get(self, agent_id: str) -> Optional[AgentCapability]:
        return self._agents.get(agent_id)

Why this matters: Without an authorization check, any agent can call any other agent. In a regulated deployment, an unauthenticated agent calling a records-access agent is an access-control violation — regardless of what the underlying model does.

2. MessageBus

The MessageBus handles all inter-agent communication. Every message is typed and carries a regulatory_context field — a propagation envelope that tells downstream agents which regulatory constraints apply to this request.

from dataclasses import dataclass, field
from typing import Any, Dict, Optional
import datetime

@dataclass
class AgentMessage:
    sender_id: str
    receiver_id: str
    payload: Any
    message_id: str
    timestamp: str = field(
        default_factory=lambda: datetime.datetime.now(datetime.timezone.utc).isoformat()
    )
    regulatory_context: Dict[str, Any] = field(default_factory=dict)
    # e.g. {"regulations": ["hipaa", "tcpa"], "consent_verified": True,
    #        "jurisdiction": "US", "session_id": "sess_abc123"}

class MessageBus:
    def route(
        self,
        message: AgentMessage,
        registry: AgentRegistry,
        privacy_filter: "PrivacyFilter",
        audit_trail: "AuditTrail",
    ) -> AgentMessage:
        # 1. Authorization check
        authorized = registry.authorize(
            caller_id=message.sender_id,
            target_id=message.receiver_id,
            context=message.regulatory_context,
        )
        if not authorized:
            raise PermissionError(
                f"Agent {message.sender_id!r} is not authorized "
                f"to invoke {message.receiver_id!r}"
            )

        # 2. Scrub PII from payload before routing
        if isinstance(message.payload, str):
            scrub_result = privacy_filter.scrub(message.payload)
            message = AgentMessage(
                **{**message.__dict__, "payload": scrub_result.clean_text}
            )

        # 3. Append routing decision to audit trail
        audit_trail.append({
            "event": "message_routed",
            "from": message.sender_id,
            "to": message.receiver_id,
            "message_id": message.message_id,
            "regulatory_context": message.regulatory_context,
        })

        return message

The key design decision: regulatory_context propagates with every message hop. An orchestrator that sets {"regulations": ["hipaa"], "consent_verified": True} will have that context available to every downstream subagent — without requiring each subagent to re-derive it.

3. ContextStore

The ContextStore holds cross-agent shared state with two MACF-specific additions: TTL enforcement (context expires, preventing stale data from leaking across sessions) and audit flagging (any read of a flagged key is automatically recorded in the AuditTrail).

import time
from typing import Any, Optional

class ContextStore:
    def __init__(self, default_ttl_seconds: int = 3600):
        self._store: dict = {}
        self._ttl: dict = {}
        self._audit_keys: set = set()
        self.default_ttl = default_ttl_seconds

    def set(
        self,
        key: str,
        value: Any,
        ttl_seconds: Optional[int] = None,
        audit: bool = False,
    ) -> None:
        self._store[key] = value
        self._ttl[key] = time.time() + (ttl_seconds or self.default_ttl)
        if audit:
            self._audit_keys.add(key)

    def get(self, key: str, audit_trail: Optional["AuditTrail"] = None) -> Optional[Any]:
        if key not in self._store:
            return None
        if time.time() > self._ttl.get(key, 0):
            del self._store[key]
            return None
        if key in self._audit_keys and audit_trail:
            audit_trail.append({"event": "context_read", "key": key})
        return self._store[key]

Practical use: A student's enrollment record retrieved during session initialization can be stored with audit=True so every downstream read is traced — satisfying FERPA §99.32 (recordkeeping of education record disclosures).

4. PrivacyFilter

The PrivacyFilter scrubs PHI and PII from any text payload before it crosses an agent boundary. It runs on every outbound message from the orchestrator to subagents, and on every context value written to the ContextStore.

import re
from dataclasses import dataclass
from typing import List, Tuple

@dataclass
class ScrubResult:
    clean_text: str
    replacements: List[Tuple[str, str]]  # (pattern_name, matched_value)

    @property
    def was_modified(self) -> bool:
        return bool(self.replacements)

class PrivacyFilter:
    # Extend this list for your regulatory scope
    _PATTERNS = [
        ("SSN",          r"\b\d{3}-\d{2}-\d{4}\b"),
        ("MRN_PARAM",    r"(?:mrn|patient_id|dob|ssn)=[^&\s]+"),
        ("EMAIL",        r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b"),
        ("PHONE",        r"\b(\+?1[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b"),
        ("STUDENT_ID",   r"\b(?:student_id|stu_id)=\S+"),
        ("DOB",          r"\b(?:dob|date_of_birth)[=:\s]+\d{1,2}[/\-]\d{1,2}[/\-]\d{2,4}"),
    ]

    def scrub(self, text: str) -> ScrubResult:
        replacements = []
        result = text
        for name, pattern in self._PATTERNS:
            matches = re.findall(pattern, result, re.IGNORECASE)
            if matches:
                result = re.sub(pattern, "[REDACTED]", result, flags=re.IGNORECASE)
                replacements.extend([(name, m) for m in matches])
        return ScrubResult(clean_text=result, replacements=replacements)

In a healthcare context, this is the difference between sending "Patient mrn=9876543 needs billing review" to a downstream agent and sending "Patient mrn=[REDACTED] needs billing review". The subagent can still process the intent without receiving the PHI.

The regulated-ai-governance and voice-ai-governance OSS packages ship production-grade implementations of this scrubbing layer with 14+ PHI patterns validated across 100% recall on a test corpus.

5. ComplianceGate

The ComplianceGate is evaluated before every agent response is returned to the caller. It enforces applicable regulations — HIPAA, TCPA, GDPR, EU AI Act Article 13/14 — based on the regulatory_context carried by the incoming message.

This is where the confidence-escalation package integrates directly into MACF. The ComplianceGate doesn't just check rules — it also evaluates the agent's confidence in its response and can trigger a human-in-the-loop handoff if confidence falls below threshold.

pip install confidence-escalation regulated-ai-governance

from confidence_escalation import (
    MultiSignalConfidenceScorer,
    ThresholdPolicy,
    EscalationAction,
    HumanInLoopHandler,
    ComplianceLoggingHandler,
    ConfidenceEscalationMiddleware,
    ScoringMethod,
)

# Build a multi-signal scorer: logprob + verbalized confidence + tool-call risk
scorer = MultiSignalConfidenceScorer(
    weights={
        ScoringMethod.LOGPROB: 0.5,
        ScoringMethod.VERBALIZED: 0.3,
        ScoringMethod.TOOL_CALL_RISK: 0.2,
    }
)

# Two-tier policy: warn below 0.65, abort below 0.30
policy = ThresholdPolicy(
    threshold=0.65,
    action=EscalationAction.HUMAN_IN_LOOP,
    critical_threshold=0.30,
    critical_action=EscalationAction.ABORT,
)

# Handlers: route to human queue OR write to compliance log
handlers = [
    HumanInLoopHandler(queue_callback=your_human_queue.enqueue),
    ComplianceLoggingHandler(logger=your_compliance_logger),
]

class ComplianceGate:
    def __init__(self, scorer, policy, handlers, audit_trail):
        self._middleware = ConfidenceEscalationMiddleware(
            scorer=scorer,
            policy=policy,
            handlers=handlers,
        )
        self._audit = audit_trail

    def enforce(self, response: str, context: dict) -> dict:
        # 1. Confidence check
        escalation_event = self._middleware.evaluate(response, context)
        self._audit.append(escalation_event.to_dict())

        if escalation_event.action == EscalationAction.ABORT:
            raise RuntimeError("Response aborted: confidence below critical threshold")

        # 2. Regulatory policy check (from regulated-ai-governance)
        # Pass response through applicable compliance policies
        # (HIPAA, TCPA quiet hours, EU AI Act disclosure, etc.)

        return {
            "response": response,
            "confidence": escalation_event.confidence_score,
            "escalated": escalation_event.triggered,
            "regulatory_context": context,
        }

The confidence-escalation package provides four framework adapters (LangChain, CrewAI, AutoGen, Google ADK) so this composes cleanly with whichever agent framework you're already using:

from confidence_escalation import LangChainEscalationAdapter

# Drop-in middleware for an existing LangChain agent
adapter = LangChainEscalationAdapter(
    scorer=scorer,
    policy=policy,
    handlers=handlers,
)
agent_with_gate = adapter.wrap(your_langchain_agent)

6. AuditTrail

Every MACF operation — agent authorization, message routing, context reads, compliance gate evaluations — appends a record to the AuditTrail. The trail uses a hash chain: each entry includes the SHA-256 hash of the previous entry, making tamper-evidence detectable without an external ledger.

import hashlib
import json
import datetime
from typing import Any, Dict, List

class AuditTrail:
    def __init__(self):
        self._entries: List[Dict[str, Any]] = []
        self._last_hash = "genesis"

    def append(self, record: Dict[str, Any]) -> str:
        entry = {
            **record,
            "timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(),
            "prev_hash": self._last_hash,
        }
        entry_hash = hashlib.sha256(
            json.dumps(entry, sort_keys=True).encode()
        ).hexdigest()
        entry["hash"] = entry_hash
        self._entries.append(entry)
        self._last_hash = entry_hash
        return entry_hash

    def verify(self) -> bool:
        """Returns True if the chain is intact (no entries have been modified)."""
        prev = "genesis"
        for entry in self._entries:
            claimed_hash = entry.get("hash")
            reconstructed = {k: v for k, v in entry.items() if k != "hash"}
            reconstructed["prev_hash"] = prev
            expected = hashlib.sha256(
                json.dumps(reconstructed, sort_keys=True).encode()
            ).hexdigest()
            if claimed_hash != expected:
                return False
            prev = claimed_hash
        return True

    def export(self) -> List[Dict[str, Any]]:
        return list(self._entries)

A three-month deployment using this audit design produced zero unresolved TCPA compliance queries — every agent decision was traceable within seconds of a compliance request.

How a Request Flows Through MACF

Here's what a single user request looks like end-to-end:

sequenceDiagram
    actor User
    participant O as Orchestrator
    participant AR as AgentRegistry
    participant PF as PrivacyFilter
    participant MB as MessageBus
    participant SA as Specialist Agent
    participant CS as ContextStore
    participant CG as ComplianceGate
    participant AT as AuditTrail

    User->>O: request
    O->>AR: authorize(caller="orchestrator", target="records_agent")
    AR-->>O: ✅ authorized

    O->>PF: scrub(payload)
    PF-->>O: clean_payload [PHI removed]

    O->>MB: route(message + regulatory_context)
    MB->>AT: append(routing_event)
    MB->>SA: deliver(message)

    SA->>CS: get("session_context", audit_trail)
    CS->>AT: append(context_read_event)
    CS-->>SA: context_data

    SA-->>O: raw_response

    O->>CG: enforce(response, regulatory_context)
    Note over CG: MultiSignalConfidenceScorer → ThresholdPolicy<br/>score ≥ 0.65 → pass / score < 0.65 → HumanInLoop
    CG->>AT: append(gate_result)
    CG-->>O: gated_response

    O-->>User: final_response
    Note over AT: AuditTrail.verify() → ✅ hash chain intact

The total overhead across all six components adds a median of under 10ms per request — less than a single LLM token generation step.

Getting Started

All compliance-enforcement components referenced in this article are available as open-source Python packages:

# Confidence scoring, threshold policies, framework adapters
pip install confidence-escalation

# Runtime compliance enforcement (HIPAA, FERPA, TCPA, EU AI Act)
pip install regulated-ai-governance

# Voice/SMS pipeline compliance (Pipecat, Twilio, A2P 10DLC)
pip install voice-ai-governance

A minimal MACF wiring for a two-agent system:

from confidence_escalation import (
    MultiSignalConfidenceScorer,
    ThresholdPolicy,
    EscalationAction,
    HumanInLoopHandler,
    ConfidenceEscalationMiddleware,
    ScoringMethod,
)

# Initialize the six components
registry = AgentRegistry()
bus = MessageBus()
store = ContextStore()
pfilter = PrivacyFilter()
audit = AuditTrail()

scorer = MultiSignalConfidenceScorer(
    weights={ScoringMethod.LOGPROB: 0.5, ScoringMethod.VERBALIZED: 0.5}
)
policy = ThresholdPolicy(threshold=0.65, action=EscalationAction.HUMAN_IN_LOOP)
gate = ComplianceGate(scorer, policy, [HumanInLoopHandler(...)], audit)

# Register agents
registry.register(AgentCapability(
    agent_id="records_agent",
    name="Records Access Agent",
    allowed_callers={"orchestrator"},
    required_context_keys={"consent_verified", "regulations"},
    regulatory_scope={"hipaa", "ferpa"},
    handler=your_records_handler,
))

# Route a message
msg = AgentMessage(
    sender_id="orchestrator",
    receiver_id="records_agent",
    payload="Retrieve enrollment history for student",
    message_id="msg_001",
    regulatory_context={"regulations": ["ferpa"], "consent_verified": True},
)
routed = bus.route(msg, registry, pfilter, audit)

# Execute and gate the response
raw_response = registry.get("records_agent").handler(routed)
result = gate.enforce(raw_response, msg.regulatory_context)

# Verify audit integrity
assert audit.verify(), "Audit trail tampered"

Why Six Components

Each component addresses a distinct failure mode:

Component	Failure Mode Addressed
AgentRegistry	Unauthorized agent invocations in multi-hop chains
MessageBus	PII leaking across agent boundaries during routing
ContextStore	Stale session data persisting beyond TTL; untracked reads
PrivacyFilter	PHI reaching downstream agents without scrubbing
ComplianceGate	Low-confidence responses reaching users without human review
AuditTrail	Unverifiable decision history when compliance queries arrive

You can adopt them incrementally. Start with the AuditTrail and ComplianceGate (the two with the highest compliance ROI), then add the others as your deployment matures.

Every Enterprise AI Framework Has a Compliance Gap — Here's the Architecture That Closes It

Ashutosh Rana — Fri, 24 Apr 2026 14:57:47 +0000

Every Enterprise AI Framework Has a Compliance Gap — Here's the Architecture That Closes It

Ashutosh Rana — Enterprise AI Architect

A survey published by Grant Thornton in 2026 found that 78% of business executives cannot pass an independent AI governance audit within 90 days. A separate S&P Global Market Intelligence study found that 42% of companies abandoned most AI initiatives in 2025 — up from 17% the year before. The most common reason: compliance and governance failures, not technical ones.

The AI systems are working. The compliance architecture around them is not.

This article explains why that gap exists, what the regulatory environment now demands, and how to build a governance layer that actually works across the major enterprise AI frameworks.

The Problem: AI Frameworks Ship Without Compliance

Pick any major agentic AI framework — CrewAI, AutoGen, LangChain, Semantic Kernel, Google ADK. Read its documentation. You will find excellent coverage of:

Tool calling and function execution
Multi-agent orchestration
Memory and context management
Model switching and routing

You will not find:

A concept of regulated data categories
An enforcement point for FERPA, HIPAA, or GDPR access rules
A structured audit record tied to a regulation citation
A mechanism for flagging decisions that require human review under the EU AI Act

This is not a criticism of those frameworks. They are general-purpose tools. But when you deploy them in a hospital, a university, a bank, or a government agency, you are operating in a regulated environment that those frameworks were not designed for.

The compliance gap is architectural. Fixing it with prompts, post-processing filters, or manual review processes does not close it.

What the Regulatory Environment Now Demands

Three regulatory developments in 2025-2026 have made this a matter of immediate financial risk.

EU AI Act — Penalties Active as of August 2025

The EU AI Act's penalty regime became enforceable on August 2, 2025. Full compliance for high-risk AI systems is mandatory by August 2, 2026.

High-risk classifications include AI systems used in education, employment, healthcare, law enforcement, and critical infrastructure. Multi-agent AI systems that make decisions affecting individuals in these domains fall squarely within scope.

OWASP Agentic AI Top 10 2026

Published in December 2025 after peer review by 100+ security experts, the OWASP Agentic AI Top 10 2026 identified the primary risks facing enterprise AI agent deployments:

ASI01 — Agent Goal Hijacking: Adversarial instructions injected into data sources (emails, PDFs, RAG documents) redirect agents to exfiltrate sensitive files
ASI02 — Tool Misuse: Agents misuse legitimate tools due to ambiguous prompts or manipulated input, calling them with destructive parameters
ASI03 — Identity and Privilege Abuse: Agents operating with broader permissions than required for a given task
ASI04 — Supply Chain Vulnerabilities: Compromised agent dependencies or external tool integrations

48% of cybersecurity professionals now identify agentic AI as the #1 attack vector heading into 2026 — ahead of ransomware and supply chain attacks.

None of the major AI frameworks have built-in mitigations for these risks. They are framework-agnostic architectural problems.

HIPAA Security Rule Update — January 2025

The HHS Office for Civil Rights proposed its first major HIPAA Security Rule update in 20 years on January 6, 2025. The proposed rule directly addresses AI:

AI tools must be included in organizational risk analysis and risk management activities
Encryption of ePHI in transit and at rest changed from "addressable" (optional) to mandatory
Business Associate Agreements must explicitly address AI vendor data use

Healthcare data breaches already cost an average of $7.42–$11.2 million per incident — the most expensive of any industry for 15 consecutive years. HIPAA-violating AI deployments now face both regulatory penalties and dramatically higher breach costs.

Why Existing Mitigation Approaches Fail

There are three common approaches to compliance in enterprise AI, and three reasons none of them are sufficient:

1. Prompt-level instructions

"Only discuss information the current user is authorized to see. 
 Do not reference other users' records."

Why it fails: Under FERPA (34 CFR § 99.30), HIPAA (45 CFR § 164.502), and GDPR (Article 5(1)(f)), unauthorized access — not unauthorized output — constitutes a violation. A document retrieved into an LLM's context window has already been disclosed, regardless of what the LLM says in its response. OWASP LLM01 (prompt injection) can also override any system prompt instruction.

2. Post-processing output filters

# Filter the response after the LLM produces it
if contains_pii(response):
    return redacted_response

Why it fails: The LLM has already processed the unauthorized data. The disclosure occurred during retrieval and inference, not in the output. Filtering the response does not undo the access.

3. Manual review workflows

Why it fails: Agentic AI systems execute hundreds of tool calls per session, often in parallel. Manual review does not scale to agent execution speeds, and audit trails produced after the fact do not satisfy real-time regulatory requirements.

The Right Architecture: A Governance Layer Before Execution

The solution is a pre-execution governance layer — a composable set of filters that evaluate every agent action, tool call, and data access decision before it executes, producing a structured audit record with a regulation citation for every decision.

Agent Action Request
      │
      ▼
GovernanceOrchestrator.evaluate(context, action)
      │
      ├── Identity / Data Protection Gate ──► FilterResult (FERPA, HIPAA, GDPR)
      │
      ├── Responsible AI Gate ─────────────► FilterResult (EU AI Act, METI, PDPC)
      │
      ├── Sector-Specific Gate ────────────► FilterResult (MAS FEAT, NIST AI RMF)
      │
      └── OWASP Agentic Top 10 Gate ───────► FilterResult (ASI01–ASI10)
                   │
                   ▼
            GovernanceReport
            ├── overall_decision: APPROVED / DENIED / REQUIRES_HUMAN_REVIEW
            ├── is_compliant: bool
            ├── regulation_citation: "34 CFR § 99.31(a)(1)"
            └── audit_record: structured log for regulatory file

Key design properties:

Pre-execution — the filter runs before the action, not after
Every filter is independently testable — adding a regulation means adding a filter class; no existing filter is modified
Immutable context — context objects are @dataclass(frozen=True); no mutable state passes through
Structured audit records — every decision produces a log entry with a regulation citation, timestamp, user identity, and data category

Implementation: Drop-In Governance for Major Frameworks

regulated-ai-governance implements this architecture as drop-in adapters for 10 major AI frameworks.

pip install regulated-ai-governance

CrewAI + FERPA

from regulated_ai_governance.integrations.crewai import EnterpriseActionGuard
from regulated_ai_governance.regulations.ferpa import make_ferpa_student_policy

guard = EnterpriseActionGuard(
    policy=make_ferpa_student_policy(
        student_id="student_789",
        institution_id="univ_001",
        authorized_categories=["transcript", "enrollment"],
    )
)

# Attach to any CrewAI agent
@guard.enforce
def advising_agent_tool(query: str, student_record: dict) -> str:
    ...

Every call is evaluated against FERPA's 34 CFR § 99.31 before execution. Unauthorized access is denied and logged. Legitimate access produces a disclosure record.

AutoGen + HIPAA

from regulated_ai_governance.integrations.autogen import AutoGenGovernanceHook
from regulated_ai_governance.regulations.hipaa import make_hipaa_phi_policy

hook = AutoGenGovernanceHook(
    policy=make_hipaa_phi_policy(
        authorized_roles=["attending_physician"],
        phi_categories=["diagnosis", "medication", "lab_results"],
        minimum_necessary=True,  # 45 CFR § 164.502(b)
    )
)

# Register as AutoGen pre-tool-call hook
agent.register_hook("process_message_before_send", hook.evaluate)

LangChain + GDPR

from regulated_ai_governance.integrations.langchain import LangChainGovernanceCallback
from regulated_ai_governance.regulations.gdpr import make_gdpr_data_policy

callback = LangChainGovernanceCallback(
    policy=make_gdpr_data_policy(
        lawful_basis="legitimate_interest",  # GDPR Article 6
        data_subject_categories=["eu_resident"],
        cross_border_transfer=False,
    )
)

chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=retriever,
    callbacks=[callback],  # governance applied at every retrieval step
)

OWASP Agentic AI Top 10 Enforcement

from regulated_ai_governance.regulations.owasp_agentic import make_owasp_agentic_policy

policy = make_owasp_agentic_policy(
    mitigations=[
        "ASI01_goal_hijacking",
        "ASI02_tool_misuse",
        "ASI03_privilege_escalation",
    ],
    human_review_threshold=0.85,  # flag decisions above confidence threshold for human review
)

What 25 Jurisdictions of Coverage Looks Like

The governance layer currently covers regulations across 25 jurisdictions:

Region	Regulations
United States	FERPA, HIPAA, GLBA, CCPA, NIST AI RMF, FedRAMP, FISMA, ITAR/EAR, FINRA/SEC, FDA 21 CFR Part 11
European Union	GDPR, EU AI Act, ePrivacy
Asia-Pacific	Singapore PDPA + MAS FEAT, Japan APPI + METI, South Korea PIPA, Australia Privacy Act, India DPDPA
Global	ISO/IEC 42001, LGPD (Brazil), PIPEDA (Canada), OWASP Agentic AI Top 10 2026

Each regulation is a standalone filter class. You compose only what your deployment requires.

The Audit Trail That Regulators Actually Want

Every governance decision produces a structured audit record:

GovernanceReport(
    overall_decision=GovernanceDecision.APPROVED,
    is_compliant=True,
    compliance_summary="Access authorized under FERPA § 99.31(a)(1): legitimate educational interest verified",
    filter_results=[
        FilterResult(
            regulation="FERPA",
            decision=GovernanceDecision.APPROVED,
            citation="34 CFR § 99.31(a)(1)",
            reason="Requesting party has legitimate educational interest in student record",
            timestamp="2026-04-24T09:15:42Z",
            data_subject="student_789",
            data_category="transcript",
        )
    ]
)

This record format satisfies the documentation requirements of FERPA's § 99.32 (disclosure record-keeping), HIPAA's § 164.528 (accounting of disclosures), and GDPR's Article 30 (records of processing activities).

What This Does Not Replace

A governance layer is not a substitute for:

Data classification — documents must be tagged with subject identity and data category before governance can enforce access rules
Identity management — the governance layer needs a verified user identity to enforce access policies
Legal counsel — the filter implementations encode a good-faith interpretation of each regulation; your organization's legal team should review the policies applied to your specific deployment

What it replaces: the assumption that an AI framework, a system prompt, or a post-processing filter provides sufficient compliance coverage for a regulated environment.

The Market Is Moving Fast

The AI governance and compliance tools market was valued at $2.2 billion in 2025 and is projected to reach $11.05 billion by 2036 at 15.8% CAGR (Future Market Insights, 2025). That growth is driven by exactly the regulatory pressure described above.

Only 20% of companies currently have a mature governance model for autonomous AI agents (Deloitte, 2026). The remaining 80% are operating without the architecture that the EU AI Act, updated HIPAA rules, and OWASP Agentic AI Top 10 2026 now effectively require.

Get Started

pip install regulated-ai-governance

The library is open source, MIT licensed, and includes 45 examples across 25 jurisdictions and 10 AI frameworks.

GitHub: github.com/ashutoshrana/regulated-ai-governance
PyPI: pypi.org/project/regulated-ai-governance
Docs: Getting started guide | API reference | Regulation coverage

If your organization is deploying AI agents in healthcare, higher education, financial services, or any other regulated environment — the governance layer is the piece that is currently missing from every major framework's documentation.

Ashutosh Rana is an enterprise cloud architect specialising in AI systems for regulated industries. He writes about enterprise AI architecture on Medium and publishes open-source governance tools on GitHub.

Google ADK Has a Compliance Gap — Here's How to Close It

Ashutosh Rana — Mon, 20 Apr 2026 02:28:27 +0000

Google's Agent Development Kit (ADK) makes it remarkably easy to build multi-agent AI systems. You can wire up an orchestrator agent, connect it to specialized sub-agents, and have a working pipeline in under 100 lines of Python.

What it does not give you — at least not yet — is a compliance layer.

In regulated industries, that gap is the difference between a production deployment and a liability.

What ADK Gives You

ADK provides a clean callback architecture:

before_model_callback — intercept before the LLM sees the prompt
before_agent_callback — intercept at agent invocation
before_tool_callback — intercept before any tool executes
after_model_callback — intercept after the LLM responds

These hooks exist precisely for this kind of instrumentation. The framework is well-designed. The gap is not architectural — it is that there is no reference implementation for compliance enforcement using these hooks.

Why This Matters for Regulated Deployments

Consider three real scenarios:

Higher Education (FERPA)
An admissions agent handles student data. FERPA requires that every disclosure of student education records be logged (34 CFR § 99.32) and that access be limited to legitimate educational interest (34 CFR § 99.31). Without a compliance layer, an ADK agent has no mechanism to enforce or record either requirement.

Healthcare (HIPAA)
An intake triage agent processes patient queries. HIPAA requires that PHI (Protected Health Information) only be accessed by authorized workforce members under a BAA (Business Associate Agreement). An ADK agent without a compliance hook cannot verify BAA status or create the audit trail required by 45 CFR § 164.312.

Enterprise AI (OWASP Agentic AI Top 10 2026)
OWASP's 2026 Agentic AI Top 10 identifies privilege escalation (ASI02), insufficient audit logging (ASI06), and uncontrolled resource consumption (ASI08) as the top risks in multi-agent systems. An ADK orchestrator that spawns sub-agents without privilege boundaries is exposed to all three.

The ADKPolicyGuard Pattern

I built ADKPolicyGuard in the regulated-ai-governance package to provide a drop-in compliance layer for ADK agents.

from regulated_ai_governance.adapters.google_adk_adapter import (
    ADKPolicyGuard,
    BigQueryAuditSink,
    Regulation,
)
from google.adk.agents import LlmAgent

# Define policy — FERPA + OWASP Agentic Top 10
guard = ADKPolicyGuard(
    regulations=[Regulation.FERPA, Regulation.OWASP_AGENTIC_TOP10],
    audit_sink=BigQueryAuditSink(
        project_id="your-gcp-project",
        dataset_id="compliance_audit",
        table_id="adk_disclosures",
    ),
    rate_limit_rpm=60,
)

# Wire into your ADK agent via callbacks
agent = LlmAgent(
    name="student_advisor",
    model="gemini-2.0-flash",
    before_agent_callback=guard.before_agent_callback,
    before_model_callback=guard.before_model_callback,
    before_tool_callback=guard.before_tool_callback,
)

Every agent invocation is now covered:

Before agent starts: identity scope is validated, rate limit is checked
Before model call: prompt is screened for policy violations (OWASP LLM01 — prompt injection)
Before tool executes: tool permissions are validated against the authorized role
All events: written to BigQuery as structured audit records

Multi-Agent Orchestration

The real value shows up in multi-agent systems. In an Orchestrator → LeadAgent → ApplicantAgent architecture, each agent hand-off is a potential privilege escalation point. ADKPolicyGuard enforces that sub-agents cannot exceed the privilege scope of the orchestrator:

from google.adk.agents import SequentialAgent

orchestrator = SequentialAgent(
    name="admissions_orchestrator",
    sub_agents=[lead_agent, applicant_agent],
    before_agent_callback=guard.before_agent_callback,
)

The guard's before_agent_callback validates each sub-agent invocation against the original identity scope. A sub-agent cannot access data the orchestrator was not authorized to access — privilege escalation is structurally prevented.

The Audit Record

Every agent interaction produces a structured compliance record:

{
  "event_id": "adk-20260418-001",
  "agent_name": "student_advisor",
  "regulation": "FERPA",
  "decision": "ALLOWED",
  "identity": {"user_id": "stu_001", "role": "student"},
  "tool_calls": ["get_transcript", "check_financial_aid"],
  "timestamp": "2026-04-18T10:30:00Z",
  "rate_limit_remaining": 58,
  "owasp_checks": {
    "ASI01_prompt_injection": "PASS",
    "ASI02_privilege_escalation": "PASS",
    "ASI06_audit_logging": "PASS"
  }
}

This record goes directly to BigQuery for compliance reporting, incident investigation, and regulatory audit response.

What This Does Not Replace

ADKPolicyGuard is a compliance enforcement layer, not an authentication system. Your application must establish the authenticated identity context before the agent runs. The guard enforces the scope; your auth layer establishes it.

It also does not replace your legal counsel's review of how your specific deployment maps to applicable regulations.

Getting Started

pip install regulated-ai-governance

from regulated_ai_governance.adapters.google_adk_adapter import ADKPolicyGuard, Regulation

GitHub: ashutoshrana/regulated-ai-governance
PyPI: regulated-ai-governance
Full ADK examples: examples/42_google_adk_ferpa_agent.py through 45_google_adk_hipaa_agent.py

If you are building ADK agents for healthcare, education, or any regulated environment and want to discuss the compliance architecture, open an issue or connect with me directly.

The Hidden Compliance Gap in Every Enterprise RAG Pipeline

Ashutosh Rana — Mon, 20 Apr 2026 02:27:44 +0000

Every week, another enterprise announces a RAG-powered AI assistant. Legal teams get a contract review bot. Hospitals get a clinical decision support tool. Banks get a loan advisory chatbot. Universities get a student advising system.

Nearly all of them have the same structural compliance problem. And almost none of their builders have noticed it yet.

What RAG Actually Does

Retrieval-Augmented Generation works like this:

User query → Embedding model → Vector store retrieval → LLM → Response

The vector store retrieves the most semantically similar documents to the query — regardless of who owns them, who is authorized to see them, or what regulatory framework governs them. Those documents land in the LLM's context window. The LLM synthesizes a response.

By the time the LLM responds, the retrieval has already happened. The documents were already in the context. If any of those documents were unauthorized for the requesting user, the disclosure has already occurred.

This is not a theoretical risk. It is a structural property of every standard RAG implementation.

The Regulations That Make This a Legal Problem

Different industries, same architectural gap:

Healthcare — HIPAA (45 CFR § 164)
Protected Health Information (PHI) may only be accessed by authorized workforce members under a valid Business Associate Agreement. A clinical RAG system that retrieves Patient A's records into a query for Patient B's provider has violated the minimum necessary standard — regardless of whether the LLM's final response mentions Patient A by name.

Financial Services — GLBA, SOX
Customer financial records, account data, and trading information carry strict access controls. A wealth management AI that retrieves one client's portfolio details during another client's session has a data segregation failure — not a prompt failure.

Higher Education — FERPA (34 CFR § 99)
Student education records — transcripts, financial aid, disciplinary files — are protected by the Family Educational Rights and Privacy Act. A student advising chatbot that retrieves another student's academic record into its context, even briefly, has made an unauthorized disclosure under § 99.31.

Europe — GDPR (Article 5(1)(f))
Personal data must be processed with appropriate security to prevent unauthorized access. A RAG pipeline that does not enforce user-level access control on retrieved documents violates the integrity and confidentiality principle at the architecture level.

The common thread: Every one of these regulations requires that unauthorized data not be accessed — not merely that it not be mentioned in the final output.

Why Prompt-Layer Controls Fail

The instinct is to fix this at the prompt:

"Only discuss information belonging to the current user. 
 Ignore anything that belongs to someone else."

This approach has three failure modes that make it insufficient for any regulated deployment:

1. The document is already disclosed.
When the vector store retrieves an unauthorized document, it enters the LLM's context window. The LLM has processed it. Under HIPAA, FERPA, and GDPR, access — not just output — constitutes disclosure. A prompt instruction cannot retroactively undo retrieval.

2. Prompt injection overrides instructions.
OWASP's LLM Top 10 (LLM01) identifies prompt injection as the primary attack vector against LLM applications. An adversarial user input can override system prompt instructions. Any compliance control implemented purely as a prompt instruction is one injection payload away from failure.

3. LLMs hallucinate and leak.
Language models occasionally surface information from their context in unexpected ways — in reasoning chains, in partial responses, in error messages. A compliance architecture that relies on the LLM "knowing not to mention" certain content is not an architecture; it is a hope.

The Right Fix: Pre-Filter, Not Post-Filter

The solution is architecturally simple: enforce access control between the retriever and the LLM, before documents enter the context window.

Before (standard, non-compliant):
User query → Retriever → [all retrieved docs] → LLM → Response

After (compliant):
User query → Retriever → [Compliance Pre-Filter] → [authorized docs only] → LLM → Response
                                    ↓
                            Audit record (disclosure log)

The pre-filter does three things:

Identity enforcement — documents tagged with a user or entity identifier are only passed to the LLM when the requesting user is authorized to see that entity's data
Category authorization — documents in restricted categories (PHI, financial records, disciplinary files) require explicit authorization, not just identity match
Audit logging — every retrieval event produces a structured disclosure record for compliance reporting

No document reaches the LLM context unless it has passed both checks. Shared content — knowledge base articles, policy documents, product documentation — passes through unchanged because it carries no identity metadata.

Implementation Across Frameworks

The enterprise-rag-patterns library implements this pattern across the major RAG frameworks:

Haystack 2.x

from haystack_integrations.components.filters.ferpa_filter import FERPAMetadataFilter

ferpa_filter = FERPAMetadataFilter(
    student_id="stu_001",
    institution_id="inst_abc",
    authorized_categories=["academic_record", "financial_aid"],
    requesting_user_id="advisor_007",
)
pipeline.add_component("ferpa_filter", ferpa_filter)
pipeline.connect("retriever.documents", "ferpa_filter.documents")

LangChain

from enterprise_rag_patterns import FERPAContextPolicy, make_enrollment_advisor_policy

policy = make_enrollment_advisor_policy(
    student_id="stu_001",
    institution_id="inst_abc",
)
filtered_docs = policy.filter(retrieved_docs)

HIPAA — any framework

from enterprise_rag_patterns.hipaa import HIPAADocumentFilter

hipaa_filter = HIPAADocumentFilter(
    patient_id="pat_001",
    provider_npi="1234567890",
    authorized_purposes=["treatment", "care_coordination"],
)

GDPR — consent-gated retrieval

from enterprise_rag_patterns.gdpr import GDPRConsentFilter

gdpr_filter = GDPRConsentFilter(
    data_subject_id="user_eu_001",
    processing_purpose="personalization",
    consent_store=your_consent_store,
)

Every filter emits a structured audit record on each run — timestamps, identity scope, categories disclosed, documents retrieved vs. disclosed. This is the disclosure log that HIPAA, FERPA, and GDPR all require in some form.

Document Metadata Design

The pattern requires that protected documents carry identity metadata at ingestion time:

# Protected record — only reaches authorized users
Document(
    content="Patient presented with chest pain, BP 140/90...",
    meta={
        "patient_id": "pat_001",
        "provider_npi": "1234567890",
        "record_type": "clinical_note",
        "data_classification": "PHI",
    }
)

# Shared content — no identity metadata, passes through for all users
Document(
    content="Standard dosing protocol for metformin...",
    meta={"record_type": "clinical_guideline"}
)

This is a design decision that must be made at the data pipeline level — not at the RAG pipeline level. If documents are ingested without identity metadata, no filter can enforce access control because there is nothing to enforce against. Getting the metadata schema right at ingestion is the prerequisite for any compliant RAG deployment.

What This Does Not Replace

Authentication — the pre-filter enforces an authorized identity scope; your application layer must establish that scope through authenticated session context
Encryption at rest — vector store encryption is outside this pattern's scope
Legal review — how your specific deployment maps to applicable regulations requires counsel; this pattern provides the technical control, not the legal interpretation

The Broader Point

The compliance gap in enterprise RAG is not a product gap — no major framework will solve this for you, because the solution requires knowing who the user is and what they are authorized to see. That context is application-specific. What a framework can provide is the enforcement mechanism; what your application must provide is the identity context.

The pre-filter pattern is that enforcement mechanism. It is not complex. It does not require a new architecture. It requires inserting one component between your retriever and your LLM — and designing your document metadata to carry the identity context the filter needs to do its job.

Every regulated enterprise deploying RAG today needs this. Most don't have it yet.

Resources:

enterprise-rag-patterns — github.com/ashutoshrana/enterprise-rag-patterns
ferpa-haystack — github.com/ashutoshrana/ferpa-haystack
regulated-ai-governance — github.com/ashutoshrana/regulated-ai-governance

FERPA Compliance in RAG Pipelines: Five Rules Your Enterprise System Probably Breaks

Ashutosh Rana — Sat, 11 Apr 2026 20:50:23 +0000

If you are building a retrieval-augmented generation (RAG) system for a higher-education institution, your pipeline is probably violating FERPA. Not because you meant to — but because the standard RAG tutorial pattern and the regulated record-access pattern are fundamentally different, and most documentation does not explain where they diverge.

This post covers five rules that most enterprise RAG implementations break, and what the correct pattern looks like for each.

What FERPA requires from a retrieval system

FERPA (Family Educational Rights and Privacy Act, 20 U.S.C. § 1232g; implementing regulations at 34 CFR Part 99) governs access to education records at institutions that receive federal funding.

The relevant requirement for a RAG pipeline is simple: a student's education records must not be accessible to another student or to an unauthorized third party.

In a vector store-backed system, "accessible" means more than whether the LLM produces the record in its response. It means whether the record enters the retrieval pipeline at all. A document that is retrieved, ranked, and then discarded by a post-filter has still been surfaced to a process that handles data for a different user.

Under FERPA's minimum-disclosure principle — and under any reasonable security posture — that is not acceptable.

Rule 1: Filter before ranking, not after

What most systems do: Retrieve the top-k documents from the vector store based on semantic similarity, then apply a metadata filter to remove documents that belong to the wrong student.

Why this breaks FERPA: The unauthorized documents are scored, ranked, and processed by the retrieval pipeline before being discarded. If the post-filter has a defect — a misconfigured field name, a missing metadata key, a swallowed exception — the unauthorized content reaches the LLM context window. The failure mode is silent and the blast radius is wide.

The correct pattern: Apply the identity constraint as a metadata pre-filter on the vector store query. Unauthorized documents should not exist in the candidate set.

# ❌ Wrong — retrieve all, then filter
all_docs = vector_store.similarity_search(query, k=20)
authorized = [d for d in all_docs if d.metadata["student_id"] == session.student_id]

# ✅ Correct — filter at query time
authorized = vector_store.similarity_search(
    query,
    k=20,
    filter={
        "student_id": session.student_id,
        "institution_id": session.institution_id,
    }
)

Most vector stores support metadata filtering natively: Pinecone, Weaviate, Qdrant, pgvector, and Chroma all support pre-filter expressions. Use them.

Rule 2: Filter on `institution_id`, not just `student_id`

What most systems do: Filter by student_id only.

Why this breaks FERPA: In a multi-tenant deployment, a student_id that is unique within Institution A may collide with a record at Institution B. More fundamentally, a student authorized to access their own records at Institution A should never retrieve records from Institution B — even if their student_id matches.

The correct pattern: Apply a compound AND filter: student_id == X AND institution_id == Y. Both conditions must be satisfied.

# ❌ Wrong — student_id alone
filter = {"student_id": session.student_id}

# ✅ Correct — compound identity predicate
filter = {
    "$and": [
        {"student_id": {"$eq": session.student_id}},
        {"institution_id": {"$eq": session.institution_id}},
    ]
}

Never query on student_id alone in a multi-institution deployment.

Rule 3: Enforce document categories as a second layer

What most systems do: Once the identity filter passes, all of the student's documents are fair game.

Why this breaks FERPA: Not all of a student's records are equally accessible. Counseling records, health records, disciplinary files, and financial aid records each have different access rules. Even if the current retrieval is authorized for identity, the category of document being retrieved matters.

A financial aid query that incidentally surfaces a counseling note is retrieving the right student's record — but the wrong type of record.

The correct pattern: After the identity pre-filter, apply a category authorization check. The authenticated session carries a set of permitted document categories. Documents outside that set are excluded.

# Session carries permitted categories (set by auth layer)
session.allowed_categories = {"academic_record", "financial_record"}

# Second enforcement layer — category filter
authorized = [
    doc for doc in identity_filtered_docs
    if doc.metadata.get("category") in session.allowed_categories
]

This is the two-layer enforcement model:

Layer 1 — Identity boundary: who owns this document?
Layer 2 — Category authorization: what type of document is this, and is the session permitted to retrieve it?

Rule 4: Every retrieval event must produce an audit record

What most systems do: Log at the application level — a timestamped entry that a user made a query.

Why this breaks FERPA: 34 CFR § 99.32 requires institutions to maintain a record of each disclosure of education records. "Disclosure" includes allowing access to records — which includes retrieval by an AI pipeline. The audit record must capture:

Who made the request
What was disclosed
The basis for disclosure
The date

An application log that records "user X made a query" does not satisfy this requirement.

The correct pattern: Produce a typed audit record for each retrieval event, containing the count of documents retrieved, the categories accessed, the policy version in effect, and the timestamp. Route it to a durable, student-accessible store — not just an application log.

audit_record = AuditRecord(
    student_id=session.student_id,
    institution_id=session.institution_id,
    documents_retrieved=len(raw_docs),
    documents_filtered=len(authorized_docs),
    categories_accessed=list(session.allowed_categories),
    policy_version="v1.2",
    timestamp=datetime.now(timezone.utc),
    requester_context={"session_id": session.id, "channel": session.channel},
)
audit_sink(audit_record)  # write to compliance database — not application log

Application logs rotate. FERPA compliance audit trails must be retained for as long as the education records themselves are retained.

Rule 5: Identity values must come from the session, not the query

What most systems do: Accept student_id and institution_id as parameters in the API request, or extract them from user-supplied query text.

Why this breaks FERPA: If the filter values come from the request, an attacker — or a misconfigured agent — can supply a different student's ID and retrieve their records. This is the most common vector for unauthorized record access in multi-tenant educational systems.

The correct pattern: The student_id and institution_id used for filtering must come from the authenticated session token — not from the request body, not from the query, not from user input.

# ❌ Wrong — accept from request body
student_id = request.params["student_id"]

# ✅ Correct — extract from verified session token
session = verify_token(request.headers["Authorization"])
student_id = session.student_id      # set by auth layer, not by user
institution_id = session.institution_id

This is not FERPA-specific — it is a basic authorization principle. In RAG systems it is easy to miss because most tutorials treat the retrieval query as the only input and ignore the access control context entirely.

What a compliant pipeline looks like

Authenticated session
(student_id + institution_id + allowed_categories — from verified token)
         │
         ▼
Vector store pre-filter query
(metadata filter: student_id AND institution_id — applied at query time)
         │
         ▼
Semantic ranking
(only authorized documents are candidates)
         │
         ▼
Category authorization check
(second enforcement layer — removes out-of-scope document types)
         │
         ▼
Context assembly → LLM call
         │
         ▼
Audit record (34 CFR § 99.32)
(student_id, institution_id, documents retrieved, categories, timestamp)
→ written to durable compliance store

The identity boundary is enforced twice — at the vector store and at the category level — before any document enters the LLM context window. The audit record is produced for every retrieval event, regardless of whether the LLM produces a response.

Reference implementation

The patterns described here are implemented in enterprise-rag-patterns, a MIT-licensed Python library:

pip install enterprise-rag-patterns

It provides:

StudentIdentityScope — defines the retrieval boundary per student and institution
FERPAContextPolicy — two-layer enforcement (pre-filter + category authorization)
AuditRecord — structured 34 CFR § 99.32 disclosure logging with a typed sink interface
make_enrollment_advisor_policy — factory for the most common higher-education RAG use case

The design is platform-agnostic (any vector store, any LLM provider) and cloud-agnostic (AWS, GCP, Azure, OCI, or on-premises). The same two-layer pattern applies to HIPAA's minimum-necessary standard and GLBA's safeguards rule.

A companion library regulated-ai-governance provides policy enforcement and audit for AI agents across FERPA, HIPAA, GDPR, CCPA, GLBA, and SOC 2.

Summary

Rule	What breaks	The fix
1. Filter before ranking	Post-retrieval filter leaves unauthorized docs in pipeline	Metadata pre-filter at vector store query time
2. Filter on `institution_id`	`student_id` alone allows cross-institution leakage	Compound `AND` filter: `student_id` + `institution_id`
3. Enforce document categories	All of student's records are accessible regardless of type	Category authorization as second enforcement layer
4. Audit every retrieval event	Application-level logs don't satisfy 34 CFR § 99.32	Typed `AuditRecord` per retrieval, routed to durable store
5. Identity from session	User-supplied filter values enable unauthorized access	Filter constructed from verified session token only

These are not edge cases. They are the default failure modes of standard RAG architectures when applied to regulated record-access environments. The fix for each is straightforward once you know where to look.

Reference implementation: github.com/ashutoshrana/enterprise-rag-patterns

DEV Community: Ashutosh Rana

EU AI Act Goes Live in 90 Days: What Developers Building AI Agents Actually Need to Do

What Actually Applies to You

The multi-agent problem

The Five Things You Actually Have to Build

1. Audit Logging (Article 12)

2. Human Oversight (Article 14)

3. Transparency (Article 13)

4. Accuracy and Robustness (Article 15)

5. Risk Management System (Article 9)

Building the Compliance Stack for Multi-Agent Systems

The August 2 Deadline: What's Actually Enforceable When

Where to Start

MACF: The 6-Component Framework Every Enterprise Multi-Agent AI System Needs

The Problem Nobody Talks About

Multi-Agent Systems in 90 Seconds

What Current Frameworks Provide (and What They Skip)

The MACF Architecture

1. AgentRegistry

2. MessageBus

3. ContextStore

4. PrivacyFilter

5. ComplianceGate

6. AuditTrail

How a Request Flows Through MACF

Getting Started

Why Six Components

Further Reading

Every Enterprise AI Framework Has a Compliance Gap — Here's the Architecture That Closes It

Every Enterprise AI Framework Has a Compliance Gap — Here's the Architecture That Closes It

The Problem: AI Frameworks Ship Without Compliance

What the Regulatory Environment Now Demands

EU AI Act — Penalties Active as of August 2025

OWASP Agentic AI Top 10 2026

HIPAA Security Rule Update — January 2025

Why Existing Mitigation Approaches Fail

1. Prompt-level instructions

2. Post-processing output filters

3. Manual review workflows

The Right Architecture: A Governance Layer Before Execution

Implementation: Drop-In Governance for Major Frameworks

CrewAI + FERPA

AutoGen + HIPAA

LangChain + GDPR

OWASP Agentic AI Top 10 Enforcement

What 25 Jurisdictions of Coverage Looks Like

The Audit Trail That Regulators Actually Want

What This Does Not Replace

The Market Is Moving Fast

Get Started

Google ADK Has a Compliance Gap — Here's How to Close It

What ADK Gives You

Why This Matters for Regulated Deployments

The ADKPolicyGuard Pattern

Multi-Agent Orchestration

The Audit Record

What This Does Not Replace

Getting Started

The Hidden Compliance Gap in Every Enterprise RAG Pipeline

What RAG Actually Does

The Regulations That Make This a Legal Problem

Why Prompt-Layer Controls Fail

The Right Fix: Pre-Filter, Not Post-Filter

Implementation Across Frameworks

Document Metadata Design

What This Does Not Replace

The Broader Point

FERPA Compliance in RAG Pipelines: Five Rules Your Enterprise System Probably Breaks

What FERPA requires from a retrieval system

Rule 1: Filter before ranking, not after

Rule 2: Filter on institution_id, not just student_id

Rule 3: Enforce document categories as a second layer

Rule 4: Every retrieval event must produce an audit record

Rule 5: Identity values must come from the session, not the query

What a compliant pipeline looks like

Reference implementation

Summary

Rule 2: Filter on `institution_id`, not just `student_id`