Build time-aware knowledge graphs in PostgreSQL for RAG

post-graph-rag

Graph RAG with a memory of time - on the PostgreSQL you already run.

post-graph-rag extracts entities and relations with an LLM, stores them as a property graph beside pgvector embeddings, and answers questions by fusing vector similarity, graph traversal and full-text search. What makes it different: a later document can close an earlier fact, so your model stops reporting that someone is both an ally and a rival.

No separate vector store. No graph engine to operate. One database, one consistency model, one backup - and transactions that span your graph and your application tables.

The graph layer underneath is post-graph, also usable on its own - more below.

⚡ Try it

pip install post-graph-rag

Point it at PostgreSQL with pgvector and any OpenAI-compatible endpoint:

from post_graph_rag import GraphRAG, RAGConfig

rag = GraphRAG(RAGConfig(
    api_base="https://your-router/v1", api_key=..., model="gemini-3.6-flash",
    embedding_model="gemini-embedding-001", embedding_dim=1536,
    db_uri="postgresql://localhost:5432/postgres", realm="my_kb"))
await rag.initialize()

await rag.index_document("Helena Voss is CFO of Calder Industries.", metadata={"source": "2019.txt"})
await rag.index_document("Priya Nair has been appointed CFO, succeeding Helena Voss.", metadata={"source": "2023.txt"})

answer = await rag.query("Who is the CFO of Calder Industries?")
# Priya Nair — the earlier fact is closed, not competing

→ Seven runnable examples · Full installation · Configuration reference

📈 Benchmarks

Paper: arXiv:2608.24921.

LongMemEval: long-horizon chat memory

On the full 500-question LongMemEval set - all six question types, nothing sampled - against the numbers Zep publish for Graphiti (arXiv:2501.13956):

Qualifications, so you can weigh them yourself: Zep judge with GPT-4o, this uses a three-model majority panel with the answering model excluded from its own jury; the generation models differ in cost class and vintage in a direction that cannot be signed; one question of 500 is excluded because neither extraction prompt could turn that session into triples. The harness, frozen config and every failing case ship in the repo - see the full write-up, which also documents the improvements that were tested and rejected.

ECT-QA: the adversarial corpus

Chat memory is the easy register. ECT-QA is the hard one - earnings call transcripts, sixteen quarters per company, every metric restated every quarter with only the date to tell the values apart. A store that merely accumulates cannot answer these at all; it has four values for "gross margin" and no way to choose.

Scored under the protocol ECT-QA's own authors use - an LLM judge comparing element-wise, with a refusal counted separately from a wrong answer:

A second judge from a different model family scores the same answers at 0.805 - two-tenths of a point apart, which matters more than either figure, since the usual objection to a judged rate is that it moves with the judge.

The breakdown is more useful than the total: incorrect elements sit at 0.14, refusals at 0.05, and relative-time questions produce no incorrect elements at all. When this system commits to a figure on this corpus, it is usually right.

Same qualifications apply, plus two specific to this comparison: their judge model is not available on our router, and their verbatim rubric is truncated in the public HTML, so ours reproduces their described categories rather than their text. Their figures are on their corpus slice; ours is 78 questions over 6 companies. Read a few points of margin as approximate rather than decisive.

🧱 The layer underneath: post-graph

A standalone library, if you want the graph without the RAG. It makes PostgreSQL behave like a graph database rather than emulating one:

• Table-per-vertex, table-per-edge - real foreign keys, real indexes, real constraints, so your graph is queryable by anything that speaks SQL
• Recursive CTE traversals - neighbours, paths and shortest-path with cycle detection, executed in the database rather than in your application
• Two levels of tenancy - realm for hard isolation (optionally schema-per-tenant), space for sub-grouping inside it
• pgvector on vertices and edges, searchable across live and historical rows
• Append-only history and trigger-based audit logging on every table, capturing old and new state with the acting user
• Promoted payload columns and server-side range queries - filter, order and bulk-delete on JSONB fields at the database, not in Python

pip install post-graph · Apache 2.0 · 635 tests

🌟 Why post-graph-rag?

Traditional Vector RAG systems suffer from "chunk isolation"-they retrieve isolated text passages based purely on semantic similarity, missing higher-level relationships and cross-document entity connections.

post-graph-rag solves this by building a dual representation inside PostgreSQL:

1. Unstructured Vector Passages: Full document chunks indexed with pgvector HNSW embeddings.
2. Knowledge Graph Triples: Extracted Subject-Predicate-Object entities connected by graph edges.
3. Structured Document Metadata: Rich metadata tracking (source, category, collection, document, page, paragraph, space).
4. Application-Level Space Sub-grouping (space): Scopes indexing and vector similarity search to application-specific environments (e.g. production, sandbox, staging, user_workspace) within {realm} tenant partitions.

Relationship quality

Extracted relations are context-specific, drawn from what the text actually states:

(Zeus) --[is_king_of]--> (Olympian gods)
(Zeus) --[son_of]--> (Cronus)
(Zeus) --[married_to]--> (Hera)
(Zeus) --[defeated]--> (Titans)

Vague connectors (relates_to, associated_with, connected_to, …), self-loops and blank endpoints are rejected at extraction time. A relation that reaches the graph always says something specific about the pair it connects, and two entities merely appearing near each other never produces an edge.

If the LLM cannot produce usable structure, indexing raises ExtractionError. Placeholder edges are never invented as a fallback: once written they are indistinguishable from genuine extracted structure.

Relations the text explicitly denies are stored with the positive predicate and negated: true, rather than as an inverted predicate like did_not_have_relationship_with. Traversal and synthesis can then exclude them instead of reading them as assertions.

Entity resolution

Entities are unique per (realm, space, lower(name)), enforced by a unique index, and are additionally resolved through aliases. Extraction records every other surface form it sees, so Babbage, Charles Babbage and C. Babbage converge on one vertex; the fuller name becomes canonical and the rest become aliases. Pronouns and relative references (he, his father, the company) are rejected outright - they cannot resolve to a stable vertex.

The same entity mentioned in many documents is one vertex, which is what allows the graph to connect chunks that share no vocabulary.

Chunking and document context

index_text() chunks a document (with overlap, so relations spanning a boundary survive) and threads a DocumentContext through the chunks - title, source, and the canonical entity names found so far. Without it, every chunk after the first is extracted blind and its pronouns become junk vertices.

Bring your own splitter by passing chunker= to GraphRAG, or use index_document() directly with your own DocumentContext.

rag = GraphRAG(config, chunker=my_splitter)
await rag.index_text(long_text, metadata=DocumentMetadata(document="babbage.txt"))

Community summarisation

Corpus-level questions - "what are the main themes here?" - cannot be answered by retrieving passages, because no single passage contains the answer. After indexing, cluster the entity graph and summarise each cluster:

await rag.index_text(doc_a, metadata=DocumentMetadata(document="a.txt"))
await rag.index_text(doc_b, metadata=DocumentMetadata(document="b.txt"))

await rag.build_communities()          # clusters + one LLM report per community

res = await rag.query("What are the main themes?", param=QueryParam(mode="global"))
print(res["retrieved_communities"])

Each report is stored as a vertex in communities with its own embedding, so global and hybrid retrieval find themes by similarity rather than by enumerating relations. Membership is recorded as community_members edges back to the entities, so a report is always traceable to the subgraph it came from.

Communities are derived data: build_communities() replaces the previous clustering for the space rather than accumulating stale clusters. Global mode degrades to relation ranking when none have been built, so it never hard-fails.

Detection uses Leiden (igraph + leidenalg, installed by default), falling back to deterministic label propagation if the native build is unavailable. Both are deterministic - a randomised partition would produce a different graph on every indexing run. Leiden is the default because partition balance matters: on the evaluation corpus the largest community holds 35% of the graph under label propagation against 17% under Leiden, and a community spanning a third of the graph summarises everything rather than a theme. Supply your own with community_detector=:

rag = GraphRAG(config, community_detector=my_detector)   # (nodes, edges) -> {node: community_id}

Repeated relations

The same triple extracted from several chunks is one edge whose weight increments, not several edges. Weight then breaks ties when ranking relations.

🧭 Exploration Support (1.10.0)

Three engine calls for exploration-first consumers - structure, coverage, change - with the agent loop staying yours:

# Structure: an opt-in topic tree above the flat communities
rag = GraphRAG(RAGConfig(community_levels=2))
tree = await rag.get_community_tree()

# Coverage: where has retrieval never looked? (opt-in telemetry, hash-only)
frontier = await rag.least_explored_communities(k=5)
dark = await rag.dark_entities(limit=100)

# Change: what moved since the last poll, from belief time
delta = await rag.changes_since(watermark)          # counts only, one round trip
if not delta.empty:
    detail = await rag.changes_since(watermark, summary=False)
watermark = delta.as_of

Hierarchy levels nest by construction (recursive supergraph clustering); level-filtered retrieval happens inside the vector search, not as a post-hoc trim; deltas are exactly-once under clock skew via database-clock watermarks; and re-indexing an unchanged document yields an empty delta. community_levels defaults to 1 - existing behaviour is untouched.

🏗️ Architecture Workflow

graph TD
    subgraph INDEXING ["1. Indexing"]
        A[Document + Metadata] --> B[Chunker]
        B --> C[Embedding Service]
        B --> D[LLM GraphExtractor<br/>validate · glean · normalise]
        C -->|chunk vectors| S[(PostgreSQL + pgvector<br/>via post-graph)]
        D -->|entities · triples · validity| S
        D -.->|later doc contradicts earlier| SUP[Supersession<br/>closes the old edge]
        SUP --> S
    end

    subgraph SCHEMA ["Tables in one database"]
        S --- V1[documents]
        S --- V2[entities]
        S --- V3[communities]
        S --- E1[relations<br/>valid_from/to · t_created/expired]
        S --- E2[doc_mentions]
        S --- E3[community_members / _children]
    end

    subgraph RETRIEVAL ["2. Retrieval — three channels, fused by RRF"]
        Q[Question] --> K[Keyword + subquery expansion]
        K --> C1[Entity vector search<br/>→ multi-hop traversal]
        K --> C2[Relation embedding search]
        K --> C3[Lexical / BM25 over relations]
        V2 --> C1
        E1 --> C1 & C2 & C3
        C1 & C2 & C3 --> F[RRF fusion<br/>MMR · node-distance rerank]
        V3 -->|global mode| F
        V1 -->|chunks| F
    end

    subgraph SYNTHESIS ["3. Synthesis"]
        F --> G[Temporal filter<br/>as_of · as_believed_at]
        G --> H[Prompt assembly<br/>renders each relation's validity]
        H --> I[LLM]
        I --> J[Answer + citations + triples]
    end

Three things in that diagram are the whole argument: supersession at write time, three retrieval channels fused rather than one, and validity rendered into the prompt - the last being worth +38 points on temporal reasoning on its own.

🧪 Runnable Examples

Seven scripts in examples/, each one capability, all runnable:

export OPENAI_API_KEY=...  OPENAI_API_BASE=...  POSTGRES_URI=...
cd examples && python 01_quickstart.py

📦 Installation

Install post-graph-rag via pip or uv:

pip install post-graph-rag

Or using uv:

uv add post-graph-rag

PostgreSQL Requirements

pgvector is required, not optional. Without it the vertex tables are created without embedding columns and every similarity search silently returns nothing. initialize() raises SchemaError if it is missing rather than degrading.

# macOS
brew install pgvector

# Debian/Ubuntu (match your server version)
sudo apt install postgresql-17-pgvector

CREATE EXTENSION IF NOT EXISTS vector;

🚀 Quick Start

1. Basic Indexing & Querying

import asyncio
from post_graph_rag import GraphRAG, RAGConfig, DocumentMetadata

async def main():
    # 1. Configure GraphRAG engine
    config = RAGConfig(
        api_base="http://localhost:4000/v1",       # OpenAI-compatible router endpoint
        api_key=os.environ["OPENAI_API_KEY"],     # Never hardcode credentials
        model="gemini-3.6-flash",                    # LLM model for extraction & synthesis
        embedding_model="gemini-embedding-001", # Embedding model
        embedding_dim=1536,                       # Must match the model's output width
        db_uri="postgresql://user:password@localhost:5432/postgres",
        realm="enterprise_kb",
        schema_per_realm=True                     # Give each tenant its own schema
    )

    rag = GraphRAG(config)
    
    # 2. Connect & initialize PostgreSQL graph schema
    await rag.initialize()

    # 3. Index unstructured documents
    doc_text = (
        "Zeus is the king of the Olympian gods, ruling sky and thunder from Mount Olympus. "
        "He is the son of Cronus and Rhea, and married to Hera. "
        "Zeus defeated the Titans in the Titanomachy to establish his rule."
    )
    
    result = await rag.index_document(doc_text, metadata={"source": "greek_mythology.txt"})
    print(f"Indexed document {result['document_id']}: Extracted {result['entities_extracted']} entities.")

    # 4. Perform Hybrid RAG Query
    response = await rag.query("Who are the parents of Zeus and what did he defeat?")
    
    print("\n=== SYNTHESIZED ANSWER ===")
    print(response["answer"])

    print("\n=== RETRIEVED GRAPH TRIPLES ===")
    for triple in response["retrieved_graph_triples"]:
        print(f"  - {triple}")

    # 5. Clean up
    await rag.close()

if __name__ == "__main__":
    asyncio.run(main())

📋 Document Metadata (DocumentMetadata)

post-graph-rag includes structured document metadata tracking via the DocumentMetadata model:

from post_graph_rag import DocumentMetadata

metadata = DocumentMetadata(
    source="https://mythology.org/zeus.html",  # Document origin (URL, filepath, API)
    category="greek_mythology",               # Document category/topic
    collection="olympian_deities",             # Collection namespace
    document="zeus_overview.pdf",              # Title or filename
    page=1,                                    # 1-based page number
    paragraph=2,                               # 1-based paragraph index
    space="production",                        # Sub-grouping within the realm
    extra={"author": "Homer", "year": -700}    # Custom metadata key-value pairs
)

await rag.index_document(chunk_text, metadata=metadata)

Design Rationale: Optional vs. Required

• All metadata fields are optional with default None. This allows seamless indexing of raw strings, short code snippets, webhooks, or unformatted text, while offering rich structural provenance tracking when indexing multi-page PDFs or categorized enterprise documents.

Document identity: source and document together

Re-indexing replaces rather than appends, so two documents that resolve to the same key are treated as one document seen twice - the second deletes the first.

The key is built from source and document together, so give at least one of them a value that is unique per document:

# Correct: source identifies this document
DocumentMetadata(source="/corpus/WDC-2022-q1.json", document="WDC-2022-q1")

# Wrong: a corpus name is not a document identity
DocumentMetadata(source="ect", document="WDC-2022-q1")   # was catastrophic before 1.8.0

Before 1.8.0 the key preferred source and ignored document entirely, so the second form collapsed an entire corpus onto one key. An 80-transcript run kept five transcripts and marked 92% of its relations dormant, with no error raised - the only visible symptom was the system declining to answer questions whose evidence had been deleted. Since 1.8.0 both parts contribute, so the second form is merely untidy rather than destructive.

Upgrading: keys computed before 1.8.0 do not match the new scheme, so re-indexing an existing document appends a copy instead of replacing it. Rebuild realms indexed on the old scheme.

⚙️ Configuration Reference (RAGConfig)

RAGConfig can be configured explicitly or automatically loaded from environment variables:

Environment variables are read when a RAGConfig is constructed, not at import time.

External extraction (extraction_fn)

Some structure should never be guessed at. Imports, call edges, ownership and deployment topology are recoverable exactly from an AST, an LSP index or a build file, and asking a model to infer them is strictly worse than reading them. extraction_fn lets you hand those in directly.

def ast_edges(text, context=None):
    return [{"subject": "checkout_service", "predicate": "calls",
             "object": "ledger_service"}]

RAGConfig(extraction_fn=ast_edges)          # replaces the LLM
RAGConfig(extraction_fn=ast_edges, extraction_fn_mode="merge")   # runs both

The function takes (text, context) and may be sync or async. It can return an ExtractionResult, a {"entities": [...], "triples": [...]} mapping, a bare sequence of triples, or None for "nothing here". Endpoints you do not also describe as entities become stubs, the same path the LLM's own unmatched endpoints take - so a triples-only source still produces a traversable graph.

merge is the interesting mode. A deterministic code graph and the model's reading of the prose around it end up in one graph, with extraction spending its cost only on the unwritten part: rationale, causation, what a decision was for. Each source is gated on its own way in, so an exact external record is not held to the LLM's vocabulary and an LLM record is not let through ungated.

Your records go through the same validation as the model's by default - pronominal and phrase-shaped names dropped, vague predicates and self-loops rejected, predicates snapped onto the vocabulary. That is what makes "extraction output is untrusted input" true of every source rather than only of the model. Set validate_external_extraction=False when the source is already exact and the vocabulary would mangle it: calls and imports from an AST do not want snapping.

Two behaviours worth knowing. An exception from your function fails the chunk, deliberately - a broken external extractor is the same event as a dead model, and swallowing it would let an import that stopped resolving read as a corpus with no call edges in it. And relations from your function carry the chunk that asserted them, so deleting that document withdraws them exactly as it would the model's.

Model roles

One model for every call is a compromise no role actually wants, so each role can take its own. Leave them unset and everything resolves to model, which is how the library behaved before roles existed - nothing else changes.

RAGConfig(
    model="gemini-3.6-flash",           # the default for anything unset
    extraction_model="MiniMax-M2.7",    # index-time: builds the graph
    synthesis_model="gemini-3.7-flash", # query-time: reads it
    auxiliary_model="gemma-4-31B-it",   # keywords and decomposition
)

Two reasons this is worth using. The roles have different optima: over one fixed graph, four answering models separate across a 23-point band, while on the extraction side the models do not even agree on what a good graph is - one builds the richest (most relations, most aliases), another the most queryable (94% predicate-vocabulary adherence but 40% fewer relations). And they have different economics: extraction is one or two calls per chunk and runs once, whereas auxiliary and synthesis run on every question for ever.

Be careful with extraction_model. A weak model here does not fail, it silently degrades the graph. Anything that depends on the model doing something subtle simply stops happening - one tested model never once emitted the negated field, collapsing "X worked with Y" and "X never met Y" into the same edge, with no error anywhere. Extraction is the role to spend on.

schema_per_realm

Off by default for backwards compatibility, but recommended for new deployments. With it off, every realm shares one physical set of tables filtered by a realm column - so the first realm to create entities fixes the embedding column width for all of them, and a second realm with a different embedding_dim cannot work.

allow_embedding_fallback

Off by default. Fallback vectors are not comparable with API embeddings, so mixing them into the same table corrupts retrieval rather than degrading it. With it off, an embedding failure raises EmbeddingError.

📖 API Reference

GraphRAG

The main orchestrator class for indexing and querying.

• await initialize(): Connects to PostgreSQL and creates the graph tables (documents, entities, relations, doc_mentions). Raises SchemaError if pgvector is unavailable or an existing table's embedding width disagrees with embedding_dim.
• await index_document(text, metadata=None, space=None) -> Dict[str, Any]: Embeds the chunk, extracts entities/triples via the LLM, resolves entities by name, and writes vertices, relations and doc_mentions edges. Raises rather than writing placeholder structure if extraction fails. Returns counts plus document_id and metadata.
• await query(question, param=None, top_k=None): Retrieves and synthesizes an answer. Returns question, answer, mode, keywords, retrieved_documents, retrieved_entities, retrieved_graph_triples, references. With QueryParam(stream=True) returns an async iterator of content chunks instead.
• await query_data(question, param=None) -> Dict[str, Any]: Structured retrieval with no synthesis - returns entities, relationships, chunks, references.
• await close(): Closes database connection pools.

Per-document views (1.13.0)

For a registry or admin UI that lists documents and shows what each one put into the graph.

• await document_stats(doc_key, space=None) -> DocumentStats: chunks, chunk_bytes, entities_mentioned split into entities_current / entities_dormant, relations contributed, and first_indexed_at / last_indexed_at. A document that was never indexed returns zeros with found=False rather than raising, so "indexed but empty" stays distinguishable from "not present".
• await documents_stats(doc_keys, space=None) -> Dict[str, DocumentStats]: the batch form, and the one to render a table with. Two SQL round trips for any number of keys - looping over document_stats would issue two per row.
• await document_graph(doc_key, space=None, max_entities=500, max_relations=1000): the entities and relations that document contributed, for a drill-down. Capped, and the caps are reported in truncated / entities_truncated / relations_truncated - a silently truncated subgraph reads as a complete one.
• await sweep_orphaned_relations(space=None) -> int: one-shot repair for graphs written before relation provenance existed. Those relations record no sources, so no document deletion can withdraw them; this retires the ones whose endpoint entities have all gone dormant. Nothing is deleted, and it is safe to re-run.

These refuse space="__all__": they resolve a caller-supplied document key, and doing that across every space would return another tenant's document under the key this one asked for.

Deletion and dormancy

RAGGraphStore.delete_document_chunks(doc_key, space=None) removes a document's chunks and mention edges. Nothing else is deleted. Entities whose last mention disappears are marked dormant; relations left with no contributing chunk are marked dormant too, and so are relations whose endpoint entities have all gone dormant. Dormant relations are excluded from every retrieval path, from query/query_data, and from community building - pass include_dormant=True to the store's read methods to see them for audit. A dormant entity or relation revives automatically if a later document brings it back.

QueryParam

• mode: one of mix, local, global, hybrid, naive, bypass. An unknown mode raises ValueError.
• top_k, max_total_tokens, max_entity_tokens, max_relation_tokens, response_type
  • The three token budgets default to None - unlimited - as of 1.11.1: everything retrieved reaches the model. Set an integer to cap context for cost or for a model with a small window.
• stream: return an async iterator of tokens instead of a dict.
• only_need_context: return retrieval output without calling the LLM.
• space: restrict retrieval to one space; __all__ queries across all spaces.
• conversation_history, hl_keywords, ll_keywords: supply keywords to skip extraction.

Errors

All inherit from RAGError, so failures surface instead of degrading into irrelevant results:

• SchemaError - pgvector missing, or embedding width mismatch.
• EmbeddingError - embedding request failed, or returned the wrong width.
• LLMError - completion or streaming call failed.
• ExtractionError - the LLM returned no usable entities or triples.

DocumentMetadata

Data container for structured document metadata.

• source: Optional[str]: Document URL, path, or origin.
• category: Optional[str]: Document category or domain.
• collection: Optional[str]: Document collection or folder.
• document: Optional[str]: File title or filename.
• page: Optional[int]: 1-based page number.
• paragraph: Optional[int]: 1-based paragraph index.
• space: Optional[str]: Sub-grouping space to index into.
• extra: Dict[str, Any]: Custom user metadata.
• to_dict() -> Dict[str, Any]: Serializes non-None fields to dictionary representation.
• from_dict(data: Dict[str, Any]) -> DocumentMetadata: Deserializes dictionary data.

RAGGraphStore

Database layer wrapping post-graph.

• add_document(text, embedding, metadata, space=None): Inserts a document vertex into the documents table.
• upsert_entity(name, entity_type, description, embedding, space=None): Upserts by canonical name within (realm, space), so an entity mentioned in many documents is a single vertex. A bare Concept stub never overwrites a richer type or description.
• find_entity_by_name(name, space=None): Resolve an entity vertex by name.
• add_relation(from_entity, to_entity, relation_type, description, space=None, embedding=None): Directed relation edge.
• add_doc_mention(doc_vertex, entity_vertex, space=None): Links a chunk to an entity it mentions.
• search_similar_entities(query_vec, top_k, space=None) / search_similar_documents(...): pgvector HNSW similarity search.
• search_similar_relations(query_vec, top_k, space=None): Semantic search over relation edges. Returns [] unless embed_relations is enabled.
• get_neighbors(entity_id, space=None, include_dormant=False): 1-hop outgoing relations, scoped to space.
• get_all_relations(limit, space=None, include_dormant=False): Relations with their endpoint vertices.
• get_neighborhood(entity_id, max_hops=1, ...) / get_relations_by_ids(...) / search_relations_text(...): the other relation read paths. All take include_dormant=False, filtered in SQL, so a retired relation cannot reach an answer through any of them.
• document_stats(...) / documents_stats(...) / document_graph(...) / sweep_orphaned_relations(space=None): as described under GraphRAG above.

🗄️ PostgreSQL Database Schema

post-graph-rag automatically provisions and manages the following graph schema in PostgreSQL powered by post-graph:

🧪 Testing

pip install -e ".[test]"
createdb postgres && psql -d postgres -c "CREATE EXTENSION IF NOT EXISTS vector;"
POSTGRES_TEST_URI="postgresql://localhost:5432/postgres" pytest

DB-backed tests create a disposable schema per test realm and drop it afterwards. They skip automatically when PostgreSQL with pgvector is not reachable.

🤝 Contributing

Bug reports, failing test cases and pull requests are all welcome. CONTRIBUTING.md covers the parts that are specific to this project rather than generic advice: what the test suite needs, what it deliberately does not need, and the invariants worth understanding before changing them.

Issues tagged good first issue are real gaps rather than make-work - each one names the file to look at, what "done" means, and what you will learn from it.

Tests need PostgreSQL with pgvector and no LLM credentials - every test uses an in-process fake, so the suite is free, offline and deterministic.

📄 License

This project is licensed under the Apache License 2.0 - see the LICENSE file for details.

Developed by Chandan Rajah (chandan.rajah@gmail.com).

examples/01_quickstart.py

"""Index three documents and ask a question that needs all three.

The point: the answer requires joining facts stated in separate documents.
A flat vector store retrieves whichever chunk is most similar to the
question; the graph walks the relations between them.
"""
import asyncio

from _shared import banner, fresh_realm, make_config

from post_graph_rag import GraphRAG

DOCS = [
    ("acquisition.txt",
     "Northwind Systems acquired Vertex Analytics in March 2021. "
     "Vertex Analytics was founded by Dr. Amara Okafor in 2014."),
    ("leadership.txt",
     "Dr. Amara Okafor was appointed Chief Technology Officer of Northwind "
     "Systems in June 2021, reporting to CEO Martin Feld."),
    ("products.txt",
     "Northwind's flagship product, Meridian, was rebuilt in 2022 on the "
     "forecasting engine that Vertex Analytics originally developed."),
]


async def main():
    rag = GraphRAG(make_config(fresh_realm("example_quickstart")))
    await rag.initialize()
    try:
        for source, text in DOCS:
            res = await rag.index_document(text, metadata={"source": source})
            print(f"indexed {source}: {res['entities_extracted']} entities")

        banner("Who built the engine behind Meridian, and where do they work now?")
        out = await rag.query(
            "Who built the technology behind Meridian, and what is their role "
            "at Northwind today?")
        print(out["answer"])

        banner("Relations the answer was built from")
        for t in out.get("retrieved_graph_triples", [])[:8]:
            print(f"  {t}")
    finally:
        await rag.close()


if __name__ == "__main__":
    asyncio.run(main())

examples/02_supersession.py

"""A later document retires an earlier fact.

This is the behaviour the library exists for. Three filings arrive in order;
each names a different CFO. A store that only accumulates will answer with
all three, or with whichever chunk embeds closest. Here the last assertion
wins and the earlier ones are marked superseded -- kept, not deleted.
"""
import asyncio

from _shared import banner, fresh_realm, make_config

from post_graph_rag import GraphRAG, QueryParam

FILINGS = [
    ("2019-annual-report.txt",
     "Helena Voss serves as Chief Financial Officer of Calder Industries."),
    ("2021-annual-report.txt",
     "Calder Industries announced that Ravi Chandran has been appointed "
     "Chief Financial Officer, succeeding Helena Voss."),
    ("2023-annual-report.txt",
     "Ravi Chandran stepped down as Chief Financial Officer of Calder "
     "Industries. Priya Nair now serves as Chief Financial Officer."),
]


async def main():
    rag = GraphRAG(make_config(fresh_realm("example_supersession")))
    await rag.initialize()
    try:
        for source, text in FILINGS:
            await rag.index_document(text, metadata={"source": source})
            print(f"indexed {source}")

        banner("Who is the CFO of Calder Industries?")
        out = await rag.query("Who is the Chief Financial Officer of Calder Industries?")
        print(out["answer"])
        print("\n-> Priya Nair is named as the current CFO. The earlier two are\n"
              "   reported as prior holders, not as competing present-tense facts --\n"
              "   because the graph closed them when the later filing arrived.")

        banner("Same question, superseded history included")
        out = await rag.query(
            "Who has served as Chief Financial Officer of Calder Industries, "
            "and in what order?",
            param=QueryParam(include_superseded=True))
        print(out["answer"])
        print("\n-> The history is retained and auditable, not overwritten.")
    finally:
        await rag.close()


if __name__ == "__main__":
    asyncio.run(main())

examples/03_bitemporal_audit.py

"""Two clocks: what was true, and what we believed.

A filing published in 2024 can restate a figure for 2019. Those are different
questions and they need different answers:

    as_of            -- the world as it was at that date (valid time)
    as_believed_at   -- the graph as this system held it at that instant
                        (belief time)

The second is the one an auditor asks: not "what is true" but "what did your
systems report in March, before the restatement". Answering it is how you
reproduce a past decision.
"""
import asyncio
from datetime import datetime, timezone

from _shared import banner, fresh_realm, make_config

from post_graph_rag import GraphRAG, QueryParam


async def main():
    rag = GraphRAG(make_config(fresh_realm("example_bitemporal")))
    await rag.initialize()
    try:
        await rag.index_document(
            "In its 2019 annual report, Calder Industries reported full-year "
            "revenue of 412 million dollars.",
            metadata={"source": "2019-annual-report.txt"})
        before_restatement = datetime.now(timezone.utc).isoformat()
        print(f"watermark taken: {before_restatement}")

        await rag.index_document(
            "Calder Industries restated its 2019 revenue to 388 million "
            "dollars following an accounting review.",
            metadata={"source": "2021-restatement.txt"})
        print("restatement indexed")

        banner("What is 2019 revenue? (today's belief)")
        print((await rag.query("What was Calder Industries' 2019 revenue?"))["answer"])

        banner("What did we believe before the restatement was filed?")
        out = await rag.query(
            "What was Calder Industries' 2019 revenue?",
            param=QueryParam(as_believed_at=before_restatement))
        print(out["answer"])
        print("\n-> The pre-restatement answer is reproducible on demand.")
    finally:
        await rag.close()


if __name__ == "__main__":
    asyncio.run(main())

examples/04_multi_tenant_spaces.py

"""One realm, many tenants: spaces keep corpora apart and let you cross them.

Each customer's documents go in their own space. A query scoped to a space
sees only that tenant. The same graph can then answer a cross-tenant question
when you deliberately ask for one -- useful for portfolio or sector views,
and the reason spaces are a query parameter rather than a separate database.
"""
import asyncio

from _shared import banner, fresh_realm, make_config
from post_graph import RESERVED_SPACE_ALL

from post_graph_rag import GraphRAG, QueryParam

TENANTS = {
    "acme": "Acme Corp reported quarterly revenue of 24 million dollars and "
            "operates two manufacturing plants in Ohio.",
    "globex": "Globex Inc reported quarterly revenue of 31 million dollars "
              "and operates a single plant in Arizona.",
}


async def main():
    rag = GraphRAG(make_config(fresh_realm("example_spaces")))
    await rag.initialize()
    try:
        for space, text in TENANTS.items():
            await rag.index_document(text, metadata={"source": f"{space}-q3.txt"},
                                     space=space)
            print(f"indexed into space '{space}'")

        banner("Scoped to acme -- globex is invisible")
        out = await rag.query("What was quarterly revenue?",
                              param=QueryParam(space="acme"))
        print(out["answer"])

        banner("Across every space -- a deliberate cross-tenant view")
        out = await rag.query(
            "Compare quarterly revenue across all companies.",
            param=QueryParam(space=RESERVED_SPACE_ALL, top_k=20))
        print(out["answer"])
    finally:
        await rag.close()


if __name__ == "__main__":
    asyncio.run(main())

examples/05_incremental_and_deltas.py

"""Re-index changed documents, and poll for what moved.

Two properties worth knowing:

  * Re-indexing an unchanged document is a no-op -- same content, no new
    chunks, no duplicated entities, and an empty delta.
  * changes_since(watermark) reports what actually moved from belief time:
    new and superseded relations, new and dormant entities, new documents.
    The default summary transfers counts only, so a poller pays one cheap
    round trip per tick.

Watermarks come from the database clock, so the poll chain stays
exactly-once even when your application server's clock drifts.
"""
import asyncio

from _shared import banner, fresh_realm, make_config

from post_graph_rag import GraphRAG


async def main():
    rag = GraphRAG(make_config(fresh_realm("example_deltas")))
    await rag.initialize()
    try:
        await rag.index_document(
            "Orion Freight operates 40 vehicles from a depot in Leeds.",
            metadata={"source": "fleet.txt"})

        watermark = await rag.watermark()          # start of the poll chain
        print(f"initial watermark: {watermark}")

        banner("Re-index the same content unchanged")
        res = await rag.index_document(
            "Orion Freight operates 40 vehicles from a depot in Leeds.",
            metadata={"source": "fleet.txt"})
        print(f"result: {res}")
        delta = await rag.changes_since(watermark)
        print(f"delta empty: {delta.empty}   -> idempotent re-indexing")

        banner("Now the fact actually changes")
        await rag.index_document(
            "Orion Freight has expanded to 65 vehicles and opened a second "
            "depot in Manchester.",
            metadata={"source": "fleet.txt"})
        delta = await rag.changes_since(watermark)
        print(f"delta empty: {delta.empty}")
        print(f"counts: {delta}")

        detail = await rag.changes_since(watermark, summary=False)
        print(f"\nsuperseded relations: {len(detail.superseded_relations)}")
        print(f"new relations:        {len(detail.new_relations)}")
    finally:
        await rag.close()


if __name__ == "__main__":
    asyncio.run(main())

examples/06_communities_and_exploration.py

"""Corpus-level questions, and deciding what to look at next.

build_communities clusters the graph and writes an LLM report per cluster, so
"what are the themes here" is answered from summaries rather than from a
top-k of chunks. With community_levels > 1 the clusters nest, giving you a
topic tree to drill through.

The exploration calls answer a different question -- not "what is the answer"
but "where has nobody looked". Coverage telemetry is opt-in and stores only a
hash of each query.
"""
import asyncio

from _shared import banner, fresh_realm, make_config

from post_graph_rag import GraphRAG, QueryParam

DOCS = [
    "Solar capacity in Spain grew 22 percent as grid connections accelerated.",
    "Portugal commissioned three offshore wind farms off the Atlantic coast.",
    "Battery storage costs fell sharply, improving solar project economics.",
    "Rail freight volumes in Germany declined amid industrial slowdown.",
    "Dutch port throughput fell as container traffic shifted south.",
    "Rail operators in France invested in electrification of regional lines.",
]


async def main():
    rag = GraphRAG(make_config(fresh_realm("example_communities"),
                               community_levels=2,
                               record_retrieval_events=True))
    await rag.initialize()
    try:
        for i, text in enumerate(DOCS):
            await rag.index_document(text, metadata={"source": f"news_{i}.txt"})...