diff --git a/backend/database/vector_db.py b/backend/database/vector_db.py index 017dcc837be..f3c787fe691 100644 --- a/backend/database/vector_db.py +++ b/backend/database/vector_db.py @@ -634,3 +634,119 @@ 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. 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" + + +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']), + }, + } + ) + + 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 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) + 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( + { + 'created_at': int(md['created_at']) if md.get('created_at') is not None else None, + 'conversation_id': md.get('conversation_id'), + '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): + """Delete all chunk vectors for one conversation (id-prefix listing on serverless).""" + if index is None: + return + 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) + 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/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/tools.py b/backend/routers/tools.py index b75e1d937dc..cb5da0a7fe5 100644 --- a/backend/routers/tools.py +++ b/backend/routers/tools.py @@ -21,7 +21,9 @@ 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.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 ( @@ -115,6 +117,32 @@ 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) + rows = hydrate_chunk_texts(uid, rows) + 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 --------------- 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: 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) diff --git a/backend/utils/conversations/transcript_chunks.py b/backend/utils/conversations/transcript_chunks.py new file mode 100644 index 00000000000..c9533fe7b1b --- /dev/null +++ b/backend/utils/conversations/transcript_chunks.py @@ -0,0 +1,98 @@ +"""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 + +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 +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 + + +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