feat: 优化问问微信检索性能与分析交互

- 补充 Worker、WCDB、sender、IPC、序列化时间账
- 增加 Agent 增量覆盖统计和重复检索停止条件
- 补充性能与交互回归测试

src/main/knowledge/knowledge-search-service.ts

@@ -21,11 +21,16 @@ import {
   emptyKnowledgeSearchTimings
 } from '../../shared/knowledge'
 import { KnowledgeService } from './knowledge-service'
-import { voiceAccountIdentity, voiceMessageIdentity } from '../voice-pipeline/voice-message-identity'
+import {
+  voiceAccountIdentity,
+  voiceMessageIdentity
+} from '../voice-pipeline/voice-message-identity'
 
 const FALLBACK_LIMIT = 240
 const MAX_SENDER_NAME_CONVERSATIONS = 8
 const MAX_CONVERSATION_FILTERS_PER_WORKER_SEARCH = 700
+const MAX_SENDER_ENRICHMENT_SESSIONS = 32
+const SENDER_ENRICHMENT_SESSION_TTL_MS = 5 * 60 * 1000
 
 type PendingVoiceTranscriptIndex = {
   update: VoiceTranscriptUpdate
@@ -35,6 +40,12 @@ type PendingVoiceTranscriptIndex = {
   }>
 }
 
+type SenderEnrichmentSession = {
+  lastUsedAt: number
+  contacts?: Awaited<ReturnType<typeof chat.listContactsAsync>>
+  groupSnapshots: Map<string, Awaited<ReturnType<typeof chat.getGroupSnapshotAsync>> | undefined>
+}
+
 function looksLikeOpaqueSenderId(value: string | undefined): boolean {
   const normalized = value?.trim() || ''
   return (
@@ -177,7 +188,10 @@ export class KnowledgeSearchService {
   private readonly indexing = new Map<string, Promise<void>>()
   private readonly statusByAccount = new Map<string, KnowledgeRuntimeStatus>()
   private readonly statusListeners = new Set<(status: KnowledgeRuntimeStatus) => void>()
+  private readonly senderEnrichmentSessions = new Map<string, SenderEnrichmentSession>()
   private wcdbReadTail: Promise<void> = Promise.resolve()
+  private wcdbQueueMsTotal = 0
+  private wcdbExecutionMsTotal = 0
   private voiceTranscriptResolver:
     | ((reference: VoiceMessageReference) => VoiceTranscriptSnapshot)
     | undefined
@@ -310,7 +324,7 @@ export class KnowledgeSearchService {
       // An existing derived database can answer while its next incremental pass is running.
       // Never turn an interactive global search into another full WCDB scan during that pass.
       if (result.state === 'ready' || result.evidence.length) {
-        return this.toKnowledgeResult(result)
+        return this.toKnowledgeResult(result, request.retrievalSessionId)
       }
       if (this.indexing.has(accountId)) {
         return {
@@ -494,9 +508,19 @@ export class KnowledgeSearchService {
         score: -score
       }))
     }
+    const beforeQueueMs = this.wcdbQueueMsTotal
+    const beforeExecutionMs = this.wcdbExecutionMsTotal
+    const enrichmentStartedAt = Date.now()
+    const evidence = await this.enrichEvidenceSenders(result.evidence, request.retrievalSessionId)
     return {
       ...result,
-      evidence: await this.enrichEvidenceSenders(result.evidence)
+      evidence,
+      timings: {
+        ...result.timings,
+        senderEnrichmentMs: Date.now() - enrichmentStartedAt,
+        wcdbQueueMs: this.wcdbQueueMsTotal - beforeQueueMs,
+        wcdbExecutionMs: this.wcdbExecutionMsTotal - beforeExecutionMs
+      }
     }
   }
 
@@ -556,7 +580,17 @@ export class KnowledgeSearchService {
         messageLoadMs: total.messageLoadMs + (result.timings?.messageLoadMs || 0),
         chunkExpandMs: total.chunkExpandMs + (result.timings?.chunkExpandMs || 0),
         rankingMs: total.rankingMs + (result.timings?.rankingMs || 0),
-        totalMs: total.totalMs + (result.timings?.totalMs || 0)
+        totalMs: total.totalMs + (result.timings?.totalMs || 0),
+        globalCountMs: (total.globalCountMs || 0) + (result.timings?.globalCountMs || 0),
+        voiceCoverageMs: (total.voiceCoverageMs || 0) + (result.timings?.voiceCoverageMs || 0),
+        workerExecutionMs:
+          (total.workerExecutionMs || 0) +
+          (result.timings?.workerExecutionMs || result.timings?.totalMs || 0),
+        workerQueueMs: (total.workerQueueMs || 0) + (result.timings?.workerQueueMs || 0),
+        ipcMs: (total.ipcMs || 0) + (result.timings?.ipcMs || result.timings?.workerIpcMs || 0),
+        serializationMs:
+          (total.serializationMs || 0) +
+          (result.timings?.serializationMs || result.timings?.responseSerializeMs || 0)
       }),
       emptyKnowledgeSearchTimings()
     )
@@ -606,12 +640,23 @@ export class KnowledgeSearchService {
     const startedAt = Date.now()
     const result = await this.service.search(request)
     const timings = result.timings || emptyKnowledgeSearchTimings()
+    const workerExecutionMs = timings.workerExecutionMs ?? timings.totalMs
+    const ipcMs = timings.ipcMs ?? timings.workerIpcMs
+    const serializationMs = timings.serializationMs ?? timings.responseSerializeMs
     return {
       ...result,
       timings: {
         ...timings,
-        workerIpcMs: timings.workerIpcMs || Math.max(0, Date.now() - startedAt - timings.totalMs),
-        workerSqlMs: timings.workerSqlMs || timings.totalMs
+        // Do not infer IPC by subtracting the Worker timer from wall clock:
+        // that previously hid unmeasured Worker execution inside “通信”.
+        workerIpcMs: timings.workerIpcMs,
+        ipcMs,
+        workerSqlMs: timings.workerSqlMs || timings.totalMs,
+        workerExecutionMs,
+        serializationMs,
+        otherMs:
+          timings.otherMs ??
+          Math.max(0, Date.now() - startedAt - workerExecutionMs - ipcMs - serializationMs)
       }
     }
   }
@@ -646,12 +691,7 @@ export class KnowledgeSearchService {
     const reference = this.voiceReferenceFromMessage(message)
     const snapshot = reference ? this.voiceTranscriptResolver?.(reference) : undefined
     const hydrated = this.withVoiceTranscript(message)
-    const source = toSourceMessage(
-      accountId,
-      conversationId,
-      hydrated,
-      transcriptOverride
-    )
+    const source = toSourceMessage(accountId, conversationId, hydrated, transcriptOverride)
     if (!source || source.kind !== 'voice') return source
     return {
       ...source,
@@ -666,7 +706,12 @@ export class KnowledgeSearchService {
   private voiceReferenceFromMessage(
     message: chat.FormattedMessage
   ): VoiceMessageReference | undefined {
-    if (message.type !== '语音' || !message.sessionId || message.localId === undefined || !message.createTime) {
+    if (
+      message.type !== '语音' ||
+      !message.sessionId ||
+      message.localId === undefined ||
+      !message.createTime
+    ) {
       return undefined
     }
     return {
@@ -730,17 +775,31 @@ export class KnowledgeSearchService {
   }
 
   private async toKnowledgeResult(
-    result: KnowledgeSearchResult
+    result: KnowledgeSearchResult,
+    retrievalSessionId?: string
   ): Promise<KnowledgeSearchIpcResult> {
+    const beforeQueueMs = this.wcdbQueueMsTotal
+    const beforeExecutionMs = this.wcdbExecutionMsTotal
+    const enrichmentStartedAt = Date.now()
+    const evidence = await this.enrichEvidenceSenders(result.evidence, retrievalSessionId)
     return {
       ...result,
-      evidence: await this.enrichEvidenceSenders(result.evidence),
+      evidence,
+      timings: {
+        ...result.timings,
+        senderEnrichmentMs: Date.now() - enrichmentStartedAt,
+        wcdbQueueMs: this.wcdbQueueMsTotal - beforeQueueMs,
+        wcdbExecutionMs: this.wcdbExecutionMsTotal - beforeExecutionMs
+      },
       source: 'knowledge',
       totalMessages: result.indexedMessageCount
     }
   }
 
-  private async enrichEvidenceSenders(evidence: KnowledgeEvidence[]): Promise<KnowledgeEvidence[]> {
+  private async enrichEvidenceSenders(
+    evidence: KnowledgeEvidence[],
+    retrievalSessionId?: string
+  ): Promise<KnowledgeEvidence[]> {
     const candidateConversationIds = Array.from(
       new Set(
         evidence
@@ -750,14 +809,22 @@ export class KnowledgeSearchService {
     ).slice(0, MAX_SENDER_NAME_CONVERSATIONS)
     if (!candidateConversationIds.length) return evidence
 
-    const contacts = await this.listContacts()
+    const session = retrievalSessionId
+      ? this.senderEnrichmentSession(retrievalSessionId)
+      : undefined
+    const contacts = session?.contacts || (await this.listContacts())
+    if (session && !session.contacts) session.contacts = contacts
     const groupConversationIds = new Set(
       contacts.filter((contact) => contact.type === 'group').map((contact) => contact.md5)
     )
     const memberNamesByConversation = new Map<string, Map<string, string>>()
     for (const conversationId of candidateConversationIds) {
       if (!groupConversationIds.has(conversationId)) continue
-      const snapshot = await this.enqueueWcdbRead(() => chat.getGroupSnapshotAsync(conversationId))
+      let snapshot = session?.groupSnapshots.get(conversationId)
+      if (!snapshot) {
+        snapshot = await this.enqueueWcdbRead(() => chat.getGroupSnapshotAsync(conversationId))
+        session?.groupSnapshots.set(conversationId, snapshot)
+      }
       const memberNames = new Map(
         (snapshot?.members || [])
           .map((member) => [member.wxid, groupMemberDisplayName(member)] as const)
@@ -772,8 +839,39 @@ export class KnowledgeSearchService {
     })
   }
 
+  private senderEnrichmentSession(retrievalSessionId: string): SenderEnrichmentSession {
+    const now = Date.now()
+    for (const [key, value] of this.senderEnrichmentSessions) {
+      if (now - value.lastUsedAt > SENDER_ENRICHMENT_SESSION_TTL_MS) {
+        this.senderEnrichmentSessions.delete(key)
+      }
+    }
+    let session = this.senderEnrichmentSessions.get(retrievalSessionId)
+    if (!session) {
+      session = { lastUsedAt: now, groupSnapshots: new Map() }
+      this.senderEnrichmentSessions.set(retrievalSessionId, session)
+    }
+    session.lastUsedAt = now
+    while (this.senderEnrichmentSessions.size > MAX_SENDER_ENRICHMENT_SESSIONS) {
+      const oldest = this.senderEnrichmentSessions.keys().next().value as string | undefined
+      if (!oldest) break
+      this.senderEnrichmentSessions.delete(oldest)
+    }
+    return session
+  }
+
   private enqueueWcdbRead<T>(operation: () => Promise<T>): Promise<T> {
-    const result = this.wcdbReadTail.then(operation, operation)
+    const enqueuedAt = Date.now()
+    const run = async (): Promise<T> => {
+      const startedAt = Date.now()
+      this.wcdbQueueMsTotal += Math.max(0, startedAt - enqueuedAt)
+      try {
+        return await operation()
+      } finally {
+        this.wcdbExecutionMsTotal += Date.now() - startedAt
+      }
+    }
+    const result = this.wcdbReadTail.then(run, run)
     // Keep the queue usable after a read failure while returning that failure to its caller.
     this.wcdbReadTail = result.then(
       () => undefined,

src/main/knowledge/knowledge-store.ts

@@ -137,6 +137,7 @@ export class KnowledgeStore {
   private readonly database: DatabaseSync
   private readonly databasePath: string
   private readonly ftsExternalContent: boolean
+  private pendingStatsRefreshMs = 0
 
   constructor(
     private readonly databaseRoot: string,
@@ -203,6 +204,7 @@ export class KnowledgeStore {
     let indexedChunks = 0
     let updatedChunks = 0
     let unchangedConversations = 0
+    this.markStatsStale()
     this.setRunState('indexing')
     try {
       for (const conversation of request.conversations) {
@@ -238,6 +240,7 @@ export class KnowledgeStore {
       }
       if (request.sourceMessageCount !== undefined) {
         this.writeMeta('source_message_count', String(request.sourceMessageCount))
+        this.refreshStatsSnapshot()
       }
       this.setRunState('ready')
       return {
@@ -257,6 +260,8 @@ export class KnowledgeStore {
         cancelled ? 'cancelled' : 'error',
         error instanceof Error ? error.message : String(error)
       )
+      // Per-conversation transactions may already have committed. Keep the
+      // snapshot stale so the next status/search/open reconciles it once.
       if (cancelled) {
         return {
           accountId: this.accountId,
@@ -288,13 +293,9 @@ export class KnowledgeStore {
   }
 
   getSearchStatus(): Omit<KnowledgeSearchResult, 'evidence'> {
-    const indexedMessageCount = Number(
-      (this.database.prepare('SELECT COUNT(*) AS count FROM knowledge_messages').get() as DbRow)
-        .count
-    )
-    const indexedChunkCount = Number(
-      (this.database.prepare('SELECT COUNT(*) AS count FROM knowledge_chunks').get() as DbRow).count
-    )
+    this.ensureStatsSnapshot()
+    const indexedMessageCount = this.readStatNumber('stats_message_count')
+    const indexedChunkCount = this.readStatNumber('stats_chunk_count')
     const runState = this.readMeta('run_state')
     return {
       state:
@@ -667,7 +668,7 @@ export class KnowledgeStore {
       evidence: asRows(
         this.database
           .prepare(
-          `SELECT m.conversation_id, m.message_id, m.create_time, m.searchable_text, m.kind, m.sender_id, m.sender_name
+            `SELECT m.conversation_id, m.message_id, m.create_time, m.searchable_text, m.kind, m.sender_id, m.sender_name
            FROM knowledge_messages m
            WHERE ${clauses.join(' AND ')}
            ORDER BY m.create_time DESC
@@ -701,16 +702,28 @@ export class KnowledgeStore {
   }
 
   searchWithStatus(query: KnowledgeQuery): KnowledgeSearchResult {
+    const startedAt = Date.now()
     const status = this.getSearchStatus()
     const measured = status.indexedChunkCount > 0 ? this.searchMeasured(query) : null
+    const voiceStartedAt = Date.now()
+    const voiceCoverage = this.getVoiceCoverage(query)
+    const voiceCoverageMs = Date.now() - voiceStartedAt
+    const workerExecutionMs = Date.now() - startedAt
+    const statsRefreshMs = this.consumeStatsRefreshMs()
     return {
       ...status,
       // A long incremental pass can already have durable chunks. Those chunks are
       // safe to query and avoid falling back to a second scan of the source archive.
       evidence: measured?.evidence || [],
-      timings: measured?.timings || emptyKnowledgeSearchTimings(),
+      timings: {
+        ...(measured?.timings || emptyKnowledgeSearchTimings()),
+        totalMs: workerExecutionMs,
+        globalCountMs: statsRefreshMs,
+        voiceCoverageMs,
+        workerExecutionMs
+      },
       conversationRetrieval: measured?.conversationRetrieval,
-      voiceCoverage: this.getVoiceCoverage(query)
+      voiceCoverage
     }
   }
 
@@ -718,8 +731,24 @@ export class KnowledgeStore {
     const clauses = ["kind = 'voice'"]
     const values: (string | number)[] = []
     const conversationIds = Array.from(
-      new Set([...(query.conversationIds || []), ...(query.conversationId ? [query.conversationId] : [])])
+      new Set([
+        ...(query.conversationIds || []),
+        ...(query.conversationId ? [query.conversationId] : [])
+      ])
     ).filter(Boolean)
+    // The common global query can use the same truthful snapshot as status.
+    // Scoped or time-bounded coverage remains a real SQL aggregation because
+    // its answer depends on the requested slice.
+    if (!conversationIds.length && query.startTime === undefined && query.endTime === undefined) {
+      return {
+        voiceMessageCount: this.readStatNumber('stats_voice_message_count'),
+        transcribedVoiceCount: this.readStatNumber('stats_transcribed_voice_count'),
+        failedVoiceCount: this.readStatNumber('stats_failed_voice_count'),
+        voiceCoverageComplete:
+          this.readStatNumber('stats_voice_message_count') ===
+          this.readStatNumber('stats_transcribed_voice_count')
+      }
+    }
     if (conversationIds.length) {
       clauses.push(`conversation_id IN (${conversationIds.map(() => '?').join(', ')})`)
       values.push(...conversationIds)
@@ -828,6 +857,7 @@ export class KnowledgeStore {
     }
     this.writeMetaIfMissing('fts_config', fingerprint)
     this.createFtsTable()
+    this.ensureStatsSnapshot()
   }
 
   private createFtsTable(): void {
@@ -1142,6 +1172,61 @@ export class KnowledgeStore {
       .run(key, value)
   }
 
+  /**
+   * Counts are a database-state snapshot, not a per-search query. The snapshot
+   * is marked stale before indexing and refreshed only after the final request
+   * in a complete source pass. If the process is reopened while stale (for
+   * example after an incremental update, cancellation or crash), the first
+   * Worker operation reconciles it once before serving status/search.
+   */
+  private ensureStatsSnapshot(): void {
+    if (this.readMeta('stats_state') === 'fresh') return
+    this.refreshStatsSnapshot()
+  }
+
+  private markStatsStale(): void {
+    this.writeMeta('stats_state', 'stale')
+  }
+
+  private refreshStatsSnapshot(): void {
+    const startedAt = Date.now()
+    const messageCount = Number(
+      (this.database.prepare('SELECT COUNT(*) AS count FROM knowledge_messages').get() as DbRow)
+        .count
+    )
+    const chunkCount = Number(
+      (this.database.prepare('SELECT COUNT(*) AS count FROM knowledge_chunks').get() as DbRow).count
+    )
+    const voice = this.database
+      .prepare(
+        `SELECT COUNT(*) AS total,
+                SUM(CASE WHEN voice_transcript IS NOT NULL AND trim(voice_transcript) <> '' THEN 1 ELSE 0 END) AS transcribed,
+                SUM(CASE WHEN voice_transcript_state = 'failed' THEN 1 ELSE 0 END) AS failed
+         FROM knowledge_messages WHERE kind = 'voice'`
+      )
+      .get() as DbRow
+    this.writeMeta('stats_message_count', String(messageCount))
+    this.writeMeta('stats_chunk_count', String(chunkCount))
+    this.writeMeta('stats_voice_message_count', String(Number(voice.total || 0)))
+    this.writeMeta('stats_transcribed_voice_count', String(Number(voice.transcribed || 0)))
+    this.writeMeta('stats_failed_voice_count', String(Number(voice.failed || 0)))
+    this.writeMeta('stats_updated_at', String(Date.now()))
+    this.writeMeta('stats_state', 'fresh')
+    this.pendingStatsRefreshMs += Date.now() - startedAt
+  }
+
+  private readStatNumber(key: string): number {
+    const raw = this.readMeta(key)
+    const value = raw === null ? NaN : Number(raw)
+    return Number.isFinite(value) && value >= 0 ? value : 0
+  }
+
+  private consumeStatsRefreshMs(): number {
+    const value = this.pendingStatsRefreshMs
+    this.pendingStatsRefreshMs = 0
+    return value
+  }
+
   private databaseBytes(): number {
     return existsSync(this.databasePath) ? statSync(this.databasePath).size : 0
   }

src/main/knowledge/knowledge-worker-host.ts

@@ -169,8 +169,12 @@ export class KnowledgeWorkerHost {
         workerBootMs,
         dispatchMs,
         workerSqlMs: result.timings.totalMs,
+        workerExecutionMs: result.timings.workerExecutionMs ?? result.timings.totalMs,
+        workerQueueMs: transport.workerQueueMs ?? 0,
         responseSerializeMs: transport.responseSerializeMs,
         responseTransferMs,
+        serializationMs: transport.responseSerializeMs,
+        ipcMs: workerBootMs + dispatchMs + responseTransferMs,
         workerIpcMs: workerBootMs + dispatchMs + responseTransferMs
       }
     }

src/main/knowledge/knowledge-worker.ts

@@ -30,7 +30,8 @@ function send(
 function sendSearchResult(
   request: KnowledgeWorkerRequest,
   payload: KnowledgeWorkerResponse['payload'],
-  workerReceivedAt: number
+  workerReceivedAt: number,
+  workerQueueMs: number
 ): void {
   const serializeStartedAt = Date.now()
   // This measures the actual payload encoding workload before Node IPC performs
@@ -39,7 +40,13 @@ function sendSearchResult(
   const responseSerializeMs = Date.now() - serializeStartedAt
   send(
     { version: 1, type: 'result', requestId: request.requestId, payload },
-    { workerReceivedAt, workerCompletedAt: Date.now(), responseSerializeMs }
+    {
+      messageReceivedAt: workerReceivedAt - workerQueueMs,
+      workerReceivedAt,
+      workerCompletedAt: Date.now(),
+      responseSerializeMs,
+      workerQueueMs
+    }
   )
 }
 
@@ -92,9 +99,12 @@ async function handlePreflight(
 
 async function handleSearch(
   request: KnowledgeWorkerRequest,
-  payload: KnowledgeSearchRequest
+  payload: KnowledgeSearchRequest,
+  messageReceivedAt: number
 ): Promise<void> {
   const workerReceivedAt = Date.now()
+  const workerQueueMs = Math.max(0, workerReceivedAt - messageReceivedAt)
+  const workerExecutionStartedAt = Date.now()
   const path = getKnowledgeDatabasePath(payload.databaseRoot, payload.accountId)
   if (!existsSync(path)) {
     sendSearchResult(
@@ -106,12 +116,24 @@ async function handleSearch(
         indexedChunkCount: 0,
         timings: emptyKnowledgeSearchTimings()
       },
-      workerReceivedAt
+      workerReceivedAt,
+      workerQueueMs
     )
     return
   }
   const result = getStore(payload).searchWithStatus(payload)
-  sendSearchResult(request, result, workerReceivedAt)
+  sendSearchResult(
+    request,
+    {
+      ...result,
+      timings: {
+        ...result.timings,
+        workerExecutionMs: Date.now() - workerExecutionStartedAt
+      }
+    },
+    workerReceivedAt,
+    workerQueueMs
+  )
 }
 
 async function handleStatus(
@@ -144,7 +166,7 @@ async function handleStatus(
   })
 }
 
-async function handle(request: KnowledgeWorkerRequest): Promise<void> {
+async function handle(request: KnowledgeWorkerRequest, messageReceivedAt: number): Promise<void> {
   try {
     if (request.type === 'cancel') {
       const payload = request.payload as { targetRequestId: string }
@@ -172,7 +194,7 @@ async function handle(request: KnowledgeWorkerRequest): Promise<void> {
       return
     }
     if (request.type === 'search') {
-      await handleSearch(request, request.payload as KnowledgeSearchRequest)
+      await handleSearch(request, request.payload as KnowledgeSearchRequest, messageReceivedAt)
       return
     }
     if (request.type === 'status') {
@@ -196,5 +218,5 @@ async function handle(request: KnowledgeWorkerRequest): Promise<void> {
 
 process.on('message', (message: KnowledgeWorkerRequest) => {
   if (message?.version !== 1) return
-  void handle(message)
+  void handle(message, Date.now())
 })