riff-kg-kit

PyPI-first Python library for staged knowledge-graph ingestion and agent-approved commits, designed to plug into Riff apps (FastAPI + Postgres + pgvector).

Status

0.7.1 — Extraction durability. run_extraction_for_signal commits staged proposals per segment (when not inside a caller transaction), marks failed runs with error_message, and keeps proposals from completed segments if a later segment errors or times out. 0.7.0 added hybrid reranking, graph-hop retrieval, and pack_context helpers.

Install

pip install riff-kg-kit

Editable (local dev):

pip install -e ".[dev,yaml]"

Quick use

from riff_kg import KgConfig

cfg = KgConfig.model_validate_json('{"embedding_dimension": 768}')
assert cfg.embedding_dimension == 768

Chunking strategy

By default, chunking uses fixed character windows. You can opt into boundary-aware chunking that prefers paragraph/sentence/whitespace splits:

from riff_kg import KgConfig

cfg = KgConfig(
    chunking_strategy="semantic",  # "char" (default) or "semantic"
    chunk_if_longer_than_chars=12000,
    chunk_max_chars=8000,
    chunk_overlap_chars=400,
)

Roadmap (summary)

1. Migrations + core tables (signal, signal_segment, staged_proposal, …)
2. Normalize → segment → embed
3. Extract → stage (LLM proposals only)
4. Approve → commit (validated canonical graph)
5. Search / pack-context / retrieval (implemented: riff_kg.search)

Retrieval (Phase 5)

from riff_kg import KgConfig
from riff_kg.search import pack_context, search_segments_vector

# After migrations and ingest with embeddings, pass a query embedding
# (same dimension as KgConfig.embedding_dimension, e.g. 768):
# hits = await search_segments_vector(conn, cfg, query_vec, scope_id="my_scope")
# text = pack_context(hits)

Hybrid rerank and graph hops

from riff_kg.search import graph_hop_subgraph, search_segments_hybrid

# Hybrid search with Reciprocal Rank Fusion (default)
# hits = await search_segments_hybrid(conn, cfg, query_embedding=qvec, fts_query="topic")

# Graph neighborhood around a committed entity
# nodes, edges = await graph_hop_subgraph(conn, root_entity_id=entity_id, max_hops=2, scope_id="my_scope")

Publishing checklist

1. Bump version in pyproject.toml and src/riff_kg/__init__.py.
2. Update CHANGELOG.md and README status/examples.
3. Run python -m ruff check src tests and python -m pytest tests -q.
4. Build distributions: python -m build.
5. Upload to TestPyPI first, verify install, then publish to PyPI.

License

Apache-2.0 — see LICENSE.