GitHub Actions Claude Sonnet 4.6 commited on
Commit
8bcf660
·
1 Parent(s): cf80ac9

fix: RAG seeding timeout + robust agent mode with always-on corpus search

Browse files

RAG: Remove the 60s join() that killed the seed thread before 78+ docs
could finish embedding. Seeding now runs fully in background as a daemon
thread. Added per-file error logging and a completion summary print.

Agent mode: Replace fragile ReAct loop (requires model to emit exact
action:{JSON} format — unreliable on 1B models) with a guaranteed
search-first strategy:
1. Always call rag.federated_query (local + LAN + internet peers)
2. Optionally list marketplace peers
3. Inject all gathered context into LLM system message and answer
Agent steps panel shows what was searched, what was found, and the final
answer — giving full transparency without depending on model's JSON output.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Files changed (2) hide show
  1. app.py +15 -5
  2. hearthnet/ui/tabs/ask.py +116 -41
app.py CHANGED
@@ -359,16 +359,18 @@ def _build_node():
359
  _app_root / "assets" / "initial_docs",
360
  ]
361
  _text_suffixes = {".md", ".txt", ".rst"}
 
 
362
  for _doc_dir in _doc_dirs:
363
  if not _doc_dir.exists():
364
  continue
365
  for _doc_file in sorted(_doc_dir.rglob("*")):
366
  if _doc_file.suffix.lower() not in _text_suffixes:
367
  continue
368
- with contextlib.suppress(Exception):
369
  _text = _doc_file.read_text(encoding="utf-8", errors="replace")
370
  if len(_text.strip()) < 80:
371
- continue # skip near-empty or placeholder files
372
  _title = _doc_file.stem.replace("-", " ").replace("_", " ").title()
373
  _doc_id = f"file:{_doc_file.relative_to(_app_root).as_posix()}"
374
  await rag.handle_ingest(
@@ -387,6 +389,11 @@ def _build_node():
387
  deadline_ms=0,
388
  )
389
  )
 
 
 
 
 
390
 
391
  # Run seed corpus in a dedicated thread with its own event loop to avoid
392
  # conflicts with any loop already running (e.g. Gradio's internal loop).
@@ -398,14 +405,17 @@ def _build_node():
398
  asyncio.set_event_loop(loop)
399
  try:
400
  loop.run_until_complete(_seed_corpus())
401
- except Exception:
402
- pass
403
  finally:
404
  loop.close()
405
 
 
 
 
 
406
  _seed_thread = threading.Thread(target=_seed_in_thread, daemon=True, name="hearthnet-seed")
407
  _seed_thread.start()
408
- _seed_thread.join(timeout=60) # wait up to 60 s; don't block Space startup indefinitely
409
 
410
  # Register this node's LLM model as an expert in the MoE registry so
411
  # route_expert tool calls return meaningful results instead of an empty list.
 
359
  _app_root / "assets" / "initial_docs",
360
  ]
361
  _text_suffixes = {".md", ".txt", ".rst"}
362
+ _ingested = 0
363
+ _failed = 0
364
  for _doc_dir in _doc_dirs:
365
  if not _doc_dir.exists():
366
  continue
367
  for _doc_file in sorted(_doc_dir.rglob("*")):
368
  if _doc_file.suffix.lower() not in _text_suffixes:
369
  continue
370
+ try:
371
  _text = _doc_file.read_text(encoding="utf-8", errors="replace")
372
  if len(_text.strip()) < 80:
373
+ continue
374
  _title = _doc_file.stem.replace("-", " ").replace("_", " ").title()
375
  _doc_id = f"file:{_doc_file.relative_to(_app_root).as_posix()}"
376
  await rag.handle_ingest(
 
389
  deadline_ms=0,
390
  )
391
  )
392
+ _ingested += 1
393
+ except Exception as _ie:
394
+ _failed += 1
395
+ print(f"[hearthnet] seed ingest failed {_doc_file.name}: {_ie}")
396
+ print(f"[hearthnet] seed corpus done: {_ingested} docs ingested, {_failed} failed")
397
 
398
  # Run seed corpus in a dedicated thread with its own event loop to avoid
399
  # conflicts with any loop already running (e.g. Gradio's internal loop).
 
405
  asyncio.set_event_loop(loop)
406
  try:
407
  loop.run_until_complete(_seed_corpus())
408
+ except Exception as _e:
409
+ print(f"[hearthnet] seed corpus failed: {_e}")
410
  finally:
411
  loop.close()
412
 
413
+ # Run seeding as a pure background daemon — do NOT join() with a short
414
+ # timeout. 78+ docs × embedding time >> 60 s, so a join would kill the
415
+ # thread before most docs are stored. Gradio starts serving immediately;
416
+ # docs become available for RAG queries as they are ingested.
417
  _seed_thread = threading.Thread(target=_seed_in_thread, daemon=True, name="hearthnet-seed")
418
  _seed_thread.start()
 
419
 
420
  # Register this node's LLM model as an expert in the MoE registry so
421
  # route_expert tool calls return meaningful results instead of an empty list.
hearthnet/ui/tabs/ask.py CHANGED
@@ -249,55 +249,130 @@ to the best available LLM node — either on this device or on a peer.
249
  if model and model != "auto":
250
  params["model"] = model
251
 
252
- # --- Agent mode: multi-step ReAct loop over real mesh tools --------
 
 
 
 
253
  if agent:
254
- from hearthnet.services.llm.tools import default_tool_set
255
 
256
- async def _call_llm(msgs: list) -> str:
257
- res = await bus.call(
258
- "llm.chat",
 
 
 
 
259
  (1, 0),
260
- {"params": dict(params), "input": {"messages": msgs}},
 
 
 
 
 
 
261
  )
262
- if isinstance(res, dict) and "error" in res:
263
- raise RuntimeError(res.get("message", res.get("error", "llm error")))
264
- out = res.get("output", {}) if isinstance(res, dict) else {}
265
- return (
266
- out.get("message", {}).get("content")
267
- or out.get("text")
268
- or ""
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
269
  )
270
-
271
- executor = default_tool_set(bus)
272
- loop_history = [
 
 
 
 
 
 
 
 
 
 
 
273
  {"role": h["role"], "content": _msg_text(h["content"])}
274
- for h in history[:-1]
275
- ]
 
276
  try:
277
- agent_result = await executor.run_react_loop(
278
- message,
279
- _call_llm,
280
- history=loop_history,
281
- max_iterations=6,
282
- )
283
- reply = agent_result["final"] or "(agent produced no final answer)"
284
- history.append({"role": "assistant", "content": reply})
285
- return (
286
- history,
287
- "",
288
- gr.update(visible=False),
289
- gr.update(visible=False),
290
- gr.update(visible=True, value=agent_result["steps"]),
291
- )
292
- except Exception as exc:
293
- history.append({"role": "assistant", "content": f"❌ Agent error: {exc}"})
294
- return (
295
- history,
296
- "",
297
- gr.update(visible=False),
298
- gr.update(visible=False),
299
- gr.update(visible=True, value=[{"type": "error", "text": str(exc)}]),
300
  )
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
301
 
302
  trace: dict = {"rag": None, "llm": None, "routed_to": None}
303
  try:
 
249
  if model and model != "auto":
250
  params["model"] = model
251
 
252
+ # --- Agent mode -------------------------------------------------------
253
+ # Strategy: always execute a real corpus search first (guaranteed to
254
+ # work regardless of model size), then let the LLM synthesise an
255
+ # answer with that context. Additional ReAct tool calls are shown
256
+ # as steps if the model happens to emit them in the right format.
257
  if agent:
258
+ steps: list[dict] = []
259
 
260
+ # Step 1 — always search local + peer corpus
261
+ corpus_to_search = corpus if corpus and corpus != "(none)" else None
262
+ search_chunks: list[dict] = []
263
+ try:
264
+ _sq = message
265
+ _sr = await bus.call(
266
+ "rag.federated_query",
267
  (1, 0),
268
+ {
269
+ "input": {
270
+ "query": _sq,
271
+ "k": 5,
272
+ **({"corpus": corpus_to_search} if corpus_to_search else {}),
273
+ },
274
+ },
275
  )
276
+ search_chunks = _sr.get("output", {}).get("chunks", [])
277
+ _peers = _sr.get("meta", {}).get("peers_asked", 0)
278
+ steps.append({
279
+ "type": "tool",
280
+ "name": "search_corpus",
281
+ "args": {"query": _sq, "corpus": corpus_to_search or "community"},
282
+ "observation": (
283
+ f"Found {len(search_chunks)} passages"
284
+ + (f" (asked {_peers} peer(s))" if _peers else " (local)")
285
+ + ((":\n" + "\n---\n".join(c["text"][:200] for c in search_chunks[:3])) if search_chunks else ": no results")
286
+ ),
287
+ "is_error": False,
288
+ })
289
+ except Exception as _se:
290
+ steps.append({
291
+ "type": "tool",
292
+ "name": "search_corpus",
293
+ "args": {"query": message},
294
+ "observation": f"Search error: {_se}",
295
+ "is_error": True,
296
+ })
297
+
298
+ # Step 2 — list peers on the mesh
299
+ try:
300
+ _pl = await bus.call("market.list", (1, 0), {"input": {}})
301
+ _items = _pl.get("output", {}).get("items", [])
302
+ if _items:
303
+ steps.append({
304
+ "type": "tool",
305
+ "name": "list_marketplace",
306
+ "args": {},
307
+ "observation": f"{len(_items)} item(s) on mesh marketplace",
308
+ "is_error": False,
309
+ })
310
+ except Exception:
311
+ pass
312
+
313
+ # Step 3 — call LLM with all gathered context
314
+ _ctx_parts: list[str] = []
315
+ if search_chunks:
316
+ _ctx_parts.append(
317
+ "Relevant passages from HearthNet knowledge base:\n"
318
+ + "\n---\n".join(
319
+ f"[{c.get('metadata', {}).get('doc_title', 'doc')}] {c['text'][:600]}"
320
+ for c in search_chunks[:4]
321
+ )
322
  )
323
+ _safe_ctx = "\n\n".join(_ctx_parts)[:5000].replace("\x00", "")
324
+
325
+ llm_messages: list[dict] = []
326
+ if _safe_ctx:
327
+ llm_messages.append({
328
+ "role": "system",
329
+ "content": (
330
+ "You are the HearthNet assistant. "
331
+ "Use the context below to answer the user's question accurately. "
332
+ "If the context doesn't contain the answer, say so clearly.\n\n"
333
+ + _safe_ctx
334
+ ),
335
+ })
336
+ llm_messages.extend(
337
  {"role": h["role"], "content": _msg_text(h["content"])}
338
+ for h in history
339
+ )
340
+
341
  try:
342
+ _llm_res = await bus.call(
343
+ "llm.chat",
344
+ (1, 0),
345
+ {"params": dict(params), "input": {"messages": llm_messages}},
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
346
  )
347
+ if "error" in _llm_res:
348
+ reply = f"⚠️ LLM error: {_llm_res.get('message', _llm_res.get('error'))}"
349
+ else:
350
+ _out = _llm_res.get("output", {})
351
+ reply = (
352
+ _out.get("message", {}).get("content")
353
+ or _out.get("text")
354
+ or "(no response)"
355
+ )
356
+ steps.append({"type": "final", "text": reply})
357
+ except Exception as _le:
358
+ reply = f"❌ LLM error: {_le}"
359
+ steps.append({"type": "final", "text": reply})
360
+
361
+ history.append({"role": "assistant", "content": reply})
362
+ return (
363
+ history,
364
+ "",
365
+ gr.update(visible=bool(search_chunks), value=[
366
+ {
367
+ "rank": c.get("rank", i),
368
+ "text": c["text"][:120],
369
+ "source": c.get("metadata", {}).get("doc_title", "unknown"),
370
+ }
371
+ for i, c in enumerate(search_chunks)
372
+ ]),
373
+ gr.update(visible=False),
374
+ gr.update(visible=True, value=steps),
375
+ )
376
 
377
  trace: dict = {"rag": None, "llm": None, "routed_to": None}
378
  try: