From e412fdc0c651e73befd35cbadc8b76abea4ebe0c Mon Sep 17 00:00:00 2001 From: dimknaf <136385722+dimknaf@users.noreply.github.com> Date: Tue, 16 Jun 2026 13:28:28 +0100 Subject: [PATCH 1/5] feat(wiki): batch related orphan triage into one maintainer call The maintainer triaged one orphan per LLM call; same-subject keywords sharing a source fact each repeated the same research. With WIKI_TRIAGE_CLUSTER on (default) it now claims a few related pending triage jobs together (non-hub keywords sharing a source fact + that fact's own job) and decides every seed in ONE agent call, one decision per seed. run_cron and the per-job lifecycle are unchanged: each job still closes with its own status; a singleton (or flag off) is a 1-seed unit, as before. Validated against the one-at-a-time path on the same snapshot: identical decision split and near-identical wikis, at ~1.6 orphans/call (~37% fewer LLM calls), with multi-seed batches deciding each seed independently (no conflation). --- .../agent/prompts/wiki_maintainer_prompt.md | 35 ++- braindb/agent/schemas.py | 35 +++ braindb/agent/tools.py | 5 +- braindb/routers/wiki.py | 288 ++++++++++-------- braindb/services/wiki_jobs.py | 88 ++++++ 5 files changed, 313 insertions(+), 138 deletions(-) diff --git a/braindb/agent/prompts/wiki_maintainer_prompt.md b/braindb/agent/prompts/wiki_maintainer_prompt.md index 746cd37..6fa7d79 100644 --- a/braindb/agent/prompts/wiki_maintainer_prompt.md +++ b/braindb/agent/prompts/wiki_maintainer_prompt.md @@ -1,13 +1,18 @@ -You are the **BrainDB Wiki Maintainer**, working on exactly ONE case. +You are the **BrainDB Wiki Maintainer**. You are given a small batch of SEEDS — +usually one, sometimes a few orphan entities that share a source fact — and you +decide ONE action for EACH seed, independently. A "wiki" is a synthesised, human-readable page (entity_type = `wiki`) about ONE real-world subject, built from the fact/thought/source entities that are genuinely about that subject. -Your case (THE SEED) and the numbered WIKIS catalog are at the **END** of -this prompt. Read the static rules here first, then act on the data there. -The single seed is rarely enough to decide correctly — you MUST investigate -the surrounding reality before deciding. +Your SEEDS and the numbered WIKIS catalog are at the **END** of this prompt. +Read the static rules here first, then act on the data there. A seed is rarely +enough to decide correctly — you MUST investigate the surrounding reality before +deciding, FOR EACH seed. The seeds may share context (they co-occur on a fact), +but **co-occurrence is NOT identity**: a person, a company, and a technique can +sit on one fact yet be three different subjects with three different decisions — +decide each seed on its OWN subject. ## Research FIRST with the powerful tools (this is mandatory) @@ -73,7 +78,7 @@ attach/consolidate to wikis that appear in that numbered catalog. You never see or emit a uuid; the harness maps your number back to the real wiki. If the subject is not in the catalog, you cannot attach/consolidate to it. -## Decide ONE action for THIS seed — STRICT PRECEDENCE, in this order +## Decide ONE action PER SEED — STRICT PRECEDENCE, in this order Evaluate top to bottom and take the FIRST that applies. `create` is the last resort, not the default. This ordering is how the wiki set heals over time — @@ -104,10 +109,13 @@ writer stage does, and it will research further. ## Output — STRICT -Finish by calling `final_answer` exactly once. Its argument is a typed -object — the tool's schema defines and validates the fields; you just fill -them (no raw JSON text, no prose): +Finish by calling `final_answer` exactly once. Its argument is a typed object +with a `decisions` array — **exactly one entry per seed** in the SEEDS list +below (do not omit a seed, do not invent seeds). The tool's schema defines and +validates the fields; you just fill them (no raw JSON text, no prose). Each +entry has: +- `entity_id` — the seed's id, copied VERBATIM from the SEEDS list. - `action` — one of `attach`, `create`, `consolidate`, `skip`, `ambiguous`. - `target_wiki_no` — required for `attach`: the catalog NUMBER of the wiki (an integer from the WIKIS list at the end); null otherwise. @@ -121,14 +129,9 @@ them (no raw JSON text, no prose): --- -## THE SEED (your one case) +## THE SEEDS (decide one action for EACH) -- entity_id: `{entity_id}` -- entity_type: `{entity_type}` -- keywords: {keywords} -- summary: {summary} -- content: -{content} +{seeds} ## WIKIS catalog (existing wikis — reference these BY NUMBER) diff --git a/braindb/agent/schemas.py b/braindb/agent/schemas.py index 062e309..e099bb2 100644 --- a/braindb/agent/schemas.py +++ b/braindb/agent/schemas.py @@ -206,6 +206,41 @@ def _coerce_consolidate_nos(cls, v): return _coerce_to_list(v) +class MaintainerClusterItem(MaintainerDecision): + """One per-seed decision inside a clustered triage. Identical to + `MaintainerDecision` (same fields + the same forgiving coercion validators, + inherited) plus the `entity_id` of the seed it applies to, so the harness + maps each decision back to its orphan.""" + entity_id: str = Field( + ..., + description=( + "The seed's entity_id, copied VERBATIM from the SEEDS list in the " + "prompt. Identifies which orphan THIS decision is for." + ), + ) + + +class MaintainerClusterDecision(BaseModel): + """The maintainer's output for a triage cluster: one decision per seed. A + singleton triage is just a cluster of one (a single-element list), so this + is the maintainer's only output schema. The harness applies each item + exactly as it applied a `MaintainerDecision` before.""" + decisions: list[MaintainerClusterItem] = Field( + ..., + description=( + "Exactly one entry per seed in the SEEDS list — do not omit a seed " + "and do not invent seeds. Each entry is a full decision (action + " + "its action-specific fields + rationale) plus the seed's entity_id." + ), + ) + + # Top-level coercion: accept JSON-string-of-dict (vLLM/Qwen quirk). + @model_validator(mode="before") + @classmethod + def _accept_json_string(cls, v): + return _maybe_parse_json_string(v) + + class WikiWriteResult(BaseModel): """The wiki writer's full output. `body` is the complete markdown page — a typed field of the schema, exactly like any other field (not loose diff --git a/braindb/agent/tools.py b/braindb/agent/tools.py index e8384fe..287603d 100644 --- a/braindb/agent/tools.py +++ b/braindb/agent/tools.py @@ -39,6 +39,7 @@ from braindb.agent.run_state import record_handoff, record_submit from braindb.agent.schemas import ( AgentAnswer, + MaintainerClusterDecision, MaintainerDecision, SubagentResult, WikiWriteResult, @@ -1180,8 +1181,8 @@ async def submit_answer(payload: AgentAnswer) -> str: @function_tool(name_override="final_answer", strict_mode=False) @_verbose("final_answer") -async def submit_maintainer(payload: MaintainerDecision) -> str: - """Submit the maintainer decision. Call this exactly once when you're done.""" +async def submit_maintainer(payload: MaintainerClusterDecision) -> str: + """Submit the maintainer decisions (one entry per seed). Call this exactly once when you're done.""" record_submit(payload) return "ok" diff --git a/braindb/routers/wiki.py b/braindb/routers/wiki.py index 0de6512..2cc2e77 100644 --- a/braindb/routers/wiki.py +++ b/braindb/routers/wiki.py @@ -13,7 +13,7 @@ from braindb.agent.agent import run_typed, get_maintainer_agent, get_writer_agent from braindb.agent.run_state import install_handoff_slot, release_handoff_slot -from braindb.agent.schemas import MaintainerDecision, WikiWriteResult +from braindb.agent.schemas import MaintainerClusterDecision, WikiWriteResult from braindb.config import settings from braindb.db import get_conn from braindb.services.activity_log import log_activity @@ -40,149 +40,197 @@ def wiki_cron(): @router.post("/maintain") async def wiki_maintain(): """ - Process EXACTLY ONE triage case (C1). Claims one pending triage job, - asks the existing agent to decide attach/create/consolidate/skip for that - single orphan, persists the resulting suggestion job, closes the triage. + Process ONE triage unit. Claims the highest-importance pending triage job + and — when WIKI_TRIAGE_CLUSTER is on and that seed is a non-hub keyword — + the other pending triage jobs whose entity shares a source fact with it + (`claim_related_triage`). The maintainer decides EVERY seed in ONE agent + call; each seed's own triage job is then closed with its own status + (skip→rejected, attach→done+suggestion, …), exactly as in the + one-at-a-time path. A singleton (or flag off) is just a 1-seed unit. """ - # 1. Claim one case + staleness guard, atomically (one transaction). + # 1. Claim the unit (primary + related) + per-seed staleness guard — one txn. with get_conn() as conn: - job = wiki_jobs.claim_one_triage(conn) - if not job: + primary = wiki_jobs.claim_one_triage(conn) + if not primary: return {"claimed": 0, "message": "no pending triage jobs"} - orphan_id = job["entity_ids"][0] - job_id = str(job["id"]) - batch_id = str(job["batch_id"]) if job["batch_id"] else None - orphan = wiki_jobs.fetch_entity_brief(conn, orphan_id) - - if not orphan: - wiki_jobs.finish_job(conn, job_id, "failed", "orphan entity not found") - return {"claimed": 1, "job_id": job_id, "result": "failed", - "reason": "orphan missing"} - - # Stale-skip: a prior writer run may have already absorbed/linked this - # entity (or it's already in an active suggestion). If so, close the - # triage with NO LLM call — the writer's broad research retired it. - if not wiki_jobs.is_orphan(conn, orphan_id, exclude_triage_job_id=job_id): - wiki_jobs.finish_job(conn, job_id, "done", - "already covered — absorbed by a wiki") - return {"claimed": 1, "job_id": job_id, "result": "skipped_stale"} - - # Catalog of existing wikis the model will reference BY NUMBER (never - # by uuid). This in-request list IS the numbering used to resolve the - # model's chosen number(s) back to ids below. + batch_id = str(primary["batch_id"]) if primary["batch_id"] else None + primary_id = primary["entity_ids"][0] + primary_brief = wiki_jobs.fetch_entity_brief(conn, primary_id) + + jobs = [primary] + if (wiki_jobs.WIKI_TRIAGE_CLUSTER and primary_brief + and primary_brief["entity_type"] == "keyword"): + jobs += wiki_jobs.claim_related_triage( + conn, primary_id, str(primary["id"]), + wiki_jobs.WIKI_TRIAGE_CLUSTER_MAX - 1) + + # Per seed: brief + stale-skip. A stale/missing seed closes its OWN job + # now (done/failed) with NO LLM call — same as the one-at-a-time path. + seeds: list[dict] = [] + for j in jobs: + jid, eid = str(j["id"]), j["entity_ids"][0] + brief = (primary_brief if eid == primary_id + else wiki_jobs.fetch_entity_brief(conn, eid)) + if not brief: + wiki_jobs.finish_job(conn, jid, "failed", "orphan entity not found") + continue + if not wiki_jobs.is_orphan(conn, eid, exclude_triage_job_id=jid): + wiki_jobs.finish_job(conn, jid, "done", + "already covered — absorbed by a wiki") + continue + seeds.append({"job_id": jid, "brief": brief}) + + if not seeds: + return {"claimed": len(jobs), "result": "skipped_stale"} + + # Catalog the model references BY NUMBER; this in-request list IS the + # numbering used to resolve its chosen number(s) back to ids below. cat = wiki_jobs.list_active_wikis(conn) - # 2. One agent call. The prompt directs it to RESEARCH the neighbourhood - # with its own tools (recall_memory / view_tree / delegate_to_subagent) - # before deciding — we give the seed, the LLM gathers the context. - # Generous turns so it can actually investigate / delegate. + # 2. ONE agent call for all live seeds. The prompt directs it to RESEARCH + # EACH seed's own subject (recall_memory / view_tree / delegate) before + # deciding — co-occurrence is not identity. Returns one decision per seed. catalog_txt = ( "\n".join(f"{i}. {w['canonical_name']}" for i, w in enumerate(cat, 1)) or "(no existing wikis yet — attach/consolidate are impossible; " "use create/skip/ambiguous)" ) prompt = _MAINTAINER_PROMPT.format( - entity_id=orphan_id, - entity_type=orphan["entity_type"], - keywords=orphan.get("keywords") or [], - summary=orphan.get("summary"), - content=(orphan.get("content") or "")[:4000], - wiki_catalog=catalog_txt, + seeds=_seeds_block(seeds), wiki_catalog=catalog_txt ) - # `run_typed` returns a SDK-validated MaintainerDecision, or raises if - # the model never submitted (e.g. max_turns hit) — that error path - # below treats it like any other agent failure (release + log + 5xx). + # `run_typed` returns an SDK-validated MaintainerClusterDecision, or raises + # if the model never submitted — handled below like any agent failure. try: - res: MaintainerDecision = await run_typed( - prompt, get_maintainer_agent(), MaintainerDecision, max_turns=30 + res: MaintainerClusterDecision = await run_typed( + prompt, get_maintainer_agent(), MaintainerClusterDecision, max_turns=30 ) except Exception as e: logger.exception("maintainer agent failed") with get_conn() as conn: - wiki_jobs.finish_job(conn, job_id, "failed", f"agent error: {e}"[:500]) - return {"claimed": 1, "job_id": job_id, "result": "failed", "reason": str(e)} + wiki_jobs.finish_jobs(conn, [s["job_id"] for s in seeds], "failed", + f"agent error: {e}"[:500]) + return {"claimed": len(jobs), "result": "failed", "reason": str(e)} - # Schema-validated; expose as a dict so the action handlers below are - # unchanged. - decision = res.model_dump() - action = decision.get("action") - rationale = decision.get("rationale") + by_id = {d.entity_id: d for d in res.decisions} - # 3. Persist the suggestion + close the triage, in one transaction. + # 3. Per seed: apply its decision + close its OWN job (same resolution as the + # one-at-a-time path). A seed the model omitted is released for a normal + # retry (never dropped). One transaction. + outcomes: dict[str, dict] = {} with get_conn() as conn: try: - if action in ("skip", "ambiguous"): - # 'ambiguous' = the data cannot disambiguate identity/scope; - # the LLM correctly refuses to mint a confident page. Treated - # as a deliberate skip (self-clears via run_cron). - wiki_jobs.finish_job(conn, job_id, "rejected", rationale) - outcome = {"action": action} - - elif action == "attach": - no = decision.get("target_wiki_no") - target = (cat[no - 1]["id"] - if isinstance(no, int) and 1 <= no <= len(cat) - else None) - if not target or not _is_wiki(conn, target): - wiki_jobs.finish_job( - conn, job_id, "failed", - f"attach: target_wiki_no {no!r} not a valid catalog number (1..{len(cat)})") - outcome = {"action": "attach", "error": "invalid target_wiki_no"} - else: - key = wiki_jobs.suggestion_dedupe_key("attach", target, [orphan_id], []) - sid = wiki_jobs.insert_suggestion( - conn, job_type="attach", target_wiki_id=target, - entity_ids=[orphan_id], dedupe_key=key, rationale=rationale, - proposed_name=None, batch_id=batch_id) - wiki_jobs.finish_job(conn, job_id, "done", rationale) - outcome = {"action": "attach", "suggestion_id": sid, "target_wiki_id": target} - - elif action == "create": - name = decision.get("proposed_name") - if not name: - wiki_jobs.finish_job(conn, job_id, "failed", "create missing proposed_name") - outcome = {"action": "create", "error": "missing proposed_name"} - else: - key = wiki_jobs.suggestion_dedupe_key("create", None, [orphan_id], []) - sid = wiki_jobs.insert_suggestion( - conn, job_type="create", target_wiki_id=None, - entity_ids=[orphan_id], dedupe_key=key, rationale=rationale, - proposed_name=name, batch_id=batch_id) - wiki_jobs.finish_job(conn, job_id, "done", rationale) - outcome = {"action": "create", "suggestion_id": sid, "proposed_name": name} - - elif action == "consolidate": - nos = decision.get("consolidate_nos") or [] - ids = [cat[n - 1]["id"] for n in nos - if isinstance(n, int) and 1 <= n <= len(cat)] - wiki_ids = list(dict.fromkeys(ids)) # dedupe, keep order - if len(wiki_ids) < 2: - wiki_jobs.finish_job( - conn, job_id, "failed", - f"consolidate: need >=2 valid catalog numbers, got {nos!r} (1..{len(cat)})") - outcome = {"action": "consolidate", "error": "need >=2 valid catalog numbers"} - else: - key = wiki_jobs.suggestion_dedupe_key("consolidate", None, [], wiki_ids) - sid = wiki_jobs.insert_suggestion( - conn, job_type="consolidate", target_wiki_id=None, - entity_ids=wiki_ids, dedupe_key=key, rationale=rationale, - proposed_name=None, batch_id=batch_id) - # The orphan itself is still unconnected; closing 'done' - # lets the next cron re-triage it after the merge. - wiki_jobs.finish_job(conn, job_id, "done", rationale) - outcome = {"action": "consolidate", "suggestion_id": sid, "wiki_ids": wiki_ids} - - else: - wiki_jobs.finish_job(conn, job_id, "failed", f"unknown action {action!r}") - outcome = {"action": action, "error": "unknown action"} - - log_activity(conn, "wiki_maintain", orphan["entity_type"], orphan_id, - details={"job_id": job_id, **outcome}) - except Exception as e: + for s in seeds: + eid = s["brief"]["id"] + d = by_id.get(eid) + if d is None: + wiki_jobs.release_or_fail_jobs( + conn, [s["job_id"]], "maintainer omitted this seed") + outcomes[eid] = {"action": "released_omitted"} + continue + outcomes[eid] = _apply_decision( + conn, decision=d.model_dump(), orphan=s["brief"], + job_id=s["job_id"], cat=cat, batch_id=batch_id) + except Exception: logger.exception("maintainer persistence failed") raise - return {"claimed": 1, "job_id": job_id, "result": outcome} + return {"claimed": len(jobs), "seeds": len(seeds), "result": outcomes} + + +def _seeds_block(seeds: list[dict]) -> str: + """Render the claimed seeds as a list the maintainer decides one-by-one. + Each carries its entity_id — the decision must echo it back.""" + out = [] + for s in seeds: + b = s["brief"] + out.append( + f"- entity_id: {b['id']}\n" + f" entity_type: {b['entity_type']}\n" + f" keywords: {b.get('keywords') or []}\n" + f" summary: {b.get('summary')}\n" + f" content: {(b.get('content') or '')[:4000]}" + ) + return "\n".join(out) + + +def _apply_decision(conn, *, decision: dict, orphan: dict, job_id: str, + cat: list[dict], batch_id: str | None) -> dict: + """Resolve ONE maintainer decision for one seed and close THAT seed's triage + job — the exact attach/create/consolidate/skip handling used in the + one-at-a-time path, applied per seed. Wikis are referenced by catalog NUMBER + (`cat` is the numbering).""" + orphan_id = orphan["id"] + action = decision.get("action") + rationale = decision.get("rationale") + + if action in ("skip", "ambiguous"): + # 'ambiguous' = the data cannot disambiguate identity/scope; the LLM + # correctly refuses to mint a confident page. Treated as a deliberate + # skip (self-clears via run_cron). + wiki_jobs.finish_job(conn, job_id, "rejected", rationale) + outcome = {"action": action} + + elif action == "attach": + no = decision.get("target_wiki_no") + target = (cat[no - 1]["id"] + if isinstance(no, int) and 1 <= no <= len(cat) + else None) + if not target or not _is_wiki(conn, target): + wiki_jobs.finish_job( + conn, job_id, "failed", + f"attach: target_wiki_no {no!r} not a valid catalog number (1..{len(cat)})") + outcome = {"action": "attach", "error": "invalid target_wiki_no"} + else: + key = wiki_jobs.suggestion_dedupe_key("attach", target, [orphan_id], []) + sid = wiki_jobs.insert_suggestion( + conn, job_type="attach", target_wiki_id=target, + entity_ids=[orphan_id], dedupe_key=key, rationale=rationale, + proposed_name=None, batch_id=batch_id) + wiki_jobs.finish_job(conn, job_id, "done", rationale) + outcome = {"action": "attach", "suggestion_id": sid, "target_wiki_id": target} + + elif action == "create": + name = decision.get("proposed_name") + if not name: + wiki_jobs.finish_job(conn, job_id, "failed", "create missing proposed_name") + outcome = {"action": "create", "error": "missing proposed_name"} + else: + key = wiki_jobs.suggestion_dedupe_key("create", None, [orphan_id], []) + sid = wiki_jobs.insert_suggestion( + conn, job_type="create", target_wiki_id=None, + entity_ids=[orphan_id], dedupe_key=key, rationale=rationale, + proposed_name=name, batch_id=batch_id) + wiki_jobs.finish_job(conn, job_id, "done", rationale) + outcome = {"action": "create", "suggestion_id": sid, "proposed_name": name} + + elif action == "consolidate": + nos = decision.get("consolidate_nos") or [] + ids = [cat[n - 1]["id"] for n in nos + if isinstance(n, int) and 1 <= n <= len(cat)] + wiki_ids = list(dict.fromkeys(ids)) # dedupe, keep order + if len(wiki_ids) < 2: + wiki_jobs.finish_job( + conn, job_id, "failed", + f"consolidate: need >=2 valid catalog numbers, got {nos!r} (1..{len(cat)})") + outcome = {"action": "consolidate", "error": "need >=2 valid catalog numbers"} + else: + key = wiki_jobs.suggestion_dedupe_key("consolidate", None, [], wiki_ids) + sid = wiki_jobs.insert_suggestion( + conn, job_type="consolidate", target_wiki_id=None, + entity_ids=wiki_ids, dedupe_key=key, rationale=rationale, + proposed_name=None, batch_id=batch_id) + # The orphan itself is still unconnected; closing 'done' lets the + # next cron re-triage it after the merge. + wiki_jobs.finish_job(conn, job_id, "done", rationale) + outcome = {"action": "consolidate", "suggestion_id": sid, "wiki_ids": wiki_ids} + + else: + wiki_jobs.finish_job(conn, job_id, "failed", f"unknown action {action!r}") + outcome = {"action": action, "error": "unknown action"} + + log_activity(conn, "wiki_maintain", orphan["entity_type"], orphan_id, + details={"job_id": job_id, **outcome}) + return outcome def _members_block(members: list[dict]) -> str: diff --git a/braindb/services/wiki_jobs.py b/braindb/services/wiki_jobs.py index e991b1b..5a71be4 100644 --- a/braindb/services/wiki_jobs.py +++ b/braindb/services/wiki_jobs.py @@ -52,6 +52,22 @@ # that wiki on each fire. ATTACH_COOLDOWN_SEC = int(os.getenv("WIKI_ATTACH_COOLDOWN_SECONDS", "300")) +# Batched triage (clustering). When the maintainer runs, let it CLAIM several +# related triage jobs at once — orphan keywords that share ONE source fact (plus +# that fact's own job when the fact is also an orphan) — and decide them in a +# single LLM call. This changes only how the maintainer GROUPS its claim; it +# never touches run_cron, the per-job lifecycle, or the decision logic (each job +# still closes with its own status). Same os.getenv pattern as the knobs above +# (config-import free). Default on; the A/B harness flips it per run. +WIKI_TRIAGE_CLUSTER = os.getenv("WIKI_TRIAGE_CLUSTER", "true").lower() == "true" +# Max seeds (the shared source fact + its orphan keywords) decided in one +# maintainer call — caps prompt size and keeps the model focused; extras are +# left for a later claim. +WIKI_TRIAGE_CLUSTER_MAX = int(os.getenv("WIKI_TRIAGE_CLUSTER_MAX", "8")) +# A keyword tagged on >= this many entities is a hub (it spans many distinct +# subjects, e.g. "CityFalcon") — triaged alone, never used to pull in a cluster. +WIKI_TRIAGE_HUB_DEGREE = int(os.getenv("WIKI_TRIAGE_HUB_DEGREE", "9")) + def _claimable(alias: str = "") -> str: """SQL predicate: a job is claimable if pending, OR assigned but its @@ -285,6 +301,78 @@ def claim_one_triage(conn) -> dict | None: return dict(row) if row else None +def claim_related_triage(conn, primary_entity_id: str, exclude_job_id: str, + limit: int) -> list[dict]: + """ + Claim up to `limit` other claimable triage jobs whose entity shares a source + fact with `primary_entity_id` — its non-hub sibling keywords (tagged on the + same fact) and the source fact(s)' own jobs — so a fact and its keywords are + decided in ONE maintainer call. `FOR UPDATE SKIP LOCKED`, importance first. + Each returned job is still single-entity and is closed individually by the + caller (per-job lifecycle unchanged). Returns [] if `limit` <= 0 or the + primary is itself a hub (a hub spans many subjects, so it is triaged alone). + """ + if limit <= 0: + return [] + pid = str(primary_entity_id) + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + # A hub primary would group too many unrelated subjects → triage it alone. + cur.execute( + "SELECT count(*) AS c FROM relations " + "WHERE relation_type='tagged_with' AND (from_entity_id=%s OR to_entity_id=%s)", + (pid, pid), + ) + if cur.fetchone()["c"] >= WIKI_TRIAGE_HUB_DEGREE: + return [] + cur.execute( + f""" + WITH src AS ( + -- the primary keyword's source facts/thoughts (tagged_with, undirected) + SELECT (CASE WHEN r.from_entity_id = %(pid)s THEN r.to_entity_id + ELSE r.from_entity_id END) AS fid + FROM relations r + JOIN entities f ON f.id = (CASE WHEN r.from_entity_id = %(pid)s + THEN r.to_entity_id ELSE r.from_entity_id END) + AND f.entity_type IN ('fact','thought') + WHERE r.relation_type = 'tagged_with' + AND (r.from_entity_id = %(pid)s OR r.to_entity_id = %(pid)s) + ), + cand AS ( + SELECT j.id + FROM wiki_job j + JOIN entities e ON e.id = j.entity_ids[1] + WHERE {_claimable("j")} AND j.job_type = 'triage' AND j.id <> %(xjob)s + AND ( + e.id IN (SELECT fid FROM src) -- a source fact's own job + OR ( + e.entity_type = 'keyword' -- a non-hub sibling keyword + AND (SELECT count(*) FROM relations rk + WHERE rk.relation_type = 'tagged_with' + AND (rk.from_entity_id = e.id OR rk.to_entity_id = e.id) + ) < %(hub)s + AND EXISTS ( + SELECT 1 FROM relations r2 + WHERE r2.relation_type = 'tagged_with' + AND ((r2.from_entity_id = e.id AND r2.to_entity_id IN (SELECT fid FROM src)) + OR (r2.to_entity_id = e.id AND r2.from_entity_id IN (SELECT fid FROM src))) + ) + ) + ) + ORDER BY e.importance DESC, j.created_at + FOR UPDATE OF j SKIP LOCKED + LIMIT %(limit)s + ) + UPDATE wiki_job + SET status = 'assigned', assigned_at = now(), attempts = attempts + 1 + WHERE id IN (SELECT id FROM cand) + RETURNING id, entity_ids::text[] AS entity_ids, batch_id + """, + {"pid": pid, "xjob": str(exclude_job_id), + "hub": WIKI_TRIAGE_HUB_DEGREE, "limit": limit}, + ) + return [dict(r) for r in cur.fetchall()] + + def finish_job(conn, job_id: str, status: str, last_error: str | None = None) -> None: """Transition a job to a terminal state (done / rejected / failed).""" with conn.cursor() as cur: From 2b85253f6629b9992714c9ac1e3118f405ef6142 Mon Sep 17 00:00:00 2001 From: dimknaf <136385722+dimknaf@users.noreply.github.com> Date: Tue, 16 Jun 2026 13:29:18 +0100 Subject: [PATCH 2/5] fix(integrations): make hermes ask-timeout env-configurable (default 600s) _ASK_TIMEOUT is now read from BRAINDB_ASK_TIMEOUT (default 600s, was a hardcoded 90s) so a slow /agent/query LLM loop doesn't time out and the ceiling is tunable per deployment. --- integrations/hermes/braindb/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/integrations/hermes/braindb/__init__.py b/integrations/hermes/braindb/__init__.py index 4a89cec..5fd2d3b 100644 --- a/integrations/hermes/braindb/__init__.py +++ b/integrations/hermes/braindb/__init__.py @@ -34,7 +34,7 @@ _DEFAULT_BASE_URL = "http://localhost:8000" _SKILL_NAME = "braindb-agent" -_ASK_TIMEOUT = 90 # /agent/query runs an LLM loop — allow headroom +_ASK_TIMEOUT = int(os.environ.get("BRAINDB_ASK_TIMEOUT", "600")) # /agent/query LLM loop; override via env _HTTP_TIMEOUT = 15 # skill fetch / file upload _HEALTH_TIMEOUT = 2 # readiness ping From 5961056b3aa759f007e8d9d4a4f20bee7a228c47 Mon Sep 17 00:00:00 2001 From: dimknaf <136385722+dimknaf@users.noreply.github.com> Date: Tue, 16 Jun 2026 13:44:30 +0100 Subject: [PATCH 3/5] test(wiki): track maintainer final_answer schema rebind to MaintainerClusterDecision submit_maintainer now carries MaintainerClusterDecision (a `decisions` list, one entry per seed) instead of the flat MaintainerDecision, so the tool-schema assertion is updated to expect `decisions` as the only required field. The base MaintainerDecision model is unchanged, so the model-direct tests stay. --- tests/test_final_answer_rename.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/tests/test_final_answer_rename.py b/tests/test_final_answer_rename.py index dc4c9ca..b881f7c 100644 --- a/tests/test_final_answer_rename.py +++ b/tests/test_final_answer_rename.py @@ -24,6 +24,7 @@ from braindb.agent import run_state from braindb.agent.schemas import ( AgentAnswer, + MaintainerClusterDecision, MaintainerDecision, SubagentResult, WikiWriteResult, @@ -441,7 +442,10 @@ def test_expected_shape_hint_covers_required_keys(model, required_keys, must_con "tool, model, pydantic_required", [ (submit_answer, AgentAnswer, ["answer"]), - (submit_maintainer, MaintainerDecision, ["action", "rationale"]), + # submit_maintainer now carries the clustered schema: a `decisions` + # list (one MaintainerClusterItem per seed); `decisions` is its only + # required field. The per-seed item still reuses MaintainerDecision. + (submit_maintainer, MaintainerClusterDecision, ["decisions"]), # body became optional with the section-edit work; only mode is # still required at the Pydantic level. (submit_wiki, WikiWriteResult, ["mode"]), From 9c814c1b5c5806c3e5d0a46ee6cd535e375a8f3c Mon Sep 17 00:00:00 2001 From: dimknaf <136385722+dimknaf@users.noreply.github.com> Date: Tue, 16 Jun 2026 23:10:26 +0100 Subject: [PATCH 4/5] feat(config): add vllm_workstation_gemma12b LLM profile A self-hosted vLLM profile for Gemma 4 12B (QAT w4a16) served at host.docker.internal:8012, mirroring the existing vllm_workstation_* profiles. Select it with LLM_PROFILE=vllm_workstation_gemma12b to run the agent on the local Gemma model. --- braindb/config.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/braindb/config.py b/braindb/config.py index 69874de..f572d0d 100644 --- a/braindb/config.py +++ b/braindb/config.py @@ -44,6 +44,14 @@ "api_key_env": "VLLM_API_KEY", "base_url": "http://host.docker.internal:8009/v1", }, + # Gemma 4 12B (QAT w4a16) on the workstation vLLM — a SMALLER, distinct model + # from the 31B above, on its own port. Reached over the laptop/Pi -> workstation + # tunnel to :8012. + "vllm_workstation_gemma12b": { + "model": "openai/google/gemma-4-12B-it-qat-w4a16-ct", + "api_key_env": "VLLM_API_KEY", + "base_url": "http://host.docker.internal:8012/v1", + }, } From 7a741f87585751a2e47450a88da56ad0099c9a36 Mon Sep 17 00:00:00 2001 From: dimknaf <136385722+dimknaf@users.noreply.github.com> Date: Tue, 16 Jun 2026 23:24:07 +0100 Subject: [PATCH 5/5] 0.8.0 --- CHANGELOG.md | 30 ++++++++++++++++++++++++++++++ braindb/main.py | 2 +- pyproject.toml | 2 +- 3 files changed, 32 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 42d2465..6bcc338 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,36 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.8.0] — 2026-06-16 + +Headline: the wiki maintainer now batches related orphan triage into ONE LLM call — +same-subject keywords that share a source fact are decided together, one decision per +seed — cutting maintainer LLM calls while keeping decisions and the job lifecycle +identical to before. Also in this release: a self-hosted Gemma 4 12B LLM profile and an +env-configurable Hermes ask-timeout. + +### Added + +- **Batched wiki triage (clustered maintainer).** When the maintainer runs and the + claimed seed is a non-hub keyword, it also claims the other pending triage jobs whose + entity shares a source fact (sibling keywords + that fact's own job) and decides them + all in ONE agent call — one decision per seed. `run_cron` and the per-job lifecycle are + unchanged (each job still closes with its own status); a singleton, or the flag off, is + a one-seed unit identical to before. Gated by `WIKI_TRIAGE_CLUSTER` (default on, with + `WIKI_TRIAGE_CLUSTER_MAX` / `WIKI_TRIAGE_HUB_DEGREE` to tune it). Validated A/B against + the one-at-a-time path on the same snapshot: identical decision split, near-identical + wikis, ~1.6 orphans decided per LLM call, and co-occurring seeds decided independently + (no conflation). +- **`vllm_workstation_gemma12b` LLM profile.** A self-hosted vLLM profile for Gemma 4 12B; + select it with `LLM_PROFILE=vllm_workstation_gemma12b` to run the internal agent on the + local Gemma model. + +### Changed + +- **Hermes `MemoryProvider` ask-timeout is env-configurable.** `BRAINDB_ASK_TIMEOUT` + (default 600s, previously a hardcoded 90s) so a slow `/agent/query` LLM loop does not + time out before the agent finishes. + ## [0.7.0] — 2026-06-14 Headline: BrainDB becomes a native long-term-memory backend for **Hermes Agent** diff --git a/braindb/main.py b/braindb/main.py index 154242a..bb092c4 100644 --- a/braindb/main.py +++ b/braindb/main.py @@ -11,7 +11,7 @@ app = FastAPI( title="BrainDB", description="Memory database and REST API for LLM agents", - version="0.7.0", + version="0.8.0", ) app.add_middleware( diff --git a/pyproject.toml b/pyproject.toml index 36d0ef2..7381158 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "braindb" -version = "0.7.0" +version = "0.8.0" description = "Persistent memory for LLM agents — thoughts, facts, sources, and behavioral rules with fuzzy + semantic search, graph traversal, and an internal agent." readme = "README.md" license = "Apache-2.0"