先週の会議の議事録はどこだっけ」「この手続きのルールって何だったか」「新入社員に同じことを何度も教えている」——こうした社内の情報管理に関する悩みは、どの組織にも共通しています。
ChatGPTのようなクラウドAIに社内文書を読み込ませる方法もありますが、機密情報・個人情報・営業秘密が外部サーバーに送信されることへの懸念が拭えません。そこで注目されているのが、RAG（Retrieval-Augmented Generation）をローカル環境で構築するというアプローチです。
本記事では、Ollamaをエンジンに、社内文書・マニュアル・議事録・ナレッジベースを丸ごと取り込んで「社内専用AI」を作る方法を、コードとともに徹底解説します。完成したシステムはインターネット不要・データは自社サーバー内に完結します。
この記事で構築するもの
- PDF・Word・Markdown・テキストファイルを一括インデックス化するパイプライン
- Ollamaの埋め込みモデルを使ったローカルベクトルデータベース（ChromaDB）
- 質問に対して関連文書を検索し、AIが回答するRAGエンジン
- 社内Slack・ブラウザから使えるWebインターフェース（Open WebUI連携）
- 新しい文書を追加すると自動的にインデックスを更新する仕組み
- RAGとは何か：仕組みと社内活用の価値
- アーキテクチャ設計：全体構成の決定
- 環境構築：インストールとセットアップ
- 文書ローダー：あらゆる形式のファイルを取り込む
- インデクサー：文書をベクトルデータベースに登録する
- RAGエンジン：検索と生成の中核
- FastAPI サーバー：Webからアクセスできるようにする
- CLIツール：ターミナルから対話する
- ファイル変更の自動監視：文書を追加したら即インデックス更新
- Open WebUI との連携：ブラウザからチャット感覚で使う
- Slack Bot との連携：チャンネルで質問できるようにする
- 精度向上のテクニック：RAGをチューニングする
- セキュリティと運用設計
- まとめ：社内ナレッジAIの効果と展望
RAGとは何か：仕組みと社内活用の価値
LLMだけでは解決できない「知識の壁」
ChatGPTをはじめとする大規模言語モデル（LLM）は、学習データに含まれる膨大な知識を持っています。しかし、いくつかの根本的な限界があります。
- 学習データのカットオフ：学習完了以降の情報は知らない
- 社内固有の知識を持たない：自社の規則・製品・顧客情報は学習していない
- ハルシネーション（幻覚）：知らないことを「知っているかのように」答えてしまう
- 情報の出典が不明確：どの文書に基づいて回答したか分からない
RAG（Retrieval-Augmented Generation：検索拡張生成）はこれらの問題を解決するアーキテクチャです。
RAGの動作原理
【RAGの処理フロー】
① ユーザーが質問する
「有給休暇の申請方法を教えて」
② 質問をベクトル化（埋め込みモデル）
「有給休暇の申請方法を教えて」→ [0.23, -0.41, 0.87, ...]（数値の配列）
③ ベクトルデータベースで類似文書を検索
→ 就業規則第15条（有給休暇の取得手続き）.pdf 類似度：0.94
→ 人事FAQ_v3.docx（Q: 有給はいつ申請する？） 類似度：0.89
→ 社内ポータル_休暇申請.md 類似度：0.85
④ 検索した文書の断片（チャンク）をプロンプトに追加
「以下の文書を参考に、質問に答えてください：
[就業規則第15条の内容...]
[人事FAQの内容...]
質問：有給休暇の申請方法を教えて」
⑤ Ollamaが文書を参照して回答を生成
「就業規則第15条に基づき、有給休暇の申請は
取得希望日の3営業日前までに、社内ポータルの
申請フォームから上長に申請します。急病の場合は...」
⑥ 出典情報と一緒に回答を返す
「参照文書：就業規則.pdf（15条）」
クラウドRAG vs ローカルRAGの比較
| 比較項目 | クラウドRAG（OpenAI等） | ローカルRAG（Ollama） |
|---|---|---|
| データのプライバシー | ❌ 文書が外部送信される | ✅ 自社内で完結 |
| 月額コスト | ❌ API料金（文書量・利用量に比例） | ✅ 初期ハードウェアコストのみ |
| ネット接続 | ❌ 必須 | ✅ 不要（オフライン可） |
| カスタマイズ性 | △ 制限あり | ✅ 完全自由 |
| 導入の手軽さ | ✅ すぐに使える | △ 構築が必要 |
| モデル品質 | ✅ GPT-4等の最高品質 | △ オープンソースモデル |
| セキュリティ認証対応 | △ ベンダー依存 | ✅ 自社ポリシーで管理 |
どんな組織にローカルRAGが向いているか
機密文書・個人情報・営業秘密を多く扱う組織（法律事務所、医療機関、金融機関、製造業の技術部門等）、クラウドサービスの利用を制限するセキュリティポリシーがある組織、長期的なコスト削減を重視する組織、インターネット接続が制限された環境（工場内LANなど）で特に有効です。
アーキテクチャ設計：全体構成の決定
使用するコンポーネント
【社内ナレッジAI の全体構成】
┌─────────────────────────────────────────────────┐
│ ユーザー層 │
│ Webブラウザ / Slack Bot / API クライアント │
└──────────────────┬──────────────────────────────┘
│ HTTP
┌──────────────────▼──────────────────────────────┐
│ Webインターフェース層 │
│ FastAPI（バックエンドAPI） │
│ Open WebUI（チャットUI） │
└──────────────────┬──────────────────────────────┘
│
┌──────────────────▼──────────────────────────────┐
│ RAGエンジン層 │
│ LangChain / LlamaIndex（オーケストレーション） │
│ ┌─────────────┐ ┌──────────────────────────┐ │
│ │ 文書ローダー │ │ テキスト分割（チャンク化） │ │
│ └─────────────┘ └──────────────────────────┘ │
└──────────┬───────────────────┬──────────────────┘
│ │
┌──────────▼──────┐ ┌─────────▼───────────────────┐
│ Ollama │ │ ベクトルDB（ChromaDB） │
│ ┌───────────┐ │ │ ・文書チャンクの保存 │
│ │生成モデル │ │ │ ・ベクトル検索 │
│ │qwen2.5:14b│ │ │ ・メタデータフィルタ │
│ ├───────────┤ │ └─────────────────────────────┘
│ │埋め込みモデ│ │
│ │nomic-embed │ │
│ └───────────┘ │
└─────────────────┘
┌────────────────────────────────────────────────┐
│ 文書ソース層 │
│ PDF / Word(.docx) / Excel / Markdown / TXT │
│ （社内ファイルサーバー / SharePoint / NAS等） │
└────────────────────────────────────────────────┘
使用するライブラリとモデルの選定
| 役割 | 選択肢 | 採用理由 |
|---|---|---|
| 生成モデル | qwen2.5:14b | 日本語品質が高く、指示追従性が優秀 |
| 埋め込みモデル | nomic-embed-text | Ollama公式対応・高精度・軽量（274MB） |
| ベクトルDB | ChromaDB | ローカル動作・Pythonネイティブ・永続化対応 |
| オーケストレーション | LangChain | 豊富なドキュメントローダー・RAG実装の充実 |
| Webフレームワーク | FastAPI | 非同期処理・自動API文書生成 |
| WebUI | Open WebUI | Docker一発で起動・ChatGPT風UI |
環境構築：インストールとセットアップ
必要なモデルのダウンロード
# 生成モデル（回答生成用） # 推奨：qwen2.5:14b（14GB RAM以上） ollama pull qwen2.5:14b # スペックが厳しい場合の代替 ollama pull qwen2.5:7b # 8GB RAM以上 ollama pull qwen2.5:3b # 4GB RAM以上（精度は低下） # 埋め込みモデル（文書のベクトル化用） # ※これは必ずダウンロードすること（274MBと軽量） ollama pull nomic-embed-text # インストール確認 ollama list # NAME ID SIZE MODIFIED # qwen2.5:14b ac89... 9.0 GB ... # nomic-embed-text:latest 0a10... 274 MB ...
Pythonプロジェクトの初期化
# プロジェクトディレクトリの作成 mkdir company-knowledge-ai cd company-knowledge-ai # 仮想環境の作成 python -m venv .venv source .venv/bin/activate # macOS/Linux # .venv\Scripts\activate # Windows # 必要ライブラリのインストール pip install langchain langchain-community langchain-ollama pip install chromadb pip install pypdf # PDF処理 pip install python-docx # Word文書処理 pip install openpyxl # Excel処理 pip install unstructured # 多形式ファイル処理 pip install fastapi uvicorn # WebAPIサーバー pip install python-multipart # ファイルアップロード pip install watchdog # ファイル変更監視 pip install python-dotenv rich # 設定・表示 # インストール確認 pip list | grep -E "langchain|chroma|pypdf|fastapi"
プロジェクト構成
company-knowledge-ai/ ├── .env # 設定ファイル ├── config.py # 設定の読み込み ├── document_loader.py # 文書ローダー ├── indexer.py # インデックス構築 ├── rag_engine.py # RAGエンジン本体 ├── api_server.py # FastAPI サーバー ├── cli.py # コマンドラインインターフェース ├── auto_watcher.py # ファイル自動監視 ├── documents/ # 社内文書の置き場所 │ ├── manuals/ │ ├── rules/ │ └── faq/ ├── chroma_db/ # ChromaDBのデータ（自動生成） └── logs/ # ログファイル
# .env ファイル（設定値） OLLAMA_BASE_URL=http://localhost:11434 EMBED_MODEL=nomic-embed-text GENERATE_MODEL=qwen2.5:14b CHROMA_DB_PATH=./chroma_db DOCUMENTS_DIR=./documents CHUNK_SIZE=800 CHUNK_OVERLAP=100 RETRIEVAL_K=5 API_HOST=127.0.0.1 API_PORT=8000
文書ローダー：あらゆる形式のファイルを取り込む
#!/usr/bin/env python3
"""
document_loader.py
PDF・Word・Excel・Markdown・TXTを読み込むユニバーサルローダー
"""
import os
from pathlib import Path
from typing import List, Dict, Any
from datetime import datetime
from langchain.schema import Document
from langchain_community.document_loaders import (
PyPDFLoader, # PDF
Docx2txtLoader, # Word (.docx)
UnstructuredExcelLoader, # Excel (.xlsx)
TextLoader, # テキスト・Markdown
UnstructuredHTMLLoader, # HTML
CSVLoader, # CSV
)
from rich.console import Console
from rich.progress import Progress, TaskID
console = Console()
class UniversalDocumentLoader:
"""あらゆる形式の文書を統一的に読み込むローダー"""
# サポートする拡張子とローダーのマッピング
LOADER_MAP = {
".pdf": PyPDFLoader,
".docx": Docx2txtLoader,
".xlsx": UnstructuredExcelLoader,
".xls": UnstructuredExcelLoader,
".txt": TextLoader,
".md": TextLoader,
".markdown": TextLoader,
".html": UnstructuredHTMLLoader,
".htm": UnstructuredHTMLLoader,
".csv": CSVLoader,
}
def __init__(self, documents_dir: str):
self.documents_dir = Path(documents_dir)
def load_file(self, file_path: Path) -> List[Document]:
"""単一ファイルを読み込む"""
suffix = file_path.suffix.lower()
loader_class = self.LOADER_MAP.get(suffix)
if not loader_class:
console.print(f" [yellow]⚠ 未対応形式をスキップ: {file_path.name}[/]")
return []
try:
# エンコーディング指定が必要なファイル形式
if suffix in (".txt", ".md", ".markdown"):
loader = loader_class(str(file_path), encoding="utf-8")
else:
loader = loader_class(str(file_path))
documents = loader.load()
# メタデータを強化する
for doc in documents:
doc.metadata.update({
"source_file": file_path.name,
"source_path": str(file_path),
"file_type": suffix.lstrip(".").upper(),
"indexed_at": datetime.now().isoformat(),
"file_size_kb": round(file_path.stat().st_size / 1024, 1),
# フォルダ名をカテゴリとして使用
"category": file_path.parent.name,
})
return documents
except Exception as e:
console.print(
f" [red]✗ 読み込みエラー {file_path.name}: {e}[/]"
)
return []
def load_directory(
self,
directory: Path = None,
recursive: bool = True
) -> List[Document]:
"""ディレクトリ内の全対応ファイルを再帰的に読み込む"""
target_dir = directory or self.documents_dir
all_docs: List[Document] = []
# 対応ファイルを列挙
pattern = "**/*" if recursive else "*"
files = [
f for f in target_dir.glob(pattern)
if f.is_file() and f.suffix.lower() in self.LOADER_MAP
and not f.name.startswith(".") # 隠しファイルを除外
and "~$" not in f.name # Officeの一時ファイルを除外
]
if not files:
console.print(
f"[yellow]⚠ {target_dir} に対応ファイルが見つかりません[/]"
)
return []
console.print(f"\n[cyan]📂 {len(files)}件のファイルを処理します...[/]")
with Progress() as progress:
task = progress.add_task("読み込み中...", total=len(files))
for file_path in files:
progress.update(task, description=f"読み込み: {file_path.name[:40]}")
docs = self.load_file(file_path)
all_docs.extend(docs)
progress.advance(task)
console.print(
f"[green]✓ 読み込み完了: {len(files)}ファイル → {len(all_docs)}ページ/セクション[/]"
)
return all_docs
インデクサー：文書をベクトルデータベースに登録する
#!/usr/bin/env python3
"""
indexer.py
文書をチャンク分割してChromaDBに登録するインデクサー
"""
import os
import hashlib
import json
from pathlib import Path
from typing import List, Optional
from datetime import datetime
import chromadb
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document
from langchain_ollama import OllamaEmbeddings
from langchain_community.vectorstores import Chroma
from rich.console import Console
from document_loader import UniversalDocumentLoader
console = Console()
# ─── 設定 ─────────────────────────────────────────
EMBED_MODEL = os.getenv("EMBED_MODEL", "nomic-embed-text")
OLLAMA_URL = os.getenv("OLLAMA_BASE_URL", "http://localhost:11434")
CHROMA_DB_PATH = os.getenv("CHROMA_DB_PATH", "./chroma_db")
DOCUMENTS_DIR = os.getenv("DOCUMENTS_DIR", "./documents")
CHUNK_SIZE = int(os.getenv("CHUNK_SIZE", "800"))
CHUNK_OVERLAP = int(os.getenv("CHUNK_OVERLAP", "100"))
COLLECTION_NAME = "company_knowledge" # ChromaDBのコレクション名
class KnowledgeIndexer:
"""社内文書をベクトルDBにインデックスするクラス"""
def __init__(self):
# 埋め込みモデルの初期化
console.print(f"[cyan]🔄 埋め込みモデルを初期化中: {EMBED_MODEL}[/]")
self.embeddings = OllamaEmbeddings(
model=EMBED_MODEL,
base_url=OLLAMA_URL
)
# テキスト分割器の設定
# 日本語文書に合わせた区切り文字の優先順位
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=CHUNK_SIZE,
chunk_overlap=CHUNK_OVERLAP,
separators=[
"\n## ", "\n### ", "\n#### ", # Markdownの見出し
"\n\n", # 段落区切り
"\n", # 改行
"。", "．", # 日本語の句点
"？", "！", # 疑問符・感嘆符
" ", "", # スペース・文字
],
length_function=len,
)
# ChromaDB クライアントの初期化
self.chroma_client = chromadb.PersistentClient(path=CHROMA_DB_PATH)
# LangChain経由のChromaVectorStore
self.vectorstore = Chroma(
collection_name=COLLECTION_NAME,
embedding_function=self.embeddings,
client=self.chroma_client,
persist_directory=CHROMA_DB_PATH
)
# 文書ローダーの初期化
self.loader = UniversalDocumentLoader(DOCUMENTS_DIR)
console.print("[green]✓ インデクサー初期化完了[/]")
def _compute_doc_hash(self, file_path: str) -> str:
"""ファイルのMD5ハッシュを計算して変更検知に使う"""
hasher = hashlib.md5()
with open(file_path, "rb") as f:
for chunk in iter(lambda: f.read(8192), b""):
hasher.update(chunk)
return hasher.hexdigest()
def _get_indexed_files(self) -> dict:
"""インデックス済みファイルの情報を取得する"""
hash_file = Path(CHROMA_DB_PATH) / "indexed_files.json"
if hash_file.exists():
with open(hash_file) as f:
return json.load(f)
return {}
def _save_indexed_files(self, indexed: dict) -> None:
"""インデックス済みファイル情報を保存する"""
hash_file = Path(CHROMA_DB_PATH) / "indexed_files.json"
Path(CHROMA_DB_PATH).mkdir(parents=True, exist_ok=True)
with open(hash_file, "w") as f:
json.dump(indexed, f, ensure_ascii=False, indent=2)
def split_documents(self, documents: List[Document]) -> List[Document]:
"""文書をチャンクに分割する"""
chunks = self.text_splitter.split_documents(documents)
console.print(
f"[cyan]✂ チャンク分割: {len(documents)}ページ → {len(chunks)}チャンク[/]"
)
return chunks
def index_documents(
self,
force_reindex: bool = False
) -> dict:
"""
documentsディレクトリの文書をインデックスする
force_reindex=True で全ファイルを強制再インデックス
"""
console.print("\n[bold cyan]📚 インデックス構築を開始します[/]")
indexed_files = {} if force_reindex else self._get_indexed_files()
loader = UniversalDocumentLoader(DOCUMENTS_DIR)
supported_files = [
f for f in Path(DOCUMENTS_DIR).rglob("**/*")
if f.is_file()
and f.suffix.lower() in loader.LOADER_MAP
and not f.name.startswith(".")
and "~$" not in f.name
]
stats = {"added": 0, "skipped": 0, "errors": 0, "total_chunks": 0}
for file_path in supported_files:
file_key = str(file_path)
current_hash = self._compute_doc_hash(file_key)
# ハッシュが変わっていなければスキップ
if not force_reindex and indexed_files.get(file_key) == current_hash:
stats["skipped"] += 1
continue
console.print(f"\n [yellow]→ 処理中: {file_path.name}[/]")
# 既存のこのファイルのチャンクを削除（更新処理）
try:
self.vectorstore._collection.delete(
where={"source_path": str(file_path)}
)
except Exception:
pass
# 文書を読み込み → チャンク化 → ベクトルDB登録
try:
docs = loader.load_file(file_path)
if not docs:
stats["errors"] += 1
continue
chunks = self.split_documents(docs)
if not chunks:
continue
# ChromaDBに追加（バッチ処理で高速化）
batch_size = 50
for i in range(0, len(chunks), batch_size):
batch = chunks[i:i + batch_size]
self.vectorstore.add_documents(batch)
indexed_files[file_key] = current_hash
stats["added"] += 1
stats["total_chunks"] += len(chunks)
console.print(
f" [green]✓ {len(chunks)}チャンク登録[/]"
)
except Exception as e:
console.print(f" [red]✗ エラー: {e}[/]")
stats["errors"] += 1
self._save_indexed_files(indexed_files)
console.print(f"""
╔══════════════════════════════════╗
║ インデックス構築完了 ║
║ 追加: {stats[‘added’]:>5}ファイル ║
║ スキップ: {stats[‘skipped’]:>3}ファイル（変更なし）║
║ エラー: {stats[‘errors’]:>4}ファイル ║
║ 総チャンク数: {stats[‘total_chunks’]:>7} ║
╚══════════════════════════════════╝
RAGエンジン：検索と生成の中核
#!/usr/bin/env python3
"""
rag_engine.py
RAGの中核：文書検索 + Ollamaによる回答生成
"""
import os
from typing import List, Optional
from dataclasses import dataclass, field
from datetime import datetime
import chromadb
from langchain_ollama import OllamaEmbeddings, OllamaLLM
from langchain_community.vectorstores import Chroma
from langchain.prompts import PromptTemplate
from langchain.schema import Document
from langchain.chains import RetrievalQA
# ─── 設定 ─────────────────────────────────────────
EMBED_MODEL = os.getenv("EMBED_MODEL", "nomic-embed-text")
GENERATE_MODEL = os.getenv("GENERATE_MODEL", "qwen2.5:14b")
OLLAMA_URL = os.getenv("OLLAMA_BASE_URL", "http://localhost:11434")
CHROMA_DB_PATH = os.getenv("CHROMA_DB_PATH", "./chroma_db")
RETRIEVAL_K = int(os.getenv("RETRIEVAL_K", "5"))
COLLECTION_NAME = "company_knowledge"
@dataclass
class RAGResponse:
"""RAGの回答を格納するデータクラス"""
answer: str
sources: List[dict] = field(default_factory=list)
query: str = ""
model: str = ""
retrieval_count: int = 0
response_time_sec: float = 0.0
timestamp: str = field(
default_factory=lambda: datetime.now().isoformat()
)
# ─── システムプロンプト ────────────────────────────
RAG_PROMPT_TEMPLATE = """あなたは社内ナレッジアシスタントです。
以下の「参考文書」に基づいてのみ質問に回答してください。
重要なルール：
1. 参考文書に記載のない情報は「文書に記載がありません」と答えること
2. 回答の最後に必ず参照した文書名を明示すること
3. 曖昧な場合は断定を避け「〜と記載されています」という表現を使うこと
4. 複数の文書から情報を得た場合はすべての出典を示すこと
5. 回答は丁寧かつ簡潔にまとめること
=== 参考文書 ===
{context}
=== 参考文書ここまで ===
質問：{question}
回答："""
RAG_PROMPT = PromptTemplate(
template=RAG_PROMPT_TEMPLATE,
input_variables=["context", "question"]
)
class RAGEngine:
"""RAGエンジンの主クラス"""
def __init__(self):
# 埋め込みモデル
self.embeddings = OllamaEmbeddings(
model=EMBED_MODEL,
base_url=OLLAMA_URL
)
# ベクトルDB接続
self.chroma_client = chromadb.PersistentClient(path=CHROMA_DB_PATH)
self.vectorstore = Chroma(
collection_name=COLLECTION_NAME,
embedding_function=self.embeddings,
client=self.chroma_client,
persist_directory=CHROMA_DB_PATH
)
# 生成モデル
self.llm = OllamaLLM(
model=GENERATE_MODEL,
base_url=OLLAMA_URL,
temperature=0.3, # 事実に基づいた一貫した回答を優先
num_ctx=8192, # 長い文書コンテキストに対応
)
# Retriever（検索器）の設定
self.retriever = self.vectorstore.as_retriever(
search_type="similarity", # コサイン類似度での検索
search_kwargs={
"k": RETRIEVAL_K, # 上位k件を取得
}
)
def retrieve(
self,
query: str,
category_filter: Optional[str] = None,
k: Optional[int] = None
) -> List[Document]:
"""クエリに関連する文書チャンクを検索する"""
search_k = k or RETRIEVAL_K
if category_filter:
# カテゴリフィルタ付き検索
docs = self.vectorstore.similarity_search(
query,
k=search_k,
filter={"category": category_filter}
)
else:
docs = self.vectorstore.similarity_search(query, k=search_k)
return docs
def _format_context(self, docs: List[Document]) -> str:
"""検索結果をプロンプト用のコンテキスト文字列に整形する"""
context_parts = []
for i, doc in enumerate(docs, 1):
meta = doc.metadata
source = meta.get("source_file", "不明")
category = meta.get("category", "")
file_type = meta.get("file_type", "")
header = f"【文書{i}】{source}"
if category:
header += f"（カテゴリ：{category}）"
if file_type:
header += f"（形式：{file_type}）"
context_parts.append(f"{header}\n{doc.page_content}")
return "\n\n---\n\n".join(context_parts)
def _extract_sources(self, docs: List[Document]) -> List[dict]:
"""検索結果からソース情報を抽出する"""
sources = []
seen = set()
for doc in docs:
meta = doc.metadata
key = meta.get("source_file", "不明")
if key not in seen:
seen.add(key)
sources.append({
"file": meta.get("source_file", "不明"),
"path": meta.get("source_path", ""),
"category": meta.get("category", ""),
"type": meta.get("file_type", ""),
"indexed": meta.get("indexed_at", ""),
})
return sources
def query(
self,
question: str,
category_filter: Optional[str] = None,
stream: bool = False
) -> RAGResponse:
"""
質問を受け取り、RAGで回答を生成して返す
"""
import time
start = time.time()
# ① 関連文書の検索
retrieved_docs = self.retrieve(question, category_filter)
if not retrieved_docs:
return RAGResponse(
answer=("関連する文書が見つかりませんでした。\n"
"質問の表現を変えるか、関連する文書が"
"インデックスに登録されているかご確認ください。"),
query=question,
model=GENERATE_MODEL,
retrieval_count=0,
response_time_sec=time.time() - start
)
# ② コンテキストの整形
context = self._format_context(retrieved_docs)
# ③ プロンプトの構築
prompt = RAG_PROMPT_TEMPLATE.format(
context=context,
question=question
)
# ④ Ollamaで回答生成
answer = self.llm.invoke(prompt)
elapsed = time.time() - start
return RAGResponse(
answer=answer,
sources=self._extract_sources(retrieved_docs),
query=question,
model=GENERATE_MODEL,
retrieval_count=len(retrieved_docs),
response_time_sec=elapsed
)
def query_with_history(
self,
question: str,
history: List[dict],
category_filter: Optional[str] = None
) -> RAGResponse:
"""
会話履歴を考慮したマルチターン対話
history: [{"role": "user"|"assistant", "content": "..."}, ...]
"""
# 履歴から文脈を補強した質問を生成
if history:
history_text = "\n".join([
f"{h['role'].upper()}: {h['content'][:200]}"
for h in history[-4:] # 直近4ターンのみ参照
])
contextualized_q = (
f"これまでの会話：\n{history_text}\n\n"
f"新しい質問：{question}"
)
else:
contextualized_q = question
return self.query(contextualized_q, category_filter)
FastAPI サーバー：Webからアクセスできるようにする
#!/usr/bin/env python3
"""
api_server.py
FastAPIによるRAG WebAPIサーバー
起動: uvicorn api_server:app --host 127.0.0.1 --port 8000 --reload
"""
from fastapi import FastAPI, UploadFile, File, HTTPException, BackgroundTasks
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from typing import List, Optional
import shutil
import os
from pathlib import Path
from rag_engine import RAGEngine
from indexer import KnowledgeIndexer
app = FastAPI(
title="社内ナレッジAI API",
description="Ollamaを使ったローカルRAGシステム",
version="1.0.0"
)
# CORS設定（同一ホストからのアクセスのみ許可）
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000", "http://127.0.0.1:3000"],
allow_credentials=True,
allow_methods=["GET", "POST", "DELETE"],
allow_headers=["*"],
)
# エンジンの初期化（起動時に一度だけ）
engine = RAGEngine()
indexer = KnowledgeIndexer()
DOCUMENTS_DIR = os.getenv("DOCUMENTS_DIR", "./documents")
# ─── リクエスト/レスポンスモデル ──────────────────
class QueryRequest(BaseModel):
question: str
category: Optional[str] = None
history: Optional[List[dict]] = None
class QueryResponse(BaseModel):
answer: str
sources: List[dict]
response_time_sec: float
retrieval_count: int
class IndexResponse(BaseModel):
status: str
stats: dict
# ─── エンドポイント ────────────────────────────────
@app.get("/health")
def health_check():
"""ヘルスチェック"""
stats = indexer.get_stats()
return {
"status": "ok",
"model": os.getenv("GENERATE_MODEL", "qwen2.5:14b"),
"indexed_chunks": stats.get("total_chunks", 0)
}
@app.post("/query", response_model=QueryResponse)
async def query(req: QueryRequest):
"""
質問に対してRAGで回答する
- question: 質問文
- category: フォルダ名でフィルタ（任意）
- history: 会話履歴（任意）
"""
if not req.question.strip():
raise HTTPException(status_code=400, detail="質問が空です")
if len(req.question) > 2000:
raise HTTPException(status_code=400, detail="質問が長すぎます（2000文字以内）")
if req.history:
response = engine.query_with_history(
req.question, req.history, req.category
)
else:
response = engine.query(req.question, req.category)
return QueryResponse(
answer=response.answer,
sources=response.sources,
response_time_sec=response.response_time_sec,
retrieval_count=response.retrieval_count
)
@app.post("/upload")
async def upload_document(
file: UploadFile = File(...),
category: str = "uploaded",
background_tasks: BackgroundTasks = None
):
"""
文書ファイルをアップロードしてインデックスに追加する
"""
# サポートする拡張子の確認
allowed = {".pdf", ".docx", ".xlsx", ".txt", ".md", ".csv"}
suffix = Path(file.filename).suffix.lower()
if suffix not in allowed:
raise HTTPException(
status_code=400,
detail=f"未対応の形式: {suffix}。対応形式: {allowed}"
)
# ファイルサイズ制限（50MB）
content = await file.read()
if len(content) > 50 * 1024 * 1024:
raise HTTPException(
status_code=413, detail="ファイルサイズは50MB以下にしてください"
)
# 保存先ディレクトリの作成
save_dir = Path(DOCUMENTS_DIR) / category
save_dir.mkdir(parents=True, exist_ok=True)
save_path = save_dir / file.filename
with open(save_path, "wb") as f:
f.write(content)
# バックグラウンドでインデックスを更新
if background_tasks:
background_tasks.add_task(indexer.index_documents)
return {
"status": "uploaded",
"filename": file.filename,
"category": category,
"size_kb": round(len(content) / 1024, 1),
"message": "バックグラウンドでインデックスを更新中です"
}
@app.post("/index", response_model=IndexResponse)
async def rebuild_index(force: bool = False):
"""インデックスを（再）構築する"""
stats = indexer.index_documents(force_reindex=force)
return IndexResponse(
status="completed",
stats=stats
)
@app.get("/stats")
def get_stats():
"""インデックスの統計情報を返す"""
return indexer.get_stats()
@app.get("/categories")
def list_categories():
"""登録されているカテゴリ（フォルダ名）の一覧を返す"""
docs_path = Path(DOCUMENTS_DIR)
if not docs_path.exists():
return {"categories": []}
categories = [
d.name for d in docs_path.iterdir()
if d.is_dir() and not d.name.startswith(".")
]
return {"categories": sorted(categories)}
# サーバーの起動
uvicorn api_server:app --host 127.0.0.1 --port 8000 --reload
# 動作確認
curl http://127.0.0.1:8000/health
# 質問APIのテスト
curl -X POST http://127.0.0.1:8000/query \
-H "Content-Type: application/json" \
-d '{"question": "有給休暇の申請方法を教えてください"}'
CLIツール：ターミナルから対話する
#!/usr/bin/env python3
"""
cli.py
コマンドラインから社内ナレッジAIと対話するツール
"""
import sys
from rich.console import Console
from rich.markdown import Markdown
from rich.panel import Panel
from rich.table import Table
from rich import box
from rag_engine import RAGEngine
from indexer import KnowledgeIndexer
console = Console()
def print_response(response):
"""RAG回答を見やすく表示する"""
# 危険度に応じた色
answer_md = Markdown(response.answer)
console.print(Panel(
answer_md,
title="[bold green]💡 回答[/]",
border_style="green"
))
# ソース情報の表示
if response.sources:
table = Table(
title="📚 参照文書",
box=box.ROUNDED,
show_header=True,
header_style="bold cyan"
)
table.add_column("ファイル名", style="white", min_width=25)
table.add_column("カテゴリ", style="cyan", min_width=12)
table.add_column("形式", style="yellow", min_width=6)
for src in response.sources:
table.add_row(
src.get("file", "不明"),
src.get("category", ""),
src.get("type", "")
)
console.print(table)
console.print(
f"[dim]検索: {response.retrieval_count}件 | "
f"応答時間: {response.response_time_sec:.1f}秒 | "
f"モデル: {response.model}[/]"
)
def interactive_mode(engine: RAGEngine, category: str = None):
"""対話モード：会話を続けながら質問できる"""
history = []
console.print(Panel(
"[bold]社内ナレッジAI[/]\n"
"質問を入力してください。終了するには [bold red]exit[/] または [bold red]quit[/]",
border_style="blue"
))
if category:
console.print(f"[cyan]カテゴリフィルタ：{category}[/]\n")
while True:
try:
question = console.input("[bold blue]❓ 質問：[/] ").strip()
except (KeyboardInterrupt, EOFError):
console.print("\n[yellow]終了します[/]")
break
if question.lower() in ("exit", "quit", "終了", "q"):
console.print("[yellow]終了します[/]")
break
if not question:
continue
if question == "/clear":
history = []
console.print("[dim]会話履歴をクリアしました[/]")
continue
if question == "/stats":
indexer = KnowledgeIndexer()
console.print(indexer.get_stats())
continue
console.print("[dim]⏳ 検索・回答生成中...[/]")
response = engine.query_with_history(question, history, category)
print_response(response)
# 会話履歴を更新（最大8ターン保持）
history.append({"role": "user", "content": question})
history.append({"role": "assistant", "content": response.answer})
history = history[-16:] # 直近16メッセージ（8往復）
def main():
import argparse
parser = argparse.ArgumentParser(
description="社内ナレッジAI CLIツール"
)
subparsers = parser.add_subparsers(dest="command")
# 対話モード
chat_parser = subparsers.add_parser("chat", help="対話モードで起動")
chat_parser.add_argument("--category", "-c", help="カテゴリフィルタ")
# 単発質問
ask_parser = subparsers.add_parser("ask", help="単発の質問")
ask_parser.add_argument("question", help="質問文")
ask_parser.add_argument("--category", "-c", help="カテゴリフィルタ")
# インデックス構築
index_parser = subparsers.add_parser("index", help="インデックスを構築する")
index_parser.add_argument("--force", "-f", action="store_true",
help="強制再インデックス")
# 統計情報
subparsers.add_parser("stats", help="インデックスの統計を表示")
args = parser.parse_args()
if args.command == "index":
indexer = KnowledgeIndexer()
indexer.index_documents(force_reindex=args.force)
elif args.command == "stats":
indexer = KnowledgeIndexer()
stats = indexer.get_stats()
console.print(stats)
elif args.command == "ask":
engine = RAGEngine()
response = engine.query(args.question, args.category)
print_response(response)
elif args.command == "chat" or args.command is None:
engine = RAGEngine()
category = getattr(args, "category", None)
interactive_mode(engine, category)
else:
parser.print_help()
if __name__ == "__main__":
main()
# CLIの使い方 # インデックスの構築（最初に必ず実行） python cli.py index # 対話モードで起動 python cli.py chat # カテゴリを指定した対話モード（rulesフォルダ内のみを検索） python cli.py chat --category rules # 単発質問 python cli.py ask "出張申請の締め切りはいつですか？" # インデックスの統計確認 python cli.py stats # 強制再インデックス（全ファイルを再処理） python cli.py index --force
ファイル変更の自動監視：文書を追加したら即インデックス更新
#!/usr/bin/env python3
"""
auto_watcher.py
documentsフォルダを監視し、ファイル変更時に自動でインデックスを更新する
"""
import time
import threading
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from rich.console import Console
from indexer import KnowledgeIndexer
console = Console()
SUPPORTED_EXTENSIONS = {
".pdf", ".docx", ".xlsx", ".xls",
".txt", ".md", ".markdown", ".html", ".csv"
}
class DocumentChangeHandler(FileSystemEventHandler):
"""ファイルシステムの変更イベントを処理するハンドラー"""
def __init__(self, indexer: KnowledgeIndexer):
self.indexer = indexer
self._pending = False # デバウンス用フラグ
self._lock = threading.Lock()
def _schedule_reindex(self, filepath: str) -> None:
"""複数の変更をまとめて1回のインデックス更新に集約する（デバウンス）"""
with self._lock:
if self._pending:
return
self._pending = True
# 3秒後にインデックス更新（この間の変更もまとめて処理）
def delayed_index():
time.sleep(3)
console.print(f"\n[cyan]🔄 変更検知：インデックスを更新中...[/]")
self.indexer.index_documents()
with self._lock:
self._pending = False
thread = threading.Thread(target=delayed_index, daemon=True)
thread.start()
def _is_supported(self, path: str) -> bool:
"""対応ファイル形式かどうかを確認する"""
p = Path(path)
return (
p.suffix.lower() in SUPPORTED_EXTENSIONS
and not p.name.startswith(".")
and "~$" not in p.name
)
def on_created(self, event):
if not event.is_directory and self._is_supported(event.src_path):
console.print(f"[green]+ ファイル追加: {Path(event.src_path).name}[/]")
self._schedule_reindex(event.src_path)
def on_modified(self, event):
if not event.is_directory and self._is_supported(event.src_path):
console.print(f"[yellow]✎ ファイル更新: {Path(event.src_path).name}[/]")
self._schedule_reindex(event.src_path)
def on_deleted(self, event):
if not event.is_directory and self._is_supported(event.src_path):
console.print(f"[red]- ファイル削除: {Path(event.src_path).name}[/]")
self._schedule_reindex(event.src_path)
def start_watcher(documents_dir: str = "./documents") -> None:
"""ファイル監視を開始する（ブロッキング）"""
indexer = KnowledgeIndexer()
handler = DocumentChangeHandler(indexer)
observer = Observer()
observer.schedule(handler, documents_dir, recursive=True)
observer.start()
console.print(f"[bold green]👁 ファイル監視開始: {documents_dir}[/]")
console.print("[dim]停止するには Ctrl+C を押してください[/]")
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
console.print("\n[yellow]ファイル監視を停止しました[/]")
observer.join()
if __name__ == "__main__":
start_watcher()
Open WebUI との連携：ブラウザからチャット感覚で使う
Open WebUI を使えば、ChatGPT そっくりの UI でローカル RAG にアクセスできます。Docker が入っていれば数コマンドで起動します。
# Open WebUI の起動（Ollamaと同じホストで動かす場合） docker run -d \ -p 127.0.0.1:3000:8080 \ --add-host=host.docker.internal:host-gateway \ -e OLLAMA_BASE_URL=http://host.docker.internal:11434 \ -v open-webui:/app/backend/data \ --name open-webui \ --restart unless-stopped \ ghcr.io/open-webui/open-webui:main # ブラウザで http://localhost:3000 にアクセス
Open WebUI の RAG 機能を使う（組み込み機能）
Open WebUI 自体にも RAG 機能（文書アップロード・検索）が組み込まれています。Ollama の埋め込みモデルと組み合わせることで、コードなしで基本的な RAG が構成できます。
# Open WebUI で Ollama 埋め込みモデルを設定する手順 # （UIから設定する場合の概要） # # 1. http://localhost:3000 にアクセスしてアカウント作成 # 2. 設定 → 管理者パネル → 文書 # 3. 埋め込みモデルプロバイダー：Ollama を選択 # 4. モデル：nomic-embed-text を選択 # 5. チャンクサイズ：800、オーバーラップ：100 に設定 # 6. 保存 # # 文書の追加： # 1. 左サイドバーの「ワークスペース」→「ナレッジ」 # 2. 「新しいナレッジを作成」 # 3. PDF/Wordファイルをドラッグ&ドロップ # 4. チャット画面で「+」→「ナレッジ」から参照して質問
Slack Bot との連携：チャンネルで質問できるようにする
#!/usr/bin/env python3
"""
slack_bot.py
Slackから社内ナレッジAIに質問できるボット
slack_bolt と slack_sdk のインストールが必要：
pip install slack-bolt slack-sdk
"""
import os
import threading
from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler
from rag_engine import RAGEngine
SLACK_BOT_TOKEN = os.getenv("SLACK_BOT_TOKEN") # xoxb-...
SLACK_APP_TOKEN = os.getenv("SLACK_APP_TOKEN") # xapp-...
app = App(token=SLACK_BOT_TOKEN)
engine = RAGEngine()
# ─── ボットへのメンション ─────────────────────────
@app.event("app_mention")
def handle_mention(event, say, client):
"""ボットへのメンションを処理する"""
# ボットのメンション部分を除いた質問テキストを取得
text = event.get("text", "")
# <@UXXXXXXXX> の部分を除去
import re
question = re.sub(r"<@[A-Z0-9]+>", "", text).strip()
if not question:
say("こんにちは！質問があればメンションしてください 😊")
return
# 処理中メッセージ
thinking_msg = say(f"⏳ 「{question[:30]}...」を検索中です...")
thread_ts = event.get("ts")
def process_and_reply():
response = engine.query(question)
# ソース情報の整形
sources_text = ""
if response.sources:
sources_list = [
f"• {s['file']}（{s['category']}）"
for s in response.sources
]
sources_text = "\n".join(sources_list)
blocks = [
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"*💡 回答*\n{response.answer}"
}
}
]
if sources_text:
blocks.append({
"type": "context",
"elements": [{
"type": "mrkdwn",
"text": f"📚 *参照文書:*\n{sources_text}"
}]
})
blocks.append({
"type": "context",
"elements": [{
"type": "mrkdwn",
"text": (f"🔍 {response.retrieval_count}件の文書を参照 | "
f"⏱ {response.response_time_sec:.1f}秒")
}]
})
client.chat_update(
channel=event["channel"],
ts=thinking_msg["ts"],
text=response.answer,
blocks=blocks
)
# 別スレッドで処理（Slackの3秒タイムアウトを回避）
thread = threading.Thread(target=process_and_reply, daemon=True)
thread.start()
# ─── ホームタブ ──────────────────────────────────
@app.event("app_home_opened")
def update_home_tab(client, event):
"""ボットのホームタブにヘルプ情報を表示する"""
client.views_publish(
user_id=event["user"],
view={
"type": "home",
"blocks": [
{
"type": "header",
"text": {"type": "plain_text", "text": "🤖 社内ナレッジAI"}
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": ("*使い方*\n"
"チャンネルでボットをメンションして質問するだけです。\n\n"
"`@ナレッジBot 有給休暇の申請方法を教えてください`\n\n"
"*特徴*\n"
"• すべての処理はローカルで完結（情報は外部送信されません）\n"
"• 社内文書に基づいた正確な回答\n"
"• 参照した文書名を必ず明示")
}
}
]
}
)
if __name__ == "__main__":
handler = SocketModeHandler(app, SLACK_APP_TOKEN)
print("⚡ Slack Botを起動しました")
handler.start()
精度向上のテクニック：RAGをチューニングする
チャンクサイズの最適化
RAG の精度に最も影響するパラメータの一つがチャンクサイズです。文書の性質に合わせて調整します。
| 文書の種類 | 推奨チャンクサイズ | オーバーラップ | 理由 |
|---|---|---|---|
| FAQ・Q&A形式 | 400〜600文字 | 50文字 | 1つの質問と回答が完結するサイズ |
| 規則・マニュアル | 700〜1000文字 | 100文字 | 条文や手順が文脈を含む必要があるため |
| 議事録・レポート | 800〜1200文字 | 150文字 | 話題の区切りをまたぐコンテキストが重要 |
| 技術仕様書 | 600〜900文字 | 100文字 | 専門用語の文脈を保持するため |
ハイブリッド検索の実装
ベクトル検索（意味的類似度）だけでなく、キーワード検索（BM25）と組み合わせることで、固有名詞・型番・日付などの完全一致検索を強化できます。
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever
def create_hybrid_retriever(vectorstore, all_chunks):
"""
ハイブリッド検索：ベクトル検索 + BM25キーワード検索
"""
# ベクトル検索Retriever（意味的類似度）
vector_retriever = vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 5}
)
# BM25 Retriever（キーワード完全一致）
bm25_retriever = BM25Retriever.from_documents(all_chunks)
bm25_retriever.k = 5
# アンサンブル（重み：ベクトル0.6、BM25 0.4）
ensemble_retriever = EnsembleRetriever(
retrievers=[vector_retriever, bm25_retriever],
weights=[0.6, 0.4]
)
return ensemble_retriever
メタデータフィルタリングの活用
# カテゴリ・更新日・ファイル形式でフィルタリングする例
# 規則文書のみを検索
results = engine.vectorstore.similarity_search(
"残業の上限時間",
filter={"category": "rules"},
k=5
)
# 複数条件のフィルタ（ChromaDB形式）
results = engine.vectorstore.similarity_search(
"プロジェクトの進捗",
filter={
"$and": [
{"category": {"$in": ["projects", "reports"]}},
{"file_type": {"$in": ["PDF", "DOCX"]}}
]
},
k=5
)
回答品質の評価スクリプト
#!/usr/bin/env python3
"""
RAG の回答品質を定量的に評価するスクリプト
"""
from rag_engine import RAGEngine
# 評価用の質問と期待される回答キーワード
EVAL_CASES = [
{
"question": "有給休暇は何日付与されますか？",
"expected_keywords": ["有給", "日", "付与", "年次"],
"expected_source_contains": "就業規則"
},
{
"question": "出張費の精算期限はいつですか？",
"expected_keywords": ["精算", "出張", "日以内", "申請"],
"expected_source_contains": "経費"
},
]
def evaluate_rag(engine: RAGEngine) -> dict:
"""RAGシステムの精度を評価する"""
results = []
for case in EVAL_CASES:
response = engine.query(case["question"])
# キーワードの存在確認
answer_lower = response.answer.lower()
keyword_hits = sum(
1 for kw in case["expected_keywords"]
if kw in response.answer
)
keyword_score = keyword_hits / len(case["expected_keywords"])
# ソースファイルの確認
source_names = [s["file"] for s in response.sources]
source_hit = any(
case["expected_source_contains"] in name
for name in source_names
)
results.append({
"question": case["question"],
"keyword_score": keyword_score,
"source_correct": source_hit,
"response_time": response.response_time_sec
})
# 集計
avg_keyword = sum(r["keyword_score"] for r in results) / len(results)
source_accuracy = sum(1 for r in results if r["source_correct"]) / len(results)
avg_time = sum(r["response_time"] for r in results) / len(results)
print(f"キーワードスコア：{avg_keyword*100:.1f}%")
print(f"ソース正答率：{source_accuracy*100:.1f}%")
print(f"平均応答時間：{avg_time:.1f}秒")
return {
"keyword_score": avg_keyword,
"source_accuracy": source_accuracy,
"avg_time": avg_time,
"details": results
}
if __name__ == "__main__":
engine = RAGEngine()
evaluate_rag(engine)
セキュリティと運用設計
アクセス制御の実装
# api_server.py にAPIキー認証を追加する例
from fastapi import Security, HTTPException, status
from fastapi.security import APIKeyHeader
import secrets
API_KEY_HEADER = APIKeyHeader(name="X-API-Key", auto_error=False)
# 有効なAPIキーのリスト（.envで管理）
VALID_API_KEYS = set(
os.getenv("API_KEYS", "").split(",")
)
def verify_api_key(api_key: str = Security(API_KEY_HEADER)):
if not api_key or api_key not in VALID_API_KEYS:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="有効なAPIキーが必要です"
)
return api_key
# エンドポイントに認証を追加
@app.post("/query", response_model=QueryResponse)
async def query(
req: QueryRequest,
api_key: str = Security(verify_api_key) # ← 追加
):
...
# .env にAPIキーを設定
# API_KEYS=key1xxxx,key2xxxx,key3xxxx
# APIキーの生成
# python -c "import secrets; print(secrets.token_hex(32))"
ログの記録と監査
import logging
import json
from datetime import datetime
# 構造化ログの設定
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s %(levelname)s %(message)s',
handlers=[
logging.FileHandler("logs/rag_audit.log", encoding="utf-8"),
logging.StreamHandler()
]
)
audit_logger = logging.getLogger("audit")
# クエリの記録（個人情報に注意して最小限のログにする）
def log_query(question: str, response, user_id: str = "anonymous"):
audit_logger.info(json.dumps({
"timestamp": datetime.now().isoformat(),
"user_id": user_id,
"question_len": len(question), # 質問内容は記録せず長さのみ
"verdict": response.retrieval_count,
"sources": [s["file"] for s in response.sources],
"response_time": response.response_time_sec
}, ensure_ascii=False))
推奨スペックとスケーリング指針
| 規模 | 文書数目安 | 推奨スペック | 同時接続数 |
|---|---|---|---|
| 小規模（個人・小チーム） | 〜500ファイル | RAM 16GB、qwen2.5:7b | 1〜3名 |
| 中規模（部門単位） | 〜5,000ファイル | RAM 32GB、RTX 3090、qwen2.5:14b | 5〜20名 |
| 大規模（全社） | 〜50,000ファイル | RAM 64GB+、A100 GPU、複数インスタンス | 50名以上 |
まとめ：社内ナレッジAIの効果と展望
本記事では、OllamaとChromaDBを組み合わせたローカルRAGシステムをゼロから構築しました。完成したシステムは以下を実現しています。
- 完全なデータプライバシー：社内文書の内容が外部に出ることは一切なし
- ランニングコストゼロ：API料金が発生しないローカル動作
- 出典付きの正確な回答：どの文書から回答したかを必ず明示
- 柔軟な拡張性：Slack / WebUI / API と自由に連携可能
- 自動更新：文書を追加するだけで知識ベースが更新される
今後の発展として特に注目される方向性は、マルチモーダル対応（図や表を含むPDFの解析）、グラフRAG（文書間の関係性を知識グラフで表現する手法）、そしてエージェント型RAG（AIが自律的に複数回検索してより深い回答を生成する手法）です。
クラウドAIへの依存から脱却し、自社の知識資産を自社でコントロールする時代が始まっています。本記事で構築したシステムを起点に、自社のニーズに合わせてカスタマイズしていただければ幸いです。
✅ 導入チェックリスト
- ☐ Ollamaのインストールと
nomic-embed-text、qwen2.5:14bのダウンロード - ☐ Pythonライブラリのインストール（langchain、chromadb 等）
- ☐
documents/フォルダへの社内文書の配置 - ☐
python cli.py indexでインデックス構築 - ☐
python cli.py chatで動作確認 - ☐ FastAPIサーバーの起動と
/healthでの確認 - ☐ Open WebUI または Slack Bot との連携設定
- ☐ APIキー認証の設定（複数人で使う場合）
- ☐ ファイル自動監視（
auto_watcher.py）のサービス化 - ☐ 定期的な回答品質の評価と文書の見直し
※ 本記事の情報は2025年3月時点のものです。LangChain・ChromaDB・Open WebUIなどはバージョンアップが頻繁です。最新のAPIは各公式ドキュメントをご確認ください。サンプルコードは動作確認の参考として提供しています。本番環境での使用には追加のセキュリティ設計・テストを実施してください。
コメント