0Pricing
AI Agents · Aula

Consultas Neo4j a partir de ferramentas de agentes

Geração de consultas Cypher, percurso de grafos e ferramentas de análise de resultados.

Consultas Neo4j a partir de ferramentas de agentes é uma aula grátis de AI Agents no CoddyKit. Esta é a aula 2 de 4. Você pode ler a aula completa abaixo gratuitamente — depois pratica ao vivo no navegador com um editor de código integrado e um tutor de IA 24/7. Faz parte do caminho de aprendizado de AI Agents, e seu progresso é sincronizado entre a web e o app CoddyKit. O curso de AI Agents inclui 4 aulas no total.

Por Que Usar Neo4j para o Conhecimento de Agentes?

O Neo4j é um banco de dados de grafos otimizado para percorrer relações. Para agentes que trabalham com grafos de conhecimento, ele permite fazer perguntas como Quem trabalha com quem? ou Quais empresas estão conectadas a esta pessoa? de forma eficiente.

Conectando-se ao Neo4j

O driver Python neo4j conecta-se a uma instância do Neo4j. Use variáveis de ambiente para o URI de conexão e as credenciais. Sempre feche o driver quando terminar.

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 em Cypher

Cypher é a linguagem de consulta do Neo4j. O padrão básico é MATCH (n:Label {property: value})-[:RELATIONSHIP]->(m) RETURN m. Os colchetes contêm o tipo de relação, e os parênteses contêm os nós.

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

Sempre use consultas parametrizadas (por exemplo, $name) em vez de interpolação de strings. Isso evita injeção de Cypher e melhora o desempenho por meio do armazenamento em cache do plano 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)')

Gravando Dados no Grafo

Use MERGE para criar ou atualizar nós e relações. MERGE cria o nó ou a relação somente se ele ou ela ainda não existir, evitando duplicatas.

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

Gerando Cypher a Partir de Linguagem Natural

Um agente pode converter perguntas em linguagem natural em consultas Cypher. Forneça ao LLM o esquema do seu grafo como contexto e peça que ele gere o Cypher apropriado.

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)

Executando Cypher Gerado com Segurança

Antes de executar Cypher gerado por um LLM, valide-o. Bloqueie instruções de mutação (CREATE, DELETE, SET), a menos que o agente precise explicitamente de acesso de escrita. Execute em uma sessão somente leitura sempre que possível.

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

Analisando e Formatando Resultados de Consultas

Formate os resultados de consultas do Neo4j em uma string legível ou em um objeto estruturado para interpretação pelo LLM. Trate os resultados vazios adequadamente.

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

Ferramenta do Agente: Consulta ao Grafo

Transforme as consultas ao Neo4j em uma ferramenta de agente. A ferramenta aceita uma pergunta em linguagem natural, gera Cypher, executa-o com segurança e retorna resultados formatados.

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

Travessia de Grafos com Múltiplos Saltos

Os bancos de dados de grafos são excelentes para consultas com múltiplos saltos: encontrar entidades que estão a N etapas de distância. Por exemplo, encontrar empresas conectadas a uma pessoa por meio de dois ou mais intermediários.

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

Agregação em Cypher

Cypher oferece funções de agregação: COUNT, COLLECT, AVG, MIN, MAX. Use-as para responder a perguntas de resumo sobre o 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ção de Conhecimento: Neo4j para Agentes

Teste sua compreensão sobre o uso do Neo4j a partir de ferramentas de agentes.

Resumo das Ferramentas de Agente do Neo4j

Integrar o Neo4j a um agente envolve: conectar-se com o driver Python, usar consultas parametrizadas para evitar injeção, gerar Cypher a partir de linguagem natural com a ajuda de um LLM, validar consultas antes da execução, formatar resultados para consumo pelo LLM e disponibilizar a consulta ao grafo como uma ferramenta de agente.

Perguntas Frequentes

A aula “Consultas Neo4j a partir de ferramentas de agentes” é grátis?

Sim — o texto completo de “Consultas Neo4j a partir de ferramentas de agentes” é grátis para ler aqui na web. Para praticá-la interativamente (um editor de código integrado e um tutor de IA 24/7) e desbloquear o restante do curso de AI Agents, atualize para CoddyKit PRO. O curso de AI Agents inclui 4 aulas no total.

O que vou aprender em “Consultas Neo4j a partir de ferramentas de agentes”?

Geração de consultas Cypher, percurso de grafos e ferramentas de análise de resultados. Você pratica AI Agents com código prático que executa diretamente no navegador, e um tutor de IA 24/7 responde suas dúvidas enquanto trabalha na aula.

Preciso ter experiência prévia para começar AI Agents?

Nenhuma experiência prévia é necessária. AI Agents no CoddyKit é estruturado para alunos iniciantes até avançados, então você pode começar aqui ou desde o início e aprender no seu ritmo. Esta é a aula 2 de 4.

Quanto tempo leva a aula “Consultas Neo4j a partir de ferramentas de agentes”?

A maioria das aulas CoddyKit leva cerca de 5–10 minutos. Cada uma é compacta e interativa, então você faz progresso constante e retoma exatamente de onde parou entre web e app.

Posso escrever e executar código nesta aula de AI Agents?

Sim. Cada aula de AI Agents inclui um editor de código integrado, então você escreve e executa código real direto no navegador e recebe feedback de IA instantaneamente — nenhuma configuração local necessária.

Todas as aulas deste curso

  1. Extração de entidades para grafos de conhecimento
  2. Consultas Neo4j a partir de ferramentas de agentes
  3. Combinando recuperação vetorial e em grafos
  4. Construindo um agente aumentado por conhecimento
← Voltar para AI Agents