adding phrase matching to ebook rag engen #36

Merged
Richie merged 36 commits from feature/adding-phrase-matching-to-ebook-rag-engen into main 2026-07-25 10:17:10 -04:00
23 changed files with 142 additions and 368 deletions
Showing only changes of commit 8f1a69529c - Show all commits
+3 -10
View File
@@ -32,11 +32,8 @@ async def answer_query(
return "No relevant sources were found."
logger.info(
"ebook_answer_request_start base_url=%s model=%s sources=%s query_length=%s",
config.vllm_base_url,
config.chat_model,
len(results),
len(query),
f"ebook_answer_request_start {config.vllm_base_url=} {config.chat_model=} sources={len(results)} "
f"query_length={len(query)}"
)
context = "\n\n".join(
f"[{index}] {result.source_title}{' - ' + result.chapter_title if result.chapter_title else ''}\n{result.text}"
@@ -57,9 +54,5 @@ async def answer_query(
],
)
logger.info(
"ebook_answer_request_complete model=%s answer_length=%s",
config.chat_model,
len(content),
)
logger.info(f"ebook_answer_request_complete {config.chat_model=} answer_length={len(content)}")
return content or "The model returned an empty answer."
+1 -4
View File
@@ -36,10 +36,7 @@ def schedule_bm25_refresh(app: FastAPI) -> None:
app.state.bm25_refresh_task = loop.create_task(refresh_bm25_for_app(app))
app.state.bm25_refresh_timer = loop.call_later(app.state.config.bm25_refresh_delay_seconds, start_refresh)
logger.info(
"ebook_bm25_refresh_scheduled delay_seconds=%s",
app.state.config.bm25_refresh_delay_seconds,
)
logger.info(f"ebook_bm25_refresh_scheduled {app.state.config.bm25_refresh_delay_seconds=}")
def cancel_bm25_refresh(app: FastAPI) -> None:
+5 -9
View File
@@ -65,12 +65,12 @@ def start_book_phrase_judgment(app: FastAPI, background_tasks: BackgroundTasks,
"""
state = get_judge_task_state(app)
if source_id in state.running_book_ids:
logger.info("ebook_book_phrase_judgment_already_running source_id=%s", source_id)
logger.info(f"ebook_book_phrase_judgment_already_running {source_id=}")
return False
state.running_book_ids.add(source_id)
state.outcome_messages.pop(source_id, None)
background_tasks.add_task(judge_book_phrases_for_app, app, source_id)
logger.info("ebook_book_phrase_judgment_queued source_id=%s", source_id)
logger.info(f"ebook_book_phrase_judgment_queued {source_id=}")
return True
@@ -85,12 +85,8 @@ async def judge_book_phrases_for_app(app: FastAPI, source_id: int) -> None:
try:
result = await judge_candidate_phrases_for_books(app.state.engine, app.state.config, source_ids=[source_id])
logger.info(
"ebook_book_phrase_judgment_complete source_id=%s judged=%s protected=%s mentions=%s failed=%s",
source_id,
result.candidates_judged,
result.protected_phrases,
result.phrase_mentions,
result.books_failed,
f"ebook_book_phrase_judgment_complete {source_id=} {result.candidates_judged=} {result.protected_phrases=} "
f"{result.phrase_mentions=} {result.books_failed=}"
)
if result.books_failed:
message = "Judging failed; see server logs for details"
@@ -99,7 +95,7 @@ async def judge_book_phrases_for_app(app: FastAPI, source_id: int) -> None:
f"Judged {result.candidates_judged} candidates; {result.protected_phrases} protected phrases promoted"
)
except Exception:
logger.exception("ebook_book_phrase_judgment_task_failed source_id=%s", source_id)
logger.exception(f"ebook_book_phrase_judgment_task_failed {source_id=}")
message = "Judging failed; see server logs for details"
state.running_book_ids.discard(source_id)
state.outcome_messages[source_id] = message
+3 -10
View File
@@ -37,16 +37,9 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
config = load_config()
app.state.config = config
logger.info(
"ebook_search_config_loaded top_k=%s embedding_model=%s embedding_base_url=%s vllm_base_url=%s "
"rerank_enabled=%s phrase_matching_enabled=%s answer_enabled=%s library_paths=%s",
config.top_k,
config.embedding_model,
config.embedding_base_url,
config.vllm_base_url,
config.rerank.enabled,
config.phrase_matching_enabled,
config.answer_enabled,
len(config.library_paths),
f"ebook_search_config_loaded {config.top_k=} {config.embedding_model=} {config.embedding_base_url=} "
f"{config.vllm_base_url=} {config.rerank.enabled=} {config.phrase_matching_enabled=} {config.answer_enabled=} "
f"library_paths={len(config.library_paths)}"
)
if not config.library_paths:
logger.warning("ebook_search_no_library_paths_configured")
+10 -30
View File
@@ -32,10 +32,8 @@ async def admin(request: Request, config: AppConfig, session: AsyncDbSession) ->
stats = await embedding_model_stats(session)
phrase_stats = await corpus_phrase_stats(session)
logger.info(
"ebook_admin_page_loaded models=%s candidate_phrases=%s protected_phrases=%s",
len(stats),
phrase_stats.candidate_phrases,
phrase_stats.protected_phrases,
f"ebook_admin_page_loaded models={len(stats)} {phrase_stats.candidate_phrases=} "
f"{phrase_stats.protected_phrases=}"
)
return templates.TemplateResponse(
request,
@@ -54,7 +52,7 @@ async def scan_library(request: Request, config: AppConfig, session: AsyncDbSess
logger.exception("ebook_admin_scan_failed")
return templates.TemplateResponse(request, "partials/error.html", {"message": str(error)}, status_code=500)
logger.info("ebook_admin_scan_complete changed_files=%s", count)
logger.info(f"ebook_admin_scan_complete {count=}")
if count > 0:
schedule_bm25_refresh(request.app)
return templates.TemplateResponse(request, "partials/admin_status.html", {"message": f"Indexed {count} EPUBs"})
@@ -70,10 +68,7 @@ async def generate_all_phrases(request: Request, config: AppConfig, engine: AppE
return templates.TemplateResponse(request, "partials/error.html", {"message": str(error)}, status_code=500)
logger.info(
"ebook_admin_generate_phrases_complete books_seen=%s books_built=%s candidates=%s",
result.books_seen,
result.books_built,
result.candidate_phrases,
f"ebook_admin_generate_phrases_complete {result.books_seen=} {result.books_built=} {result.candidate_phrases=}"
)
return templates.TemplateResponse(
request,
@@ -136,14 +131,8 @@ async def run_phrase_judgment(
return templates.TemplateResponse(request, "partials/error.html", {"message": str(error)}, status_code=500)
logger.info(
"ebook_admin_judge_phrases_complete books_seen=%s books_judged=%s books_failed=%s candidates_judged=%s "
"protected=%s mentions=%s",
result.books_seen,
result.books_judged,
result.books_failed,
result.candidates_judged,
result.protected_phrases,
result.phrase_mentions,
f"ebook_admin_judge_phrases_complete {result.books_seen=} {result.books_judged=} {result.books_failed=} "
f"{result.candidates_judged=} {result.protected_phrases=} {result.phrase_mentions=}"
)
return templates.TemplateResponse(
request,
@@ -174,7 +163,7 @@ async def embed_missing(
logger.exception("ebook_admin_embed_missing_failed")
return templates.TemplateResponse(request, "partials/error.html", {"message": str(error)}, status_code=500)
logger.info("ebook_admin_embed_missing_complete chunks=%s", count)
logger.info(f"ebook_admin_embed_missing_complete {count=}")
return templates.TemplateResponse(
request,
"partials/admin_status.html",
@@ -200,18 +189,9 @@ async def embed_all(
await session.commit()
total += count
batches += 1
logger.info(
"ebook_admin_embed_all_batch_complete batch=%s chunks=%s total_chunks=%s",
batches,
count,
total,
)
logger.info(f"ebook_admin_embed_all_batch_complete {batches=} {count=} {total=}")
except Exception as error:
logger.exception(
"ebook_admin_embed_all_failed batches=%s chunks=%s",
batches,
total,
)
logger.exception(f"ebook_admin_embed_all_failed {batches=} {total=}")
return templates.TemplateResponse(
request,
"partials/error.html",
@@ -219,7 +199,7 @@ async def embed_all(
status_code=500,
)
logger.info("ebook_admin_embed_all_complete batches=%s chunks=%s", batches, total)
logger.info(f"ebook_admin_embed_all_complete {batches=} {total=}")
return templates.TemplateResponse(
request,
"partials/admin_status.html",
+2 -9
View File
@@ -60,14 +60,7 @@ async def ready(config: AppConfig, session: AsyncDbSession, client: AppHttpClien
status = "ready"
status_code = HTTPStatus.OK
logger.info(
"ebook_ready_check status=%s database=%s embedding=%s chat=%s bm25=%s",
status,
database_ok,
embedding_ok,
chat_status,
bm25_status,
)
logger.info(f"ebook_ready_check {status=} {database_ok=} {embedding_ok=} {chat_status=} {bm25_status=}")
return JSONResponse(content={"status": status, "checks": checks}, status_code=status_code)
@@ -83,7 +76,7 @@ async def check_database(session: AsyncSession) -> bool:
try:
await session.execute(select(literal(1)))
except SQLAlchemyError as error:
logger.warning("ebook_ready_database_unavailable error=%s", error)
logger.warning(f"ebook_ready_database_unavailable {error=}")
return False
return True
+7 -18
View File
@@ -36,7 +36,7 @@ async def index(request: Request, config: AppConfig) -> HTMLResponse:
async def books(request: Request, session: AsyncDbSession) -> HTMLResponse:
"""Render the indexed books page."""
sources = list((await session.scalars(select(EbookSource).order_by(EbookSource.title))).all())
logger.info("ebook_books_page_loaded count=%s", len(sources))
logger.info(f"ebook_books_page_loaded count={len(sources)}")
return templates.TemplateResponse(request, "books.html", {"sources": sources})
@@ -134,14 +134,8 @@ async def book_detail(source_id: int, request: Request, session: AsyncDbSession)
candidates = []
protected_phrases = []
logger.info(
"ebook_book_detail_loaded source_id=%s found=%s chapters=%s chunks=%s candidates=%s judged=%s protected=%s",
source_id,
source is not None,
chapter_count,
chunk_count,
candidate_count,
judged_candidate_count,
protected_count,
f"ebook_book_detail_loaded {source_id=} found={source is not None} {chapter_count=} {chunk_count=} "
f"{candidate_count=} {judged_candidate_count=} {protected_count=}"
)
return templates.TemplateResponse(
request,
@@ -173,14 +167,9 @@ async def recalculate_book_phrases(source_id: int, config: AppConfig, session: A
except ValueError as error:
raise HTTPException(status_code=409, detail=str(error)) from error
logger.info(
"ebook_book_phrase_recalculation_complete source_id=%s candidates=%s deleted_candidates=%s "
"deleted_protected=%s deleted_aliases=%s deleted_mentions=%s",
source_id,
result.candidate_phrases,
result.deleted_candidates,
result.deleted_protected_phrases,
result.deleted_aliases,
result.deleted_mentions,
f"ebook_book_phrase_recalculation_complete {source_id=} {result.candidate_phrases=} "
f"{result.deleted_candidates=} {result.deleted_protected_phrases=} {result.deleted_aliases=} "
f"{result.deleted_mentions=}"
)
return RedirectResponse(
url=f"/books/{source_id}?phrases_recalculated={result.candidate_phrases}",
@@ -201,5 +190,5 @@ async def judge_book_phrases(
raise HTTPException(status_code=404, detail="Book not found")
started = start_book_phrase_judgment(request.app, background_tasks, source.id)
logger.info("ebook_book_phrase_judgment_requested source_id=%s started=%s", source_id, started)
logger.info(f"ebook_book_phrase_judgment_requested {source_id=} {started=}")
return RedirectResponse(url=f"/books/{source_id}", status_code=303)
+7 -14
View File
@@ -49,9 +49,8 @@ async def build_answer(
if not is_confident(response.results, config):
logger.info(
"ebook_answer_low_confidence confidence=%.4f threshold=%.4f",
retrieval_confidence(response.results),
config.min_retrieval_confidence,
f"ebook_answer_low_confidence confidence={retrieval_confidence(response.results):.4f} "
f"{config.min_retrieval_confidence=:.4f}"
)
answer = (
"Retrieval confidence is low for this query, so answer generation was skipped. "
@@ -62,18 +61,14 @@ async def build_answer(
try:
answer = await answer_query(client, query, response.results, config)
except RuntimeError as error:
logger.warning("ebook_answer_request_failed_falling_back error=%s", error)
logger.warning(f"ebook_answer_request_failed_falling_back {error=}")
return "Answer generation failed. Source chunks are still shown below.", False, None
citation_report = None
if config.validate_citations_enabled and response.results:
citation_report = validate_citations(answer, len(response.results))
if citation_report.invalid or not citation_report.grounded:
logger.warning(
"ebook_answer_citation_issue invalid=%s grounded=%s",
citation_report.invalid,
citation_report.grounded,
)
logger.warning(f"ebook_answer_citation_issue {citation_report.invalid=} {citation_report.grounded=}")
return answer, False, citation_report
@@ -110,12 +105,10 @@ async def search(
)
for step in response.timings:
logger.info("ebook_search_timing step=%r runtime_ms=%.1f", step.name, step.duration_ms)
logger.info(f"ebook_search_timing {step.name=} {step.duration_ms=:.1f}")
logger.info(
"ebook_search_request_complete results=%s rank_label=%s runtime_ms=%.1f",
len(response.results),
response.rank_label,
response.total_runtime_ms,
f"ebook_search_request_complete results={len(response.results)} {response.rank_label=} "
f"{response.total_runtime_ms=:.1f}"
)
return templates.TemplateResponse(
request,
+7 -14
View File
@@ -80,23 +80,19 @@ async def ensure_bm25_corpus(session: AsyncSession, config: EbookSearchConfig) -
manifest = read_bm25_manifest(index_path)
db_updated_at = await corpus_last_updated_at(session)
if not bm25_index_exists(index_path, manifest):
logger.info("ebook_bm25_index_missing path=%s", index_path)
logger.info(f"ebook_bm25_index_missing {index_path=}")
await refresh_bm25_corpus(session, config, db_updated_at=db_updated_at)
return
if db_updated_at is not None and manifest is not None and manifest.created_at < db_updated_at:
logger.info(
"ebook_bm25_index_stale path=%s created_at=%s db_updated_at=%s",
index_path,
manifest.created_at.isoformat(),
db_updated_at.isoformat(),
f"ebook_bm25_index_stale {index_path=} created_at={manifest.created_at.isoformat()} "
f"db_updated_at={db_updated_at.isoformat()}"
)
await refresh_bm25_corpus(session, config, db_updated_at=db_updated_at)
return
logger.info(
"ebook_bm25_index_current path=%s chunks=%s created_at=%s",
index_path,
manifest.chunk_count if manifest else 0,
manifest.created_at.isoformat() if manifest else None,
f"ebook_bm25_index_current {index_path=} chunks={manifest.chunk_count if manifest else 0} "
f"created_at={manifest.created_at.isoformat() if manifest else None}"
)
@@ -119,10 +115,7 @@ async def refresh_bm25_corpus(
)
await asyncio.to_thread(write_bm25_corpus, index_path, records, texts, manifest)
logger.info(
"ebook_bm25_index_refreshed path=%s chunks=%s created_at=%s",
index_path,
manifest.chunk_count,
manifest.created_at.isoformat(),
f"ebook_bm25_index_refreshed {index_path=} {manifest.chunk_count=} created_at={manifest.created_at.isoformat()}"
)
return manifest
@@ -135,7 +128,7 @@ def load_bm25_corpus(config: EbookSearchConfig) -> BM25Corpus:
"""
index_path = bm25_index_path(config)
active_index_path = get_current_bm25_index(index_path)
logger.info("ebook_bm25_corpus_cache_load path=%s active_path=%s", index_path, active_index_path)
logger.info(f"ebook_bm25_corpus_cache_load {index_path=} {active_index_path=}")
manifest = read_bm25_manifest(index_path)
if manifest is None or not bm25_index_exists(index_path, manifest):
msg = f"BM25 corpus is not available: {index_path}"
+1 -1
View File
@@ -32,7 +32,7 @@ def docker_run(
capture_output: bool = False,
) -> subprocess.CompletedProcess[str]:
"""Run docker with repo-root cwd and consistent error handling."""
logger.info("docker %s", " ".join(arguments))
logger.info(f"docker {' '.join(arguments)}")
return subprocess.run(
["docker", *arguments],
cwd=get_repo_dir(),
+6 -16
View File
@@ -72,24 +72,14 @@ async def embed_texts(
config: EbookSearchConfig,
) -> list[list[float]]:
"""Embed text with the configured vLLM embedding model."""
logger.info(
"ebook_embed_request_start base_url=%s model=%s count=%s",
config.embedding_base_url,
config.embedding_model,
len(texts),
)
logger.info(f"ebook_embed_request_start {config.embedding_base_url=} {config.embedding_model=} count={len(texts)}")
vectors = await request_embeddings(client, texts, config)
expected_dimension = MODEL_DIMENSIONS[config.embedding_model]
for vector in vectors:
if len(vector) != expected_dimension:
msg = f"Expected {expected_dimension} dimensions, got {len(vector)}"
raise ValueError(msg)
logger.info(
"ebook_embed_request_complete model=%s count=%s dimension=%s",
config.embedding_model,
len(vectors),
expected_dimension,
)
logger.info(f"ebook_embed_request_complete {config.embedding_model=} count={len(vectors)} {expected_dimension=}")
return vectors
@@ -105,7 +95,7 @@ async def ensure_embedding_models(session: AsyncSession) -> None:
existing = await session.scalar(select(EbookEmbeddingModel).where(EbookEmbeddingModel.name == name))
if existing is None:
session.add(EbookEmbeddingModel(name=name, dimension=dimension, is_default=name == "qwen3-embedding-0.6b"))
logger.info("ebook_embedding_model_created model=%s dimension=%s", name, dimension)
logger.info(f"ebook_embedding_model_created {name=} {dimension=}")
await session.flush()
@@ -159,10 +149,10 @@ async def embed_missing_chunks(session: AsyncSession, client: httpx.AsyncClient,
)
)
if not chunks:
logger.info("ebook_embed_missing_none model=%s", config.embedding_model)
logger.info(f"ebook_embed_missing_none {config.embedding_model=}")
return 0
logger.info("ebook_embed_missing_batch_start model=%s count=%s", config.embedding_model, len(chunks))
logger.info(f"ebook_embed_missing_batch_start {config.embedding_model=} count={len(chunks)}")
vectors = await embed_texts(client, [chunk.text for chunk in chunks], config)
rows = [
{"chunk_id": chunk.id, "model_id": model.id, "embedding": vector}
@@ -171,5 +161,5 @@ async def embed_missing_chunks(session: AsyncSession, client: httpx.AsyncClient,
statement = insert(table).values(rows).on_conflict_do_nothing(index_elements=["chunk_id", "model_id"])
await session.execute(statement)
await session.flush()
logger.info("ebook_embed_missing_batch_complete model=%s count=%s", config.embedding_model, len(rows))
logger.info(f"ebook_embed_missing_batch_complete {config.embedding_model=} count={len(rows)}")
return len(rows)
+9 -13
View File
@@ -94,13 +94,13 @@ async def ingest_configured_paths(session: AsyncSession, config: EbookSearchConf
count = 0
for library_path in config.library_paths:
path, epub_paths = await asyncio.to_thread(find_library_epubs, library_path)
logger.info("ebook_ingest_path_start path=%s", path)
logger.info(f"ebook_ingest_path_start {path=}")
if epub_paths is None:
logger.warning("ebook_ingest_path_missing path=%s", path)
logger.warning(f"ebook_ingest_path_missing {path=}")
continue
for epub_path in epub_paths:
count += int(await ingest_file(session, epub_path, config))
logger.info("ebook_ingest_paths_complete changed_files=%s configured_paths=%s", count, len(config.library_paths))
logger.info(f"ebook_ingest_paths_complete {count=} configured_paths={len(config.library_paths)}")
return count
@@ -113,7 +113,7 @@ async def ingest_file(session: AsyncSession, path: Path, config: EbookSearchConf
"""Ingest one EPUB file. Return True when the database changed."""
try:
resolved_path = await asyncio.to_thread(resolve_ingest_path, path)
logger.info("ebook_ingest_file_start path=%s", resolved_path)
logger.info(f"ebook_ingest_file_start {resolved_path=}")
file_hash = await asyncio.to_thread(sha256_file, resolved_path)
existing = await find_existing_source(session, resolved_path, file_hash)
if existing is not None and existing.file_sha256 == file_hash:
@@ -122,10 +122,10 @@ async def ingest_file(session: AsyncSession, path: Path, config: EbookSearchConf
existing.file_mtime = datetime.fromtimestamp(stat.st_mtime, tz=UTC)
existing.file_size = stat.st_size
await session.flush()
logger.info("ebook_ingest_file_unchanged source_id=%s path=%s", existing.id, resolved_path)
logger.info(f"ebook_ingest_file_unchanged {existing.id=} {resolved_path=}")
return False
if existing is not None:
logger.info("ebook_ingest_file_replacing source_id=%s path=%s", existing.id, resolved_path)
logger.info(f"ebook_ingest_file_replacing {existing.id=} {resolved_path=}")
await session.delete(existing)
await session.flush()
@@ -160,15 +160,11 @@ async def ingest_file(session: AsyncSession, path: Path, config: EbookSearchConf
await session.commit()
mention_count = await index_chunk_phrase_mentions_for_book(session, source.id, config)
logger.info(
"ebook_ingest_file_complete source_id=%s path=%s chapters=%s chunks=%s phrase_mentions=%s",
source.id,
resolved_path,
len(parsed.chapters),
chunk_index,
mention_count,
f"ebook_ingest_file_complete {source.id=} {resolved_path=} chapters={len(parsed.chapters)} {chunk_index=} "
f"{mention_count=}"
)
except Exception:
logger.exception(f"ebook_ingest_file_error path={path}")
logger.exception(f"ebook_ingest_file_error {path=}")
return False
else:
return True
+3 -6
View File
@@ -51,10 +51,7 @@ async def request_embeddings(
return embedding_vectors_from_response(response.json())
except (httpx.HTTPError, ValueError, KeyError, TypeError) as error:
logger.exception(
"ebook_embed_request_failed base_url=%s model=%s count=%s",
config.embedding_base_url,
config.embedding_model,
len(texts),
f"ebook_embed_request_failed {config.embedding_base_url=} {config.embedding_model=} count={len(texts)}"
)
msg = f"Embedding request failed. base_url={config.embedding_base_url} model={config.embedding_model}"
raise RuntimeError(msg) from error
@@ -75,7 +72,7 @@ async def check_embedding_endpoint(
)
response.raise_for_status()
except httpx.HTTPError as error:
logger.warning("ebook_embedding_endpoint_unreachable base_url=%s error=%s", config.embedding_base_url, error)
logger.warning(f"ebook_embedding_endpoint_unreachable {config.embedding_base_url=} {error=}")
return False
return True
@@ -95,7 +92,7 @@ async def check_chat_endpoint(
)
response.raise_for_status()
except httpx.HTTPError as error:
logger.warning("ebook_chat_endpoint_unreachable base_url=%s error=%s", config.vllm_base_url, error)
logger.warning(f"ebook_chat_endpoint_unreachable {config.vllm_base_url=} {error=}")
return False
return True
+2 -9
View File
@@ -112,7 +112,7 @@ async def send_search(client: httpx.AsyncClient, query: str, *, rerank: bool) ->
try:
response = await client.post("/search", data=data)
except httpx.HTTPError as error:
logger.warning("ebook_loadtest_request_failed error=%s", error)
logger.warning(f"ebook_loadtest_request_failed {error=}")
return RequestResult(status_code=0, latency_ms=(time.perf_counter() - start) * 1000, ok=False)
return RequestResult(
status_code=response.status_code,
@@ -192,14 +192,7 @@ def main(
"""Load test the search endpoint and report latency and throughput."""
configure_logger(log_level)
queries = load_queries(queries_file)
logger.info(
"ebook_loadtest_start base_url=%s requests=%s concurrency=%s rerank=%s queries=%s",
base_url,
request_count,
concurrency,
rerank,
len(queries),
)
logger.info(f"ebook_loadtest_start {base_url=} {request_count=} {concurrency=} {rerank=} queries={len(queries)}")
summary = asyncio.run(
run_load(
base_url=base_url,
@@ -0,0 +1 @@
"""Protected phrase extraction and matching for ebook search."""
@@ -682,33 +682,26 @@ def extract_phrase_candidates_for_book(
"""
started_at = perf_counter()
logger.info(
"ebook_phrase_candidate_extract_start chapters=%s chars=%s min_tokens=%s max_tokens=%s max_candidates=%s",
len(chapters),
len(book_text),
config.phrase_min_tokens,
config.phrase_max_tokens,
config.protected_phrase_max_candidates_per_book,
f"ebook_phrase_candidate_extract_start chapters={len(chapters)} chars={len(book_text)} "
f"{config.phrase_min_tokens=} {config.phrase_max_tokens=} {config.protected_phrase_max_candidates_per_book=}"
)
raw_started_at = perf_counter()
raw = extract_raw_ngrams_by_chapter(chapters, config)
logger.info(
"ebook_phrase_candidate_extract_raw_complete candidates=%s duration_ms=%.1f",
len(raw),
(perf_counter() - raw_started_at) * 1000,
f"ebook_phrase_candidate_extract_raw_complete candidates={len(raw)} duration_ms={(perf_counter() - "
f"raw_started_at) * 1000:.1f}"
)
yake_started_at = perf_counter()
yake_candidates = extract_yake_candidates(book_text, config)
logger.info(
"ebook_phrase_candidate_extract_yake_complete candidates=%s duration_ms=%.1f",
len(yake_candidates),
(perf_counter() - yake_started_at) * 1000,
f"ebook_phrase_candidate_extract_yake_complete candidates={len(yake_candidates)} duration_ms={(perf_counter() - "
f"yake_started_at) * 1000:.1f}"
)
capitalized_started_at = perf_counter()
capitalized = extract_capitalized_phrases(book_text, config)
logger.info(
"ebook_phrase_candidate_extract_capitalized_complete candidates=%s duration_ms=%.1f",
len(capitalized),
(perf_counter() - capitalized_started_at) * 1000,
f"ebook_phrase_candidate_extract_capitalized_complete candidates={len(capitalized)} "
f"duration_ms={(perf_counter() - capitalized_started_at) * 1000:.1f}"
)
metadata_candidates = extract_metadata_candidates(metadata, config)
@@ -732,22 +725,10 @@ def extract_phrase_candidates_for_book(
: config.protected_phrase_max_candidates_per_book
]
logger.info(
"ebook_phrase_candidate_extract_complete raw=%s yake=%s capitalized=%s metadata=%s "
"merged=%s filtered_too_short=%s filtered_too_rare=%s filtered_too_common=%s filtered_junk=%s "
"min_uses=%s storable=%s limited=%s enrich_score_ms=%.1f duration_ms=%.1f",
len(raw),
len(yake_candidates),
len(capitalized),
len(metadata_candidates),
pre_filter_count,
filtered_too_short,
filtered_too_rare,
filtered_too_common,
filtered_junk,
minimum_candidate_raw_count(config),
len(candidates),
len(limited),
(perf_counter() - enriched_started_at) * 1000,
(perf_counter() - started_at) * 1000,
f"ebook_phrase_candidate_extract_complete raw={len(raw)} yake={len(yake_candidates)} "
f"capitalized={len(capitalized)} metadata={len(metadata_candidates)} {pre_filter_count=} {filtered_too_short=} "
f"{filtered_too_rare=} {filtered_too_common=} {filtered_junk=} min_uses={minimum_candidate_raw_count(config)} "
f"storable={len(candidates)} limited={len(limited)} enrich_score_ms={(perf_counter() - enriched_started_at) * "
f"1000:.1f} duration_ms={(perf_counter() - started_at) * 1000:.1f}"
)
return limited
@@ -68,12 +68,8 @@ async def generate_candidate_phrases_for_books(
source_ids = (await session.scalars(source_query)).all()
books_seen = len(source_ids)
logger.info(
"ebook_candidate_phrase_generation_start books_seen=%s min_tokens=%s max_tokens=%s "
"max_candidates_per_book=%s",
books_seen,
config.phrase_min_tokens,
config.phrase_max_tokens,
config.protected_phrase_max_candidates_per_book,
f"ebook_candidate_phrase_generation_start {books_seen=} {config.phrase_min_tokens=} {config.phrase_max_tokens=} "
f"{config.protected_phrase_max_candidates_per_book=}"
)
pool = get_extraction_pool(config.protected_phrase_extraction_workers)
@@ -89,19 +85,11 @@ async def generate_candidate_phrases_for_books(
await asyncio.wait([wrapped_future])
exception = wrapped_future.exception()
if exception is not None:
logger.error(
"ebook_candidate_phrase_generation_book_failed source_id=%s",
source_id,
exc_info=exception,
)
logger.error(f"ebook_candidate_phrase_generation_book_failed {source_id=}")
outcomes.append(BookCandidateResult())
continue
saved_count = wrapped_future.result()
logger.info(
"ebook_candidate_phrase_generation_book_committed source_id=%s candidates=%s",
source_id,
saved_count,
)
logger.info(f"ebook_candidate_phrase_generation_book_committed {source_id=} {saved_count=}")
outcomes.append(BookCandidateResult(candidates=saved_count, built=True))
result = PhraseCandidateGenerationResult(
@@ -110,10 +98,8 @@ async def generate_candidate_phrases_for_books(
candidate_phrases=sum(outcome.candidates for outcome in outcomes),
)
logger.info(
"ebook_candidate_phrase_generation_complete books_seen=%s books_built=%s candidate_total=%s",
result.books_seen,
result.books_built,
result.candidate_phrases,
f"ebook_candidate_phrase_generation_complete {result.books_seen=} {result.books_built=} "
f"{result.candidate_phrases=}"
)
return result
@@ -141,11 +127,7 @@ async def recalculate_candidate_phrases_for_book(
regeneration failure rolls the deletion back.
"""
started_at = perf_counter()
logger.info(
"ebook_candidate_phrase_recalculation_start source_id=%s title=%r",
source.id,
source.title,
)
logger.info(f"ebook_candidate_phrase_recalculation_start {source.id=} {source.title=}")
deleted = await delete_phrase_data_for_book(session, source.id)
candidate_count = await generate_candidate_phrases_for_book(
session,
@@ -164,15 +146,9 @@ async def recalculate_candidate_phrases_for_book(
candidate_phrases=candidate_count,
)
logger.info(
"ebook_candidate_phrase_recalculation_complete source_id=%s deleted_candidates=%s "
"deleted_protected=%s deleted_aliases=%s deleted_mentions=%s candidates=%s duration_ms=%.1f",
source.id,
result.deleted_candidates,
result.deleted_protected_phrases,
result.deleted_aliases,
result.deleted_mentions,
result.candidate_phrases,
(perf_counter() - started_at) * 1000,
f"ebook_candidate_phrase_recalculation_complete {source.id=} {result.deleted_candidates=} "
f"{result.deleted_protected_phrases=} {result.deleted_aliases=} {result.deleted_mentions=} "
f"{result.candidate_phrases=} duration_ms={(perf_counter() - started_at) * 1000:.1f}"
)
return result
@@ -268,10 +244,8 @@ async def generate_candidate_phrases_for_book(
await session.rollback()
raise
logger.info(
"ebook_candidate_phrase_generation_book_duration book_id=%s candidates=%s duration_ms=%.1f",
book_id,
saved_count,
(perf_counter() - started_at) * 1000,
f"ebook_candidate_phrase_generation_book_duration {book_id=} {saved_count=} duration_ms={(perf_counter() - "
f"started_at) * 1000:.1f}"
)
return saved_count
@@ -306,23 +280,16 @@ async def store_candidate_phrases_for_book(
await session.flush()
saved_count = len(rows)
logger.info(
"ebook_candidate_phrase_save_start book_id=%s candidates=%s mode=bulk_insert",
book_id,
len(limited_candidates),
f"ebook_candidate_phrase_save_start {book_id=} candidates={len(limited_candidates)} mode=bulk_insert"
)
else:
pruned_count = await prune_unstorable_unjudged_candidate_phrases(session, book_id, config)
logger.info(
"ebook_candidate_phrase_save_start book_id=%s candidates=%s pruned_unstorable=%s",
book_id,
len(limited_candidates),
pruned_count,
f"ebook_candidate_phrase_save_start {book_id=} candidates={len(limited_candidates)} {pruned_count=}"
)
saved_count = await bulk_upsert_unjudged_candidates(session, book_id, series_id, limited_candidates)
logger.info(
"ebook_candidate_phrase_save_complete book_id=%s candidates=%s save_ms=%.1f",
book_id,
saved_count,
(perf_counter() - save_started_at) * 1000,
f"ebook_candidate_phrase_save_complete {book_id=} {saved_count=} save_ms={(perf_counter() - save_started_at) * "
f"1000:.1f}"
)
return saved_count
@@ -80,12 +80,8 @@ async def judge_candidate_phrases_for_books(
book_workers = max(1, config.phrase_judge_book_workers)
phrase_workers = max(1, config.phrase_judge_phrase_workers)
logger.info(
"ebook_candidate_phrase_judgment_start books_seen=%s book_workers=%s phrase_workers=%s "
"confidence_threshold=%.2f",
books_seen,
book_workers,
phrase_workers,
config.protected_phrase_confidence_threshold,
f"ebook_candidate_phrase_judgment_start {books_seen=} {book_workers=} {phrase_workers=} "
f"{config.protected_phrase_confidence_threshold=:.2f}"
)
book_semaphore = asyncio.Semaphore(book_workers)
@@ -105,14 +101,8 @@ async def judge_candidate_phrases_for_books(
phrase_mentions=sum(outcome.mentions for outcome in outcomes),
)
logger.info(
"ebook_candidate_phrase_judgment_complete books_seen=%s books_judged=%s books_failed=%s "
"candidates_judged=%s protected=%s mentions=%s",
result.books_seen,
result.books_judged,
result.books_failed,
result.candidates_judged,
result.protected_phrases,
result.phrase_mentions,
f"ebook_candidate_phrase_judgment_complete {result.books_seen=} {result.books_judged=} {result.books_failed=} "
f"{result.candidates_judged=} {result.protected_phrases=} {result.phrase_mentions=}"
)
return result
@@ -147,7 +137,7 @@ async def judge_one_book_async(
return BookJudgmentResult()
return await persist_book_judgments(engine, source_id, config, judged)
except Exception:
logger.exception("ebook_candidate_phrase_judgment_book_failed source_id=%s", source_id)
logger.exception(f"ebook_candidate_phrase_judgment_book_failed {source_id=}")
return BookJudgmentResult(failed=True)
@@ -173,7 +163,7 @@ async def prepare_book_judgment(
return None
async with AsyncSession(engine) as session:
if not await count_unjudged_candidates(session, source_id, config):
logger.info("ebook_candidate_phrase_judgment_book_skip_no_unjudged source_id=%s", source_id)
logger.info(f"ebook_candidate_phrase_judgment_book_skip_no_unjudged {source_id=}")
return None
existing_protected = await count_protected_phrases(session, source_id)
target_remaining: int | None = None
@@ -181,15 +171,13 @@ async def prepare_book_judgment(
target_remaining = max(config.phrase_target_protected_per_book - existing_protected, 0)
if target_remaining == 0:
logger.info(
"ebook_candidate_phrase_judgment_skipped_target_met source_id=%s existing_protected=%s target=%s",
source_id,
existing_protected,
config.phrase_target_protected_per_book,
f"ebook_candidate_phrase_judgment_skipped_target_met {source_id=} {existing_protected=} "
f"{config.phrase_target_protected_per_book=}"
)
return None
book_text = await load_book_text(session, source_id)
if not book_text:
logger.warning("ebook_candidate_phrase_judgment_book_empty source_id=%s", source_id)
logger.warning(f"ebook_candidate_phrase_judgment_book_empty {source_id=}")
return None
normalized_book_text = normalize_text(book_text)
# Stored rows may predate the current junk filters and score weights, so re-filter and
@@ -211,15 +199,8 @@ async def prepare_book_judgment(
normalized_book_text, candidate.phrase_norm
)
logger.info(
"ebook_candidate_phrase_judgment_candidates_loaded source_id=%s candidates=%s skipped_junk=%s "
"unjudged_rows=%s existing_protected=%s target_remaining=%s judgment_limit=%s",
source_id,
len(work_items),
skipped_junk,
len(rows),
existing_protected,
target_remaining,
judgment_limit,
f"ebook_candidate_phrase_judgment_candidates_loaded {source_id=} candidates={len(work_items)} {skipped_junk=} "
f"unjudged_rows={len(rows)} {existing_protected=} {target_remaining=} {judgment_limit=}"
)
return work_items, target_remaining
@@ -313,31 +294,20 @@ async def persist_book_judgments(
await upsert_protected_phrase(session, source_id, None, candidate, judgment, candidate_row)
)
logger.info(
"ebook_candidate_phrase_judgment_candidate_complete source_id=%s candidate_id=%s phrase=%r "
"keep=%s confidence=%.3f category=%r promoted=%s",
source_id,
candidate_id,
candidate.phrase_norm,
judgment.keep,
judgment.confidence,
judgment.category,
promote,
f"ebook_candidate_phrase_judgment_candidate_complete {source_id=} {candidate_id=} "
f"{candidate.phrase_norm=} {judgment.keep=} {judgment.confidence=:.3f} {judgment.category=} "
f"{promote=}"
)
await session.flush()
mentions = await index_chunk_phrase_mentions_for_book(session, source_id, config) if protected else 0
await session.commit()
except Exception:
await session.rollback()
logger.exception("ebook_candidate_phrase_judgment_book_persist_failed source_id=%s", source_id)
logger.exception(f"ebook_candidate_phrase_judgment_book_persist_failed {source_id=}")
return BookJudgmentResult(failed=True)
logger.info(
"ebook_candidate_phrase_judgment_book_committed source_id=%s judged=%s protected=%s mentions=%s "
"duration_ms=%.1f",
source_id,
len(judged),
len(protected),
mentions,
(perf_counter() - book_started_at) * 1000,
f"ebook_candidate_phrase_judgment_book_committed {source_id=} judged={len(judged)} protected={len(protected)} "
f"{mentions=} duration_ms={(perf_counter() - book_started_at) * 1000:.1f}"
)
return BookJudgmentResult(judged=len(judged), protected=len(protected), mentions=mentions, committed=True)
@@ -369,24 +339,14 @@ def should_protect_judged_candidate(
accepted_token_count = len(accepted_tokens)
if accepted_token_count < config.phrase_min_tokens:
logger.info(
"ebook_candidate_phrase_judgment_candidate_skip_short_canonical book_id=%s candidate_id=%s "
"phrase=%r canonical=%r token_count=%s min_tokens=%s",
book_id,
candidate_id,
candidate.phrase_norm,
accepted_norm,
accepted_token_count,
config.phrase_min_tokens,
f"ebook_candidate_phrase_judgment_candidate_skip_short_canonical {book_id=} {candidate_id=} "
f"{candidate.phrase_norm=} {accepted_norm=} {accepted_token_count=} {config.phrase_min_tokens=}"
)
return False
if is_most_common_word_phrase(accepted_tokens):
logger.info(
"ebook_candidate_phrase_judgment_candidate_skip_common_canonical book_id=%s candidate_id=%s "
"phrase=%r canonical=%r",
book_id,
candidate_id,
candidate.phrase_norm,
accepted_norm,
f"ebook_candidate_phrase_judgment_candidate_skip_common_canonical {book_id=} {candidate_id=} "
f"{candidate.phrase_norm=} {accepted_norm=}"
)
return False
return True
@@ -393,7 +393,7 @@ async def index_chunk_phrase_mentions_for_book(
for chunk in chunks:
count += await index_chunk_phrase_mentions(session, chunk, lookup=active_lookup)
await session.flush()
logger.info("ebook_chunk_phrase_mentions_indexed book_id=%s mentions=%s", book_id, count)
logger.info(f"ebook_chunk_phrase_mentions_indexed {book_id=} {count=}")
return count
@@ -45,7 +45,7 @@ def get_extraction_pool(max_workers: int) -> ProcessPoolExecutor:
max_workers=workers,
mp_context=multiprocessing.get_context("spawn"),
)
logger.info("ebook_phrase_extraction_pool_started workers=%s", workers)
logger.info(f"ebook_phrase_extraction_pool_started {workers=}")
return _extraction_pool.pool
+4 -11
View File
@@ -615,11 +615,8 @@ async def prune_unstorable_unjudged_candidate_phrases(
)
if deleted:
logger.info(
"ebook_candidate_phrase_unstorable_pruned book_id=%s deleted=%s min_tokens=%s min_uses=%s",
book_id,
deleted,
config.phrase_min_tokens,
minimum_candidate_raw_count(config),
f"ebook_candidate_phrase_unstorable_pruned {book_id=} {deleted=} {config.phrase_min_tokens=} "
f"min_uses={minimum_candidate_raw_count(config)}"
)
return deleted
@@ -661,12 +658,8 @@ async def delete_phrase_data_for_book(session: AsyncSession, book_id: int) -> Ph
)
await session.flush()
logger.info(
"ebook_candidate_phrase_data_deleted book_id=%s candidates=%s protected=%s aliases=%s mentions=%s",
book_id,
deleted_candidates,
deleted_protected,
deleted_aliases,
deleted_mentions,
f"ebook_candidate_phrase_data_deleted {book_id=} {deleted_candidates=} {deleted_protected=} {deleted_aliases=} "
f"{deleted_mentions=}"
)
return PhraseRecalculationResult(
book_id=book_id,
+3 -13
View File
@@ -35,12 +35,7 @@ async def rerank_chunks(
if not candidates:
return []
logger.info(
"ebook_rerank_request_start base_url=%s model=%s candidates=%s",
config.base_url,
config.model,
len(candidates),
)
logger.info(f"ebook_rerank_request_start {config.base_url=} {config.model=} candidates={len(candidates)}")
scores = await score_candidates(client, query, candidates, config)
results = sorted(
(
@@ -54,12 +49,7 @@ async def rerank_chunks(
key=lambda result: result.score,
reverse=True,
)
logger.info(
"ebook_rerank_request_complete base_url=%s model=%s candidates=%s",
config.base_url,
config.model,
len(results),
)
logger.info(f"ebook_rerank_request_complete {config.base_url=} {config.model=} candidates={len(results)}")
return results
@@ -76,7 +66,7 @@ async def score_candidates(
scores = parse_vllm_scores(body, candidates)
for result in scores.values():
logger.debug("ebook_rerank_candidate_scored chunk_id=%s score=%s", result.chunk_id, result.score)
logger.debug(f"ebook_rerank_candidate_scored {result.chunk_id=} {result.score=}")
return scores
+16 -37
View File
@@ -106,12 +106,7 @@ async def search_ebooks(
return SearchResponse(query=query, results=[], rank_label="Hybrid")
phrase_matching_enabled = config.phrase_matching_enabled if phrase_matching is None else phrase_matching
logger.info(
"ebook_search_start query_length=%s rerank=%s phrase_matching=%s",
len(query),
rerank,
phrase_matching_enabled,
)
logger.info(f"ebook_search_start query_length={len(query)} {rerank=} {phrase_matching_enabled=}")
timings: list[RuntimeStep] = []
if phrase_matching_enabled:
phrase_matches, timing = await async_timed_result(
@@ -149,16 +144,10 @@ async def search_ebooks(
timings.append(timing)
response = replace(response, timings=tuple(timings), phrase_matches=tuple(phrase_matches))
logger.info(
"ebook_search_complete vector_candidates=%s lexical_candidates=%s "
"fused_candidates=%s phrase_matching=%s phrase_matches=%s returned=%s rank_label=%s runtime_ms=%.1f",
len(retrieval.vector_results),
len(retrieval.lexical_results),
len(fused),
phrase_matching_enabled,
len(phrase_matches),
len(response.results),
response.rank_label,
response.total_runtime_ms,
f"ebook_search_complete vector_candidates={len(retrieval.vector_results)} "
f"lexical_candidates={len(retrieval.lexical_results)} fused_candidates={len(fused)} {phrase_matching_enabled=} "
f"phrase_matches={len(phrase_matches)} returned={len(response.results)} {response.rank_label=} "
f"{response.total_runtime_ms=:.1f}"
)
return response
@@ -179,13 +168,13 @@ async def query_phrase_matches(
async with AsyncSession(engine) as session:
return await detect_protected_phrases_for_query(session, query, config)
except SQLAlchemyError as error:
logger.warning("ebook_protected_phrase_detection_unavailable error=%s", error)
logger.warning(f"ebook_protected_phrase_detection_unavailable {error=}")
return []
def skip_phrase_mention_boosts(candidates: list[SearchResult]) -> list[SearchResult]:
"""Return candidates unchanged when phrase matching is disabled."""
logger.info("ebook_phrase_boost_skipped candidates=%s", len(candidates))
logger.info(f"ebook_phrase_boost_skipped candidates={len(candidates)}")
return candidates
@@ -205,7 +194,7 @@ async def apply_phrase_mention_boosts(
async with AsyncSession(engine) as session:
phrase_hits = await phrase_hits_for_chunks(session, chunk_ids=chunk_ids, phrase_ids=phrase_ids)
except SQLAlchemyError as error:
logger.warning("ebook_phrase_boost_unavailable error=%s", error)
logger.warning(f"ebook_phrase_boost_unavailable {error=}")
return candidates
if not phrase_hits:
@@ -259,9 +248,8 @@ async def parallel_retrieval(
)
logger.info(
"ebook_parallel_retrieval_complete vector_candidates=%s lexical_candidates=%s",
len(vector_results),
len(lexical_results),
f"ebook_parallel_retrieval_complete vector_candidates={len(vector_results)} "
f"lexical_candidates={len(lexical_results)}"
)
return RetrievalResponse(
vector_results=vector_results,
@@ -279,7 +267,7 @@ def skip_rerank(
config: EbookSearchConfig,
) -> SearchResponse:
"""Return fused hybrid results without reranking."""
logger.info("ebook_rerank_skipped candidates=%s", len(candidates))
logger.info(f"ebook_rerank_skipped candidates={len(candidates)}")
return SearchResponse(query=query, results=candidates[: config.top_k], rank_label="Hybrid")
@@ -292,9 +280,8 @@ async def apply_rerank(
"""Rerank already-fused hybrid candidates."""
reranked = await rerank_chunks(client, query, candidates[: config.rerank.candidates], config.rerank)
logger.info(
"ebook_rerank_complete input_candidates=%s returned=%s",
min(len(candidates), config.rerank.candidates),
len(reranked),
f"ebook_rerank_complete input_candidates={min(len(candidates), config.rerank.candidates)} "
f"returned={len(reranked)}"
)
return SearchResponse(
query=query,
@@ -352,10 +339,7 @@ async def vector_candidates(
rows = (await session.execute(statement)).mappings()
results = [search_result_from_row(row) for row in rows]
logger.info(
"ebook_vector_search_complete model=%s dimension=%s candidates=%s",
config.embedding_model,
model.dimension,
len(results),
f"ebook_vector_search_complete {config.embedding_model=} {model.dimension=} candidates={len(results)}"
)
return results
@@ -365,7 +349,7 @@ def bm25_candidates(query: str, config: EbookSearchConfig) -> list[SearchResult]
try:
corpus = load_bm25_corpus(config)
except BM25CorpusUnavailableError as error:
logger.warning("ebook_bm25_index_unavailable_skipping error=%s", error)
logger.warning(f"ebook_bm25_index_unavailable_skipping {error=}")
return []
if not corpus.records:
@@ -380,12 +364,7 @@ def bm25_candidates(query: str, config: EbookSearchConfig) -> list[SearchResult]
]
max_score = results[0].bm25_score if results else 0.0
logger.info(
"ebook_bm25_search_complete corpus=%s candidates=%s max_score=%.6f",
len(corpus.records),
len(results),
max_score,
)
logger.info(f"ebook_bm25_search_complete corpus={len(corpus.records)} candidates={len(results)} {max_score=:.6f}")
return results