2
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@skillogy(久保田 勇一)inClickHouse株式会社

ClickHouse×MCPで「学び続けるAIエージェント」を作る【後編】

Last updated at Posted at 2026-01-26

前編ではログ用MCPサーバを作り、オーケストレータからClickHouseへイベントを強制記録する仕組みを構築しました。後編では、貯まったログを「使える資産」に変えていきます。

後編のゴール

後編で実装するのは以下の最小ループです。

  1. knowledge生成 — eventsから有用な情報を抽出し、検索可能な形で保存
  2. RAG検索 — 会話中にknowledgeを引いて回答品質を上げる
  3. 評価記録 — ユーザーの反応や検索精度を自動で記録
  4. 設定更新 — 評価結果に基づきRAGパラメータを調整

このループが回れば、AIエージェントは「使うほど賢くなる」状態になります。

テーブル追加（knowledge / retrieval_metrics）

前編で作ったeventsテーブルに、2つのテーブルを追加します。

テーブル 役割
events 生ログ（前編で作成済み）
knowledge eventsから抽出したナレッジ
retrieval_metrics RAG検索の評価データ 
events（生ログ）
    ↓ 抽出
knowledge（構造化ナレッジ）
    ↓ 検索・利用
retrieval_metrics（評価）
    ↓ 分析
RAG設定更新

DDL例：knowledge

CREATE TABLE IF NOT EXISTS knowledge (
    knowledge_id       UUID DEFAULT generateUUIDv4(),
    text               String,                           -- ナレッジ本文
    tags               Array(String) DEFAULT [],         -- 分類タグ
    embedding          Array(Float32) DEFAULT [],        -- 埋め込み（オプション）
    source_event_ids   Array(String) DEFAULT [],         -- 元イベントID群
    source_type        Enum8(                            -- ナレッジの出典
        'conversation' = 1,
        'tool_output' = 2,
        'user_feedback' = 3,
        'document' = 4
    ) DEFAULT 'conversation',
    created_at         DateTime64(3) DEFAULT now64(3),
    updated_at         DateTime64(3) DEFAULT now64(3)
)
ENGINE = MergeTree()
ORDER BY (created_at, knowledge_id)
TTL toDateTime(created_at) + INTERVAL 365 DAY
SETTINGS index_granularity = 8192;

設計ポイント

  • embeddingはオプション。まずはtext検索で始め、必要なら追加
  • source_event_idsで出典を追跡。問題時の調査や重複防止に使用
  • source_typeで抽出元の種別を記録（会話、ツール出力、フィードバック等）

DDL例：retrieval_metrics

CREATE TABLE IF NOT EXISTS retrieval_metrics (
    metric_id          UUID DEFAULT generateUUIDv4(),
    trace_id           String,
    turn_id            String,
    query              String,
    retrieved_ids      Array(String) DEFAULT [],
    retrieved_scores   Array(Float32) DEFAULT [],

    -- 評価指標
    user_feedback      Enum8(
        'none' = 0,
        'positive' = 1,
        'negative' = 2,
        'regenerated' = 3
    ) DEFAULT 'none',
    accepted           UInt8 DEFAULT 0,
    regenerated        UInt8 DEFAULT 0,
    citation_coverage  Float32 DEFAULT 0.0,

    -- RAG設定スナップショット
    rag_config_version String DEFAULT '',
    top_k              UInt8 DEFAULT 5,

    created_at         DateTime64(3) DEFAULT now64(3)
)
ENGINE = MergeTree()
ORDER BY (created_at, trace_id, turn_id)
TTL toDateTime(created_at) + INTERVAL 90 DAY
SETTINGS index_granularity = 8192;

評価指標の取得方法

  • user_feedback: 明示的なフィードバックUI（Enum8で状態管理）
  • accepted: 会話継続パターンから推定（「ありがとう」→1、「違う」→0）
  • regenerated: 再生成リクエストの有無
  • citation_coverage: 回答と検索結果の照合

実装例：knowledge生成ワーカー

eventsから定期的にナレッジを抽出するワーカーです。

処理フロー

1. チェックポイント読み込み（前回処理位置）
2. events増分取得
3. 抽出対象の判定（ツール結果、長い応答など）
4. LLMで要点抽出
5. knowledge INSERT
6. チェックポイント更新

核心部分

def extract_knowledge(self, events: list[dict]) -> list[dict]:
    """イベント群から要点を抽出"""
    prompt = ChatPromptTemplate.from_messages([
        ("system", """あなたはAIエージェントの会話ログから、将来の回答に役立つナレッジを抽出するエキスパートです。

抽出基準
- ツール実行結果から得られた事実情報
- ユーザーからのフィードバックや訂正
- 成功したタスク完了のパターン
- エラーからの学び

出力形式（JSON配列）
[{{"text": "抽出したナレッジ", "tags": ["タグ"], "source_type": "conversation|tool_output|user_feedback"}}, ...]

ナレッジがない場合は空配列 [] を返してください。"""),
        ("user", "イベントログ:\n{events_text}")
    ])

    events_text = "\n".join([
        f"[{e['event_type']}] {e.get('tool_name', '')}: {e['content'][:500]}"
        for e in events
    ])

    chain = prompt | self.llm | JsonOutputParser()
    return chain.invoke({"events_text": events_text})

ポイント

  • チェックポイント方式で障害時も再開可能
  • LangChainのJsonOutputParserでJSON配列を直接パース
  • 軽量モデル（gpt-4o-mini等）でコスト管理

実装例：RAG検索

検索戦略（段階的に高度化）

フェーズ 方式 特徴
Phase 1 タグ・キーワード検索 実装が簡単、高速、説明可能
Phase 2 ベクトル類似検索 意味的な類似を捉える
Phase 3 ハイブリッド検索 キーワード＋ベクトルの組み合わせ

まずはPhase 1から始めるのが現実的です。

キーワード検索の例

def search_by_keywords(self, query: str, top_k: int = 5) -> list[dict]:
    # キーワード抽出とサニタイズ（SQLインジェクション対策）
    keywords = [kw.replace("'", "''") for kw in query.split()[:10] if kw.strip()]

    if not keywords:
        score_expr = "0"  # 空クエリ時はスコア0
    else:
        score_expr = " + ".join([f"multiSearchAnyCaseInsensitive(text, ['{kw}'])" for kw in keywords])

    sql = f"""
        SELECT knowledge_id, text, tags, source_type, ({score_expr}) as score
        FROM knowledge
        ORDER BY score DESC
        LIMIT {top_k}
    """
    return self.execute(sql)

ベクトル検索の例

def similarity_search(self, query: str, top_k: int = 5, threshold: float = 0.7) -> list[dict]:
    """類似検索を実行"""
    query_embedding = self.embeddings.embed_query(query)

    result = self.client.execute(f"""
        SELECT knowledge_id, text, tags, source_type,
               1 - cosineDistance(embedding, %(query_embedding)s) as similarity
        FROM knowledge
        WHERE length(embedding) > 0
        ORDER BY similarity DESC
        LIMIT %(top_k)s
    """, {"query_embedding": query_embedding, "top_k": top_k})

    # 閾値でフィルタリング
    return [row for row in result if row[4] >= threshold]

注意: embeddingの次元はknowledgeテーブルとクエリで揃える必要があります（例: text-embedding-3-smallは1536次元）。次元が異なるとcosineDistanceがエラーになります。パラメータ化クエリを使用してSQLインジェクションを防止しています。

評価の自動記録（3指標）

RAGの品質を継続的に測定するため、3つの指標から始めます。

指標 計算方法 用途
accept率 countIf(accepted = 1) / count() 回答の受容度
再生成率 countIf(regenerated = 1) / count() 不満の検出（低いほど良い）
引用カバレッジ 回答中で検索結果が使われた割合 検索精度の評価

ダッシュボード用クエリ

SELECT
    toDate(created_at) as date,
    count() as total,
    -- 分母0対策: NULLIFで0をNULLに変換し、0除算を回避
    countIf(accepted = 1) / nullIf(count(), 0) as accept_rate,
    countIf(regenerated = 1) / nullIf(count(), 0) as regeneration_rate,
    avg(citation_coverage) as avg_coverage,
    rag_config_version
FROM retrieval_metrics
WHERE created_at >= now() - INTERVAL 30 DAY
GROUP BY date, rag_config_version
ORDER BY date DESC;

設定の更新

評価結果に基づいてRAG設定を調整します。小さく動かすが鉄則です。

RAG設定テーブル

CREATE TABLE IF NOT EXISTS rag_config (
    config_id            UUID DEFAULT generateUUIDv4(),
    version              String,
    top_k                UInt8 DEFAULT 5,
    similarity_threshold Float32 DEFAULT 0.7,
    tag_filter           Array(String) DEFAULT [],
    prompt_template      String DEFAULT '',
    is_active            UInt8 DEFAULT 0,
    created_at           DateTime64(3) DEFAULT now64(3),

    -- メタデータ
    description          String DEFAULT '',
    created_by           String DEFAULT 'system'
)
ENGINE = ReplacingMergeTree(created_at)
ORDER BY (version)
SETTINGS index_granularity = 8192;

設計意図: is_active = 1は常に1件だけにします。設定更新時は旧設定をis_active = 0にしてから新設定をINSERTしてください。読み込み側はWHERE is_active = 1 ORDER BY created_at DESC LIMIT 1で取得し、万が一複数存在しても最新1件を使用します。ReplacingMergeTreeにより同一versionの重複が自動的にマージされます。

更新ルール例

条件 アクション
accept率 < 0.6 top_k を +1（最大10）
再生成率 > 0.2 similarity_threshold を +0.05
ヒット数 < 1 similarity_threshold を -0.05（最小0.5）

運用メモ

TTL設計

テーブル 推奨TTL 理由
events 90日 生ログは容量大。必要情報はknowledgeに抽出済み
knowledge 365日 ナレッジは長期的な資産だが、古すぎる情報は削除
retrieval_metrics 90日 傾向分析に十分
rag_config 無期限 設定履歴は残す（ReplacingMergeTreeで重複排除）

その他の注意点

  • PIIマスキング: knowledge生成時にメール・電話番号・カード番号を除去
  • プロンプトバージョン管理: 抽出時のモデル名とプロンプトハッシュを記録
  • embeddingコスト: OpenAI text-embedding-3-smallで10万件≒$1。まずはキーワード検索で始める

検証結果：PoCで改善効果を確認

本記事のシステムが実際に機能するか、ローカル環境でPoCを実施しました。

検証環境

  • ClickHouse（OSS版）: Docker (clickhouse/clickhouse-server:24.1)
  • LLM: ローカルLLM (Ollama, gpt-oss:20b) + OpenAI API (gpt-4o-mini)
  • Embedding: OpenAI text-embedding-3-small (1536次元)

検証手順

Step 1: 会話データの生成

ローカルLLMと34会話×3ターン＝約100往復の会話を自動生成し、eventsテーブルに記録しました。会話トピックは技術質問、エラー対処、設計相談など多様なものを用意しました。

総イベント数: 306
ユーザー入力: 102
LLMレスポンス: 102
会話数: 34

Step 2: ナレッジ抽出

knowledge_workerで会話からナレッジを自動抽出しました。LLM（gpt-4o-mini）が会話を分析し、再利用可能な知識を判定・抽出します。

抽出されたナレッジ: 58件
- tool_output由来: 47件
- その他: 11件

100往復の会話から58件のナレッジが抽出されました。抽出率は約57%で、LLMが「再利用価値がない」と判断した会話はスキップされています。

Step 3: 検索精度の確認

ベクトル検索の精度を確認しました。

クエリ 最高類似度 ヒット数
APIエラーの対処法 0.70 3件
データベースのインデックス 0.73 3件
Dockerとは 0.65 2件
コードレビューのポイント 0.72 3件

会話から抽出したナレッジは、単発で投入したサンプルデータ（類似度0.5程度）より高い精度でヒットしました。これは「質問→回答」のペアが文脈として保持されているためです。

Step 4: RAGあり/なしの比較

同じ質問に対するRAGあり/なしの回答を比較しました。

Q: データベースのインデックスはいつ作るべき？

【RAGなし】
データベースのインデックスは、特定の条件や状況に応じて作成するべきです。
以下のような場合にインデックスを作成することを検討すると良いでしょう。
1. 検索性能の向上: 大量のデータがあるテーブルで...
（一般的で冗長な説明が続く）

【RAGあり】取得文書: 1件
  [1] 類似度: 0.72
  データベースインデックスは、検索、結合、ソート、集計の際に作成すべきである。

--- 回答 ---
データベースのインデックスは、以下のような状況で作成すべきです。
1. 検索: 特定のデータを迅速に取得する必要がある場合
2. 結合: 複数のテーブルを結合する際
3. ソート: データを特定の順序で表示する必要がある場合
4. 集計: 集計関数を使用するクエリにおいて

RAGありの回答は、ナレッジの「検索、結合、ソート、集計」という4分類をそのまま採用し、簡潔で一貫性のある回答になっています。

Step 5: 設定調整の効果

similarity_thresholdを0.7→0.65に調整した結果を確認しました。

クエリ 閾値0.7 閾値0.65
APIエラーの対処法 0件 2件
インデックスはいつ作るべき？ 1件 1件
Dockerコンテナとは？ 0件 1件

閾値を下げることで、より多くのナレッジを回答に活用できるようになりました。設定変更はrag_configテーブルに記録され、バージョン管理されています（v1.0.0 → v1.0.1 → v1.0.2）。

検証で得られた知見

1. 会話からの抽出は精度が高い

単発のドキュメントより、会話の文脈を持つナレッジの方が検索精度が高くなります。「質問→回答」のペアが自然言語として類似性を持つためです。

2. 閾値のチューニングは必須

初期値0.7では厳しすぎて、有用なナレッジがヒットしないケースがありました。0.65程度に下げることで、精度と網羅性のバランスが取れます。

3. ナレッジの量が重要

5件のサンプルデータでは類似度0.5程度でしたが、58件に増やすと0.7以上でヒットするようになりました。ナレッジの量が一定数を超えると、検索精度が大きく向上します。

4. 改善ループの可視化

retrieval_metricsテーブルに検索結果が記録されるため、「どのクエリでナレッジが使われたか」「類似度の分布はどうか」を後から分析できます。これが継続的な改善の基盤になります。

まとめ

前後編を通じて構築したのは以下の最小ループです。

ユーザー会話
    ↓
オーケストレータ ──→ events（前編）
    ↓
knowledge生成ワーカー ──→ knowledge（後編）
    ↓
RAG検索 ──→ retrieval_metrics（後編）
    ↓
設定更新ワーカー ──→ rag_config（後編）
    ↓
改善されたRAG検索 ←─┘

このループが回れば、後は改善の自動化範囲を広げていくだけです。

次のステップ例

  • ナレッジの自動分類・階層化
  • A/Bテストによる設定最適化
  • ユーザーセグメント別の設定
  • 異常検知（急激な品質低下のアラート）

最小構成から始めて、必要に応じて拡張する。それがこのシステムの設計思想です。

2
1
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
Sign upLogin
2
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?