Run retrieval for *user_message* and return a detailed breakdown for inspection. Intended for the /rag command — never called during normal inference. Returns a dict with: query — the cleaned query sent to the search backend backend — "semantic" | "keywor
(user_message: str)
| 176 | |
| 177 | |
| 178 | def retrieve_debug(user_message: str) -> dict: |
| 179 | """ |
| 180 | Run retrieval for *user_message* and return a detailed breakdown for |
| 181 | inspection. Intended for the /rag command — never called during normal |
| 182 | inference. |
| 183 | |
| 184 | Returns a dict with: |
| 185 | query — the cleaned query sent to the search backend |
| 186 | backend — "semantic" | "keyword" | "unavailable" |
| 187 | all_chunks — every result before score filtering, with score + source |
| 188 | kept_chunks — results that passed the score threshold |
| 189 | block — the ## Reference Material string that would be injected |
| 190 | (empty string if nothing passed) |
| 191 | """ |
| 192 | budget_chars = RETRIEVAL_CONFIG.get("budget_chars", 2400) |
| 193 | max_chunks = RETRIEVAL_CONFIG.get("max_chunks", 4) |
| 194 | min_score = RETRIEVAL_CONFIG.get("min_score", 0.0) |
| 195 | use_semantic = RETRIEVAL_CONFIG.get("semantic_search", True) |
| 196 | semantic_threshold = RETRIEVAL_CONFIG.get("semantic_threshold", 0.3) |
| 197 | |
| 198 | query = extract_query(user_message) |
| 199 | if not query.strip(): |
| 200 | return {"query": query, "backend": "none", "all_chunks": [], |
| 201 | "kept_chunks": [], "block": ""} |
| 202 | |
| 203 | all_chunks = [] |
| 204 | backend = "unavailable" |
| 205 | try: |
| 206 | if use_semantic: |
| 207 | from tools.kb_semantic import semantic_search, has_index, keyword_fallback |
| 208 | if has_index(): |
| 209 | all_chunks = semantic_search(query, top_k=max_chunks * 2) |
| 210 | backend = "semantic" |
| 211 | else: |
| 212 | all_chunks = keyword_fallback(query, top_k=max_chunks * 2) |
| 213 | backend = "keyword" |
| 214 | else: |
| 215 | from tools.kb_semantic import keyword_fallback |
| 216 | all_chunks = keyword_fallback(query, top_k=max_chunks * 2) |
| 217 | backend = "keyword" |
| 218 | except Exception as e: |
| 219 | return {"query": query, "backend": "unavailable", "error": str(e), |
| 220 | "all_chunks": [], "kept_chunks": [], "block": ""} |
| 221 | |
| 222 | # Apply the same score filter as retrieve() |
| 223 | kept = list(all_chunks) |
| 224 | if use_semantic and backend == "semantic": |
| 225 | if kept and kept[0].get("score", 0) <= 1.0: |
| 226 | kept = [r for r in kept if r.get("score", 0) >= semantic_threshold] |
| 227 | kept = kept[:max_chunks] |
| 228 | |
| 229 | block = retrieve(user_message, budget_chars=budget_chars) |
| 230 | |
| 231 | return { |
| 232 | "query": query, |
| 233 | "backend": backend, |
| 234 | "threshold": semantic_threshold if backend == "semantic" else min_score, |
| 235 | "all_chunks": all_chunks, |
no test coverage detected