Spaces:
Sleeping
Sleeping
| # src/analyzer/chat/insight_tools.py | |
| from __future__ import annotations | |
| from typing import Dict, Any, List, Optional | |
| from ..config import load_config | |
| try: | |
| from ..llm_client import LLMClient | |
| except Exception: | |
| LLMClient = None # type: ignore | |
| try: | |
| from ..search.hybrid_index import load_index, search, top_supporting_for_grant | |
| except Exception: | |
| load_index = None # type: ignore | |
| search = None # type: ignore | |
| top_supporting_for_grant = None # type: ignore | |
| from ..prompt_templates import build_open_prompt | |
| class InsightTools: | |
| """ | |
| Retrieval + LLM glue for open, lightly-guarded answers. | |
| - Accepts an existing LLMClient (preferred) to ensure identical config across entrypoints | |
| - Falls back gracefully if LLM or index is unavailable | |
| """ | |
| def __init__(self, llm_client: Optional["LLMClient"] = None): | |
| self._cfg = load_config() | |
| self._client = llm_client | |
| if self._client is None and LLMClient is not None: | |
| try: | |
| self._client = LLMClient(self._cfg) | |
| except Exception: | |
| self._client = None | |
| self._idx = None | |
| if callable(load_index): | |
| try: | |
| self._idx = load_index() | |
| except Exception: | |
| self._idx = None | |
| # Optional helper some callers use to append snippets under a summary | |
| def supporting_snippets_md(self, grant_id: str, k: int = 5) -> str: | |
| if not self._idx or not callable(top_supporting_for_grant): | |
| return "" | |
| try: | |
| hits = top_supporting_for_grant(self._idx, grant_id, k=k) | |
| except Exception: | |
| return "" | |
| items: List[str] = [] | |
| for doc, score in hits: | |
| if doc.get("_source") != "supporting": | |
| continue | |
| sec = doc.get("section") or "(Supporting)" | |
| url = doc.get("url","") | |
| txt = (doc.get("text","") or "").replace("\n"," ") | |
| snippet = (txt[:400] + "…") if len(txt) > 400 else txt | |
| items.append(f"- **{sec}** — {url}\n > {snippet}") | |
| if not items: | |
| return "" | |
| return "\n\n---\n**Supporting info (top snippets)**\n" + "\n".join(items) | |
| def insight_search(self, question: str, *, k: int = 8, use_llm: bool = True) -> Dict[str, Any]: | |
| """Open-style grounded QA: supply question + top snippets; let the LLM pick format and length.""" | |
| if not self._idx or not callable(search): | |
| return {"answer_md": "Search index not available."} | |
| hits = search(self._idx, question, k=k, filters=None) | |
| ev_lines: List[str] = [] | |
| raw_context: List[str] = [] | |
| for doc, score in hits[:k]: | |
| sec = doc.get("section") or doc.get("title","") | |
| url = doc.get("url","") | |
| txt = (doc.get("text","") or "") | |
| ev_lines.append(f"- **{sec}** — {url}") | |
| raw_context.append(f"[{sec}] {url}\n{txt}") | |
| evidence_md = "**Sources**\n" + "\n".join(ev_lines) if ev_lines else "" | |
| if use_llm and self._client: | |
| payload = build_open_prompt( | |
| provider=getattr(self._cfg, "provider", "openai") if self._cfg else "openai", | |
| question=question, | |
| context="\n\n---\n".join(raw_context[:6]), | |
| ) | |
| try: | |
| # Slightly higher budget; model decides format naturally | |
| answer = self._client.chat(payload["messages"], max_tokens=1200, temperature=0.3) | |
| return {"answer_md": f"{answer}\n\n---\n{evidence_md}"} | |
| except Exception: | |
| pass | |
| return {"answer_md": "No LLM available. See sources below.\n\n---\n" + evidence_md} |