0Pricing
AI Agents · Lektion

Neo4j-Abfragen aus Agent-Tools

Generierung von Cypher-Abfragen, Traversierung von Graphen und Tools zum Parsen von Ergebnissen.

Neo4j-Abfragen aus Agent-Tools ist eine kostenlose AI Agents-Lektion auf CoddyKit. Dies ist Lektion 2 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des AI Agents-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der AI Agents-Kurs umfasst insgesamt 4 Lektionen.

Warum Neo4j für Agentenwissen?

Neo4j ist eine Graphdatenbank, die für das Durchlaufen von Beziehungen optimiert ist. Für Agenten, die mit Wissensgraphen arbeiten, ermöglicht sie effizient Fragen wie Wer arbeitet mit wem zusammen? oder Welche Unternehmen sind mit dieser Person verbunden?

Mit Neo4j verbinden

Der neo4j-Python-Treiber stellt eine Verbindung zu einer Neo4j-Instanz her. Verwenden Sie Umgebungsvariablen für die Verbindungs-URI und die Zugangsdaten. Schließen Sie den Treiber immer, wenn Sie ihn nicht mehr benötigen.

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

Grundlegende Cypher-Abfragen

Cypher ist die Abfragesprache von Neo4j. Das grundlegende Muster lautet MATCH (n:Label {property: value})-[:RELATIONSHIP]->(m) RETURN m. In eckigen Klammern wird der Beziehungstyp angegeben, in runden Klammern stehen die Knoten.

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)

Parametrisierte Abfragen

Verwenden Sie immer parametrisierte Abfragen (z. B. $name) statt String-Interpolation. Dadurch verhindern Sie Cypher-Injection und verbessern die Leistung durch das Caching von Abfrageplänen.

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

Graphdaten schreiben

Verwenden Sie MERGE, um Knoten und Beziehungen per Upsert anzulegen. MERGE erstellt den Knoten oder die Beziehung nur, wenn sie noch nicht vorhanden sind, und verhindert so Duplikate.

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 aus natürlicher Sprache generieren

Ein Agent kann Fragen in natürlicher Sprache in Cypher-Abfragen umwandeln. Stellen Sie dem LLM das Schema Ihres Graphen als Kontext bereit und bitten Sie es anschließend, die passende Cypher-Abfrage zu generieren.

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)

Generiertes Cypher sicher ausführen

Validieren Sie von einem LLM generiertes Cypher, bevor Sie es ausführen. Blockieren Sie Anweisungen, die Änderungen vornehmen (CREATE, DELETE, SET), sofern der Agent nicht ausdrücklich Schreibzugriff benötigt. Führen Sie die Abfrage nach Möglichkeit in einer schreibgeschützten Sitzung aus.

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

Abfrageergebnisse analysieren und formatieren

Formatieren Sie Neo4j-Abfrageergebnisse als für Menschen lesbare Zeichenkette oder als strukturiertes Objekt, das vom LLM interpretiert werden kann. Behandeln Sie leere Ergebnisse angemessen.

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

Agenten-Tool: Graphabfrage

Kapseln Sie Neo4j-Abfragen als Agenten-Tool. Das Tool nimmt eine Frage in natürlicher Sprache entgegen, generiert Cypher, führt es sicher aus und gibt formatierte Ergebnisse zurück.

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

Multi-Hop-Graphdurchlauf

Graphdatenbanken eignen sich besonders für Multi-Hop-Abfragen: Sie finden Entitäten, die N Schritte entfernt sind. Ein Beispiel ist die Suche nach Unternehmen, die über zwei oder mehr Zwischenstationen mit einer Person verbunden sind.

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

Aggregation in Cypher

Cypher unterstützt Aggregationsfunktionen: COUNT, COLLECT, AVG, MIN, MAX. Verwenden Sie diese, um zusammenfassende Fragen zum Graphen zu beantworten.

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

Wissenscheck: Neo4j für Agenten

Überprüfen Sie Ihr Verständnis der Verwendung von Neo4j in Agenten-Tools.

Zusammenfassung der Neo4j-Agenten-Tools

Die Integration von Neo4j in einen Agenten umfasst: die Verbindung über den Python-Treiber, die Verwendung parametrisierter Abfragen zur Vermeidung von Injection, die Generierung von Cypher aus natürlicher Sprache mit Unterstützung eines LLM, die Validierung von Abfragen vor ihrer Ausführung, die Formatierung von Ergebnissen für die Verarbeitung durch ein LLM und die Bereitstellung der Graphabfrage als Agenten-Tool.

Häufig gestellte Fragen

Ist die Lektion „Neo4j-Abfragen aus Agent-Tools“ kostenlos?

Ja — der vollständige Text von „Neo4j-Abfragen aus Agent-Tools“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des AI Agents-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der AI Agents-Kurs umfasst insgesamt 4 Lektionen.

Was lerne ich in „Neo4j-Abfragen aus Agent-Tools“?

Generierung von Cypher-Abfragen, Traversierung von Graphen und Tools zum Parsen von Ergebnissen. Du übst AI Agents mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.

Brauche ich Erfahrung, um AI Agents zu starten?

Keine Vorkenntnisse erforderlich. AI Agents auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 2 von 4.

Wie lange dauert die Lektion „Neo4j-Abfragen aus Agent-Tools“?

Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.

Kann ich in dieser AI Agents-Lektion Code schreiben und ausführen?

Ja. Jede AI Agents-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.

Alle Lektionen in diesem Kurs

  1. Entitätsextraktion für Wissensgraphen
  2. Neo4j-Abfragen aus Agent-Tools
  3. Vektor- und Graphen-Retrieval kombinieren
  4. Einen wissensgestützten Agenten erstellen
← Zurück zu AI Agents