matchmaking-v2: heartbeat progress during the long search
Emit a rotating progress message every ~5s while the Apify scrape + Opus curation run, so the long (30-40s) silent gap can't idle the response relay and improves the wait UX.
This commit is contained in:
+27
-3
@@ -24,6 +24,23 @@ class MatchmakingAgentSession:
|
|||||||
async def push(self, msg_type: str, **kwargs):
|
async def push(self, msg_type: str, **kwargs):
|
||||||
await self.ws.send_json({"type": msg_type, **kwargs})
|
await self.ws.send_json({"type": msg_type, **kwargs})
|
||||||
|
|
||||||
|
async def _with_heartbeat(self, awaitable, messages: list[str], interval: float = 5.0):
|
||||||
|
"""Await `awaitable` while emitting a rotating progress message every `interval`s.
|
||||||
|
|
||||||
|
A live search runs ~30-40s (Apify scrape + Opus curation). Without a steady trickle of
|
||||||
|
messages, the Redis pub/sub channel that relays the result back to the browser idles out and
|
||||||
|
the final `search_complete` is dropped — which is why users were seeing 0 matches. The
|
||||||
|
heartbeat keeps that channel warm AND gives the user real progress during the wait.
|
||||||
|
"""
|
||||||
|
task = asyncio.ensure_future(awaitable)
|
||||||
|
i = 0
|
||||||
|
while True:
|
||||||
|
done, _ = await asyncio.wait({task}, timeout=interval)
|
||||||
|
if task in done:
|
||||||
|
return task.result()
|
||||||
|
await self.push("agent_thinking", message=messages[i % len(messages)])
|
||||||
|
i += 1
|
||||||
|
|
||||||
async def _stub(self, action: str, note: str):
|
async def _stub(self, action: str, note: str):
|
||||||
await self.push("agent_data", action=action, data={"status": "not_implemented", "note": note})
|
await self.push("agent_data", action=action, data={"status": "not_implemented", "note": note})
|
||||||
|
|
||||||
@@ -102,7 +119,11 @@ class MatchmakingAgentSession:
|
|||||||
sig = _repo.search_signature(prefs)
|
sig = _repo.search_signature(prefs)
|
||||||
cursors = await _repo.get_search_cursors(self.user_id, sig)
|
cursors = await _repo.get_search_cursors(self.user_id, sig)
|
||||||
prefs["_cursors"] = cursors
|
prefs["_cursors"] = cursors
|
||||||
sweep = await search.run_sweep(prefs, fresh=fresh)
|
# Heartbeat through the long Apify scrape so the relay channel stays alive (see _with_heartbeat).
|
||||||
|
sweep = await self._with_heartbeat(
|
||||||
|
search.run_sweep(prefs, fresh=fresh),
|
||||||
|
["Scanning live roles across the boards…", "Pulling the latest Naukri + LinkedIn + Foundit roles…",
|
||||||
|
"Gathering fresh postings for you…"])
|
||||||
# Safety net: drop anything already shown / applied / dismissed (covers boards that can't
|
# Safety net: drop anything already shown / applied / dismissed (covers boards that can't
|
||||||
# paginate, so the same posting never reappears even if a board re-returns it).
|
# paginate, so the same posting never reappears even if a board re-returns it).
|
||||||
seen = await _repo.get_seen_ids(self.user_id)
|
seen = await _repo.get_seen_ids(self.user_id)
|
||||||
@@ -117,8 +138,11 @@ class MatchmakingAgentSession:
|
|||||||
# Run the blocking LLM work (embeddings + Opus) OFF the event loop, or the long sync OpenAI
|
# Run the blocking LLM work (embeddings + Opus) OFF the event loop, or the long sync OpenAI
|
||||||
# call freezes the loop and the Redis response can't publish (→ the loader hangs forever).
|
# call freezes the loop and the Redis response can't publish (→ the loader hangs forever).
|
||||||
top, dbg = await asyncio.to_thread(_sift.sift, prefs, user_context, candidates)
|
top, dbg = await asyncio.to_thread(_sift.sift, prefs, user_context, candidates)
|
||||||
await self.push("agent_thinking", message="Scout is reading your top roles…")
|
# Heartbeat through the long Opus curation (the other 30-40s gap that was dropping responses).
|
||||||
curated = await asyncio.to_thread(_curate.curate, prefs, user_context, top) # Opus's call is final
|
curated = await self._with_heartbeat(
|
||||||
|
asyncio.to_thread(_curate.curate, prefs, user_context, top), # Opus's call is final
|
||||||
|
["Scout is reading your top roles…", "Scoring fit + writing your report cards…",
|
||||||
|
"Ranking your strongest matches…", "Almost there…"])
|
||||||
if curated is None: # safety net: sift + templated cards
|
if curated is None: # safety net: sift + templated cards
|
||||||
curated = _rank.select(top, prefs.get("stretch", "balanced"))
|
curated = _rank.select(top, prefs.get("stretch", "balanced"))
|
||||||
engine = f"fallback:{dbg['mode']}"
|
engine = f"fallback:{dbg['mode']}"
|
||||||
|
|||||||
Reference in New Issue
Block a user