diff --git a/docs/decisions/042-pr-100-lazy-init-embedder.md b/docs/decisions/042-pr-100-lazy-init-embedder.md new file mode 100644 index 0000000..c5ffc02 --- /dev/null +++ b/docs/decisions/042-pr-100-lazy-init-embedder.md @@ -0,0 +1,24 @@ +# ADR-042: lazy-init embedder + reindex logging + +## Статус +Accepted (2026-07-26) + +## Контекст +`src/memory/embedder.py` падал с `RuntimeError("OPENAI_BASE_URL env var not set")` при импорте если env var не задана. Любой модуль, импортирующий embedder (`index.py:8`, `search.py:7`), падал ещё до вызова функции. TS tools, вызывающие Python через `spawnSync("python3", [...])`, получали traceback и падали целиком — memory backend был неработоспособен без OpenRouter credentials. + +Логирование reindex отсутствовало: при таймауте/ошибке процесс молча умирал без диагностики (см. `technical/openrouter-qwen3-embedding-index-timeout.md` — 120с процесса timeout при batch=50 × 47 батчей × ~16с = ~750с, процесс молча завершается без вывода). + +Это блокирует миграцию к единой облачной памяти с keyword fallback (issue #94): backend должен работать всегда, деградируя к keyword при отсутствии embeddings, а не падать. + +## Решение +1. **Lazy-init embedder**: `API_URL`/`API_KEY` читаются внутри `embed_texts()` при вызове, не при импорте. При отсутствии `OPENAI_BASE_URL` → `return None` (не raise). При API error после 5 retries → `return None` (catch `httpx.HTTPError`, `RetryError`, `ValueError`, `KeyError`). Сигнатура: `embed_texts(texts: list[str]) -> list[list[float]] | None`. `EMBEDDING_MODEL`/`BATCH_SIZE` остались module-level с безопасными дефолтами (`"gemini-embedding-2-preview"`, `"2048"`) — не падают без env. + +2. **Reindex logging**: `src/memory/index.py` пишет лог в `${output_dir}/reindex.log` (append mode, ISO-8601 UTC timestamps). Формат: `2026-07-26T20:15:30Z start: changed=3 total=82`. Логирует start, per-batch (model, status), done (total, took, changed/unchanged/deleted), failed (embeddings unavailable). Per-batch вызовы `embed_texts` вместо одного bulk-вызова — позволяет логировать прогресс каждого batch. + +3. **None handling в callers**: `search.py` возвращает `[]` если `embed_texts` → None. `index.py` логирует "failed: embeddings unavailable" и выходит с 0 (не пишет index.json). Caller (TS tool) делает fallback на keyword search. + +## Альтернативы +- **`reraise=True` в `@retry`** — tenacity пробрасывает последнюю exception вместо оборачивания в `RetryError`. Отклонено: потребовало бы ловить конкретные `httpx` exceptions по отдельности (`HTTPStatusError`, `ConnectError`, `TimeoutException`), хрупче. Catch `RetryError` + `httpx.HTTPError` покрывает все случаи. +- **Keyword fallback внутри Python (`search.py`)** — если embedder None, делать ripgrep в `search.py`. Отклонено: keyword fallback уже реализован в TS tool (plugin scoring logic), дублирование не нужно. Python возвращает `[]` — TS tool решает fallback. +- **Лог в memory root (`reindex.log` рядом с `.md`)** — отклонено: `.rag/` уже gitignored в памяти, лог рядом с memory files засоряет директорию и попадает в индексацию. Лог в `.rag/` изолирован. +- **`logging` module вместо ручного `_log_line`** — отклонено: `logging` требует config (handlers, formatters), избыточно для одного файла. `_log_line` = 4 строки, append mode, явный timestamp. \ No newline at end of file diff --git a/docs/handoff/pr-100-lazy-init-embedder.md b/docs/handoff/pr-100-lazy-init-embedder.md new file mode 100644 index 0000000..e48a1e6 --- /dev/null +++ b/docs/handoff/pr-100-lazy-init-embedder.md @@ -0,0 +1,32 @@ +--- +pr: 100 +title: refactor(memory): lazy-init embedder + reindex logging +--- + +## Что сделано +Рефактор Python memory backend чтобы он был безопасен при отсутствии env vars и логировал reindex. + +- `src/memory/embedder.py` — `API_URL`/`API_KEY` перенесены внутрь `embed_texts()` (lazy-init). `EMBEDDING_MODEL`/`BATCH_SIZE` остались module-level с безопасными дефолтами. При отсутствии `OPENAI_BASE_URL` → `return None` (не raise). При API error после 5 retries → `return None` (catch `httpx.HTTPError`, `RetryError`, `ValueError`, `KeyError`). Сигнатура: `embed_texts(texts: list[str]) -> list[list[float]] | None`. Reuse логики retries (tenacity) и batch сохранён. +- `src/memory/index.py` — добавлено логирование reindex в `${output_dir}/reindex.log` (append mode, ISO-8601 timestamps UTC). Логирует: start (changed/total), per-batch (model, status=ok/error), done (total, took, changed/unchanged/deleted), failed (embeddings unavailable). Per-batch вызовы `embed_texts` вместо одного bulk-вызова — позволяет логировать прогресс каждого batch. При `embed_texts` → None: логирует "failed: embeddings unavailable" и выходит с 0 (не падает, не пишет index.json). Helper `_log_line()` + `_embed_in_batches()`. +- `src/memory/search.py` — проверка `None` от `embed_texts`: возвращает `[]` (не падает на `[0]` index). Caller (TS tool) делает fallback на keyword. +- `tests/test_embedder.py` — `test_embed_texts_api_error` обновлён: ожидает `None` вместо `pytest.raises(HTTPStatusError)`. Добавлены `test_embed_texts_no_env_returns_none` и `test_embed_texts_empty_input`. +- `tests/test_search.py` — добавлен `test_search_returns_empty_when_embedder_none`: embedder возвращает None → search возвращает `[]`. +- `tests/test_index.py` — добавлены `test_index_writes_reindex_log` (проверяет наличие `.rag/reindex.log` со строками start/done/total) и `test_index_embedder_none_logs_and_exits_zero` (проверяет лог "failed: embeddings unavailable" + отсутствие index.json + сообщение в stdout). + +## Почему +`src/memory/embedder.py` падал с `RuntimeError("OPENAI_BASE_URL env var not set")` при импорте если env не задан. Из-за этого любой модуль, импортирующий embedder (`index.py`, `search.py`), падал ещё до вызова. TS tools, вызывающие Python через `spawnSync`, получали traceback и падали целиком. Это блокировало работу memory при отсутствии OpenRouter credentials. + +Логирование reindex отсутствовало — при таймауте/ошибке (см. ADR-024, PR#57; `technical/openrouter-qwen3-embedding-index-timeout.md` — 120с процесса timeout при batch=50 × 47 батчей × ~16с = ~750с) процесс молча умирал без диагностики. Лог в `.rag/reindex.log` (gitignored в памяти) даёт observable trail для дебага. + +Решение возвращает `None` (не raises) чтобы caller мог сделать fallback на keyword search — это часть миграции к единой облачной памяти с keyword fallback (issue #94, ADR про keyword-vs-semantic decision в `technical/memory-keyword-vs-semantic-decision.md`). + +## Pending +— Frontmatter `pr: ` — будет заполнен номером PR после `create-pr` отдельным коммитом `docs(handoff): set PR number`. + +## Watch out +- **Сигнатура `embed_texts` изменилась**: `list[list[float]]` → `list[list[float]] | None`. Все callers должны проверять `None`. В этом PR обновлены `index.py` и `search.py`. Внешних callers в репо нет (TS tools вызывают через subprocess, парсят JSON). +- **Per-batch вызовы `embed_texts`** — `_embed_in_batches` в index.py вызывает `embed_texts(batch)` в цикле вместо одного `embed_texts(all_texts)`. Каждый batch создаёт свой `httpx.Client` (внутри `embed_texts`). Для ~47 батчей это 47 TCP-соединений вместо 1. Приемлемо для offline reindex, но если потребуется оптимизация — можно вынести client наружу. Сохраняет retry-логику per-batch. +- **`RetryError` catch** — tenacity оборачивает последнюю exception в `RetryError` после исчерпания попыток. Ловим его чтобы вернуть `None`. Альтернатива — `reraise=True` в `@retry`, но это потребовало бы ловить конкретные `httpx` exceptions по отдельности. +- **`from datetime import UTC`** (не `datetime.UTC`) — `datetime.UTC` доступен только в 3.11+ как атрибут модуля, не класса. Ruff UP017 предлагает `datetime.UTC` но это требует `import datetime` (модуль). Использован `from datetime import UTC, datetime` — работает в 3.12+ (target-version). +- **Лог-файл в `.rag/`** — `reindex.log` пишется в `output_dir` (аргумент `--output`, обычно `.rag/`), НЕ в memory root. `.rag/` gitignored в памяти. Append mode — лог растёт со временем, не очищается автоматически. При необходимости добавить rotation. +- **`test_embed_texts_default_model`/`test_embed_texts_custom_model`/`test_embed_texts_trailing_slash`** используют `importlib.reload(embedder_mod)` — после рефактора reload пересчитывает `EMBEDDING_MODEL`/`BATCH_SIZE` (module-level), но `API_URL`/`API_KEY` больше не module-level. Тесты работают: env меняется через monkeypatch → `embed_texts` читает при вызове. Reload избыточен но безвреден. \ No newline at end of file diff --git a/src/memory/embedder.py b/src/memory/embedder.py index dd551f1..de9b23e 100644 --- a/src/memory/embedder.py +++ b/src/memory/embedder.py @@ -3,13 +3,8 @@ import os import time import httpx -from tenacity import retry, retry_if_exception, stop_after_attempt, wait_exponential +from tenacity import RetryError, retry, retry_if_exception, stop_after_attempt, wait_exponential -API_URL = os.environ.get("OPENAI_BASE_URL") -if not API_URL: - raise RuntimeError("OPENAI_BASE_URL env var not set") -API_URL = API_URL.rstrip("/") -API_KEY = os.environ.get("OPENAI_API_KEY", "") EMBEDDING_MODEL = os.environ.get("OPENAI_EMBEDDING_MODEL", "gemini-embedding-2-preview") BATCH_SIZE = int(os.environ.get("OPENAI_EMBEDDING_BATCH_SIZE", "2048")) @@ -45,23 +40,32 @@ def _call_embedding_api( return [d["embedding"] for d in data["data"]] -def embed_texts(texts: list[str]) -> list[list[float]]: +def embed_texts(texts: list[str]) -> list[list[float]] | None: if not texts: return [] - url = f"{API_URL}/embeddings" - headers = {"Authorization": f"Bearer {API_KEY}"} + api_url = os.environ.get("OPENAI_BASE_URL") + if not api_url: + return None + api_url = api_url.rstrip("/") + api_key = os.environ.get("OPENAI_API_KEY", "") - with httpx.Client(timeout=120.0) as client: - if len(texts) <= BATCH_SIZE: - payload: dict[str, str | list[str]] = {"model": EMBEDDING_MODEL, "input": texts} - return _call_embedding_api(client, url, headers, payload) + url = f"{api_url}/embeddings" + headers = {"Authorization": f"Bearer {api_key}"} - n_batches = (len(texts) + BATCH_SIZE - 1) // BATCH_SIZE - results: list[list[float]] = [] - for i in range(0, len(texts), BATCH_SIZE): - batch = texts[i : i + BATCH_SIZE] - payload = {"model": EMBEDDING_MODEL, "input": batch} - print(f" batch {i // BATCH_SIZE + 1}/{n_batches}...", flush=True) - results.extend(_call_embedding_api(client, url, headers, payload)) - return results + try: + with httpx.Client(timeout=120.0) as client: + if len(texts) <= BATCH_SIZE: + payload: dict[str, str | list[str]] = {"model": EMBEDDING_MODEL, "input": texts} + return _call_embedding_api(client, url, headers, payload) + + n_batches = (len(texts) + BATCH_SIZE - 1) // BATCH_SIZE + results: list[list[float]] = [] + for i in range(0, len(texts), BATCH_SIZE): + batch = texts[i : i + BATCH_SIZE] + payload = {"model": EMBEDDING_MODEL, "input": batch} + print(f" batch {i // BATCH_SIZE + 1}/{n_batches}...", flush=True) + results.extend(_call_embedding_api(client, url, headers, payload)) + return results + except (httpx.HTTPError, RetryError, ValueError, KeyError): + return None diff --git a/src/memory/index.py b/src/memory/index.py index 07fc9f7..6e48842 100644 --- a/src/memory/index.py +++ b/src/memory/index.py @@ -3,6 +3,8 @@ import fcntl import hashlib import json import os +import time +from datetime import UTC, datetime from pathlib import Path from src.memory.embedder import BATCH_SIZE, EMBEDDING_MODEL, embed_texts @@ -10,6 +12,7 @@ from src.memory.embedder import BATCH_SIZE, EMBEDDING_MODEL, embed_texts INDEX_FILENAME = "index.json" META_FILENAME = "meta.json" LOCK_FILENAME = ".lock" +LOG_FILENAME = "reindex.log" FileEntry = dict[str, str | int] IndexedEntry = dict[str, str | int | list[float]] @@ -52,6 +55,12 @@ def _atomic_write(path: Path, content: str) -> None: os.replace(tmp, path) +def _log_line(log_path: Path, message: str) -> None: + ts = datetime.now(tz=UTC).strftime("%Y-%m-%dT%H:%M:%SZ") + with open(log_path, "a", encoding="utf-8") as f: + f.write(f"{ts} {message}\n") + + def _load_meta(meta_path: Path) -> Meta: if not meta_path.exists(): return {} @@ -201,6 +210,27 @@ def run_index(args: argparse.Namespace) -> None: ) +def _embed_in_batches( + texts: list[str], + log_path: Path, +) -> list[list[float]] | None: + n_batches = (len(texts) + BATCH_SIZE - 1) // BATCH_SIZE + results: list[list[float]] = [] + for batch_idx, i in enumerate(range(0, len(texts), BATCH_SIZE), start=1): + batch = texts[i : i + BATCH_SIZE] + _log_line(log_path, f"batch {batch_idx}/{n_batches}, model={EMBEDDING_MODEL}, start") + chunk = embed_texts(batch) + if chunk is None: + _log_line( + log_path, + f"batch {batch_idx}/{n_batches}, model={EMBEDDING_MODEL}, status=error", + ) + return None + _log_line(log_path, f"batch {batch_idx}/{n_batches}, model={EMBEDDING_MODEL}, status=ok") + results.extend(chunk) + return results + + def _run_index_locked( *, output_dir: Path, @@ -208,6 +238,8 @@ def _run_index_locked( memory_dir: Path, chunk_config: tuple[int, int], ) -> None: + log_path = output_dir / LOG_FILENAME + start_ts = time.monotonic() chunk_size, chunk_overlap = chunk_config current_version = _current_version(chunk_size, chunk_overlap) index_path = output_dir / INDEX_FILENAME @@ -227,10 +259,16 @@ def _run_index_locked( ) deleted_files = [rel for rel in old_files_meta if rel not in file_map] + _log_line( + log_path, + f"start: changed={len(changed_files)} total={len(md_files)}", + ) + old_entries_by_source = _load_old_entries_by_source(index_path) kept_entries = _collect_kept_entries(unchanged_files, old_entries_by_source) if not changed_files and not deleted_files: + _log_line(log_path, f"done: changed=0 total={len(md_files)} took=0s noop") print(f"No changes detected ({len(md_files)} files, {len(unchanged_files)} unchanged)") return @@ -239,9 +277,18 @@ def _run_index_locked( ) if changed_files: - n_batches = (len(new_texts) + BATCH_SIZE - 1) // BATCH_SIZE - print(f"Embedding {len(new_texts)} chunks in {n_batches} batches...", flush=True) - new_embeddings = embed_texts(new_texts) + print(f"Embedding {len(new_texts)} chunks...", flush=True) + new_embeddings = _embed_in_batches(new_texts, log_path) + if new_embeddings is None: + _log_line( + log_path, + "failed: embeddings unavailable (OPENAI_BASE_URL not set or API error)", + ) + print( + "embeddings unavailable (OPENAI_BASE_URL not set or API error), " + "skipping semantic index" + ) + return else: new_embeddings = [] @@ -254,6 +301,13 @@ def _run_index_locked( json.dumps({"version": current_version, "files": full_meta}, ensure_ascii=False), ) + elapsed = int(time.monotonic() - start_ts) + _log_line( + log_path, + f"done: total={len(md_files)} took={elapsed}s " + f"changed={len(changed_files)} unchanged={len(unchanged_files)} " + f"deleted={len(deleted_files)}", + ) print( f"Indexed {len(md_files)} files to {index_path} " f"({len(changed_files)} changed, {len(unchanged_files)} unchanged, " diff --git a/src/memory/search.py b/src/memory/search.py index cc77810..0cafc65 100644 --- a/src/memory/search.py +++ b/src/memory/search.py @@ -15,29 +15,33 @@ def _cosine_sim(a: np.ndarray, b: np.ndarray) -> float: return float(np.dot(a, b) / (norm_a * norm_b)) -def run_search(args: argparse.Namespace) -> None: - index_path = Path(args.index_dir) / "index.json" +def _load_index(index_path: Path) -> list[dict[str, object]]: if not index_path.exists(): - json.dump([], sys.stdout) - return - + return [] index = json.loads(index_path.read_text(encoding="utf-8")) - files = index.get("files", []) - if not files: - json.dump([], sys.stdout) - return + files = index.get("files", []) if isinstance(index, dict) else [] + return files if isinstance(files, list) else [] - query_emb = embed_texts([args.query])[0] - query_vec = np.array(query_emb) +def _rank_by_similarity( + query_vec: np.ndarray, files: list[dict[str, object]] +) -> list[dict[str, str | float]]: results: list[dict[str, str | float]] = [] for f in files: file_vec = np.array(f["embedding"]) sim = _cosine_sim(query_vec, file_vec) - results.append({"source": f["source"], "score": sim, "text": f["text"]}) - + results.append( + { + "source": str(f["source"]), + "score": sim, + "text": str(f["text"]), + } + ) results.sort(key=lambda r: r["score"], reverse=True) + return results + +def _dedup_top(results: list[dict[str, str | float]], k: int) -> list[dict[str, str | float]]: seen: set[str] = set() top: list[dict[str, str | float]] = [] for r in results: @@ -46,7 +50,24 @@ def run_search(args: argparse.Namespace) -> None: continue seen.add(src) top.append(r) - if len(top) >= args.k: + if len(top) >= k: break + return top + +def run_search(args: argparse.Namespace) -> None: + index_path = Path(args.index_dir) / "index.json" + files = _load_index(index_path) + if not files: + json.dump([], sys.stdout) + return + + query_emb = embed_texts([args.query]) + if query_emb is None: + json.dump([], sys.stdout) + return + query_vec = np.array(query_emb[0]) + + results = _rank_by_similarity(query_vec, files) + top = _dedup_top(results, args.k) print(json.dumps(top, ensure_ascii=False)) diff --git a/tests/test_embedder.py b/tests/test_embedder.py index b652803..0f69553 100644 --- a/tests/test_embedder.py +++ b/tests/test_embedder.py @@ -60,8 +60,19 @@ def test_embed_texts_api_error() -> None: def mock_post(self, url, **kwargs): return FakeErrorResponse() - with patch.object(httpx.Client, "post", mock_post), pytest.raises(httpx.HTTPStatusError): - embed_texts(["test"]) + with patch.object(httpx.Client, "post", mock_post): + result = embed_texts(["test"]) + + assert result is None + + +def test_embed_texts_no_env_returns_none(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("OPENAI_BASE_URL", raising=False) + assert embed_texts(["test"]) is None + + +def test_embed_texts_empty_input() -> None: + assert embed_texts([]) == [] def test_embed_texts_default_model(monkeypatch: pytest.MonkeyPatch) -> None: diff --git a/tests/test_index.py b/tests/test_index.py index cc64624..2cd430c 100644 --- a/tests/test_index.py +++ b/tests/test_index.py @@ -337,6 +337,44 @@ class TestIncrementalIndex: meta2 = json.loads((index_dir / "meta.json").read_text()) assert meta2["version"] != "old-model:512:64" + def test_index_writes_reindex_log(self, tmp_path: Path) -> None: + memory_dir = tmp_path / "memory" + index_dir = tmp_path / ".rag" + memory_dir.mkdir(parents=True) + (memory_dir / "a.md").write_text("file A") + (memory_dir / "b.md").write_text("file B") + + with patch("src.memory.index.embed_texts", self._fake_embed()): + run_index(self._make_args(memory_dir, index_dir)) + + log_path = index_dir / "reindex.log" + assert log_path.exists() + log_content = log_path.read_text(encoding="utf-8") + assert "start:" in log_content + assert "done:" in log_content + assert "total=2" in log_content + + def test_index_embedder_none_logs_and_exits_zero(self, tmp_path: Path, capsys) -> None: + memory_dir = tmp_path / "memory" + index_dir = tmp_path / ".rag" + memory_dir.mkdir(parents=True) + (memory_dir / "a.md").write_text("file A") + + def none_embed(texts): + return None + + with patch("src.memory.index.embed_texts", none_embed): + run_index(self._make_args(memory_dir, index_dir)) + + log_path = index_dir / "reindex.log" + log_content = log_path.read_text(encoding="utf-8") + assert "failed:" in log_content + assert "embeddings unavailable" in log_content + + captured = capsys.readouterr() + assert "embeddings unavailable" in captured.out + assert not (index_dir / "index.json").exists() + class TestAtomicWrite: def test_atomic_write_replaces(self, tmp_path: Path) -> None: diff --git a/tests/test_search.py b/tests/test_search.py index 276aa72..5716632 100644 --- a/tests/test_search.py +++ b/tests/test_search.py @@ -101,3 +101,28 @@ class TestSearchOutput: assert sources.count("doc.md") == 1 doc_entry = next(r for r in result if r["source"] == "doc.md") assert doc_entry["text"] == "chunk 0" + + def test_search_returns_empty_when_embedder_none(self, tmp_path: Path, capsys) -> None: + index_dir = tmp_path / ".rag" + index_dir.mkdir() + index = { + "files": [ + { + "source": "test.md", + "text": "hello world", + "embedding": [1.0, 0.0, 0.0], + }, + ], + } + (index_dir / "index.json").write_text(json.dumps(index)) + + def fake_embed_texts(texts): + return None + + with patch.object(search_mod, "embed_texts", fake_embed_texts): + args = Namespace(index_dir=str(index_dir), query="hello", k=5, json=True) + search_mod.run_search(args) + + captured = capsys.readouterr() + result = json.loads(captured.out) + assert result == []