Graph Databases Agent

A multi-agent knowledge graph comparison that exercises 4 graph database instrumentors -- FalkorDB (Graph.query, Graph.ro_query), ArangoDB (AQL.execute, StandardCollection.insert), Memgraph (Memgraph.execute, Memgraph.execute_and_fetch), and Neptune (BaseClient._make_api_call for Gremlin and openCypher) -- through 3 child agents: a graph-builder (creates nodes and edges in all 4 databases), a graph-querier (queries all databases and ranks results with @retrieval and @reasoning), and a graph-synthesizer (LLM synthesis with quality assessment). Builds an AI knowledge graph with 6 entities and 6 relationships.

Environment variables

This example runs in dry-run mode by default (no API key needed). For live mode, set OPENAI_API_KEY, WAXELL_API_KEY, and WAXELL_API_URL.

Architecture

Key Code

Graph database tool operations

Each database's exact instrumentor methods are wrapped with @waxell.tool(tool_type="graph_db").

@waxell.tool(tool_type="graph_db")
def falkordb_create_graph(graph, entities: list, edges: list):
    for entity in entities:
        graph.query(f"CREATE (n:{entity['type']} {{name: '{entity['name']}'}})")
    for src, rel, tgt in edges:
        graph.query(f"MATCH (a {{name: '{src}'}}), (b {{name: '{tgt}'}}) CREATE (a)-[:{rel}]->(b)")
    return {"nodes_created": len(entities), "edges_created": len(edges)}

@waxell.tool(tool_type="graph_db")
def arangodb_traverse(aql, start_vertex: str):
    result = aql.execute("FOR v, e IN 1..2 OUTBOUND @start edges RETURN {vertex: v, edge: e}",
                         bind_vars={"start": start_vertex})
    return {"traversal_results": len(result), "results": list(result)}

@waxell.tool(tool_type="graph_db")
def neptune_gremlin_query(neptune, gremlin_query: str):
    result = neptune.execute_gremlin_query(gremlinQuery=gremlin_query)
    return {"result_count": len(result["result"]["data"]), "results": result["result"]["data"]}

Result evaluation and quality assessment

The querier evaluates results with @reasoning; the synthesizer assesses answer quality.

@waxell.reasoning_dec(step="evaluate_results")
async def evaluate_graph_results(db_results: dict) -> dict:
    fastest = min(db_results.items(), key=lambda x: x[1]["latency_ms"])
    most_results = max(db_results.items(), key=lambda x: x[1]["results"])
    return {
        "thought": f"Compared {len(db_results)} databases. {fastest[0]} was fastest.",
        "evidence": [f"{name}: {info['results']} results in {info['latency_ms']}ms"
                     for name, info in db_results.items()],
        "conclusion": f"Best for low-latency: {fastest[0]}. Best coverage: {most_results[0]}.",
    }

@waxell.reasoning_dec(step="quality_assessment")
async def assess_synthesis_quality(answer: str, db_count: int) -> dict:
    dbs_mentioned = sum(1 for db in ["falkordb", "arangodb", "memgraph", "neptune"]
                        if db.lower() in answer.lower())
    return {
        "thought": f"Answer mentions {dbs_mentioned}/{db_count} databases.",
        "conclusion": "Answer covers all databases" if dbs_mentioned >= db_count
                      else f"Answer missing {db_count - dbs_mentioned} database(s)",
    }

What this demonstrates

• @waxell.tool(tool_type="graph_db") -- 8 graph database tool calls covering all 4 instrumentors' wrapt-target methods.
• @waxell.retrieval(source="graph_db") -- cross-database result collection and ranking by latency.
• @waxell.reasoning_dec -- 2 reasoning chains (result evaluation + quality assessment).
• @waxell.decision -- query strategy selection (traversal/pattern_match/aggregation) via OpenAI + output format decision.
• waxell.decide() -- inline database routing decision.
• @waxell.step_dec -- query preprocessing step.
• waxell.score() -- query_coverage, synthesis_quality, and completeness scores.
• Auto-instrumented LLM calls -- 2 OpenAI calls (strategy + synthesis).
• Nested @waxell.observe -- orchestrator + 3 child agents (graph-builder, graph-querier, graph-synthesizer).
• 4 graph databases compared -- FalkorDB (in-memory Cypher), ArangoDB (multi-model AQL), Memgraph (in-memory Cypher), Neptune (managed Gremlin + openCypher).
• AI knowledge graph -- 6 entities (BERT, GPT-4, Transformer, Attention, OpenAI, Google) and 6 relationships.

Run it

# Dry-run (no API key needed)
python -m app.demos.graph_db_agent --dry-run

# Live mode with OpenAI
OPENAI_API_KEY=sk-... python -m app.demos.graph_db_agent

Source

dev/waxell-dev/app/demos/graph_db_agent.py