LLM RAG + AI Agentエンタープライズ実装実践：検索拡張生成アーキテクチャと本番デプロイ完全ガイド
人工智能
概要
- RAGシステムのコアアーキテクチャとドキュメント処理パイプラインを習得し、生ドキュメントから高品質な検索結果までの完全なチェーンを理解する
- ハイブリッド検索（ベクトル + キーワード + ナレッジグラフ）とリランキング技術を深く掘り下げ、エンタープライズ級95%以上の検索精度を実現する
- RAG + AI Agentの深度融合実践：ツール拡張検索、マルチターン会話メモリ、エンタープライズナレッジベースのアクセス制御と本番デプロイ
目次
- 一、RAGシステムアーキテクチャとコアフロー
- 二、ドキュメント処理とチャンキング戦略
- 三、Embeddingモデル選定と最適化
- 四、ハイブリッド検索とリランキング
- 五、RAG + AI Agent深度融合
- 六、エンタープライズナレッジベースの権限とセキュリティ
- 七、本番級デプロイとパフォーマンス最適化
- 八、まとめと展望
一、RAGシステムアーキテクチャとコアフロー
1.1 RAGのコアバリューと限界
検索拡張生成（Retrieval-Augmented Generation, RAG）は、2024〜2026年における大規模モデル応用で最も広く採用されている技術パラダイムである。そのコアバリューは、モデルのファインチューニングなしに、外部ナレッジベースの検索により大規模モデルの回答品質と事実正確性を向上させる点にある。
然而，生产级RAG系统面临的核心挑战远不止"检索+拼接"那么简单：
- 検索品質のボトルネック：ベクトル検索の意味的類似性は回答の関連性とイコールではなく、Top-K結果に大量のノイズが含まれる可能性がある
- コンテキストウィンドウの浪費：無関係な検索結果が限られたコンテキストウィンドウを占有し、モデルの推論品質を低下させる
- マルチホップ推論の欠如：複雑な問題には複数ステップの検索と推論が必要であり、単一の検索では不十分である
- リアルタイム性の要件：エンタープライズナレッジベースは頻繁に更新され、インデックスのリアルタイム同期が必要である
1.2 本番級RAGアーキテクチャ
一个生产级RAG系统的架构远比简单的"查询→检索→生成"复杂：
┌──────────────────────────────────────────────────┐
│ Query Understanding │
│ 意図識別 · クエリ書き換え · エンティティ抽出 · マルチホップ分解 │
├──────────────────────────────────────────────────┤
│ Hybrid Retrieval │
│ ベクトル検索 · BM25キーワード · ナレッジグラフ · SQLクエリ │
├──────────────────────────────────────────────────┤
│ Reranking & Fusion │
│ クロスエンコーダリランキング · 相互情報量最大化 · 結果重複排除 │
├──────────────────────────────────────────────────┤
│ Context Assembly │
│ 関連性フィルタリング · コンテキスト圧縮 · 構造化組織 │
├──────────────────────────────────────────────────┤
│ Generation & Verification │
│ Chain-of-Thought · 事実検証 · ハルシネーション検出 │
└──────────────────────────────────────────────────┘
Query Understanding層は、クエリ書き換え、エンティティ抽出、マルチホップ問題分解など、ユーザーの真の意図を理解する責務を担う。Hybrid Retrieval層は複数の検索戦略の融合を使用する。Reranking層は初期検索結果の精密な並べ替えを行う。Context Assembly層は最適なコンテキストを組み立てる。Generation層は最終的な回答を生成し、事実検証を行う。
1.3 RAG vs ファインチューニング vs 事前学習
| 次元 | RAG | ファインチューニング | 事前学習 |
|---|---|---|---|
| 知識更新 | リアルタイム | 再学習が必要 | 再学習が必要 |
| コスト | 低 | 中 | 非常に高い |
| 事実正確性 | 高 | 中 | 低 |
| ドメイン適応性 | 強 | 強 | 最強 |
| デプロイ複雑度 | 中 | 低 | 高 |
| ハルシネーション制御 | 良好 | 普通 | 不良 |
エンタープライズシナリオでは、RAG + ファインチューニングの組み合わせがベストプラクティスである：RAGが事実正確性を保証し、ファインチューニングが特定ドメインに対するモデルの理解と出力スタイルを最適化する。
二、ドキュメント処理とチャンキング戦略
2.1 ドキュメント解析パイプライン
エンタープライズナレッジベースのドキュメント形式は多様（PDF、Word、Excel、PPT、HTML、Markdown）であり、統一的な解析パイプラインが必要である：
from dataclasses import dataclass
from pathlib import Path
from typing import Optional
import hashlib
@dataclass
class Parsedドキュメント:
doc_id: str
title: str
content: str
metadata: dict
source_path: str
checksum: str
page_count: int
language: str
class DocumentParser:
def __init__(self, ocr_enabled: bool = True, table_enabled: bool = True):
self.ocr_enabled = ocr_enabled
self.table_enabled = table_enabled
def parse(self, file_path: str) -> Parsedドキュメント:
path = Path(file_path)
suffix = path.suffix.lower()
match suffix:
case ".pdf":
return self._parse_pdf(file_path)
case ".docx" | ".doc":
return self._parse_docx(file_path)
case ".xlsx" | ".xls":
return self._parse_excel(file_path)
case ".pptx":
return self._parse_pptx(file_path)
case ".md":
return self._parse_markdown(file_path)
case ".html" | ".htm":
return self._parse_html(file_path)
case _:
return self._parse_plain_text(file_path)
def _parse_pdf(self, file_path: str) -> Parsedドキュメント:
import fitz
doc = fitz.open(file_path)
content_parts = []
page_count = len(doc)
for page_num in range(page_count):
page = doc[page_num]
text = page.get_text()
if text.strip():
content_parts.append(text)
if self.table_enabled:
tables = page.find_tables()
for table in tables:
table_text = table.to_pandas().to_markdown()
content_parts.append(f"\n[テーブル on page {page_num + 1}]\n{table_text}")
if self.ocr_enabled:
images = page.get_images()
for img_idx, img in enumerate(images):
xref = img[0]
base_image = doc.extract_image(xref)
if base_image:
image_bytes = base_image["image"]
ocr_text = self._ocr_image(image_bytes)
if ocr_text:
content_parts.append(
f"\n[画像 {img_idx + 1} on page {page_num + 1}]\n{ocr_text}"
)
content = "\n\n".join(content_parts)
checksum = hashlib.sha256(content.encode()).hexdigest()[:16]
return ParsedDocument(
doc_id=f"doc_{checksum}",
title=Path(file_path).stem,
content=content,
metadata={"format": "pdf", "page_count": page_count},
source_path=file_path,
checksum=checksum,
page_count=page_count,
language=self._detect_language(content),
)
def _ocr_image(self, image_bytes: bytes) -> Optional[str]:
try:
import pytesseract
from PIL import Image
import io
image = Image.open(io.BytesIO(image_bytes))
return pytesseract.image_to_string(image, lang='chi_sim+eng')
except Exception:
return None
def _detect_language(self, text: str) -> str:
sample = text[:500]
chinese_chars = sum(1 for c in sample if '\u4e00' <= c <= '\u9fff')
if chinese_chars / max(len(sample), 1) > 0.3:
return "ja"
return "en"
2.2 チャンキング戦略の深い比較
ドキュメントチャンキングはRAGシステムにおいて最も重要な要素の一つであり、チャンキング戦略が検索品質に直接影響する：
固定サイズチャンキング（Fixed-Size Chunking）：固定トークン数で分割する。実装はシンプルだが、意味的整合性を損なう可能性がある。
再帰的文字チャンキング（Recursive Character Chunking）：段落→文→文字の優先順位で再帰的に分割し、意味的整合性を維持する。LangChainのRecursiveCharacterTextSplitterがこの戦略を採用している。
意味的チャンキング（Semantic Chunking）：Embeddingモデルを使用して隣接する文の意味的類似度を計算し、意味的ブレークポイントで分割する。品質は最も高いが計算コストが大きい。
構造認識チャンキング（Structure-Aware Chunking）：ドキュメントの見出し階層とセクション構造を利用してチャンキングし、ドキュメントの論理構造を維持する。
from dataclasses import dataclass
from typing import Callable
import numpy as np
@dataclass
class Chunk:
chunk_id: str
content: str
metadata: dict
start_index: int
end_index: int
token_count: int
parent_doc_id: str
class SemanticChunker:
def __init__(
self,
embedding_fn: Callable[[str], list[float]],
similarity_threshold: float = 0.5,
min_chunk_size: int = 100,
max_chunk_size: int = 1000,
):
self.embedding_fn = embedding_fn
self.similarity_threshold = similarity_threshold
self.min_chunk_size = min_chunk_size
self.max_chunk_size = max_chunk_size
def chunk(self, text: str, doc_id: str) -> list[Chunk]:
sentences = self._split_sentences(text)
if len(sentences) <= 1:
return [Chunk(
chunk_id=f"{doc_id}_0",
content=text,
metadata={"chunk_type": "semantic"},
start_index=0,
end_index=len(text),
token_count=len(text) // 4,
parent_doc_id=doc_id,
)]
embeddings = [self.embedding_fn(s) for s in sentences]
similarities = [
self._cosine_similarity(embeddings[i], embeddings[i + 1])
for i in range(len(embeddings) - 1)
]
breakpoints = []
for i, sim in enumerate(similarities):
if sim < self.similarity_threshold:
breakpoints.append(i + 1)
chunks = []
current_start = 0
chunk_idx = 0
for bp in breakpoints + [len(sentences)]:
chunk_text = " ".join(sentences[current_start:bp])
token_count = len(chunk_text) // 4
if token_count >= self.min_chunk_size:
chunks.append(Chunk(
chunk_id=f"{doc_id}_{chunk_idx}",
content=chunk_text,
metadata={
"chunk_type": "semantic",
"sentence_count": bp - current_start,
},
start_index=current_start,
end_index=bp,
token_count=token_count,
parent_doc_id=doc_id,
))
chunk_idx += 1
elif chunks:
chunks[-1].content += " " + chunk_text
chunks[-1].end_index = bp
chunks[-1].token_count += token_count
current_start = bp
for chunk in chunks:
if chunk.token_count > self.max_chunk_size:
self._split_oversized(chunk, doc_id, chunks)
return chunks
def _split_sentences(self, text: str) -> list[str]:
import re
sentences = re.split(r'(?<=[。！？.!?])\s*', text)
return [s.strip() for s in sentences if s.strip()]
def _cosine_similarity(self, a: list[float], b: list[float]) -> float:
a_arr = np.array(a)
b_arr = np.array(b)
return float(np.dot(a_arr, b_arr) / (np.linalg.norm(a_arr) * np.linalg.norm(b_arr) + 1e-8))
def _split_oversized(self, chunk: Chunk, doc_id: str, chunks: list[Chunk]) -> None:
pass
2.3 メタデータ拡張チャンキング
各チャンクに豊富なメタデータを追加し、検索時の精密なフィルタリングをサポートする：
@dataclass
class EnhancedChunk(Chunk):
heading_path: list[str]
keywords: list[str]
entities: list[dict]
summary: str
access_level: int
department: str
doc_type: str
created_at: str
updated_at: str
class MetadataEnricher:
def __init__(self, llm_client):
self.llm = llm_client
async def enrich(self, chunk: Chunk, doc: ParsedDocument) -> EnhancedChunk:
keywords = await self._extract_keywords(chunk.content)
entities = await self._extract_entities(chunk.content)
summary = await self._generate_summary(chunk.content)
return EnhancedChunk(
**chunk.__dict__,
heading_path=self._extract_heading_path(doc, chunk),
keywords=keywords,
entities=entities,
summary=summary,
access_level=doc.metadata.get("access_level", 0),
department=doc.metadata.get("department", ""),
doc_type=doc.metadata.get("format", ""),
created_at=doc.metadata.get("created_at", ""),
updated_at=doc.metadata.get("updated_at", ""),
)
async def _extract_keywords(self, text: str) -> list[str]:
prompt = f"以下のテキストから5〜10個のキーワードを抽出し、JSON配列形式で返してください：\n\n{text[:1000]}"
response = await self.llm.generate(prompt)
import json
try:
return json.loads(response)
except:
return []
async def _extract_entities(self, text: str) -> list[dict]:
prompt = f"""以下のテキストから固有表現を抽出し、JSON配列で返してください。各エンティティはname、type、valueフィールドを含むこと：
エンティティタイプ：PERSON, ORGANIZATION, PRODUCT, DATE, LOCATION, TECHNOLOGY
文本：
{text[:2000]}"""
response = await self.llm.generate(prompt)
import json
try:
return json.loads(response)
except:
return []
async def _generate_summary(self, text: str) -> str:
prompt = f"以下のテキストの核心内容を一文で要約してください：\n\n{text[:500]}"
return await self.llm.generate(prompt)
def _extract_heading_path(self, doc: ParsedDocument, chunk: Chunk) -> list[str]:
return doc.metadata.get("heading_path", [])
三、Embeddingモデル選定と最適化
3.1 主要Embeddingモデル比較
| モデル | 次元 | MTEBスコア | 日本語/中国語能力 | 推論速度 | ライセンス |
|---|---|---|---|---|---|
| BGE-M3 | 1024 | 73.5 | 優秀 | 中 | MIT |
| GTE-Qwen2-7B | 3584 | 76.2 | 優秀 | 遅 | Apache 2.0 |
| text-embedding-3-large | 3072 | 74.5 | 良好 | 高速 | 商用 |
| Jina-Embeddings-v3 | 1024 | 72.8 | 良好 | 中 | CC-BY-4.0 |
| BCE-Embedding | 768 | 71.2 | 優秀 | 高速 | MIT |
選定の推奨：
- 日本語/中国語シナリオではBGE-M3が最適、コストパフォーマンス最強
- 最高性能を追求するならGTE-Qwen2-7Bだが、推論コストが高い
- 多言語サポートが必要な場合はJina-Embeddings-v3
- OpenAI APIを使用するシナリオではtext-embedding-3-large
3.2 Embeddingサービスデプロイ
from fastapi import FastAPI
from pydantic import BaseModel
import numpy as np
from sentence_transformers import SentenceTransformer
app = FastAPI()
class EmbedRequest(BaseModel):
texts: list[str]
normalize: bool = True
class EmbedResponse(BaseModel):
embeddings: list[list[float]]
model: str
dimension: int
model = SentenceTransformer("BAAI/bge-m3")
@app.post("/embed", response_model=EmbedResponse)
async def embed(request: EmbedRequest):
embeddings = model.encode(
request.texts,
normalize_embeddings=request.normalize,
show_progress_bar=False,
)
return EmbedResponse(
embeddings=embeddings.tolist(),
model="bge-m3",
dimension=embeddings.shape[1],
)
@app.post("/embed/batch", response_model=EmbedResponse)
async def embed_batch(request: EmbedRequest):
batch_size = 64
all_embeddings = []
for i in range(0, len(request.texts), batch_size):
batch = request.texts[i:i + batch_size]
batch_embeddings = model.encode(
batch,
normalize_embeddings=request.normalize,
batch_size=len(batch),
)
all_embeddings.append(batch_embeddings)
embeddings = np.vstack(all_embeddings)
return EmbedResponse(
embeddings=embeddings.tolist(),
model="bge-m3",
dimension=embeddings.shape[1],
)
3.3 クエリ側Embedding最適化
クエリ側のEmbedding最適化は、検索品質を向上させるための重要なテクニックである：
クエリ拡張（Query Expansion）：LLMを使用してユーザーのクエリを複数の関連クエリに拡張し、検索カバレッジを増やす。
仮説的ドキュメント埋め込み（HyDE）：まずLLMに仮説的回答を生成させ、その仮説回答のEmbeddingで検索する。直接クエリで検索するよりも効果的である。
命令プレフィックス（Instruction Prefix）：クエリの前にタスク命令を追加する。例：「以下のクエリに関連するドキュメントを検索：」とし、訓練時の入力形式に合わせる。
class QueryOptimizer:
def __init__(self, llm_client, embedding_fn):
self.llm = llm_client
self.embedding_fn = embedding_fn
async def expand_query(self, query: str, num_expansions: int = 3) -> list[str]:
prompt = f"""以下のクエリを{num_expansions}つの異なる角度の等価なクエリに書き換え、JSON配列形式で返してください。
元のクエリ：{query}
要件：
1. 元のクエリのコア意図を維持する
2. 異なる表現とキーワードを使用する
3. 異なる専門用語と一般的な表現をカバーする"""
response = await self.llm.generate(prompt)
import json
try:
expansions = json.loads(response)
return [query] + expansions[:num_expansions]
except:
return [query]
async def hyde_embed(self, query: str) -> list[float]:
prompt = f"""以下の質問に対して詳細な回答を書いてください（答えが確かでない場合でも、合理的な仮説的回答を提示してください）：
質問：{query}"""
hypothetical_answer = await self.llm.generate(prompt)
return self.embedding_fn(hypothetical_answer)
def instruction_embed(self, query: str, task: str = "search") -> list[float]:
prefixes = {
"search": "以下のクエリに関連するドキュメントを検索：",
"similarity": "以下の内容に類似するドキュメントを検索：",
"classification": "以下の内容を分類：",
}
prefix = prefixes.get(task, "")
return self.embedding_fn(f"{prefix}{query}")
四、ハイブリッド検索とリランキング
4.1 ハイブリッド検索アーキテクチャ
単一のベクトル検索ではすべてのシナリオをカバーできない。キーワード検索（BM25）は精密マッチング（製品型番、固有名詞）に優れ、ベクトル検索は意味的マッチング（概念の類似性）に優れ、ナレッジグラフ検索は関係推論に優れている。ハイブリッド検索は本番級RAGの必須要件である。
from dataclasses import dataclass
from typing import Optional
@dataclass
class RetrievalResult:
chunk_id: str
content: str
score: float
source: str
metadata: dict
class HybridRetriever:
def __init__(
self,
vector_store,
bm25_store,
kg_store=None,
vector_weight: float = 0.6,
bm25_weight: float = 0.3,
kg_weight: float = 0.1,
):
self.vector_store = vector_store
self.bm25_store = bm25_store
self.kg_store = kg_store
self.vector_weight = vector_weight
self.bm25_weight = bm25_weight
self.kg_weight = kg_weight
async def retrieve(
self,
query: str,
query_embedding: list[float],
top_k: int = 20,
filters: Optional[dict] = None,
) -> list[RetrievalResult]:
vector_results = await self.vector_store.search(
query_embedding, top_k=top_k * 2, filters=filters
)
bm25_results = await self.bm25_store.search(
query, top_k=top_k * 2, filters=filters
)
kg_results = []
if self.kg_store:
kg_results = await self.kg_store.search(
query, top_k=top_k
)
merged = self._reciprocal_rank_fusion(
vector_results, bm25_results, kg_results
)
return merged[:top_k]
def _reciprocal_rank_fusion(
self,
vector_results: list[RetrievalResult],
bm25_results: list[RetrievalResult],
kg_results: list[RetrievalResult],
k: int = 60,
) -> list[RetrievalResult]:
scores: dict[str, float] = {}
result_map: dict[str, RetrievalResult] = {}
for rank, result in enumerate(vector_results):
rrf_score = self.vector_weight / (k + rank + 1)
scores[result.chunk_id] = scores.get(result.chunk_id, 0.0) + rrf_score
result_map[result.chunk_id] = result
for rank, result in enumerate(bm25_results):
rrf_score = self.bm25_weight / (k + rank + 1)
scores[result.chunk_id] = scores.get(result.chunk_id, 0.0) + rrf_score
if result.chunk_id not in result_map:
result_map[result.chunk_id] = result
for rank, result in enumerate(kg_results):
rrf_score = self.kg_weight / (k + rank + 1)
scores[result.chunk_id] = scores.get(result.chunk_id, 0.0) + rrf_score
if result.chunk_id not in result_map:
result_map[result.chunk_id] = result
sorted_ids = sorted(scores.keys(), key=lambda x: scores[x], reverse=True)
return [
RetrievalResult(
chunk_id=cid,
content=result_map[cid].content,
score=scores[cid],
source=result_map[cid].source,
metadata=result_map[cid].metadata,
)
for cid in sorted_ids
]
4.2 クロスエンコーダリランキング
初期検索結果はバイエンコーダ（Bi-Encoder）で計算される。高速だが精度に限界がある。クロスエンコーダ（Cross-Encoder）はクエリとドキュメントを一緒にモデルに入力するため、精度は高いが速度が遅く、Top-K結果の精密ランキングに適している：
from sentence_transformers import CrossEncoder
class Reranker:
def __init__(self, model_name: str = "BAAI/bge-reranker-v2-m3"):
self.model = CrossEncoder(model_name, max_length=512)
def rerank(
self,
query: str,
results: list[RetrievalResult],
top_k: int = 10,
) -> list[RetrievalResult]:
pairs = [(query, result.content) for result in results]
scores = self.model.predict(pairs)
scored_results = list(zip(results, scores))
scored_results.sort(key=lambda x: x[1], reverse=True)
return [
RetrievalResult(
chunk_id=result.chunk_id,
content=result.content,
score=float(score),
source=result.source,
metadata={**result.metadata, "rerank_score": float(score)},
)
for result, score in scored_results[:top_k]
]
4.3 関連性フィルタリング
リランキング後も低関連性の結果をフィルタリングし、ノイズがコンテキストに入るのを防ぐ必要がある：
class RelevanceFilter:
def __init__(self, min_score: float = 0.3, max_chunks: int = 8):
self.min_score = min_score
self.max_chunks = max_chunks
def filter(self, results: list[RetrievalResult]) -> list[RetrievalResult]:
filtered = [r for r in results if r.score >= self.min_score]
return filtered[:self.max_chunks]
def adaptive_filter(
self,
results: list[RetrievalResult],
max_context_tokens: int = 4000,
) -> list[RetrievalResult]:
selected = []
total_tokens = 0
for result in results:
if result.score < self.min_score:
continue
chunk_tokens = len(result.content) // 4
if total_tokens + chunk_tokens > max_context_tokens:
break
selected.append(result)
total_tokens += chunk_tokens
return selected
五、RAG + AI Agent深度融合
5.1 ツール拡張検索Agent
RAG検索機能をAI Agentが呼び出し可能なツールとしてカプセル化し、よりインテリジェントな検索戦略を実現する：
from typing import Annotated
class RAGAgentTools:
def __init__(self, retriever: HybridRetriever, reranker: Reranker):
self.retriever = retriever
self.reranker = reranker
def search_knowledge_base(
self,
query: Annotated[str, "ナレッジベース検索クエリ"],
top_k: Annotated[int, "返却する結果数"] = 5,
filters: Annotated[dict | None, "メタデータフィルタ"] = None,
) -> str:
"""エンタープライズナレッジベースから関連ドキュメントを検索する。"""
query_embedding = self.get_embedding(query)
results = self.retriever.retrieve(query, query_embedding, top_k=top_k * 2, filters=filters)
reranked = self.reranker.rerank(query, results, top_k=top_k)
if not reranked:
return "関連ドキュメントが見つかりませんでした。"
formatted = []
for i, result in enumerate(reranked):
formatted.append(
f"[ドキュメント {i + 1}] (スコア: {result.score:.3f})\n"
f"ソース: {result.metadata.get('source', '不明')}\n"
f"内容: {result.content}\n"
)
return "\n---\n".join(formatted)
def search_by_entity(
self,
entity_name: Annotated[str, "検索するエンティティ名"],
entity_type: Annotated[str, "エンティティタイプ：PERSON, ORGANIZATION, PRODUCTなど"] = "",
) -> str:
"""特定のエンティティに言及しているドキュメントを検索する。"""
filters = {"entities": {"name": entity_name}}
if entity_type:
filters["entities"]["type"] = entity_type
return self.search_knowledge_base(entity_name, filters=filters)
def compare_documents(
self,
topic: Annotated[str, "ドキュメント間で比較するトピック"],
doc_ids: Annotated[list[str], "比較するドキュメントID"] = None,
) -> str:
"""複数ドキュメントにわたるトピックの情報を比較する。"""
query_embedding = self.get_embedding(topic)
filters = {"parent_doc_id": {"$in": doc_ids}} if doc_ids else None
results = self.retriever.retrieve(topic, query_embedding, top_k=20, filters=filters)
reranked = self.reranker.rerank(topic, results, top_k=10)
grouped = {}
for result in reranked:
doc_id = result.metadata.get("parent_doc_id", "unknown")
if doc_id not in grouped:
grouped[doc_id] = []
grouped[doc_id].append(result)
output = []
for doc_id, chunks in grouped.items():
output.append(f"ドキュメント: {doc_id}")
for chunk in chunks:
output.append(f" - {chunk.content[:200]}...")
return "\n\n".join(output)
5.2 マルチターン会話RAG
エンタープライズ級RAGシステムはマルチターン会話をサポートし、会話コンテキストと検索履歴を維持する必要がある：
class ConversationalRAG:
def __init__(self, llm_client, retriever, reranker):
self.llm = llm_client
self.retriever = retriever
self.reranker = reranker
async def chat(
self,
query: str,
conversation_history: list[dict],
max_context_tokens: int = 4000,
) -> dict:
rewritten_query = await self._rewrite_query(query, conversation_history)
query_embedding = self.get_embedding(rewritten_query)
results = await self.retriever.retrieve(
rewritten_query, query_embedding, top_k=20
)
reranked = self.reranker.rerank(rewritten_query, results, top_k=10)
context = self._assemble_context(reranked, max_context_tokens)
prompt = self._build_prompt(query, context, conversation_history)
answer = await self.llm.generate(prompt)
return {
"answer": answer,
"sources": [
{
"chunk_id": r.chunk_id,
"content": r.content[:200],
"score": r.score,
"source": r.metadata.get("source", ""),
}
for r in reranked[:5]
],
"rewritten_query": rewritten_query,
}
async def _rewrite_query(self, query: str, history: list[dict]) -> str:
if not history:
return query
history_text = "\n".join([
f"{'User' if h['role'] == 'user' else 'Assistant'}: {h['content']}"
for h in history[-6:]
])
prompt = f"""基于以下对话历史，将用户的最新问题改写为独立的、完整的检索查询。
只返回改写后的查询，不要解释。
对话历史：
{history_text}
最新質問：{query}
書き換えたクエリ："""
return await self.llm.generate(prompt)
def _assemble_context(self, results: list[RetrievalResult], max_tokens: int) -> str:
parts = []
total = 0
for result in results:
tokens = len(result.content) // 4
if total + tokens > max_tokens:
break
parts.append(f"[ソース: {result.metadata.get('source', '不明')}]\n{result.content}")
total += tokens
return "\n\n---\n\n".join(parts)
def _build_prompt(self, query: str, context: str, history: list[dict]) -> str:
return f"""你是一个专业的企业知识库助手。请基于以下检索到的文档内容回答用户的问题。
要件：
1. 只基于提供的文档内容回答，不要编造信息
2. 如果文档中没有相关信息，明确告知用户
3. 引用信息时标注来源文档
4. 使用清晰的结构化格式
检索到的文档：
{context}
用户質問：{query}
回答："""
5.3 マルチホップ推論RAG
複雑な問題には複数ステップの検索と推論が必要であり、単一のRAGでは不十分である。マルチホップRAGはAgentの計画能力を活用し、複雑な問題を複数ステップの検索チェーンに分解する：
class MultiHopRAG:
def __init__(self, llm_client, retriever, reranker, max_hops: int = 3):
self.llm = llm_client
self.retriever = retriever
self.reranker = reranker
self.max_hops = max_hops
async def answer(self, query: str) -> dict:
hop_results = []
current_query = query
all_contexts = []
for hop in range(self.max_hops):
query_embedding = self.get_embedding(current_query)
results = await self.retriever.retrieve(
current_query, query_embedding, top_k=10
)
reranked = self.reranker.rerank(current_query, results, top_k=5)
all_contexts.extend(reranked)
hop_results.append({
"hop": hop + 1,
"query": current_query,
"results_count": len(reranked),
})
next_action = await self._decide_next_hop(query, all_contexts, hop)
if next_action["action"] == "answer":
break
elif next_action["action"] == "search":
current_query = next_action["query"]
context = self._assemble_context(all_contexts)
answer = await self._generate_answer(query, context)
return {
"answer": answer,
"hops": hop_results,
"total_contexts": len(all_contexts),
}
async def _decide_next_hop(self, original_query: str, contexts: list, hop: int) -> dict:
if hop >= self.max_hops - 1:
return {"action": "answer"}
context_summary = "\n".join([
f"- {c.content[:200]}" for c in contexts[-5:]
])
prompt = f"""基于原始问题和已检索到的信息，判断是否需要继续检索。
原始質問：{original_query}
検索済み情報：
{context_summary}
请判断：
1. 如果已有足够信息回答问题，返回 {{"action": "answer"}}
2. さらに情報が必要な場合、 {{"action": "search", "query": "次の検索クエリ"}}
JSON形式で返してください："""
response = await self.llm.generate(prompt)
import json
try:
return json.loads(response)
except:
return {"action": "answer"}
六、エンタープライズナレッジベースの権限とセキュリティ
6.1 ドキュメントレベルのアクセス制御
エンタープライズナレッジベースのドキュメントには通常、厳格なアクセス制御があり、RAG検索は権限ルールに従う必要がある：
class PermissionAwareRetriever:
def __init__(self, base_retriever, permission_service):
self.base_retriever = base_retriever
self.permission_service = permission_service
async def retrieve(
self,
query: str,
query_embedding: list[float],
user_id: str,
top_k: int = 20,
) -> list[RetrievalResult]:
user_permissions = await self.permission_service.get_user_permissions(user_id)
accessible_departments = user_permissions.get("departments", [])
access_level = user_permissions.get("access_level", 0)
filters = {
"$or": [
{"department": {"$in": accessible_departments}},
{"access_level": {"$lte": access_level}},
]
}
results = await self.base_retriever.retrieve(
query, query_embedding, top_k=top_k * 2, filters=filters
)
verified_results = []
for result in results:
if await self._verify_access(result, user_permissions):
verified_results.append(result)
return verified_results[:top_k]
async def _verify_access(self, result: RetrievalResult, permissions: dict) -> bool:
doc_department = result.metadata.get("department", "")
doc_access_level = result.metadata.get("access_level", 99)
if doc_access_level <= permissions.get("access_level", 0):
return True
if doc_department in permissions.get("departments", []):
return True
return False
6.2 データマスキング
検索結果はLLMに入力される前に、機密情報を自動的にマスキングする必要がある：
import re
class DataMasker:
PATTERNS = {
"phone": (r'1[3-9]\d{9}', lambda m: m.group()[:3] + "****" + m.group()[-4:]),
"email": (r'[\w.-]+@[\w.-]+\.\w+', lambda m: m.group()[0] + "***@" + m.group().split("@")[1]),
"id_card": (r'\d{17}[\dXx]', lambda m: m.group()[:6] + "********" + m.group()[-4:]),
"bank_card": (r'\d{16,19}', lambda m: m.group()[:4] + "****" + m.group()[-4:]),
}
def mask(self, text: str, enabled_types: list[str] | None = None) -> str:
types = enabled_types or list(self.PATTERNS.keys())
for type_name in types:
if type_name in self.PATTERNS:
pattern, replacer = self.PATTERNS[type_name]
text = re.sub(pattern, replacer, text)
return text
6.3 監査ログ
すべてのRAGクエリと検索操作は監査ログに記録する必要がある：
from datetime import datetime
class RAGAuditLogger:
def __init__(self, log_store):
self.log_store = log_store
async def log_query(
self,
user_id: str,
query: str,
rewritten_query: str | None,
results_count: int,
answer_preview: str,
latency_ms: int,
):
entry = {
"timestamp": datetime.utcnow().isoformat(),
"user_id": user_id,
"original_query": query,
"rewritten_query": rewritten_query,
"results_count": results_count,
"answer_preview": answer_preview[:200],
"latency_ms": latency_ms,
}
await self.log_store.insert(entry)
async def log_access_denied(self, user_id: str, query: str, denied_doc_ids: list[str]):
entry = {
"timestamp": datetime.utcnow().isoformat(),
"event_type": "access_denied",
"user_id": user_id,
"query": query,
"denied_documents": denied_doc_ids,
}
await self.log_store.insert(entry)
七、本番級デプロイとパフォーマンス最適化
7.1 キャッシュ戦略
RAGシステムのマルチレイヤーキャッシュ戦略：
import hashlib
from functools import lru_cache
class RAGCache:
def __init__(self, redis_client, ttl: int = 3600):
self.redis = redis_client
self.ttl = ttl
def _cache_key(self, query: str, user_id: str, filters: dict | None = None) -> str:
raw = f"{query}:{user_id}:{filters}"
return f"rag:cache:{hashlib.md5(raw.encode()).hexdigest()}"
async def get(self, query: str, user_id: str, filters: dict | None = None) -> dict | None:
key = self._cache_key(query, user_id, filters)
cached = await self.redis.get(key)
if cached:
import json
return json.loads(cached)
return None
async def set(self, query: str, user_id: str, result: dict, filters: dict | None = None):
key = self._cache_key(query, user_id, filters)
import json
await self.redis.setex(key, self.ttl, json.dumps(result, ensure_ascii=False))
async def invalidate_doc(self, doc_id: str):
pattern = f"rag:cache:*"
async for key in self.redis.scan_iter(pattern):
cached = await self.redis.get(key)
if cached and doc_id in cached.decode():
await self.redis.delete(key)
7.2 非同期インデックス更新
エンタープライズナレッジベースのドキュメントは頻繁に更新されるため、非同期インデックス更新の仕組みが必要である：
import asyncio
from typing import Callable
class AsyncIndexUpdater:
def __init__(
self,
embedding_fn: Callable,
vector_store,
chunker,
batch_size: int = 100,
flush_interval: int = 30,
):
self.embedding_fn = embedding_fn
self.vector_store = vector_store
self.chunker = chunker
self.batch_size = batch_size
self.flush_interval = flush_interval
self.pending_updates: list[dict] = []
self._running = False
async def start(self):
self._running = True
asyncio.create_task(self._flush_loop())
async def stop(self):
self._running = False
if self.pending_updates:
await self._flush()
async def add_document(self, doc: ParsedDocument):
self.pending_updates.append({
"action": "add",
"doc_id": doc.doc_id,
"content": doc.content,
"metadata": doc.metadata,
})
if len(self.pending_updates) >= self.batch_size:
await self._flush()
async def delete_document(self, doc_id: str):
self.pending_updates.append({
"action": "delete",
"doc_id": doc_id,
})
async def _flush_loop(self):
while self._running:
await asyncio.sleep(self.flush_interval)
if self.pending_updates:
await self._flush()
async def _flush(self):
updates = self.pending_updates[:]
self.pending_updates.clear()
to_add = [u for u in updates if u["action"] == "add"]
to_delete = [u for u in updates if u["action"] == "delete"]
if to_add:
all_chunks = []
for update in to_add:
chunks = self.chunker.chunk(update["content"], update["doc_id"])
all_chunks.extend(chunks)
texts = [c.content for c in all_chunks]
embeddings = [self.embedding_fn(t) for t in texts]
await self.vector_store.upsert(
ids=[c.chunk_id for c in all_chunks],
embeddings=embeddings,
metadatas=[c.metadata for c in all_chunks],
documents=texts,
)
if to_delete:
doc_ids = [u["doc_id"] for u in to_delete]
await self.vector_store.delete_by_doc_ids(doc_ids)
7.3 パフォーマンス指標とSLA
| 指標 | SLA目標 | 監視方法 |
|---|---|---|
| エンドツーエンドレイテンシP95 | < 3s | Prometheus histogram |
| 検索レイテンシP95 | < 500ms | カスタムメトリクス |
| 検索精度 | > 95% | 手動サンプリング + 自動評価 |
| インデックス更新レイテンシ | < 60s | ドキュメント書き込み→検索可能までの時間差 |
| システム可用性 | > 99.9% | ヘルスチェック + アラート |
| 同時QPS | > 100 | 負荷テスト検証 |
八、まとめと展望
LLM RAG + AI Agentのエンタープライズ実装は、2026年のAIアプリケーションにおける最も核心的な技術方向の一つである。本記事では、RAGアーキテクチャ、ドキュメント処理、Embedding最適化、ハイブリッド検索、Agent統合、権限セキュリティ、本番デプロイの7つの次元から、エンタープライズ級RAGシステムの構築方法を体系的に解説した。
重要ポイントの振り返り：
- ハイブリッド検索：ベクトル + BM25 + ナレッジグラフのRRF融合は、本番級RAG検索の標準構成である
- リランキング：クロスエンコーダによる精密ランキング + 関連性フィルタリングにより、検索精度を70%から95%以上に向上させる
- クエリ最適化：クエリ拡張 + HyDE + 命令プレフィックスにより、検索リコールを大幅に向上させる
- Agent統合：RAG機能をAgentツールとしてカプセル化し、マルチターン会話とマルチホップ推論をサポートする
- セキュリティコンプライアンス：ドキュメントレベルの権限 + データマスキング + 監査ログは、エンタープライズデプロイの必須条件である
今後、RAG技術はよりインテリジェントな方向へ進化するだろう：適応型検索戦略（クエリタイプに基づいて検索方法を自動選択）、継続学習（ユーザーフィードバックから検索品質を最適化）、マルチモーダルRAG（画像、テーブル、コードなどの非テキストコンテンツの検索と生成をサポート）。AI AgentとRAGの深度融合により、ナレッジベースは受動的な検索ツールから、能動的なナレッジアシスタントへと進化するだろう。
関連記事
権威参考
ブラウザローカルツールを無料で試す →
#RAG#AI Agent#大模型#向量检索#企业级落地#2026