0Pricing
AI Agents · Leçon

Requêtes Neo4j depuis les outils d’un agent

Génération de requêtes Cypher, parcours du graphe et outils d’analyse des résultats.

Requêtes Neo4j depuis les outils d’un agent est une leçon AI Agents gratuite sur CoddyKit. Ceci est la leçon 2 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage AI Agents, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours AI Agents comprend 4 leçons au total.

Pourquoi utiliser Neo4j pour les connaissances des agents ?

Neo4j est une base de données graphe optimisée pour parcourir les relations. Pour les agents qui travaillent avec des graphes de connaissances, elle permet de poser efficacement des questions comme Qui travaille avec qui ? ou Quelles entreprises sont liées à cette personne ?

Se connecter à Neo4j

Le pilote Python neo4j se connecte à une instance Neo4j. Utilisez des variables d’environnement pour l’URI de connexion et les identifiants. Fermez toujours le pilote lorsque vous avez terminé.

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()

Requêtes Cypher de base

Cypher est le langage de requête de Neo4j. Le modèle de base est MATCH (n:Label {property: value})-[:RELATIONSHIP]->(m) RETURN m. Les crochets contiennent le type de relation et les parenthèses contiennent les nœuds.

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)

Requêtes paramétrées

Utilisez toujours des requêtes paramétrées (par exemple, $name) plutôt qu’une interpolation de chaînes. Cela empêche les injections Cypher et améliore les performances grâce à la mise en cache des plans de requête.

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)')

Écrire des données dans le graphe

Utilisez MERGE pour créer ou mettre à jour des nœuds et des relations. MERGE ne crée le nœud ou la relation que s’il ou elle n’existe pas déjà, ce qui évite les doublons.

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')

Générer du Cypher à partir du langage naturel

Un agent peut convertir des questions en langage naturel en requêtes Cypher. Fournissez au LLM le schéma de votre graphe comme contexte, puis demandez-lui de générer le Cypher approprié.

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)

Exécuter le Cypher généré en toute sécurité

Avant d’exécuter du Cypher généré par un LLM, validez-le. Bloquez les instructions de modification (CREATE, DELETE, SET), sauf si l’agent a explicitement besoin d’un accès en écriture. Exécutez-les dans une session en lecture seule lorsque cela est possible.

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')

Analyser et formater les résultats des requêtes

Formatez les résultats des requêtes Neo4j sous forme de chaîne lisible par un humain ou d’objet structuré que le LLM pourra interpréter. Gérez correctement les résultats vides.

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?'))

Outil d’agent : recherche dans le graphe

Transformez les requêtes Neo4j en outil d’agent. Cet outil accepte une question en langage naturel, génère du Cypher, l’exécute en toute sécurité et renvoie des résultats formatés.

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')

Parcours du graphe sur plusieurs niveaux

Les bases de données graphe excellent dans les requêtes sur plusieurs niveaux : trouver des entités situées à N étapes de distance. Par exemple, trouver des entreprises liées à une personne par l’intermédiaire de deux intermédiaires ou plus.

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')

Agrégation en Cypher

Cypher prend en charge les fonctions d’agrégation : COUNT, COLLECT, AVG, MIN, MAX. Utilisez-les pour répondre aux questions récapitulatives sur le graphe.

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])

Vérification des connaissances : Neo4j pour les agents

Testez votre compréhension de l’utilisation de Neo4j depuis des outils d’agent.

Résumé des outils d’agent Neo4j

Intégrer Neo4j dans un agent consiste à : se connecter avec le pilote Python, utiliser des requêtes paramétrées pour empêcher les injections, générer du Cypher à partir du langage naturel avec l’aide d’un LLM, valider les requêtes avant leur exécution, formater les résultats pour que le LLM puisse les exploiter et exposer la recherche dans le graphe comme outil d’agent.

Questions Fréquemment Posées

La leçon « Requêtes Neo4j depuis les outils d’un agent » est-elle gratuite ?

Oui — le texte complet de « Requêtes Neo4j depuis les outils d’un agent » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours AI Agents, passe à CoddyKit PRO. Le cours AI Agents comprend 4 leçons au total.

Qu'est-ce que j'apprendrai dans « Requêtes Neo4j depuis les outils d’un agent » ?

Génération de requêtes Cypher, parcours du graphe et outils d’analyse des résultats. Tu pratiques AI Agents avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.

Dois-je avoir de l'expérience pour commencer AI Agents ?

Aucune expérience préalable n'est requise. AI Agents sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 2 sur 4.

Combien de temps prend la leçon « Requêtes Neo4j depuis les outils d’un agent » ?

La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.

Peux-tu écrire et exécuter du code dans cette leçon AI Agents ?

Oui. Chaque leçon AI Agents inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.

Toutes les leçons de ce cours

  1. Extraction d’entités pour les graphes de connaissances
  2. Requêtes Neo4j depuis les outils d’un agent
  3. Combiner la recherche vectorielle et la recherche dans les graphes
  4. Construire un agent enrichi par les connaissances
← Retour à AI Agents