corpus_feeder.py
Rama main · réplica de lectura del repositorio original.
"""piSynapse — Corpus Feeder
Batch script that mines tool_audit_log for correction/confirmation signals
and feeds them into the intent-classification corpus as new examples.
Usage:
python corpus_feeder.py [--dry-run] [--db PATH]
State tracking: corpus_data/state.json stores last processed audit_id so
each run only processes new rows.
"""
import argparse
import hashlib
import json
import logging
import os
import re
import sys
from datetime import datetime, timezone
from pathlib import Path
import httpx
import numpy as np
from textnorm import normalize_signature
# ── paths ──────────────────────────────────────────────────────────────────────
ROOT = Path(__file__).resolve().parent
DATA_DIR = ROOT / "corpus_data"
STATE_FILE = DATA_DIR / "state.json"
ADDITIONS_FILE = DATA_DIR / "additions.jsonl"
EMBEDDINGS_FILE = DATA_DIR / "additions_embeddings.npy"
PENDING_REVIEW_FILE = DATA_DIR / "pending_review.json"
GENUINELY_AMBIGUOUS_FILE = DATA_DIR / "genuinely_ambiguous.json"
# ── logging ────────────────────────────────────────────────────────────────────
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
log = logging.getLogger("corpus_feeder")
# ── helpers ────────────────────────────────────────────────────────────────────
# Import config after ROOT is on sys.path so config.py can be found.
sys.path.insert(0, str(ROOT))
import config # noqa: E402
# Context-dependency gate lives in llm.intent — the SAME patterns the runtime
# classifier uses to defer anaphoric follow-ups, so the feeder (which must
# NEVER feed such fragments into the corpus) and the live router can never
# drift apart.
from llm.intent import _is_context_dependent # noqa: E402
def _db_path(args_db: str | None = None) -> str:
return args_db or os.getenv("DB_PATH", config.DB_PATH)
def _ensure_data_dir():
DATA_DIR.mkdir(parents=True, exist_ok=True)
def _load_state() -> dict:
if STATE_FILE.exists():
return json.loads(STATE_FILE.read_text())
return {"last_audit_id": 0, "last_run": None}
def _save_state(state: dict):
_ensure_data_dir()
STATE_FILE.write_text(json.dumps(state, indent=2))
def _load_jsonl(path: Path) -> list[dict]:
if not path.exists():
return []
rows = []
for line in path.read_text().splitlines():
line = line.strip()
if line:
rows.append(json.loads(line))
return rows
def _append_jsonl(path: Path, record: dict):
_ensure_data_dir()
with open(path, "a") as f:
f.write(json.dumps(record, ensure_ascii=False) + "\n")
def _deterministic_split(signature: str) -> str:
"""Assign a stable train/eval split (~10% eval) from the signature.
Hash-based and deterministic per signature: re-runs and dedup never
reshuffle rows, so the eval holdout stays a FIXED, honest test set — the
data-flywheel rule "never train on your eval split" only means something
if the split cannot drift. Eval rows still count for dedup/conflict
checks inside the feeder; only LIVE intent routing excludes them
(llm/intent._additional_corpus skips split=="eval").
"""
h = int(hashlib.sha256(signature.encode("utf-8")).hexdigest(), 16)
return "eval" if h % 10 == 0 else "train"
def _write_jsonl(path: Path, records: list[dict]):
"""Atomically replace the JSONL file with the given records (temp + rename)."""
_ensure_data_dir()
tmp = path.with_name(path.name + ".tmp")
with open(tmp, "w") as f:
for rec in records:
f.write(json.dumps(rec, ensure_ascii=False) + "\n")
os.replace(str(tmp), str(path))
def _remove_addition_by_audit_id(audit_id):
"""Drop the addition record for ``audit_id`` from ADDITIONS_FILE.
Used to roll back a jsonl append when its embedding could not be computed,
keeping additions.jsonl and additions_embeddings.npy index-aligned.
"""
rows = _load_jsonl(ADDITIONS_FILE)
kept = [r for r in rows if r.get("audit_id") != audit_id]
if len(kept) != len(rows):
_write_jsonl(ADDITIONS_FILE, kept)
def _load_pending_review() -> list[dict]:
return _load_jsonl(PENDING_REVIEW_FILE)
def _append_pending_review(record: dict):
_append_jsonl(PENDING_REVIEW_FILE, record)
def _load_genuinely_ambiguous() -> list[dict]:
return _load_jsonl(GENUINELY_AMBIGUOUS_FILE)
def _append_genuinely_ambiguous(record: dict):
_append_jsonl(GENUINELY_AMBIGUOUS_FILE, record)
# ── DB helpers ─────────────────────────────────────────────────────────────────
import sqlite3 # noqa: E402
def _connect(db_path: str) -> sqlite3.Connection:
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
return conn
def _fetch_new_audits(conn: sqlite3.Connection, last_id: int) -> list[dict]:
"""Fetch audit rows with confirmed_at or expected_group newer than last_id."""
rows = conn.execute(
"""
SELECT a.id, a.tool_name, a.conversation_id, a.expected_group,
a.expected_tool, a.confirmed_at, a.corrected_at
FROM tool_audit_log a
WHERE a.id > ?
AND (a.confirmed_at IS NOT NULL OR a.expected_group IS NOT NULL)
ORDER BY a.id ASC
""",
(last_id,),
).fetchall()
return [dict(r) for r in rows]
def _resolve_message_text(conn: sqlite3.Connection, conversation_id: int | None) -> str | None:
"""Resolve the USER message that triggered a given audit row.
`tool_audit_log.conversation_id` is the *assistant* message id (see
`link_audits_to_message`); the corpus must NOT be fed the assistant reply
(that is tool output, not a user intent). We therefore walk back to the
preceding `user` message in the same session. Returns None when unresolved.
"""
if conversation_id is None:
return None
row = conn.execute(
"SELECT session_id FROM conversations WHERE id = ?", (conversation_id,)
).fetchone()
if row is None or not row["session_id"]:
return None
session_id = row["session_id"]
user_row = conn.execute(
"""SELECT content FROM conversations
WHERE session_id = ? AND id < ? AND role = 'user'
ORDER BY id DESC LIMIT 1""",
(session_id, conversation_id),
).fetchone()
if user_row is None:
return None
return user_row["content"]
# Assistant outputs (list dumps, greetings, tool-summary prose) are NOT valid
# intent examples even after user-message resolution; they indicate the resolved
# text was the reply itself or a degenerate short message. Multi-line list
# markers, assistant-style greetings, and a heavy prose ratio mark such rows.
_ASST_GREETINGS = ("merhaba", "elbette", "işte", "aşağıda", "tamam,", "tabii,",
"hazırlıyorum", "listeliyorum", "bulabilirsin", "gönderdim",
"oldu", "kaydedildi", "eklendi", "silindi", "hatırlatıcı kuruldu",
"here you", "here's", "heres", "here are", "i've",
"i found", "i'll", "of course", "there you")
def _is_user_command_like(text: str) -> bool:
"""Heuristics to reject assistant-output / degenerate text as an intent example.
Returns True when the text plausibly is a user command suitable as a corpus
example. Conservative: when in doubt, returns False (the row is skipped to
genuinely_ambiguous instead of poisoning the corpus).
"""
t = text.strip().lower()
if not t:
return False
if "\n" in t and len(t.splitlines()) > 1:
return False # multi-line output/response
if len(t) > 200:
return False # long prose — almost certainly a reply
# A single word that is not a clear verb-like command is too ambiguous.
if len(t.split()) <= 1:
return False
# Assistant-style openers strongly suggest a reply, not a command.
for g in _ASST_GREETINGS:
if t.startswith(g):
return False
return True
# Lightweight language tag for additions.jsonl records. Not authoritative —
# it is a corpus metadata hint (TR/EN/DE/FR/ES) used for analytics and future
# language-aware gating. Heuristics: distinctive functional words first, then
# language-specific characters; English is the fallback (most common neutral).
_TR_WORDS = ("için", "ama", "değil", "var mı", "lütfen",
"yap", "gönder", "kaydet", "sil", "oluştur", "ekle", "liste", "hatırlat",
"hava", "notlar", "görev", "toplantı", "takvim")
_DE_WORDS = ("und", "oder", "der", "die", "das", "nicht", "ich", "bitte", "wetter",
"erinner", "aufgabe", "termin", "notiz", "schick", "lösch")
_FR_WORDS = ("je", "vous", "pour", "avec", "pas", "météo", "rappelle", "tâche",
"rendez", "note", "envoie", "supprime", "crée", "ajoute")
_ES_WORDS = ("yo", "para", "con", "el tiempo", "recuerda", "tarea", "reunión",
"nota", "envía", "borra", "crea", "añade", "por favor", "haz")
# Short Turkish words are matched on word boundaries (bare "not" would collide
# with English "note"/"notes"; "bir" ⊂ "birçok" would over-trigger).
_TR_BOUNDARY = ("bir", "ile", "şu", "bunu", "notlar")
def _detect_lang(text: str) -> str:
"""Best-effort ISO-639-1 language tag for a corpus example (tr/en/de/fr/es).
Falls back to 'en'. Runs on lowercased text; counts distinctive stops per
language and key-language characters to break near-ties (e.g. Turkish
accented characters are a strong TR signal).
"""
t = text.strip().lower()
# Words/phrases containing a space are matched as plain substrings (a
# phrase like "var mı" or "el tiempo" spans boundaries).
def _score_part(_words: tuple[str, ...]) -> int:
n = 0
for w in _words:
if " " in w:
n += int(w in t)
else:
n += int(bool(re.search(r"\b" + re.escape(w) + r"\b", t)))
return n
scores = {
"tr": _score_part(_TR_WORDS) + _score_part(_TR_BOUNDARY),
"de": _score_part(_DE_WORDS),
"fr": _score_part(_FR_WORDS),
"es": _score_part(_ES_WORDS),
}
# Turkish/English character-set tie-breakers.
if "ğ" in t or "ı" in t or "ş" in t or ("ç" in t and "chat" not in t):
scores["tr"] += 2
if "ö" in t:
scores["de"] += 1
scores["tr"] += 1
if re.match(r"\b(el|la|los) ", t):
scores["es"] += 1
if re.match(r"\b(le|la|les) ", t):
scores["fr"] += 1
rank = sorted(scores.items(), key=lambda kv: kv[1], reverse=True)
if rank[0][1] <= 0:
return "en"
return rank[0][0]
# ── contextual dependency ──────────────────────────────────────────────────────
# Some user utterances are ANAPHORIC follow-ups: they only make sense inside the
# conversation they reference ("devam edelim en son epostayı gönderiyordun").
# As standalone intent examples they say nothing about the requested tool and
# actively poison embedding routing (confirmed: they drag unrelated future
# queries into the tool group). Industry practice (curated few-shot sets +
# quality gates) rejects such examples instead of feeding them. The gate
# itself lives in llm.intent (`_is_context_dependent`) and is the same gate
# the live classifier uses to defer such messages to the session resolver.
# Only anaphoric follow-ups never enter the corpus; explicit domain keywords
# inside them are still resolvable at runtime.
# ── TOOL_TO_GROUP (mirrors tools/definitions.py) ──────────────────────────────
def _load_tool_to_group() -> dict[str, str]:
from tools.definitions import TOOL_TO_GROUP
return TOOL_TO_GROUP
def _load_corpus_groups() -> set[str]:
"""All group labels that exist in the hardcoded corpus."""
from llm.intent import _TOOL_EMBED_CORPUS
return {g for g, _ in _TOOL_EMBED_CORPUS if g is not None}
# ── conflict detection ────────────────────────────────────────────────────────
def _cosine(a: np.ndarray, b: np.ndarray) -> float:
dot = np.dot(a, b)
norm_a = np.linalg.norm(a)
norm_b = np.linalg.norm(b)
if norm_a == 0 or norm_b == 0:
return 0.0
return float(dot / (norm_a * norm_b))
async def _load_base_corpus_embeddings() -> tuple[list[str], np.ndarray | None]:
"""Load the base corpus texts and embed them. Returns (groups, matrix or None)."""
from embedding import embed_batch_async
from llm.intent import _TOOL_EMBED_CORPUS
groups = [g for g, _ in _TOOL_EMBED_CORPUS]
texts = [desc for _, desc in _TOOL_EMBED_CORPUS]
try:
vecs = await embed_batch_async(texts)
matrix = np.array([np.frombuffer(v, dtype="float32") for v in vecs])
return groups, matrix
except Exception as e:
log.warning(f"Could not embed base corpus: {e}")
return groups, None
def _rebuild_matrix(records: list[dict]) -> np.ndarray | None:
"""Re-embed every record text in jsonl order → index-aligned matrix.
additons_embeddings.npy stores vectors positionally; rebuilding the whole
matrix from the JSONL (the source of truth) guarantees rows and records
never slip out of alignment after a partial write or a crash.
"""
from embedding import embed as embed_one
if not records:
return None
vecs = []
for rec in records:
try:
v = np.frombuffer(embed_one(rec["text"]), dtype="float32").reshape(1, -1)
except Exception as exc: # noqa: BLE001
log.error("embed failed during rebuild for record=%r: %s", rec.get("audit_id"), exc)
return None
vecs.append(v)
return np.vstack(vecs)
def _load_addition_embeddings() -> tuple[list[dict], np.ndarray | None]:
"""Load persisted addition records and their embeddings, self-healing drift.
If the persisted vector count does not match the JSONL record count (a crash
or a failed embed may have slipped jsonl ahead of npy), the matrix is rebuilt
index-aligned from the JSONL and atomically persisted before being returned.
"""
additions = _load_jsonl(ADDITIONS_FILE)
if not additions:
if EMBEDDINGS_FILE.exists():
EMBEDDINGS_FILE.unlink(missing_ok=True)
return additions, None
if not EMBEDDINGS_FILE.exists():
matrix = _rebuild_matrix(additions)
if matrix is not None:
_save_addition_embeddings(matrix)
return additions, matrix
matrix = np.load(str(EMBEDDINGS_FILE))
if matrix.ndim == 1:
matrix = matrix.reshape(-1, 1)
if len(matrix) != len(additions):
log.warning(
"jsonl/npy misalignment: %d records vs %d vectors — rebuilding index-aligned",
len(additions), len(matrix),
)
matrix = _rebuild_matrix(additions)
if matrix is not None:
_save_addition_embeddings(matrix)
return additions, matrix if (matrix is not None and len(matrix) > 0) else None
def _save_addition_embeddings(matrix: np.ndarray):
_ensure_data_dir()
tmp = EMBEDDINGS_FILE.with_name(EMBEDDINGS_FILE.name + ".tmp.npy")
np.save(str(tmp), matrix)
os.replace(str(tmp), str(EMBEDDINGS_FILE))
def _check_conflict(
text: str,
proposed_group: str,
base_groups: list[str],
base_matrix: np.ndarray | None,
addition_records: list[dict],
addition_matrix: np.ndarray | None,
conflict_threshold: float | None = None,
) -> dict | None:
"""Check if text conflicts with existing corpus or additions.
A "conflict" is flagged when the text's nearest example of a DIFFERENT
group scores at or above the (configurable) cosine threshold — i.e. when
the existing corpus routing would disagree with the user's proposed group.
The threshold defaults to config.CONFLICT_COSINE (calibrated, see config.py)
so this flow fires on real data instead of sitting dead at an unreachable
0.85. Returns a conflict dict or None.
"""
from embedding import embed as embed_one
if conflict_threshold is None:
conflict_threshold = float(config.get("CONFLICT_COSINE", 0.50))
vec = np.frombuffer(embed_one(text), dtype="float32")
# Check against base corpus
if base_matrix is not None and len(base_matrix) > 0:
sims = np.array([_cosine(vec, row) for row in base_matrix])
best_idx = int(np.argmax(sims))
best_sim = float(sims[best_idx])
best_group = base_groups[best_idx]
if best_group != proposed_group and best_sim >= conflict_threshold:
return {
"text": text,
"proposed_group": proposed_group,
"conflicts_with_group": best_group,
"similarity": round(best_sim, 4),
"source": "base_corpus",
}
# Check against existing additions
if addition_matrix is not None and len(addition_matrix) > 0:
sims = np.array([_cosine(vec, row) for row in addition_matrix])
best_idx = int(np.argmax(sims))
best_sim = float(sims[best_idx])
best_group = addition_records[best_idx].get("group")
if best_group != proposed_group and best_sim >= conflict_threshold:
return {
"text": text,
"proposed_group": proposed_group,
"conflicts_with_group": best_group,
"similarity": round(best_sim, 4),
"source": "additions",
"conflict_audit_id": addition_records[best_idx].get("audit_id"),
}
return None
# ── quorum helpers (collective learning Phase 3) ─────────────────────────────
# Sync mirrors of db.get_pattern_support / db.user_reputation (different DB
# driver here — raw sqlite3 vs aiosqlite). Thresholds mirror db.py.
QUORUM_MIN_USERS = 2
REPUTATION_FLOOR = 0.5
def _vote_support(conn: sqlite3.Connection, signature: str, group: str) -> dict:
"""Distinct-user support + contradictor list for a (signature, group).
Only approved users count (admins always do) — mirrors
db.get_pattern_support (different driver here).
"""
try:
rows = conn.execute(
"SELECT v.proposed_group, v.user_id FROM feedback_votes v "
"JOIN users u ON u.id = v.user_id "
"WHERE v.signature = ? AND (u.is_approved = 1 OR u.is_admin = 1)",
(signature,),
).fetchall()
except Exception:
return {"supports": 0, "contradictors": []}
supports = len({u for g, u in rows if g == group})
contradictors = sorted({u for g, u in rows if g != group})
return {"supports": supports, "contradictors": contradictors}
def _pattern_decision(conn: sqlite3.Connection, signature: str) -> str | None:
"""Admin decision for a signature (approved/rejected/None)."""
try:
row = conn.execute(
"SELECT status FROM pattern_approvals WHERE signature = ?", (signature,)
).fetchone()
except Exception:
return None
return row[0] if row else None
def _voter_reputation(conn: sqlite3.Connection, user_id: str) -> float:
"""Agreement rate with decided patterns (sync mirror of db.user_reputation).
Only approved users' votes decide patterns.
"""
try:
rows = conn.execute(
"SELECT v.signature, v.proposed_group, v.user_id FROM feedback_votes v "
"JOIN users u ON u.id = v.user_id "
"WHERE v.signature != '' AND (u.is_approved = 1 OR u.is_admin = 1)"
).fetchall()
except Exception:
return 1.0
by_sig: dict[str, list] = {}
for s, g, u in rows:
by_sig.setdefault(s, []).append((g, u))
decided: dict[str, str] = {}
for s, votes in by_sig.items():
per_group: dict[str, set] = {}
for g, u in votes:
per_group.setdefault(g, set()).add(u)
total = len({u for _, u in votes})
if total < 2:
continue
top = max(len(us) for us in per_group.values())
winners = [g for g, us in per_group.items() if len(us) == top]
if len(winners) == 1 and top > total / 2:
decided[s] = winners[0]
mine = [(s, g) for s, g, u in rows if u == user_id and s in decided]
if not mine:
return 1.0
return sum(1 for s, g in mine if decided[s] == g) / len(mine)
# ── duplicate check ────────────────────────────────────────────────────────────
def _is_duplicate(
text: str,
base_groups: list[str],
base_matrix: np.ndarray | None,
addition_records: list[dict],
addition_matrix: np.ndarray | None,
dup_threshold: float = 0.98,
signature: str | None = None,
) -> bool:
"""Check if text already exists in corpus or additions (near-exact match).
A signature-equality fast path catches cross-user PII variants sharing
one routing shape — without an embedding call.
"""
if signature:
for rec in addition_records:
old = rec.get("signature")
if not old and rec.get("text"):
old = normalize_signature(rec["text"])
if old and old == signature:
return True
from embedding import embed as embed_one
vec = np.frombuffer(embed_one(text), dtype="float32")
if base_matrix is not None and len(base_matrix) > 0:
sims = np.array([_cosine(vec, row) for row in base_matrix])
if float(np.max(sims)) >= dup_threshold:
return True
if addition_matrix is not None and len(addition_matrix) > 0:
sims = np.array([_cosine(vec, row) for row in addition_matrix])
if float(np.max(sims)) >= dup_threshold:
return True
return False
# ── LLM auto-resolution ────────────────────────────────────────────────────────
async def _llm_resolve(text: str, group_a: str, group_b: str) -> str | None:
"""Ask the LLM which group fits the text better. Returns chosen group or None."""
prompt = (
f'Bu cümle daha çok hangi konuya ait?\n'
f' A) {group_a}\n'
f' B) {group_b}\n\n'
f'Cümle: "{text}"\n\n'
f'Sadece tek kelime yaz: {group_a} veya {group_b}'
)
try:
client = httpx.AsyncClient(timeout=30)
backend = getattr(config, "LLM_BACKEND", "litert").strip().lower()
model = getattr(config, "LLM_MODEL", "gemma4-e2b")
payload = {
"model": model,
"messages": [
{"role": "system", "content": "Sen bir sınıflandırma asistanısın. "
"Verilen cümleyi iki kategoriden birine sınıflandır. "
"SADECE kategori adını yaz, başka bir şey yazma."},
{"role": "user", "content": prompt},
],
"stream": False,
"max_tokens": 20,
"max_completion_tokens": 20,
}
if backend == "litert":
url = f"{getattr(config, 'LITERT_BASE_URL', 'http://localhost:9379')}/v1/chat/completions"
else:
payload["think"] = False
payload["options"] = {"temperature": 0, "num_predict": 20, "num_ctx": 512}
payload["keep_alive"] = getattr(config, "LLM_KEEP_ALIVE", "4h")
url = f"{getattr(config, 'OLLAMA_BASE_URL', 'http://localhost:11434')}/api/chat"
resp = await client.post(url, json=payload)
resp.raise_for_status()
rj = resp.json()
if backend == "litert":
answer = rj["choices"][0]["message"]["content"].strip().lower()
else:
answer = rj["message"]["content"].strip().lower()
await client.aclose()
# Normalize: the LLM might output the group name or a nearby string
if group_a.lower() in answer and group_b.lower() not in answer:
return group_a
if group_b.lower() in answer and group_a.lower() not in answer:
return group_b
return None # ambiguous / unrecognizable
except Exception as e:
log.warning(f"LLM resolution failed: {e}")
return None
# ── main logic ─────────────────────────────────────────────────────────────────
async def _process_audit_row(
row: dict,
tool_to_group: dict[str, str],
conn: sqlite3.Connection,
base_groups: list[str],
base_matrix: np.ndarray | None,
existing_additions: list[dict],
addition_matrix: np.ndarray | None,
dry_run: bool = False,
) -> dict:
"""Process a single audit row. Returns a summary dict."""
audit_id = row["id"]
tool_name = row["tool_name"]
conversation_id = row["conversation_id"]
confirmed = row["confirmed_at"] is not None
corrected = row["expected_group"] is not None
expected_group = row.get("expected_group")
message_text = _resolve_message_text(conn, conversation_id)
if not message_text:
return {"id": audit_id, "status": "skip_no_text"}
# Determine the proposed group
if corrected and expected_group:
proposed_group = expected_group
signal = "negative"
# Same-group "correction" is a no-op: the tool already belongs to the
# group the user "corrected" to (case: BUG-5 UI noise). Not a real
# signal — skip, don't feed it into the corpus.
current_group = tool_to_group.get(tool_name)
if current_group and current_group == expected_group:
return {"id": audit_id, "status": "skip_noop_negative",
"group": expected_group, "tool": tool_name}
elif confirmed:
proposed_group = tool_to_group.get(tool_name)
if not proposed_group:
return {"id": audit_id, "status": "skip_no_group_mapping"}
signal = "positive"
else:
return {"id": audit_id, "status": "skip_no_signal"}
# Normalize text: strip whitespace, lowercase for dedup
text = message_text.strip()
if not text:
return {"id": audit_id, "status": "skip_empty_text"}
# Shareable routing signature (collective learning Phase 2): contact PII
# becomes placeholders, so cross-user variants dedup to one shape and raw
# PII never reaches the global corpus entry.
sig = normalize_signature(text)
# Reject assistant-output / degenerate text: feeding the model's own reply
# or a context-only fragment as an intent example would poison routing.
if not _is_user_command_like(text):
record = {
"text": text,
"proposed_group": proposed_group,
"audit_id": audit_id,
"reason": "not_user_command",
"timestamp": datetime.now(timezone.utc).isoformat(),
}
if not dry_run:
_append_genuinely_ambiguous(record)
return {"id": audit_id, "status": "genuinely_ambiguous_not_command",
"group": proposed_group, "text": text}
# Context-dependent anaphoric follow-ups ("devam edelim en son epostayı
# gönderiyordun") are NOT standalone intents — they are meaningless outside
# the conversation and warp embedding routing (their "devam edelim" core
# drags unrelated queries into the group). Skip without feeding.
if _is_context_dependent(text):
record = {
"text": text,
"proposed_group": proposed_group,
"audit_id": audit_id,
"reason": "context_dependent",
"timestamp": datetime.now(timezone.utc).isoformat(),
}
if not dry_run:
_append_genuinely_ambiguous(record)
return {"id": audit_id, "status": "skip_context_dependent",
"group": proposed_group, "text": text}
# Contradiction freeze (collective learning Phase 3): reputable voters
# disagree on this shape and no side has quorum — never auto-add; the
# admin queue decides. Single-user flows (no contradictors) pass through.
# An admin decision overrides: approved skips the freeze, rejected skips
# the row entirely.
decision = _pattern_decision(conn, sig)
if decision == "rejected":
return {"id": audit_id, "status": "skip_admin_rejected",
"group": proposed_group, "text": text}
if decision != "approved":
sup = _vote_support(conn, sig, proposed_group)
reputable = [u for u in sup["contradictors"] if _voter_reputation(conn, u) >= REPUTATION_FLOOR]
if reputable and sup["supports"] < QUORUM_MIN_USERS:
record = {
"text": text,
"signature": sig,
"proposed_group": proposed_group,
"audit_id": audit_id,
"reason": "frozen_contradiction",
"supports": sup["supports"],
"contradictors": reputable,
"timestamp": datetime.now(timezone.utc).isoformat(),
}
if not dry_run:
_append_pending_review(record)
return {"id": audit_id, "status": "frozen_contradiction",
"group": proposed_group, "text": text}
# Check exact duplicate
if _is_duplicate(text, base_groups, base_matrix, existing_additions, addition_matrix,
signature=sig):
return {"id": audit_id, "status": "skip_duplicate", "group": proposed_group}
# Check conflict
conflict = _check_conflict(
text, proposed_group, base_groups, base_matrix,
existing_additions, addition_matrix,
)
if conflict:
# For corrections: the original tool_name maps to a group — that's the "wrong" group
original_group = tool_to_group.get(tool_name) if corrected else None
# LLM auto-resolution: ask which group is correct
if original_group and original_group != proposed_group:
llm_choice = await _llm_resolve(text, proposed_group, original_group)
elif conflict["conflicts_with_group"] != proposed_group:
llm_choice = await _llm_resolve(text, proposed_group, conflict["conflicts_with_group"])
else:
llm_choice = None
if llm_choice == proposed_group:
# LLM agrees with user — auto-add, resolve the conflict
record = {
"text": text,
"signature": sig,
"group": proposed_group,
"source": signal,
"lang": _detect_lang(text),
"audit_id": audit_id,
"split": _deterministic_split(sig),
"timestamp": datetime.now(timezone.utc).isoformat(),
"resolution": "llm_confirmed",
"original_conflict": conflict,
}
if not dry_run:
_append_jsonl(ADDITIONS_FILE, record)
return {"id": audit_id, "status": "added_llm_resolved", "group": proposed_group,
"text": text, "conflict": conflict, "llm_choice": llm_choice}
else:
# LLM disagrees or ambiguous → genuinely ambiguous
record = {
"text": text,
"proposed_group": proposed_group,
"conflicts_with": conflict,
"audit_id": audit_id,
"timestamp": datetime.now(timezone.utc).isoformat(),
"llm_choice": llm_choice,
}
if not dry_run:
_append_genuinely_ambiguous(record)
return {"id": audit_id, "status": "genuinely_ambiguous", "group": proposed_group,
"conflict": conflict, "llm_choice": llm_choice}
# No conflict → add directly
record = {
"text": text,
"signature": sig,
"group": proposed_group,
"source": signal,
"lang": _detect_lang(text),
"audit_id": audit_id,
"split": _deterministic_split(sig),
"timestamp": datetime.now(timezone.utc).isoformat(),
}
if not dry_run:
_append_jsonl(ADDITIONS_FILE, record)
return {"id": audit_id, "status": "added", "group": proposed_group, "text": text}
async def run(db_path: str | None = None, dry_run: bool = False) -> dict:
"""Main entry point. Returns summary dict."""
_ensure_data_dir()
state = _load_state()
last_id = state.get("last_audit_id", 0)
resolved_db = _db_path(db_path)
conn = _connect(resolved_db)
rows = _fetch_new_audits(conn, last_id)
if not rows:
log.info("No new audit rows to process.")
conn.close()
return {"processed": 0, "added": 0, "conflicts": 0, "ambiguous": 0}
log.info(f"Found {len(rows)} new audit rows (id > {last_id})")
tool_to_group = _load_tool_to_group()
base_groups, base_matrix = await _load_base_corpus_embeddings()
existing_additions, addition_matrix = _load_addition_embeddings()
summary = {"processed": 0, "added": 0, "skipped": 0, "conflicts": 0, "ambiguous": 0,
"details": []}
for row in rows:
result = await _process_audit_row(
row, tool_to_group, conn,
base_groups, base_matrix,
existing_additions, addition_matrix,
dry_run=dry_run,
)
summary["processed"] += 1
status = result["status"]
if status in ("added", "added_llm_resolved"):
text = result.get("text") or ""
group = result.get("group") or ""
if dry_run:
summary["added"] += 1
else:
# Guard the embed: _process_audit_row already appended the jsonl
# record; if its vector cannot be computed we roll the record back
# so jsonl and npy never diverge for this run.
try:
from embedding import embed as embed_one
vec = np.frombuffer(embed_one(text), dtype="float32")
except Exception as exc: # noqa: BLE001
log.error("embed failed for audit=%s [%s]: %s — rolling back jsonl",
result.get("id"), status, exc)
_remove_addition_by_audit_id(result.get("id"))
summary["skipped"] += 1
result["status"] = "skip_embed_error"
else:
summary["added"] += 1
existing_additions.append({"text": text, "group": group})
if addition_matrix is None:
addition_matrix = vec.reshape(1, -1)
else:
addition_matrix = np.vstack([addition_matrix, vec.reshape(1, -1)])
elif status == "genuinely_ambiguous":
summary["ambiguous"] += 1
elif status == "genuinely_ambiguous_not_command":
summary["ambiguous"] += 1
elif status == "skip_context_dependent":
summary["skipped"] += 1
elif status == "skip_duplicate":
summary["skipped"] += 1
summary["conflicts"] += 1
else:
summary["skipped"] += 1
summary["details"].append(result)
# Persist embeddings
if not dry_run and addition_matrix is not None:
_save_addition_embeddings(addition_matrix)
# Update state
max_id = max(r["id"] for r in rows)
if not dry_run:
_save_state({
"last_audit_id": max_id,
"last_run": datetime.now(timezone.utc).isoformat(),
})
conn.close()
log.info(
f"Done: {summary['processed']} processed, {summary['added']} added, "
f"{summary['conflicts']} conflicts, {summary['ambiguous']} genuinely ambiguous, "
f"{summary['skipped']} skipped"
)
return summary
# ── CLI ────────────────────────────────────────────────────────────────────────
def main():
import asyncio
parser = argparse.ArgumentParser(description="piSynapse corpus feeder")
parser.add_argument("--dry-run", action="store_true",
help="Process but don't write any files")
parser.add_argument("--db", type=str, default=None,
help="Path to assistant.db (default: config.DB_PATH)")
parser.add_argument("--reset", action="store_true",
help="Reset state to reprocess all rows")
args = parser.parse_args()
if args.reset:
_ensure_data_dir()
_save_state({"last_audit_id": 0, "last_run": None})
log.info("State reset. Next run will reprocess all rows.")
return
summary = asyncio.run(run(db_path=args.db, dry_run=args.dry_run))
# Pretty-print summary
print("\n" + "=" * 60)
print("CORPUS FEEDER SUMMARY")
print("=" * 60)
print(f" Processed: {summary['processed']}")
print(f" Added to corpus: {summary['added']}")
print(f" Conflicts resolved: {summary['conflicts']}")
print(f" Genuinely ambiguous:{summary['ambiguous']}")
print(f" Skipped: {summary['skipped']}")
if summary.get("details"):
print("\n Details:")
for d in summary["details"]:
status = d.get("status", "unknown")
group = d.get("group", "?")
aid = d.get("id", "?")
if status == "added":
print(f" #{aid} → +{group} (positive/negative)")
elif status == "added_llm_resolved":
print(f" #{aid} → +{group} (conflict resolved by LLM)")
elif status == "genuinely_ambiguous":
print(f" #{aid} → AMBIGUOUS (proposed {group}, LLM disagreed)")
elif status == "genuinely_ambiguous_not_command":
print(f" #{aid} → AMBIGUOUS (not a user command: {group!r})")
elif status.startswith("skip"):
print(f" #{aid} → {status}")
print("=" * 60)
if __name__ == "__main__":
main()