File size: 10,558 Bytes
5655a42 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 | """Dialectic layer for HonchoMemoryProvider: multi-pass reasoning, cadence/backoff, liveness."""
from __future__ import annotations
import logging
import re
import threading
import time
from plugins.memory.honcho.client import spawn_context_thread
logger = logging.getLogger("plugins.memory.honcho")
_STRUCTURE_RE = re.compile(r"^(?:[*-] |\s*\d+\. )", re.MULTILINE) # bullet / ordered-list line
class DialecticMixin:
"""Dialectic prefetch state machine. State attributes are initialised in
HonchoMemoryProvider.__init__; this mixin only reads/writes them."""
# Reasoning level per (depth, pass) when dialecticDepthLevels is not configured;
# "base" means dialecticReasoningLevel. Early passes of deeper runs are lighter.
_PROPORTIONAL_LEVELS: dict[tuple[int, int], str] = {
(1, 0): "base", (2, 0): "minimal", (2, 1): "base", (3, 0): "minimal", (3, 1): "base", (3, 2): "low",
}
_LEVEL_ORDER = ("minimal", "low", "medium", "high", "max")
# Char-count thresholds for the query-length reasoning heuristic.
_HEURISTIC_LENGTH_MEDIUM, _HEURISTIC_LENGTH_HIGH = 120, 400
# A thread older than timeout × multiplier is treated as dead so a hung Honcho call
# can't block future fires indefinitely.
_STALE_THREAD_MULTIPLIER = 2.0
# A pending result fired more than cadence × multiplier turns ago is discarded on read,
# so a stale conversational pivot isn't injected after trivial-prompt turns.
_STALE_RESULT_MULTIPLIER = 2
_BACKOFF_MAX = 8 # ceiling for the empty-streak backoff (× base cadence)
# Turn-1 budgets: session init + base retrieval share the first; dialectic gets the second.
_FIRST_TURN_BASE_TIMEOUT, _FIRST_TURN_DIALECTIC_CAP = 3.0, 2.0
def _thread_is_live(self) -> bool:
"""Thread-alive guard that treats threads past the stale threshold as dead."""
if not self._prefetch_thread or not self._prefetch_thread.is_alive():
return False
timeout = self._config.timeout if self._config and self._config.timeout else 8.0
stale_after = timeout * self._STALE_THREAD_MULTIPLIER
age = time.monotonic() - self._prefetch_thread_started_at
if age > stale_after:
logger.debug("Honcho prefetch thread age %.1fs exceeds stale threshold %.1fs — treating as dead", age, stale_after)
return False
return True
def _effective_cadence(self) -> int:
"""Cadence plus empty-streak backoff, capped at _BACKOFF_MAX × base."""
cadence, streak = self._dialectic_cadence, self._dialectic_empty_streak
return min(cadence + streak, cadence * self._BACKOFF_MAX) if streak > 0 else cadence
def _note_dialectic_failure(self, exc: BaseException) -> None:
"""Widen the empty-streak backoff after a failed cycle; auth failures are exempt
because waiting cannot fix a dead token."""
from plugins.memory.honcho.session import HonchoAuthError
if isinstance(exc, HonchoAuthError):
logger.warning("Honcho dialectic auth failure (not counted toward cadence backoff): %s", exc)
return
self._dialectic_empty_streak += 1
def _spawn_dialectic(
self, query: str, *, thread_name: str, fired_at: int, log_label: str, use_query_rewrite: bool = True,
) -> threading.Thread:
"""Start a background dialectic run that publishes into the pending-result slot. Only a
non-empty result advances ``_last_dialectic_turn`` (so empty returns retry next turn) and
resets the empty streak; failures widen the backoff."""
def _run() -> None:
try:
r = self._run_dialectic_depth(query, use_query_rewrite=use_query_rewrite)
except Exception as exc:
logger.debug("Honcho %s failed: %s", log_label, exc)
self._note_dialectic_failure(exc)
return
if r and r.strip():
with self._prefetch_lock:
self._prefetch_result = r
self._prefetch_result_fired_at = fired_at
self._last_dialectic_turn = fired_at
self._dialectic_empty_streak = 0
else:
self._dialectic_empty_streak += 1
self._prefetch_thread_started_at = time.monotonic()
thread = spawn_context_thread(_run, name=thread_name, owner=self)
thread.start()
self._prefetch_thread = thread
return thread
def _consume_pending_dialectic(self) -> str:
"""Pop the pending dialectic result, or "" when none is ready or it is stale."""
with self._prefetch_lock:
dialectic_result, fired_at = self._prefetch_result, self._prefetch_result_fired_at
self._prefetch_result, self._prefetch_result_fired_at = "", -999
stale_limit = self._dialectic_cadence * self._STALE_RESULT_MULTIPLIER
if dialectic_result and fired_at >= 0 and (self._turn_count - fired_at) > stale_limit:
logger.debug("Honcho pending dialectic discarded as stale: fired_at=%d, turn=%d, limit=%d",
fired_at, self._turn_count, stale_limit)
return ""
return dialectic_result if (dialectic_result and dialectic_result.strip()) else ""
def _apply_reasoning_heuristic(self, base: str, query: str) -> str:
"""Scale ``base`` up by query length (+1 at >=120 chars, +2 at >=400), clamped at reasoning_level_cap."""
if not self._reasoning_heuristic or not query or base not in self._LEVEL_ORDER:
return base
n = len(query)
bump = 0 if n < self._HEURISTIC_LENGTH_MEDIUM else 1 if n < self._HEURISTIC_LENGTH_HIGH else 2
cap_idx = self._LEVEL_ORDER.index(self._reasoning_level_cap)
return self._LEVEL_ORDER[min(self._LEVEL_ORDER.index(base) + bump, cap_idx)]
def _resolve_pass_level(self, pass_idx: int, query: str = "") -> str:
"""Reasoning level for a pass: explicit dialecticDepthLevels win, then the
proportional table, then the base level scaled by the length heuristic."""
if self._dialectic_depth_levels and pass_idx < len(self._dialectic_depth_levels):
return self._dialectic_depth_levels[pass_idx]
mapping = self._PROPORTIONAL_LEVELS.get((self._dialectic_depth, pass_idx))
if mapping is None or mapping == "base":
base = self._config.dialectic_reasoning_level if self._config else "low"
return self._apply_reasoning_heuristic(base, query)
return mapping
def _build_dialectic_prompt(self, pass_idx: int, prior_results: list[str], is_cold: bool) -> str:
"""Pass 0: cold (general) or warm (session-scoped) query; pass 1: gap audit
against the prior result; pass 2: reconciliation across prior passes."""
if pass_idx == 0:
if is_cold:
return ("Who is this person? What are their preferences, goals, and working style? "
"Focus on facts that would help an AI assistant be immediately useful.")
return ("Given what's been discussed in this session so far, what context about this user is most "
"relevant to the current conversation? Prioritize active context over biographical facts.")
if pass_idx == 1:
prior = prior_results[-1] if prior_results else ""
return (f"Given this initial assessment:\n\n{prior}\n\n"
"What gaps remain in your understanding that would help going forward? Synthesize what you "
"actually know about the user's current state and immediate needs, grounded in evidence from "
"recent sessions.")
p0, p1 = [prior_results[i] if len(prior_results) > i else "(empty)" for i in (0, 1)]
return (f"Prior passes produced:\n\nPass 1:\n{p0}\n\nPass 2:\n{p1}\n\n"
"Do these assessments cohere? Reconcile any contradictions and produce a final, concise "
"synthesis of what matters most for the current conversation.")
@staticmethod
def _signal_sufficient(result: str) -> bool:
"""True when a pass returned enough signal to skip further passes: >100 chars
with structure (headers/bullets/ordered list), or >300 chars regardless."""
if not result or len(result.strip()) < 100:
return False
structured = "\n" in result and bool("##" in result or "•" in result or _STRUCTURE_RE.search(result))
return structured or len(result.strip()) > 300
def _run_dialectic_depth(self, query: str, *, use_query_rewrite: bool = True) -> str:
"""Run up to dialecticDepth .chat() passes, bailing early once a pass returns
strong signal. Returns the last non-empty result."""
if not self._manager or not self._session_key:
return ""
is_cold = not self._base_context_cache
results: list[str] = []
rewritten_query = ""
if use_query_rewrite and self._query_rewrite_enabled and self._query_rewriter:
try:
rewritten_query = self._query_rewriter(query).strip()
except Exception as exc:
logger.debug("Honcho query rewriter failed: %s", exc)
depth = self._dialectic_depth
for i in range(depth):
# Dependent prompts require a non-empty prior result; without one, later
# passes retry the independent base prompt.
prior_results = [r for r in results if r and r.strip()]
if prior_results and self._signal_sufficient(prior_results[-1]):
logger.debug("Honcho dialectic depth %d: pass %d skipped, prior signal sufficient", depth, i)
break
if prior_results:
prompt = self._build_dialectic_prompt(i, prior_results, is_cold)
else:
if i:
logger.debug("Honcho dialectic depth %d: pass %d has no non-empty prior — "
"falling back to base prompt", depth, i)
prompt = rewritten_query or self._build_dialectic_prompt(0, [], is_cold)
level = self._resolve_pass_level(i, query=query)
logger.debug("Honcho dialectic depth %d: pass %d, level=%s, cold=%s", depth, i, level, is_cold)
result = self._manager.dialectic_query(self._session_key, prompt, reasoning_level=level, peer="user")
results.append(result or "")
return next((r for r in reversed(results) if r and r.strip()), "")
|