Neo4j-query's vanuit agenttools
Genereren van Cypher-query's, doorlopen van grafen en tools voor het parseren van resultaten.
Neo4j-query's vanuit agenttools is een gratis AI-agenten-les op CoddyKit. Dit is les 2 van 4. Je kunt de volledige les hieronder gratis lezen en daarna in de browser praktisch oefenen met een ingebouwde code-editor en een AI-begeleider die 24/7 beschikbaar is. Deze les maakt deel uit van het leertraject AI-agenten. Je voortgang wordt gesynchroniseerd op het web en in de CoddyKit-app. De cursus AI-agenten bevat in totaal 4 lessen.
Waarom Neo4j voor kennis van agents?
Neo4j is een graafdatabase die is geoptimaliseerd voor het doorlopen van relaties. Voor agents die met kennisgrafieken werken, maakt deze database het mogelijk om efficiënt vragen te stellen als Wie werkt met wie? of Met welke bedrijven is deze persoon verbonden?
Verbinding maken met Neo4j
Het Python-stuurprogramma neo4j maakt verbinding met een Neo4j-instantie. Gebruik omgevingsvariabelen voor de verbindings-URI en inloggegevens. Sluit het stuurprogramma altijd wanneer je klaar bent.
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()Basisopvragingen in Cypher
Cypher is de opvraagtaal van Neo4j. Het basispatroon is MATCH (n:Label {property: value})-[:RELATIONSHIP]->(m) RETURN m. Vierkante haken bevatten het relatietype en ronde haakjes bevatten knopen.
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)Geparametriseerde opvragingen
Gebruik altijd geparametriseerde opvragingen (bijvoorbeeld $name) in plaats van tekenreeksinterpolatie. Dit voorkomt Cypher-injectie en verbetert de prestaties doordat queryplannen in de cache worden opgeslagen.
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)')Graafgegevens schrijven
Gebruik MERGE om knopen en relaties toe te voegen of bij te werken. MERGE maakt de knoop of relatie alleen als deze nog niet bestaat, zodat dubbele entiteiten worden voorkomen.
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 genereren uit natuurlijke taal
Een agent kan vragen in natuurlijke taal omzetten naar Cypher-opvragingen. Geef de LLM het schema van je graaf als context en vraag vervolgens om de juiste Cypher te genereren.
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)Gegenereerde Cypher veilig uitvoeren
Valideer door een LLM gegenereerde Cypher voordat je deze uitvoert. Blokkeer mutatie-instructies (CREATE, DELETE, SET), tenzij de agent expliciet schrijftoegang nodig heeft. Voer de opvraging indien mogelijk uit in een sessie met alleen-lezenrechten.
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')Resultaten van opvragingen parseren en opmaken
Maak van Neo4j-opvragingsresultaten een voor mensen leesbare tekenreeks of een gestructureerd object dat de LLM kan interpreteren. Handel lege resultaten netjes af.
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?'))Agenthulpmiddel: zoeken in de graaf
Verpak Neo4j-opvragingen als een hulpmiddel voor een agent. Het hulpmiddel accepteert een vraag in natuurlijke taal, genereert Cypher, voert deze veilig uit en retourneert opgemaakte resultaten.
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')Een graaf doorlopen met meerdere sprongen
Graafdatabases zijn uitstekend geschikt voor opvragingen met meerdere sprongen: entiteiten vinden die N stappen van elkaar verwijderd zijn. Je kunt bijvoorbeeld bedrijven vinden die via twee of meer tussenpersonen met een persoon verbonden zijn.
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')Aggregatie in Cypher
Cypher ondersteunt aggregatiefuncties: COUNT, COLLECT, AVG, MIN, MAX. Gebruik ze om samenvattende vragen over de graaf te beantwoorden.
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])Kennistoets: Neo4j voor agents
Test je begrip van het gebruik van Neo4j vanuit hulpmiddelen van agents.
Samenvatting van Neo4j-agenthulpmiddelen
Neo4j integreren in een agent omvat: verbinding maken met het Python-stuurprogramma, geparametriseerde opvragingen gebruiken om injectie te voorkomen, Cypher uit natuurlijke taal genereren met hulp van een LLM, opvragingen vóór uitvoering valideren, resultaten opmaken voor gebruik door de LLM en zoeken in de graaf beschikbaar maken als hulpmiddel voor een agent.
Leer AI-agenten met een AI-tutor — gratis
Schrijf echte code en voer die uit in je browser, krijg direct hulp van een AI-tutor die 24/7 beschikbaar is en ga verder waar je gebleven bent op het web of in de app.
- Cursussen
- 60
- Lessen
- 239
Veelgestelde vragen
Is de les “Neo4j-query's vanuit agenttools” gratis?
Ja — de volledige tekst van “Neo4j-query's vanuit agenttools” kun je hier gratis op het web lezen. Als je interactief wilt oefenen met een ingebouwde code-editor en een AI-begeleider die 24/7 beschikbaar is, en de rest van de cursus AI-agenten wilt ontgrendelen, kun je upgraden naar CoddyKit PRO. De cursus AI-agenten bevat in totaal 4 lessen.
Wat leer ik in “Neo4j-query's vanuit agenttools”?
Genereren van Cypher-query's, doorlopen van grafen en tools voor het parseren van resultaten. Je oefent met AI-agenten door code rechtstreeks in de browser uit te voeren. Een AI-begeleider die 24/7 beschikbaar is beantwoordt je vragen terwijl je de les doorwerkt.
Heb ik ervaring nodig om met AI-agenten te beginnen?
Ervaring vooraf is niet nodig. AI-agenten op CoddyKit is opgebouwd voor beginners tot gevorderden, zodat je hier of bij het begin kunt starten en in je eigen tempo kunt leren. Dit is les 2 van 4.
Hoe lang duurt de les “Neo4j-query's vanuit agenttools”?
De meeste lessen van CoddyKit duren ongeveer 5–10 minuten. Elke les is kort en interactief, zodat je gestaag vooruitgaat en op het web en in de app precies verdergaat waar je was gebleven.
Kan ik code schrijven en uitvoeren in deze les over AI-agenten?
Ja. Elke les over AI-agenten bevat een ingebouwde code-editor, zodat je rechtstreeks in je browser echte code kunt schrijven en uitvoeren en direct feedback van AI krijgt — lokale installatie is niet nodig.
Alle lessen in deze cursus
- Entiteiten extraheren voor kennisgrafieken
- Neo4j-query's vanuit agenttools
- Vector- en graafretrieval combineren
- Een kennisverrijkte agent bouwen