NemoClawで社内文書検索AIを構築する｜RAGアーキテクチャ実装ガイド

NemoClaw + RAGで解決できる課題

社内文書検索AIの構築において、NemoClawとRAG（Retrieval-Augmented Generation）を組み合わせることは現在最も実用的なアプローチです。NemoClawのOpenShellサンドボックスとローカルNIMプロファイルにより、データを社内ネットワーク外に一切出さずに構築できます。

システムアーキテクチャの設計

NemoClaw + RAGシステムの全体構成と、各コンポーネントの役割を整理します。

┌────────────────────────────────────────────────────────────┐
│                    社内ネットワーク                          │
│                                                            │
│  ┌──────────┐    ┌────────────────┐    ┌────────────────┐ │
│  │  ユーザー │───▶│  NemoClaw      │───▶│  ベクトルDB    │ │
│  │ (ブラウザ │    │  (OpenShell    │    │  (pgvector or  │ │
│  │  /Slack)  │    │   Runtime)     │◀───│   Milvus)      │ │
│  └──────────┘    └───────┬────────┘    └────────────────┘ │
│                           │                      ▲          │
│                           ▼                      │          │
│                  ┌────────────────┐   ┌──────────────────┐ │
│                  │  Nemotron NIM  │   │ Embedding Model  │ │
│                  │  (推論エンジン  │   │ (NV-Embed-v2     │ │
│                  │  local or API) │   │  or local)       │ │
│                  └────────────────┘   └──────────────────┘ │
│                                                ▲            │
│                                    ┌──────────────────────┐ │
│                                    │  文書取り込みパイプ   │ │
│                                    │  ライン（PDF/Word/    │ │
│                                    │  Excel/Confluence）  │ │
│                                    └──────────────────────┘ │
└────────────────────────────────────────────────────────────┘

ベクトルDBのセットアップ（pgvector と Pinecone）

本ガイドでは pgvector（オンプレミス向け）と Pinecone（クラウドマネージド向け）の両方を解説します。

pgvector のセットアップ（オンプレミス推奨）

pgvector は PostgreSQL の拡張モジュールです。既存の PostgreSQL 環境に追加できるため、運用コストを抑えられます。

# Ubuntu / Debian での pgvector インストール
sudo apt-get install postgresql-16-pgvector

# PostgreSQL に接続して拡張を有効化
psql -U postgres -d your_database -c "CREATE EXTENSION IF NOT EXISTS vector;"

# 文書ベクトル格納テーブルの作成
psql -U postgres -d your_database << "SQL"
CREATE TABLE document_chunks (
    id          BIGSERIAL PRIMARY KEY,
    content     TEXT NOT NULL,
    embedding   VECTOR(1024),          -- NV-Embed-v2 は 1024 次元
    source_file TEXT NOT NULL,
    page_num    INTEGER,
    department  TEXT,
    access_level TEXT DEFAULT 'internal',
    chunk_index INTEGER,
    created_at  TIMESTAMP DEFAULT NOW()
);

-- ベクトル近傍検索用インデックス（HNSW アルゴリズム）
CREATE INDEX ON document_chunks
    USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 64);
SQL

# Python から pgvector に接続（langchain-postgres 使用）
pip install langchain-postgres psycopg2-binary

Pinecone のセットアップ（クラウドマネージド）

Pinecone はサーバーレスのベクトルDBサービスです。インフラ管理が不要で、スモールスタートに適しています。ただし、データがクラウドに送信されるため、機密情報を扱う場合は pgvector または Milvus を使用してください。

# Pinecone クライアントのインストール
pip install pinecone-client langchain-pinecone

# インデックス作成（初回のみ）
import pinecone
import os

pc = pinecone.Pinecone(api_key=os.environ["PINECONE_API_KEY"])

pc.create_index(
    name="company-docs",
    dimension=1024,           # NV-Embed-v2 の次元数
    metric="cosine",
    spec=pinecone.ServerlessSpec(
        cloud="aws",
        region="ap-northeast-1"   # 東京リージョン
    )
)

文書インデキシングパイプラインの実装

PDF・Word・Excel 等の社内文書をベクトルDBに取り込むパイプラインを実装します。

必要パッケージのインストール

# 文書解析・ベクトル化に必要なパッケージ
pip install \
  langchain \
  langchain-community \
  langchain-nvidia-ai-endpoints \
  langchain-postgres \
  pypdf \
  python-docx \
  openpyxl \
  unstructured \
  "unstructured[pdf,docx]"

取り込みスクリプトの実装

# ingest_documents.py
import os
import glob
from typing import List
from pathlib import Path

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import (
    PyPDFLoader, Docx2txtLoader, UnstructuredExcelLoader
)
from langchain_nvidia_ai_endpoints import NVIDIAEmbeddings
from langchain_postgres import PGVector

# 環境変数から接続情報を取得
NVIDIA_API_KEY  = os.environ["NVIDIA_API_KEY"]
PG_CONN         = os.environ["PGVECTOR_CONNECTION_STRING"]
# 例: "postgresql+psycopg2://user:pass@localhost:5432/your_database"
DOCS_DIR        = "./documents"


def load_documents(docs_dir: str) -> list:
    """PDF・Word・Excel を再帰的にロードする"""
    docs = []
    for pdf in glob.glob(f"{docs_dir}/**/*.pdf", recursive=True):
        loader = PyPDFLoader(pdf)
        pages  = loader.load()
        for page in pages:
            page.metadata["department"]    = infer_department(pdf)
            page.metadata["access_level"]  = infer_access_level(pdf)
        docs.extend(pages)

    for docx in glob.glob(f"{docs_dir}/**/*.docx", recursive=True):
        loader = Docx2txtLoader(docx)
        items  = loader.load()
        for item in items:
            item.metadata["department"]   = infer_department(docx)
            item.metadata["access_level"] = infer_access_level(docx)
        docs.extend(items)

    for xlsx in glob.glob(f"{docs_dir}/**/*.xlsx", recursive=True):
        loader = UnstructuredExcelLoader(xlsx, mode="elements")
        items  = loader.load()
        for item in items:
            item.metadata["department"]   = infer_department(xlsx)
            item.metadata["access_level"] = infer_access_level(xlsx)
        docs.extend(items)

    print(f"ロード完了: {len(docs)} ドキュメントチャンク")
    return docs


def infer_department(file_path: str) -> str:
    """ファイルパスから部門を推定する（ディレクトリ名規則を利用）"""
    path = Path(file_path).parts
    dept_map = {"hr": "hr", "finance": "finance", "legal": "legal", "it": "it"}
    for part in path:
        if part.lower() in dept_map:
            return dept_map[part.lower()]
    return "general"


def infer_access_level(file_path: str) -> str:
    """ファイルパスからアクセスレベルを推定する"""
    path_str = file_path.lower()
    if "confidential" in path_str or "secret" in path_str:
        return "confidential"
    elif "internal" in path_str:
        return "internal"
    return "public"


def split_documents(docs: list) -> list:
    """日本語対応チャンク分割"""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=600,
        chunk_overlap=60,
        separators=["\n\n", "\n", "。", "．", "、", " ", ""],
    )
    chunks = splitter.split_documents(docs)
    print(f"チャンク分割完了: {len(chunks)} チャンク")
    return chunks


def create_vector_store(chunks: list) -> PGVector:
    """pgvector にベクトルを格納する"""
    embeddings = NVIDIAEmbeddings(
        model="nvidia/nv-embed-v2",
        api_key=NVIDIA_API_KEY,
    )
    vector_store = PGVector.from_documents(
        documents=chunks,
        embedding=embeddings,
        connection=PG_CONN,
        collection_name="company_docs",
        pre_delete_collection=True,   # 再実行時は既存を削除
    )
    print(f"ベクトルDB格納完了: {len(chunks)} チャンク")
    return vector_store


if __name__ == "__main__":
    raw_docs     = load_documents(DOCS_DIR)
    chunks       = split_documents(raw_docs)
    vector_store = create_vector_store(chunks)
    print("ドキュメント取り込み完了")

# 実行コマンド
python ingest_documents.py

NemoClaw の blueprint.yaml に RAGツールを統合する

NemoClaw の blueprint.yaml に文書検索ツールを定義し、エージェントが質問を受けたときに自動的にベクトルDB検索を行うよう設定します。

# blueprint.yaml（RAG文書検索AI統合版）
profile: local-nim

agent:
  name: document-search-ai
  description: "社内文書検索AIエージェント"

model:
  provider: local
  name: nemotron-3-super-120b
  device: cuda:0
  precision: bf16

sandbox:
  enabled: true
  filesystem:
    allow:
      - "./workspace/"
      - "./prompts/"
    deny:
      - "./.env"
      - "/etc/"
      - "/home/"
  network:
    allow:
      - "localhost"         # pgvector（同一サーバー）
      - "pgvector.internal" # pgvector（内部ドメイン）
    deny:
      - "*"

tools:
  - name: search_company_docs
    description: |
      社内文書データベースを検索して関連情報を取得します。
      社内規程・マニュアル・議事録・技術仕様書に関する質問に使用してください。
      回答の末尾に参照元ファイル名とページ番号を必ず記載すること。
    command: "python3 ./tools/rag_search.py"
    input_schema:
      type: object
      properties:
        query:
          type: string
          description: "検索クエリ（自然言語で記述）"
        top_k:
          type: integer
          description: "取得する検索結果数（デフォルト: 5）"
          default: 5
        department:
          type: string
          description: "絞り込む部門（任意）: hr / finance / legal / it / general"
        access_level:
          type: string
          description: "ユーザーのアクセスレベル: public / internal / confidential"
          default: "internal"
      required: ["query"]
    retry: 2
    timeout_seconds: 15

memory:
  enabled: true
  backend: redis
  redis:
    host: "redis.internal"
    port: 6379
    password: "${REDIS_PASSWORD}"
    ttl: 3600
  max_turns: 30

guardrails:
  enabled: true
  pii_detection: true
  injection_guard: true

logging:
  level: info
  file: "./workspace/rag-agent.log"

# tools/rag_search.py
import sys
import json
import os
from langchain_nvidia_ai_endpoints import NVIDIAEmbeddings
from langchain_postgres import PGVector

NVIDIA_API_KEY = os.environ["NVIDIA_API_KEY"]
PG_CONN        = os.environ["PGVECTOR_CONNECTION_STRING"]

CLEARANCE_LEVELS = {"public": 0, "internal": 1, "confidential": 2}


def search_company_docs(
    query: str,
    top_k: int = 5,
    department: str = None,
    access_level: str = "internal",
) -> dict:
    embeddings = NVIDIAEmbeddings(
        model="nvidia/nv-embed-v2",
        api_key=NVIDIA_API_KEY,
    )
    vector_store = PGVector(
        embeddings=embeddings,
        collection_name="company_docs",
        connection=PG_CONN,
    )

    # アクセス制御フィルター構築
    user_level    = CLEARANCE_LEVELS.get(access_level, 0)
    allowed_levels = [k for k, v in CLEARANCE_LEVELS.items() if v <= user_level]

    filter_conditions = {"access_level": {"$in": allowed_levels}}
    if department:
        filter_conditions["department"] = {"$in": [department, "general"]}

    results = vector_store.similarity_search_with_score(
        query=query,
        k=top_k,
        filter=filter_conditions,
    )

    formatted = []
    for doc, score in results:
        formatted.append({
            "content":         doc.page_content,
            "source_file":     doc.metadata.get("source", "不明"),
            "page_num":        doc.metadata.get("page", 0),
            "department":      doc.metadata.get("department", "general"),
            "relevance_score": round(float(score), 4),
        })

    return {"results": formatted, "total": len(formatted)}


if __name__ == "__main__":
    input_data = json.loads(sys.stdin.read())
    result     = search_company_docs(**input_data)
    print(json.dumps(result, ensure_ascii=False))

検索精度チューニング

RAGシステムの精度は適切なパラメーター調整で大幅に改善できます。以下の4つの観点からチューニングを行います。

チャンク分割戦略の最適化

ハイブリッド検索の導入

ベクトル検索単体よりも BM25 キーワード検索を組み合わせたハイブリッド検索の方が、製品コードや固有名詞を含む質問への精度が向上します。

# ハイブリッド検索の実装（LangChain EnsembleRetriever）
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever

# ベクトル検索リトリーバー
vector_retriever = vector_store.as_retriever(
    search_kwargs={"k": 5, "filter": filter_conditions}
)

# BM25 キーワード検索リトリーバー（全チャンクを渡す）
all_docs       = fetch_all_docs_from_pgvector()
bm25_retriever = BM25Retriever.from_documents(all_docs, k=5)

# アンサンブル（ベクトル60% + BM25 40%）
ensemble_retriever = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.6, 0.4],
)

results = ensemble_retriever.invoke("有給休暇の申請手順")

Re-ranking による精度向上

検索結果の上位N件に対してクロスエンコーダーモデルで再スコアリングを行うと、最終的な精度が向上します。NVIDIAが提供する nvidia/nv-rerankqa-mistral-4b-v3 が利用可能です。

# Re-ranker の実装
from langchain_nvidia_ai_endpoints import NVIDIARerank

reranker = NVIDIARerank(
    model="nvidia/nv-rerankqa-mistral-4b-v3",
    api_key=os.environ["NVIDIA_API_KEY"],
    top_n=3,       # 上位3件に絞り込む
)

# 一次検索で10件取得 → Re-ranking で上位3件に絞り込む
initial_results = vector_retriever.invoke(query, config={"k": 10})
reranked        = reranker.compress_documents(initial_results, query)

アクセス権制御の実装

社内文書検索AIでは「誰がどの文書を閲覧できるか」の制御が不可欠です。ベクトルDBのメタデータフィルターと NemoClaw のサンドボックスを組み合わせた多層防御を実装します。

# ドキュメント取り込み時にアクセスレベルとメタデータを付与する例
def ingest_hr_documents():
    """人事部の機密文書を取り込む"""
    docs = load_documents("./documents/hr/")
    for doc in docs:
        doc.metadata["department"]    = "hr"
        doc.metadata["access_level"]  = "confidential"
    chunks = split_documents(docs)
    create_vector_store(chunks)

def ingest_public_documents():
    """全社公開の規程集を取り込む"""
    docs = load_documents("./documents/public/")
    for doc in docs:
        doc.metadata["department"]   = "general"
        doc.metadata["access_level"] = "public"
    chunks = split_documents(docs)
    create_vector_store(chunks)

# NemoClaw エージェント呼び出し時にユーザー権限をセッションコンテキストで渡す
# （blueprint.yaml の sandbox.session_context で LDAP/AD 連携が可能）
POST /v1/chat HTTP/1.1
Content-Type: application/json
X-Session-Id: user-12345
X-User-Department: hr
X-User-Access-Level: confidential

{
  "message": "育児休業申請に必要な書類を教えてください"
}

インデックスの更新と運用保守

社内文書は日常的に更新されます。ベクトルDBを常に最新状態に保つ差分更新パイプラインを実装します。

# update_index.py（差分更新スクリプト）
import os
import hashlib
import json
from pathlib import Path

DOCS_DIR   = "./documents"
HASH_CACHE = "./workspace/doc_hashes.json"


def get_file_hash(file_path: str) -> str:
    with open(file_path, "rb") as f:
        return hashlib.sha256(f.read()).hexdigest()


def detect_changes() -> dict:
    """追加・変更・削除されたファイルを検出する"""
    old_hashes = {}
    if Path(HASH_CACHE).exists():
        with open(HASH_CACHE) as f:
            old_hashes = json.load(f)

    current_hashes = {}
    for path in Path(DOCS_DIR).rglob("*"):
        if path.is_file() and path.suffix in [".pdf", ".docx", ".xlsx"]:
            current_hashes[str(path)] = get_file_hash(str(path))

    added   = set(current_hashes) - set(old_hashes)
    deleted = set(old_hashes) - set(current_hashes)
    changed = {
        p for p in current_hashes
        if p in old_hashes and current_hashes[p] != old_hashes[p]
    }

    return {
        "added":   list(added),
        "deleted": list(deleted),
        "changed": list(changed),
        "current": current_hashes,
    }


if __name__ == "__main__":
    changes = detect_changes()
    total   = len(changes["added"]) + len(changes["deleted"]) + len(changes["changed"])
    if total == 0:
        print("変更なし。インデックス更新をスキップします。")
    else:
        print(
            f"変更検出: 追加 {len(changes['added'])} 件 / "
            f"削除 {len(changes['deleted'])} 件 / "
            f"更新 {len(changes['changed'])} 件"
        )
        # 削除ファイルはベクトルDBから該当ドキュメントを削除
        # 追加・変更ファイルは再インジェスト
        # （実装省略: source_file フィルターで削除 → 再インジェスト）
        with open(HASH_CACHE, "w") as f:
            json.dump(changes["current"], f, ensure_ascii=False)

# cron で毎日深夜2時に差分更新を実行する設定例
0 2 * * * cd /opt/nemoclaw/doc-search && python3 update_index.py >> /var/log/nemoclaw-index.log 2>&1