七层记忆的 全息大脑内核

trust_score = (helpful_count + α * prior) / (retrieval_count + α)

Where:
  α = 2          (strength of the prior, i.e. "virtual observations")
  prior = 0.50  (initial belief before any evidence)

def _search_facts(query, top_k=3):
    # Uses FTS5 over memory_store.db
    # Tokenizes query, builds OR-query from tokens ≥4 chars
    # Returns top-3 facts by rank, truncated to 200 chars
    # Fail-open: returns [] on any error

{
  "vectors": {
    "dense": { "size": 4096, "distance": "Cosine" }
  },
  "sparse_vectors": {
    "sparse": { "index": { "on_disk": false } }
  }
}

search_with_fallback(dense_vector, sparse_vector, query_text, top_k, score_threshold)
  └── t0 = time.perf_counter()
  │
  ├─── [Level 1] if sparse_vector is not None:
  │       POST /collections/knowledge_base/points/query
  │       prefetch: [dense (top_k*3), sparse (top_k*3)]
  │       query: {"fusion": "rrf"}
  │       → RRF fusion: score = Σ 1/(k + rank_i) across dense + sparse
  │
  ├─── [Level 2] elif dense_vector is not None:
  │       POST /collections/knowledge_base/points/query
  │       using: "dense"
  │       → Pure cosine similarity, no fusion
  │
  └─── [Levels 3+4] Exception caught:
          lexical_search_in_vault() → sqlite_keyword_search() → []

  Returns: (results, fallback_level, qdrant_latency_ms, fallback_latency_ms)

def _search_qdrant(query, top_k=2, threshold=0.55):
    # 1. embed_query() → dense 4096d vector via OpenRouter
    # 2. embed_query_sparse() → BM25 via FastEmbed subprocess
    # 3. search_with_fallback() → 4-level cascade
    # 4. Results labeled [qdrant] in system prompt
    # 5. Per-session dedup by point ID (_injected_qdrant set)
    #
    # Key pitfall: context_enhancer.py reads OPENROUTER_API_KEY (singular).
    # Before importing, hooks.py injects: os.environ["OPENROUTER_API_KEY"] = resolved_key

_SOCIAL_CLOSERS = frozenset({
    "ok", "obrigado", "valeu", "beleza", "blz", "tks", "thanks",
    "👍", "👌", "✅", "feito", "certo", "confirmo", "entendido",
    "sim", "não", "claro", "perfeito", "ótimo"
})

def _is_social_close(text):
    stripped = text.strip().lower()
    if stripped in _SOCIAL_CLOSERS: return True
    # Very short ASCII-only without technical markers
    if len(stripped) < 6 and stripped.isascii() and not any(
        c in stripped for c in "://.@#$_?"): return True
    return False

def on_session_start(session_id="", platform="", **kwargs):
    global _last_query_tokens, _injected_fabric, _injected_qdrant, _injected_sessions
    _last_query_tokens = set()
    _injected_fabric.clear()
    _injected_qdrant.clear()
    _injected_sessions.clear()
    state.session_id = session_id
    state.exchanges = []

    # 1. Load SOUL (the agent's identity/values document)
    soul = state.load_soul()
    if soul: parts.append(soul.strip())

    # 2. Pending handoffs from other agents (tasks, reviews, tickets)
    open_tasks, reviews, open_tickets = state.read_pending()
    if open_tasks:  # → [fabric] items assigned to you
    if reviews:     # → [fabric] reviews of your work
    if open_tickets: # → [fabric] tickets assigned to you

    # 3. Cross-agent feedback (non-pending items from other agents)
    feedback = state.read_cross_agent(3)

    # 4. Recent entries from this agent
    entries = state.read_recent(limit=5)  # → [fabric] recent activity

    # 5. Creative state (open questions + recent learnings)
    if creative["questions"]:  # → [fabric] open questions
    if creative["learnings"]:  # → [fabric] learnings

    return {"context": "\n".join(parts)} if parts else None

def pre_llm_call(session_id="", user_message="", is_first_turn=False, **kwargs):
    # ── Overlap gate ──
    # Only suppress injection on NEAR-LITERAL repetition (>0.85 token overlap).
    # The old 0.6 gate killed injection in long single-topic sessions.
    tokens = _tokenize(user_message)   # alphanumeric tokens ≥4 chars, stopwords removed
    if _last_query_tokens:
        overlap = len(tokens & _last_query_tokens) / max(len(tokens), 1)
        if overlap > 0.85: return None   # suppress — near-identical turn

    _last_query_tokens = tokens
    is_social = _is_social_close(user_message)

    # ── 4 independent searches ──
    results          = state.recall(user_message, max_results=5, agent=agent)  # Fabric
    qdrant_results   = _search_qdrant(user_message, top_k=2, threshold=0.55)  # Qdrant
    session_results  = _search_sessions(user_message, session_id, top_k=2)   # Sessions FTS5
    fact_results     = _search_facts(user_message, top_k=3) if is_first_turn else []  # Facts

    # ── Per-source dedup (within session) ──
    # Fabric:   dedup by entry id (str) — _injected_fabric set
    # Qdrant:   dedup by point id — _injected_qdrant set
    # Sessions: dedup by session_id — _injected_sessions set
    # Facts:    no dedup needed (first-turn only)

    # → Returns {"context": "[fabric]...\n\n[qdrant]...\n\n[sessions]...\n\n[facts]..."}

def post_llm_call(session_id="", user_message="", assistant_response="", platform="", **kwargs):
    state.exchanges.append({"user": user_message[:200], "assistant": assistant_response[:500]})

    # ── Decision capture ──
    # Requires: decision keyword + outcome keyword + substantial response (>200 chars)
    #           + substantial user request (>50 chars)
    if (DECISION_RE.search(assistant_response)
        and OUTCOME_RE.search(assistant_response)
        and len(assistant_response) > 200
        and len(user_text) > 50):
        body = f"Task: {user_text[:300]}\n\nResult: {assistant_response[:500]}"
        state.write_entry("decision", body, summary[:80],
                          platform=plat, status=entry_status, training_value="high")

    # ── Creative tracking (broader _THEME_RE — no writes) ──
    if _THEME_RE.search(assistant_response):
        theme = _extract_theme(assistant_response)      # top-3 non-stopword words, ≥4 chars
        if theme not in creative["themes"]: creative["themes"].append(theme)

    if _EVAL_RE.search(assistant_response):
        learning = _extract_sentence(assistant_response, _EVAL_RE)  # first matching sentence
        if learning and learning not in creative["learnings"]: creative["learnings"].append(learning)

    if _QUESTION_RE.search(assistant_response):
        question = _extract_sentence(assistant_response, _QUESTION_RE)
        if question and question not in creative["questions"]: creative["questions"].append(question)

    state.save_creative(creative)

def on_session_end(session_id="", platform="", completed=False, **kwargs):
    state.write_memory_file(creative)    # persist creative state to disk
    if not state.exchanges: return

    scores = state.score_session()
    if scores["total"] < 0.2: return    # too trivial — skip

    # ── LLM extraction (primary) ──
    transcript = _build_transcript(state.exchanges)   # compact [Turn N — User/Agent] format
    entries = _llm_extract_entries(transcript)        # LLM analyzes → structured entries

    if entries:
        for entry in entries:
            state.write_entry(entry["type"], entry["content"], entry["summary"],
                              platform=plat, training_value=entry["training_value"],
                              status="completed")
        logger.info("LLM extracted %d entries from session", len(entries))
    else:
        # ── Legacy fallback ──
        _legacy_session_write(platform, scores)  # truncated original behavior

{
  "type": "decision" | "resolution" | "note",
  "summary": "one line, max 80 chars, original language",
  "content": "markdown with ## Context, ## Action/Decision, ## Outcome",
  "training_value": "high" | "normal" | "low"
}

def _search_sessions(query, current_session_id="", top_k=2):
    # FTS5 over state.db (Hermes session DB)
    # Builds OR-query from meaningful tokens (≥4 chars, stopwords removed)
    toks = [t for t in _tokenize(query) if len(t) >= 4]
    fts_query = " OR ".join(toks[:8])   # max 8 terms to avoid query complexity

    # FTS5 snippet() + GROUP BY are incompatible in same SELECT
    # → fetch top-20 ranked rows, dedup by session_id in Python
    SELECT m.session_id, s.title, s.started_at,
           snippet(messages_fts, 0, '', '', '…', 12) AS snip
    FROM messages_fts
    JOIN messages m ON m.id = messages_fts.rowid
    LEFT JOIN sessions s ON s.id = m.session_id
    WHERE messages_fts MATCH ?
      AND m.session_id != ?
      AND m.role IN ('user','assistant')
    ORDER BY rank
    LIMIT 20

def _validate_safe_content(text: str) -> str:
    if not text or len(text) < 20: return text
    directivess = len(re.findall(
        r"(?i)\b(ignore|forget|disregard|override|replace|pretend|"
        r"act\s+as|you\s+(are|must|will|shall))\b", text))
    if directivess >= 3 and directivess / max(len(text), 1) > 0.02:
        return "[SANITIZED]"
    return text

$HERMES_HOME/.icarus-telemetry.jsonl

{"ts": "2026-06-10T14:23:01Z", "event": "recall",
 "source": "pre_llm_call", "query": "how do I configure Qdrant...",
 "result_count": 3, "result_ids": ["a1b2", "c3d4", "e5f6"],
 "session_id": "sess-abc123", "agent": "hermes"}

{"ts": "2026-06-10T14:23:05Z", "event": "usage",
 "action": "referenced", "entry_id": "c3d4",
 "session_id": "sess-abc123", "agent": "hermes"}

def _summarize_telemetry_events(events):
    recalls = [e for e in events if e.get("event") == "recall"]
    usages  = [e for e in events if e.get("event") == "usage"]
    recalled_ids = set(id for r in recalls for id in r.get("result_ids", []))
    used_ids    = set(u.get("entry_id", "") for u in usages) - {""}
    used_from_recall = used_ids & recalled_ids
    return {
        "total_recalls": len(recalls),
        "total_usages": sum(1 for u in usages if u.get("entry_id", "") in recalled_ids),
        "unique_entries_recalled": len(recalled_ids),
        "unique_entries_used": len(used_from_recall),
        "usage_rate": round(len(used_from_recall) / max(len(recalled_ids), 1), 2),
    }

brief = state.build_brief()
# {
#   "pending": { "open_tasks": 2, "reviews_of_my_work": 1, "open_tickets": 0, "items": [...] },
#   "recent_work": [{ "summary": "...", "timestamp": "2026-06-09T15:30" }],
#   "from_others": ["agent-x: Fixed the Qdrant upsert bug"],
#   "suggested_action": "Pick up: Review deployment script from hermes (id a1b2c3)",
#   "recall_stats": { "total_recalls": 12, "usage_rate": 0.58, ... }
# }