Database-Per-Service Pattern for AI Agent Microservices: Data Isolation and Consistency

The Shared Database Anti-Pattern

Many teams decompose a monolithic agent into microservices but leave the database shared. The conversation service, tool execution engine, and memory service all read from and write to the same PostgreSQL instance, the same tables, sometimes the same rows.

This defeats the purpose of microservices. A schema change by the memory team can break the conversation service. A slow query from the analytics service can lock rows needed by the tool execution engine. Deployments remain coupled because services share data structures.

The database-per-service pattern gives each microservice its own database that only it can access directly. Other services interact with that data through the owning service's API.

Data Ownership Boundaries

Each service owns the data it needs to fulfill its responsibilities. For an AI agent system, ownership maps naturally:

# Conversation Service — owns session and message data
# Database: PostgreSQL (relational, good for structured sessions)
"""
Tables:
  sessions (id, user_id, created_at, status, metadata)
  messages (id, session_id, role, content, tokens, created_at)
  routing_decisions (id, message_id, intent, confidence, tool_name)
"""

# RAG Retrieval Service — owns document and embedding data
# Database: PostgreSQL + pgvector (vector search)
"""
Tables:
  documents (id, source, content, chunk_index, metadata)
  embeddings (id, document_id, vector, model_name)
  retrieval_logs (id, query_hash, results, latency_ms)
"""

# Tool Execution Service — owns tool registry and execution logs
# Database: PostgreSQL
"""
Tables:
  tools (id, name, description, schema, enabled, version)
  executions (id, tool_id, params, result, duration_ms, status)
  rate_limits (tool_id, client_id, window_start, count)
"""

# Memory Service — owns long-term agent memory
# Database: Redis + PostgreSQL
"""
Redis: short-term working memory (session context, recent facts)
PostgreSQL:
  memory_entries (id, user_id, content, category, importance, created_at)
  memory_relationships (id, source_id, target_id, relation_type)
"""

Kubernetes Deployment with Separate Databases

Each service gets its own database instance. Here is the Kubernetes configuration for the conversation service and its dedicated database:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: conversation-db
  namespace: agent-system
spec:
  replicas: 1
  selector:
    matchLabels:
      app: conversation-db
  template:
    metadata:
      labels:
        app: conversation-db
    spec:
      containers:
        - name: postgres
          image: postgres:16
          env:
            - name: POSTGRES_DB
              value: conversation
            - name: POSTGRES_USER
              valueFrom:
                secretKeyRef:
                  name: conversation-db-creds
                  key: username
            - name: POSTGRES_PASSWORD
              valueFrom:
                secretKeyRef:
                  name: conversation-db-creds
                  key: password
          volumeMounts:
            - name: data
              mountPath: /var/lib/postgresql/data
      volumes:
        - name: data
          persistentVolumeClaim:
            claimName: conversation-db-pvc
---
apiVersion: v1
kind: Service
metadata:
  name: conversation-db
  namespace: agent-system
spec:
  selector:
    app: conversation-db
  ports:
    - port: 5432
  # No external access — only the conversation service connects
  type: ClusterIP

The conversation service is the only service with credentials to conversation-db. If the RAG service needs session data, it calls the conversation service's API.

Handling Cross-Service Queries with API Composition

When a dashboard needs data from multiple services — session count from the conversation service, retrieval latency from the RAG service, tool success rate from the tool service — use an API composition layer:

import asyncio
import httpx

class AgentDashboardComposer:
    def __init__(self):
        self.client = httpx.AsyncClient(timeout=10.0)
        self.services = {
            "conversation": "http://conversation-manager:8000",
            "rag": "http://rag-retrieval:8002",
            "tools": "http://tool-execution:8001",
        }

    async def get_dashboard_stats(self, time_range: str) -> dict:
        # Fetch from all services in parallel
        results = await asyncio.gather(
            self.client.get(
                f"{self.services['conversation']}/stats",
                params={"range": time_range},
            ),
            self.client.get(
                f"{self.services['rag']}/stats",
                params={"range": time_range},
            ),
            self.client.get(
                f"{self.services['tools']}/stats",
                params={"range": time_range},
            ),
            return_exceptions=True,
        )

        stats = {}
        for name, result in zip(self.services.keys(), results):
            if isinstance(result, Exception):
                stats[name] = {"error": str(result)}
            else:
                stats[name] = result.json()

        return stats

The Saga Pattern for Multi-Service Transactions

When an operation must update data across multiple services atomically — for example, creating a new session (conversation service), initializing memory (memory service), and registering usage (billing service) — use the saga pattern:

class CreateSessionSaga:
    def __init__(self, conversation_client, memory_client, billing_client):
        self.conversation = conversation_client
        self.memory = memory_client
        self.billing = billing_client

    async def execute(self, user_id: str, config: dict) -> dict:
        session = None
        memory_initialized = False

        try:
            # Step 1: Create session
            session = await self.conversation.create_session(
                user_id, config
            )

            # Step 2: Initialize memory for session
            await self.memory.initialize(
                session_id=session["id"],
                user_id=user_id,
            )
            memory_initialized = True

            # Step 3: Register usage
            await self.billing.register_session(
                user_id=user_id,
                session_id=session["id"],
            )

            return session

        except Exception as e:
            # Compensating transactions (rollback in reverse)
            if memory_initialized:
                await self.memory.cleanup(session["id"])
            if session:
                await self.conversation.delete_session(session["id"])
            raise e

Each step has a compensating action. If step 3 fails, the saga rolls back steps 2 and 1. This gives eventual consistency without distributed transactions.

Eventual Consistency Considerations

With separate databases, data will be temporarily inconsistent across services. The conversation service might record a new message before the memory service indexes it. This is acceptable as long as the system converges to a consistent state.

Design your APIs to be tolerant of temporary inconsistency. If the memory service returns stale results, the agent's response might be slightly less contextual — but the system does not break.

FAQ

Does database-per-service mean I need to run and manage many database instances?

Yes, but managed database services (RDS, Cloud SQL) reduce the operational burden. Alternatively, you can run one PostgreSQL cluster with separate databases (not just schemas) per service. Each service gets its own database with its own credentials, preventing cross-service access while sharing the same database server.

How do I handle reporting that needs data from multiple services?

Use event-driven data replication. Each service publishes events when its data changes. A dedicated analytics service consumes these events and builds a denormalized read model optimized for reporting queries. This keeps operational databases fast while providing the cross-service joins that dashboards need.

What about referential integrity across service boundaries?

You cannot enforce foreign keys across databases. Instead, validate references at the application level. When the conversation service references a tool by ID, it calls the tool service to verify the tool exists before storing the reference. Accept that cross-service references can become stale and design your error handling to gracefully handle missing references.

---

#Database #Microservices #SagaPattern #DataIsolation #AgenticAI #EventualConsistency #LearnAI #AIEngineering