RAGとNemoClawの組み合わせが最強な理由

RAG（Retrieval-Augmented Generation：検索拡張生成）は、AIモデルが学習していない社内固有の情報を、リアルタイムに検索してから回答を生成する手法です。NemoClawとRAGを組み合わせることで、以下のメリットが得られます。

課題RAGなし（単体LLM）NemoClaw + RAG
社内規程・マニュアル学習データに含まれず回答不可リアルタイム検索で正確に回答
最新情報への対応カットオフ日以降は不正確ドキュメント更新が即時反映
情報の根拠ハルシネーション（幻覚）のリスク参照元ドキュメントを明示
データの機密性クラウドLLMへのデータ送信が必要ローカルNIMで完全オンプレミス処理
アクセス制御モデル側での制御不可ユーザー権限に応じた検索範囲制限

NemoClawのローカルNIMプロファイルとオンプレミスのベクトルDBを組み合わせると、すべてのデータ処理を社内ネットワーク内に閉じることができます。金融・医療・法律分野など、厳格なデータ管理が求められる業種に最適です。

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

NemoClaw + RAGシステムの全体像を示します。

┌─────────────────────────────────────────────────────────────┐
│                    社内ネットワーク                           │
│                                                             │
│  ┌──────────┐    ┌──────────────┐    ┌──────────────────┐ │
│  │ ユーザー  │───▶│  NemoClaw    │───▶│  ベクトルDB      │ │
│  │ (OpenClaw│    │  (OpenShell  │    │  (Milvus/pgvector│ │
│  │  Chat)   │    │   Runtime)   │    │   /Weaviate)      │ │
│  └──────────┘    └──────┬───────┘    └──────────────────┘ │
│                         │                       ▲           │
│                         │                       │           │
│                         ▼                       │           │
│                  ┌──────────────┐    ┌──────────────────┐ │
│                  │  Nemotron    │    │  Embedding API   │ │
│                  │  (ローカルNIM│    │  (NV-Embed-v2)   │ │
│                  │   または     │    └──────────────────┘ │
│                  │  クラウドAPI)│               ▲           │
│                  └──────────────┘               │           │
│                                      ┌──────────────────┐ │
│                                      │  ドキュメント    │ │
│                                      │  取り込みパイプ  │ │
│                                      │  ライン          │ │
│                                      └──────────────────┘ │
└─────────────────────────────────────────────────────────────┘

ドキュメント取り込みパイプラインの構築

社内文書（PDF・Word・Excelなど）をベクトルDBに取り込むパイプラインを構築します。

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

# Python環境のセットアップ
pip install \
  langchain \
  langchain-community \
  langchain-nvidia-ai-endpoints \
  pymilvus \
  pypdf \
  python-docx \
  openpyxl \
  unstructured \
  sentence-transformers

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

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

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

NVIDIA_API_KEY = os.environ["NVIDIA_API_KEY"]
MILVUS_HOST   = os.environ.get("MILVUS_HOST", "localhost")
MILVUS_PORT   = int(os.environ.get("MILVUS_PORT", 19530))
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)
        docs.extend(loader.load())
    for docx in glob.glob(f"{docs_dir}/**/*.docx", recursive=True):
        loader = Docx2txtLoader(docx)
        docs.extend(loader.load())
    for xlsx in glob.glob(f"{docs_dir}/**/*.xlsx", recursive=True):
        loader = UnstructuredExcelLoader(xlsx)
        docs.extend(loader.load())
    print(f"ロード完了: {len(docs)} チャンク")
    return docs

def split_documents(docs: list) -> list:
    """チャンク分割（日本語対応）"""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,           # 500文字ごとに分割
        chunk_overlap=50,         # 前後50文字のオーバーラップ
        separators=["\n\n", "\n", "。", "．", "、", " ", ""],
    )
    return splitter.split_documents(docs)

def create_vector_store(chunks: list) -> Milvus:
    """ベクトルDBへの格納"""
    embeddings = NVIDIAEmbeddings(
        model="nvidia/nv-embed-v2",
        api_key=NVIDIA_API_KEY,
    )
    vector_store = Milvus.from_documents(
        documents=chunks,
        embedding=embeddings,
        connection_args={"host": MILVUS_HOST, "port": MILVUS_PORT},
        collection_name="company_docs",
        drop_old=True,   # 再インジェスト時は既存コレクションを削除
    )
    print(f"ベクトルDB格納完了: {len(chunks)} チャンク")
    return vector_store

if __name__ == "__main__":
    raw_docs = load_documents(DOCS_DIR)
    chunks   = split_documents(raw_docs)
    vs       = create_vector_store(chunks)
    print("ドキュメント取り込み完了")
# ドキュメントディレクトリにファイルを置いて実行
mkdir -p documents
cp /path/to/company/manuals/*.pdf documents/
python ingest_documents.py

NemoClawとベクトルDBの統合設定

NemoClawのblueprintに検索ツールを追加し、Milvusへの問い合わせを実行できるようにします。

# blueprint.yaml（RAG統合版）
profile: local-nim

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

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

tools:
  - name: search_documents
    description: |
      社内文書データベースを検索して関連情報を取得します。
      社内規程・マニュアル・議事録・技術仕様書などの質問に使用してください。
    command: "python3 ./tools/rag_search.py"
    input_schema:
      type: object
      properties:
        query:
          type: string
          description: "検索クエリ（自然言語で記述）"
        top_k:
          type: integer
          description: "取得する検索結果数（デフォルト: 5）"
          default: 5
        filter_department:
          type: string
          description: "部門フィルター（任意）: hr, finance, it, legal"
      required: ["query"]
# tools/rag_search.py
import sys
import json
import os
from langchain_nvidia_ai_endpoints import NVIDIAEmbeddings
from langchain_community.vectorstores import Milvus

def search_documents(query: str, top_k: int = 5, filter_department: str = None) -> dict:
    embeddings = NVIDIAEmbeddings(
        model="nvidia/nv-embed-v2",
        api_key=os.environ["NVIDIA_API_KEY"],
    )
    vector_store = Milvus(
        embedding_function=embeddings,
        connection_args={
            "host": os.environ.get("MILVUS_HOST", "localhost"),
            "port": int(os.environ.get("MILVUS_PORT", 19530)),
        },
        collection_name="company_docs",
    )

    # 部門フィルターがある場合はメタデータで絞り込む
    search_kwargs = {"k": top_k}
    if filter_department:
        search_kwargs["filter"] = "department == " + filter_department

    results = vector_store.similarity_search_with_score(query, **search_kwargs)

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

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

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

検索精度チューニング

RAGの検索精度は適切なチューニングで大幅に改善できます。以下の観点から調整を行います。

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

ドキュメント種別推奨チャンクサイズオーバーラップ分割基準
社内規程・就業規則800〜1,200文字100文字条項・見出し単位
技術マニュアル400〜600文字80文字手順ステップ単位
議事録・報告書300〜500文字50文字段落単位
FAQ・ナレッジベース200〜400文字30文字Q&A単位

ベクトル検索単体よりも、キーワード検索（BM25）と組み合わせたハイブリッド検索の方が精度が高い場合があります。特に固有名詞・製品コードなど正確なキーワードマッチが必要な場面で有効です。

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

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

# BM25キーワード検索リトリーバー（全チャンクが必要）
all_docs = load_all_chunks_from_milvus()
bm25_retriever = BM25Retriever.from_documents(all_docs, k=5)

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

results = ensemble_retriever.invoke("有給休暇の取得ルール")

セキュリティ設定とアクセス制御

社内文書検索AIでは、「誰がどの文書にアクセスできるか」の制御が非常に重要です。NemoClawのサンドボックスと組み合わせた権限制御の実装方法を解説します。

# ドキュメント取り込み時にメタデータで権限を付与
def ingest_with_acl(docs_dir: str, department: str, access_level: str):
    """
    access_level: "public" | "internal" | "confidential" | "secret"
    """
    docs = load_documents(docs_dir)
    for doc in docs:
        doc.metadata["department"]    = department
        doc.metadata["access_level"]  = access_level
    chunks = split_documents(docs)
    create_vector_store(chunks)

# 人事部の機密文書（HR部門限定）
ingest_with_acl("./documents/hr/confidential/", "hr", "confidential")

# 全社公開の規程集
ingest_with_acl("./documents/company-wide/", "all", "public")
# rag_search.pyでのアクセス制御フィルタリング
def search_with_acl(query: str, user_department: str, user_clearance: str, top_k: int = 5) -> dict:
    CLEARANCE_LEVELS = {"public": 0, "internal": 1, "confidential": 2, "secret": 3}
    user_level = CLEARANCE_LEVELS.get(user_clearance, 0)

    # アクセス可能なレベルのみ検索
    allowed_levels = [k for k, v in CLEARANCE_LEVELS.items() if v <= user_level]

    filter_expr = (
        "(department in [\"all\", \"" + user_department + "\"]) and "
        "(access_level in " + json.dumps(allowed_levels) + ")"
    )

    results = vector_store.similarity_search_with_score(
        query,
        k=top_k,
        filter=filter_expr,
    )
    return {"results": [...]}

ユーザーの部門・権限情報はNemoClawのセッションコンテキストから取得します。blueprint.yamlのsandbox.session_context設定でLDAP/Active Directoryとの連携が可能です。

ドキュメント更新と運用保守

社内文書は頻繁に更新されます。ベクトルDBを常に最新状態に保つための運用設計が必要です。

# 差分更新スクリプト（cronで定期実行）
#!/usr/bin/env python3
# update_documents.py

import os
import hashlib
import json
from pathlib import Path

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

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

def detect_changes() -> dict:
    """変更・追加・削除されたファイルを検出"""
    old_hashes = json.loads(open(HASH_CACHE).read()) if Path(HASH_CACHE).exists() else {}
    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": added, "deleted": deleted, "changed": changed, "current": current_hashes}

if __name__ == "__main__":
    changes = detect_changes()
    if any([changes["added"], changes["deleted"], changes["changed"]]):
        added_count = len(changes["added"])
        deleted_count = len(changes["deleted"])
        changed_count = len(changes["changed"])
        print(f"変更検出: 追加{added_count}件, 削除{deleted_count}件, 更新{changed_count}件")
        # 差分のみ再インジェスト処理...
        # （省略: 削除ファイルはMilvusから該当ベクトルを削除、追加/変更ファイルは再インジェスト）
        json.dump(changes["current"], open(HASH_CACHE, "w"))
    else:
        print("変更なし")
# cronの設定例（毎日深夜2時に差分更新）
0 2 * * * cd /opt/rag-system && python3 update_documents.py >> /var/log/rag-update.log 2>&1