refactor(memory): lazy-init embedder + reindex logging (#100)

* refactor(memory): lazy-init embedder env reads

* feat(memory): reindex log to .rag/reindex.log

* refactor(memory): search handles None embedder

* test(memory): cover None embedder in search and index

* docs(handoff): add handoff and ADR for lazy-init embedder

* docs(handoff): set PR number

* docs(handoff): rename to pr-100 prefix for pipeline detection

* refactor(memory): split run_search to satisfy xenon rank A

* fix(memory): cast search result fields to satisfy mypy

* style(memory): ruff format search.py

---------

Co-authored-by: opencode-agent <agent@opencode.local>
This commit is contained in:
Sergey 2026-07-27 00:30:16 +03:00 committed by GitHub
parent e6c378410f
commit a6491df5a6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 249 additions and 40 deletions

View file

@ -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.

View file

@ -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-NUMBER>` — будет заполнен номером 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 избыточен но безвреден.

View file

@ -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

View file

@ -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, "

View file

@ -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))

View file

@ -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:

View file

@ -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:

View file

@ -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 == []