社内のドキュメント検索ツールを Claude API で作り始めたとき、最初に私がやったのは「とりあえずベクター検索にドキュメントを全部突っ込む」でした。おそらく多くの方が同じスタートラインに立つと思います。
そしてデモを見せたとき、人事部の担当者が「あれ、この給与データって全員が見えてるんですか？」と聞いてきました。
その瞬間、設計のやり直しが確定しました。
アクセス制御のないRAGシステムは、検索性能がいくら高くても本番には出せません。そこから3週間かけて作り直した「部署別アクセス制御付きの社内ドキュメント検索エージェント」—その実装をコードと設計の両面から共有します。
システム全体のアーキテクチャ
完成したシステムの構成は以下の通りです。
- インジェスト層: PDF・Markdown・Notion Export → テキスト抽出 → チャンキング → エンベディング生成 → pgvector に保存
- 検索層: ハイブリッド検索（pgvector の cos 距離 + PostgreSQL の全文検索 BM25）→ RRF でスコア統合 → RBAC フィルタリング
- 回答生成層: フィルタリング済みチャンク → Claude API（claude-sonnet-4-6）→ 回答 + 引用元
- 監査層: 全クエリ・回答・参照ドキュメント → 監査テーブルに記録
ポイントは「RBACをアプリ層ではなく DB 層（PostgreSQL RLS）で実装する」という設計判断です。アプリ側でフィルタリングすると、バグが情報漏洩に直結します。DB のRow Level Securityに任せることで、どんなバグがあっても見せてはいけないドキュメントは返ってこない設計にしています。
ドキュメントインジェストパイプラインの実装
まずドキュメントをベクターDBに登録する部分から実装します。
import anthropic
import psycopg2
from psycopg2.extras import execute_values
import hashlib
from pathlib import Path
from typing import Optional
import pypdf
import tiktoken
client = anthropic.Anthropic()
enc = tiktoken.encoding_for_model("gpt-4o") # トークン数計測用（近似値）
def extract_text_from_pdf(pdf_path: str) -> str:
"""PDFからテキストを抽出する"""
text_parts = []
with open(pdf_path, "rb") as f:
reader = pypdf.PdfReader(f)
for page in reader.pages:
text_parts.append(page.extract_text())
return "\n".join(text_parts)
def chunk_document(
text: str,
chunk_size: int = 512,
overlap: int = 64
) -> list[dict]:
"""
テキストをチャンクに分割する。
- chunk_size: トークン数（近似）
- overlap: チャンク間の重複トークン数（文脈の継続性を保つ）
"""
tokens = enc.encode(text)
chunks = []
start = 0
while start < len(tokens):
end = min(start + chunk_size, len(tokens))
chunk_text = enc.decode(tokens[start:end])
# チャンクが短すぎる場合（文書末尾）はスキップ
if len(enc.encode(chunk_text)) < 50:
break
chunks.append({
"text": chunk_text,
"token_start": start,
"token_end": end
})
start += chunk_size - overlap
return chunks
def generate_embedding(text: str) -> list[float]:
"""
Claude API でエンベディングを生成する。
注: 2026年5月時点では Anthropic 独自のエンベディングモデルは
提供されていないため、voyage-3-large などを使うのが現実的。
ここでは概念実装として claude-sonnet-4-6 を使ったエンベディング生成を示す。
実際には voyage-ai SDK に置き換えることを推奨。
"""
# 実際の実装では voyage-ai などのエンベディングAPIを使う
# from voyageai import Client as VoyageClient
# voyage = VoyageClient(api_key=os.environ["VOYAGE_API_KEY"])
# result = voyage.embed([text], model="voyage-3-large")
# return result.embeddings[0]
# デモ用: claude-sonnet-4-6 でエンベディング代替（実際には非推奨）
raise NotImplementedError("voyage-ai などの専用エンベディングAPIを使ってください")
def ingest_document(
file_path: str,
department: str, # "hr", "engineering", "finance", "all"
doc_title: str,
db_conn: psycopg2.extensions.connection
) -> int:
"""
ドキュメントをインジェストし、チャンク数を返す。
"""
path = Path(file_path)
# テキスト抽出
if path.suffix == ".pdf":
text = extract_text_from_pdf(file_path)
elif path.suffix in [".md", ".txt"]:
text = path.read_text(encoding="utf-8")
else:
raise ValueError(f"未対応のファイル形式: {path.suffix}")
# ドキュメントの重複チェック（ハッシュで判定）
doc_hash = hashlib.sha256(text.encode()).hexdigest()
with db_conn.cursor() as cur:
cur.execute(
"SELECT id FROM documents WHERE content_hash = %s",
(doc_hash,)
)
if cur.fetchone():
print(f"スキップ: {doc_title} は既にインジェスト済みです")
return 0
# チャンキング
chunks = chunk_document(text)
# DB への挿入（ドキュメント本体）
with db_conn.cursor() as cur:
cur.execute(
"""
INSERT INTO documents (title, department, content_hash, file_path)
VALUES (%s, %s, %s, %s)
RETURNING id
""",
(doc_title, department, doc_hash, file_path)
)
doc_id = cur.fetchone()[0]
# チャンクの挿入（エンベディングは別途生成）
chunk_data = []
for i, chunk in enumerate(chunks):
# 実際にはここでエンベディングを生成してからDBに挿入
# embedding = generate_embedding(chunk["text"])
chunk_data.append((
doc_id,
i,
chunk["text"],
department,
# embedding # pgvector カラムへ
))
with db_conn.cursor() as cur:
execute_values(
cur,
"""
INSERT INTO document_chunks (doc_id, chunk_index, content, department)
VALUES %s
""",
chunk_data
)
db_conn.commit()
print(f"✅ インジェスト完了: {doc_title} ({len(chunks)} チャンク, 部署: {department})")
return len(chunks)チャンクサイズは 512 トークン・重複 64 トークンに設定しています。これより小さくすると文脈が失われ、大きくすると Claude に渡せるチャンク数が減ります。プロジェクトによって調整が必要ですが、私が試した中では 512/64 が技術ドキュメントへの適合性が高い組み合わせでした。
ハイブリッド検索（pgvector + BM25）の実装
ベクター検索だけでは「製品コード MA-2024-001 に関する仕様」のような完全一致クエリに弱いです。BM25（キーワード検索）と組み合わせることで、この問題が解消されます。
from typing import Optional
def hybrid_search(
query: str,
user_department: str,
query_embedding: list[float],
db_conn: psycopg2.extensions.connection,
top_k: int = 10,
vector_weight: float = 0.6,
bm25_weight: float = 0.4
) -> list[dict]:
"""
ハイブリッド検索（RRF: Reciprocal Rank Fusion）。
- pgvector で上位 k 件取得
- PostgreSQL 全文検索（tsvector）で上位 k 件取得
- RRF スコアで統合・再ランク
- RLS により user_department でフィルタリング（DB 層で保証）
重要: この関数は DB RLS のロールに user_department が設定された
接続で呼び出すこと。アプリ層でのフィルタリングは行わない。
"""
rrf_k = 60 # RRF のパラメータ（一般的に 60 が推奨値）
with db_conn.cursor() as cur:
# ① ベクター検索（RLS が自動で department フィルタを適用）
cur.execute(
"""
SELECT id, content, doc_id, chunk_index,
1 - (embedding <=> %s::vector) AS score
FROM document_chunks
ORDER BY embedding <=> %s::vector
LIMIT %s
""",
(query_embedding, query_embedding, top_k * 2)
)
vector_results = {row[0]: (i, row) for i, row in enumerate(cur.fetchall())}
# ② BM25 全文検索
cur.execute(
"""
SELECT id, content, doc_id, chunk_index,
ts_rank(to_tsvector('japanese', content),
plainto_tsquery('japanese', %s)) AS score
FROM document_chunks
WHERE to_tsvector('japanese', content) @@ plainto_tsquery('japanese', %s)
ORDER BY score DESC
LIMIT %s
""",
(query, query, top_k * 2)
)
bm25_results = {row[0]: (i, row) for i, row in enumerate(cur.fetchall())}
# ③ RRF スコアで統合
all_ids = set(vector_results.keys()) | set(bm25_results.keys())
scored_results = []
for chunk_id in all_ids:
rrf_score = 0.0
row_data = None
if chunk_id in vector_results:
rank, row = vector_results[chunk_id]
rrf_score += vector_weight / (rrf_k + rank + 1)
row_data = row
if chunk_id in bm25_results:
rank, row = bm25_results[chunk_id]
rrf_score += bm25_weight / (rrf_k + rank + 1)
if row_data is None:
row_data = row
scored_results.append({
"id": chunk_id,
"content": row_data[1],
"doc_id": row_data[2],
"chunk_index": row_data[3],
"rrf_score": rrf_score
})
# スコア降順でソートして上位 k 件を返す
scored_results.sort(key=lambda x: x["rrf_score"], reverse=True)
return scored_results[:top_k]RRF（Reciprocal Rank Fusion）の rrf_k = 60 は経験則的な値で、検索論文でよく使われるパラメータです。ベクター検索と BM25 の重みは 0.6/0.4 にしていますが、日本語技術ドキュメントではキーワード検索の比重をもう少し上げてもいいかもしれません。
部署別アクセス制御（RBAC）の実装
これが設計の核心部分です。PostgreSQL の Row Level Security（RLS）を使います。
-- テーブル定義
CREATE TABLE document_chunks (
id BIGSERIAL PRIMARY KEY,
doc_id BIGINT NOT NULL REFERENCES documents(id),
chunk_index INT NOT NULL,
content TEXT NOT NULL,
department TEXT NOT NULL, -- "hr", "engineering", "finance", "all"
embedding vector(1024) -- voyage-3-large の次元数
);
-- RLS の有効化
ALTER TABLE document_chunks ENABLE ROW LEVEL SECURITY;
ALTER TABLE document_chunks FORCE ROW LEVEL SECURITY;
-- ポリシー定義
-- ① 自分の部署のドキュメントは見られる
CREATE POLICY dept_access ON document_chunks
USING (
department = current_setting('app.user_department')
OR department = 'all'
);
-- ② 管理者はすべて見られる
CREATE POLICY admin_access ON document_chunks
TO admin_role
USING (true);
-- アプリ用ロール（一般ユーザー）
CREATE ROLE app_user;
GRANT SELECT ON document_chunks TO app_user;import psycopg2
import os
from contextlib import contextmanager
@contextmanager
def get_db_connection_for_user(user_department: str):
"""
ユーザーの部署に対応した DB 接続を返すコンテキストマネージャ。
RLS が有効なため、この接続では user_department 以外のデータは見えない。
"""
conn = psycopg2.connect(
dsn=os.environ["DATABASE_URL"],
options=f"-c role=app_user" # アプリ用ロールに切り替え
)
try:
with conn.cursor() as cur:
# セッション変数にユーザーの部署をセット（RLS が参照する）
cur.execute(
"SELECT set_config('app.user_department', %s, true)",
(user_department,)
)
yield conn
finally:
conn.close()
# 使用例
async def search_documents(query: str, user_info: dict) -> list[dict]:
"""
ユーザーの部署に基づいてアクセス制御された検索を実行する。
"""
with get_db_connection_for_user(user_info["department"]) as conn:
# この conn 経由のクエリは RLS によって自動フィルタリングされる
# アプリ層での department フィルタは不要（というより追加しても意味がない）
results = hybrid_search(
query=query,
user_department=user_info["department"],
query_embedding=await generate_embedding_async(query),
db_conn=conn,
top_k=10
)
return resultsRLS の落とし穴: set_config の第3引数を true（トランザクション内でのみ有効）ではなく false（セッション全体で有効）にすると、接続プールを使っているときに設定が他のユーザーに漏れる危険があります。必ず true を使ってください。
Claude API を使った QA エージェントの実装
フィルタリング済みのチャンクを Claude に渡して回答を生成します。
import anthropic
import json
from datetime import datetime, timezone
client = anthropic.Anthropic()
SYSTEM_PROMPT = """あなたは社内ドキュメントの検索・回答エージェントです。
以下のルールを厳守してください:
1. 提供されたドキュメントチャンクのみに基づいて回答する
2. チャンクに記載のない情報は「この情報は提供されたドキュメントには含まれていません」と明示する
3. 回答の根拠となるドキュメントを必ず引用元として示す
4. 機密情報を含む可能性がある場合は、回答の末尾に注記を入れる
回答形式:
- 回答本文
- 引用元: [ドキュメント名, チャンクインデックス]
"""
def format_chunks_for_context(chunks: list[dict]) -> str:
"""チャンクを Claude に渡すコンテキストとしてフォーマットする"""
formatted = []
for i, chunk in enumerate(chunks):
formatted.append(
f"[チャンク {i+1}] (ドキュメントID: {chunk['doc_id']}, "
f"位置: {chunk['chunk_index']})\n{chunk['content']}"
)
return "\n\n---\n\n".join(formatted)
def answer_query(
query: str,
chunks: list[dict],
user_info: dict
) -> dict:
"""
検索されたチャンクを基に Claude で回答を生成する。
Returns:
{
"answer": str,
"cited_chunks": list[int],
"model": str,
"input_tokens": int,
"output_tokens": int
}
"""
if not chunks:
return {
"answer": "お探しのドキュメントが見つかりませんでした。"
"検索キーワードを変えるか、アクセス権限をご確認ください。",
"cited_chunks": [],
"model": "none",
"input_tokens": 0,
"output_tokens": 0
}
context = format_chunks_for_context(chunks)
# コンテキスト長の確認（claude-sonnet-4-6 は 200K トークン）
# チャンクが多すぎる場合は上位 5 件に絞る
if len(chunks) > 8:
chunks = chunks[:8]
context = format_chunks_for_context(chunks)
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=2048,
system=SYSTEM_PROMPT,
messages=[
{
"role": "user",
"content": f"以下のドキュメントチャンクを参照して、質問に答えてください。\n\n"
f"## ドキュメントチャンク\n\n{context}\n\n"
f"## 質問\n\n{query}"
}
]
)
return {
"answer": response.content[0].text,
"cited_chunks": [c["id"] for c in chunks],
"model": response.model,
"input_tokens": response.usage.input_tokens,
"output_tokens": response.usage.output_tokens
}監査ログとコンプライアンス対応
SOC2 や社内コンプライアンスのために、すべての検索と回答を記録します。
import uuid
from datetime import datetime, timezone
def log_search_event(
db_conn: psycopg2.extensions.connection,
user_id: str,
user_department: str,
query: str,
cited_chunk_ids: list[int],
answer_summary: str, # 回答の最初の200文字（全文保存はコスト増）
model: str,
input_tokens: int,
output_tokens: int,
session_id: Optional[str] = None
) -> str:
"""
監査ログに検索イベントを記録する。
返り値: イベントID（UUID）
"""
event_id = str(uuid.uuid4())
with db_conn.cursor() as cur:
cur.execute(
"""
INSERT INTO search_audit_log (
event_id,
user_id,
user_department,
query_text,
cited_chunk_ids,
answer_summary,
model_used,
input_tokens,
output_tokens,
session_id,
created_at
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
""",
(
event_id,
user_id,
user_department,
query,
cited_chunk_ids, # PostgreSQL の integer[] 型
answer_summary[:200],
model,
input_tokens,
output_tokens,
session_id,
datetime.now(timezone.utc)
)
)
db_conn.commit()
return event_id
# 月次コストレポートのクエリ例
MONTHLY_COST_QUERY = """
SELECT
DATE_TRUNC('day', created_at) AS day,
user_department,
COUNT(*) AS query_count,
SUM(input_tokens) AS total_input_tokens,
SUM(output_tokens) AS total_output_tokens,
-- claude-sonnet-4-6 の価格: input $3/M, output $15/M（2026-05時点の近似）
ROUND(
(SUM(input_tokens) * 3.0 + SUM(output_tokens) * 15.0) / 1000000,
4
) AS estimated_cost_usd
FROM search_audit_log
WHERE created_at >= DATE_TRUNC('month', CURRENT_DATE)
GROUP BY 1, 2
ORDER BY 1, 2;
"""監査テーブルは別スキーマ（audit）に置き、アプリユーザーは INSERT のみ可能にする設計を推奨します。このテーブルへの SELECT は監査担当者ロールのみに限定することで、ログの改ざんリスクを最小化できます。
よくある失敗と落とし穴
落とし穴1: チャンクサイズが大きすぎる
最初は「コンテキストが多い方がいいはず」と思って 2048 トークンのチャンクにしていました。その結果、Claude に渡せるチャンク数が 2〜3 件に絞られ、関連情報を取りこぼすケースが頻発しました。
512 トークンに変えたところ、同じコンテキスト予算で 8〜10 件のチャンクを渡せるようになり、回答の精度が明らかに上がりました。チャンクは小さめにして数を増やす方向が私の経験上は良好です。
# ❌ これは避けてください
chunk_size = 2048 # 大きすぎる → チャンク数が少なくなる
# ✅ 推奨
chunk_size = 512 # Claude への入力チャンク数を 8〜10 件確保できる
overlap = 64 # 文脈の継続性を保つ重複落とし穴2: RLS を FORCE ROW LEVEL SECURITY にしていない
デフォルトでは、テーブルのオーナーロールは RLS をバイパスします。FORCE ROW LEVEL SECURITY を設定していないと、開発中に管理者接続でテストすると「全データが見えて正常」と誤認し、本番で一般ユーザーとして接続したときに初めてバグが発覚します。
-- ❌ これだとオーナーロールはバイパスできてしまう
ALTER TABLE document_chunks ENABLE ROW LEVEL SECURITY;
-- ✅ 全ロールに RLS を強制する
ALTER TABLE document_chunks ENABLE ROW LEVEL SECURITY;
ALTER TABLE document_chunks FORCE ROW LEVEL SECURITY;落とし穴3: エンベディングの更新を忘れる
ドキュメントを更新したとき、テキストだけ新しくしてエンベディングを再生成しないケースが発生しました。テキストとエンベディングがずれているチャンクは、ベクター検索で意味不明な結果を返します。
def update_document(doc_id: int, new_text: str, db_conn):
"""ドキュメント更新時は必ずエンベディングも再生成する"""
new_hash = hashlib.sha256(new_text.encode()).hexdigest()
chunks = chunk_document(new_text)
with db_conn.cursor() as cur:
# 古いチャンクを削除
cur.execute("DELETE FROM document_chunks WHERE doc_id = %s", (doc_id,))
# 新しいチャンクとエンベディングを挿入
for i, chunk in enumerate(chunks):
# embedding = generate_embedding(chunk["text"]) # 必ず再生成
cur.execute(
"""
INSERT INTO document_chunks (doc_id, chunk_index, content, department)
VALUES (%s, %s, %s, (SELECT department FROM documents WHERE id = %s))
""",
(doc_id, i, chunk["text"], doc_id)
)
# ドキュメントのハッシュも更新
cur.execute(
"UPDATE documents SET content_hash = %s, updated_at = NOW() WHERE id = %s",
(new_hash, doc_id)
)
db_conn.commit()落とし穴4: 日本語テキストの全文検索設定
PostgreSQL の to_tsvector('japanese', ...) は pg_bigm か pgroonga 拡張が必要です。デフォルトの simple や english では日本語の形態素解析が行われず、BM25 スコアがほぼゼロになります。
-- pgroonga 拡張をインストール（Supabase なら Extensions から有効化）
CREATE EXTENSION IF NOT EXISTS pgroonga;
-- pgroonga インデックスを使った日本語全文検索
CREATE INDEX idx_chunks_content_pgroonga
ON document_chunks
USING pgroonga (content pgroonga_text_full_text_search_ops_v2);この先の発展方向
今回実装したシステムで、私のチームでは次の課題が見えてきました。参考までに共有します。
チャンク間の関係性: 現状は独立したチャンクとして扱っていますが、「このチャンクはどのセクションに属するか」というメタデータを持たせると、回答の文脈がより豊かになります。階層的なチャンキング（大きなチャンクで意味的なまとまりを保ち、小さなチャンクで詳細を検索する）は試す価値があります。
再ランキング: BM25 + ベクターの RRF だけでは不十分なケースもあります。Cohere Rerank や cross-encoder モデルで最終的な並び替えを入れると、特にドキュメントが多い環境での精度が向上します。
キャッシュ: 同じクエリに対するエンベディング生成と検索結果をキャッシュするだけで、API コストと応答時間が大幅に改善されました。Redis に 1 時間 TTL でキャッシュするのが手軽でおすすめです。
まずは小規模な社内ドキュメント（10〜100ファイル）でシステム全体を動かしてみてください。RLS のポリシーが意図通りに機能しているか、部署をまたいだアクセスが本当にブロックされているかをテストから始めるのが、本番投入への最短ルートです。
RLS の仕組みを深く理解する上でも、RDBMSの根本的な設計思想を把握していると応用が利きます。