はじめに
前回は GraphCypherQAChain を使って自然言語からNeo4jに問い合わせる仕組みを作りました。
今回はその中身を自分で実装します。「アダプター」とは何かを理解しながら、LLMとNeo4jを繋ぐ処理を一から書いていきます。
アダプターって何？
前回使った GraphCypherQAChain は内部でこういうことをしていた：
自然言語
↓ LLMに「Cypherに変換して」と頼む（スキーマも渡す）
Cypherクエリ
↓ Neo4jで実行
クエリ結果（JSON）
↓ LLMに「自然言語にして」と頼む
回答
この「LLMとNeo4jの間で変換処理を担う層」がアダプターです。
LangChainが隠してくれていた部分を自分で書くことで：
- どんなプロンプトでCypherを生成しているか
- エラーハンドリングをどこでやるか
- 変換ロジックをどうカスタマイズするか
が自分でコントロールできるようになります。
対象者
- 前回の記事でNeo4j + LLMの基本を理解した人
- LangChainの抽象化の中身が気になる人
- アダプターパターンを実装レベルで理解したい人
今回作るもの
NaturalLanguageToNeo4jAdapter
├── get_schema() # Neo4jのスキーマを取得
├── generate_cypher() # 自然言語 → Cypher（LLM）
├── execute_cypher() # Cypherを実行
├── generate_answer() # 結果 → 自然言語（LLM）
└── query() # 上記を一括で実行するメイン関数
セットアップ
前回のプロジェクトをそのまま使う。
cd neo4j-llm-tutorial
uv add ollama
Ollamaが起動していてモデルが取得済みであることを確認：
ollama list
# llama3.1:8b が表示されればOK
1. スキーマ取得
アダプターの最初の仕事は、Neo4jのスキーマをLLMに渡せる形で取得することです。スキーマがないとLLMは正しいCypherを生成できません。
adapter.py を作る：
from neo4j import GraphDatabase
import ollama
import json
NEO4J_URI = "bolt://localhost:7687"
NEO4J_AUTH = ("neo4j", "password")
OLLAMA_MODEL = "llama3.1:8b"
class NaturalLanguageToNeo4jAdapter:
def __init__(self):
self.driver = GraphDatabase.driver(NEO4J_URI, auth=NEO4J_AUTH)
def get_schema(self) -> str:
"""Neo4jのノードとリレーションシップ情報を取得する"""
with self.driver.session() as session:
labels = session.run("CALL db.labels()").value()
rel_types = session.run("CALL db.relationshipTypes()").value()
property_keys = session.run("CALL db.propertyKeys()").value()
schema = f"""
ノードラベル: {', '.join(labels)}
リレーションシップ: {', '.join(rel_types)}
プロパティキー: {', '.join(property_keys)}
""".strip()
return schema
def close(self):
self.driver.close()
スキーマ取得の確認：
if __name__ == "__main__":
adapter = NaturalLanguageToNeo4jAdapter()
print(adapter.get_schema())
adapter.close()
uv run adapter.py
ノードラベル: Person
リレーションシップ: FRIENDS_WITH
プロパティキー: name, age
2. 自然言語 → Cypherの変換
次にLLMにCypherを生成させる部分です。ここがアダプターの核心。
def generate_cypher(self, question: str) -> str:
"""自然言語をCypherクエリに変換する"""
schema = self.get_schema()
prompt = f"""あなたはNeo4jのCypherクエリの専門家です。
以下のグラフDBのスキーマを参照して、質問に答えるCypherクエリを生成してください。
スキーマ:
{schema}
ルール:
- Cypherクエリのみを返す（説明文は不要）
- マークダウンのコードブロックは使わない
- 存在しないノードやリレーションは使わない
質問: {question}
"""
response = ollama.chat(
model=OLLAMA_MODEL,
messages=[{"role": "user", "content": prompt}],
options={"temperature": 0},
)
cypher = response["message"]["content"].strip()
# LLMがコードブロック付きで返すことがあるので除去
cypher = cypher.replace("```cypher", "").replace("```", "").strip()
return cypher
temperature=0 にしているのは、クエリ生成のブレを抑えるため。創造性より正確さが重要な場面。
3. Cypherの実行
生成したCypherをNeo4jで実行する部分です。
def execute_cypher(self, cypher: str) -> list[dict]:
"""Cypherクエリを実行して結果を返す"""
with self.driver.session() as session:
result = session.run(cypher)
return [dict(record) for record in result]
4. 結果 → 自然言語の変換
クエリ結果をそのままユーザーに返しても読みにくい。LLMで自然な回答に変換します。
def generate_answer(self, question: str, cypher: str, results: list[dict]) -> str:
"""クエリ結果を自然言語の回答に変換する"""
prompt = f"""以下の情報をもとに、質問に対する自然な日本語の回答を生成してください。
質問: {question}
実行したCypherクエリ: {cypher}
クエリの結果: {json.dumps(results, ensure_ascii=False)}
回答は簡潔に、1〜2文でまとめてください。
"""
response = ollama.chat(
model=OLLAMA_MODEL,
messages=[{"role": "user", "content": prompt}],
options={"temperature": 0.3},
)
return response["message"]["content"].strip()
5. メイン関数でまとめる
ここまでのステップを1つの query() にまとめます。
def query(self, question: str) -> dict:
"""自然言語の質問からNeo4jに問い合わせて回答を返す"""
print(f"\n質問: {question}")
# Step1: Cypher生成
cypher = self.generate_cypher(question)
print(f"生成されたCypher: {cypher}")
# Step2: 実行
results = self.execute_cypher(cypher)
print(f"クエリ結果: {results}")
# Step3: 自然言語に変換
answer = self.generate_answer(question, cypher, results)
print(f"回答: {answer}")
return {
"question": question,
"cypher": cypher,
"results": results,
"answer": answer,
}
6. 動作確認
if __name__ == "__main__":
adapter = NaturalLanguageToNeo4jAdapter()
questions = [
"Aliceの友達は誰ですか？",
"Aliceの友達の友達は誰ですか？",
"25歳の人は誰ですか？",
]
for q in questions:
adapter.query(q)
print("-" * 40)
adapter.close()
uv run adapter.py
実行するとこんな感じ：
質問: Aliceの友達の友達は誰ですか？
生成されたCypher: MATCH (:Person {name: 'Alice'})-[:FRIENDS_WITH*2]->(fof) RETURN fof.name AS name
クエリ結果: [{'name': 'Carol'}]
回答: Aliceの友達の友達はCarolです。
7. エラーハンドリングを加える
LLMが不正なCypherを生成することがある。そのままNeo4jに渡すとエラーになるので対処する。
def execute_cypher(self, cypher: str) -> list[dict]:
"""Cypherクエリを実行して結果を返す（エラーハンドリング付き）"""
try:
with self.driver.session() as session:
result = session.run(cypher)
return [dict(record) for record in result]
except Exception as e:
print(f"Cypher実行エラー: {e}")
return []
def query(self, question: str) -> dict:
cypher = self.generate_cypher(question)
print(f"生成されたCypher: {cypher}")
results = self.execute_cypher(cypher)
if not results:
return {
"question": question,
"cypher": cypher,
"results": [],
"answer": "該当するデータが見つかりませんでした。",
}
answer = self.generate_answer(question, cypher, results)
print(f"回答: {answer}")
return {
"question": question,
"cypher": cypher,
"results": results,
"answer": answer,
}
GraphCypherQAChainと何が違うか
| GraphCypherQAChain | 自作アダプター | |
|---|---|---|
| プロンプト | 固定（カスタム困難） | 自由に変更できる |
| エラーハンドリング | ライブラリ任せ | 自分でコントロール |
| ログ | verbose=Trueのみ | 好きなだけ出せる |
| 拡張性 | 低い | 高い |
実運用するなら自作の方が柔軟に対応できます。
まとめ
今回作ったアダプターの流れ：
自然言語
↓ generate_cypher() ← スキーマを渡してLLMに変換させる
Cypherクエリ
↓ execute_cypher() ← Neo4jで実行
クエリ結果
↓ generate_answer() ← LLMで自然言語に戻す
回答
GraphCypherQAChain がやっていたことを自分で実装することで、各ステップでのカスタマイズや拡張がしやすくなります。プロンプトを変えたい、リトライ処理を入れたい、ログを細かく残したいといった場合に自作アダプターが活きます。