Federated & Privacy-Preserving Agents
Every agent needs data to be useful. A customer support agent needs access to account records. A medical agent needs patient histories. A legal agent needs confidential contracts. The problem is that the data most valuable to an agent is often the data most dangerous to centralize. Sending raw patient records to a cloud inference endpoint, piping confidential contracts through a third-party model API, aggregating financial data from multiple subsidiaries into a single context window — each of these is a compliance incident waiting to happen.
The traditional agent architecture assumes that data flows freely to wherever the model runs. Privacy-preserving agents follow a different principle: bring computation to the data. The model can run where the data already lives, receive a transformed and safe representation, or collaborate across data holders while each party sees only its authorized portion.
These techniques are structural requirements for deploying agents in healthcare, finance, legal, government, and any domain where compliance mandates restrict how data moves, who sees it, and where inference happens.
The Data Residency Problem #
Data residency is the simplest constraint and the most common blocker. A regulation may require data to remain within a jurisdiction or processing to stay inside a private network. The agent needs the data to do its job, so its model placement must satisfy those boundaries.
The problem:
┌──────────────────────┐ ┌──────────────────────┐
│ Data (EU region) │ │ Model (US cloud) │
│ │ ───► │ │
│ Patient records │ upload │ Large inference │
│ Financial reports │ │ endpoint │
│ Personnel files │ │ │
└──────────────────────┘ └──────────────────────┘
GDPR says no.
The fix:
┌──────────────────────┐
│ Data + Model │
│ (EU region) │
│ │
│ On-premise or │
│ same-region │
│ inference │
└──────────────────────┘
Data never leaves.
The architectural options, in order of increasing complexity:
Same-region cloud inference. Run the model on cloud infrastructure within the same jurisdiction as the data. Most cloud providers offer region-specific model endpoints. The data crosses a network boundary but stays within the regulated jurisdiction.
On-premise inference. Run the model on hardware you control, inside your own network. All data remains inside that environment. This requires hosting infrastructure and managing model deployment, while giving you complete control over data flow.
Edge inference. Run a smaller model directly on the device where the data originates. A medical device agent that processes vitals on the device itself eliminates network transmission entirely. The trade-off is model capability — edge models are smaller and less capable.
Architecture: Data-Residency-Aware Tool Routing #
When an agent uses tools that access data from different jurisdictions or sensitivity levels, the tool routing layer needs to enforce residency constraints. Each tool call must route to an eligible backend.
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field
from enum import Enum
class DataRegion(Enum):
EU = "eu"
US = "us"
APAC = "apac"
ON_PREMISE = "on_premise"
class SensitivityLevel(Enum):
PUBLIC = "public"
INTERNAL = "internal"
CONFIDENTIAL = "confidential"
RESTRICTED = "restricted"
@dataclass
class DataPolicy:
allowed_regions: list[DataRegion]
max_sensitivity: SensitivityLevel
requires_encryption: bool = False
requires_audit_log: bool = False
allowed_model_endpoints: list[str] = field(default_factory=list)
@dataclass
class ToolMetadata:
name: str
data_region: DataRegion
sensitivity: SensitivityLevel
fn: Callable[..., Awaitable[str]]
class PolicyAwareRouter:
"""
Routes tool calls only to backends that satisfy
data residency and sensitivity constraints.
"""
def __init__(self, policies: dict[str, DataPolicy]):
self.policies = policies
def can_execute(
self,
tool: ToolMetadata,
execution_region: DataRegion,
model_endpoint: str,
*,
encrypted: bool,
audit_log_enabled: bool,
) -> tuple[bool, str]:
policy = self.policies.get(tool.name)
if policy is None:
return False, f"No policy defined for tool '{tool.name}'"
if execution_region not in policy.allowed_regions:
return False, (
f"Tool '{tool.name}' cannot execute in {execution_region.value}. "
f"Allowed: {[r.value for r in policy.allowed_regions]}"
)
sensitivity_order = list(SensitivityLevel)
if sensitivity_order.index(tool.sensitivity) > sensitivity_order.index(policy.max_sensitivity):
return False, (
f"Tool '{tool.name}' handles {tool.sensitivity.value} data, "
f"exceeding max {policy.max_sensitivity.value}"
)
if policy.allowed_model_endpoints and model_endpoint not in policy.allowed_model_endpoints:
return False, (
f"Model endpoint '{model_endpoint}' not authorized "
f"for tool '{tool.name}'"
)
if policy.requires_encryption and not encrypted:
return False, f"Tool '{tool.name}' requires encrypted transport"
if policy.requires_audit_log and not audit_log_enabled:
return False, f"Tool '{tool.name}' requires audit logging"
return True, "OK"
async def execute_tool(
self,
tool: ToolMetadata,
arguments: dict,
execution_region: DataRegion,
model_endpoint: str,
*,
encrypted: bool,
audit_log_enabled: bool,
) -> str:
allowed, reason = self.can_execute(
tool,
execution_region,
model_endpoint,
encrypted=encrypted,
audit_log_enabled=audit_log_enabled,
)
if not allowed:
return f"Policy violation: {reason}"
return await tool.fn(**arguments)
The router sits between the agent's orchestration loop and the tool execution layer. Every tool call passes through a policy check before it runs. If the check fails, the agent gets an error message explaining why — which lets the model adjust its approach by choosing another tool, requesting data through a compliant channel, or explaining the available compliant options to the user.
On-Premise Inference #
Running models inside your own network is the most direct way to keep data private. All processing stays within the perimeter. On-premise inference also introduces engineering challenges that cloud endpoints abstract away.
Model hosting. You need GPU infrastructure to run inference at acceptable latency. For small models (1B–7B parameters), a single modern GPU handles real-time inference. For larger models (30B+), you need multiple GPUs or inference-optimized hardware. Quantized models reduce the hardware requirement significantly — a 70B model quantized to INT4 fits on a single high-end GPU.
Model updates. Cloud endpoints update models silently. On-premise, you manage versions explicitly. This is actually an advantage for compliance — you control exactly which model version is running, and you can pin versions for reproducibility and audit purposes.
Scaling. Cloud inference scales automatically with load. On-premise, you provision capacity in advance. Over-provisioning wastes money; under-provisioning creates latency spikes or request failures.
┌──────────────────────────────────────────────────────┐
│ On-Premise Agent Architecture │
│ │
│ ┌─────────────────┐ ┌──────────────────────────┐ │
│ │ Agent │ │ Internal Data Stores │ │
│ │ Orchestrator │◄──►│ (database, files, APIs) │ │
│ │ │ └──────────────────────────┘ │
│ │ ┌───────────┐ │ │
│ │ │ Policy │ │ ┌──────────────────────────┐ │
│ │ │ Router │ │ │ Local Model Server │ │
│ │ └───────────┘ │◄──►│ (vLLM, TGI, Ollama) │ │
│ │ │ │ │ │
│ └─────────────────┘ │ GPU: A100 / H100 / L40 │ │
│ │ Model: 7B–70B quantized │ │
│ └──────────────────────────┘ │
│ │
│ ═══════════════ Network Boundary ═════════════════ │
│ │
│ Nothing crosses this line. │
└──────────────────────────────────────────────────────┘
The agent orchestrator, the model server, and the data stores all live inside the same network. Tool calls to internal databases stay within the perimeter. External communication is limited to public services such as web search and documentation lookup, using sanitized payloads.
Redaction and Data Masking #
Redaction enables the use of a powerful cloud model by stripping or masking sensitive fields before they reach it, then restoring them in the output.
import re
from dataclasses import dataclass
@dataclass
class RedactionMapping:
"""Maps redacted placeholders back to original values."""
mappings: dict[str, str] # placeholder -> original
def restore(self, text: str) -> str:
result = text
for placeholder, original in self.mappings.items():
result = result.replace(placeholder, original)
return result
class Redactor:
"""
Replaces sensitive data with consistent placeholders
before sending to the model.
"""
PII_PATTERNS = {
"email": r'\b[\w.+-]+@[\w-]+\.[\w.-]+\b',
"phone": r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b',
"ssn": r'\b\d{3}-\d{2}-\d{4}\b',
"credit_card": r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b',
}
def __init__(self):
self._counter = 0
self._seen: dict[str, str] = {}
def redact(self, text: str) -> tuple[str, RedactionMapping]:
result = text
mappings = {}
for field_type, pattern in self.PII_PATTERNS.items():
for match in re.finditer(pattern, result):
original = match.group()
if original not in self._seen:
self._counter += 1
placeholder = f"[{field_type.upper()}_{self._counter}]"
self._seen[original] = placeholder
placeholder = self._seen[original]
mappings[placeholder] = original
result = result.replace(original, placeholder)
return result, RedactionMapping(mappings=mappings)
The agent loop integrates redaction as a transparent layer:
async def privacy_aware_agent_step(
model: "ModelClient",
messages: list[dict],
redactor: Redactor,
) -> tuple[str, list[dict]]:
# Redact all messages before sending to the model
redacted_messages = []
all_mappings = RedactionMapping(mappings={})
for msg in messages:
redacted_text, mapping = redactor.redact(msg["content"])
redacted_messages.append({**msg, "content": redacted_text})
all_mappings.mappings.update(mapping.mappings)
# The model only sees redacted content
response = await model.chat(
system="You are an assistant. References like [EMAIL_1] are placeholders.",
messages=redacted_messages,
)
# Restore original values in the output
restored_text = all_mappings.restore(response.text)
return restored_text, redacted_messages
The model sees safe placeholders such as [EMAIL_1] and [PHONE_2]. It can still reason about the content — "the customer's email is [EMAIL_1]" — and the restoration layer swaps placeholders back to real values before the user sees the response.
Limitations. Redaction works for structured PII — emails, phone numbers, IDs. Unstructured sensitive information, such as a patient's description of symptoms, a detailed legal dispute, or a confidential negotiation, often carries its meaning in the sensitive content itself. These cases call for on-premise inference or federated approaches.
Federated Agent Architectures #
Federated architectures let multiple data holders collaborate while each party retains its raw data locally. Each party runs part of the agent workflow and shares only derived results — summaries, scores, classifications.
Hub-and-Spoke Federation #
A central orchestrator coordinates the task while data remains at each spoke. Every spoke runs a local agent or tool that processes its own data and returns a sanitized result.
┌───────────────────┐
│ Orchestrator │
│ (central hub) │
│ │
│ Sees: summaries │
│ Never sees: │
│ raw data │
└─────┬──────┬──────┘
│ │
┌───────────┘ └───────────┐
▼ ▼
┌───────────────────────┐ ┌───────────────────────┐
│ Site A │ │ Site B │
│ Local agent + │ │ Local agent + │
│ local data │ │ local data │
│ │ │ │
│ Processes patient │ │ Processes patient │
│ records locally. │ │ records locally. │
│ Returns: aggregate │ │ Returns: aggregate │
│ stats, no PII. │ │ stats, no PII. │
└───────────────────────┘ └───────────────────────┘
@dataclass
class FederatedQuery:
task: str
required_output_format: str
privacy_constraints: list[str]
@dataclass
class SiteResult:
site_id: str
summary: str
record_count: int
aggregates: dict
# No raw records — only derived values
class FederatedOrchestrator:
"""
Coordinates agent work across multiple data sites
without centralizing raw data.
"""
def __init__(self, sites: dict[str, "SiteAgent"], model: "ModelClient"):
self.sites = sites
self.model = model
async def execute(self, query: FederatedQuery) -> str:
# Fan out: each site processes the query against its local data
import asyncio
import json
results = await asyncio.gather(*(
site_agent.process_locally(query)
for site_agent in self.sites.values()
))
# Aggregate: the orchestrator sees only summaries
combined = self._merge_results(results)
# Synthesize: use the model to produce the final answer
response = await self.model.chat(
system="Synthesize findings from multiple sites. You see only aggregates.",
messages=[{"role": "user", "content": json.dumps(combined)}],
)
return response.text
def _merge_results(self, results: list[SiteResult]) -> dict:
return {
"sites_queried": len(results),
"total_records": sum(r.record_count for r in results),
"site_summaries": [
{"site": r.site_id, "summary": r.summary, "aggregates": r.aggregates}
for r in results
],
}
The orchestrator's model sees aggregated summaries such as "Site A analyzed 1,200 patient records and found a 15% prevalence rate," while individual records remain at their source. Each site runs its own local agent (potentially a smaller model or even a deterministic pipeline) with full access to its local data.
Peer-to-Peer Federation #
In a peer-to-peer federation, trust is distributed across the participating sites. Each site processes its data, shares derived results with peers, and the final synthesis happens locally at whichever site initiated the query.
This pattern is more complex to coordinate and distributes aggregation risk across the consortium. It suits multiple hospitals collaborating on research or financial institutions sharing fraud signals while each member keeps its detailed records private.
Differential Privacy in Agent Memory #
Agents with long-term memory accumulate information across sessions. Over many interactions, that memory can become a detailed profile of a user — their preferences, habits, medical conditions, financial situation. Even if individual interactions are private, the aggregate memory can reveal more than any single session should.
Differential privacy adds calibrated noise to data so that every individual contribution receives statistical protection. Applied to agent memory, it preserves useful aggregate knowledge while protecting each user's information.
import random
import math
_secure_random = random.SystemRandom()
def add_laplace_noise(value: float, sensitivity: float, epsilon: float) -> float:
"""
Add Laplace noise for differential privacy.
Lower epsilon = more privacy, more noise.
"""
if epsilon <= 0:
raise ValueError("epsilon must be greater than zero")
if sensitivity < 0:
raise ValueError("sensitivity must be non-negative")
if sensitivity == 0:
return value
scale = sensitivity / epsilon
uniform = _secure_random.random() - 0.5
while abs(uniform) == 0.5:
uniform = _secure_random.random() - 0.5
noise = (
-scale
* math.copysign(1.0, uniform)
* math.log1p(-2 * abs(uniform))
)
return value + noise
class DifferentiallyPrivateMemory:
"""
Agent memory that applies differential privacy
to aggregate queries over stored facts.
"""
def __init__(self, epsilon: float = 1.0, total_epsilon_budget: float = 10.0):
if epsilon <= 0:
raise ValueError("epsilon must be greater than zero")
if total_epsilon_budget < epsilon:
raise ValueError("total_epsilon_budget must cover at least one query")
self.epsilon = epsilon
self.remaining_epsilon = total_epsilon_budget
self._records: list[dict] = []
def store(self, record: dict) -> None:
self._records.append(record)
def _consume_budget(self) -> None:
if self.remaining_epsilon < self.epsilon:
raise RuntimeError("Differential privacy budget exhausted")
self.remaining_epsilon -= self.epsilon
def count_matching(self, predicate) -> float:
"""Return a noisy count of records matching a condition."""
self._consume_budget()
true_count = sum(1 for r in self._records if predicate(r))
return add_laplace_noise(true_count, sensitivity=1.0, epsilon=self.epsilon)
def average_field(
self,
field: str,
lower_bound: float,
upper_bound: float,
) -> float | None:
"""Return a noisy average after clipping values to public bounds."""
if lower_bound >= upper_bound:
raise ValueError("lower_bound must be less than upper_bound")
self._consume_budget()
values = [
min(max(float(r[field]), lower_bound), upper_bound)
for r in self._records
if field in r
]
if not values:
return None
true_avg = sum(values) / len(values)
# Conservative global sensitivity for a bounded mean. A tighter
# range / n bound requires a fixed, public group size.
sensitivity = upper_bound - lower_bound
return add_laplace_noise(true_avg, sensitivity, self.epsilon)
The epsilon parameter controls the privacy-utility trade-off. A small epsilon (strong privacy) adds more noise, making individual contributions undetectable but aggregate queries less accurate. A large epsilon (weak privacy) adds less noise, preserving accuracy but offering less protection. Typical values range from 0.1 (strong) to 10.0 (weak).
Where this applies in practice: an agent remembers aggregate patterns across customers ("customers in this segment typically prefer X") while individual details stay private. Calibrated noise makes the inclusion of any specific customer's data statistically ambiguous.
Secure Computation Patterns #
For the most sensitive operations, keep sensitive data behind a tool boundary and give the model only the computed result.
Tool-Side Computation #
Push sensitive computation into the tool. The model describes what it wants to know; the tool computes the answer locally and returns only the result.
# Bad: model sees raw data
import json
async def bad_check_eligibility(patient_id: str) -> str:
records = db.query(
"SELECT * FROM patients WHERE id = %s",
(patient_id,),
)
return json.dumps(records) # Raw patient data goes to the model
# Good: model gets only the answer
async def good_check_eligibility(patient_id: str) -> str:
row = db.fetch_one(
"SELECT CASE WHEN age >= 18 AND consent_signed = TRUE "
"THEN 'eligible' ELSE 'not_eligible' END AS status "
"FROM patients WHERE id = %s",
(patient_id,),
)
status = row["status"] if row else "patient_not_found"
return json.dumps({"patient_id": patient_id, "status": status})
The model asks "is this patient eligible?" and gets back "eligible" or "not_eligible." The patient's age, consent status, medical history, and other source fields remain inside the database, where SQL logic determines eligibility behind the tool boundary.
This is the most practical privacy technique for most agent systems. Design tools that answer questions and retain source data behind the tool boundary. The model reasons about the answers.
Split Inference #
For scenarios where the model needs to process sensitive content — classifying a medical image, extracting information from a legal document — split inference separates the work between a local model and a remote model.
┌──────────────────────────────────┐
│ Local Environment │
│ │
│ Sensitive ┌──────────┐ │
│ document ────────►│ Local │ │
│ │ model │ │
│ │ (small) │ │
│ └────┬─────┘ │
│ │ │
│ Extracted │ │
│ features │ │
│ (no PII) │ │
└─────────────────────────┼────────┘
│
▼
┌──────────────────────────────────┐
│ Cloud Environment │
│ │
│ ┌──────────┐ │
│ Features ────────►│ Large │ │
│ only │ model │ │
│ │ (cloud) │ │
│ └────┬─────┘ │
│ │ │
│ Final │ │
│ answer │ │
└─────────────────────────┼────────┘
│
▼
Response
The local model extracts safe structured features — "this is a contract, dated 2024-03-15, between two parties, with a termination clause in section 7." The cloud model receives that structure and produces reasoning based on it, while names, financial terms, full clauses, and other sensitive content remain in the local environment.
Privacy-Aware Agent Design Checklist #
When building an agent that handles sensitive data, work through these questions at design time:
- Where does the data live, and where can it move? Map data residency requirements before choosing your inference architecture.
- What does the model actually need to see? Design tools that answer questions and keep raw data behind the tool boundary. Minimize what enters the context window.
- Can structured PII be redacted? If the sensitive fields are identifiers (names, emails, account numbers), redaction lets you use cloud models safely.
- Is the sensitive content itself what the model needs to reason about? If so, use on-premise inference or split inference so the model can reason directly while preserving the privacy boundary.
- Does the agent's memory accumulate sensitive patterns? Apply differential privacy to aggregate queries over memory, or set retention limits that delete individual records after use.
- Are multiple parties involved? Use federated patterns where each party processes its own data locally and shares only derived results.
- Can you audit what data the model saw? Log redacted inputs and tool queries so you can demonstrate compliance during audits.
Conclusion #
Privacy-preserving agents are regular agents with additional constraints on how data flows through the system. The techniques map to different threat profiles:
- Data residency is solved by running inference where the data lives — same-region cloud, on-premise, or edge. The policy-aware router enforces these constraints at the tool level.
- Redaction protects structured PII such as identifiers and contact information when using cloud models. Free-form sensitive content requires on-premise or federated processing.
- Federation lets multiple data holders collaborate while each site retains its raw records. The orchestrator sees summaries.
- Differential privacy protects individual contributions in agent memory, making it useful for aggregate knowledge with user-level privacy.
- Tool-side computation is the most practical technique for most systems. Design tools that answer questions and keep sensitive data within the tool.
- Split inference handles cases where the model must process sensitive content. A local model extracts features; a cloud model reasons about them.
The common thread: minimize what the model sees. Every token of sensitive data in the context window is a token that could leak through logs, model training feedback, or a prompt injection attack. The less the model knows, the less there is to protect.