Consultas de Neo4j desde herramientas de agentes
Generación de consultas Cypher, recorrido de grafos y herramientas para analizar resultados.
Consultas de Neo4j desde herramientas de agentes es una lección gratuita de AI Agents en CoddyKit. Esta es la lección 2 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de AI Agents, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de AI Agents incluye 4 lecciones en total.
¿Por qué Neo4j para el conocimiento de los agentes?
Neo4j es una base de datos de grafos optimizada para recorrer relaciones. Para los agentes que trabajan con grafos de conocimiento, permite responder eficazmente a preguntas como ¿Quién trabaja con quién? o ¿Qué empresas están relacionadas con esta persona?
Conexión con Neo4j
El controlador de Python neo4j se conecta a una instancia de Neo4j. Utilice variables de entorno para el URI de conexión y las credenciales. Cierre siempre el controlador cuando 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()Consultas básicas de Cypher
Cypher es el lenguaje de consulta de Neo4j. El patrón principal es MATCH (n:Label {property: value})-[:RELATIONSHIP]->(m) RETURN m. Los corchetes contienen el tipo de relación y los paréntesis contienen los nodos.
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)Consultas parametrizadas
Utilice siempre consultas parametrizadas (por ejemplo, $name) en lugar de interpolación de cadenas. Esto evita la inyección de Cypher y mejora el rendimiento gracias a la caché de los planes de consulta.
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)')Escritura de datos en el grafo
Utilice MERGE para hacer upsert de nodos y relaciones. MERGE crea el nodo o la relación solo si aún no existe, lo que evita duplicados.
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')Generación de Cypher a partir del lenguaje natural
Un agente puede convertir preguntas en lenguaje natural en consultas de Cypher. Proporcione al LLM el esquema de su grafo como contexto y pídale que genere el Cypher adecuado.
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)Ejecución segura del Cypher generado
Antes de ejecutar el Cypher generado por un LLM, valídelo. Bloquee las instrucciones de modificación (CREATE, DELETE, SET) a menos que el agente necesite explícitamente acceso de escritura. Ejecútelo en una sesión de solo lectura siempre que sea posible.
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')Análisis y formato de los resultados de las consultas
Formatee los resultados de las consultas de Neo4j como una cadena legible o como un objeto estructurado que el LLM pueda interpretar. Gestione los resultados vacíos correctamente.
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?'))Herramienta del agente: consulta del grafo
Envoltorva las consultas de Neo4j en una herramienta del agente. La herramienta acepta una pregunta en lenguaje natural, genera Cypher, lo ejecuta de forma segura y devuelve los resultados con formato.
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')Recorrido de grafos de varios saltos
Las bases de datos de grafos son excelentes para las consultas de varios saltos: permiten encontrar entidades que están a N pasos de distancia. Por ejemplo, permiten encontrar empresas conectadas con una persona a través de dos o más intermediarios.
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')Agregación en Cypher
Cypher admite funciones de agregación: COUNT, COLLECT, AVG, MIN y MAX. Utilícelas para responder preguntas de resumen sobre el 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])Comprobación de conocimientos: Neo4j para agentes
Compruebe su comprensión del uso de Neo4j desde herramientas de agentes.
Resumen de las herramientas de agente de Neo4j
La integración de Neo4j en un agente implica: conectarse mediante el controlador de Python, utilizar consultas parametrizadas para evitar la inyección, generar Cypher a partir del lenguaje natural con ayuda de un LLM, validar las consultas antes de ejecutarlas, dar formato a los resultados para que el LLM los consuma y exponer la consulta del grafo como una herramienta del agente.
Aprende AI Agents con un tutor de IA — gratis
Escribe y ejecuta código real en tu navegador, obtén ayuda instantánea de un tutor de IA disponible 24/7 y continúa donde lo dejaste en la web o en la aplicación.
- Cursos
- 60
- Lecciones
- 239
Preguntas frecuentes
¿La lección «Consultas de Neo4j desde herramientas de agentes» es gratis?
Sí — el texto completo de «Consultas de Neo4j desde herramientas de agentes» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de AI Agents, actualiza a CoddyKit PRO. El curso de AI Agents incluye 4 lecciones en total.
¿Qué aprenderé en «Consultas de Neo4j desde herramientas de agentes»?
Generación de consultas Cypher, recorrido de grafos y herramientas para analizar resultados. Practicas AI Agents con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.
¿Necesito experiencia previa para empezar AI Agents?
No se requiere experiencia previa. AI Agents en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 2 de 4.
¿Cuánto tiempo toma la lección «Consultas de Neo4j desde herramientas de agentes»?
La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.
¿Puedo escribir y ejecutar código en esta lección de AI Agents?
Sí. Cada lección de AI Agents incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.
Todas las lecciones de este curso
- Extracción de entidades para grafos de conocimiento
- Consultas de Neo4j desde herramientas de agentes
- Combinación de recuperación vectorial y de grafos
- Creación de un agente aumentado con conocimiento