text2cypher

Generate validated Cypher queries from natural language text with LLMs Ask a question in plain language and get back a read-only Cypher query that has been checked against your FalkorDB graph.

text2cypher asks an LLM to write the query, checks it against the real graph with cypher-guardrail, and sends what is wrong back to the LLM until the query passes or the attempts run out. A query that fails the checks is never handed back as if it were fine.

• Grounded in your graph: the prompt carries the schema read from the graph itself (labels, relationship types, properties with their value types, the paths that exist, and the indexes), cut down to what the question is about.
• Validated before it is returned: syntax, schema (labels, relationship types, paths and their direction) and property checks. A query that could write to the graph is never valid.
• Self-correcting: the problems found are written back to the LLM. If the cut-down schema does not get there, the loop moves to the full one.
• Honest outcomes: valid, unanswerable (the LLM declined) or failed (the corrections ran out), with every attempt, its issues, the schema it used and its token usage.
• Optional execution: run the final query read-only, with a timeout, and get the rows.
• Bring your own LLM: a one-method protocol, with an OpenAI client included. Async throughout.

Installation

Python 3.12 or later.

uv add text2cypher        # or: pip install text2cypher

You need a FalkorDB graph that has data, reachable with the async client (falkordb.asyncio), and a model. The included client uses OpenAI's API and reads OPENAI_API_KEY.

Quickstart

import asyncio

from falkordb.asyncio import FalkorDB
from text2cypher import OpenAIClient, Text2Cypher


async def main() -> None:
    client = FalkorDB(host="localhost", port=6379)  # one client per event loop
    agent = Text2Cypher(client, OpenAIClient("your-model"))

    result = await agent.run("Which actors appeared in The Matrix?", graph="movies")
    print(result.status)  # "valid", "unanswerable" or "failed"
    print(
        result.query
    )  # MATCH (p:Person)-[:ACTED_IN]->(m:Movie {title: 'The Matrix'}) RETURN p.name

    await client.aclose()


asyncio.run(main())

Always look at result.status. valid means the query passed the checks, not that it means what you meant (see below).

There is also a quickstart that needs no model and no server: it starts an embedded FalkorDB and answers with a scripted LLM, including one correction.

What is and is not checked

• Checked: the query is valid Cypher for FalkorDB, it cannot write, its parameters are provided (the prompt asks for values written in the query, so a $name is a problem), the labels, relationship types and paths exist in the direction used, and every property read exists on the label or relationship type it is read from.
• Not checked: that the query means what the question meant. A query can be valid and still answer another question, for example by using a property with a similar name. Look at the query, or at the rows, before you rely on them.
• Early days: the prompts and the loop have been tried with small models and small graphs, and the filter has not been measured on a real large schema. Expect them to need work.

Documentation

The full documentation is at text2cypher.readthedocs.io:

• Using text2cypher: the result, the configuration, running the query, errors
• How it works: the loop, the escalation and what it costs
• The schema and its filter: what the LLM is shown, your own filter, keeping the schema between runs
• LLM clients: your own client, other OpenAI-compatible models, FakeLLM for tests
• What is and is not checked and the API reference

For contributors: how it is built, and why, and a harness that replays questions through the real pipeline with replies written by a model or a person.

Development

uv sync                              # create the environment (Python >= 3.12 also installs FalkorDBLite for the tests)
uv run pytest                        # the tests, on an embedded FalkorDB
uv run pytest -m "not db"            # only the tests that need no database (under ten seconds)
uv run ruff check . && uv run ruff format --check . && uv run ty check && uv run lint-imports
uv run coverage run -m pytest && uv run coverage report

Contributions are welcome: please open an issue or a pull request.

License

Apache License 2.0: see LICENSE.