fix: isolate per-text embedding failures in reconciler

The message-embedding reconciler sent all pending texts in a single
simple_batch_embed call. If any one text exceeded the provider's context
window, the entire batch 400'd and every row was marked failed, dragging
down perfectly-embeddable small messages as collateral.

Embed each text individually so a single oversized text fails on its own
and the rest of the batch stays embeddable.
This commit is contained in:
cadamec 2026-08-08 18:10:40 -05:00
parent 3ee890fa6f
commit e70537cbfb
1 changed files with 25 additions and 27 deletions

View File

@ -425,33 +425,31 @@ async def _sync_message_embeddings(
freshly_embedded: dict[int, list[float]] = {}
if embs_needing_embed:
try:
contents = [emb.content for emb in embs_needing_embed]
# MESSAGE_CREATE (not VECTOR_SYNC): these rows come from create_messages
# as pending chunks; document re-embeds stay on VECTOR_SYNC below.
workspaces = {emb.workspace_name for emb in embs_needing_embed}
with embedding_call_purpose(
EmbeddingCallPurpose.MESSAGE_CREATE.value,
workspace_name=workspaces.pop() if len(workspaces) == 1 else None,
parent_category="reconciliation",
):
new_embeddings = await embedding_client.simple_batch_embed(contents)
if len(new_embeddings) != len(embs_needing_embed):
logger.warning(
"Re-embedded %s/%s message embeddings; remaining will be retried",
len(new_embeddings),
len(embs_needing_embed),
)
for emb, new_emb in zip(embs_needing_embed, new_embeddings, strict=False):
freshly_embedded[emb.id] = new_emb
if store_in_postgres:
emb.embedding = new_emb
except Exception:
logger.exception(
"Failed to re-embed %s message embeddings", len(embs_needing_embed)
)
# Embed each text individually so a single oversized text (rejected by
# the provider) can't poison the whole batch. The chunking tokenizer
# can undercount a provider's real token count (e.g. tiktoken o200k vs
# nomic-embed-text), so oversized texts can still slip through the chunk
# cap and get rejected by the provider. Isolating per-text keeps the
# rest of the batch embeddable and lets genuinely-oversized rows fail
# on their own instead of taking the whole batch down with them.
workspaces = {emb.workspace_name for emb in embs_needing_embed}
with embedding_call_purpose(
EmbeddingCallPurpose.MESSAGE_CREATE.value,
workspace_name=workspaces.pop() if len(workspaces) == 1 else None,
parent_category="reconciliation",
):
for emb in embs_needing_embed:
try:
new_emb = await embedding_client.simple_batch_embed([emb.content])
freshly_embedded[emb.id] = new_emb[0]
if store_in_postgres:
emb.embedding = new_emb[0]
except Exception:
logger.warning(
"Failed to embed message %s chunk %s; will retry",
emb.message_id,
emb.id,
)
# Mark embeddings that failed to get a vector
failed_to_embed: list[models.MessageEmbedding] = [