Lars
011cc9dd81
Identity-only senses no longer lock a new cue. Learning stages, journal-generate review, and additive docs follow the closed section 6. Co-authored-by: Cursor <cursoragent@cursor.com>
1105 lines
37 KiB
Python
1105 lines
37 KiB
Python
"""Transitional learning detect: senses grow from dialogue confirmations.
|
|
|
|
Not a global word list. Detect still does not auto-activate identities.
|
|
Student path: cues, kinship, homonym passage, then GLiNER. Teacher is optional.
|
|
Uncertain mentions go as a mini-passage to a chosen LLM profile, not to Detect.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import uuid
|
|
from typing import Any
|
|
|
|
from db import get_db, row_to_dict
|
|
from identity_store import (
|
|
DETERMINERS,
|
|
ENTITY_TYPES,
|
|
KINSHIP,
|
|
confirm_identity,
|
|
is_maskable_label,
|
|
normalize_label,
|
|
)
|
|
|
|
MODE_STUDENT = "student"
|
|
MODE_LEARNING_FULL = "learning_full"
|
|
MODE_LEARNING_REDUCED = "learning_reduced"
|
|
MODE_LEARNING_MANUAL = "learning_manual"
|
|
MODE_BYPASS = "bypass"
|
|
MODE_SEMANTIC = "semantic"
|
|
MODE_LEARNING = "learning"
|
|
CANONICAL_MODES = (
|
|
MODE_STUDENT,
|
|
MODE_LEARNING_FULL,
|
|
MODE_LEARNING_REDUCED,
|
|
MODE_LEARNING_MANUAL,
|
|
MODE_BYPASS,
|
|
)
|
|
MODE_ALIASES = {
|
|
MODE_SEMANTIC: MODE_BYPASS,
|
|
MODE_LEARNING: MODE_LEARNING_FULL,
|
|
}
|
|
MODES = CANONICAL_MODES + (MODE_SEMANTIC, MODE_LEARNING)
|
|
LEARNING_MODES = {MODE_LEARNING_FULL, MODE_LEARNING_REDUCED, MODE_LEARNING_MANUAL}
|
|
SETTING_KEY = "detect_operating_mode"
|
|
GLINER_ENABLED_KEY = "gliner_enabled"
|
|
GLINER_URL_KEY = "gliner_url"
|
|
PASSAGE_PROFILE_KEY = "passage_profile_id"
|
|
GLINER_DEFAULT_URL = "http://192.168.2.144:8090"
|
|
_LETTER = r"A-Za-zÄÖÜäöüß"
|
|
WINDOW = 80
|
|
|
|
_DECISION_IDENTITY = "identity"
|
|
_DECISION_NOT = "not_identity"
|
|
_DECISION_ASK = "ask"
|
|
_CUE_BARE = "_bare"
|
|
|
|
|
|
def canonicalize_detect_mode(mode: str | None) -> str | None:
|
|
chosen = (mode or "").strip().lower()
|
|
if not chosen:
|
|
return None
|
|
mapped = MODE_ALIASES.get(chosen, chosen)
|
|
return mapped if mapped in CANONICAL_MODES else None
|
|
|
|
|
|
def _default_detect_mode() -> str:
|
|
if os.environ.get("KANSHO_FAKE_PROVIDER") or os.environ.get("KANSHO_FAKE_DETECT"):
|
|
return MODE_BYPASS
|
|
return MODE_LEARNING_FULL
|
|
|
|
|
|
def get_detect_operating_mode() -> str:
|
|
with get_db() as conn:
|
|
row = row_to_dict(
|
|
conn.execute("SELECT value FROM app_settings WHERE key = ?", (SETTING_KEY,)).fetchone()
|
|
)
|
|
stored = canonicalize_detect_mode((row or {}).get("value") or "")
|
|
return stored or _default_detect_mode()
|
|
|
|
|
|
def set_detect_operating_mode(mode: str) -> str:
|
|
chosen = canonicalize_detect_mode(mode)
|
|
if not chosen:
|
|
raise ValueError("invalid_detect_operating_mode")
|
|
with get_db() as conn:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO app_settings (key, value, updated)
|
|
VALUES (?, ?, datetime('now'))
|
|
ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated = datetime('now')
|
|
""",
|
|
(SETTING_KEY, chosen),
|
|
)
|
|
return chosen
|
|
|
|
|
|
def is_bypass_mode(mode: str | None = None) -> bool:
|
|
return (mode or get_detect_operating_mode()) == MODE_BYPASS
|
|
|
|
|
|
def is_learning_mode(mode: str | None = None) -> bool:
|
|
return (mode or get_detect_operating_mode()) in LEARNING_MODES
|
|
|
|
|
|
def learning_stage(mode: str | None = None) -> int:
|
|
current = mode or get_detect_operating_mode()
|
|
if current == MODE_LEARNING_REDUCED:
|
|
return 2
|
|
return 1
|
|
|
|
|
|
def _app_setting(key: str, default: str = "") -> str:
|
|
with get_db() as conn:
|
|
row = row_to_dict(
|
|
conn.execute("SELECT value FROM app_settings WHERE key = ?", (key,)).fetchone()
|
|
)
|
|
if not row or row.get("value") is None:
|
|
return default
|
|
return str(row.get("value") or "").strip() or default
|
|
|
|
|
|
def _put_app_setting(key: str, value: str) -> None:
|
|
with get_db() as conn:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO app_settings (key, value, updated)
|
|
VALUES (?, ?, datetime('now'))
|
|
ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated = datetime('now')
|
|
""",
|
|
(key, value),
|
|
)
|
|
|
|
|
|
def get_local_ner_settings() -> dict:
|
|
from gliner_client import health, normalize_base_url
|
|
from providers import is_local_url
|
|
|
|
enabled_raw = _app_setting(GLINER_ENABLED_KEY, "0").lower()
|
|
enabled = enabled_raw in {"1", "true", "yes", "on"}
|
|
url = normalize_base_url(_app_setting(GLINER_URL_KEY, GLINER_DEFAULT_URL) or GLINER_DEFAULT_URL)
|
|
passage_profile_id = _app_setting(PASSAGE_PROFILE_KEY, "")
|
|
ping = health(url) if enabled and is_local_url(url) else {"ok": False, "status": "off"}
|
|
return {
|
|
"enabled": enabled,
|
|
"url": url,
|
|
"passage_profile_id": passage_profile_id,
|
|
"ready": bool(enabled and ping.get("ok")),
|
|
"health": ping,
|
|
}
|
|
|
|
|
|
def set_local_ner_settings(
|
|
*,
|
|
enabled: bool | None = None,
|
|
url: str | None = None,
|
|
passage_profile_id: str | None = None,
|
|
) -> dict:
|
|
from gliner_client import normalize_base_url
|
|
from providers import is_local_url
|
|
|
|
if enabled is not None:
|
|
_put_app_setting(GLINER_ENABLED_KEY, "1" if enabled else "0")
|
|
if url is not None:
|
|
chosen = normalize_base_url(url) if (url or "").strip() else GLINER_DEFAULT_URL
|
|
if not is_local_url(chosen):
|
|
raise ValueError("gliner_url_not_local")
|
|
_put_app_setting(GLINER_URL_KEY, chosen)
|
|
if passage_profile_id is not None:
|
|
token = (passage_profile_id or "").strip()
|
|
if token:
|
|
from provider_settings import get_profile
|
|
|
|
if not get_profile(token):
|
|
raise ValueError("passage_profile_missing")
|
|
_put_app_setting(PASSAGE_PROFILE_KEY, token)
|
|
return get_local_ner_settings()
|
|
|
|
|
|
def get_sense(profile_id: str, label: str) -> dict:
|
|
key = normalize_label(label) or (label or "").strip()
|
|
if not key:
|
|
return {
|
|
"normalized_label": "",
|
|
"identity_hits": 0,
|
|
"non_identity_hits": 0,
|
|
"ambiguous": False,
|
|
}
|
|
with get_db() as conn:
|
|
row = row_to_dict(
|
|
conn.execute(
|
|
"""
|
|
SELECT normalized_label, identity_hits, non_identity_hits
|
|
FROM label_senses
|
|
WHERE profile_id = ? AND lower(normalized_label) = lower(?)
|
|
""",
|
|
(profile_id, key),
|
|
).fetchone()
|
|
)
|
|
if not row:
|
|
return {
|
|
"normalized_label": key,
|
|
"identity_hits": 0,
|
|
"non_identity_hits": 0,
|
|
"ambiguous": False,
|
|
}
|
|
identity = int(row.get("identity_hits") or 0)
|
|
other = int(row.get("non_identity_hits") or 0)
|
|
return {
|
|
"normalized_label": row.get("normalized_label") or key,
|
|
"identity_hits": identity,
|
|
"non_identity_hits": other,
|
|
"ambiguous": identity > 0 and other > 0,
|
|
}
|
|
|
|
|
|
def record_sense(profile_id: str, label: str, *, identity: bool) -> dict:
|
|
key = normalize_label(label) or (label or "").strip()
|
|
if not key:
|
|
raise ValueError("empty_label")
|
|
with get_db() as conn:
|
|
existing = row_to_dict(
|
|
conn.execute(
|
|
"""
|
|
SELECT normalized_label, identity_hits, non_identity_hits FROM label_senses
|
|
WHERE profile_id = ? AND lower(normalized_label) = lower(?)
|
|
""",
|
|
(profile_id, key),
|
|
).fetchone()
|
|
)
|
|
if existing:
|
|
key = existing.get("normalized_label") or key
|
|
identity_hits = int((existing or {}).get("identity_hits") or 0)
|
|
non_identity_hits = int((existing or {}).get("non_identity_hits") or 0)
|
|
if identity:
|
|
identity_hits += 1
|
|
else:
|
|
non_identity_hits += 1
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO label_senses (
|
|
profile_id, normalized_label, identity_hits, non_identity_hits, updated
|
|
)
|
|
VALUES (?, ?, ?, ?, datetime('now'))
|
|
ON CONFLICT(profile_id, normalized_label) DO UPDATE SET
|
|
identity_hits = excluded.identity_hits,
|
|
non_identity_hits = excluded.non_identity_hits,
|
|
updated = datetime('now')
|
|
""",
|
|
(profile_id, key, identity_hits, non_identity_hits),
|
|
)
|
|
return get_sense(profile_id, key)
|
|
|
|
|
|
def _in_user_text(label: str, user_body: str) -> bool:
|
|
if not label or not user_body:
|
|
return False
|
|
return bool(
|
|
re.search(
|
|
rf"(?<![{_LETTER}]){re.escape(label)}(?![{_LETTER}])",
|
|
user_body,
|
|
re.IGNORECASE,
|
|
)
|
|
)
|
|
|
|
|
|
def _resolve_user_span(
|
|
user_body: str,
|
|
start: int | None,
|
|
end: int | None,
|
|
label: str,
|
|
rendered: str = "",
|
|
) -> tuple[int, int] | None:
|
|
text = user_body or ""
|
|
needle = (label or "").casefold()
|
|
if start is not None and end is not None:
|
|
left, right = int(start), int(end)
|
|
if 0 <= left < right <= len(text) and text[left:right].casefold() == needle:
|
|
return left, right
|
|
if rendered and text:
|
|
origin = rendered.rfind(text)
|
|
if origin >= 0:
|
|
user_left, user_right = left - origin, right - origin
|
|
if 0 <= user_left < user_right <= len(text) and text[user_left:user_right].casefold() == needle:
|
|
return user_left, user_right
|
|
if not needle:
|
|
return None
|
|
match = re.search(rf"(?<![{_LETTER}]){re.escape(label)}(?![{_LETTER}])", text, re.IGNORECASE)
|
|
if not match:
|
|
return None
|
|
return match.start(), match.end()
|
|
|
|
|
|
def excerpt_view(
|
|
user_body: str,
|
|
start: int | None,
|
|
end: int | None,
|
|
label: str,
|
|
rendered: str = "",
|
|
) -> dict:
|
|
text = user_body or ""
|
|
span = _resolve_user_span(text, start, end, label, rendered)
|
|
if not span:
|
|
snippet = text[:WINDOW]
|
|
return {"excerpt": snippet, "highlight_start": None, "highlight_end": None}
|
|
left_i, right_i = span
|
|
clip_left = max(0, left_i - WINDOW)
|
|
clip_right = min(len(text), right_i + WINDOW)
|
|
prefix = "…" if clip_left > 0 else ""
|
|
suffix = "…" if clip_right < len(text) else ""
|
|
snippet = text[clip_left:clip_right]
|
|
return {
|
|
"excerpt": f"{prefix}{snippet}{suffix}",
|
|
"highlight_start": left_i - clip_left + len(prefix),
|
|
"highlight_end": right_i - clip_left + len(prefix),
|
|
}
|
|
|
|
|
|
def _excerpt(user_body: str, start: int | None, end: int | None, label: str) -> str:
|
|
return excerpt_view(user_body, start, end, label)["excerpt"]
|
|
|
|
|
|
def needs_review(profile_id: str, mapping: dict, user_body: str, rendered: str = "") -> bool:
|
|
label = mapping.get("local_label") or ""
|
|
if mapping.get("source") == "confirmed_registry":
|
|
return (
|
|
_in_user_text(label, user_body)
|
|
and decision_for_mention(
|
|
profile_id, label, user_body, mapping.get("start"), mapping.get("end"), rendered
|
|
)
|
|
== _DECISION_ASK
|
|
)
|
|
if mapping.get("source") != "request_local":
|
|
return False
|
|
if not is_maskable_label(label) or not _in_user_text(label, user_body):
|
|
return False
|
|
return (
|
|
decision_for_mention(
|
|
profile_id, label, user_body, mapping.get("start"), mapping.get("end"), rendered
|
|
)
|
|
== _DECISION_ASK
|
|
)
|
|
|
|
|
|
def _word_matches(text: str):
|
|
return list(re.finditer(rf"[{_LETTER}]+", text or ""))
|
|
|
|
|
|
def _prev_word(text: str, index: int) -> str:
|
|
words = re.findall(rf"[{_LETTER}]+", text[:index])
|
|
return (words[-1].lower() if words else "")
|
|
|
|
|
|
def _token_index_at(tokens, start: int) -> int | None:
|
|
for index, match in enumerate(tokens):
|
|
if match.start() <= start < match.end() or match.start() == start:
|
|
return index
|
|
return None
|
|
|
|
|
|
def attachment_kinship(text: str, start: int, end: int) -> str:
|
|
"""Kinship in the local NP: left of the mention or right apposition, not the whole sentence."""
|
|
tokens = _word_matches(text)
|
|
index = _token_index_at(tokens, start)
|
|
if index is None:
|
|
return ""
|
|
left = index - 1
|
|
while left >= 0 and tokens[left].group(0).lower() in DETERMINERS:
|
|
left -= 1
|
|
if left >= 0 and tokens[left].group(0).lower() in KINSHIP:
|
|
return tokens[left].group(0).lower()
|
|
right = index + 1
|
|
while right < len(tokens) and tokens[right].group(0).lower() in DETERMINERS:
|
|
right += 1
|
|
if right < len(tokens) and tokens[right].group(0).lower() in KINSHIP:
|
|
return tokens[right].group(0).lower()
|
|
return ""
|
|
|
|
|
|
def cue_for_mention(
|
|
user_body: str,
|
|
start: int | None,
|
|
end: int | None,
|
|
label: str,
|
|
rendered: str = "",
|
|
) -> str:
|
|
span = _resolve_user_span(user_body, start, end, label, rendered)
|
|
if not span:
|
|
return _CUE_BARE
|
|
attached = attachment_kinship(user_body, span[0], span[1])
|
|
if attached:
|
|
return attached
|
|
return _prev_word(user_body, span[0]) or _CUE_BARE
|
|
|
|
|
|
def _mention_fields(
|
|
user_body: str,
|
|
start: int | None,
|
|
end: int | None,
|
|
label: str,
|
|
rendered: str = "",
|
|
) -> dict:
|
|
span = _resolve_user_span(user_body, start, end, label, rendered)
|
|
return {
|
|
**excerpt_view(user_body, start, end, label, rendered),
|
|
"user_start": span[0] if span else None,
|
|
"user_end": span[1] if span else None,
|
|
"cue": cue_for_mention(user_body, start, end, label, rendered),
|
|
}
|
|
|
|
|
|
def get_cue_decision(profile_id: str, label: str, cue: str) -> str | None:
|
|
key = normalize_label(label) or (label or "").strip()
|
|
token = (cue or _CUE_BARE).strip().casefold() or _CUE_BARE
|
|
if not key:
|
|
return None
|
|
with get_db() as conn:
|
|
row = row_to_dict(
|
|
conn.execute(
|
|
"""
|
|
SELECT decision FROM label_sense_cues
|
|
WHERE profile_id = ? AND lower(normalized_label) = lower(?) AND cue = ?
|
|
""",
|
|
(profile_id, key, token),
|
|
).fetchone()
|
|
)
|
|
value = ((row or {}).get("decision") or "").strip()
|
|
return value if value in {_DECISION_IDENTITY, _DECISION_NOT} else None
|
|
|
|
|
|
def record_cue(profile_id: str, label: str, cue: str, decision: str, excerpt: str = "") -> None:
|
|
key = normalize_label(label) or (label or "").strip()
|
|
token = (cue or _CUE_BARE).strip().casefold() or _CUE_BARE
|
|
snippet = (excerpt or "").strip()
|
|
if not key or decision not in {_DECISION_IDENTITY, _DECISION_NOT}:
|
|
return
|
|
with get_db() as conn:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO label_sense_cues (
|
|
profile_id, normalized_label, cue, decision, hits, excerpt, updated
|
|
)
|
|
VALUES (?, ?, ?, ?, 1, ?, datetime('now'))
|
|
ON CONFLICT(profile_id, normalized_label, cue) DO UPDATE SET
|
|
decision = excluded.decision,
|
|
hits = label_sense_cues.hits + 1,
|
|
excerpt = CASE
|
|
WHEN excluded.excerpt <> '' THEN excluded.excerpt
|
|
ELSE label_sense_cues.excerpt
|
|
END,
|
|
updated = datetime('now')
|
|
""",
|
|
(profile_id, key, token, decision, snippet),
|
|
)
|
|
|
|
|
|
def list_cue_examples(profile_id: str, label: str, limit: int = 6) -> list[dict]:
|
|
key = normalize_label(label) or (label or "").strip()
|
|
if not key:
|
|
return []
|
|
with get_db() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT cue, decision, excerpt, hits
|
|
FROM label_sense_cues
|
|
WHERE profile_id = ? AND lower(normalized_label) = lower(?)
|
|
ORDER BY hits DESC, updated DESC
|
|
""",
|
|
(profile_id, key),
|
|
).fetchall()
|
|
items = []
|
|
for raw in rows:
|
|
row = row_to_dict(raw) or {}
|
|
decision = (row.get("decision") or "").strip()
|
|
if decision not in {_DECISION_IDENTITY, _DECISION_NOT}:
|
|
continue
|
|
items.append(
|
|
{
|
|
"cue": row.get("cue") or _CUE_BARE,
|
|
"decision": decision,
|
|
"excerpt": (row.get("excerpt") or "").strip(),
|
|
}
|
|
)
|
|
if len(items) >= limit:
|
|
break
|
|
return items
|
|
|
|
|
|
def decision_for_mention(
|
|
profile_id: str,
|
|
label: str,
|
|
user_body: str,
|
|
start: int | None,
|
|
end: int | None,
|
|
rendered: str = "",
|
|
) -> str:
|
|
"""identity / not_identity from a stored cue, otherwise ask.
|
|
|
|
A sense with only identity hits must not lock a new cue (Sushi day 21).
|
|
"""
|
|
if not label:
|
|
return _DECISION_ASK
|
|
cue = cue_for_mention(user_body, start, end, label, rendered)
|
|
learned = get_cue_decision(profile_id, label, cue)
|
|
if learned:
|
|
return learned
|
|
return _DECISION_ASK
|
|
|
|
|
|
def apply_learned_decisions(
|
|
profile_id: str,
|
|
candidates: list[dict],
|
|
mappings: list[dict],
|
|
user_body: str,
|
|
rendered: str = "",
|
|
) -> tuple[list[dict], list[dict]]:
|
|
remaining = []
|
|
for item in candidates:
|
|
verdict = decision_for_mention(
|
|
profile_id,
|
|
item.get("text") or "",
|
|
user_body,
|
|
item.get("user_start", item.get("start")),
|
|
item.get("user_end", item.get("end")),
|
|
rendered,
|
|
)
|
|
if verdict == _DECISION_ASK:
|
|
remaining.append(item)
|
|
kept = []
|
|
for mapping in mappings:
|
|
label = mapping.get("local_label") or ""
|
|
verdict = decision_for_mention(
|
|
profile_id,
|
|
label,
|
|
user_body,
|
|
mapping.get("start"),
|
|
mapping.get("end"),
|
|
rendered,
|
|
)
|
|
if verdict == _DECISION_NOT:
|
|
continue
|
|
kept.append(mapping)
|
|
if verdict == _DECISION_IDENTITY:
|
|
kind = mapping.get("entity_type") if mapping.get("entity_type") in ENTITY_TYPES else "PERSON"
|
|
try:
|
|
confirm_identity(profile_id, label, entity_type=kind)
|
|
except ValueError:
|
|
continue
|
|
return remaining, kept
|
|
|
|
|
|
def kinship_governed_labels(user_body: str) -> set[str]:
|
|
"""Labels attached to Frau/Sohn/… in the current user line. Not a food word list."""
|
|
found: set[str] = set()
|
|
for match in _word_matches(user_body):
|
|
label = match.group(0)
|
|
if not is_maskable_label(label):
|
|
continue
|
|
if attachment_kinship(user_body, match.start(), match.end()):
|
|
found.add(label.casefold())
|
|
return found
|
|
|
|
|
|
def _align_user_span(rendered: str, user_body: str, start: int, end: int) -> tuple[int, int]:
|
|
if not user_body:
|
|
return start, end
|
|
pos = (rendered or "").rfind(user_body)
|
|
if pos < 0:
|
|
return start, end
|
|
return pos + start, pos + end
|
|
|
|
|
|
def _token_for_label(mappings: list[dict], label: str, entity_type: str = "PERSON") -> str:
|
|
needle = label.casefold()
|
|
for mapping in mappings:
|
|
if (mapping.get("local_label") or "").casefold() == needle and mapping.get("token"):
|
|
return mapping["token"]
|
|
return f"{entity_type}:L{uuid.uuid4().hex[:6].upper()}"
|
|
|
|
|
|
def supplement_kinship_candidates(
|
|
profile_id: str,
|
|
candidates: list[dict],
|
|
mappings: list[dict],
|
|
user_body: str,
|
|
rendered: str = "",
|
|
) -> tuple[list[dict], list[dict]]:
|
|
"""If Detect misses Frau X / Sohn X, still offer those user-line mentions for review."""
|
|
wanted = kinship_governed_labels(user_body)
|
|
if not wanted:
|
|
return candidates, mappings
|
|
occupied: set[tuple[int, int, str]] = set()
|
|
for item in candidates:
|
|
span = _resolve_user_span(
|
|
user_body,
|
|
item.get("start"),
|
|
item.get("end"),
|
|
item.get("text") or "",
|
|
rendered,
|
|
)
|
|
if span:
|
|
occupied.add((span[0], span[1], (item.get("text") or "").casefold()))
|
|
extra_mappings = list(mappings)
|
|
extra_candidates = list(candidates)
|
|
for match in _word_matches(user_body):
|
|
label = match.group(0)
|
|
if label.casefold() not in wanted:
|
|
continue
|
|
user_key = (match.start(), match.end(), label.casefold())
|
|
if user_key in occupied:
|
|
continue
|
|
if (
|
|
decision_for_mention(profile_id, label, user_body, match.start(), match.end(), rendered)
|
|
!= _DECISION_ASK
|
|
):
|
|
continue
|
|
occupied.add(user_key)
|
|
start, end = _align_user_span(rendered, user_body, match.start(), match.end())
|
|
identity_shaped = bool(attachment_kinship(user_body, match.start(), match.end()))
|
|
extra_candidates.append(
|
|
{
|
|
"id": str(uuid.uuid4()),
|
|
"text": label,
|
|
"entity_type": "PERSON",
|
|
"start": start,
|
|
"end": end,
|
|
**_mention_fields(user_body, match.start(), match.end(), label, rendered),
|
|
"ambiguous": get_sense(profile_id, label)["ambiguous"],
|
|
"suggested": _DECISION_IDENTITY if identity_shaped else _DECISION_NOT,
|
|
}
|
|
)
|
|
extra_mappings.append(
|
|
{
|
|
"token": _token_for_label(extra_mappings, label),
|
|
"local_label": label,
|
|
"canonical_label": label,
|
|
"demask_label": label,
|
|
"entity_type": "PERSON",
|
|
"source": "request_local",
|
|
"start": start,
|
|
"end": end,
|
|
"aliases": [],
|
|
"labels": [label],
|
|
}
|
|
)
|
|
return extra_candidates, extra_mappings
|
|
|
|
|
|
def merge_gliner_candidates(
|
|
profile_id: str,
|
|
candidates: list[dict],
|
|
mappings: list[dict],
|
|
user_body: str,
|
|
rendered: str,
|
|
spans: list[dict] | None,
|
|
) -> tuple[list[dict], list[dict]]:
|
|
"""Add GLiNER spans from the user line. Learned cues still win afterwards."""
|
|
from gliner_client import gliner_identity_type, is_gliner_non_identity
|
|
|
|
extra_candidates = list(candidates)
|
|
extra_mappings = list(mappings)
|
|
if not spans:
|
|
return extra_candidates, extra_mappings
|
|
occupied: set[tuple[int, int, str]] = set()
|
|
for item in extra_candidates:
|
|
span = _resolve_user_span(
|
|
user_body,
|
|
item.get("user_start", item.get("start")),
|
|
item.get("user_end", item.get("end")),
|
|
item.get("text") or "",
|
|
rendered,
|
|
)
|
|
if span:
|
|
occupied.add((span[0], span[1], (item.get("text") or "").casefold()))
|
|
food_keys: set[tuple[int, int, str]] = set()
|
|
for raw in spans:
|
|
label = raw.get("text") or ""
|
|
kind = raw.get("label") or ""
|
|
try:
|
|
start, end = int(raw["start"]), int(raw["end"])
|
|
except (KeyError, TypeError, ValueError):
|
|
continue
|
|
if start < 0 or end > len(user_body) or end <= start:
|
|
continue
|
|
if user_body[start:end].casefold() != label.casefold():
|
|
resolved = _resolve_user_span(user_body, start, end, label, rendered)
|
|
if not resolved:
|
|
continue
|
|
start, end = resolved
|
|
label = user_body[start:end]
|
|
if not is_maskable_label(label):
|
|
continue
|
|
key = (start, end, label.casefold())
|
|
if is_gliner_non_identity(kind):
|
|
food_keys.add(key)
|
|
continue
|
|
entity_type = gliner_identity_type(kind)
|
|
if not entity_type or key in occupied:
|
|
continue
|
|
if decision_for_mention(profile_id, label, user_body, start, end, rendered) != _DECISION_ASK:
|
|
continue
|
|
occupied.add(key)
|
|
aligned = _align_user_span(rendered, user_body, start, end)
|
|
extra_candidates.append(
|
|
{
|
|
"id": str(uuid.uuid4()),
|
|
"text": label,
|
|
"entity_type": entity_type,
|
|
"start": aligned[0],
|
|
"end": aligned[1],
|
|
**_mention_fields(user_body, start, end, label, rendered),
|
|
"ambiguous": get_sense(profile_id, label)["ambiguous"],
|
|
"suggested": _DECISION_IDENTITY,
|
|
}
|
|
)
|
|
extra_mappings.append(
|
|
{
|
|
"token": _token_for_label(extra_mappings, label, entity_type),
|
|
"local_label": label,
|
|
"canonical_label": label,
|
|
"demask_label": label,
|
|
"entity_type": entity_type,
|
|
"source": "request_local",
|
|
"start": aligned[0],
|
|
"end": aligned[1],
|
|
"aliases": [],
|
|
"labels": [label],
|
|
}
|
|
)
|
|
for item in extra_candidates:
|
|
span = _resolve_user_span(
|
|
user_body,
|
|
item.get("user_start", item.get("start")),
|
|
item.get("user_end", item.get("end")),
|
|
item.get("text") or "",
|
|
rendered,
|
|
)
|
|
if not span:
|
|
continue
|
|
key = (span[0], span[1], (item.get("text") or "").casefold())
|
|
if key in food_keys and not attachment_kinship(user_body, span[0], span[1]):
|
|
item["suggested"] = _DECISION_NOT
|
|
for start, end, folded in food_keys:
|
|
if (start, end, folded) in occupied:
|
|
continue
|
|
label = user_body[start:end]
|
|
if decision_for_mention(profile_id, label, user_body, start, end, rendered) != _DECISION_ASK:
|
|
continue
|
|
occupied.add((start, end, folded))
|
|
aligned = _align_user_span(rendered, user_body, start, end)
|
|
extra_candidates.append(
|
|
{
|
|
"id": str(uuid.uuid4()),
|
|
"text": label,
|
|
"entity_type": "PERSON",
|
|
"start": aligned[0],
|
|
"end": aligned[1],
|
|
**_mention_fields(user_body, start, end, label, rendered),
|
|
"ambiguous": get_sense(profile_id, label)["ambiguous"],
|
|
"suggested": _DECISION_NOT,
|
|
}
|
|
)
|
|
extra_mappings.append(
|
|
{
|
|
"token": _token_for_label(extra_mappings, label),
|
|
"local_label": label,
|
|
"canonical_label": label,
|
|
"demask_label": label,
|
|
"entity_type": "PERSON",
|
|
"source": "request_local",
|
|
"start": aligned[0],
|
|
"end": aligned[1],
|
|
"aliases": [],
|
|
"labels": [label],
|
|
}
|
|
)
|
|
return extra_candidates, extra_mappings
|
|
|
|
|
|
def apply_gliner_if_enabled(
|
|
profile_id: str,
|
|
candidates: list[dict],
|
|
mappings: list[dict],
|
|
user_body: str,
|
|
rendered: str = "",
|
|
) -> tuple[list[dict], list[dict]]:
|
|
settings = get_local_ner_settings()
|
|
if not settings.get("enabled"):
|
|
return candidates, mappings
|
|
from gliner_client import extract_entities
|
|
|
|
spans = extract_entities(user_body, url=settings.get("url"))
|
|
if spans is None:
|
|
return candidates, mappings
|
|
return merge_gliner_candidates(profile_id, candidates, mappings, user_body, rendered, spans)
|
|
|
|
|
|
def passage_provider():
|
|
"""LLM profile for mini-passage decisions. Independent of the detect role."""
|
|
import os
|
|
|
|
from providers import ProviderConfig, is_local_url
|
|
from provider_settings import get_profile
|
|
|
|
profile_id = _app_setting(PASSAGE_PROFILE_KEY, "")
|
|
if not profile_id:
|
|
return None
|
|
profile = get_profile(profile_id)
|
|
if not profile:
|
|
return None
|
|
url = profile.get("url") or ""
|
|
model = profile.get("model") or ""
|
|
if not url or not model:
|
|
return None
|
|
local = is_local_url(url)
|
|
key = ""
|
|
if not local:
|
|
key = (os.environ.get("KANSHO_PROVIDER_KEY") or os.environ.get("KANSHO_DETECT_PROVIDER_KEY") or "").strip()
|
|
if not key:
|
|
return None
|
|
if not profile.get("zdr") or not profile.get("no_train"):
|
|
return None
|
|
return ProviderConfig(
|
|
role="passage",
|
|
name=profile.get("name") or ("ollama" if local else "openrouter"),
|
|
mode="http",
|
|
url=url,
|
|
model=model,
|
|
key=key,
|
|
local=local,
|
|
zdr=True if local else bool(profile.get("zdr")),
|
|
no_train=True if local else bool(profile.get("no_train")),
|
|
)
|
|
|
|
|
|
def build_candidates(profile_id: str, mappings: list[dict], user_body: str, rendered: str = "") -> list[dict]:
|
|
seen: set[tuple[int | None, int | None, str]] = set()
|
|
items: list[dict] = []
|
|
for mapping in mappings:
|
|
if not needs_review(profile_id, mapping, user_body, rendered):
|
|
continue
|
|
label = mapping.get("local_label") or ""
|
|
start = mapping.get("start")
|
|
end = mapping.get("end")
|
|
key = (start, end, label.casefold())
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
items.append(
|
|
{
|
|
"id": str(uuid.uuid4()),
|
|
"text": label,
|
|
"entity_type": mapping.get("entity_type") or "PERSON",
|
|
"start": start,
|
|
"end": end,
|
|
**_mention_fields(user_body, start, end, label, rendered),
|
|
"ambiguous": get_sense(profile_id, label)["ambiguous"],
|
|
}
|
|
)
|
|
return items
|
|
|
|
|
|
def suppress_known_non_identity(profile_id: str, mappings: list[dict], user_body: str) -> list[dict]:
|
|
kept = []
|
|
for mapping in mappings:
|
|
label = mapping.get("local_label") or ""
|
|
sense = get_sense(profile_id, label)
|
|
if (
|
|
mapping.get("source") == "request_local"
|
|
and sense["non_identity_hits"] > 0
|
|
and not sense["identity_hits"]
|
|
and _in_user_text(label, user_body)
|
|
):
|
|
continue
|
|
kept.append(mapping)
|
|
return kept
|
|
|
|
|
|
def save_pending(
|
|
profile_id: str,
|
|
conversation_id: str,
|
|
user_message_id: str,
|
|
payload: dict,
|
|
) -> str:
|
|
review_id = str(uuid.uuid4())
|
|
with get_db() as conn:
|
|
conn.execute(
|
|
"DELETE FROM pending_mask_reviews WHERE profile_id = ? AND conversation_id = ?",
|
|
(profile_id, conversation_id),
|
|
)
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO pending_mask_reviews (
|
|
id, profile_id, conversation_id, user_message_id, payload_json, created
|
|
)
|
|
VALUES (?, ?, ?, ?, ?, datetime('now'))
|
|
""",
|
|
(review_id, profile_id, conversation_id, user_message_id, json.dumps(payload, ensure_ascii=False)),
|
|
)
|
|
return review_id
|
|
|
|
|
|
def load_pending(profile_id: str, review_id: str) -> dict | None:
|
|
with get_db() as conn:
|
|
row = row_to_dict(
|
|
conn.execute(
|
|
"""
|
|
SELECT * FROM pending_mask_reviews
|
|
WHERE id = ? AND profile_id = ?
|
|
""",
|
|
(review_id, profile_id),
|
|
).fetchone()
|
|
)
|
|
if not row:
|
|
return None
|
|
payload = json.loads(row.get("payload_json") or "{}")
|
|
payload["id"] = row["id"]
|
|
payload["conversation_id"] = row["conversation_id"]
|
|
payload["user_message_id"] = row["user_message_id"]
|
|
return payload
|
|
|
|
|
|
def drop_pending(profile_id: str, review_id: str) -> None:
|
|
with get_db() as conn:
|
|
conn.execute(
|
|
"DELETE FROM pending_mask_reviews WHERE id = ? AND profile_id = ?",
|
|
(review_id, profile_id),
|
|
)
|
|
|
|
|
|
def apply_review_decisions(profile_id: str, pending: dict, decisions: list[dict]) -> list[dict]:
|
|
by_id = {item["id"]: item for item in pending.get("candidates") or []}
|
|
drop_keys: set[tuple] = set()
|
|
for raw in decisions:
|
|
candidate = by_id.get(raw.get("id") or "")
|
|
if not candidate:
|
|
continue
|
|
decision = (raw.get("decision") or "").strip()
|
|
label = candidate.get("text") or ""
|
|
kind = candidate.get("entity_type") if candidate.get("entity_type") in ENTITY_TYPES else "PERSON"
|
|
cue = candidate.get("cue") or cue_for_mention(
|
|
pending.get("user_body") or "",
|
|
candidate.get("user_start", candidate.get("start")),
|
|
candidate.get("user_end", candidate.get("end")),
|
|
label,
|
|
)
|
|
if decision == _DECISION_IDENTITY:
|
|
record_sense(profile_id, label, identity=True)
|
|
record_cue(profile_id, label, cue, _DECISION_IDENTITY, candidate.get("excerpt") or "")
|
|
confirm_identity(profile_id, label, entity_type=kind)
|
|
elif decision == _DECISION_NOT:
|
|
record_sense(profile_id, label, identity=False)
|
|
record_cue(profile_id, label, cue, _DECISION_NOT, candidate.get("excerpt") or "")
|
|
drop_keys.add((candidate.get("start"), candidate.get("end"), label.casefold()))
|
|
mappings = []
|
|
for mapping in pending.get("mappings") or []:
|
|
key = (mapping.get("start"), mapping.get("end"), (mapping.get("local_label") or "").casefold())
|
|
if key in drop_keys:
|
|
continue
|
|
mappings.append(mapping)
|
|
return suppress_known_non_identity(profile_id, mappings, pending.get("user_body") or "")
|
|
|
|
|
|
def confirm_known_identity_spans(profile_id: str, mappings: list[dict], user_body: str) -> None:
|
|
for mapping in mappings:
|
|
label = mapping.get("local_label") or ""
|
|
sense = get_sense(profile_id, label)
|
|
if not (
|
|
sense["identity_hits"] > 0
|
|
and not sense["ambiguous"]
|
|
and _in_user_text(label, user_body)
|
|
):
|
|
continue
|
|
kind = mapping.get("entity_type") if mapping.get("entity_type") in ENTITY_TYPES else "PERSON"
|
|
try:
|
|
confirm_identity(profile_id, label, entity_type=kind)
|
|
except ValueError:
|
|
continue
|
|
|
|
|
|
def list_senses(profile_id: str) -> list[dict]:
|
|
with get_db() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT normalized_label, identity_hits, non_identity_hits
|
|
FROM label_senses WHERE profile_id = ?
|
|
ORDER BY lower(normalized_label)
|
|
""",
|
|
(profile_id,),
|
|
).fetchall()
|
|
items = []
|
|
for raw in rows:
|
|
row = row_to_dict(raw) or {}
|
|
identity = int(row.get("identity_hits") or 0)
|
|
other = int(row.get("non_identity_hits") or 0)
|
|
items.append(
|
|
{
|
|
"normalized_label": row.get("normalized_label") or "",
|
|
"identity_hits": identity,
|
|
"non_identity_hits": other,
|
|
"ambiguous": identity > 0 and other > 0,
|
|
}
|
|
)
|
|
return items
|
|
|
|
|
|
def pending_for_conversation(profile_id: str, conversation_id: str) -> dict | None:
|
|
with get_db() as conn:
|
|
row = row_to_dict(
|
|
conn.execute(
|
|
"""
|
|
SELECT id, payload_json FROM pending_mask_reviews
|
|
WHERE profile_id = ? AND conversation_id = ?
|
|
""",
|
|
(profile_id, conversation_id),
|
|
).fetchone()
|
|
)
|
|
if not row:
|
|
return None
|
|
payload = json.loads(row.get("payload_json") or "{}")
|
|
return {
|
|
"id": row["id"],
|
|
"candidates": payload.get("candidates") or [],
|
|
"kind": payload.get("kind") or "diff",
|
|
"purpose": payload.get("purpose") or "dialogue_turn",
|
|
}
|
|
|
|
|
|
def auto_resolve_ambiguous(profile_id: str, candidates: list[dict], mappings: list[dict]) -> tuple[list[dict], list[dict]]:
|
|
"""Local passage LLM for already-ambiguous spellings. Unresolved stay for the popup."""
|
|
remaining = []
|
|
drop_keys: set[tuple] = set()
|
|
for item in candidates:
|
|
if not item.get("ambiguous"):
|
|
remaining.append(item)
|
|
continue
|
|
label = item.get("text") or ""
|
|
examples = list_cue_examples(profile_id, label)
|
|
decision = try_local_passage_decision(item.get("excerpt") or "", label, examples)
|
|
if decision == _DECISION_NOT:
|
|
record_sense(profile_id, label, identity=False)
|
|
record_cue(profile_id, label, item.get("cue") or _CUE_BARE, _DECISION_NOT, item.get("excerpt") or "")
|
|
drop_keys.add((item.get("start"), item.get("end"), label.casefold()))
|
|
elif decision == _DECISION_IDENTITY:
|
|
record_sense(profile_id, label, identity=True)
|
|
record_cue(profile_id, label, item.get("cue") or _CUE_BARE, _DECISION_IDENTITY, item.get("excerpt") or "")
|
|
else:
|
|
remaining.append(item)
|
|
if not drop_keys:
|
|
return remaining, mappings
|
|
kept = []
|
|
for mapping in mappings:
|
|
key = (mapping.get("start"), mapping.get("end"), (mapping.get("local_label") or "").casefold())
|
|
if key in drop_keys:
|
|
continue
|
|
kept.append(mapping)
|
|
return remaining, kept
|
|
|
|
|
|
def try_local_passage_decision(excerpt: str, label: str, examples: list[dict] | None = None) -> str | None:
|
|
"""Return identity, not_identity, or None. Uses the admin passage profile, not Detect."""
|
|
from providers import ProviderError, complete_chat
|
|
|
|
config = passage_provider()
|
|
if not config or config.mode != "http":
|
|
return None
|
|
few_shot = ""
|
|
lines = []
|
|
for example in examples or []:
|
|
snippet = (example.get("excerpt") or "").strip()
|
|
decision = (example.get("decision") or "").strip()
|
|
if snippet and decision in {_DECISION_IDENTITY, _DECISION_NOT}:
|
|
lines.append(f'- "{snippet}" → {decision}')
|
|
if lines:
|
|
few_shot = "Bisherige lokale Bestätigungen als Beispiele, nicht als Wortliste:\n" + "\n".join(lines) + "\n"
|
|
prompt = (
|
|
f"{few_shot}"
|
|
"Entscheide nur für die markierte Nennung in diesem kurzen Ausschnitt. "
|
|
"Nutze den Satzkontext, nicht nur das Wort davor. "
|
|
"Ist sie eine schützenswerte Identität (Person, Ort, Organisation, privates Projekt) "
|
|
f"oder eine Sache/Allgemeinbedeutung? Wort: {label}\n"
|
|
f"Ausschnitt: {excerpt}\n"
|
|
'Antworte nur mit JSON {"decision":"identity"} oder {"decision":"not_identity"}.'
|
|
)
|
|
try:
|
|
result = complete_chat(
|
|
config,
|
|
[{"role": "user", "content": prompt}],
|
|
timeout=30.0,
|
|
max_tokens=32,
|
|
)
|
|
except ProviderError:
|
|
return None
|
|
match = re.search(r"\{.*\}", result.content or "", re.DOTALL)
|
|
if not match:
|
|
return None
|
|
try:
|
|
data = json.loads(match.group(0))
|
|
except json.JSONDecodeError:
|
|
return None
|
|
value = (data.get("decision") or "").strip()
|
|
if value in {_DECISION_IDENTITY, _DECISION_NOT}:
|
|
return value
|
|
return None
|