perf: move all slow operations to background threads
Blocking operations in the request path caused SSE timeouts: - Goal classification (~4.3s LLM call) → background thread - Hierarchy rebuild (loads all objects) → background thread - Per-segment find_duplicate (async per-object) → skipped - Per-object goal relevance loop → disabled Request path now only does: ingest, segment, admission (fast), then forwards to Anthropic immediately.
This commit is contained in:
parent
2c14ec09b2
commit
a0822b0f6d
1 changed files with 49 additions and 41 deletions
|
|
@ -1988,13 +1988,10 @@ def create_app(
|
|||
rejected_count = 0
|
||||
admitted_objects = []
|
||||
for seg_obj in new_objects:
|
||||
# Phase 4d: admission control — score and gate
|
||||
# Phase 4d: admission control — score and gate.
|
||||
# Skip async find_duplicate() in the hot path — it blocks
|
||||
# per-segment. Admission still gates on type/novelty/utility.
|
||||
has_dup = False
|
||||
if seg_obj.source_key:
|
||||
dup = _run_async(
|
||||
session.object_store.find_duplicate(session.id, seg_obj.source_key)
|
||||
)
|
||||
has_dup = dup is not None
|
||||
with Timer(session.benchmark.latency["admission"]):
|
||||
admitted, _score = session.admission.should_admit(
|
||||
seg_obj.content,
|
||||
|
|
@ -2074,30 +2071,30 @@ def create_app(
|
|||
# Enable hierarchy after turn 50
|
||||
session.hierarchy.enabled = True
|
||||
|
||||
# Feed all session objects into hierarchy if not yet populated
|
||||
if session.hierarchy.object_count == 0:
|
||||
all_stored = _run_async(session.object_store.get_session_objects(session.id))
|
||||
if all_stored:
|
||||
session.hierarchy.rebuild(all_stored)
|
||||
print(
|
||||
f" {_DIM}[{session.id}] hierarchy: initial build "
|
||||
f"({session.hierarchy.object_count} objects, "
|
||||
f"{session.hierarchy.episode_count} episodes, "
|
||||
f"{session.hierarchy.theme_count} themes){_RESET}",
|
||||
file=sys.stderr,
|
||||
)
|
||||
else:
|
||||
# Incremental: add any new objects from this turn
|
||||
all_stored = _run_async(session.object_store.get_session_objects(session.id))
|
||||
for obj in all_stored:
|
||||
session.hierarchy.add_object(obj)
|
||||
# Feed all session objects into hierarchy in background
|
||||
# to avoid blocking the request path with get_session_objects().
|
||||
def _hierarchy_bg(s=session, t=turn):
|
||||
try:
|
||||
if s.hierarchy.object_count == 0:
|
||||
all_stored = _run_async(s.object_store.get_session_objects(s.id))
|
||||
if all_stored:
|
||||
s.hierarchy.rebuild(all_stored)
|
||||
print(
|
||||
f" {_DIM}[{s.id}] hierarchy: initial build "
|
||||
f"({s.hierarchy.object_count} objects){_RESET}",
|
||||
file=sys.stderr,
|
||||
)
|
||||
else:
|
||||
all_stored = _run_async(s.object_store.get_session_objects(s.id))
|
||||
for obj in all_stored:
|
||||
s.hierarchy.add_object(obj)
|
||||
goal_hash = s._current_goal.goal if s._current_goal else None
|
||||
s.hierarchy.maintenance(t, goal_hash=goal_hash)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Periodic maintenance (re-cluster every 20 turns or on goal change)
|
||||
goal_hash = None
|
||||
if session._current_goal is not None:
|
||||
goal_hash = session._current_goal.goal
|
||||
rebuilt = session.hierarchy.maintenance(turn, goal_hash=goal_hash)
|
||||
if rebuilt:
|
||||
_threading.Thread(target=_hierarchy_bg, daemon=True).start()
|
||||
if False: # Dead code — keep for reference
|
||||
print(
|
||||
f" {_DIM}[{session.id}] hierarchy: rebuilt at turn {turn} "
|
||||
f"({session.hierarchy.episode_count} episodes, "
|
||||
|
|
@ -2140,30 +2137,41 @@ def create_app(
|
|||
|
||||
if goal_changed and helper_llm is not None:
|
||||
session.benchmark.goals.record_topic_shift()
|
||||
# Build recent context (last 2 messages)
|
||||
# Fire goal classification in background thread to avoid
|
||||
# blocking the request path (~4s LLM API call).
|
||||
recent = (
|
||||
incoming_messages[-4:] if len(incoming_messages) > 4 else incoming_messages
|
||||
)
|
||||
recent_text = "\n".join(str(m.get("content", ""))[:500] for m in recent)
|
||||
with Timer(session.benchmark.latency["goal_classification"]):
|
||||
goal = _run_async(helper_llm.classify_goal(user_text, recent_text))
|
||||
session._current_goal = goal
|
||||
session.benchmark.goals.record_reclassification()
|
||||
print(
|
||||
f" {_DIM}[{session.id}] goal transition: {goal.goal[:80]}, "
|
||||
f"types={goal.relevant_types}, tags={goal.relevant_tags}{_RESET}",
|
||||
file=sys.stderr,
|
||||
)
|
||||
|
||||
# Adjust fidelity based on goal relevance
|
||||
def _classify_goal_bg(s=session, ut=user_text, rt=recent_text, hlm=helper_llm):
|
||||
try:
|
||||
with Timer(s.benchmark.latency["goal_classification"]):
|
||||
goal = _run_async(hlm.classify_goal(ut, rt))
|
||||
s._current_goal = goal
|
||||
s.benchmark.goals.record_reclassification()
|
||||
print(
|
||||
f" {_DIM}[{s.id}] goal: {goal.goal[:80]}, "
|
||||
f"types={goal.relevant_types}{_RESET}",
|
||||
file=sys.stderr,
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
_threading.Thread(target=_classify_goal_bg, daemon=True).start()
|
||||
|
||||
# Skip the per-object goal relevance loop — it does N object_store.get()
|
||||
# calls synchronously which blocks the request path. The fidelity system
|
||||
# handles degradation/promotion via pressure zones instead.
|
||||
if False and session._current_goal is not None:
|
||||
turn = ts.get("turn", 0)
|
||||
fm = session.fidelity_manager
|
||||
promotions = 0
|
||||
goal = session._current_goal
|
||||
for obj_id in list(fm._objects.keys()):
|
||||
obj = fm.get_object(obj_id)
|
||||
if obj is None or obj.pinned:
|
||||
continue
|
||||
# Check if object matches goal via stored object tags
|
||||
stored = _run_async(session.object_store.get(obj_id))
|
||||
obj_tags = stored.tags if stored else []
|
||||
is_relevant = obj.object_type in goal.relevant_types or any(
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue