Tự dựng Text-to-SPARQL truy vấn knowledge graph bằng LLM mã nguồn mở

flowchart TD
    RDF[("File RDF")]

    subgraph INGEST ["1 · ingest — chụp ontology (chạy một lần / đồ thị)"]
        direction LR
        I_INTRO["introspect:<br/>lớp + hình dạng thuộc tính"] --> REG[("Registry JSON<br/>ontology + prefixes")]
    end

    subgraph ASK ["2 · ask — sinh SPARQL từ câu hỏi"]
        direction TB
        A_Q["Câu hỏi<br/>(ngôn ngữ tự nhiên)"] --> A_PROMPT["Prompt = ontology (từ registry)<br/>+ BỘ BA DỮ LIỆU THẬT + câu hỏi"]
        A_SAMPLE["sample_triples:<br/>vài bộ ba thật"] --> A_PROMPT
        A_PROMPT --> A_LLM["LLM"] -->|sinh| A_SPARQL["SPARQL"]
    end

    subgraph EXEC ["3 · exec — thực thi truy vấn"]
        direction LR
        E_RUN["run_query<br/>(chỉ-đọc)"] --> E_RESULT["Kết quả"]
    end

    RDF -->|introspect| I_INTRO
    REG -.->|ontology đã cache| A_PROMPT
    RDF -.->|lấy mẫu| A_SAMPLE
    A_SPARQL --> E_RUN
    RDF -->|truy vấn| E_RUN

text2sparql-simple/
├── pyproject.toml          # metadata + phụ thuộc + entry point CLI
├── .env.example            # cấu hình mẫu (endpoint LLM, model, …)
├── text2sparql/
│   ├── __init__.py
│   ├── config.py           # đọc cấu hình từ biến môi trường
│   ├── graph.py            # RDF: nội suy ontology + thực thi SPARQL  ← xử lý chính
│   ├── llm.py              # gọi LLM qua OpenAI SDK
│   ├── engine.py           # điều phối: ingest + build prompt + self-repair
│   ├── cli.py              # giao diện dòng lệnh
└── examples/
    ├── shop_demo.ttl       # đồ thị mẫu 1: cửa hàng (schema.org)
    ├── acme_kg.ttl         # đồ thị mẫu 2: công ty (cây phân cấp, property path)

[project]
name = "text2sparql"
requires-python = ">=3.10"
dependencies = [
    "rdflib>=7.0",        # đọc mọi định dạng RDF + chạy SPARQL trong bộ nhớ
    "openai>=1.40",       # client cho MỌI endpoint tương thích OpenAI (kể cả Ollama)
    "python-dotenv>=1.0", # nạp .env
]

[project.scripts]
text2sparql = "text2sparql.cli:main"   # tạo lệnh `text2sparql` sau khi cài

cd text2sparql-simple
python -m venv .venv && source .venv/bin/activate
pip install -e .
cp .env.example .env       # trỏ tới endpoint LLM của bạn

# LLM chạy nội bộ qua Ollama:
ollama pull gemma4:31b
ollama serve &

@dataclass
class Settings:
    # LLM sinh SPARQL, gọi qua OpenAI SDK. Bất kỳ endpoint tương thích OpenAI nào cũng chạy:
    llm_base_url: str = os.getenv("LLM_BASE_URL", "http://localhost:11434/v1")
    llm_model: str = os.getenv("LLM_MODEL", "gemma4:31b")
    llm_api_key: str = os.getenv("LLM_API_KEY", "")   # chỉ cần cho endpoint hosted
    request_timeout: float = float(os.getenv("TEXT2SPARQL_TIMEOUT", "180"))

    # Self-repair: tổng số lần sinh. Nếu truy vấn báo LỖI, lỗi được đưa ngược cho model sửa.
    max_attempts: int = int(os.getenv("TEXT2SPARQL_MAX_ATTEMPTS", "2"))

    data_dir: Path = Path(os.getenv("TEXT2SPARQL_DATA_DIR", ".text2sparql"))

SCHEMA = Namespace("https://schema.org/")

class GraphSource:
    def __init__(self, source: str):
        self.source = source
        # bind_namespaces="none": KHÔNG nạp sẵn bảng prefix khổng lồ của rdflib
        self.graph = Graph(bind_namespaces="none")
        self.graph.parse(source)   # rdflib tự nhận diện định dạng
        # Tạo sẵn vài prefix cơ bản để SPARQL sinh ra đọc tự nhiên; @prefix của đồ thị được ưu tiên.
        for pfx, ns in (("rdf", RDF), ("rdfs", RDFS), ("owl", OWL),
                        ("xsd", XSD), ("skos", SKOS), ("schema", SCHEMA)):
            self.graph.namespace_manager.bind(pfx, ns, replace=False)

meta_filter = " ".join(f'FILTER(!STRSTARTS(STR(?cls), "{ns}"))' for ns in _META_NAMESPACES)
class_q = f"""
SELECT DISTINCT ?cls WHERE {{
  {{ ?cls a <{OWL.Class}> }} UNION {{ ?cls a <{RDFS.Class}> }} UNION {{ ?s a ?cls }}
  FILTER(isIRI(?cls))
  {meta_filter}
}} LIMIT {MAX_CLASSES}
"""

prop_q = f"""
SELECT DISTINCT ?p ?dt ?otype WHERE {{
  ?s a <{iri}> ; ?p ?o .
  OPTIONAL {{ ?o a ?otype }}      # nếu đối tượng là một thực thể có kiểu
  BIND(DATATYPE(?o) AS ?dt)       # nếu đối tượng là literal
}} LIMIT {SAMPLE_LIMIT}
"""

sup_q = f"SELECT DISTINCT ?sup WHERE {{ <{iri}> <{RDFS.subClassOf}> ?sup . FILTER(isIRI(?sup)) }}"
for r in gs.query(sup_q):
    info.super_classes.append(gs.curie(str(r[0])))

Class: schema:Order
Instances: 10
Properties (predicate -> range):
  - schema:customer -> schema:Person
  - schema:orderStatus -> xsd:string
  - schema:orderDate -> xsd:dateTime
  - schema:orderedItem -> schema:OrderItem

def sample_triples(source, class_iris, max_subjects=12, per_class=2) -> str:
    gs = GraphSource(source)
    for iri in class_iris:
        # ORDER BY ?s để mẫu ỔN ĐỊNH giữa các lần chạy — LIMIT không sắp xếp sẽ tuỳ tiện,
        # khiến việc sinh truy vấn (và do đó cả eval) chập chờn.
        for r in gs.query(f"SELECT DISTINCT ?s WHERE {{ ?s a <{iri}> }} ORDER BY ?s LIMIT {per_class}"):
            # ... gom mọi bộ ba của ?s, in ra dạng Turtle rút gọn với CURIE

ex:raj  a org:Manager, org:Employee, org:Person, org:Agent ;
  schema:name "Raj Patel" ; org:reportsTo ex:sofia ; org:hasSkill ex:python, ex:ml .

def run_query(source: str, sparql: str) -> QueryResult:
    """Chạy truy vấn SELECT/ASK chỉ-đọc và chuẩn hoá kết quả. Giới hạn MAX_ROWS dòng."""
    gs = GraphSource(source)
    result = gs.query(sparql)

    if result.type == "ASK":                       # ASK → một boolean
        return QueryResult(["ask"], [[bool(result.askAnswer)]], 1, False)

    columns = [str(v) for v in result.vars]        # SELECT → binding của biến
    fetched = list(result)
    truncated = len(fetched) > MAX_ROWS
    rows = [[_json_safe(row[v]) for v in result.vars] for row in fetched[:MAX_ROWS]]
    return QueryResult(columns, rows, len(rows), truncated)

def generate_sparql(system_prompt: str, user_prompt: str) -> dict:
    """Hỏi LLM sinh SPARQL. Trả về {'sparql', 'explanation', 'classes_used'}."""
    client = OpenAI(
        base_url=settings.llm_base_url,
        api_key=settings.llm_api_key or "ollama",   # Ollama bỏ qua key nhưng SDK cần chuỗi khác rỗng
        timeout=settings.request_timeout,
    )
    resp = client.chat.completions.create(
        model=settings.llm_model,
        messages=[{"role": "system", "content": system_prompt},
                  {"role": "user", "content": user_prompt}],
        temperature=0,                               # xác định, để eval ổn định
        response_format={"type": "json_object"},     # ép trả JSON
    )
    return _parse(resp.choices[0].message.content or "")

def _parse(content: str) -> dict:
    try:
        data = json.loads(content)
    except (json.JSONDecodeError, TypeError):
        i, j = content.find("{"), content.rfind("}")
        data = json.loads(content[i:j+1]) if i != -1 and j > i else None
    if not isinstance(data, dict):
        return {"sparql": "", "explanation": "", "classes_used": []}   # để vòng self-repair xử lý
    return {"sparql": (data.get("sparql") or "").strip(), ...}

SYSTEM_PROMPT = """You are an expert in SPARQL and RDF knowledge graphs. Convert the user's
question into ONE SPARQL query for the graph described below.

You are given the graph's prefixes, its ontology (classes + properties), and a block of REAL
example triples. Ground your query in that example data: use its exact prefixes and IRIs, and
follow how the entities actually link together. Never invent classes, properties, or IRIs.

Respond with ONLY a JSON object:
{"sparql": "...", "explanation": "...", "classes_used": ["prefix:Class", ...]}
..."""

reg[graph_id] = {
    "name": name or graph_id,
    "source": source,
    "classes": ids,
    "documents": documents,   # tài liệu mô tả từng lớp — chính là context đưa cho model
    "prefixes": prefixes,
}
reg["_latest"] = graph_id     # nhớ đồ thị vừa nạp để lệnh sau khỏi phải chỉ --graph-id

def _build_prompt(question: str, meta: dict) -> str:
    context = "\n\n".join(meta.get("documents", []))         # ontology đã cache
    prefix_table = _prefix_block(meta.get("prefixes", {}))
    context_iris = [_expand(i, prefixes) for i in meta["classes"]]
    try:
        examples = graph.sample_triples(meta["source"], context_iris)
    except Exception:
        examples = ""   # lấy mẫu là best-effort; không bao giờ chặn việc sinh vì nó
    return (
        f"# Prefixes available:\n{prefix_table}\n\n"
        f"# Ontology (classes + property shapes):\n{context}\n\n"
        f"# Example instance data (REAL triples ...):\n{examples}\n\n"
        f"# Question: {question}"
    )

def ask(question: str, graph_id=None) -> str:
    _, meta = _resolve(graph_id)
    return _to_ask_result(llm.generate_sparql(SYSTEM_PROMPT, _build_prompt(question, meta))).sparql

def execute(sparql: str, graph_id=None) -> "graph.QueryResult":
    _, meta = _resolve(graph_id)
    return graph.run_query(meta["source"], sparql)   # chỉ-đọc

for attempt in range(1, settings.max_attempts + 1):
    last = _to_ask_result(llm.generate_sparql(SYSTEM_PROMPT, prompt))
    if not last.sparql:
        error = "model produced no SPARQL"
    else:
        try:
            qr = graph.run_query(meta["source"], last.sparql)
        except Exception as e:
            error = str(e)                 # lỗi → nạp vào prompt sửa
        else:
            return AnswerResult(ask=last, result=qr, attempts=attempt, error=None)
    if attempt < settings.max_attempts:
        prompt = base_prompt + _REPAIR.format(sparql=last.sparql or "(empty)", problem=error)

if args.cmd == "ingest":
    res = engine.ingest(args.source, args.name)
    print(f"Ingested graph_id={res.graph_id} ... with {res.class_count} classes.")
    if not args.quiet:
        print("\nOntology captured (this is the context handed to the model):")
        for doc in res.documents:
            print(doc); print("-" * 60)

text2sparql ingest examples/shop_demo.ttl
text2sparql ask "Which customer placed the most orders?"

PREFIX ex: <http://example.org/shop#>
PREFIX schema: <https://schema.org/>
SELECT ?customerName
WHERE {
  ?order a schema:Order ;
          schema:customer ?customer .
  ?customer schema:name ?customerName .
}
GROUP BY ?customer ?customerName
ORDER BY DESC(COUNT(?order))
LIMIT 1

text2sparql exec "PREFIX ex: <http://example.org/shop#> PREFIX schema: <https://schema.org/> SELECT ?customerName WHERE { ?order a schema:Order ; schema:customer ?customer . ?customer schema:name ?customerName . } GROUP BY ?customer ?customerName ORDER BY DESC(COUNT(?order)) LIMIT 1"

customerName
------------------------------------------------------------
Alice Nguyen

text2sparql ingest examples/acme_kg.ttl
text2sparql ask "Who does Tom Becker report to, directly or indirectly?"

PREFIX ex: <http://example.org/data#>
PREFIX org: <http://www.w3.org/ns/org#>
PREFIX schema: <https://schema.org/>
SELECT ?managerName WHERE {
  ?tom schema:name "Tom Becker" .
  ?tom org:reportsTo+ ?manager .
  ?manager schema:name ?managerName .
}

managerName
------------------------------------------------------------
Mei Lin
Raj Patel
Sofia Reyes