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 sync as _sync 

154 

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

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

157 register_sources(resolved, force=True) 

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

159 

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

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

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

163 remove_documents_durably([name]) 

164 

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

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

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

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

169 return { 

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

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

172 "document_count": len(sources), 

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

174 } 

175 

176 def rebuild(self) -> SyncResult: 

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

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

179 from lilbee.data.ingest import sync as _sync 

180 

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

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

183 

184 def remember( 

185 self, 

186 text: str, 

187 *, 

188 kind: MemoryKind = MemoryKind.FACT, 

189 shared: bool = False, 

190 ) -> str: 

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

192 

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

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

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

196 operate as a pair regardless of the flag. 

197 """ 

198 from lilbee.app.memory import make_memory_row 

199 

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

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

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

203 

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

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

206 from lilbee.data.store import human_recall_predicate 

207 

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

209 return self._services.store.search_memories( 

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

211 owner_predicate=human_recall_predicate(), 

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

213 max_distance=self._config.memory_max_distance, 

214 ) 

215 

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

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

218 from lilbee.data.store import local_owner_predicate 

219 

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

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

222 

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

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

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

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

227 

228 def close(self) -> None: 

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

230 if self._closed: 

231 return 

232 self._closed = True 

233 self._services.provider.shutdown() 

234 self._services.store.close()