0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

PM/EMでもClaude Codeに任せてローカルRAGを構築できた話 — 作業ログの横断検索

0
Posted at

はじめに

以前の記事(前編 / 後編)で、Claude Codeの作業記録を自動で蓄積し、改善提案を生成する仕組みを紹介しました。

本記事は、蓄積したログの別の活用方法です。RAG(Retrieval-Augmented Generation)は以前から気になっていたものの、プロジェクトで扱う機会がありませんでした。はっきりとした目的があったわけではなく、せっかくデータが溜まっていたので活用するならこのあたりかな、くらいの温度感で構築してみました。

私はPM/EMで、普段コードを書く仕事はしていません。RAGは経験として一度やっておきたかったこともあり、構築はClaude Codeに任せて取り組みました。構築自体は1〜2日で、データ量によってはインデックス構築にさらに時間がかかります。この記事では、読者が同じものを作れるレベルでコードと設定を出します。

この記事を読んで得られること:

  • ローカル環境でClaude Codeの作業ログを意味検索できるMCPサーバーの作り方
  • 技術スタックの選定理由と設計判断
  • そのまま使えるコードと設定
  • grepとの違いがわかる検索の実例

何を作ったか

Claude Codeの作業ログをプロジェクト横断で意味検索できるMCPサーバーを作りました。

検索対象

データ 形式 説明
transcript JSONL Claude Codeの会話履歴
worklog Markdown 作業記録
蓄積層 Markdown ナレッジ蓄積
会議文字起こし Markdown 会議メモ

現在、12プロジェクト + 5会議体、合計約45万チャンクをインデックス化しています。元データはtranscript(JSONL)だけで数百MB規模です。

grepとの違い

transcriptファイルを直接grep検索することはしていたのですが、ワードをはっきり覚えていないと検索が難しいという問題がありました。ベクトル検索なら、キーワードが一致しなくても意味が近ければヒットします。具体的な検索例は「検索してみる」のセクションで紹介します。

構成

[Claude Code] → MCP → [knowledge-search サーバー]
                           ↓
                    [LanceDB(ベクトルDB)]
                           ↑
                    [ingest.py(インデックス構築)]
                           ↑
           [transcript / worklog / 蓄積層 / 会議文字起こし]

提供するツール

MCPサーバーは3つのツールを提供します。Claude Codeとの会話中に、これらのツールが自動的に呼び出されます。

ツール 機能
search_knowledge 自然言語クエリでログを横断検索する
get_context 検索結果の前後のチャンクを取得する。検索結果は約800文字のチャンク単位で返るため、1チャンクだけでは経緯がわからないことがある
reindex インデックスを再構築する(新しいファイルを追加したいとき等)

技術スタックと設計判断

技術スタック

埋め込みモデルはテキストをベクトル(数値の配列)に変換するもの、ベクトルDBはそのベクトルを保存・検索するためのDBです。

選定 理由
埋め込みモデル intfloat/multilingual-e5-base pip install1コマンドで導入可能。日本語対応。ローカル完結
ベクトルDB LanceDB Rust実装で軽量。ローカルで動作し、外部サーバー不要。差分追加可能
Claude Code統合 MCPサーバー(公式mcpパッケージ) Claude Code標準の統合方式
チャンク化 Markdown: ヘッダーベース分割 → 再帰的文字分割。JSONL: メッセージ単位 → 再帰的文字分割

ローカル完結にしたのは、はっきりとした目的があるわけではないので、単純に費用の問題です。お金を払ってまで試すような温度感ではありませんでした。

設計判断

構築中のClaude Codeとのやりとりの中で、いくつかの判断をしています。

1. tool_use / tool_resultの除外

Claude Codeのtranscriptは、テキストのやりとりだけで構成されているわけではありません。ファイル読み込み、コマンド実行、コード編集といったツール呼び出しの詳細がJSON形式で大量に含まれています。

要素 中身 扱い
userのテキスト ユーザーの発言 保持
assistantのテキスト AIの応答 保持
tool_use {"type":"tool_use","name":"Edit","input":{"file_path":"...","old_string":"..."}}のようなJSON 除外
tool_result ツール実行結果(ファイル内容、コマンド出力等) 除外
thinking AIの思考過程 除外

これをそのままインデックスに入れると、JSONの断片が検索ノイズになります。埋め込みモデルは自然言語に対して最適化されているので、JSONが混ざると検索精度が落ちます。

ツールの使用結果はAIの応答テキストに反映されているので、tool_use / tool_resultを別途保持する必要はありません。除外することでDBサイズも小さくなります。

2. MCP登録はopt-in

MCPサーバーをClaude Codeに登録する方式は3通りあります。

方式 設定ファイル スコープ
プロジェクト単位 .mcp.json そのプロジェクトのみ
ユーザーグローバル ~/.claude.json 全プロジェクトで常時有効
ローカル上書き .claude/settings.local.json そのプロジェクト&そのマシン

全プロジェクトで常時有効にすると、MCPのツール定義がコンテキストを消費します。必要なプロジェクトだけで有効化するopt-in方式にしました。

3. インデックス更新は手動

自動化(セッション終了時のhookやcron)は複雑さを増やします。今回作るMCPサーバーには、検索だけでなくインデックスの再構築もツールとして含めています。Claude Codeとの会話中に「最近のファイルをインデックスに追加して」と言えば実行してくれるので、まずは手動で十分です。

構築する

ここからは実際の構築手順です。環境構築 → インデックス構築 → 検索エンジン → MCPサーバー → Claude Codeへの登録の順に進めます。

ファイル構成

knowledge-search/
├── venv/              # Python仮想環境
├── db/                # LanceDB永続化ディレクトリ
├── ingest.py          # データ取り込み・インデックス構築
├── search.py          # CLI検索(動作確認用)
└── server.py          # MCPサーバー

環境構築

mkdir knowledge-search && cd knowledge-search
python3 -m venv venv
source venv/bin/activate
pip install sentence-transformers lancedb pyarrow mcp

インデックス構築(ingest.py)

ingest.pyは以下の流れで動作します。

  1. 対象ファイル(transcript, worklog, 蓄積層, 会議文字起こし)を収集
  2. テキストをチャンク化(約800文字単位)
  3. 埋め込みベクトルを生成
  4. LanceDBに格納

transcriptの処理

transcript(JSONL)からは、userとassistantのテキストのみを抽出します。thinking、tool_use、tool_resultは除外します。

def extract_text_from_content(content) -> str:
    """テキストブロックのみ抽出。tool_use, tool_result, thinking は除外"""
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts = []
        for item in content:
            if isinstance(item, dict) and item.get("type") == "text":
                parts.append(item.get("text", ""))
        return "\n".join(p for p in parts if p)
    return str(content)

この関数を使って、transcriptの各行からテキストを抽出します。抽出したメッセージは[User] / [Assistant]のプレフィックス付きで連結し、約800文字ごとにチャンク化します。

for line in path.read_text().splitlines():
    obj = json.loads(line)
    if obj.get("type") not in ("user", "assistant"):
        continue
    text = extract_text_from_content(obj["message"]["content"])
    role = "User" if obj["type"] == "user" else "Assistant"
    messages.append(f"[{role}] {text.strip()}")

Markdownの処理

worklog、蓄積層、会議文字起こし(Markdown)は、ヘッダー(# ##等)でセクション分割した後、大きすぎるセクションを再帰的に分割します。分割の優先順位は段落区切り(\n\n)→ 改行(\n)→ 句点()→ スペースの順です。隣接チャンク間には100文字のオーバーラップを持たせています。オーバーラップがないとチャンクの境界で文脈が切れてしまい、検索でヒットすべき内容を取りこぼすことがあります。

埋め込みとDB格納

multilingual-e5-baseは、格納するテキスト(document)には"passage: "、検索クエリには"query: "というプレフィックスを付ける仕様です。これはe5モデルが「検索する側」と「検索される側」を区別して扱うための仕組みで、付け忘れると検索精度が落ちます。

from sentence_transformers import SentenceTransformer
import lancedb

model = SentenceTransformer("intfloat/multilingual-e5-base")

# e5 の仕様: document には "passage: " プレフィックス
texts = [f"passage: {r['text']}" for r in records]
embeddings = model.encode(
    texts, batch_size=64, normalize_embeddings=True, convert_to_numpy=True
)

# LanceDB に格納
db = lancedb.connect("db")
table = db.create_table("knowledge", data=records)

各チャンクには以下のメタデータを付与します。

フィールド 説明
text チャンクのテキスト
project プロジェクト名
source_file 元ファイルの相対パス
source_type transcript / worklog / accumulation / meeting
header_path Markdownのヘッダー階層(例: 設計 > 技術スタック
chunk_index ファイル内のチャンク番号
date ファイル名から抽出した日付

実行

# 全プロジェクト + 全会議体のインデックスを構築
python3 ingest.py

# 特定プロジェクトのみ
python3 ingest.py --projects project-a,project-b

# 会議文字起こしのみ
python3 ingest.py --meetings

# フルリビルド(既存インデックスを破棄して再構築)
python3 ingest.py --rebuild

差分追加に対応しているので、2回目以降は新しいファイルのみ処理されます。

検索エンジン(search.py)

CLIから検索して動作確認するためのスクリプトです。e5モデルはクエリ側に"query: "プレフィックスを付けます。

def search(query, limit=10, source_type=None, project=None):
    db = lancedb.connect("db")
    table = db.open_table("knowledge")
    model = SentenceTransformer("intfloat/multilingual-e5-base")

    # e5 の仕様: query には "query: " プレフィックス
    query_vec = model.encode(
        [f"query: {query}"], normalize_embeddings=True, convert_to_numpy=True
    )[0]

    q = table.search(query_vec.tolist()).limit(limit)
    if source_type:
        q = q.where(f"source_type = '{source_type}'")
    if project:
        q = q.where(f"project = '{project}'")
    return q.to_list()
python3 search.py "AIを助手化しようと思った動機" --limit 5
python3 search.py "体験" --source-type meeting

MCPサーバー(server.py)

公式mcpパッケージでMCPサーバーを作ります。FastMCPはMCPサーバーを簡単に作るためのクラスで、@mcp.tool()デコレータでツールを定義できます。先述の3ツール(search_knowledgeget_contextreindex)を実装します。

from mcp.server.fastmcp import FastMCP
from sentence_transformers import SentenceTransformer
import lancedb

mcp = FastMCP("knowledge-search")

# Lazy initialization(初回呼び出し時にロード)
_model = None
_table = None

def _get_model():
    global _model
    if _model is None:
        _model = SentenceTransformer("intfloat/multilingual-e5-base")
    return _model

@mcp.tool()
def search_knowledge(query: str, limit: int = 10,
                     source_type: str | None = None,
                     project: str | None = None) -> dict:
    """プロジェクト横断の意味検索"""
    model = _get_model()
    table = _get_table()
    query_vec = model.encode(
        [f"query: {query}"], normalize_embeddings=True, convert_to_numpy=True
    )[0]
    q = table.search(query_vec.tolist()).limit(limit)
    # source_type / project でフィルタ
    if source_type:
        q = q.where(f"source_type = '{source_type}'")
    if project:
        q = q.where(f"project = '{project}'")
    rows = q.to_list()
    results = []
    for r in rows:
        similarity = max(0.0, 1.0 - r.get("_distance", 0.0) / 2)
        results.append({
            "score": round(similarity, 4),
            "project": r.get("project", ""),
            "source_file": r.get("source_file", ""),
            "source_type": r.get("source_type", ""),
            "date": r.get("date", ""),
            "text": r.get("text", ""),
        })
    return {"query": query, "count": len(results), "results": results}

@mcp.tool()
def get_context(source_file: str, chunk_index: int, window: int = 3) -> dict:
    """検索結果の前後チャンクを取得"""
    table = _get_table()
    rows = table.search().where(f"source_file = '{source_file}'") \
                .select(["chunk_index", "header_path", "text"]).limit(10000).to_list()
    rows.sort(key=lambda r: r["chunk_index"])
    selected = [r for r in rows
                if chunk_index - window <= r["chunk_index"] <= chunk_index + window]
    return {"source_file": source_file, "chunks": selected}

@mcp.tool()
def reindex(projects: str | None = None, meetings: str | None = None) -> dict:
    """インデックスを再構築(ingest.py を実行)"""
    global _table
    _table = None  # 再構築後にテーブルを再読み込みさせる
    cmd = [sys.executable, str(INGEST_SCRIPT), "--rebuild"]
    if projects:
        cmd.extend(["--projects", projects])
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=21600)
    return {"status": "ok" if result.returncode == 0 else "error"}

if __name__ == "__main__":
    mcp.run()

モデルとテーブルはlazy initializationにしています。MCPサーバーの起動時にはモデルをロードせず、最初のツール呼び出し時にロードします。

Claude Codeへの登録

使いたいプロジェクトの.mcp.jsonに以下を追加します。

{
  "mcpServers": {
    "knowledge-search": {
      "command": "/path/to/knowledge-search/venv/bin/python3",
      "args": ["/path/to/knowledge-search/server.py"]
    }
  }
}

/path/to/は自分の環境に合わせて絶対パスに置き換えてください。登録後、Claude Codeを再起動するとsearch_knowledgeget_contextreindexの3ツールが使えるようになります。

検索してみる

前編・後編の記事作成で使った例

前編・後編の記事を作成する際に、実際にこの検索を使っていました。スコアは0〜1の範囲で、1に近いほど意味的に近いことを表します。体感としては0.85以上だと関連性の高い結果が返ってきます。

AI助手化プロジェクトの議論を引き出す

前編・後編で扱ったAI助手化プロジェクトの背景を書くために、当時の議論の経緯が必要でした。「AIを助手化しようと思った動機」で検索した結果です。

# スコア source_type ヒットしたテキスト(抜粋)
1 0.896 transcript 「コンテキストを、助手のようにしようとやったのですが、コンテキストだけではやりきれないことがわかったんですよね」
2 0.870 transcript 「知識の偏りもわかってくると思うので、私が知らないことを補完してくれる助手になってほしいのです」

記事作成プロジェクトから、AI助手化プロジェクトのtranscriptに対して横断検索できています。

ワードを正確に覚えていないが関連する議論を探す

記事に引用したい議論があるが、正確なワードを覚えていない場面です。「体験」で検索した結果、会議文字起こしから「経験していないことを想像するのは難しい」という議論がヒットしました。

# スコア source_type ヒットしたテキスト(抜粋)
1 0.832 meeting 「経験していないことを想像するのは難しい」

「体験」と「経験」はキーワードとしては異なりますが、意味的に近いためベクトル検索で拾えます。grepでは「体験」で検索しても「経験」を含む文はヒットしません。

get_contextで前後を確認する

検索結果は1チャンク(約800文字)単位で返ります。1発言だけだと経緯を追いきれない場合があるので、get_contextで前後のチャンクを取得できます。

search_knowledge("体験")
  → 結果の source_file, chunk_index を取得
    → get_context(source_file, chunk_index, window=2)
      → 前後2チャンクを取得

たとえば上の「体験」の検索結果(chunk_index=170)に対してget_contextを呼ぶと、以下のように前後の会話が見えます。

chunk_index テキスト(抜粋)
168 [User] 全プロジェクトを横断したいです。
169 [Assistant] 了解です。全プロジェクトのtranscriptをインデックス対象にします...
170 [User] 試しに「体験」で検索してください ← 検索でヒットしたチャンク
171 [Assistant] 結果です: score=0.832 「経験していないことを想像するのは難しい」...
172 [User] この機能に名称をつけていましたっけ?

1チャンクだけでは「体験」で検索した意図がわかりませんが、前後を見ることで「全プロジェクト横断の検索テスト中だった」という経緯がわかります。

フィルタの使い方

source_typeprojectでフィルタできます。

# transcriptのみ検索
search_knowledge("設計判断", source_type="transcript")

# 特定プロジェクトのみ
search_knowledge("テスト設計", project="work-base")

# 会議文字起こしのみ
search_knowledge("体験", source_type="meeting")

おわりに

前編で作った蓄積の仕組みがあったから、その上に検索を載せるだけで済みました。データが溜まっていれば、構築自体はClaude Codeに任せて1〜2日です。

今のところ私は、過去に自分がした作業の経緯を確認する用途で使っています。「あのときなぜこう判断したか」をすぐに引き出せるのは便利です。会議文字起こしもインデックスに入れていますが、まだ活用しきれていません。

AIはセッションをまたいで記憶を永続化できないので、過去の議論をAI自身が検索できる仕組みは、その補完にもなりえます。作業中に関連する過去の知見を自動で提示する、といった使い方も考えられそうです。

本記事のコードはそのまま使えるようにしていますので、プロジェクトのパスやインデックス対象を自分の環境に合わせて試してみてください。

参考

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?