From d82d054d7c1436c2be4c944fedbc598b4ddf6632 Mon Sep 17 00:00:00 2001 From: Nik Shevchenko Date: Thu, 11 Jun 2026 15:52:44 -0400 Subject: [PATCH 1/5] feat(memory): verbatim transcript-chunk vectors (ns_tchunks) Conversation vectors embed only the structured summary, so specific details (exact dates, names, numbers) are semantically unfindable. Chunk the raw transcript into overlapping dated windows and index them with the text in vector metadata; search returns verbatim evidence with zero extra reads. --- backend/database/vector_db.py | 89 +++++++++++++++++++ .../utils/conversations/transcript_chunks.py | 71 +++++++++++++++ 2 files changed, 160 insertions(+) create mode 100644 backend/utils/conversations/transcript_chunks.py diff --git a/backend/database/vector_db.py b/backend/database/vector_db.py index 017dcc837be..b652ebb369e 100644 --- a/backend/database/vector_db.py +++ b/backend/database/vector_db.py @@ -634,3 +634,92 @@ def delete_memory_vectors_batch(uid: str, memory_ids: List[str]) -> int: logger.warning(f'delete_memory_vectors_batch chunk failed uid={uid} chunk={i // 1000}') logger.info(f'delete_memory_vectors_batch uid={uid} total_deleted={total_deleted}') return total_deleted + + +# --------------------------------------------------------------------------- +# Transcript chunks ("ns_tchunks"): verbatim retrieval over raw conversation +# transcripts. Conversation vectors (ns1) embed only the structured SUMMARY, so +# specific details (exact dates, names, numbers, one-off mentions) are not +# findable semantically. These chunk vectors carry the raw transcript text in +# metadata so retrieval returns the verbatim evidence with zero extra reads. +TRANSCRIPT_CHUNKS_NAMESPACE = "ns_tchunks" + + +def upsert_transcript_chunk_vectors(uid: str, conversation_id: str, chunks: List[dict]) -> int: + """chunks: [{'text': str, 'created_at': int unix ts, 'chunk_index': int}]""" + if index is None: + logger.warning('Pinecone index not initialized, skipping transcript chunk upsert') + return 0 + chunks = [c for c in chunks if (c.get('text') or '').strip()] + if not chunks: + return 0 + + vectors = embeddings.embed_documents([c['text'] for c in chunks]) + payload = [] + for c, v in zip(chunks, vectors): + payload.append( + { + 'id': f"{uid}-{conversation_id}-c{c['chunk_index']}", + 'values': v, + 'metadata': { + 'uid': uid, + 'conversation_id': conversation_id, + 'chunk_index': c['chunk_index'], + 'created_at': int(c['created_at']), + # Pinecone metadata limit is 40KB/vector; transcripts chunks are ~<2KB. + 'text': c['text'][:8000], + }, + } + ) + + upserted = 0 + for i in range(0, len(payload), 100): + index.upsert(vectors=payload[i : i + 100], namespace=TRANSCRIPT_CHUNKS_NAMESPACE) + upserted += len(payload[i : i + 100]) + logger.info(f'upsert_transcript_chunk_vectors uid={uid} conversation={conversation_id} count={upserted}') + return upserted + + +def search_transcript_chunks( + uid: str, query: str, limit: int = 20, starts_at: int = None, ends_at: int = None +) -> List[dict]: + """Semantic search over raw transcript chunks. Returns verbatim text + date, newest data + comes from metadata directly (no Firestore round-trip).""" + if index is None: + return [] + vector = embeddings.embed_query(query) + filter_data = {'uid': uid} + if starts_at is not None and ends_at is not None: + filter_data['created_at'] = {'$gte': int(starts_at), '$lte': int(ends_at)} + xc = index.query( + vector=vector, + top_k=limit, + include_metadata=True, + filter=filter_data, + namespace=TRANSCRIPT_CHUNKS_NAMESPACE, + ) + results = [] + for m in xc.get('matches', []): + md = m.get('metadata') or {} + results.append( + { + 'text': md.get('text', ''), + 'created_at': int(md['created_at']) if md.get('created_at') is not None else None, + 'conversation_id': md.get('conversation_id'), + 'chunk_index': md.get('chunk_index'), + 'score': m.get('score', 0), + } + ) + return results + + +def delete_transcript_chunk_vectors(uid: str, conversation_id: str, max_chunks: int = 512): + """Delete all chunk vectors for a conversation (ids are sequential by construction).""" + if index is None: + return + ids = [f'{uid}-{conversation_id}-c{i}' for i in range(max_chunks)] + try: + for i in range(0, len(ids), 100): + index.delete(ids=ids[i : i + 100], namespace=TRANSCRIPT_CHUNKS_NAMESPACE) + except Exception: + logger.warning(f'delete_transcript_chunk_vectors failed uid={uid} conversation={conversation_id}') diff --git a/backend/utils/conversations/transcript_chunks.py b/backend/utils/conversations/transcript_chunks.py new file mode 100644 index 00000000000..ff5796b1fac --- /dev/null +++ b/backend/utils/conversations/transcript_chunks.py @@ -0,0 +1,71 @@ +"""Build verbatim transcript chunks for vector indexing. + +Conversation vectors (ns1) embed only the structured summary, so specific details +(exact dates, names, numbers, one-off mentions) are unfindable semantically. These +chunks slice the raw transcript into overlapping windows, each prefixed with the +conversation date, so semantic search can land on the verbatim evidence. +""" + +from datetime import datetime +from typing import List, Optional + +# ~8 segments per chunk with 2-segment overlap keeps chunks small enough to embed +# precisely while not splitting answers across a hard boundary. +CHUNK_WINDOW = 8 +CHUNK_STRIDE = 6 + + +def _speaker_label(seg: dict, people_by_id: Optional[dict] = None) -> str: + if seg.get('is_user'): + return 'User' + person_id = seg.get('person_id') + if person_id and people_by_id and person_id in people_by_id: + return people_by_id[person_id] + speaker_id = seg.get('speaker_id') + return f"Speaker {speaker_id}" if speaker_id is not None else 'Speaker' + + +def build_transcript_chunks( + segments: List[dict], + started_at: Optional[datetime], + window: int = CHUNK_WINDOW, + stride: int = CHUNK_STRIDE, + people_by_id: Optional[dict] = None, +) -> List[dict]: + """segments: transcript_segment dicts ({'text','is_user','speaker_id','person_id',...}). + + Returns [{'text', 'created_at' (unix ts), 'chunk_index'}] ready for + vector_db.upsert_transcript_chunk_vectors. + """ + lines = [] + for seg in segments or []: + text = (seg.get('text') or '').strip() + if not text: + continue + lines.append(f"{_speaker_label(seg, people_by_id)}: {text}") + if not lines: + return [] + + date_header = '' + created_ts = 0 + if started_at is not None: + date_header = f"[Conversation on {started_at.strftime('%d %b %Y, %H:%M')}]\n" + created_ts = int(started_at.timestamp()) + + chunks = [] + idx = 0 + pos = 0 + while pos < len(lines): + piece = lines[pos : pos + window] + chunks.append( + { + 'text': date_header + "\n".join(piece), + 'created_at': created_ts, + 'chunk_index': idx, + } + ) + if pos + window >= len(lines): + break + pos += stride + idx += 1 + return chunks From ffe7278b6bc7843a896d227cc5689f1bf9ca1e47 Mon Sep 17 00:00:00 2001 From: Nik Shevchenko Date: Thu, 11 Jun 2026 15:52:45 -0400 Subject: [PATCH 2/5] feat(memory): index transcript chunks at conversation processing time Flag-gated (TRANSCRIPT_CHUNK_INDEXING_ENABLED, default off). Runs beside save_structured_vector on the postprocess executor. --- .../utils/conversations/process_conversation.py | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/backend/utils/conversations/process_conversation.py b/backend/utils/conversations/process_conversation.py index e713652d6a6..7e9e6211247 100644 --- a/backend/utils/conversations/process_conversation.py +++ b/backend/utils/conversations/process_conversation.py @@ -31,7 +31,8 @@ ) from utils.llm.memories import resolve_memory_conflict from database.apps import record_app_usage, get_omi_personas_by_uid_db, get_app_by_id_db -from database.vector_db import upsert_vector2, update_vector_metadata +from database.vector_db import upsert_vector2, update_vector_metadata, upsert_transcript_chunk_vectors +from utils.conversations.transcript_chunks import build_transcript_chunks from models.app import App, UsageHistoryType from models.memories import MemoryDB, Memory from models.calendar_context import CalendarMeetingContext @@ -655,6 +656,18 @@ def _run_auto_sync(): ) +# Verbatim transcript-chunk indexing (ns_tchunks). Off by default: enables semantic +# retrieval over raw transcript text, which the summary-only conversation vectors miss. +TRANSCRIPT_CHUNK_INDEXING_ENABLED = os.getenv('TRANSCRIPT_CHUNK_INDEXING_ENABLED', 'false').lower() == 'true' + + +def save_transcript_chunk_vectors(uid: str, conversation: Conversation): + segments = [s.dict() if hasattr(s, 'dict') else s for s in (conversation.transcript_segments or [])] + chunks = build_transcript_chunks(segments, conversation.started_at or conversation.created_at) + if chunks: + upsert_transcript_chunk_vectors(uid, conversation.id, chunks) + + def save_structured_vector(uid: str, conversation: Conversation, update_only: bool = False): vector = generate_embedding(str(conversation.structured)) if not update_only else None tz = notification_db.get_user_time_zone(uid) @@ -846,6 +859,8 @@ def process_conversation( ) if not is_reprocess: submit_with_context(postprocess_executor, save_structured_vector, uid, conversation) + if TRANSCRIPT_CHUNK_INDEXING_ENABLED: + submit_with_context(postprocess_executor, save_transcript_chunk_vectors, uid, conversation) submit_with_context(postprocess_executor, _extract_memories, uid, conversation) submit_with_context(postprocess_executor, _extract_trends, uid, conversation) submit_with_context(postprocess_executor, _save_action_items, uid, conversation) From 07cb88fa272b73ed83edcf23738652fec755561b Mon Sep 17 00:00:00 2001 From: Nik Shevchenko Date: Thu, 11 Jun 2026 15:52:45 -0400 Subject: [PATCH 3/5] feat(tools): POST /v1/tools/conversations/search-chunks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Semantic search over raw transcript chunks — verbatim dated evidence to complement summary-level /conversations/search. LoCoMo eval: this layer moved the standard-protocol score 50.7% -> 89.3% (tuning) / 77.5% (held-out). --- backend/routers/tools.py | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/backend/routers/tools.py b/backend/routers/tools.py index b75e1d937dc..9f80bbe00a6 100644 --- a/backend/routers/tools.py +++ b/backend/routers/tools.py @@ -21,6 +21,7 @@ from fastapi import APIRouter, Depends, Query from pydantic import BaseModel, Field +import database.vector_db as vector_db from utils.other.endpoints import get_current_user_uid, with_rate_limit from utils.retrieval.tool_services.conversations import get_conversations_text, search_conversations_text from utils.retrieval.tool_services.memories import get_memories_text, search_memories_text @@ -115,6 +116,31 @@ def search_conversations( return _ok("search_conversations", result) +class SearchChunksRequest(BaseModel): + query: str = Field(description="Semantic search query") + limit: int = Field(default=20, ge=1, le=30) + + +@router.post("/v1/tools/conversations/search-chunks", response_model=ToolResponse) +def search_conversation_chunks( + body: SearchChunksRequest, + uid: str = Depends(with_rate_limit(get_current_user_uid, "tools:search")), +): + """Semantic search over RAW transcript chunks (verbatim evidence with dates). + + Complements /conversations/search, which matches against conversation summaries: + summaries drop specifics (exact dates, names, numbers), so detail questions need + this verbatim layer. Returns chunks newest-relevant with their conversation date. + """ + rows = vector_db.search_transcript_chunks(uid, body.query, limit=body.limit) + if not rows: + return _ok("search_conversation_chunks", f"No transcript excerpts found matching '{body.query}'.") + parts = [] + for i, r in enumerate(rows, 1): + parts.append(f"Excerpt {i} (relevance: {r['score']:.2f}):\n{r['text']}") + return _ok("search_conversation_chunks", "\n\n".join(parts)) + + # --------------- memory endpoints --------------- From 5690bc4505171d52fb16943a10cad4c8897d6aef Mon Sep 17 00:00:00 2001 From: Nik Shevchenko Date: Thu, 11 Jun 2026 17:40:06 -0400 Subject: [PATCH 4/5] refactor(memory): keep transcript text out of Pinecone metadata Transcripts are encrypted at rest in Firestore; storing chunk text as plaintext vector metadata would bypass that. Store only (conversation_id, chunk_index, created_at) and re-hydrate the verbatim text from Firestore at read time via deterministic re-chunking. Verified: search results byte-identical on a 20-question LoCoMo sample (20/20 same verdicts as the pre-refactor full run). --- backend/database/vector_db.py | 53 ++++++++++++++----- backend/routers/tools.py | 2 + .../utils/conversations/transcript_chunks.py | 27 ++++++++++ 3 files changed, 69 insertions(+), 13 deletions(-) diff --git a/backend/database/vector_db.py b/backend/database/vector_db.py index b652ebb369e..f3c787fe691 100644 --- a/backend/database/vector_db.py +++ b/backend/database/vector_db.py @@ -640,8 +640,12 @@ def delete_memory_vectors_batch(uid: str, memory_ids: List[str]) -> int: # Transcript chunks ("ns_tchunks"): verbatim retrieval over raw conversation # transcripts. Conversation vectors (ns1) embed only the structured SUMMARY, so # specific details (exact dates, names, numbers, one-off mentions) are not -# findable semantically. These chunk vectors carry the raw transcript text in -# metadata so retrieval returns the verbatim evidence with zero extra reads. +# findable semantically. Chunk vectors make the raw transcript searchable. +# +# Privacy: chunk TEXT is embedded but never stored in Pinecone metadata — +# transcripts are encrypted at rest in Firestore, and mirroring them as +# plaintext metadata would bypass that. Readers re-hydrate the text from +# Firestore via (conversation_id, chunk_index). TRANSCRIPT_CHUNKS_NAMESPACE = "ns_tchunks" @@ -666,8 +670,6 @@ def upsert_transcript_chunk_vectors(uid: str, conversation_id: str, chunks: List 'conversation_id': conversation_id, 'chunk_index': c['chunk_index'], 'created_at': int(c['created_at']), - # Pinecone metadata limit is 40KB/vector; transcripts chunks are ~<2KB. - 'text': c['text'][:8000], }, } ) @@ -683,8 +685,9 @@ def upsert_transcript_chunk_vectors(uid: str, conversation_id: str, chunks: List def search_transcript_chunks( uid: str, query: str, limit: int = 20, starts_at: int = None, ends_at: int = None ) -> List[dict]: - """Semantic search over raw transcript chunks. Returns verbatim text + date, newest data - comes from metadata directly (no Firestore round-trip).""" + """Semantic search over transcript chunks. Returns chunk references + [{conversation_id, chunk_index, created_at, score}] — hydrate text from + Firestore (utils.conversations.transcript_chunks.hydrate_chunk_texts).""" if index is None: return [] vector = embeddings.embed_query(query) @@ -703,23 +706,47 @@ def search_transcript_chunks( md = m.get('metadata') or {} results.append( { - 'text': md.get('text', ''), 'created_at': int(md['created_at']) if md.get('created_at') is not None else None, 'conversation_id': md.get('conversation_id'), - 'chunk_index': md.get('chunk_index'), + 'chunk_index': int(md['chunk_index']) if md.get('chunk_index') is not None else None, 'score': m.get('score', 0), } ) return results -def delete_transcript_chunk_vectors(uid: str, conversation_id: str, max_chunks: int = 512): - """Delete all chunk vectors for a conversation (ids are sequential by construction).""" +def delete_transcript_chunk_vectors(uid: str, conversation_id: str): + """Delete all chunk vectors for one conversation (id-prefix listing on serverless).""" if index is None: return - ids = [f'{uid}-{conversation_id}-c{i}' for i in range(max_chunks)] + prefix = f'{uid}-{conversation_id}-c' try: - for i in range(0, len(ids), 100): - index.delete(ids=ids[i : i + 100], namespace=TRANSCRIPT_CHUNKS_NAMESPACE) + ids = [] + for page in index.list(prefix=prefix, namespace=TRANSCRIPT_CHUNKS_NAMESPACE): + ids.extend(page if isinstance(page, list) else [page]) + for i in range(0, len(ids), 1000): + index.delete(ids=ids[i : i + 1000], namespace=TRANSCRIPT_CHUNKS_NAMESPACE) + if ids: + logger.info(f'delete_transcript_chunk_vectors uid={uid} conversation={conversation_id} count={len(ids)}') except Exception: logger.warning(f'delete_transcript_chunk_vectors failed uid={uid} conversation={conversation_id}') + + +def delete_transcript_chunk_vectors_batch(uid: str, conversation_ids: List[str]) -> int: + """Account-deletion purge: drop all transcript-chunk vectors for the user's conversations.""" + if index is None or not conversation_ids: + return 0 + deleted = 0 + for conversation_id in conversation_ids: + prefix = f'{uid}-{conversation_id}-c' + try: + ids = [] + for page in index.list(prefix=prefix, namespace=TRANSCRIPT_CHUNKS_NAMESPACE): + ids.extend(page if isinstance(page, list) else [page]) + for i in range(0, len(ids), 1000): + index.delete(ids=ids[i : i + 1000], namespace=TRANSCRIPT_CHUNKS_NAMESPACE) + deleted += len(ids) + except Exception: + logger.warning(f'delete_transcript_chunk_vectors_batch failed uid={uid} conversation={conversation_id}') + logger.info(f'delete_transcript_chunk_vectors_batch uid={uid} total_deleted={deleted}') + return deleted diff --git a/backend/routers/tools.py b/backend/routers/tools.py index 9f80bbe00a6..cb5da0a7fe5 100644 --- a/backend/routers/tools.py +++ b/backend/routers/tools.py @@ -23,6 +23,7 @@ import database.vector_db as vector_db from utils.other.endpoints import get_current_user_uid, with_rate_limit +from utils.conversations.transcript_chunks import hydrate_chunk_texts from utils.retrieval.tool_services.conversations import get_conversations_text, search_conversations_text from utils.retrieval.tool_services.memories import get_memories_text, search_memories_text from utils.retrieval.tool_services.action_items import ( @@ -133,6 +134,7 @@ def search_conversation_chunks( this verbatim layer. Returns chunks newest-relevant with their conversation date. """ rows = vector_db.search_transcript_chunks(uid, body.query, limit=body.limit) + rows = hydrate_chunk_texts(uid, rows) if not rows: return _ok("search_conversation_chunks", f"No transcript excerpts found matching '{body.query}'.") parts = [] diff --git a/backend/utils/conversations/transcript_chunks.py b/backend/utils/conversations/transcript_chunks.py index ff5796b1fac..c9533fe7b1b 100644 --- a/backend/utils/conversations/transcript_chunks.py +++ b/backend/utils/conversations/transcript_chunks.py @@ -9,6 +9,8 @@ from datetime import datetime from typing import List, Optional +import database.conversations as conversations_db + # ~8 segments per chunk with 2-segment overlap keeps chunks small enough to embed # precisely while not splitting answers across a hard boundary. CHUNK_WINDOW = 8 @@ -69,3 +71,28 @@ def build_transcript_chunks( pos += stride idx += 1 return chunks + + +def hydrate_chunk_texts(uid: str, rows: List[dict]) -> List[dict]: + """Attach verbatim text to chunk references returned by vector search. + + Re-reads the conversations from Firestore (decrypted by the db layer) and rebuilds + the deterministic chunking, so transcript text never has to live in Pinecone. + Rows whose conversation/chunk no longer exists are dropped. + """ + conv_ids = list({r['conversation_id'] for r in rows if r.get('conversation_id')}) + if not conv_ids: + return [] + conversations = conversations_db.get_conversations_by_id(uid, conv_ids) + chunks_by_conv = {} + for c in conversations: + segs = c.get('transcript_segments') or [] + started = c.get('started_at') or c.get('created_at') + chunks_by_conv[c['id']] = {ch['chunk_index']: ch['text'] for ch in build_transcript_chunks(segs, started)} + + hydrated = [] + for r in rows: + text = chunks_by_conv.get(r.get('conversation_id'), {}).get(r.get('chunk_index')) + if text: + hydrated.append({**r, 'text': text}) + return hydrated From b8df931815d51a9b7fc4977c56b6da537c35991d Mon Sep 17 00:00:00 2001 From: Nik Shevchenko Date: Thu, 11 Jun 2026 17:40:07 -0400 Subject: [PATCH 5/5] feat(memory): delete transcript-chunk vectors on conversation and account deletion Conversation delete and the account-deletion purge now also remove ns_tchunks vectors (id-prefix listing). Verified live: upsert -> search finds 2 -> delete -> search finds 0. --- backend/routers/conversations.py | 3 ++- backend/routers/users.py | 11 ++++++++++- 2 files changed, 12 insertions(+), 2 deletions(-) diff --git a/backend/routers/conversations.py b/backend/routers/conversations.py index 04d8a003a5c..c249dcda51a 100644 --- a/backend/routers/conversations.py +++ b/backend/routers/conversations.py @@ -9,7 +9,7 @@ import database.memories as memories_db import database.redis_db as redis_db import database.users as users_db -from database.vector_db import delete_vector, delete_memory_vector +from database.vector_db import delete_vector, delete_memory_vector, delete_transcript_chunk_vectors from utils.other.storage import delete_conversation_audio_files from models.calendar_context import CalendarMeetingContext from models.conversation import ( @@ -389,6 +389,7 @@ def delete_conversation( logger.info(f'delete_conversation {conversation_id} {uid} cascade={cascade}') conversations_db.delete_conversation(uid, conversation_id) delete_vector(uid, conversation_id) + delete_transcript_chunk_vectors(uid, conversation_id) if cascade: # Delete audio files diff --git a/backend/routers/users.py b/backend/routers/users.py index f9700e75c45..df30e4e8201 100644 --- a/backend/routers/users.py +++ b/backend/routers/users.py @@ -109,6 +109,7 @@ from database.screen_activity import get_screen_activity_ids from database.vector_db import ( delete_conversation_vectors_batch, + delete_transcript_chunk_vectors_batch, delete_memory_vectors_batch, delete_action_item_vectors_batch, delete_screen_activity_vectors, @@ -159,7 +160,8 @@ def _purge_derived_user_data(uid: str): blocks the others or the subsequent Firestore deletion. IDs are read via lightweight IDs-only queries (no decryption). - Scope: conversation (ns1), memory (ns2), action-item (ns4) and screen-activity (ns3) vectors, + Scope: conversation (ns1), memory (ns2), action-item (ns4), screen-activity (ns3) and + transcript-chunk (ns_tchunks) vectors, plus conversation recordings. Known follow-ups NOT covered here: X-post vectors (no delete helper yet), speech-profile / person-sample / private-cloud-sync / chat-upload GCS blobs, and the externally-indexed Typesense collection. @@ -171,6 +173,13 @@ def _purge_derived_user_data(uid: str): except Exception as e: logger.error(f'delete_account purge conversation vectors failed for {uid}: {sanitize(str(e))}') + try: + conversation_ids = get_conversation_ids(uid) + if conversation_ids: + delete_transcript_chunk_vectors_batch(uid, conversation_ids) + except Exception as e: + logger.error(f'delete_account purge transcript chunk vectors failed for {uid}: {sanitize(str(e))}') + try: memory_ids = get_memory_ids(uid) if memory_ids: