はじめに
こんにちは。開発部の雨です。
今回は、ローカル環境で動く「長期記憶付きチャットボット」をゼロから実装する方法を紹介します。
最初はシンプルに「全会話履歴をそのままLLMに渡す」素朴な実装からスタートし、その問題点を実際に確認しながら、RAG（Retrieval-Augmented Generation）とChromaDBを使って段階的に改善していきます。最終的には、会話がどれだけ長くなっても関連情報だけを的確に引き出せる、実用的なチャットボットが完成します。
使用するツールはすべて完全ローカル・無料。APIキーは一切不要です。
問題の本質
LLMを使ったチャットボットを作るとき、多くの人が最初にこう実装します。
messages = []
while True:
user_input = input("You: ")
messages.append({"role": "user", "content": user_input})
response = call_llm(messages)
messages.append({"role": "assistant", "content": response})
シンプルで直感的ですが、これには大きな問題があります。
会話が長くなるほど、毎回すべての履歴をLLMに渡すことになる。
その結果として何が起きるか：
- トークン上限に達する — LLMは処理できるテキスト量に上限がある
- コストが増える — APIを使う場合、トークン数=料金
- 精度が下がる — コンテキストが長すぎると、LLMは重要な情報を見失う（Lost in the Middle問題）
- レスポンスが遅くなる — 処理するトークンが多いほど推論時間が増える
解決策：RAG（Retrieval-Augmented Generation）
RAGの核心的なアイデアはシンプルです。
「全部渡すのではなく、今の質問に関係する情報だけを取り出して渡す」
会話履歴をベクトルデータベースに保存しておき、新しい質問が来たときに「意味的に近いもの」だけを検索して取り出す。これによって、コンテキストウィンドウを最小限に保ちながら、関連情報を正確に参照できます。
この記事では、素朴な実装から出発して、次の5段階で少しずつ「実用的な長期記憶」に近づけていきます。
- 素朴な実装 — 全履歴を渡す（問題の出発点）
- 基本的なRAG — 関連する履歴だけを取り出す
-
記憶抽出 — 生ログではなく「重要な事実」だけを保存する
- 3.5 重複検出 — 保存前に、意味的に同じ事実がすでにないか確認する
- クエリリファクタリング — 「それ」「あれ」などの文脈依存クエリを解決する
- 記憶の更新 — 古い情報と新しい情報をタイムスタンプで区別する（簡易版）
使用技術スタック
| 役割 | ツール | 説明 |
|---|---|---|
| LLM実行 | Ollama + llama3 | ローカルでLLMを動かす |
| 埋め込み生成 | nomic-embed-text | テキストをベクトルに変換 |
| ベクトルDB | ChromaDB | ベクトルの保存と検索 |
| 言語 | Python 3.10+ | 全体の接着剤 |
すべて完全ローカル・無料で動きます。APIキー不要です。
環境構築
1. Ollamaのインストール
ollama.com からインストーラーをダウンロードするか、以下を実行します。
macOS / Linux:
curl -fsSL https://ollama.com/install.sh | sh
Windows:
公式サイトからインストーラーをダウンロード。
インストール後、以下でモデルをダウンロードします：
# チャット用LLM
ollama pull llama3
# 埋め込み用モデル
ollama pull nomic-embed-text
Ollamaが起動しているか確認：
ollama list
# NAME ID SIZE MODIFIED
# llama3:latest ... 4.7 GB ...
# nomic-embed-text:latest ... 274 MB ...
2. Pythonのインストール
Python 3.10以上が必要です。まずバージョンを確認してください：
python --version
# Python 3.11.x など 3.10以上であればOK
インストールされていない、またはバージョンが古い場合：
macOS（Homebrewを使う場合）:
brew install python@3.11
Windows / Linux:
python.org から最新版をダウンロードしてインストール。
💡 Tips: バージョン管理ツールの
pyenvを使うと複数バージョンの切り替えが楽になります。
3. Pythonライブラリのインストール
仮想環境を作ってからインストールするのがおすすめです：
# 仮想環境を作成・有効化
python -m venv rag-env
source rag-env/bin/activate # Windows: rag-env\Scripts\activate
# ライブラリをインストール
pip install chromadb requests
requestsはOllamaのREST APIを叩くために使います。
4. Ollamaの動作確認
import requests
def chat_with_ollama(messages: list[dict]) -> str:
response = requests.post(
"http://localhost:11434/api/chat",
json={
"model": "llama3",
"messages": messages,
"stream": False
}
)
return response.json()["message"]["content"]
def get_embedding(text: str) -> list[float]:
response = requests.post(
"http://localhost:11434/api/embeddings",
json={
"model": "nomic-embed-text",
"prompt": text
}
)
return response.json()["embedding"]
# テスト
print(chat_with_ollama([{"role": "user", "content": "こんにちは！"}]))
ステップ①：素朴な実装（全履歴を渡す）
まず「普通の」実装を作ります。これが後で比較の基準になります。
import requests
def chat_with_ollama(messages: list[dict]) -> str:
response = requests.post(
"http://localhost:11434/api/chat",
json={
"model": "llama3",
"messages": messages,
"stream": False
}
)
return response.json()["message"]["content"]
def naive_chatbot():
"""全会話履歴をそのまま渡すチャットボット"""
conversation_history = []
print("チャットボット起動（終了: 'quit'）\n")
while True:
user_input = input("You: ").strip()
if user_input.lower() == "quit":
break
conversation_history.append({
"role": "user",
"content": user_input
})
# 毎回、全履歴を渡す
response = chat_with_ollama(conversation_history)
conversation_history.append({
"role": "assistant",
"content": response
})
# トークン数の概算（英語の場合、1トークン≒4文字）
total_chars = sum(len(m["content"]) for m in conversation_history)
estimated_tokens = total_chars // 4
print(f"AI: {response}")
print(f"[現在の履歴: {len(conversation_history)}メッセージ, 推定{estimated_tokens}トークン]\n")
if __name__ == "__main__":
naive_chatbot()
試してみると：
You: 私の名前は田中太郎です。東京在住のエンジニアです。
AI: はじめまして、田中太郎さん！東京のエンジニアなんですね。
[現在の履歴: 2メッセージ, 推定30トークン]
You: Pythonのベストプラクティスについて教えてください。
AI: Pythonのベストプラクティスとして、型ヒントの使用、...（長い回答）
[現在の履歴: 4メッセージ, 推定800トークン]
You: （さらに20回会話を続けると...）
[現在の履歴: 44メッセージ, 推定8000トークン]
この実装の限界
なぜ全履歴渡しがダメなのか、具体的に見てみます。
問題1：トークン数の爆発
def analyze_token_growth():
"""会話が長くなるほどトークン数がどう増えるかシミュレート"""
# 各メッセージが平均200トークンと仮定
avg_tokens_per_message = 200
print("会話数 | 全履歴トークン数 | RAGトークン数（上位3件）")
print("-" * 55)
for num_messages in [5, 10, 20, 50, 100]:
naive_tokens = num_messages * avg_tokens_per_message
rag_tokens = 3 * avg_tokens_per_message # 常に上位3件のみ
print(f"{num_messages:6}件 | {naive_tokens:16,} | {rag_tokens:14,}")
analyze_token_growth()
会話数 | 全履歴トークン数 | RAGトークン数（上位3件）
-------------------------------------------------------
5件 | 1,000 | 600
10件 | 2,000 | 600
20件 | 4,000 | 600
50件 | 10,000 | 600
100件 | 20,000 | 600
RAGを使えば、会話がどれだけ長くなってもコンテキストサイズは一定に保てます。
問題2：Lost in the Middle
研究によると、LLMは長いコンテキストの最初と最後の情報は覚えていますが、中間の情報を見落としやすいことが知られています。
100回の会話の中に埋もれた「私の名前は田中太郎です」という情報を、LLMが正確に参照できる保証はありません。
ステップ②：基本的なRAG（関連する履歴だけを取り出す）
では、ChromaDBを使って「関連する会話だけを取り出す」仕組みを作ります。
まず、基本的な構造を理解しましょう：
新しい質問
↓
埋め込みベクトルに変換
↓
ChromaDBで類似検索
↓
上位N件の過去の会話を取得
↓
取得した会話を「参考情報」として渡す + 新しい質問をLLMに渡す
↓
回答を生成 → ChromaDBに保存
ここで一つ設計上のポイントがあります。取り出した過去の会話は、role 付きの会話ターンとして差し込まず、システムプロンプトの「参考情報」として渡します。 検索で取り出した断片は本来バラバラの記録であり、それを assistant / user のターンとして並べ直すと、前後関係の壊れた不自然な会話（宙に浮いた assistant 発言など）になってモデルが混乱するためです。
import requests
import chromadb
from datetime import datetime
# --- Ollama関数 ---
def chat_with_ollama(messages: list[dict]) -> str:
response = requests.post(
"http://localhost:11434/api/chat",
json={"model": "llama3", "messages": messages, "stream": False}
)
return response.json()["message"]["content"]
def get_embedding(text: str) -> list[float]:
response = requests.post(
"http://localhost:11434/api/embeddings",
json={"model": "nomic-embed-text", "prompt": text}
)
return response.json()["embedding"]
# --- ChromaDB セットアップ ---
client = chromadb.Client()
collection = client.get_or_create_collection(
name="basic_rag",
metadata={"hnsw:space": "cosine"}
)
# --- メモリ操作 ---
def save_turn(role: str, content: str, message_id: str):
"""会話ターンをそのままChromaDBに保存する"""
collection.add(
ids=[message_id],
embeddings=[get_embedding(content)],
documents=[content],
metadatas=[{"role": role, "timestamp": datetime.now().isoformat()}]
)
def retrieve_relevant_history(query: str, n_results: int = 3) -> list[str]:
"""クエリに関連する過去の会話を「参考テキスト」として取得"""
if collection.count() == 0:
return []
results = collection.query(
query_embeddings=[get_embedding(query)],
n_results=min(n_results, collection.count())
)
# 役割つきテキストの塊として返す（会話ターンには戻さない）
snippets = []
for doc, meta in zip(results["documents"][0], results["metadatas"][0]):
label = "ユーザー" if meta["role"] == "user" else "アシスタント"
snippets.append(f"{label}: {doc}")
return snippets
# --- 基本RAGチャットボット ---
def rag_chatbot_basic():
message_counter = 0
print("基本RAGチャットボット起動（終了: 'quit'）\n")
while True:
user_input = input("You: ").strip()
if user_input.lower() == "quit":
break
# 1. 関連する過去の会話を検索
relevant = retrieve_relevant_history(user_input, n_results=3)
context = "\n".join(relevant) if relevant else "（まだ履歴がありません）"
# 2. 参考情報としてシステムプロンプトに埋め込む
messages = [
{
"role": "system",
"content": f"""あなたは親切なアシスタントです。
以下は過去の会話から検索された関連性の高いやり取りです。必要なときだけ参考にしてください。
--- 過去の関連会話 ---
{context}
--- ここまで ---"""
},
{"role": "user", "content": user_input},
]
# 3. 回答を生成
response = chat_with_ollama(messages)
# 4. 今回のやり取りを保存（ユーザー発言とAI発言の両方）
message_counter += 1
save_turn("user", user_input, f"msg_{message_counter}_user")
message_counter += 1
save_turn("assistant", response, f"msg_{message_counter}_assistant")
print(f"AI: {response}")
print(f"[検索: {len(relevant)}件参照 / 総保存数: {collection.count()}件]\n")
if __name__ == "__main__":
rag_chatbot_basic()
基本RAGの限界
これで「全部渡し」からは脱却できましたが、まだ2つの問題が残っています。
- 保存しているのが生ログそのもの — LLMの回答は長く冗長で、あいさつや相槌のような「覚えなくていい情報」も全部保存されます。検索で上位N件を取り出すとき、こうしたノイズが貴重な枠を食い、埋め込みも要点がぼやけて精度が落ちます。
- 文脈依存のクエリに弱い — 「それについてもっと教えて」の「それ」が何を指すか分からないまま検索してしまいます。
この2つを、次の2ステップで順番に解決します。まずは保存する内容を改善しましょう。
ステップ③：記憶抽出 — 重要な情報だけを保存する
生ログをそのまま貯めるのではなく、LLM自身に「後で思い出す価値のある事実」だけを抽出させて保存します。これは memory extraction（記憶抽出／事実抽出） と呼ばれる手法で、Mem0・Letta(MemGPT)・Zep など実用的なメモリシステムの中核になっている考え方です。生のやり取りではなく、そこから抽出した「事実（メモリ）」を保存・検索することで、トークン量を大きく削減できることが報告されています（Mem0は素朴な全文渡しに対して約9割のトークン削減を報告）。
イメージはこうです：
You: 私の名前は田中太郎です。東京在住のエンジニアです。
↓ LLMが事実だけ抽出
保存される記憶：
- ユーザーの名前は田中太郎
- ユーザーは東京在住
- ユーザーの職業はエンジニア
（「はじめまして！」のような相槌は保存しない）
この方式のいいところは、ステップ②で悩んだ「ユーザー発言とAI発言、どちらを保存するか」という問題が自然に消えることです。抽出器は両方を読み、どちら側の発言であっても「事実」であれば残し、そうでなければ捨てます。
import requests
import chromadb
import json
from datetime import datetime
# --- Ollama関数 ---
def chat_with_ollama(messages: list[dict]) -> str:
response = requests.post(
"http://localhost:11434/api/chat",
json={"model": "llama3", "messages": messages, "stream": False}
)
return response.json()["message"]["content"]
def chat_json(prompt: str) -> dict:
"""Ollamaに必ず有効なJSONを返させる（format="json"）。前置き文の混入を防ぐ。"""
response = requests.post(
"http://localhost:11434/api/chat",
json={
"model": "llama3",
"messages": [{"role": "user", "content": prompt}],
"stream": False,
"format": "json", # ← 出力を有効なJSONに制約する
}
)
try:
return json.loads(response.json()["message"]["content"])
except (json.JSONDecodeError, KeyError):
return {}
def get_embedding(text: str) -> list[float]:
response = requests.post(
"http://localhost:11434/api/embeddings",
json={"model": "nomic-embed-text", "prompt": text}
)
return response.json()["embedding"]
# --- ChromaDB セットアップ ---
client = chromadb.Client()
collection = client.get_or_create_collection(
name="memory_extraction",
metadata={"hnsw:space": "cosine"}
)
# --- 記憶抽出 ---
def extract_memories(user_text: str, assistant_text: str) -> list[str]:
"""
1往復のやり取りから、長期的に覚えておくべき「事実」だけを抽出する。
あいさつ・相槌・一時的な雑談は捨て、後で参照したくなる情報だけを残す。
"""
prompt = f"""以下の会話から、後で思い出す価値のある「重要な事実」だけを抽出してください。
ルール：
- 事実は「ユーザーの発言」を最優先で抽出する。AIの返答はあくまで文脈の補助で、AIの言い換えでユーザーの発言の意味を上書きしない
- ユーザーに関する事実（名前・職業・居住地・好み・目標など）
- 意味と方向（肯定/否定・好き/嫌い）をそのまま正確に保つ。ぼかさない
- 「〜が変化した」のような曖昧な要約は禁止。変化したなら「変化後の内容」を具体的に書く
- AIが提示した具体的な推奨・決定も、価値があれば記録してよい
- あいさつ・相槌・一時的な雑談は抽出しない
- 重要な事実がなければ、空のリストを返す
例：
- 入力「もう犬は好きじゃない」→ {{"facts": ["ユーザーは犬が好きではない"]}}
- ❌ 悪い例：「ユーザーの犬への好みが変化した」（好き嫌いの方向が消えている）
次のJSON形式だけで出力してください（前置きや説明は禁止）：
{{"facts": ["事実1", "事実2"]}}
会話：
User: {user_text}
Assistant: {assistant_text}"""
data = chat_json(prompt)
facts = data.get("facts", [])
# 文字列だけを残し、前後の空白を除去（万一の混入に対する保険）
return [f.strip() for f in facts if isinstance(f, str) and f.strip()]
def save_memories(facts: list[str], turn_id: int) -> list[str]:
"""抽出した事実を、それぞれ独立したドキュメントとして保存する"""
saved = []
for i, fact in enumerate(facts):
collection.add(
ids=[f"mem_{turn_id}_{i}"],
embeddings=[get_embedding(fact)],
documents=[fact],
metadatas=[{"timestamp": datetime.now().isoformat()}]
)
saved.append(fact)
print(f" 💾 保存: {fact}")
return saved
# これ未満の類似度の記憶は「関連が薄い」とみなして渡さない（高いほど厳しい・要調整）
RELEVANCE_THRESHOLD = 0.6
def retrieve_relevant_memories(query: str, n_results: int = 5) -> list[str]:
if collection.count() == 0:
return []
results = collection.query(
query_embeddings=[get_embedding(query)],
n_results=min(n_results, collection.count())
)
# cosine: similarity = 1 - distance。しきい値以上のものだけ残す
relevant = []
for doc, dist in zip(results["documents"][0], results["distances"][0]):
if (1 - dist) >= RELEVANCE_THRESHOLD:
relevant.append(doc)
return relevant
# --- 記憶抽出チャットボット ---
def memory_chatbot():
turn = 0
print("記憶抽出チャットボット起動（終了: 'quit'）\n")
while True:
user_input = input("You: ").strip()
if user_input.lower() == "quit":
break
# 1. 関連する記憶を検索
memories = retrieve_relevant_memories(user_input, n_results=5)
context = "\n".join(f"- {m}" for m in memories) if memories else "（まだ記憶がありません）"
# 2. 記憶を参考情報として渡す
messages = [
{
"role": "system",
"content": f"""あなたは親切なアシスタントです。
以下はユーザーに関する記憶です。必要なときだけ参考にして回答してください。
--- 記憶 ---
{context}
--- ここまで ---"""
},
{"role": "user", "content": user_input},
]
# 3. 回答を生成
response = chat_with_ollama(messages)
# 4. 今回のやり取りから事実を抽出して保存
facts = extract_memories(user_input, response)
print(f"AI: {response}")
if facts:
print("\n📝 抽出 → 保存:")
save_memories(facts, turn)
else:
print("\n📝 今回保存する事実はありませんでした")
turn += 1
print(f"[総記憶数: {collection.count()}件]\n")
if __name__ == "__main__":
memory_chatbot()
各「事実」が短く単一トピックになるため、埋め込みがシャープになり、検索精度が上がります。生の冗長な回答を丸ごと埋め込んでいたときの「ベクトルがぼやける」問題も解消されます。
💡 抽出は必ずJSONで受け取る。 「前置きは付けないで」と指示しても、llama3 のような小さいモデルは
以下は重要な事実です：のような前置きを返しがちです。出力を行で分割して保存すると、この前置きがそのまま「事実」として保存されてしまいます。そこで Ollama のformat="json"を使って出力を{"facts": [...]}形式に強制し、facts配列だけを読み取ります。プロンプトの指示に頼らず構造で縛るので、前置きが混入しません。
💡 関連度のしきい値を入れる。
collection.queryは「上位N件」を必ず返すので、関係の薄い記憶でも枠が余っていれば紛れ込みます。無関係な記憶を渡すとモデルが混乱するので、RELEVANCE_THRESHOLD（類似度 =1 - distance）でふるいにかけ、十分に近い記憶だけを渡します。しきい値が高いほど厳しく（＝渡す件数が減る）なります。最適値は埋め込みモデルによって変わるため、実際のdistanceを見ながら調整してください。本記事では0.6をデフォルトにしています。
⚠️ トレードオフ： 抽出のために毎ターンLLM呼び出しが1回増えます（ローカルなので料金ではなく推論時間のコスト）。また、抽出は本質的にロスのある圧縮です。「さっきのコードをそのまま見せて」のような逐語的な再現は苦手なので、それが必要なら生ログも別レイヤーで保持する必要があります。
ステップ3.5：保存前の重複検出
ステップ③で保存はきれいになりましたが、まだ「同じ事実を何度も保存してしまう」問題が残っています。会話を続けると、ユーザーは同じことを何度も言います（「私はエンジニアです」を別の機会にまた言う、など）。毎回そのまま保存すると、同じ意味の記憶が増殖して検索の枠を無駄に食います。そこで保存する前に、「これ、もう似たのがDBにあるか？」をチェックし、十分に似ていればスキップします。
ただし、ここに重大な落とし穴があります。
⚠️ 埋め込みの類似度だけで判定してはいけない。
「猫が好き」と「猫が好きではない」は、埋め込み上はとても近い（同じ単語をほぼ共有し、話題も同じ）のに、意味は正反対です。類似度だけで「重複」と判断すると、好きではないという重要な変化を「同じだからスキップ」してしまい、情報が失われます。
つまり必要なのは2段構えです。埋め込みの類似度で「候補」を絞り込み（再現率）、最終的に「本当に同じ意味か」はLLMに判定させる（適合率）。 こうすれば、否定・反対・変化はちゃんと「別物」として保存されます。
新しい事実
↓
埋め込みで近い既存の記憶を検索
↓
類似度がしきい値より低い → 重複ではない → 保存
↓
類似度が高い候補がある
↓
LLMに「本当に同じ意味か？」を確認
├─ SAME（言い回し違いだけ） → スキップ
└─ DIFFERENT（否定・変化・別対象）→ 保存
# cosine space では distance = 1 - 類似度。類似度がこの値以上の候補だけLLMで確認する。
# ※ しきい値は埋め込みモデルによって最適値が違う。実際のdistanceを見て調整すること。
SIMILARITY_THRESHOLD = 0.85
def find_near_duplicates(embedding: list[float], n_results: int = 3) -> list[str]:
"""埋め込みが十分に近い既存の記憶だけを「重複候補」として返す"""
if collection.count() == 0:
return []
results = collection.query(
query_embeddings=[embedding],
n_results=min(n_results, collection.count())
)
candidates = []
for doc, dist in zip(results["documents"][0], results["distances"][0]):
similarity = 1 - dist # cosine: distanceが小さいほど似ている
if similarity >= SIMILARITY_THRESHOLD:
candidates.append(doc)
return candidates
def is_true_duplicate(new_fact: str, candidates: list[str]) -> bool:
"""
埋め込み上は近い候補が、意味として「本当に同じ」かをLLMで確認する。
★ ここが肝：否定・反対・変化（「好き」↔「好きではない」）は同じ扱いにしない。
"""
if not candidates:
return False
existing_text = "\n".join(f"- {c}" for c in candidates)
prompt = f"""新しい事実が、既存の事実のどれかと「まったく同じ意味」かどうかを判定してください。
判定ルール：
- 言い回しが違っても意味が同じなら「同じ」（例：「猫が好き」と「猫が大好き」）
- 否定・反対・変化は「同じではない」（例：「猫が好き」と「猫が好きではない」は正反対なので別物）
- 対象が違えば「同じではない」（例：「猫が好き」と「犬が好き」）
既存の事実：
{existing_text}
新しい事実：
{new_fact}
完全に同じ意味なら SAME、少しでも違う・反対・変化していれば DIFFERENT とだけ出力してください。"""
answer = chat_with_ollama([{"role": "user", "content": prompt}]).strip().upper()
return answer.startswith("SAME")
save_memories に組み込みます。埋め込みは重複チェックと保存で使い回して、計算の二度手間を避けます。
def save_memories(facts: list[str], turn_id: int) -> list[str]:
saved = []
for i, fact in enumerate(facts):
embedding = get_embedding(fact)
# --- ステップ3.5：重複チェック ---
candidates = find_near_duplicates(embedding)
if is_true_duplicate(fact, candidates):
print(f" 🔁 スキップ（重複）: {fact}")
continue
# --- ここまで ---
collection.add(
ids=[f"mem_{turn_id}_{i}"],
embeddings=[embedding],
documents=[fact],
metadatas=[{"timestamp": datetime.now().isoformat()}]
)
saved.append(fact)
print(f" 💾 保存: {fact}")
return saved
動作イメージ：
（既存の記憶）ユーザーは猫が好き
📝 抽出 → 保存:
🔁 スキップ（重複）: ユーザーは猫が大好き # 類似度 高 / LLM判定 SAME
💾 保存: ユーザーは猫が好きではない # 類似度 高（埋め込みは似てる！）だが LLM判定 DIFFERENT
💾 保存: ユーザーは犬が好き # 類似度 しきい値未満
# ポイント：「好きではない」は埋め込みが近くても否定なので保存される
メモ： これは「重複を保存しない」だけの仕組みで、古い「猫が好き」を消すわけではありません。新旧の矛盾をどう扱うかはステップ⑤（タイムスタンプで新旧を区別）の担当です。以降のステップ④・⑤の
save_memoriesも、この重複チェックを含んだものとして扱います。
このステップの限界
保存内容はきれいになりましたが、検索クエリ自体はまだユーザーの生の入力のままです。「それをFastAPIで使う方法は？」のような文脈依存の質問では、「それ」を埋め込んでも過去の記憶とうまくマッチしません。次のステップで検索クエリ側を改善します。
ステップ④：クエリリファクタリングで検索精度を上げる
解決策は、LLM自身に検索クエリを書き直してもらうことです。
ユーザーの質問を、文脈を補完した「自己完結したクエリ」に変換してから検索します。
「それをFastAPIで使う方法は？」
↓ LLMがリファクタリング
「PythonのFastAPIで型ヒントを使う方法」
↓ ベクトル検索
関連する過去の記憶を正確に取得できる！
ステップ③の記憶抽出に、この「クエリ書き直し」を足したものが以下です。
import requests
import chromadb
import json
from datetime import datetime
# --- Ollama関数 ---
def chat_with_ollama(messages: list[dict]) -> str:
response = requests.post(
"http://localhost:11434/api/chat",
json={"model": "llama3", "messages": messages, "stream": False}
)
return response.json()["message"]["content"]
def chat_json(prompt: str) -> dict:
"""Ollamaに必ず有効なJSONを返させる（format="json"）。前置き文の混入を防ぐ。"""
response = requests.post(
"http://localhost:11434/api/chat",
json={
"model": "llama3",
"messages": [{"role": "user", "content": prompt}],
"stream": False,
"format": "json",
}
)
try:
return json.loads(response.json()["message"]["content"])
except (json.JSONDecodeError, KeyError):
return {}
def get_embedding(text: str) -> list[float]:
response = requests.post(
"http://localhost:11434/api/embeddings",
json={"model": "nomic-embed-text", "prompt": text}
)
return response.json()["embedding"]
# --- ChromaDB セットアップ ---
client = chromadb.Client()
collection = client.get_or_create_collection(
name="memory_with_refactoring",
metadata={"hnsw:space": "cosine"}
)
# --- クエリリファクタリング ---
def refactor_query(user_input: str, recent_history: list[dict]) -> str:
"""
最近の会話履歴を踏まえて、ユーザーの質問を自己完結したクエリに書き直す。
ここで渡す recent_history は「直近数件」だけ。全履歴ではない。
"""
if not recent_history:
return user_input # 履歴がなければそのまま返す
history_text = "\n".join([
f"{m['role']}: {m['content']}"
for m in recent_history[-4:] # 直近4件だけ使う
])
refactor_prompt = f"""以下の会話の流れを踏まえて、ユーザーの最新の質問を「文脈がなくても意味が通じる、自己完結した検索クエリ」に書き直してください。
重要：
- 必ずユーザーの最新の質問と同じ言語で書く（日本語の質問なら日本語で。英語に翻訳しない）
- 引用符・かっこ・記号で囲まない。説明や前置きも付けない
- 書き直したクエリの文字列だけを出力する
会話の流れ：
{history_text}
ユーザーの最新の質問：
{user_input}"""
refined_query = chat_with_ollama([{"role": "user", "content": refactor_prompt}])
# 念のため、前後の引用符・記号を除去（モデルが付けてくることがある）
return refined_query.strip().strip('"\'「」 ')
# --- 記憶抽出（ステップ③と同じ） ---
def extract_memories(user_text: str, assistant_text: str) -> list[str]:
prompt = f"""以下の会話から、後で思い出す価値のある「重要な事実」だけを抽出してください。
ルール：
- 事実は「ユーザーの発言」を最優先で抽出する。AIの返答はあくまで文脈の補助で、AIの言い換えでユーザーの発言の意味を上書きしない
- ユーザーに関する事実（名前・職業・居住地・好み・目標など）
- 意味と方向（肯定/否定・好き/嫌い）をそのまま正確に保つ。ぼかさない
- 「〜が変化した」のような曖昧な要約は禁止。変化したなら「変化後の内容」を具体的に書く
- AIが提示した具体的な推奨・決定も、価値があれば記録してよい
- あいさつ・相槌・一時的な雑談は抽出しない
- 重要な事実がなければ、空のリストを返す
例：
- 入力「もう犬は好きじゃない」→ {{"facts": ["ユーザーは犬が好きではない"]}}
- ❌ 悪い例：「ユーザーの犬への好みが変化した」（好き嫌いの方向が消えている）
次のJSON形式だけで出力してください（前置きや説明は禁止）：
{{"facts": ["事実1", "事実2"]}}
会話：
User: {user_text}
Assistant: {assistant_text}"""
data = chat_json(prompt)
facts = data.get("facts", [])
return [f.strip() for f in facts if isinstance(f, str) and f.strip()]
# --- 重複検出（ステップ3.5と同じ） ---
SIMILARITY_THRESHOLD = 0.85
def find_near_duplicates(embedding: list[float], n_results: int = 3) -> list[str]:
if collection.count() == 0:
return []
results = collection.query(
query_embeddings=[embedding],
n_results=min(n_results, collection.count())
)
candidates = []
for doc, dist in zip(results["documents"][0], results["distances"][0]):
if (1 - dist) >= SIMILARITY_THRESHOLD:
candidates.append(doc)
return candidates
def is_true_duplicate(new_fact: str, candidates: list[str]) -> bool:
if not candidates:
return False
existing_text = "\n".join(f"- {c}" for c in candidates)
prompt = f"""新しい事実が、既存の事実のどれかと「まったく同じ意味」かどうかを判定してください。
判定ルール：
- 言い回しが違っても意味が同じなら「同じ」（例：「猫が好き」と「猫が大好き」）
- 否定・反対・変化は「同じではない」（例：「猫が好き」と「猫が好きではない」は正反対なので別物）
- 対象が違えば「同じではない」（例：「猫が好き」と「犬が好き」）
既存の事実：
{existing_text}
新しい事実：
{new_fact}
完全に同じ意味なら SAME、少しでも違う・反対・変化していれば DIFFERENT とだけ出力してください。"""
answer = chat_with_ollama([{"role": "user", "content": prompt}]).strip().upper()
return answer.startswith("SAME")
def save_memories(facts: list[str], turn_id: int) -> list[str]:
saved = []
for i, fact in enumerate(facts):
embedding = get_embedding(fact)
# ステップ3.5：重複チェック
candidates = find_near_duplicates(embedding)
if is_true_duplicate(fact, candidates):
print(f" 🔁 スキップ（重複）: {fact}")
continue
collection.add(
ids=[f"mem_{turn_id}_{i}"],
embeddings=[embedding],
documents=[fact],
metadatas=[{"timestamp": datetime.now().isoformat()}]
)
saved.append(fact)
print(f" 💾 保存: {fact}")
return saved
# これ未満の類似度の記憶は「関連が薄い」とみなして渡さない（高いほど厳しい・要調整）
RELEVANCE_THRESHOLD = 0.6
def retrieve_relevant_memories(query: str, n_results: int = 5) -> list[str]:
if collection.count() == 0:
return []
results = collection.query(
query_embeddings=[get_embedding(query)],
n_results=min(n_results, collection.count())
)
relevant = []
for doc, dist in zip(results["documents"][0], results["distances"][0]):
if (1 - dist) >= RELEVANCE_THRESHOLD:
relevant.append(doc)
return relevant
# --- RAG + クエリリファクタリング チャットボット ---
def rag_chatbot_full():
turn = 0
recent_history = [] # 直近の会話（リファクタリング用）
print("RAG + 記憶抽出 + クエリリファクタリング 起動（終了: 'quit'）\n")
while True:
user_input = input("You: ").strip()
if user_input.lower() == "quit":
break
# 1. クエリをリファクタリング（文脈を補完）
refined_query = refactor_query(user_input, recent_history)
if refined_query != user_input:
YELLOW, GREEN, BOLD, RESET = "\033[33m", "\033[32m", "\033[1m", "\033[0m"
print(f"\n{BOLD}🔍 クエリ変換{RESET}")
print(f" {YELLOW}Before: {user_input}{RESET}")
print(f" {GREEN}After : {refined_query}{RESET}\n")
# 2. リファクタリング後のクエリで記憶を検索
memories = retrieve_relevant_memories(refined_query, n_results=5)
context = "\n".join(f"- {m}" for m in memories) if memories else "（まだ記憶がありません）"
# 3. 記憶を参考情報として渡す
messages = [
{
"role": "system",
"content": f"""あなたは親切なアシスタントです。
以下はユーザーに関する記憶です。必要なときだけ参考にして回答してください。
--- 記憶 ---
{context}
--- ここまで ---"""
},
{"role": "user", "content": user_input},
]
# 4. 回答を生成
response = chat_with_ollama(messages)
# 5. 今回のやり取りから事実を抽出して保存
facts = extract_memories(user_input, response)
print(f"AI: {response}")
if facts:
print("\n📝 抽出 → 保存:")
save_memories(facts, turn)
else:
print("\n📝 今回保存する事実はありませんでした")
turn += 1
# 6. 直近履歴を更新（リファクタリング用、直近8件のみ保持）
recent_history.append({"role": "user", "content": user_input})
recent_history.append({"role": "assistant", "content": response})
recent_history = recent_history[-8:]
CYAN, GRAY, BOLD, RESET = "\033[36m", "\033[90m", "\033[1m", "\033[0m"
print(f"\n{BOLD}📚 参照した記憶（上位{len(memories)}件）{RESET}")
if memories:
for i, m in enumerate(memories, 1):
preview = m[:60] + "..." if len(m) > 60 else m
print(f" {CYAN}[{i}] {preview}{RESET}")
else:
print(f" {GRAY}（まだ記憶がありません）{RESET}")
print()
if __name__ == "__main__":
rag_chatbot_full()
動作例
You: Pythonの型ヒントについて教えて
AI: Pythonの型ヒントは、変数や関数の引数・戻り値に型情報を付与する機能です...
📝 抽出 → 保存:
💾 保存: ユーザーはPythonの型ヒントに興味がある
📚 参照した記憶（上位0件）
（まだ記憶がありません）
You: それをFastAPIで使う方法は？
🔍 クエリ変換
Before: それをFastAPIで使う方法は？ # 黄色で表示
After : PythonのFastAPIで型ヒントを活用する方法 # 緑色で表示
AI: FastAPIはPythonの型ヒントを中心に設計されており...
📝 抽出 → 保存:
💾 保存: ユーザーはFastAPIに興味がある
📚 参照した記憶（上位1件）
[1] ユーザーはPythonの型ヒントに興味がある # シアン色で表示
You: 私の名前覚えてる？（← 100回前に伝えた情報）
🔍 クエリ変換
Before: 私の名前覚えてる？
After : ユーザーの名前や自己紹介に関する情報
AI: はい、田中太郎さんですね！
📝 今回保存する事実はありませんでした
📚 参照した記憶（上位1件）
[1] ユーザーの名前は田中太郎
Before/Afterを並べることで、LLMが代名詞をどう解釈してクエリを補完しているか一目で分かります。クエリが適切に変換されることで、遠い過去の情報も正確に引き出せるようになりました。
💡 書き直すクエリの言語を固定する。
llama3は英語中心のモデルなので、「検索クエリに書き直して」とだけ頼むと、日本語の質問でも"dog dislike"のように英語に翻訳して返すことがあります。記憶は日本語で保存されているので、英語クエリで検索すると類似度が大きく落ち、せっかくのリファクタリングが逆効果になります。対策として、プロンプトで「ユーザーの質問と同じ言語で書く（英語に翻訳しない）」「引用符で囲まない」を明示し、出力の前後の記号もstripで除去しています。多言語に強い埋め込みモデル（bge-m3 など）に替えると言語ずれの影響は和らぎますが、まずは出力言語を固定するのが確実です。
ステップ⑤：記憶の更新 — タイムスタンプで新旧を区別する（簡易版）
⚠️ 先に断っておくと、このステップは「ちゃんとした記憶の更新」ではなく、あくまで簡単なトリックです。 本格的な記憶の更新（矛盾の検出・古い情報の上書き・削除）はそれ自体が大きなテーマで、後述する「次のステップ」として扱います。ここで紹介するのは、最小限の手間で「古い情報」と「今の情報」をそれなりに区別する方法です。
ここまでの実装には、地味だが厄介な問題が残っています。事実がどんどん追加されるだけで、古くなった事実が残り続けることです。
（過去の記憶）ユーザーは東京在住
You: 先月大阪に引っ越したんだ
↓ 抽出
（追加される記憶）ユーザーは大阪在住
→ いま記憶には「東京在住」と「大阪在住」の両方が残る
→ 「今どこに住んでる？」と聞くと、AIが混乱する可能性がある
ひとつの考え方は「古い事実を削除・上書きする」ことですが、それは矛盾の検出が必要で、削除を間違えると情報が失われます。
もっと単純で、しかも過去の情報を捨てずに済むやり方があります。全部の事実を残したまま、検索結果をタイムスタンプ順（新しい順）に並べ替えて、「新しいものが現在の事実」だとモデルに伝えるだけです。
記憶（新しい順に提示）：
- (2026-05-01) ユーザーは大阪在住 ← 現在
- (2025-01-01) ユーザーは東京在住 ← 過去
→ AI: 今は大阪、以前は東京にお住まいですね
このやり方には嬉しいおまけがあります。古い事実を消さないので、「前はどこに住んでた？」という過去についての質問にも答えられるのです（削除方式ではこれができません）。
実装
変更はわずかです。retrieve_relevant_memories をタイムスタンプ付きで返し、新しい順に並べ替えるだけ。ステップ④の他の部分はそのままで構いません。
def retrieve_relevant_memories(query: str, n_results: int = 5) -> list[dict]:
if collection.count() == 0:
return []
results = collection.query(
query_embeddings=[get_embedding(query)],
n_results=min(n_results, collection.count())
)
# ステップ④と同じ関連度フィルタ（RELEVANCE_THRESHOLD 以上のみ）
memories = [
{"fact": doc, "timestamp": meta.get("timestamp", "")}
for doc, meta, dist in zip(
results["documents"][0], results["metadatas"][0], results["distances"][0]
)
if (1 - dist) >= RELEVANCE_THRESHOLD
]
# ※ ChromaDBは「類似度順」で返すので、明示的に新しい順へ並べ替える
memories.sort(key=lambda m: m["timestamp"], reverse=True)
return memories
そして、システムプロンプトでは日付を添えて、「新しいものが現在」というルールを明示します（順番だけのヒントは小さいモデルには弱いので、日付を見せる方が効きます）。
context = "\n".join(
f"- ({m['timestamp'][:10]}) {m['fact']}" for m in memories
) if memories else "（まだ記憶がありません）"
system_prompt = f"""あなたは親切なアシスタントです。
以下はユーザーに関する記憶で、新しい順に並んでいます。
矛盾する情報があるときは、最も新しい記憶を「現在の事実」として扱い、
古いものは「過去の情報」として区別してください。
--- 記憶（新しい順）---
{context}
--- ここまで ---"""
この「トリック」の限界
-
検索で新旧の両方がヒットしないと意味がない。 「今どこ住んでる？」で「大阪在住」しか取れなければ、並べ替えても比較できません（同じ属性なので両方ヒットしやすいですが、保証はない。
n_resultsを多めに）。 - 判断をモデルに丸投げしている。 「新しい=現在」を llama3 が常に守ってくれるとは限らず、「大阪と東京の両方に住んでいる」のように混同することもあります。日付提示で改善しますが、自分のモデルで必ずテストを。
-
記憶が溜まり続ける。 引っ越しを5回すれば住所の事実が5件残ります。
n_resultsで上限を切れば実用上は問題になりにくいですが、古い情報の比率は少しずつ増えていきます。
本格的にやるなら（次のステップ）
「トリック」を越えて本気で更新を扱うなら、新しい事実ごとに既存の近い記憶を引き、LLMに ADD / UPDATE / DELETE / NOOP のどれかを選ばせる方式になります（Mem0 が採用している「Update フェーズ」がこれ）。さらに進むと、Zep のように時間的な有効期間を持つ知識グラフで「いつからいつまで正しかった事実か」を管理します。いずれも実装は重くなるので、この記事では簡易版にとどめ、別記事で扱う予定です。
まとめ
この記事で実装したことを振り返ります：
| アプローチ | トークン数 | 検索精度 | 長期記憶 | 矛盾・更新 |
|---|---|---|---|---|
| ①全履歴渡し | 会話数に比例して増加 | 中（Lost in the Middle） | × 上限あり | × |
| ②基本RAG（生ログ保存） | 一定（上位N件） | 中（ノイズ多・代名詞に弱い） | ○ | × |
| ③＋記憶抽出 | 一定（さらに小さい） | 高（要点だけを保存） | ○ | × |
| ③.5＋重複検出 | 一定（重複を貯めない） | 高（記憶がすっきり） | ○ | × |
| ④＋クエリリファクタリング | 一定 | 高（代名詞も解決） | ○ | × |
| ⑤＋タイムスタンプ管理 | 一定 | 高 | ○ | △（簡易的に新旧を区別） |
実装のポイント：
- ChromaDB で記憶をベクトルとして保存し、意味的類似検索を実現
- nomic-embed-text（Ollama）でローカルに埋め込みを生成
- 記憶抽出 で生ログではなく「重要な事実」だけを保存し、ノイズを削減
- 重複検出 で保存前に意味的な重複をスキップ（否定・変化は別物として保存）
- クエリリファクタリング で「それ」「あれ」などの代名詞問題を解消
- 取り出した記憶は会話ターンではなく「参考情報」としてシステムプロンプトに渡す
- タイムスタンプ で古い情報と新しい情報を簡易的に区別する
次のステップ
さらに精度・実用性を上げたい場合は、以下のテクニックが有効です：
- 本格的な記憶更新 — ADD/UPDATE/DELETEや、Zepのような時間的知識グラフで矛盾を解決する
- Re-ranking — 検索結果をクロスエンコーダーで再スコアリングし、本当に関連性の高いものだけを残す
- 会話の要約記憶 — 古い会話を要約してコンパクトに保持する
- ハイブリッド検索 — ベクトル検索とBM25（キーワード検索）を組み合わせて精度を上げる
- HyDE — 仮想的な回答を生成してから検索クエリとして使う
⚠️ 注意：RAGはモデルの「賢さ」までは肩代わりしない
最後に、いちばん大事な前提を。この記事の工夫（RAG・記憶抽出・クエリリファクタリング・重複検出）は、すべて「モデルに何を見せるか」を改善するものであって、モデルそのものを賢くするわけではありません。
このチャットボットは、いたるところでLLMに判断を任せています。
- 事実の抽出 — どれが「重要な事実」か
- クエリのリファクタリング — 「それ」「あれ」が何を指すか
- 重複判定 — SAME / DIFFERENT
- そして何より、最終的な回答の生成
これらの品質は、すべて土台となるLLMの能力に乗っています。つまり、検索が完璧に「正しい記憶」を引っ張ってきても、モデルが弱ければこうなります。
- 抽出がスカスカ、あるいは見当違いの事実を拾う
- クエリの書き直しが的外れになる
- 「猫が好き」と「猫が好きではない」の区別を誤る
- 文脈は揃っているのに、回答そのものが浅い・的外れになる
要するに Garbage in, garbage out です。RAGをいくら磨いても、モデルの天井は超えられません。本記事では手軽に動かせる llama3（8B）を使っていますが、より大きなモデル（llama3:70b など）やクラウドのモデル（GPT・Claude など）に差し替えるだけで、上記のLLM依存ステップすべての精度が大きく変わります。format="json" のような工夫は出力の安定性を上げますが、賢さまでは足せません。
💡 デバッグのコツ： 回答がイマイチなときは、正しい記憶が検索できていないのか（RAG側の問題）」と「記憶は合っているのに回答が悪いのか（モデル側の問題）」を切り分けて考えると、どこを直すべきかが見えてきます。前者はこの記事の手法で、後者はモデルの差し替えで改善します。
（次のステップの各テクニックは、別の記事で紹介予定です。）
参考
- 【初心者向け】Ollamaってなに？自分のPCでAIを動かす「ローカルLLM」をゼロから解説（Qiita）
- ChromaDB入門：保存できるデータと個人開発での活用法（Qiita）
- エンジニアのための RAG入門（Qiita）
- コンテキストウィンドウの罠｜LLMの構造的弱点「Lost in the Middle」と「Context Rot」を整理する（Qiita）
- Mem0 で AI チャットにメモリ機能を実装してみた（Zenn）
- AIエージェントメモリとは？Mem0/Zep/Letta/LangMemの比較【2026年版】（renue）
最後に
はつかぜ株式会社では、IT学習や業務に役立つ情報を定期的にお届けしていきたいと思っています。システム開発のお問い合わせ・ご相談はこちら
https://www.hatsukaze.co.jp/inquire/