Query Neo4j dagli strumenti dell’agente
Generazione di query Cypher, attraversamento del grafo e strumenti per l’analisi dei risultati.
Query Neo4j dagli strumenti dell’agente è una lezione AI Agents gratuita su CoddyKit. Questa è la lezione 2 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Agents, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Agents include 4 lezioni in totale.
Perché Neo4j per la conoscenza degli agenti
Neo4j è un database a grafo ottimizzato per attraversare le relazioni. Per gli agenti che lavorano con i grafi di conoscenza, consente di porre in modo efficiente domande come Chi lavora con chi? o Quali aziende sono collegate a questa persona?
Connessione a Neo4j
Il driver Python neo4j si connette a un'istanza Neo4j. Utilizzi le variabili d'ambiente per l'URI di connessione e le credenziali. Chiuda sempre il driver al termine.
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()Query Cypher di base
Cypher è il linguaggio di query di Neo4j. Il modello fondamentale è MATCH (n:Label {property: value})-[:RELATIONSHIP]->(m) RETURN m. Le parentesi quadre contengono il tipo di relazione, mentre le parentesi tonde contengono i nodi.
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)Query parametrizzate
Utilizzi sempre query parametrizzate (ad esempio, $name) invece dell'interpolazione di stringhe. In questo modo previene l'injection in Cypher e migliora le prestazioni grazie alla memorizzazione nella cache del piano di esecuzione.
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)')Scrittura dei dati nel grafo
Utilizzi MERGE per eseguire l'upsert di nodi e relazioni. MERGE crea il nodo o la relazione solo se non esiste già, evitando i duplicati.
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')Generazione di Cypher dal linguaggio naturale
Un agente può convertire le domande in linguaggio naturale in query Cypher. Fornisca all'LLM lo schema del grafo come contesto, quindi gli chieda di generare la query Cypher appropriata.
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)Esecuzione sicura del Cypher generato
Prima di eseguire Cypher generato da un LLM, lo convalidi. Blocchi le istruzioni che modificano i dati (CREATE, DELETE, SET), a meno che l'agente non necessiti esplicitamente dell'accesso in scrittura. Quando possibile, esegua le query in una sessione di sola lettura.
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')Analisi e formattazione dei risultati delle query
Formatti i risultati delle query Neo4j come stringa leggibile da una persona o come oggetto strutturato che l'LLM possa interpretare. Gestisca con eleganza i risultati vuoti.
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?'))Strumento dell'agente: ricerca nel grafo
Racchiuda le query Neo4j in uno strumento per l'agente. Lo strumento accetta una domanda in linguaggio naturale, genera Cypher, lo esegue in modo sicuro e restituisce risultati formattati.
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')Attraversamento multi-hop del grafo
I database a grafo eccellono nelle query multi-hop: consentono di trovare entità distanti N passaggi. Ad esempio, permettono di trovare aziende collegate a una persona attraverso due o più intermediari.
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')Aggregazione in Cypher
Cypher supporta le funzioni di aggregazione: COUNT, COLLECT, AVG, MIN, MAX. Le utilizzi per rispondere a domande riepilogative sul grafo.
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])Verifica delle conoscenze: Neo4j per gli agenti
Verifichi la sua comprensione dell'utilizzo di Neo4j dagli strumenti degli agenti.
Riepilogo degli strumenti Neo4j per gli agenti
L'integrazione di Neo4j in un agente comprende: la connessione tramite il driver Python, l'uso di query parametrizzate per prevenire l'injection, la generazione di Cypher dal linguaggio naturale con l'assistenza di un LLM, la convalida delle query prima dell'esecuzione, la formattazione dei risultati per l'utilizzo da parte dell'LLM e l'esposizione della ricerca nel grafo come strumento dell'agente.
Domande Frequenti
La lezione «Query Neo4j dagli strumenti dell’agente» è gratuita?
Sì — il testo completo di «Query Neo4j dagli strumenti dell’agente» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Agents, passa a CoddyKit PRO. Il corso AI Agents include 4 lezioni in totale.
Cosa imparerò in «Query Neo4j dagli strumenti dell’agente»?
Generazione di query Cypher, attraversamento del grafo e strumenti per l’analisi dei risultati. Eserciti AI Agents con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.
Ho bisogno di esperienza per iniziare AI Agents?
Non è richiesta alcuna esperienza precedente. AI Agents su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 2 di 4.
Quanto tempo richiede la lezione «Query Neo4j dagli strumenti dell’agente»?
La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.
Posso scrivere ed eseguire codice in questa lezione AI Agents?
Sì. Ogni lezione AI Agents include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.
Tutte le lezioni di questo corso
- Estrazione di entità per grafi della conoscenza
- Query Neo4j dagli strumenti dell’agente
- Combinazione del recupero vettoriale e a grafo
- Creazione di un agente potenziato dalla conoscenza