graphrag

Graph-structured retrieval-augmented generation with semantic operations over a knowledge graph.

Run it from sema/:

sema check examples/graphrag
SEMA_STRICT=1 sema run examples/graphrag
sema assure examples/graphrag --grade silver

Source

src/main.sema

# GraphRAG backend in Sema — entry point. The implementation is split across
# modules (types / embed / similarity / store / api) to demonstrate proper
# folder + namespace modularity (§5.35). One-to-one with the Python reference
# (experiments/graphrag/SPEC.md); deterministic and bit-comparable.

# Standard-library modules are imported explicitly: `math` (via the libraries),
# `io` (files + stdio), `http` (the API server). Effect capabilities (fs, net,
# ...) stay declared in the `!{...}` effect rows.
import io
import http
from graphrag.store import GraphRAG, build_index
from graphrag.embed import embed
from graphrag.api import handle, json_ints, json_floats, json_answer

# ---- host entry points (called in-process by the Python/TS bridge) ---------
# The index is built once and cached in a module-level dict, so repeated host
# calls reuse it (constant work per call). These take/return strings (JSON), the
# neutral boundary the embedding API uses (INTEROP.md §4).

_CACHE = dict([])

def load_corpus() -> list[str] !{fs.read}:
    raw = io.lines("corpus.txt")
    lines = []
    for l in raw:
        if len(l.strip()) > 0:
            lines.append(l)
    return lines

def get_index() -> GraphRAG !{fs.read}:
    if not _CACHE.has("g"):
        _CACHE.set("g", build_index(load_corpus(), 96, 32, 3))
    return _CACHE.get("g")

def api_query(q: str) -> str !{fs.read}:
    return json_answer(get_index().query(q, 3))

def api_search(q: str) -> str !{fs.read}:
    return "{\"ids\": " + json_ints(get_index().full_text_search(q)) + "}"

def main() -> None !{fs.read, net.connect, net.listen, observe.record, clock.read, env.read, ui.render}:
    dim = 96
    kdim = 32
    kn = 3
    top_k = 3
    lines = load_corpus()

    t0 = clock.mono_ms()
    g = build_index(lines, dim, kdim, kn)
    t1 = clock.mono_ms()
    log.info("indexed corpus", n=len(g.chunks), build_ms=round(t1 - t0))

    queries = [
        "approximate nearest neighbor graph",
        "how does retrieval augmented generation work",
        "quantized models on apple gpu",
        "cosine similarity of normalized embeddings",
        "memory safety in language runtimes",
    ]
    t2 = clock.mono_ms()
    for q in queries:
        a = g.query(q, top_k)
        log.info("query", q=q, sources=a.sources, answer=a.answer)
    t3 = clock.mono_ms()
    log.info("query set done", ms=round(t3 - t2))

    log.info("full-text 'graph'", ids=g.full_text_search("graph"))
    log.info("full-text 'cosine'", ids=g.full_text_search("cosine"))

    # Native tensor showcase: project one embedding via the stored matrix.
    e = tensor(g.chunks.first().vec)
    log.info("tensor matmul projection", dims=shape(matmul(g.proj, e)))

    # Machine-readable parity dump for cross-language checking (SEMA_PARITY=1).
    if env.var("SEMA_PARITY") == "1":
        probes = ["transformer self attention", "vector database cosine", "rust memory safety"]
        pj = "{\"embeddings\": ["
        i = 0
        for p in probes:
            if i > 0:
                pj = pj + ", "
            pj = pj + json_floats(embed(p, dim))
            i = i + 1
        pj = pj + "], \"queries\": ["
        i = 0
        for q in queries:
            if i > 0:
                pj = pj + ", "
            pj = pj + json_ints(g.query(q, top_k).sources)
            i = i + 1
        pj = pj + "]}"
        io.print("PARITY:" + pj + "\n")

    # Serve the API when asked (SEMA_SERVE=1); the default demo run just exits.
    # The port is configurable (SEMA_PORT) so an embedding host can pick a free
    # one — this is what the Python/TS interop bridge uses (INTEROP.md).
    if env.var("SEMA_SERVE") == "1":
        p = env.var("SEMA_PORT")
        port = 8080
        if len(p) > 0:
            port = int(p)
        log.info("serving", url="http://127.0.0.1:" + str(port) + "  (/query?q=...  /search?q=...)")
        http.serve(port, lambda req: handle(g, req))

src/api.sema

# The HTTP API surface: JSON encoding + the request handler. Depends on the
# store (for the GraphRAG type) and the shared Answer type (§5.35).

from graphrag.types import Answer
from graphrag.store import GraphRAG

def json_ints(ids: list[int]) -> str !{}:
    out = "["
    i = 0
    for x in ids:
        if i > 0:
            out = out + ", "
        out = out + str(x)
        i = i + 1
    return out + "]"

def json_floats(xs: list[f64]) -> str !{}:
    out = "["
    i = 0
    for x in xs:
        if i > 0:
            out = out + ", "
        out = out + str(x)
        i = i + 1
    return out + "]"

def json_answer(a: Answer) -> str !{}:
    txt = a.answer.replace("\"", "'")
    return "{\"answer\": \"" + txt + "\", \"sources\": " + json_ints(a.sources) + "}"

def qparam(query: str, key: str) -> str !{}:
    for p in query.split("&"):
        kv = p.split("=")
        if kv.first() == key:
            if len(kv) > 1:
                return kv[1]
            return ""
    return ""

# GET /query?q=...  -> {answer, sources};  GET /search?q=... -> {query, ids}
def handle(g: GraphRAG, req: dict) -> str !{}:
    q = qparam(req.get("query"), "q")
    if len(q) == 0:
        return "{\"error\": \"missing q parameter\"}"
    if req.get("path") == "/search":
        return "{\"query\": \"" + q + "\", \"ids\": " + json_ints(g.full_text_search(q)) + "}"
    return json_answer(g.query(q, 3))

src/embed.sema

# Feature-hashing embeddings + deterministic Johnson-Lindenstrauss projection.
# Pure numeric code — imports the `math` library explicitly (§5.35).

import math

# ---- feature-hashing embedding (dim D) -----------------------------------

def embed(text: str, dim: int) -> list[f64] !{}:
    v = [0.0] * dim                    # native fill, not an append loop
    for w in words(text):
        b = hash_int(w) % dim
        v[b] = v[b] + 1.0
    return l2norm(v)

# Normalize via native tensor ops: one dot for the norm, one scalar division
# for the whole vector (no interpreted per-element loop).
def l2norm(v: list[f64]) -> list[f64] !{}:
    t = tensor(v)
    n = math.sqrt(dot(t, t))
    if n == 0.0:
        return v
    return t / n

# ---- deterministic Johnson-Lindenstrauss projection (D -> K) --------------

def proj_entry(i: int, j: int, kdim: int) -> f64 !{}:
    h = hash_int("proj:" + str(i) + ":" + str(j))
    u = (h % 1000003) / 1000003.0
    return (u * 2.0 - 1.0) / math.sqrt(float(kdim))

def build_proj(kdim: int, dim: int) -> list[list[f64]] !{}:
    p = []
    for i in range(kdim):
        row = []
        for j in range(dim):
            row.append(proj_entry(i, j, kdim))
        p.append(row)
    return p

# Interpreted reference projection (kept for benchmarking vs the native path).
def project_loop(v: list[f64], proj: list[list[f64]], kdim: int, dim: int) -> list[f64] !{}:
    out = []
    for i in range(kdim):
        s = 0.0
        row = proj[i]
        for j in range(dim):
            s = s + row[j] * v[j]
        out.append(s)
    return out

# Native path: the whole K x D projection runs as one Rust-speed matmul on
# tensors. Same arithmetic (i-outer/j-inner) as project_loop, so bit-identical.
# `v` may be a list or a tensor; matmul coerces either.
def project(v, proj_t, kdim: int, dim: int) !{}:
    return matmul(proj_t, v)

def round6(x: f64) -> f64 !{}:
    return round(x * 1000000.0) / 1000000.0

src/similarity.sema

# Similarity + ranking helpers.

import math
from graphrag.types import Scored

def cosine(a: list[f64], b: list[f64]) -> f64 !{}:
    na = math.sqrt(dot(a, a))
    nb = math.sqrt(dot(b, b))
    if na == 0.0 or nb == 0.0:
        return 0.0
    return dot(a, b) / (na * nb)

# Descending by score, ties broken by ascending id — the stdlib sort provides
# exactly this total order (a Scored has .score and .id), so we use it directly
# instead of hand-rolling a comparator loop.
def rank(items: list[Scored]) -> list[Scored] !{}:
    return sort_by_score_desc(items)

def take(items: list[Scored], n: int) -> list[Scored] !{}:
    out = []
    i = 0
    for it in items:
        if i < n:
            out.append(it)
        i = i + 1
    return out

src/store.sema

# The vector store + kNN similarity graph + retrieval — the GraphRAG core.
# Imports its data types and the embed/similarity libraries (§5.35).

from graphrag.types import Chunk, Scored, Answer
from graphrag.embed import embed, project, build_proj, round6
from graphrag.similarity import cosine, rank, take

struct GraphRAG:
    chunks: list[Chunk]
    adj: dict                 # str(id) -> list[int]  (kNN neighbours)
    proj: Tensor              # projection matrix, K rows x D cols
    dim: int
    kdim: int

    # Coarse-to-fine vector search: rank all by projected cosine (cheap), then
    # re-rank the shortlist by full-dimensional cosine (accurate).
    def search(self, qv: list[f64], qp: list[f64], top_k: int) -> list[Scored] !{}:
        coarse = []
        for c in self.chunks:
            coarse.append(Scored(id=c.id, score=cosine(qp, c.pvec)))
        coarse = rank(coarse)
        m = top_k * 3
        fine = []
        i = 0
        for s in coarse:
            if i < m:
                fine.append(Scored(id=s.id, score=cosine(qv, self.chunks[s.id].vec)))
            i = i + 1
        fine = rank(fine)
        return take(fine, top_k)

    # Graph-expanded retrieval: seed via vector search, pull in 1-hop graph
    # neighbours, then re-rank the union by full cosine to the query.
    def query(self, question: str, top_k: int) -> Answer !{}:
        qv = embed(question, self.dim)
        qp = project(qv, self.proj, self.kdim, self.dim)
        seeds = self.search(qv, qp, top_k)
        seen = dict([])
        cand = []
        for s in seeds:
            add_unique(seen, cand, s.id)
            for nb in self.adj.get(str(s.id)):
                add_unique(seen, cand, nb)
        scored = []
        for cid in cand:
            scored.append(Scored(id=cid, score=cosine(qv, self.chunks[cid].vec)))
        ranked = rank(scored)
        top = take(ranked, top_k)
        sources = []
        scores = []
        for s in top:
            sources.append(s.id)
            scores.append(round6(s.score))
        return Answer(answer=self.chunks[top.first().id].text, sources=sources, scores=scores)

    # Literal (lexical) full-text search, ascending id.
    def full_text_search(self, term: str) -> list[int] !{}:
        t = term.lower()
        hits = []
        for c in self.chunks:
            if c.text.lower().contains(t):
                hits.append(c.id)
        return hits

def add_unique(seen: dict, out: list[int], id: int) -> None !{}:
    k = str(id)
    if not seen.has(k):
        seen.set(k, true)
        out.append(id)

# Build the whole index: embed + project every chunk, then a kNN graph over the
# full-dimensional cosine similarity.
def build_index(lines: list[str], dim: int, kdim: int, kn: int) -> GraphRAG !{}:
    proj = tensor(build_proj(kdim, dim))   # K x D projection matrix as a tensor
    chunks = []
    cid = 0
    for line in lines:
        vec = embed(line, dim)
        pvec = project(vec, proj, kdim, dim)
        chunks.append(Chunk(id=cid, text=line, vec=vec, pvec=pvec))
        cid = cid + 1
    adj = dict([])
    for c in chunks:
        scored = []
        for other in chunks:
            if other.id != c.id:
                scored.append(Scored(id=other.id, score=cosine(c.vec, other.vec)))
        ranked = rank(scored)
        neigh = []
        i = 0
        for s in ranked:
            if i < kn:
                neigh.append(s.id)
            i = i + 1
        adj.set(str(c.id), neigh)
    return GraphRAG(chunks=chunks, adj=adj, proj=proj, dim=dim, kdim=kdim)

src/types.sema

# Shared data types for the GraphRAG backend. Imported by the other modules —
# a single source of truth for the record shapes (demonstrates cross-module
# struct sharing, §5.35 modularity).

struct Chunk:
    id: int
    text: str
    vec: list[f64]      # full embedding, dim D
    pvec: list[f64]     # projected embedding, dim K

struct Scored:
    id: int
    score: f64

struct Answer:
    answer: str
    sources: list[int]
    scores: list[f64]

Reflected API

api

json_ints

def json_ints(ids: list[int]) -> str !{}

Parameters

name	type
ids	list[int]

Returns str

Effects !{}

json_floats

def json_floats(xs: list[f64]) -> str !{}

Parameters

name	type
xs	list[f64]

Returns str

Effects !{}

json_answer

def json_answer(a: Answer) -> str !{}

Parameters

name	type
a	Answer

Returns str

Effects !{}

qparam

def qparam(query: str, key: str) -> str !{}

Parameters

name	type
query	str
key	str

Returns str

Effects !{}

handle

def handle(g: GraphRAG, req: dict) -> str !{}

Parameters

name	type
g	GraphRAG
req	dict

Returns str

Effects !{}

embed

embed

def embed(text: str, dim: int) -> list[f64] !{}

Parameters

name	type
text	str
dim	int

Returns list[f64]

Effects !{}

l2norm

def l2norm(v: list[f64]) -> list[f64] !{}

Parameters

name	type
v	list[f64]

Returns list[f64]

Effects !{}

proj_entry

def proj_entry(i: int, j: int, kdim: int) -> f64 !{}

Parameters

name	type
i	int
j	int
kdim	int

Returns f64

Effects !{}

build_proj

def build_proj(kdim: int, dim: int) -> list[list[f64]] !{}

Parameters

name	type
kdim	int
dim	int

Returns list[list[f64]]

Effects !{}

project_loop

def project_loop(v: list[f64], proj: list[list[f64]], kdim: int, dim: int) -> list[f64] !{}

Parameters

name	type
v	list[f64]
proj	list[list[f64]]
kdim	int
dim	int

Returns list[f64]

Effects !{}

project

def project(v, proj_t, kdim: int, dim: int) !{}

Parameters

name	type
v	any
proj_t	any
kdim	int
dim	int

Effects !{}

round6

def round6(x: f64) -> f64 !{}

Parameters

name	type
x	f64

Returns f64

Effects !{}

main

load_corpus

def load_corpus() -> list[str] !{fs.read}

Returns list[str]

Effects !{fs.read}

get_index

def get_index() -> GraphRAG !{fs.read}

Returns GraphRAG

Effects !{fs.read}

api_query

def api_query(q: str) -> str !{fs.read}

Parameters

name	type
q	str

Returns str

Effects !{fs.read}

api_search

def api_search(q: str) -> str !{fs.read}

Parameters

name	type
q	str

Returns str

Effects !{fs.read}

main

def main() -> None !{fs.read, net.connect, net.listen, observe.record, clock.read, env.read, ui.render}

Returns None

Effects !{fs.read, net.connect, net.listen, observe.record, clock.read, env.read, ui.render}

similarity

cosine

def cosine(a: list[f64], b: list[f64]) -> f64 !{}

Parameters

name	type
a	list[f64]
b	list[f64]

Returns f64

Effects !{}

rank

def rank(items: list[Scored]) -> list[Scored] !{}

Parameters

name	type
items	list[Scored]

Returns list[Scored]

Effects !{}

take

def take(items: list[Scored], n: int) -> list[Scored] !{}

Parameters

name	type
items	list[Scored]
n	int

Returns list[Scored]

Effects !{}

store

struct GraphRAG

Fields

field	type	descriptor
chunks	list[Chunk]
adj	dict
proj	Tensor
dim	int
kdim	int

add_unique

def add_unique(seen: dict, out: list[int], id: int) -> None !{}

Parameters

name	type
seen	dict
out	list[int]
id	int

Returns None

Effects !{}

build_index

def build_index(lines: list[str], dim: int, kdim: int, kn: int) -> GraphRAG !{}

Parameters

name	type
lines	list[str]
dim	int
kdim	int
kn	int

Returns GraphRAG

Effects !{}

types

struct Chunk

Fields

field	type	descriptor
id	int
text	str
vec	list[f64]
pvec	list[f64]

struct Scored

Fields

field	type	descriptor
id	int
score	f64

struct Answer

Fields

field	type	descriptor
answer	str
sources	list[int]
scores	list[f64]