AIエージェント入門

AIエージェント マルチテナント設計パターン

AIエージェント マルチテナント設計パターン

この記事の結論

SaaS型AIエージェントのデータ隔離・テナント別設定。PostgreSQL RLS実装コード付き。

「AIエージェントのマルチテナント設計、何から始めればいいんですか?」

SaaS型のAIエージェント製品を作ろうとしているチームから、最近よく受ける質問だ。

実際に複数社のAIエージェント基盤構築を支援してきた経験からいうと、この問いには3つの層がある。データの隔離、権限の分離、そしてエージェント固有の記憶とクレデンシャルの管理だ。通常のSaaSとの最大の違いは3つ目で、ここで設計ミスをするとテナント間でデータが漏れる。

この記事では、PostgreSQL RLSによるデータレイヤーの隔離から、エージェント固有のクレデンシャル管理まで、コピペして使えるコード例と一緒に順を追って解説する。

AIエージェントの設計パターンの基礎については AIエージェント構築完全ガイドでもまとめているので、基本から押さえたい方はそちらも参照してほしい。

まず試したい:RLSで最も簡単なテナント隔離

PostgreSQL 9.5以降に搭載されているRow Level Security(RLS)は、アプリケーション層のバグがあってもデータベースレベルでクロステナントアクセスをブロックする。これが最も確実な防衛ラインだ。

セットアップは3ステップ。

-- ステップ1: テーブルにtenant_idカラムを追加
CREATE TABLE agent_conversations (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  tenant_id UUID NOT NULL,
  user_id UUID NOT NULL,
  message TEXT NOT NULL,
  role VARCHAR(20) NOT NULL,  -- 'user' | 'assistant'
  created_at TIMESTAMPTZ DEFAULT NOW()
);

-- ステップ2: RLSを有効化
ALTER TABLE agent_conversations ENABLE ROW LEVEL SECURITY;
ALTER TABLE agent_conversations FORCE ROW LEVEL SECURITY;  -- テーブルオーナーにも適用

-- ステップ3: ポリシーを定義
-- USING: SELECT/UPDATE/DELETE で表示する行を制限
-- WITH CHECK: INSERT/UPDATE で書き込める行を制限
CREATE POLICY tenant_isolation ON agent_conversations
  USING (tenant_id = current_setting('app.current_tenant_id')::UUID)
  WITH CHECK (tenant_id = current_setting('app.current_tenant_id')::UUID);

動作環境: PostgreSQL 9.5以上

アプリケーション層では、リクエストの最初にセッション変数を設定する。

import asyncpg
import uuid

async def get_db_connection(tenant_id: str):
    """テナントIDをセッション変数に設定したDB接続を返す"""
    conn = await asyncpg.connect(dsn="postgresql://...")

    # 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください
    # tenant_idは必ずバリデーション済みのUUIDを使うこと
    try:
        uuid.UUID(tenant_id)  # UUIDフォーマット検証
    except ValueError:
        raise ValueError(f"Invalid tenant_id format: {tenant_id}")

    await conn.execute(
        "SELECT set_config('app.current_tenant_id', $1, true)",
        tenant_id
    )
    return conn

# 使い方
async def get_agent_history(tenant_id: str, user_id: str):
    async with get_db_connection(tenant_id) as conn:
        # RLSが自動でtenant_idをフィルタリングする
        # アプリ層でWHERE tenant_id = ... を書く必要がない
        rows = await conn.fetch(
            "SELECT * FROM agent_conversations WHERE user_id = $1 ORDER BY created_at",
            user_id
        )
    return rows

これだけで、アプリ側でWHERE句を書き忘れても、データベースが自動でクロステナントアクセスをブロックする。

3つのテナント分離モデルと選び方

モデル 構成 スケーラビリティ コスト 向いているケース
共有DB + RLS(プールモデル) 1DB、全テナント共有 高(水平スケーリング容易) SMB・スタートアップ向けSaaS
スキーマ分離(ブリッジモデル) 1DB、テナント別スキーマ 中規模、マイグレーション管理が重要な場合
DB分離(サイロモデル) テナント別DB 低(管理複雑) 大企業・規制業界・コンプライアンス要件が厳しい場合

2026年のSaaS設計では「ハイブリッドモデル」が増えている。スタンダードプランは共有DB+RLS、エンタープライズプランはDB分離、という構成だ。

エージェント固有の問題:記憶とクレデンシャルの管理

通常のSaaSと違う最大の問題がここにある。AIエージェントは会話を記憶する。そのメモリが別テナントの会話に混入したら致命的だ。

エージェントのメモリをテナントスコープに閉じ込める設計を見てみよう。

import uuid
from dataclasses import dataclass, field
from typing import Optional
import json

@dataclass
class TenantScopedMemory:
    """テナント境界を持つエージェントメモリ"""
    tenant_id: str
    user_id: str
    conversation_id: str = field(default_factory=lambda: str(uuid.uuid4()))

    def __post_init__(self):
        # conversation_idはテナントとユーザーでスコープを付ける
        # グローバルなIDではなくテナント内でユニークなIDを使う
        self._memory_key = f"memory:{self.tenant_id}:{self.user_id}:{self.conversation_id}"

    def get_memory_key(self) -> str:
        return self._memory_key

class AgentMemoryStore:
    """Redisを使ったテナント分離メモリストア"""

    def __init__(self, redis_client):
        self.redis = redis_client

    async def save_memory(self, memory: TenantScopedMemory, data: dict, ttl_seconds: int = 86400):
        """メモリを保存(TTL付き)"""
        key = memory.get_memory_key()
        # 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください
        await self.redis.setex(key, ttl_seconds, json.dumps(data))

    async def get_memory(self, memory: TenantScopedMemory) -> Optional[dict]:
        """メモリを取得(必ずテナントスコープ内でのみ取得される)"""
        key = memory.get_memory_key()
        data = await self.redis.get(key)
        return json.loads(data) if data else None

    async def delete_all_tenant_memory(self, tenant_id: str):
        """テナント退会時の全メモリ削除(GDPR対応)"""
        pattern = f"memory:{tenant_id}:*"
        # SCAN + DELでAtomicに削除
        cursor = 0
        while True:
            cursor, keys = await self.redis.scan(cursor, match=pattern, count=100)
            if keys:
                await self.redis.delete(*keys)
            if cursor == 0:
                break

もう一つの問題はエージェントのクレデンシャルだ。テナントが自分のAPIキーを持ち込む(BYOK: Bring Your Own Key)設計では、あるテナントのリクエストが別テナントのキーで実行されると請求の汚染が起きる。

from typing import Dict

class TenantCredentialRouter:
    """テナント別APIキーのルーティング"""

    def __init__(self, credential_store):
        self.store = credential_store

    async def get_llm_client(self, tenant_id: str, provider: str):
        """テナント固有のクレデンシャルでLLMクライアントを返す"""
        # テナントのAPIキーを暗号化ストアから取得
        credentials = await self.store.get_credentials(tenant_id, provider)

        if not credentials:
            # フォールバック: プラットフォーム共用キーを使う(請求はテナントに課金)
            credentials = await self.store.get_platform_credentials(provider)

        # プロバイダー別にクライアントを初期化
        if provider == "openai":
            from openai import AsyncOpenAI
            return AsyncOpenAI(api_key=credentials["api_key"])
        elif provider == "anthropic":
            from anthropic import AsyncAnthropic
            return AsyncAnthropic(api_key=credentials["api_key"])
        else:
            raise ValueError(f"Unknown provider: {provider}")

    async def rotate_credentials(self, tenant_id: str, provider: str, new_key: str):
        """クレデンシャルのローテーション(古いキーのセッションを無効化)"""
        await self.store.update_credentials(tenant_id, provider, new_key)
        # TODO: 既存のアクティブセッションを全て無効化

権限分離:テナント内のRBAC設計

テナント間の隔離だけでなく、テナント内での権限分離も必要だ。エージェントは「誰の代わりに」動作しているかを常に明示的に持つべきだ。

from enum import Enum
from dataclasses import dataclass
from typing import List, Set

class AgentPermission(Enum):
    READ_DOCS = "read_docs"
    WRITE_DOCS = "write_docs"
    SEND_EMAIL = "send_email"
    ACCESS_CRM = "access_crm"
    EXECUTE_CODE = "execute_code"

@dataclass
class AgentContext:
    """エージェントの実行コンテキスト(テナント+ユーザー+権限のセット)"""
    tenant_id: str
    user_id: str
    role: str  # "admin" | "editor" | "viewer"
    granted_permissions: Set[AgentPermission]

    def can(self, permission: AgentPermission) -> bool:
        return permission in self.granted_permissions

    def require(self, permission: AgentPermission):
        if not self.can(permission):
            raise PermissionError(
                f"User {self.user_id} in tenant {self.tenant_id} "
                f"does not have permission: {permission.value}"
            )

# エージェントツールでの使い方
async def send_email_tool(context: AgentContext, to: str, subject: str, body: str):
    """メール送信ツール(権限チェック付き)"""
    # 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください
    context.require(AgentPermission.SEND_EMAIL)

    # 送信元をテナントのドメインに固定(他テナントのドメインで送れないようにする)
    tenant_config = await get_tenant_config(context.tenant_id)
    from_email = f"agent@{tenant_config['email_domain']}"

    # 実際の送信処理
    await email_service.send(from_email=from_email, to=to, subject=subject, body=body)

    # 監査ログ(テナント+ユーザーIDを記録)
    await audit_log.write(
        tenant_id=context.tenant_id,
        user_id=context.user_id,
        action="email_sent",
        metadata={"to": to, "subject": subject}
    )

【要注意】よくある設計ミスと回避策

失敗1:グローバルなconversation_idを使う

❌ UUIDを単純採番してテナント横断で使い回す
⭕ conversation_idにテナントIDをプレフィックスとして含める、またはDBレベルでtenant_idとセットでユニーク制約を張る

なぜ重要か: グローバルなIDがあると、IDを知っていれば別テナントのデータにアクセスできるInsecure Direct Object Reference(IDOR)脆弱性になる。

失敗2:エージェントメモリにテナントスコープを付けない

❌ Redisのキーをmemory:{user_id}:{conversation_id}にする
memory:{tenant_id}:{user_id}:{conversation_id}にする

なぜ重要か: user_idが別テナント間で衝突した場合(異なるテナントで同じメールアドレスを使う等)、メモリが混入する。テナントIDは必ず最上位のスコープにすること。

失敗3:RLSをスーパーユーザーに適用しない

❌ テーブルにENABLE ROW LEVEL SECURITYだけ設定する
FORCE ROW LEVEL SECURITYも設定する

なぜ重要か: RLSはデフォルトでスーパーユーザーとテーブルオーナーに適用されない。FORCEをつけないと管理者ユーザーがバイパスできてしまう。

失敗4:テナント削除時に関連データを消し忘れる

❌ usersテーブルのレコードだけ削除する
⭕ メモリストア(Redis)、ベクトルDB、ストレージ(S3等)を含む全データを削除するオフボーディングフローを用意する

なぜ重要か: GDPR/個人情報保護法の「忘れられる権利」対応。AIエージェントはベクトルDBや外部ストレージにデータを分散させるため、削除漏れが起きやすい。

セキュリティと運用のチェックリスト

カテゴリ チェック項目 優先度
データ隔離 全テーブルにRLSポリシーを設定済みか 必須
データ隔離 FORCE ROW LEVEL SECURITYを設定済みか 必須
メモリ Redisキーにtenant_idをプレフィックスとして含むか 必須
メモリ テナント削除時のメモリ一括削除フローがあるか 必須
クレデンシャル テナントAPIキーを暗号化して保存しているか 必須
権限 全エージェントツールに権限チェックがあるか 必須
監査 全ツール実行でtenant_id+user_idを監査ログに記録するか 推奨
コスト テナント別のAPI使用量を追跡しているか 推奨

まとめ:今日から始める3つのアクション

  1. 今日やること: 既存のエージェントテーブルにRLSを追加する。まず最も機密性の高いテーブル1つから始めてみる。
  2. 今週中: エージェントメモリのキー設計を見直し、tenant_idが先頭に来ているか確認する。なければリファクタリング計画を立てる。
  3. 今月中: テナントオフボーディングフローを実装する。DB・Redis・S3・ベクトルDBの全データを削除するスクリプトをテストする。

参考・出典

あわせて読みたい:
AIエージェントのメモリ設計|Short/Long/Episodic実装ガイド — メモリ設計の詳細はこちら
AIエージェントのコスト最適化|トークン削減5つの戦略 — マルチテナントでのコスト配分も参考に

著者: 佐藤傑(さとう・すぐる)
株式会社Uravation代表取締役。X(@SuguruKun_ai)フォロワー10万人超。
100社以上の企業向けAI研修・導入支援。著書累計3万部突破。
SoftBank IT連載7回執筆(NewsPicks最大1,125ピックス)。

ご質問・ご相談は お問い合わせフォーム からお気軽にどうぞ。

Need help moving from reading to rollout?

この記事を読んで導入イメージが固まってきた方へ

Uravationでは、AIエージェントの要件整理、PoC設計、社内導入、研修まで一気通貫で支援しています。

この記事をシェア

X Facebook LINE

※ 本記事の情報は2026年4月時点のものです。サービスの料金・仕様は変更される可能性があります。最新情報は各サービスの公式サイトをご確認ください。

関連記事