Coverage for src/lilbee/api.py: 100%

1"""Programmatic access to lilbee's retrieval pipeline. 

2 

3Retrieval only -- no LLM chat. Search your indexed documents from Python. 

4Optional features (concept graph, reranker) activate automatically when 

5their dependencies are installed. 

6 

7Usage:: 

8 

9 from lilbee import Lilbee 

10 

11 bee = Lilbee("./docs") 

12 bee.sync() 

13 results = bee.search("authentication") 

14 bee.close() 

15 

16Each instance binds its own Config and Services for the duration of every call 

17via contextvar scopes, so it runs against its own data root without mutating the 

18process-global cfg or the shared services singleton. The scope is per task, so 

19two instances may be driven from different threads or asyncio tasks without 

20clobbering one another, and the global HTTP daemon's fleet is untouched. An 

21instance holds a live engine between calls; call :meth:`Lilbee.close` to release 

22it. 

23""" 

24 

25from __future__ import annotations 

26 

27import asyncio 

28from pathlib import Path 

29from typing import TYPE_CHECKING 

30 

31# app.ingest stays at module top: registering source roots is a locked 

32# config.toml read-modify-write over the config singleton (no copy, no symlink), 

33# cheap to import. data.ingest is deferred at each callsite below because it 

34# transitively imports spaCy via the wiki package and adds ~3s on first touch. 

35from lilbee.app.ingest import register_sources, remove_documents_durably 

36from lilbee.app.services import build_services, services_scope 

37from lilbee.core.config import Config, cfg, config_scope 

38from lilbee.core.system import canonical_data_root 

39from lilbee.data.store import LOCAL_OWNER, MemoryKind, MemoryRow, SearchScope, scope_to_chunk_type 

40 

41if TYPE_CHECKING: 

42 from lilbee.app.services import Services 

43 from lilbee.data.ingest import SyncResult 

44 from lilbee.data.store import SearchChunk, Store 

45 from lilbee.providers.base import LLMProvider 

46 from lilbee.retrieval.embedder import Embedder 

47 from lilbee.retrieval.query import Searcher 

48 

49 

50class Lilbee: 

51 """Programmatic access to lilbee's retrieval pipeline. 

52 

53 Usage:: 

54 

55 from lilbee import Lilbee 

56 

57 bee = Lilbee("./docs") 

58 bee.sync() 

59 results = bee.search("authentication") 

60 

61 Retrieval only. Wiki building and browsing stay off this surface by design: 

62 they are interactive, long-running operations that belong to the CLI, TUI, 

63 HTTP API, and MCP tools. ``search(scope=...)`` is the library's whole wiki 

64 story, since scoping a query is retrieval. 

65 """ 

66 

67 def __init__( 

68 self, 

69 documents_dir: str | Path | None = None, 

70 *, 

71 config: Config | None = None, 

72 provider: LLMProvider | None = None, 

73 ) -> None: 

74 """Create a lilbee instance. 

75 Args: 

76 documents_dir: Path to documents folder. Creates a default Config 

77 with derived data and lancedb directories. 

78 config: Full Config instance for complete control. 

79 provider: LLM provider instance. If not given, creates one from config. 

80 

81 Pass documents_dir or config, not both. If neither is given, uses 

82 ``Config()`` (same defaults as the CLI). 

83 """ 

84 if documents_dir is not None and config is not None: 

85 raise ValueError("Pass documents_dir or config, not both") 

86 

87 if config is not None: 

88 self._config = config 

89 elif documents_dir is not None: 

90 root = canonical_data_root(documents_dir) 

91 self._config = cfg.model_copy( 

92 update={ 

93 "data_root": root, 

94 "documents_dir": root / "documents", 

95 "data_dir": root / "data", 

96 "lancedb_dir": root / "data" / "lancedb", 

97 }, 

98 ) 

99 else: 

100 self._config = Config() 

101 

102 self._config.documents_dir.mkdir(parents=True, exist_ok=True) 

103 self._config.data_dir.mkdir(parents=True, exist_ok=True) 

104 

105 self._services: Services = build_services(self._config, provider=provider) 

106 self._closed = False 

107 

108 @property 

109 def config(self) -> Config: 

110 """The Config instance backing this Lilbee.""" 

111 return self._config 

112 

113 @property 

114 def store(self) -> Store: 

115 """The Store component.""" 

116 return self._services.store 

117 

118 @property 

119 def embedder(self) -> Embedder: 

120 """The Embedder component.""" 

121 return self._services.embedder 

122 

123 @property 

124 def searcher(self) -> Searcher: 

125 """The Searcher component.""" 

126 return self._services.searcher 

127 

128 def sync(self, *, quiet: bool = True) -> SyncResult: 

129 """Sync documents to the vector store. Returns what changed.""" 

130 # heavy: data.ingest transitively imports spaCy via wiki 

131 from lilbee.data.ingest import sync as _sync 

132 

133 with config_scope(self._config), services_scope(self._services): 

134 return asyncio.run(_sync(quiet=quiet)) 

135 

136 def search( 

137 self, query: str, *, top_k: int = 0, scope: SearchScope = SearchScope.BOTH 

138 ) -> list[SearchChunk]: 

139 """Search indexed documents. Returns ranked chunks. 

140 

141 ``scope`` selects raw document chunks, generated wiki chunks, or both. 

142 """ 

143 with config_scope(self._config), services_scope(self._services): 

144 return self._services.searcher.search( 

145 query, top_k=top_k, chunk_type=scope_to_chunk_type(scope) 

146 ) 

147 

148 def add(self, paths: list[str | Path]) -> SyncResult: 

149 """Add files to the knowledge base and sync. 

150 Registers each path as a source root (indexed in place), then syncs. 

151 """ 

152 # heavy: data.ingest transitively imports spaCy via wiki 

153 from lilbee.data.ingest import SyncResult 

154 from lilbee.data.ingest import sync as _sync 

155 

156 resolved = [Path(p).resolve() for p in paths] 

157 with config_scope(self._config), services_scope(self._services): 

158 if not register_sources(resolved, force=True).reached_corpus: 

159 return SyncResult() 

160 return asyncio.run(_sync(quiet=True)) 

161 

162 def remove(self, name: str) -> None: 

163 """Remove a document from the index by source name (source bytes are kept).""" 

164 with config_scope(self._config), services_scope(self._services): 

165 remove_documents_durably([name]) 

166 

167 def status(self) -> dict[str, object]: 

168 """Return index stats (document count, data directory, etc.).""" 

169 with config_scope(self._config), services_scope(self._services): 

170 sources = self._services.store.get_sources() 

171 return { 

172 "documents_dir": str(self._config.documents_dir), 

173 "data_dir": str(self._config.data_dir), 

174 "document_count": len(sources), 

175 "sources": [s["filename"] for s in sources], 

176 } 

177 

178 def rebuild(self) -> SyncResult: 

179 """Rebuild the entire index from scratch.""" 

180 # heavy: data.ingest transitively imports spaCy via wiki 

181 from lilbee.data.ingest import sync as _sync 

182 

183 with config_scope(self._config), services_scope(self._services): 

184 return asyncio.run(_sync(force_rebuild=True, quiet=True)) 

185 

186 def remember( 

187 self, 

188 text: str, 

189 *, 

190 kind: MemoryKind = MemoryKind.FACT, 

191 shared: bool = False, 

192 ) -> str: 

193 """Store a fact or preference in long-term memory; returns its id. 

194 

195 This library primitive does not consult ``memory_enabled``: that flag 

196 gates the interactive surfaces (TUI/CLI/MCP/REST) and the chat-prompt 

197 injection, not direct programmatic access. ``remember`` and ``recall`` 

198 operate as a pair regardless of the flag. 

199 """ 

200 from lilbee.app.memory import make_memory_row 

201 

202 with config_scope(self._config), services_scope(self._services): 

203 record = make_memory_row(text, self._services.embedder.embed, kind=kind, shared=shared) 

204 return self._services.store.add_memory(record) 

205 

206 def recall(self, query: str, *, top_k: int | None = None) -> list[MemoryRow]: 

207 """Recall facts relevant to *query* (own memories plus agent-shared).""" 

208 from lilbee.data.store import human_recall_predicate 

209 

210 with config_scope(self._config), services_scope(self._services): 

211 return self._services.store.search_memories( 

212 self._services.embedder.embed_query(query), 

213 owner_predicate=human_recall_predicate(), 

214 top_k=self._config.memory_top_k if top_k is None else top_k, 

215 max_distance=self._config.memory_max_distance, 

216 ) 

217 

218 def memories(self) -> list[MemoryRow]: 

219 """List all stored memories, newest first.""" 

220 from lilbee.data.store import local_owner_predicate 

221 

222 with config_scope(self._config), services_scope(self._services): 

223 return self._services.store.get_memories(owner_predicate=local_owner_predicate()) 

224 

225 def forget(self, memory_id: str) -> bool: 

226 """Delete a local memory by id; True when it existed and was removed.""" 

227 with config_scope(self._config), services_scope(self._services): 

228 return self._services.store.delete_memory(memory_id, owner=LOCAL_OWNER) 

229 

230 def close(self) -> None: 

231 """Release the engine and store this instance holds. Idempotent.""" 

232 if self._closed: 

233 return 

234 self._closed = True 

235 self._services.provider.shutdown() 

236 self._services.store.close()