エージェントツールから Neo4j にクエリを実行する
Cypher クエリの生成、グラフ走査、結果解析ツールを扱います。
「エージェントツールから Neo4j にクエリを実行する」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン2/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
エージェントの知識に Neo4j を使う理由
Neo4j は、関係の辿り歩きに最適化されたグラフデータベースです。ナレッジグラフを扱うエージェントでは、誰と誰が一緒に働いていますか？やこの人物とつながりのある企業はどこですか？といった質問に効率よく答えられます。
Neo4j への接続
neo4j Python ドライバーで Neo4j インスタンスに接続します。接続 URI と認証情報には環境変数を使います。処理が終わったら、必ずドライバーを閉じます。
from neo4j import GraphDatabase
import os
URI = os.environ.get('NEO4J_URI', 'bolt://localhost:7687')
USER = os.environ.get('NEO4J_USER', 'neo4j')
PASSWORD = os.environ.get('NEO4J_PASSWORD', 'password')
driver = GraphDatabase.driver(URI, auth=(USER, PASSWORD))
def test_connection():
with driver.session() as session:
result = session.run('RETURN "Connected to Neo4j" AS message')
record = result.single()
print(record['message'])
test_connection()
# Always close driver when application exits
# driver.close()Cypher クエリの基礎
Cypher は Neo4j のクエリ言語です。基本的なパターンは MATCH (n:Label {property: value})-[:RELATIONSHIP]->(m) RETURN m です。角括弧には関係タイプを、丸括弧にはノードを記述します。
from neo4j import GraphDatabase
driver = GraphDatabase.driver('bolt://localhost:7687', auth=('neo4j', 'password'))
def find_company_for_person(person_name: str) -> list:
with driver.session() as session:
result = session.run(
'MATCH (p:Person {name: $name})-[:WORKS_AT]->(c:Company) '
'RETURN c.name AS company, c.industry AS industry',
name=person_name
)
return [dict(record) for record in result]
def find_colleagues(person_name: str) -> list:
with driver.session() as session:
result = session.run(
'MATCH (p:Person {name: $name})-[:WORKS_AT]->(c:Company) '
'<-[:WORKS_AT]-(colleague:Person) '
'WHERE colleague.name <> $name '
'RETURN DISTINCT colleague.name AS name',
name=person_name
)
return [r['name'] for r in result]
companies = find_company_for_person('Alice Johnson')
print('Works at:', companies)パラメーター化クエリ
文字列補間ではなく、必ずパラメーター化クエリ（例：$name）を使います。これにより Cypher インジェクションを防ぎ、クエリプランのキャッシュによってパフォーマンスも向上します。
from neo4j import GraphDatabase
driver = GraphDatabase.driver('bolt://localhost:7687', auth=('neo4j', 'password'))
# WRONG: vulnerable to injection
def bad_query(name):
query = f'MATCH (p:Person {{name: "{name}"}}) RETURN p'
# Never do this
pass
# RIGHT: parameterized
def good_query(name: str, company: str) -> list:
with driver.session() as session:
result = session.run(
'MATCH (p:Person {name: $name})-[:WORKS_AT]->(c:Company {name: $company}) '
'RETURN p.name AS person, p.title AS title, c.name AS company',
name=name,
company=company
)
return [dict(r) for r in result]
# Multiple parameters via dict
def find_by_params(params: dict) -> list:
with driver.session() as session:
result = session.run(
'MATCH (p:Person) WHERE p.name = $name AND p.department = $dept RETURN p',
**params
)
return [dict(r) for r in result]
print('Good query defined (parameterized)')グラフデータの書き込み
ノードと関係の upsert には MERGE を使います。MERGE は、ノードや関係がまだ存在しない場合にのみ作成するため、重複を防げます。
from neo4j import GraphDatabase
driver = GraphDatabase.driver('bolt://localhost:7687', auth=('neo4j', 'password'))
def upsert_person_works_at_company(person_name: str, company_name: str, title: str):
with driver.session() as session:
session.run(
'MERGE (p:Person {name: $person}) '
'MERGE (c:Company {name: $company}) '
'MERGE (p)-[r:WORKS_AT]->(c) '
'SET r.title = $title, r.updated_at = datetime()',
person=person_name,
company=company_name,
title=title
)
def create_entity_with_properties(label: str, properties: dict):
props_string = ', '.join([f'{k}: ${k}' for k in properties.keys()])
query = f'MERGE (n:{label} {{{props_string}}}) RETURN n'
with driver.session() as session:
result = session.run(query, **properties)
return result.single()
upsert_person_works_at_company('Alice', 'Acme Corp', 'Senior Engineer')
print('Graph data written')自然言語から Cypher を生成する
エージェントは自然言語の質問を Cypher クエリに変換できます。グラフスキーマをコンテキストとして LLM に渡し、適切な Cypher を生成するよう依頼します。
import openai
client = openai.OpenAI(api_key='sk-...')
GRAPH_SCHEMA = '''
Nodes:
- Person: {name, title, email}
- Company: {name, industry, founded_year}
- Product: {name, category, version}
Relationships:
- (Person)-[:WORKS_AT {title, start_date}]->(Company)
- (Person)-[:FOUNDED]->(Company)
- (Company)-[:MAKES]->(Product)
- (Person)-[:USES]->(Product)
'''
def nl_to_cypher(question: str) -> str:
prompt = (
f'Graph schema:\n{GRAPH_SCHEMA}\n\n'
f'Convert this natural language question to a Cypher query:\n{question}\n\n'
'Return only the Cypher query, no explanation.'
)
response = client.chat.completions.create(
model='gpt-4o-mini',
messages=[{'role': 'user', 'content': prompt}]
)
return response.choices[0].message.content.strip()
question = 'Who are all the people who work at companies that make AI products?'
cypher = nl_to_cypher(question)
print('Generated Cypher:')
print(cypher)生成された Cypher を安全に実行する
LLM が生成した Cypher を実行する前に検証します。エージェントに明示的な書き込み権限が必要な場合を除き、変更を行う文（CREATE、DELETE、SET）をブロックします。可能な場合は読み取り専用セッションで実行します。
import re
from neo4j import GraphDatabase
driver = GraphDatabase.driver('bolt://localhost:7687', auth=('neo4j', 'password'))
MUTATION_KEYWORDS = ['CREATE', 'DELETE', 'MERGE', 'SET', 'REMOVE', 'DROP']
def is_read_only_cypher(cypher: str) -> bool:
upper = cypher.upper()
for keyword in MUTATION_KEYWORDS:
# Check if mutation keyword appears outside of comments
if re.search(r'\b' + keyword + r'\b', upper):
return False
return True
def execute_agent_query(cypher: str, allow_writes=False) -> list:
if not allow_writes and not is_read_only_cypher(cypher):
raise ValueError(f'Mutation query blocked. Query: {cypher[:100]}')
with driver.session() as session:
result = session.run(cypher)
return [dict(r) for r in result]
# Read query: allowed
read_cypher = 'MATCH (p:Person)-[:WORKS_AT]->(c:Company) RETURN p.name, c.name LIMIT 10'
if is_read_only_cypher(read_cypher):
print('Read query: safe to execute')
# Write query: blocked
write_cypher = 'DELETE (p:Person {name: "Alice"})'
if not is_read_only_cypher(write_cypher):
print('Write query: blocked')クエリ結果の解析と整形
Neo4j のクエリ結果を、人間が読みやすい文字列、または LLM が解釈できる構造化オブジェクトに整形します。結果が空の場合も適切に処理します。
def format_graph_results(records: list, question: str) -> str:
if not records:
return f'No results found for: {question}'
# Format as a simple table
if not records[0]:
return 'Query returned no data'
headers = list(records[0].keys())
rows = []
for record in records:
row = [str(record.get(h, '')) for h in headers]
rows.append(' | '.join(row))
header_line = ' | '.join(headers)
separator = '-' * len(header_line)
table = '\n'.join([header_line, separator] + rows[:20]) # Cap at 20 rows
result = f'Results for: {question}\n{table}'
if len(records) > 20:
result += f'\n... and {len(records) - 20} more results'
return result
# Simulate some results
sample = [
{'person': 'Alice', 'company': 'Acme Corp'},
{'person': 'Bob', 'company': 'TechCo'},
]
print(format_graph_results(sample, 'Who works where?'))エージェントツール：グラフ照会
Neo4j クエリをエージェントツールとしてラップします。ツールは自然言語の質問を受け取り、Cypher を生成して安全に実行し、整形した結果を返します。
import openai
import json
client = openai.OpenAI(api_key='sk-...')
def graph_lookup_tool(question: str) -> str:
try:
# Step 1: Generate Cypher
cypher = nl_to_cypher(question)
print(f'Generated Cypher: {cypher}')
# Step 2: Validate
if not is_read_only_cypher(cypher):
return 'Error: Generated query contains write operations'
# Step 3: Execute
records = execute_agent_query(cypher)
# Step 4: Format
return format_graph_results(records, question)
except Exception as e:
return f'Graph lookup failed: {str(e)}'
# Register as OpenAI tool
graph_lookup_schema = {
'type': 'function',
'function': {
'name': 'graph_lookup',
'description': 'Query the knowledge graph to answer questions about entities and their relationships',
'parameters': {
'type': 'object',
'properties': {
'question': {
'type': 'string',
'description': 'Natural language question about entities or relationships'
}
},
'required': ['question']
}
}
}
print('Graph lookup tool registered')マルチホップグラフ探索
グラフデータベースは、N ステップ離れたエンティティを見つけるようなマルチホップクエリを得意とします。たとえば、2 人以上の仲介者を介して人物とつながっている企業を見つけられます。
from neo4j import GraphDatabase
driver = GraphDatabase.driver('bolt://localhost:7687', auth=('neo4j', 'password'))
def find_connected_companies(person_name: str, max_hops: int = 3) -> list:
with driver.session() as session:
# Variable-length path: 1 to max_hops relationships
result = session.run(
f'MATCH (p:Person {{name: $name}})-[:WORKS_AT|FOUNDED*1..{max_hops}]->(c:Company) '
'RETURN DISTINCT c.name AS company, c.industry AS industry',
name=person_name
)
return [dict(r) for r in result]
def find_shortest_path(entity1: str, entity2: str) -> dict:
with driver.session() as session:
result = session.run(
'MATCH path = shortestPath((a {name: $name1})-[*..6]-(b {name: $name2})) '
'RETURN [node in nodes(path) | node.name] AS path_nodes, '
'length(path) AS hops',
name1=entity1,
name2=entity2
)
record = result.single()
if record:
return {'path': record['path_nodes'], 'hops': record['hops']}
return {'path': [], 'hops': -1}
print('Multi-hop traversal functions defined')Cypher での集計
Cypher は COUNT、COLLECT、AVG、MIN、MAX などの集計関数をサポートしています。これらを使って、グラフに関する要約的な質問に答えます。
from neo4j import GraphDatabase
driver = GraphDatabase.driver('bolt://localhost:7687', auth=('neo4j', 'password'))
def company_employee_stats() -> list:
with driver.session() as session:
result = session.run(
'MATCH (p:Person)-[:WORKS_AT]->(c:Company) '
'RETURN c.name AS company, '
'COUNT(p) AS employee_count, '
'COLLECT(p.name) AS employees '
'ORDER BY employee_count DESC '
'LIMIT 10'
)
return [dict(r) for r in result]
def count_connections(person_name: str) -> dict:
with driver.session() as session:
result = session.run(
'MATCH (p:Person {name: $name}) '
'OPTIONAL MATCH (p)-[:WORKS_AT]->(c:Company) '
'OPTIONAL MATCH (p)-[:FOUNDED]->(fc:Company) '
'RETURN COUNT(DISTINCT c) AS employers, COUNT(DISTINCT fc) AS founded_companies',
name=person_name
)
record = result.single()
return dict(record) if record else {}
stats = company_employee_stats()
print('Company stats:', stats[:3])理解度チェック：エージェントのための Neo4j
エージェントツールから Neo4j を利用する方法についての理解度を確認します。
Neo4j エージェントツールのまとめ
Neo4j をエージェントに統合するには、Python ドライバーによる接続、インジェクションを防ぐパラメーター化クエリの使用、LLM の支援による自然言語からの Cypher 生成、実行前のクエリ検証、LLM が利用できる形式への結果整形、グラフ照会のエージェントツールとしての公開が必要です。
よくある質問
「エージェントツールから Neo4j にクエリを実行する」レッスンは無料ですか？
はい。「エージェントツールから Neo4j にクエリを実行する」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「エージェントツールから Neo4j にクエリを実行する」で何を学びますか？
Cypher クエリの生成、グラフ走査、結果解析ツールを扱います。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン2/4です。
「エージェントツールから Neo4j にクエリを実行する」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか？
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- ナレッジグラフのためのエンティティ抽出
- エージェントツールから Neo4j にクエリを実行する
- ベクトル検索とグラフ検索の組み合わせ
- ナレッジ拡張エージェントを構築する