zleap-sag

Local-first memory & knowledge engine for AI agents. Ingest documents, extract an event/entity graph with an LLM, and retrieve over it — fully local by default, progressive to production databases.

Distribution zleap-sag · import zleap.sag · Python ≥ 3.11 · MIT

Highlights

• Zero-infra by default — embedded SQLite + LanceDB (built-in BM25), no services to run. Existing default paths remain ./.zleap/sag.db and ./.zleap/lancedb/.
• Progressive to production — swap in MySQL / PostgreSQL / OceanBase (relational) and Elasticsearch / pgvector (vector) by changing config only; pipeline code is unchanged.
• Freely composable storage — combine any relational backend with LanceDB, Elasticsearch, pgvector, or OceanBase Vector. A physical single-database deployment shares only the target; relation and vector still use separate tables, runtimes, pools, and transactions.
• Explicit five-stage API — Parse → Chunking → Index → Extract → Search, with immutable, JSON-serializable hand-off contracts and no "last run" engine state.
• Profile-driven retrieval — five explicit profiles: vector, atomic, full_expand, pruned_expand_llm, and pruned_expand_rff.

Quick Start

1. Install

pip install zleap-sag        # runs as-is: embedded SQLite + LanceDB, no services needed

Add an extra only when you switch that backend on (quote the brackets in shells):

pip install "zleap-sag[es]"        # Elasticsearch vector store
pip install "zleap-sag[mysql]"     # MySQL or OceanBase       (aiomysql driver)
pip install "zleap-sag[postgres]"  # PostgreSQL, incl. pgvector single-DB
pip install "zleap-sag[summary]"   # optional Sumy article summaries
pip install "zleap-sag[tables]"    # CSV/XLSX via MarkItDown, including multi-sheet workbooks
pip install "zleap-sag[all]"       # all backends + litellm

2. Run

Point it at any OpenAI-compatible LLM + embedding endpoint. Storage defaults to ./.zleap/ (SQLite + LanceDB); the schema is created automatically on first run.

import asyncio
from zleap.sag import DataEngine, EngineConfig
from zleap.sag.config import LLMConfig, EmbeddingConfig
from zleap.sag.pipeline import SearchOptions, SearchRequest, SearchScope


async def main():
    config = EngineConfig(
        storage_mode="normal",  # required: normal | lite
        llm=LLMConfig(api_key="sk-...", base_url="https://your-gateway/v1", model="qwen3.6-flash"),
        embedding=EmbeddingConfig(model="bge-large-en-v1.5"),
    )
    async with DataEngine(config) as engine:  # start(): create local tables on first run
        chunk_ref = await engine.ingest("your_document.md")
        event_ref = await engine.extract(chunk_ref)
        result = await engine.search(
            SearchRequest(
                query="Who founded Acme?",
                scope=SearchScope(
                    data_source_ids=(event_ref.data_source_id,),
                    source_ids=(event_ref.source_id,),
                ),
                options=SearchOptions(strategy="full_expand", top_k=5, return_type="event"),
            )
        )
        for event in result.events:
            print(event.content[:200])


asyncio.run(main())

Storage lands in ./.zleap/ — add it to your .gitignore. storage_mode selects the schema contract but does not rewrite data_dir; use an explicit different data_dir if you intentionally maintain a second embedded store. Runnable scripts: examples/.

The two-command example stores the extracted events and reusable search scope in one JSON file:

python examples/05_extract_and_search.py extract article.md --env-file .env
python examples/05_extract_and_search.py search "Who led the round?" \
  --scope extracted_events.json --env-file .env

Both modes store EventEntity embeddings in the formal event_entity_vectors collection—never as relation-database bytes. One vector record is keyed by the event_entity.id and carries event_id, entity_id, source metadata, description, status, and timestamps. Normal uses the configured embedding dimension; Lite stores and queries the first 128 float values. The physical column is provider-native (dense_vector, vector(N), OceanBase VECTOR(N), or a fixed-size Arrow float32 list in LanceDB).

Configuration

Two ways to configure — pick one

zleap-sag always builds a single EngineConfig. Provide it by parameter injection or environment variables — they are alternatives, not layered.

Does setting env vars remove the need to pass keys? Only if you call EngineConfig.from_env(). Plain EngineConfig(...) never reads the environment — you pass keys there explicitly. Don't mix the two: use from_env() or inject params.

Environment-variable path, minimal set (zero-infra):

export OPENAI_API_KEY=sk-...
export SAG_STORAGE_MODE=normal                     # required: normal | lite
export OPENAI_BASE_URL=https://your-gateway/v1     # optional; defaults to OpenAI
export LLM_MODEL=qwen3.6-flash
export EMBEDDING_MODEL=bge-large-en-v1.5

config = EngineConfig.from_env()  # or EngineConfig.from_env(env_file=".env")

Required vs optional

• Required: storage_mode (normal or lite), llm, and embedding.
• Optional (all default to the local stack): storage backend, data_dir (./.zleap), rerank, language, log_level.

Two gotchas

• Separate embedding endpoint? Set EmbeddingConfig(base_url=..., api_key=...) (or EMBEDDING_BASE_URL / EMBEDDING_API_KEY). If omitted, embedding reuses the LLM's key and URL.
• Don't set EmbeddingConfig(dimensions=...) unless your model supports it — many embedding models reject a dimension override.

Full variable reference (storage, backends, rerank): .env.example.

Custom entity types (optional)

Extraction keeps only entities whose type is defined. A generic set (person, organization, location, product, event, time, …) is seeded automatically. To add domain types, declare them in config — they are seeded on schema init, idempotently: existing types are skipped, only new ones are added.

config = EngineConfig(
    storage_mode="normal",
    entity_types=[
        "contract",
        "invoice",
        "party",
    ],  # str, or EntityTypeConfig(type=..., description=...)
    llm=LLMConfig(...),
    embedding=EmbeddingConfig(...),
)

Storage backends

Switch backends by changing EngineConfig only — the ingest/extract/search code is identical. Spin up local backends for testing with make up (docker compose).

Initialization

• Local SQLite — start() creates the schema and seeds default entity types automatically. Nothing to call.
• Production backends — call await engine.init_schema() before start() once. It only creates completely missing tables in the selected mode and never alters or migrates an existing table.
• Production mutations — multi-worker deployments must use process_source, delete_source, and delete_data_source with caller-owned OperationContext values. Direct ingest → extract calls are stage-level APIs for local workflows and custom orchestration; they do not provide the durable fence, lease, checkpoint, generation switch, or exact replay contract.

Example — MySQL + Elasticsearch storage wiring

from zleap.sag import DataEngine
from zleap.sag.pipeline import SearchOptions, SearchRequest, SearchScope
from zleap.sag.config import (
    EmbeddingConfig,
    ElasticsearchVectorConfig,
    EngineConfig,
    LLMConfig,
    RelationalConfig,
)

config = EngineConfig(
    storage_mode="normal",
    relational=RelationalConfig(
        provider="mysql", host="localhost", user="sag2", password="sag2", database="sag2"
    ),
    vector=ElasticsearchVectorConfig(hosts=["http://localhost:9200"]),
    llm=LLMConfig(api_key="sk-...", base_url="https://your-gateway/v1", model="qwen3.6-flash"),
    embedding=EmbeddingConfig(model="bge-large-en-v1.5"),
)

engine = DataEngine(config)
await engine.init_schema()  # once, before start(), for production backends
async with engine:
    # Stage-level example. A production worker uses process_source(), shown below.
    chunk_ref = await engine.ingest("your_document.md")
    event_ref = await engine.extract(chunk_ref)
    result = await engine.search(
        SearchRequest(
            query="Who founded Acme?",
            scope=SearchScope(data_source_ids=(event_ref.data_source_id,)),
            options=SearchOptions(strategy="full_expand", top_k=5, return_type="event"),
        )
    )

Production mutation lifecycle

from hashlib import sha256

from zleap.sag.operations import OperationContext, ProcessSourceRequest
from zleap.sag.pipeline import SourceDescriptor, TextSource

data_source_id = "11111111-1111-1111-1111-111111111111"
source_id = "22222222-2222-2222-2222-222222222222"
markdown = "# Acme\nAcme was founded by Jane."

request = ProcessSourceRequest(
    context=OperationContext(
        operation_id="publish-job-42-attempt-1",
        idempotency_key="publish-job-42",
        request_digest=sha256(markdown.encode()).hexdigest(),
        fence_scope=data_source_id,
        fence_token=7,
        owner_id="knowledge-worker-3",
    ),
    source=TextSource(
        text=markdown,
        descriptor=SourceDescriptor(
            data_source_id=data_source_id,
            source_id=source_id,
            source_type="article",
        ),
    ),
)

async with DataEngine(config, data_source_id=data_source_id) as engine:
    result = await engine.process_source(request)
    if result.status == "failed":
        raise RuntimeError(f"{result.failure_code}: retryable={result.retryable}")
    # An uncertain caller response is recovered with get_operation_status(operation_id)
    # or by replaying the exact same request.

Example — physical single database

# One PostgreSQL for relational + vector (pip install "zleap-sag[postgres]")
from zleap.sag.config import PgVectorConfig, PostgresConnectionConfig

config = EngineConfig(
    storage_mode="normal",
    relational=RelationalConfig(
        provider="postgres",
        host="localhost",
        port=5432,
        user="sag2",
        password="sag2",
        database="sag2",
    ),
    vector=PgVectorConfig(
        connection=PostgresConnectionConfig(
            host="localhost", port=5432, user="sag2", password="sag2", database="sag2"
        )
    ),
    llm=LLMConfig(...),
    embedding=EmbeddingConfig(...),
)

# One OceanBase for SQL + vector (pip install "zleap-sag[mysql]", OceanBase ≥ 4.3.3)
from zleap.sag.config import OceanBaseConnectionConfig, OceanBaseVectorConfig

config = EngineConfig(
    storage_mode="lite",
    relational=RelationalConfig(
        provider="oceanbase", host="localhost", port=2881, user="root", password="", database="sag2"
    ),
    vector=OceanBaseVectorConfig(
        connection=OceanBaseConnectionConfig(
            host="localhost", port=2881, user="root", password="", database="sag2"
        )
    ),
    llm=LLMConfig(...),
    embedding=EmbeddingConfig(...),
)

OceanBase ANN indexes need the tenant setting ob_vector_memory_limit_percentage > 0; if unset, the engine falls back to exact vector search automatically.

Matching connection values above mean “same physical database”; they do not trigger runtime reuse. To deploy PostgreSQL + Elasticsearch or OceanBase + Elasticsearch, keep the relational config and replace only vector with ElasticsearchVectorConfig(...).

How it works

The engine exposes five independent stages. Each output is passed explicitly to the next stage:

• parse(SourceInput, ParseOptions) → ParsedSource; built-in Markdown, HTML, and plain text parsing, plus automatic CSV/XLSX → Markdown conversion with the [tables] extra.
• chunk(ParsedSource, ChunkOptions) → deterministic in-memory ChunkSet.
• index(ChunkSet, SourceDescriptor, IndexOptions) → ChunkSetRef.
• extract(ChunkSetRef | PersistedChunkSelector, ExtractionOptions) → EventSetRef.
• load_events(EventSetRef) → portable EventDetail records scoped to that exact source.
• search(SearchRequest) → SearchResult; scope always contains non-empty data_source_ids, optional source_ids/source_types/creator_ids, and optional timezone-aware time bounds. creator_ids requires an injected SearchScopeResolver.

For Markdown documents, ChunkOptions(strategy="heading_strict") creates one chunk for every non-empty heading block. Consecutive headings with the same text remain separate, and a heading block is never split again by sentences or max_tokens; chunk text preserves the source block after trimming its outer whitespace. This matches SAG-Benchmark's heading_strict corpus semantics.

In standard and overlap modes, valid Markdown tables are recognized as TABLE blocks. Rows stay intact, every split table chunk repeats the complete header, and table chunks never mix with neighboring prose or a different table. ArticleSection keeps one header evidence record that all chunks from the same table reference. CSV and XLSX files enter this exact path after MarkItDown normalization; parse() remains storage-free and works before engine.start().

ingest() remains a Parse → Chunking → Index convenience method, but does not save its result on the engine. The no-argument extract() and implicit-scope search() forms are removed. SearchOptions.strategy is required. Old names multi/multi1/hopllm/multi_es are no longer accepted by the public dispatcher.

Search profiles and typed overrides

Multi-hop expansion is tuned through graph.max_hops on the same strategy (e.g. full_expand + graph=GraphSearchOptions(max_hops=2)); there is no separate multi-hop strategy name.

The built-in search runtime has dedicated Vector, Atomic, and Production Executors. Production performs direct Event-vector plus Entity lexical/vector recall; both precise and fast run bounded one-hop graph expansion. The precise variants (pruned_expand_llm / pruned_expand_rerank) rank by LLM or an external rerank model respectively; the fast variant (pruned_expand_rff) uses deterministic RRF, and full_expand also runs on Production. Atomic invokes its concrete searcher directly; there is no second SAGSearcher dispatcher. Hosts may replace the engine-local Production Executor without importing host code into this package. The Atomic executor honors score thresholds, candidate limits, graph hops, and selection.enabled; unsupported query rewrite, custom Selection prompts, rationale output, or non-vector ranking fail explicitly instead of being silently ignored.

Profiles provide stable defaults. Requests may override supported behavior through typed sub-options; there is no strategy_options dictionary:

from zleap.sag.pipeline import (
    GraphSearchOptions,
    RankingOptions,
    SearchOptions,
    SearchOutputOptions,
)

options = SearchOptions(
    strategy="pruned_expand_llm",
    top_k=10,
    graph=GraphSearchOptions(enabled=True, max_hops=2),
    ranking=RankingOptions(rerank_threshold=0.6),
    output=SearchOutputOptions(return_graph=True),
)

graph.enabled controls retrieval expansion; output.return_graph only controls whether the computed graph is returned. Search does not filter content by default; an optional EngineConfig.guard provider can filter queries and returned search results.

Extract options and atomic batches

Extract has one public strategy, sag_extract, and one response schema. Customize behavior with typed options instead of an unvalidated strategy_options dictionary:

from zleap.sag.pipeline import ExtractionLimits, ExtractionOptions

event_ref = await engine.extract(
    chunk_ref,
    ExtractionOptions(
        background="Preserve monetary values in their original currency.",
        guidance_rules=("Ignore headers and footers.",),
        limits=ExtractionLimits(
            max_events_per_chunk=20,
            min_entities_per_event=1,
            max_entities_per_event=20,
        ),
        max_retries=5,
        enable_parent_summary=True,
    ),
)

Each Chunk gets one initial generation plus up to five validation-repair generations. Transport retries remain controlled by LLMConfig.max_retries. The built-in Extract adapter commits atomically per source: if any Chunk exhausts its retries, or an enabled parent summary fails, no new events or vectors are written and the previous source snapshot remains active. Failure details are available on ExtractionBatchFailure.failures; EventSetRef represents only a fully committed success.

enable_article_summary=True is valid only for text/article sources and requires pip install "zleap-sag[summary]". Sumy is imported lazily, never downloads NLTK data at runtime, and the extractive summary is cached by source_version.

DataEngine is an async context manager: async with DataEngine(config) as engine runs start() on enter and aclose() on exit. Stage contracts are imported from zleap.sag.pipeline; durable production contracts come from zleap.sag.operations, zleap.sag.queries, zleap.sag.records, and zleap.sag.maintenance. Parse and Chunking need no started storage; use parse(TextSource(...)) followed by chunk(parsed), or chunk_text(...). See MIGRATING_PIPELINE.md for breaking API changes.

Notes

• Multiple DataEngine instances may coexist in one process. Each owns its relational and vector resources; closing one does not close another instance's pools or client.
• pruned_expand_llm / pruned_expand_rerank / pruned_expand_rff resolve to the registered production Executor. The LLM and rerank variants need real lexical retrieval (LanceDB or Elasticsearch). Algorithms are registered as SearchExecutorSpec values before start() and the engine-local catalog is frozen after startup. Use register_search_algorithm(...) for a complete versioned contract; Missing capabilities or executors fail explicitly unless the request declares a valid fallback_strategy.
• All engine errors derive from SagError — catch it at the boundary.

Links

Examples · API Reference · Changelog · Contributing · Config reference