Coverage for src / lilbee / cli / commands / search_chat.py: 100%
145 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-05-15 20:55 +0000
1"""Search, ask, chat, and topics commands."""
3from __future__ import annotations
5import sys
6from pathlib import Path
8import typer
9from rich.table import Table
11from lilbee.app.search import clean_result
12from lilbee.app.services import get_services
13from lilbee.cli import theme
14from lilbee.cli.app import (
15 apply_overrides,
16 console,
17 data_dir_option,
18 global_option,
19 model_option,
20 num_ctx_option,
21 repeat_penalty_option,
22 seed_option,
23 temperature_option,
24 top_k_sampling_option,
25 top_p_option,
26)
27from lilbee.cli.commands._shared import CHUNK_PREVIEW_LEN
28from lilbee.cli.helpers import (
29 auto_sync,
30 json_output,
31)
32from lilbee.core.config import cfg
33from lilbee.data.store import SearchScope, scope_to_chunk_type
34from lilbee.providers.base import ProviderError
36# How many top concepts to show inline before truncating with a ``+N more`` tail.
37_TOPIC_PREVIEW_LIMIT = 5
39_scope_option = typer.Option(
40 SearchScope.BOTH,
41 "--scope",
42 "-s",
43 help="Restrict the pool to raw chunks, wiki pages, or both (default).",
44 case_sensitive=False,
45)
48def search(
49 query: str = typer.Argument(..., help="Search query"),
50 top_k: int = typer.Option(None, "--top-k", "-k", help="Number of results"),
51 scope: SearchScope = _scope_option,
52 data_dir: Path | None = data_dir_option,
53 use_global: bool = global_option,
54) -> None:
55 """Search the knowledge base for relevant chunks."""
56 apply_overrides(data_dir=data_dir, use_global=use_global)
58 if not query or not query.strip():
59 if cfg.json_mode:
60 json_output({"error": "query must not be empty"})
61 raise SystemExit(1)
62 console.print(f"[{theme.ERROR}]Error:[/{theme.ERROR}] query must not be empty")
63 raise SystemExit(1)
65 try:
66 results = get_services().searcher.search(
67 query,
68 top_k=top_k or cfg.top_k,
69 chunk_type=scope_to_chunk_type(scope),
70 )
71 except Exception as exc:
72 if cfg.json_mode:
73 json_output({"error": str(exc)})
74 raise SystemExit(1) from None
75 console.print(f"[{theme.ERROR}]Error:[/{theme.ERROR}] {exc}")
76 raise SystemExit(1) from None
77 cleaned = [clean_result(r) for r in results]
79 if cfg.json_mode:
80 json_output({"command": "search", "query": query, "results": cleaned})
81 return
83 if not cleaned:
84 console.print("No results found.")
85 return
87 has_relevance = any("relevance_score" in r for r in cleaned)
88 table = Table(title="Search Results")
89 table.add_column("Source", style=theme.ACCENT)
90 table.add_column("Chunk", max_width=80)
91 score_label = "Score" if has_relevance else "Distance"
92 table.add_column(score_label, justify="right", style=theme.MUTED)
94 for r in cleaned:
95 chunk_text = r["chunk"]
96 preview = chunk_text[:CHUNK_PREVIEW_LEN]
97 if len(chunk_text) > CHUNK_PREVIEW_LEN:
98 preview += "..."
99 score = r.get("relevance_score") or r.get("distance") or 0
100 table.add_row(r["source"], preview, f"{score:.4f}")
101 console.print(table)
104def ask(
105 question: str = typer.Argument(..., help="Question to ask"),
106 scope: SearchScope = _scope_option,
107 data_dir: Path | None = data_dir_option,
108 model: str | None = model_option,
109 use_global: bool = global_option,
110 temperature: float | None = temperature_option,
111 top_p: float | None = top_p_option,
112 top_k_sampling: int | None = top_k_sampling_option,
113 repeat_penalty: float | None = repeat_penalty_option,
114 num_ctx: int | None = num_ctx_option,
115 seed: int | None = seed_option,
116) -> None:
117 """Ask a one-shot question (auto-syncs first)."""
118 apply_overrides(
119 data_dir=data_dir,
120 model=model,
121 use_global=use_global,
122 temperature=temperature,
123 top_p=top_p,
124 top_k_sampling=top_k_sampling,
125 repeat_penalty=repeat_penalty,
126 num_ctx=num_ctx,
127 seed=seed,
128 )
130 try:
131 from lilbee.modelhub.models import ensure_chat_model
133 ensure_chat_model()
134 get_services().embedder.validate_model()
135 if cfg.json_mode:
136 from rich.console import Console as _QuietConsole
138 auto_sync(_QuietConsole(quiet=True))
139 else:
140 auto_sync(console)
142 chunk_type = scope_to_chunk_type(scope)
144 if cfg.json_mode:
145 result = get_services().searcher.ask_raw(question, chunk_type=chunk_type)
146 json_output(
147 {
148 "command": "ask",
149 "question": question,
150 "answer": result.answer,
151 "sources": [clean_result(s) for s in result.sources],
152 }
153 )
154 return
156 for token in get_services().searcher.ask_stream(question, chunk_type=chunk_type):
157 console.print(token.content, end="")
158 console.print()
159 except (RuntimeError, ProviderError) as exc:
160 if cfg.json_mode:
161 json_output({"error": str(exc)})
162 raise SystemExit(1) from None
163 console.print(f"[{theme.ERROR}]Error:[/{theme.ERROR}] {exc}")
164 raise SystemExit(1) from None
167def chat(
168 data_dir: Path | None = data_dir_option,
169 model: str | None = model_option,
170 use_global: bool = global_option,
171 temperature: float | None = temperature_option,
172 top_p: float | None = top_p_option,
173 top_k_sampling: int | None = top_k_sampling_option,
174 repeat_penalty: float | None = repeat_penalty_option,
175 num_ctx: int | None = num_ctx_option,
176 seed: int | None = seed_option,
177) -> None:
178 """Interactive chat loop. Press S in the TUI to sync pending documents."""
179 apply_overrides(
180 data_dir=data_dir,
181 model=model,
182 use_global=use_global,
183 temperature=temperature,
184 top_p=top_p,
185 top_k_sampling=top_k_sampling,
186 repeat_penalty=repeat_penalty,
187 num_ctx=num_ctx,
188 seed=seed,
189 )
191 if cfg.json_mode:
192 json_output({"error": "Chat requires a terminal, not --json"})
193 raise SystemExit(1)
194 if not sys.stdin.isatty() or not sys.stdout.isatty():
195 console.print(f"[{theme.ERROR}]Error:[/{theme.ERROR}] Chat requires a terminal.")
196 raise SystemExit(1)
197 from lilbee.cli.tui import run_tui
199 run_tui()
202def topics(
203 query: str = typer.Argument(None, help="Optional query to find related concepts."),
204 top_k: int = typer.Option(10, "--top-k", "-k", help="Number of results."),
205 data_dir: Path | None = data_dir_option,
206 use_global: bool = global_option,
207) -> None:
208 """Show top concept communities or concepts related to a query."""
209 apply_overrides(data_dir=data_dir, use_global=use_global)
211 from lilbee.retrieval.concepts import concepts_available
213 if not concepts_available():
214 msg = "Concept graph requires: pip install 'lilbee[graph]'"
215 if cfg.json_mode:
216 json_output({"error": msg})
217 raise SystemExit(1)
218 console.print(f"[{theme.ERROR}]{msg}[/{theme.ERROR}]")
219 raise SystemExit(1)
221 if not cfg.concept_graph:
222 if cfg.json_mode:
223 json_output({"error": "Concept graph is disabled (LILBEE_CONCEPT_GRAPH=false)"})
224 raise SystemExit(1)
225 console.print(
226 f"[{theme.ERROR}]Concept graph is disabled.[/{theme.ERROR}] "
227 "Enable with LILBEE_CONCEPT_GRAPH=true"
228 )
229 raise SystemExit(1)
231 if not get_services().concepts.get_graph():
232 if cfg.json_mode:
233 json_output({"error": "Concept graph not available"})
234 raise SystemExit(1)
235 console.print(f"[{theme.ERROR}]Concept graph not available.[/{theme.ERROR}]")
236 raise SystemExit(1)
238 if query:
239 _topics_for_query(query)
240 else:
241 _topics_overview(top_k)
244def _topics_for_query(query: str) -> None:
245 """Show concepts related to a query."""
246 cg = get_services().concepts
247 concepts = cg.extract_concepts(query)
248 related = cg.expand_query(query)
249 all_concepts = concepts + [r for r in related if r not in concepts]
251 if cfg.json_mode:
252 json_output({"command": "topics", "query": query, "concepts": all_concepts})
253 return
254 if not all_concepts:
255 console.print("No concepts found for this query.")
256 return
257 console.print(f"Concepts related to [{theme.ACCENT}]{query}[/{theme.ACCENT}]:")
258 for c in all_concepts:
259 console.print(f" {c}")
262def _topics_overview(top_k: int) -> None:
263 """Show top concept communities."""
264 from dataclasses import asdict
266 communities = get_services().concepts.top_communities(k=top_k)
267 if cfg.json_mode:
268 json_output({"command": "topics", "communities": [asdict(c) for c in communities]})
269 return
270 if not communities:
271 console.print("No concept communities found. Try syncing some documents first.")
272 return
273 table = Table(title="Concept Communities")
274 table.add_column("Cluster", justify="right", style=theme.MUTED)
275 table.add_column("Size", justify="right")
276 table.add_column("Top Concepts", style=theme.ACCENT)
277 for comm in communities:
278 preview = ", ".join(comm.concepts[:_TOPIC_PREVIEW_LIMIT])
279 if len(comm.concepts) > _TOPIC_PREVIEW_LIMIT:
280 preview += f" (+{len(comm.concepts) - _TOPIC_PREVIEW_LIMIT} more)"
281 table.add_row(str(comm.cluster_id), str(comm.size), preview)
282 console.print(table)