graph_engine

zocrates.rag.graph_engine ¶

Evidence-Gated Hybrid & Curated Paper Relations Engine.

Discovers, verifies, and transactionally persists high-confidence undirected edges between academic papers in conversion_meta.db without LLMs or cross-encoders: 1. Sub-millisecond 5-feature statistical keywords and Zotero user tags. 2. Chunk-level hybrid search (Chroma dense + SQLite FTS5 BM25 with RRF). 3. The Canonical Evidence Gate: - Method == 'hybrid' (dense semantic + BM25 keyword corroboration), OR - High single-chunk semantic similarity >= 0.65, OR - Multi-chunk dense cohesion (>= 2 chunks and max_sem >= 0.58). 4. Exponential diminishing-returns score rescaling to intuitive [0.50, 0.99]. 5. Ground truth ingestion (human_curated, shared_tags, human_curated_and_hybrid_search). 6. Undirected edge symmetrization with canonical max score and ON CONFLICT semantics.

DEFAULT_GRAPH_RELATIONS `module-attribute` ¶

DEFAULT_GRAPH_RELATIONS = {'enabled': True, 'strategy': 'hybrid_corroborated', 'candidate_pool_size': 50, 'min_hybrid_score': 0.015, 'score_tau': 0.1, 'shared_tags_score': 0.7, 'include_human_curated': True, 'include_shared_tags': True, 'min_shared_tags': 2, 'sync_bundle_files': True}

GENERIC_TAGS `module-attribute` ¶

GENERIC_TAGS = {'read', 'to-read', 'toread', 'unread', 'favorite', 'important', '_reading'}

logger `module-attribute` ¶

logger = logging.getLogger(__name__)

build_search_query ¶

build_search_query(title, keywords, tags=None)

Build a deduplicated, high-signal search query combining title, keywords, and tags.

Guarantees high-IDF lexical coverage without repeating words or bloating the query.

Parameters:

Name	Type	Description	Default
`title`	`str`	Document title.	required
`keywords`	`list[str]`	Extracted statistical keywords.	required
`tags`	`list[str] \| None`	Optional user or automatic Zotero tags.	`None`

Returns:

Type	Description
`str`	Deduplicated space-separated query string capped at 35 terms.

Source code in zocrates/rag/keyword_extractor.py

def build_search_query(
    title: str,
    keywords: list[str],
    tags: list[str] | None = None,
) -> str:
    """Build a deduplicated, high-signal search query combining title, keywords, and tags.

    Guarantees high-IDF lexical coverage without repeating words or bloating the query.

    Args:
        title: Document title.
        keywords: Extracted statistical keywords.
        tags: Optional user or automatic Zotero tags.

    Returns:
        Deduplicated space-separated query string capped at 35 terms.
    """
    clean_title = re.sub(r"[^\w\s\-]", " ", title or "").strip()
    title_words = [w for w in clean_title.split() if len(w) > 1]

    seen_lower = {w.lower() for w in title_words}
    query_parts = list(title_words)

    extra_items = list(keywords or [])
    if tags:
        extra_items.extend(tags)

    for item in extra_items:
        clean_item = re.sub(r"[^\w\s\-]", " ", str(item)).strip()
        item_words = [w for w in clean_item.split() if len(w) > 1]
        new_words = [w for w in item_words if w.lower() not in seen_lower]
        if new_words:
            query_parts.extend(new_words)
            seen_lower.update(w.lower() for w in new_words)

    return " ".join(query_parts[:35])

check_evidence_gate ¶

check_evidence_gate(chunks, min_hybrid_score=0.015)

Check if retrieved chunks satisfy the canonical Evidence Gate.

A candidate paper connection clears the Evidence Gate if: agg_score >= min_hybrid_score AND at least one of: 1. Any chunk has retrieval_method == 'hybrid' (lexical BM25 + dense cos-sim >= 0.50). 2. High single-chunk semantic confidence: max_sem_score >= 0.65. 3. Multi-chunk dense cohesion: len(chunks) >= 2 and max_sem_score >= 0.58.

Parameters:

Name	Type	Description	Default
`chunks`	`list[Any]`	List of retrieved SearchResult chunks or dict equivalents.	required
`min_hybrid_score`	`float`	Minimum aggregated raw RRF score threshold.	`0.015`

Returns:

Type	Description
`tuple[bool, float, float, set[str]]`	Tuple of (is_valid, agg_score, max_sem_score, retrieval_methods).

Source code in zocrates/rag/graph_engine.py

def check_evidence_gate(
    chunks: list[Any],
    min_hybrid_score: float = 0.015,
) -> tuple[bool, float, float, set[str]]:
    """Check if retrieved chunks satisfy the canonical Evidence Gate.

    A candidate paper connection clears the Evidence Gate if:
    agg_score >= min_hybrid_score AND at least one of:
    1. Any chunk has retrieval_method == 'hybrid' (lexical BM25 + dense cos-sim >= 0.50).
    2. High single-chunk semantic confidence: max_sem_score >= 0.65.
    3. Multi-chunk dense cohesion: len(chunks) >= 2 and max_sem_score >= 0.58.

    Args:
        chunks: List of retrieved SearchResult chunks or dict equivalents.
        min_hybrid_score: Minimum aggregated raw RRF score threshold.

    Returns:
        Tuple of (is_valid, agg_score, max_sem_score, retrieval_methods).
    """
    if not chunks:
        return False, 0.0, 0.0, set()

    agg_score = 0.0
    sem_scores: list[float] = []
    methods: set[str] = set()

    for c in chunks:
        if isinstance(c, dict):
            s = float(c.get("score") or 0.0)
            raw_sem = c.get("raw_semantic_score")
            method = str(c.get("retrieval_method") or "unknown")
        else:
            s = float(getattr(c, "score", 0.0) or 0.0)
            raw_sem = getattr(c, "raw_semantic_score", None)
            method = str(getattr(c, "retrieval_method", "unknown") or "unknown")

        agg_score += s
        methods.add(method)
        if raw_sem is not None:
            sem_scores.append(float(raw_sem))

    max_sem_score = max(sem_scores) if sem_scores else 0.0
    has_hybrid = "hybrid" in methods

    # Multi-chunk cohesion requires at least 2 distinct/independent chunks
    chunk_identifiers = set()
    for c in chunks:
        cid = c.get("chunk_id") if isinstance(c, dict) else getattr(c, "chunk_id", None)
        if not cid:
            cid = c.get("chunk_index") if isinstance(c, dict) else getattr(c, "chunk_index", None)
        if not cid:
            cid = id(c)
        chunk_identifiers.add(cid)
    is_multi_chunk = len(chunk_identifiers) >= 2 and max_sem_score >= 0.58

    corroborated = (
        has_hybrid
        or max_sem_score >= 0.65
        or is_multi_chunk
    )

    is_valid = bool(agg_score >= min_hybrid_score and corroborated)
    return is_valid, agg_score, max_sem_score, methods

evaluate_and_persist_edges ¶

evaluate_and_persist_edges(item_key, syncer, sync_bundle_files=True)

Evaluate and persist relations for an item using hybrid search and ground truth.

Looks up item_key metadata, keywords, and tags.
Builds high-signal search query via build_search_query.
Runs syncer.store.search isolating body chunks and excluding Chunk 0 frontmatter.
Groups candidates by parent key, strictly excluding own parent key and sibling attachments.
Clears candidates through the canonical Evidence Gate and exponential score rescaling.
Persists edges transactionally with ON CONFLICT max score and undirected symmetrization.
Syncs manual links and shared user tags if enabled.
If sync_bundle_files is True, regenerates item_relations.md for item_key and all connected neighbors.

Parameters:

Name	Type	Description	Default
`item_key`	`str`	Unique Zotero attachment or item key.	required
`syncer`	`Any`	Active LibrarySyncer instance with vector store and database.	required
`sync_bundle_files`	`bool`	If True, regenerates item_relations.md on disk.	`True`

Returns:

Type	Description
`list[dict[str, Any]]`	List of all verified relation dictionaries for item_key.

Source code in zocrates/rag/graph_engine.py

def evaluate_and_persist_edges(
    item_key: str,
    syncer: Any,
    sync_bundle_files: bool = True,
) -> list[dict[str, Any]]:
    """Evaluate and persist relations for an item using hybrid search and ground truth.

    1. Looks up item_key metadata, keywords, and tags.
    2. Builds high-signal search query via build_search_query.
    3. Runs syncer.store.search isolating body chunks and excluding Chunk 0 frontmatter.
    4. Groups candidates by parent key, strictly excluding own parent key and sibling attachments.
    5. Clears candidates through the canonical Evidence Gate and exponential score rescaling.
    6. Persists edges transactionally with ON CONFLICT max score and undirected symmetrization.
    7. Syncs manual links and shared user tags if enabled.
    8. If sync_bundle_files is True, regenerates item_relations.md for item_key and all connected neighbors.

    Args:
        item_key: Unique Zotero attachment or item key.
        syncer: Active LibrarySyncer instance with vector store and database.
        sync_bundle_files: If True, regenerates item_relations.md on disk.

    Returns:
        List of all verified relation dictionaries for item_key.
    """
    zotero_dir = getattr(syncer, "_zotero_data_dir", None) or getattr(syncer, "zotero_dir", None)
    if zotero_dir:
        zotero_dir = Path(zotero_dir)

    processor = getattr(syncer, "processor", None)
    cache_dir = getattr(processor, "_cache_dir", None) or (get_corpus_dir(zotero_dir) if zotero_dir else None)
    if not cache_dir:
        return []

    cache_dir = Path(cache_dir)
    meta_db_path = cache_dir / "conversion_meta.db"
    if not meta_db_path.exists():
        return []

    cfg = get_graph_relations_config(zotero_dir)
    if not cfg.get("enabled", True):
        return []

    conn = sqlite3.connect(str(meta_db_path), timeout=5)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode = WAL;")
    conn.execute("PRAGMA busy_timeout = 5000;")

    try:
        item_row = conn.execute(
            """
            SELECT item_key, parent_item_key, title, abstract, keywords, content_type
            FROM conversion_meta
            WHERE item_key = ?
            """,
            (item_key,),
        ).fetchone()

        if not item_row:
            return []

        title = str(item_row["title"] or "").strip()
        abstract = str(item_row["abstract"] or "").strip()
        keywords: list[str] = []

        if item_row["keywords"]:
            try:
                parsed_kw = json.loads(item_row["keywords"])
                if isinstance(parsed_kw, list):
                    keywords = [str(k) for k in parsed_kw]
            except Exception:
                keywords = []

        if not keywords:
            # Fall back to extracting keywords from document.md
            doc_path = cache_dir / item_key / "document.md"
            if doc_path.exists():
                try:
                    doc_text = doc_path.read_text(encoding="utf-8", errors="ignore")
                    digest = extract_structural_digest(doc_text)
                    keywords = extract_statistical_keywords(title, digest, max_keywords=6)
                except Exception:
                    pass
            if not keywords and abstract:
                keywords = extract_statistical_keywords(title, abstract, max_keywords=6)

        tags = get_item_tags(zotero_dir, item_key=item_key) if zotero_dir else []
        query = build_search_query(title=title, keywords=keywords, tags=tags)

        # Build mapping for parent isolation and primary bundle resolution
        all_meta_rows = conn.execute(
            "SELECT item_key, parent_item_key, content_type FROM conversion_meta"
        ).fetchall()

        parent_map: dict[str, str] = {}
        parent_to_bundle: dict[str, str] = {}

        for r in all_meta_rows:
            k = r["item_key"]
            pk = r["parent_item_key"] or k
            is_pdf = r["content_type"] == "application/pdf"
            parent_map[k] = pk
            if pk not in parent_to_bundle or is_pdf:
                parent_to_bundle[pk] = k

        source_parent_key = parent_map.get(item_key, item_key)

        # Hybrid search edge discovery
        strategy = cfg.get("strategy", "hybrid_corroborated")
        if strategy != "curated_only" and query and len(query.strip()) >= 3 and hasattr(syncer, "store"):
            candidate_pool_size = int(cfg.get("candidate_pool_size", 50))
            min_hybrid_score = float(cfg.get("min_hybrid_score", 0.015))
            score_tau = float(cfg.get("score_tau", 0.10))

            try:
                search_results = syncer.store.search(
                    query=query,
                    top_k=candidate_pool_size,
                    final_k=candidate_pool_size,
                    chunk_type_filter=["text", "table", "equation", "code", "figure"],
                )
            except Exception as exc:
                logger.warning("Graph hybrid search failed for %s: %s", item_key, exc)
                search_results = []

            # Group chunks by target parent key, strictly excluding own parent and sibling attachments
            by_target: dict[str, list[Any]] = {}
            for chunk in search_results:
                chunk_key = chunk.item_key if hasattr(chunk, "item_key") else chunk.get("item_key")
                target_parent = parent_map.get(chunk_key, chunk_key)

                # Strict sibling guard: exclude self-matches, child notes, and companion attachments
                if target_parent == source_parent_key or chunk_key == item_key:
                    continue

                by_target.setdefault(target_parent, []).append(chunk)

            # Evaluate each candidate group against the Evidence Gate
            for target_parent, chunks in by_target.items():
                is_valid, agg_score, max_sem, methods = check_evidence_gate(
                    chunks, min_hybrid_score=min_hybrid_score
                )
                if not is_valid:
                    continue

                target_bundle_key = parent_to_bundle.get(target_parent, target_parent)
                rescaled_score = rescale_hybrid_score(
                    raw_rrf_score=agg_score,
                    min_hybrid_score=min_hybrid_score,
                    score_tau=score_tau,
                )

                # Check if manual link exists to determine promotion
                is_manual_corroborated = False
                if cfg.get("include_human_curated", True) and zotero_dir:
                    try:
                        from zocrates.rag.zotero_db import _open_db

                        sqlite_path = zotero_dir / "zotero.sqlite"
                        if sqlite_path.exists():
                            with _open_db(sqlite_path) as z_conn:
                                q = """
                                    SELECT 1 FROM itemRelations ir
                                    JOIN items i ON ir.itemID = i.itemID
                                    WHERE (i.key = ? AND ir.object LIKE '%/items/' || ?)
                                       OR (i.key = ? AND ir.object LIKE '%/items/' || ?)
                                """
                                row = z_conn.execute(
                                    q,
                                    (source_parent_key, target_parent, target_parent, source_parent_key),
                                ).fetchone()
                                if row:
                                    is_manual_corroborated = True
                    except Exception:
                        pass

                if is_manual_corroborated:
                    rel_type = "human_curated_and_hybrid_search"
                    final_score = 1.0
                else:
                    rel_type = "hybrid_search"
                    final_score = rescaled_score

                match_evidence = json.dumps(
                    {
                        "raw_rrf_score": round(agg_score, 4),
                        "matched_chunks": len(chunks),
                        "max_semantic_score": round(max_sem, 4),
                        "retrieval_methods": sorted(list(methods)),
                    }
                )

                upsert_edge(
                    conn=conn,
                    source_key=item_key,
                    target_key=target_bundle_key,
                    score=final_score,
                    relation_type=rel_type,
                    match_evidence=match_evidence,
                    symmetrize=True,
                )

        # Ground truth ingestion: manual links & shared tags
        if cfg.get("include_human_curated", True) and zotero_dir:
            sync_manual_relations(conn=conn, zotero_dir=zotero_dir, target_key=item_key)

        if cfg.get("include_shared_tags", True) and zotero_dir:
            sync_shared_tags(
                conn=conn,
                zotero_dir=zotero_dir,
                min_shared_tags=int(cfg.get("min_shared_tags", 2)),
                target_key=item_key,
                score=float(cfg.get("shared_tags_score", 0.70)),
            )

        # Query all verified edges for item_key
        edges_cursor = conn.execute(
            """
            SELECT source_key, target_key, score, relation_type, match_evidence, updated_at
            FROM item_relations
            WHERE source_key = ?
            ORDER BY score DESC, target_key ASC
            """,
            (item_key,),
        )
        persisted_edges = [dict(r) for r in edges_cursor.fetchall()]

        # Atomic bundle file synchronization
        if sync_bundle_files and cfg.get("sync_bundle_files", True):
            try:
                from zocrates.rag.bundle_catalog import regenerate_item_relations_md

                # Regenerate for item_key
                regenerate_item_relations_md(item_key=item_key, cache_dir=cache_dir)

                # Regenerate for all connected neighbors
                connected_neighbors = {e["target_key"] for e in persisted_edges if e.get("target_key")}
                for neighbor_key in connected_neighbors:
                    regenerate_item_relations_md(item_key=neighbor_key, cache_dir=cache_dir)
            except Exception as exc:
                logger.warning(
                    "Failed to synchronize item_relations.md files for %s: %s", item_key, exc
                )

        return persisted_edges
    finally:
        conn.close()

extract_statistical_keywords ¶

extract_statistical_keywords(title, text, max_keywords=6)

Pure-Python 5-feature statistical keyphrase extractor (RAKE + YAKE principles).

Extracts high-precision compound keyphrases in < 0.2 ms on CPU using: 1. Co-occurrence degree / frequency ratio. 2. Casing & acronym multipliers (1.8x ALL_CAPS, 1.3x proper nouns). 3. Title positional boost (2.0x). 4. Sentence spread log-multiplier. 5. Length normalization favoring 2-3 word technical noun phrases.

Parameters:

Name	Type	Description	Default
`title`	`str`	Document title.	required
`text`	`str`	Text to extract from (e.g. structural digest or abstract).	required
`max_keywords`	`int`	Maximum number of keywords to return.	`6`

Returns:

Type	Description
`list[str]`	List of extracted keyphrase strings.

Source code in zocrates/rag/keyword_extractor.py

def extract_statistical_keywords(title: str, text: str, max_keywords: int = 6) -> list[str]:
    """Pure-Python 5-feature statistical keyphrase extractor (RAKE + YAKE principles).

    Extracts high-precision compound keyphrases in < 0.2 ms on CPU using:
    1. Co-occurrence degree / frequency ratio.
    2. Casing & acronym multipliers (1.8x ALL_CAPS, 1.3x proper nouns).
    3. Title positional boost (2.0x).
    4. Sentence spread log-multiplier.
    5. Length normalization favoring 2-3 word technical noun phrases.

    Args:
        title: Document title.
        text: Text to extract from (e.g. structural digest or abstract).
        max_keywords: Maximum number of keywords to return.

    Returns:
        List of extracted keyphrase strings.
    """
    full_text = f"{title}. {text}" if title else text
    sentences = [s.strip() for s in re.split(r"[\n.!?]+", full_text) if s.strip()]
    if not sentences:
        return []

    title_words = set(re.findall(r"\b[A-Za-z0-9\-_]+\b", title.lower())) if title else set()
    sentence_phrases: list[tuple[str, ...]] = []
    word_casing_boost: dict[str, float] = {}
    word_sentence_counts: dict[str, set[int]] = defaultdict(set)

    for sent_idx, sent in enumerate(sentences):
        raw_tokens = re.findall(r"\b[A-Za-z0-9\-_]+\b", sent)
        curr_phrase: list[str] = []
        for raw_tok in raw_tokens:
            clean_tok = raw_tok.lower()
            if clean_tok not in STOPWORDS and not clean_tok.isdigit() and len(clean_tok) > 2:
                curr_phrase.append(clean_tok)
                # Feature 1: Casing & Acronym boost
                if raw_tok.isupper() and len(raw_tok) >= 2:
                    word_casing_boost[clean_tok] = max(word_casing_boost.get(clean_tok, 1.0), 1.8)
                elif raw_tok[0].isupper() and sent_idx > 0:
                    word_casing_boost[clean_tok] = max(word_casing_boost.get(clean_tok, 1.0), 1.3)
                else:
                    word_casing_boost.setdefault(clean_tok, 1.0)
                word_sentence_counts[clean_tok].add(sent_idx)
            else:
                if curr_phrase and len(curr_phrase) <= 4:
                    sentence_phrases.append(tuple(curr_phrase))
                curr_phrase = []
        if curr_phrase and len(curr_phrase) <= 4:
            sentence_phrases.append(tuple(curr_phrase))

    if not sentence_phrases:
        return []

    word_freq: Counter[str] = Counter()
    word_degree: Counter[str] = Counter()
    for phrase in sentence_phrases:
        length = len(phrase)
        for word in phrase:
            word_freq[word] += 1
            word_degree[word] += length

    term_scores: dict[str, float] = {}
    for w in word_freq:
        base_rake = word_degree[w] / word_freq[w]
        casing = word_casing_boost.get(w, 1.0)
        pos_boost = 2.0 if w in title_words else 1.0
        spread = 1.0 + math.log(1 + len(word_sentence_counts[w]))
        term_scores[w] = base_rake * casing * pos_boost * spread

    phrase_scores: dict[str, float] = {}
    for phrase in sentence_phrases:
        p_str = " ".join(phrase)
        length_factor = 1.0 if len(phrase) in (2, 3) else (0.85 if len(phrase) == 1 else 0.75)
        raw_sum = sum(term_scores[w] for w in phrase)
        phrase_scores[p_str] = raw_sum * length_factor

    sorted_phrases = sorted(phrase_scores.items(), key=lambda x: -x[1])
    results: list[str] = []
    seen_words: set[str] = set()
    for p_str, _ in sorted_phrases:
        words = set(p_str.split())
        if not (words & seen_words):
            results.append(p_str)
            seen_words.update(words)
        elif len(words - seen_words) >= 2:
            results.append(p_str)
            seen_words.update(words)
        if len(results) >= max_keywords:
            break

    return results

extract_structural_digest ¶

extract_structural_digest(doc_text, max_lead=2000)

Extract lead abstract/intro plus topical H1/H2/H3 section headers.

Filters out standard academic IMRAD boilerplate headers so only substantive topical section headings are retained alongside lead text.

Parameters:

Name	Type	Description	Default
`doc_text`	`str`	Full document markdown string.	required
`max_lead`	`int`	Maximum character length for lead prose.	`2000`

Returns:

Type	Description
`str`	Structural digest string combining lead text and topical section headers.

Source code in zocrates/rag/keyword_extractor.py

def extract_structural_digest(doc_text: str, max_lead: int = 2000) -> str:
    """Extract lead abstract/intro plus topical H1/H2/H3 section headers.

    Filters out standard academic IMRAD boilerplate headers so only substantive
    topical section headings are retained alongside lead text.

    Args:
        doc_text: Full document markdown string.
        max_lead: Maximum character length for lead prose.

    Returns:
        Structural digest string combining lead text and topical section headers.
    """
    # Strip YAML frontmatter if present
    body = re.sub(r"^---\n.*?\n---\n", "", doc_text, flags=re.DOTALL)
    cleaned = clean_markdown_for_kw(body)

    # Extract section headers
    headers = re.findall(r"^#{1,3}\s+(.+)$", cleaned, flags=re.MULTILINE)
    topical_headers = [h.strip() for h in headers if not is_generic_section_header(h)]

    # Extract lead paragraphs
    lead = cleaned[:max_lead].strip()
    header_block = "\n".join(topical_headers)

    return f"{lead}\n\n{header_block}" if header_block else lead

get_corpus_dir ¶

get_corpus_dir(zotero_data_dir)

Return (and create if needed) the document corpus directory.

Parameters:

Name	Type	Description	Default
`zotero_data_dir`	`Path`	Root Zotero data directory.	required

Returns:

Type	Description
`Path`	Path to `<zocrates_data>/corpus/`, created if absent.

Source code in zocrates/config.py

def get_corpus_dir(zotero_data_dir: Path) -> Path:
    """Return (and create if needed) the document corpus directory.

    Args:
        zotero_data_dir: Root Zotero data directory.

    Returns:
        Path to ``<zocrates_data>/corpus/``, created if absent.
    """
    zocrates_dir = get_zocrates_data_dir(zotero_data_dir)
    legacy_dir = zocrates_dir / "markdown_cache"
    corpus_dir = zocrates_dir / "corpus"
    if legacy_dir.is_dir() and not corpus_dir.exists():
        try:
            legacy_dir.rename(corpus_dir)
            logger.info("Migrated legacy markdown_cache to %s", corpus_dir)
        except OSError as exc:
            logger.warning(
                "Failed to automatically migrate legacy markdown_cache to corpus: %s. "
                "Please rename '%s' to '%s' manually.",
                exc,
                legacy_dir,
                corpus_dir,
            )
    elif legacy_dir.is_dir() and corpus_dir.is_dir():
        logger.debug("Both corpus/ and legacy markdown_cache/ exist. Using %s.", corpus_dir)
    corpus_dir.mkdir(parents=True, exist_ok=True)
    return corpus_dir

get_graph_relations_config ¶

get_graph_relations_config(zotero_data_dir=None, cfg=None)

Return the graph relations configuration dict from zocrates_config.json merged with defaults.

Parameters:

Name	Type	Description	Default
`zotero_data_dir`	`Path \| str \| None`	Path to the Zotero data directory.	`None`
`cfg`	`dict[str, Any] \| None`	Optional in-memory config dict. If provided, bypasses disk read.	`None`

Returns:

Type	Description
`dict[str, Any]`	Dict containing graph relations parameters.

Source code in zocrates/config.py

def get_graph_relations_config(
    zotero_data_dir: Path | str | None = None,
    cfg: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Return the graph relations configuration dict from zocrates_config.json merged with defaults.

    Args:
        zotero_data_dir: Path to the Zotero data directory.
        cfg: Optional in-memory config dict. If provided, bypasses disk read.

    Returns:
        Dict containing graph relations parameters.
    """
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    res = copy.deepcopy(DEFAULT_GRAPH_RELATIONS)
    graph_cfg = cfg.get("graph_relations")
    if isinstance(graph_cfg, dict):
        res.update(graph_cfg)
    return res

get_item_tags ¶

get_item_tags(zotero_dir, item_id=None, item_key=None, tag_type=None)

Retrieve user and automatic tags for an item from zotero.sqlite.

Excludes generic workflow tags (e.g. 'read', 'unread', 'to-read', 'favorite'). If item is an attachment, also resolves tags from its parent item.

Parameters:

Name	Type	Description	Default
`zotero_dir`	`Path \| str`	Path to the root Zotero data directory.	required
`item_id`	`int \| None`	Optional integer itemID in zotero.sqlite.	`None`
`item_key`	`str \| None`	Optional string key in zotero.sqlite (items.key).	`None`
`tag_type`	`int \| None`	Optional integer tag type filter (0 = user/manual tag, 1 = automatic tag). If None (default), returns all tag types.	`None`

Returns:

Type	Description
`list[str]`	List of non-generic tag name strings.

Source code in zocrates/rag/keyword_extractor.py

def get_item_tags(
    zotero_dir: Path | str,
    item_id: int | None = None,
    item_key: str | None = None,
    tag_type: int | None = None,
) -> list[str]:
    """Retrieve user and automatic tags for an item from zotero.sqlite.

    Excludes generic workflow tags (e.g. 'read', 'unread', 'to-read', 'favorite').
    If item is an attachment, also resolves tags from its parent item.

    Args:
        zotero_dir: Path to the root Zotero data directory.
        item_id: Optional integer itemID in zotero.sqlite.
        item_key: Optional string key in zotero.sqlite (items.key).
        tag_type: Optional integer tag type filter (0 = user/manual tag, 1 = automatic tag).
            If None (default), returns all tag types.

    Returns:
        List of non-generic tag name strings.
    """
    if item_id is None and item_key is None:
        return []

    zotero_path = Path(zotero_dir)
    sqlite_path = zotero_path / "zotero.sqlite"
    if not sqlite_path.exists():
        return []

    tags: list[str] = []
    seen: set[str] = set()

    try:
        from zocrates.rag.zotero_db import _open_db

        with _open_db(sqlite_path) as conn:
            target_item_ids: list[int] = []
            if item_id is not None:
                target_item_ids.append(item_id)
                try:
                    parent_row = conn.execute(
                        "SELECT parentItemID FROM itemAttachments WHERE itemID = ?",
                        (item_id,),
                    ).fetchone()
                    if parent_row and parent_row["parentItemID"]:
                        target_item_ids.append(parent_row["parentItemID"])
                except Exception:
                    pass
            elif item_key is not None:
                try:
                    row = conn.execute(
                        "SELECT itemID FROM items WHERE key = ?",
                        (item_key,),
                    ).fetchone()
                    if row and row["itemID"]:
                        target_item_ids.append(row["itemID"])
                        try:
                            parent_row = conn.execute(
                                "SELECT parentItemID FROM itemAttachments WHERE itemID = ?",
                                (row["itemID"],),
                            ).fetchone()
                            if parent_row and parent_row["parentItemID"]:
                                target_item_ids.append(parent_row["parentItemID"])
                        except Exception:
                            pass
                except Exception:
                    pass

            if not target_item_ids:
                return []

            placeholders = ",".join("?" for _ in target_item_ids)
            sql_params: list[Any] = list(target_item_ids)
            type_clause = ""
            if tag_type is not None:
                cols = {c[1] for c in conn.execute("PRAGMA table_info(itemTags)").fetchall()}
                if "type" in cols:
                    type_clause = " AND it.type = ?"
                    sql_params.append(tag_type)

            sql = f"""
                SELECT t.name
                FROM itemTags it
                JOIN tags t ON it.tagID = t.tagID
                WHERE it.itemID IN ({placeholders}){type_clause}
            """
            rows = conn.execute(sql, sql_params).fetchall()
            for r in rows:
                tag_name = str(r["name"]).strip()
                if tag_name.lower() not in GENERIC_TAGS and tag_name.lower() not in seen:
                    tags.append(tag_name)
                    seen.add(tag_name.lower())
    except Exception as exc:
        logger.warning("Could not fetch tags from %s: %s", sqlite_path, exc)

    return tags

rescale_hybrid_score ¶

rescale_hybrid_score(raw_rrf_score, min_hybrid_score=0.015, score_tau=0.1)

Rescale raw aggregated RRF score to [0.50, 0.99] using exponential diminishing returns.

Formula

score = round(0.50 + 0.49 * (1.0 - exp(-(raw_rrf_score - min_hybrid_score) / score_tau)), 2)

Clamped strictly to [0.50, 0.99].

Parameters:

Name	Type	Description	Default
`raw_rrf_score`	`float`	Aggregated raw Reciprocal Rank Fusion score.	required
`min_hybrid_score`	`float`	Minimum raw score threshold to qualify.	`0.015`
`score_tau`	`float`	Exponential decay constant governing progression.	`0.1`

Returns:

Type	Description
`float`	Rescaled confidence score in [0.50, 0.99] rounded to 2 decimals.

Source code in zocrates/rag/graph_engine.py

def rescale_hybrid_score(
    raw_rrf_score: float,
    min_hybrid_score: float = 0.015,
    score_tau: float = 0.10,
) -> float:
    """Rescale raw aggregated RRF score to [0.50, 0.99] using exponential diminishing returns.

    Formula:
        score = round(0.50 + 0.49 * (1.0 - exp(-(raw_rrf_score - min_hybrid_score) / score_tau)), 2)
    Clamped strictly to [0.50, 0.99].

    Args:
        raw_rrf_score: Aggregated raw Reciprocal Rank Fusion score.
        min_hybrid_score: Minimum raw score threshold to qualify.
        score_tau: Exponential decay constant governing progression.

    Returns:
        Rescaled confidence score in [0.50, 0.99] rounded to 2 decimals.
    """
    if score_tau <= 0:
        score_tau = 0.10

    diff = raw_rrf_score - min_hybrid_score
    scaled = 0.50 + 0.49 * (1.0 - math.exp(-diff / score_tau))
    clamped = max(0.50, min(0.99, scaled))
    return round(clamped, 2)

sync_manual_relations ¶

sync_manual_relations(conn, zotero_dir, target_key=None)

Read itemRelations from zotero.sqlite and persist human_curated edges.

If a manual link matches an existing hybrid_search edge, promotes it to human_curated_and_hybrid_search with score 1.0 while preserving underlying evidence. Otherwise, inserts human_curated with score 1.0.

Parameters:

Name	Type	Description	Default
`conn`	`Connection`	Open SQLite connection to conversion_meta.db.	required
`zotero_dir`	`Path`	Root Zotero data directory containing zotero.sqlite.	required
`target_key`	`str \| None`	Optional filter to restrict sync to relations involving this key.	`None`

Returns:

Type	Description
`int`	Number of manual edges processed and persisted.

Source code in zocrates/rag/graph_engine.py

def sync_manual_relations(
    conn: sqlite3.Connection,
    zotero_dir: Path,
    target_key: str | None = None,
) -> int:
    """Read itemRelations from zotero.sqlite and persist human_curated edges.

    If a manual link matches an existing hybrid_search edge, promotes it to
    human_curated_and_hybrid_search with score 1.0 while preserving underlying
    evidence. Otherwise, inserts human_curated with score 1.0.

    Args:
        conn: Open SQLite connection to conversion_meta.db.
        zotero_dir: Root Zotero data directory containing zotero.sqlite.
        target_key: Optional filter to restrict sync to relations involving this key.

    Returns:
        Number of manual edges processed and persisted.
    """
    sqlite_path = zotero_dir / "zotero.sqlite"
    if not sqlite_path.exists():
        return 0

    # Build key mapping from conversion_meta
    key_map: dict[str, str] = {}
    try:
        rows = conn.execute(
            "SELECT item_key, parent_item_key, content_type FROM conversion_meta"
        ).fetchall()
        for r in rows:
            k = r["item_key"]
            pk = r["parent_item_key"]
            is_pdf = r["content_type"] == "application/pdf"
            key_map[k] = k
            if pk:
                if pk not in key_map or is_pdf:
                    key_map[pk] = k
    except Exception:
        pass

    target_parent = None
    if target_key:
        try:
            p_row = conn.execute(
                "SELECT parent_item_key FROM conversion_meta WHERE item_key = ?",
                (target_key,),
            ).fetchone()
            if p_row and p_row["parent_item_key"]:
                target_parent = p_row["parent_item_key"]
        except Exception:
            pass

    synced_count = 0
    try:
        from zocrates.rag.zotero_db import _open_db

        with _open_db(sqlite_path) as z_conn:
            has_deleted_items = bool(
                z_conn.execute(
                    "SELECT 1 FROM sqlite_master WHERE type='table' AND name='deletedItems'"
                ).fetchone()
            )
            del_clause = "WHERE i.itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted_items else ""
            sql = f"""
                SELECT i.key AS source_key, ir.object AS target_uri
                FROM itemRelations ir
                JOIN items i ON ir.itemID = i.itemID
                {del_clause}
            """
            manual_rows = z_conn.execute(sql).fetchall()

            for r in manual_rows:
                raw_src = str(r["source_key"]).strip()
                raw_uri = str(r["target_uri"]).strip()
                if "/items/" not in raw_uri:
                    continue

                raw_dest = raw_uri.split("/items/")[-1].strip()
                if not raw_dest:
                    continue

                # Filter by target_key if requested
                if target_key:
                    matches_src = raw_src in (target_key, target_parent)
                    matches_dst = raw_dest in (target_key, target_parent)
                    if not (matches_src or matches_dst):
                        continue

                eff_src = key_map.get(raw_src, raw_src)
                eff_dst = key_map.get(raw_dest, raw_dest)
                if eff_src == eff_dst:
                    continue

                # Check if edge already exists for this pair
                existing_hybrid = None
                already_dual = False
                try:
                    ex_dual = conn.execute(
                        """
                        SELECT 1 FROM item_relations
                        WHERE source_key = ? AND target_key = ? AND relation_type = 'human_curated_and_hybrid_search'
                        """,
                        (eff_src, eff_dst),
                    ).fetchone()
                    if ex_dual:
                        already_dual = True
                    else:
                        ex_row = conn.execute(
                            """
                            SELECT match_evidence FROM item_relations
                            WHERE source_key = ? AND target_key = ? AND relation_type = 'hybrid_search'
                            """,
                            (eff_src, eff_dst),
                        ).fetchone()
                        if ex_row:
                            existing_hybrid = ex_row["match_evidence"]
                except Exception:
                    pass

                if already_dual:
                    continue

                if existing_hybrid is not None:
                    rel_type = "human_curated_and_hybrid_search"
                    match_evidence = existing_hybrid
                else:
                    rel_type = "human_curated"
                    match_evidence = json.dumps({"source": "zotero_itemRelations"})

                upsert_edge(
                    conn=conn,
                    source_key=eff_src,
                    target_key=eff_dst,
                    score=1.0,
                    relation_type=rel_type,
                    match_evidence=match_evidence,
                    symmetrize=True,
                )
                synced_count += 1
    except Exception as exc:
        logger.warning("Failed to sync manual relations from %s: %s", sqlite_path, exc)

    return synced_count

sync_shared_tags ¶

sync_shared_tags(conn, zotero_dir, min_shared_tags=2, target_key=None, score=None)

Check for shared user tags (type=0) and persist shared_tags edges.

Requires at least min_shared_tags (default: 2) distinctive non-generic user tags.

Parameters:

Name	Type	Description	Default
`conn`	`Connection`	Open SQLite connection to conversion_meta.db.	required
`zotero_dir`	`Path`	Root Zotero data directory containing zotero.sqlite.	required
`min_shared_tags`	`int`	Minimum count of shared distinctive tags required.	`2`
`target_key`	`str \| None`	Optional filter to restrict comparison to this item key.	`None`
`score`	`float \| None`	Optional score override (defaults to config shared_tags_score or 0.70).	`None`

Returns:

Type	Description
`int`	Number of shared_tags edges created.

Source code in zocrates/rag/graph_engine.py

def sync_shared_tags(
    conn: sqlite3.Connection,
    zotero_dir: Path,
    min_shared_tags: int = 2,
    target_key: str | None = None,
    score: float | None = None,
) -> int:
    """Check for shared user tags (type=0) and persist shared_tags edges.

    Requires at least min_shared_tags (default: 2) distinctive non-generic user tags.

    Args:
        conn: Open SQLite connection to conversion_meta.db.
        zotero_dir: Root Zotero data directory containing zotero.sqlite.
        min_shared_tags: Minimum count of shared distinctive tags required.
        target_key: Optional filter to restrict comparison to this item key.
        score: Optional score override (defaults to config shared_tags_score or 0.70).

    Returns:
        Number of shared_tags edges created.
    """
    sqlite_path = zotero_dir / "zotero.sqlite"
    if not sqlite_path.exists():
        return 0

    effective_score = score
    if effective_score is None:
        try:
            from zocrates.config import get_graph_relations_config
            cfg = get_graph_relations_config(zotero_dir)
            effective_score = float(cfg.get("shared_tags_score", 0.70))
        except Exception:
            effective_score = 0.70

    try:
        rows = conn.execute("SELECT item_key, parent_item_key FROM conversion_meta").fetchall()
        items = [(r["item_key"], r["parent_item_key"]) for r in rows]
    except Exception:
        return 0

    if not items:
        return 0

    # Retrieve user tags (tag_type=0) for all items
    item_tags_map: dict[str, set[str]] = {}
    for k, pk in items:
        tags = set(get_item_tags(zotero_dir, item_key=k, tag_type=0))
        if pk and not tags:
            tags = set(get_item_tags(zotero_dir, item_key=pk, tag_type=0))
        item_tags_map[k] = tags

    created_count = 0
    if target_key:
        src_tags = item_tags_map.get(target_key, set())
        if not src_tags:
            return 0
        for other_key, other_tags in item_tags_map.items():
            if other_key == target_key:
                continue
            shared = src_tags & other_tags
            if len(shared) >= min_shared_tags:
                evidence = json.dumps({"shared_tags": sorted(list(shared))})
                upsert_edge(
                    conn=conn,
                    source_key=target_key,
                    target_key=other_key,
                    score=effective_score,
                    relation_type="shared_tags",
                    match_evidence=evidence,
                    symmetrize=True,
                )
                created_count += 1
    else:
        all_keys = sorted(item_tags_map.keys())
        for i in range(len(all_keys)):
            k1 = all_keys[i]
            t1 = item_tags_map[k1]
            if len(t1) < min_shared_tags:
                continue
            for j in range(i + 1, len(all_keys)):
                k2 = all_keys[j]
                t2 = item_tags_map[k2]
                shared = t1 & t2
                if len(shared) >= min_shared_tags:
                    evidence = json.dumps({"shared_tags": sorted(list(shared))})
                    upsert_edge(
                        conn=conn,
                        source_key=k1,
                        target_key=k2,
                        score=effective_score,
                        relation_type="shared_tags",
                        match_evidence=evidence,
                        symmetrize=True,
                    )
                    created_count += 1

    return created_count

upsert_edge ¶

upsert_edge(conn, source_key, target_key, score, relation_type, match_evidence=None, symmetrize=True)

Insert or update an edge in SQLite item_relations with max-score semantics.

Always enforces undirected edge symmetrization by persisting both (A, B) and (B, A) with the canonical max score when symmetrize is True.

Parameters:

Name	Type	Description	Default
`conn`	`Connection`	Open SQLite connection to conversion_meta.db.	required
`source_key`	`str`	Source item or attachment key.	required
`target_key`	`str`	Target item or attachment key.	required
`score`	`float`	Edge confidence score in [0.50, 1.0].	required
`relation_type`	`str`	Relation category ('human_curated_and_hybrid_search', 'human_curated', 'hybrid_search', or 'shared_tags').	required
`match_evidence`	`str \| None`	Optional JSON evidence debug summary string.	`None`
`symmetrize`	`bool`	If True, writes both directional rows with canonical max score.	`True`

Source code in zocrates/rag/graph_engine.py

def upsert_edge(
    conn: sqlite3.Connection,
    source_key: str,
    target_key: str,
    score: float,
    relation_type: str,
    match_evidence: str | None = None,
    symmetrize: bool = True,
) -> None:
    """Insert or update an edge in SQLite item_relations with max-score semantics.

    Always enforces undirected edge symmetrization by persisting both (A, B)
    and (B, A) with the canonical max score when symmetrize is True.

    Args:
        conn: Open SQLite connection to conversion_meta.db.
        source_key: Source item or attachment key.
        target_key: Target item or attachment key.
        score: Edge confidence score in [0.50, 1.0].
        relation_type: Relation category ('human_curated_and_hybrid_search',
            'human_curated', 'hybrid_search', or 'shared_tags').
        match_evidence: Optional JSON evidence debug summary string.
        symmetrize: If True, writes both directional rows with canonical max score.
    """
    if source_key == target_key:
        return

    conn.execute("PRAGMA busy_timeout = 5000;")
    updated_at = datetime.now(UTC).isoformat()

    if symmetrize:
        # Determine canonical max score across existing forward and backward rows
        existing_scores = [score]
        try:
            cursor = conn.execute(
                """
                SELECT score FROM item_relations
                WHERE (source_key = ? AND target_key = ? AND relation_type = ?)
                   OR (source_key = ? AND target_key = ? AND relation_type = ?)
                """,
                (source_key, target_key, relation_type, target_key, source_key, relation_type),
            )
            for row in cursor.fetchall():
                existing_scores.append(float(row["score"] if isinstance(row, sqlite3.Row) else row[0]))
        except Exception:
            pass

        canonical_score = max(existing_scores)
        pairs = [(source_key, target_key), (target_key, source_key)]
    else:
        canonical_score = score
        pairs = [(source_key, target_key)]

    sql = """
        INSERT INTO item_relations (
            source_key, target_key, score, relation_type, match_evidence, updated_at
        ) VALUES (?, ?, ?, ?, ?, ?)
        ON CONFLICT(source_key, target_key, relation_type) DO UPDATE SET
            score = max(item_relations.score, excluded.score),
            match_evidence = excluded.match_evidence,
            updated_at = excluded.updated_at;
    """
    with conn:
        for src, dst in pairs:
            conn.execute(
                sql,
                (src, dst, canonical_score, relation_type, match_evidence, updated_at),
            )

graph_engine

zocrates.rag.graph_engine ¶

DEFAULT_GRAPH_RELATIONS module-attribute ¶

GENERIC_TAGS module-attribute ¶

logger module-attribute ¶

build_search_query ¶

check_evidence_gate ¶

evaluate_and_persist_edges ¶

extract_statistical_keywords ¶

extract_structural_digest ¶

get_corpus_dir ¶

get_graph_relations_config ¶

get_item_tags ¶

rescale_hybrid_score ¶

sync_manual_relations ¶

sync_shared_tags ¶

upsert_edge ¶

DEFAULT_GRAPH_RELATIONS `module-attribute` ¶

GENERIC_TAGS `module-attribute` ¶

logger `module-attribute` ¶