Implement Experience V1 reflect, retrieval boost, and product metrics.

Add rule-based reflect() to aggregate task outcomes, environment-aware search scoring, experience_saved/recalled events, and structured injection blocks per the V1 schema. Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-02 14:30:59 +08:00
parent 6d48480b1c
commit 2ec6fdbe11
11 changed files with 675 additions and 58 deletions
@@ -1,11 +1,24 @@
 import crypto from 'node:crypto';

 import {
-  experienceSearchHaystack,
  mapExperienceRow,
  normalizeExperienceRecordInput,
-  serializeExperienceJson,
 } from './experience-schema.mjs';
+import {
+  DEFAULT_REFLECT_MIN_GROUP_SIZE,
+  planExperienceReflection,
+  scoreExperienceRow,
+} from './experience-reflect.mjs';
+import {
+  MEMORY_V2_PRODUCT_EVENT_TYPES,
+  recordMemoryV2ProductEvent,
+} from './memory-v2-product-events.mjs';
+
+const PG_EXPERIENCE_SELECT = `SELECT id, scope, kind, title, body, tags, source_session_id,
+              source_user_id, use_count, problem, environment_json, hypothesis,
+              action_json, result, confidence, evidence_json, parent_experience_id,
+              supersedes_id, status, created_at, updated_at
+       FROM h5_experience`;

 /**
 * PostgreSQL + pgvector backed experience store (etat C, polyglot mode).
@@ -27,6 +40,7 @@ import {
 * @param {number} [options.embeddingDim=1536] - vector dimension (must match embed)
 * @param {() => number} [options.now]
 * @param {number} [options.recencyHalfLifeMs]
+ * @param {import('mysql2/promise').Pool} [options.productEventsPool] - MySQL pool for product events
 */
 export async function createPgExperienceService(options = {}) {
  const connectionString = options.connectionString;
@@ -34,6 +48,7 @@ export async function createPgExperienceService(options = {}) {
    throw new Error('createPgExperienceService 需要 connectionString（建议用 EXPERIENCE_PG_URL）');
  }
  const now = options.now ?? (() => Date.now());
+  const productEventsPool = options.productEventsPool ?? null;
  const embed = options.embed ?? null;
  const embeddingDim = Math.max(1, Number(options.embeddingDim ?? 1536));
  const halfLifeMs = Math.max(
@@ -110,6 +125,45 @@ export async function createPgExperienceService(options = {}) {
    });
  }

+  async function emitExperienceSaved(saved, input) {
+    if (!productEventsPool?.query) return;
+    await recordMemoryV2ProductEvent(productEventsPool, {
+      eventType: MEMORY_V2_PRODUCT_EVENT_TYPES.EXPERIENCE_SAVED,
+      userId: saved.sourceUserId,
+      sessionId: saved.sourceSessionId,
+      data: {
+        experienceId: saved.id,
+        kind: saved.kind,
+        scope: saved.scope,
+        result: saved.result ?? null,
+        source: input?.source ?? null,
+      },
+      createdAt: saved.createdAt,
+    }).catch(() => {});
+  }
+
+  async function emitExperienceRecalled({ query, hits, scope }) {
+    if (!productEventsPool?.query || !hits.length) return;
+    await recordMemoryV2ProductEvent(productEventsPool, {
+      eventType: MEMORY_V2_PRODUCT_EVENT_TYPES.EXPERIENCE_RECALLED,
+      userId: hits[0]?.sourceUserId ?? null,
+      data: {
+        query: String(query ?? '').slice(0, 500),
+        scope,
+        hitCount: hits.length,
+        experienceIds: hits.map((hit) => hit.id),
+        kinds: hits.map((hit) => hit.kind),
+      },
+    }).catch(() => {});
+  }
+
+  function rankRows(rows, queryTerms, nowMs) {
+    return rows
+      .map((row) => scoreExperienceRow(row, queryTerms, nowMs, halfLifeMs))
+      .filter((entry) => entry.score > 0)
+      .sort((a, b) => b.score - a.score);
+  }
+
  async function record(input) {
    const normalized = normalizeExperienceRecordInput(input);
    if (!normalized.title) throw experienceError('经验标题不能为空', 'invalid_experience_input');
@@ -160,7 +214,7 @@ export async function createPgExperienceService(options = {}) {
        ts,
      ],
    );
-    return rowToExperience({
+    const saved = rowToExperience({
      id,
      scope: normalized.scope,
      kind: normalized.kind,
@@ -183,6 +237,8 @@ export async function createPgExperienceService(options = {}) {
      created_at: ts,
      updated_at: ts,
    });
+    void emitExperienceSaved(saved, input);
+    return saved;
  }

  async function search(query, { scope = 'global', limit = 5 } = {}) {
@@ -208,7 +264,9 @@ export async function createPgExperienceService(options = {}) {
          );
          if (rows.length > 0) {
            await bumpUseCount(rows.map((r) => r.id));
-            return rows.map(rowToExperience);
+            const hits = rows.map(rowToExperience);
+            void emitExperienceRecalled({ query: text, hits, scope });
+            return hits;
          }
        }
      } catch {
@@ -216,24 +274,25 @@ export async function createPgExperienceService(options = {}) {
      }
    }

-    // Keyword fallback: ILIKE any term, ordered by recency.
+    // Keyword fallback: fetch candidate rows, rank with environment/problem boost.
    const terms = tokenize(text);
    if (terms.length === 0) return [];
-    const likeClauses = terms.map((_, i) => `(title ILIKE $${i + 2} OR body ILIKE $${i + 2})`);
-    const params = [scope, ...terms.map((t) => `%${t}%`), max];
+    const likeClauses = terms.map(
+      (_, i) => `(title ILIKE $${i + 2} OR body ILIKE $${i + 2} OR problem ILIKE $${i + 2} OR environment_json::text ILIKE $${i + 2})`,
+    );
+    const params = [scope, ...terms.map((t) => `%${t}%`)];
    const { rows } = await pool.query(
-      `SELECT id, scope, kind, title, body, tags, source_session_id,
-              source_user_id, use_count, problem, environment_json, hypothesis,
-              action_json, result, confidence, evidence_json, parent_experience_id,
-              supersedes_id, status, created_at, updated_at
-       FROM h5_experience
+      `${PG_EXPERIENCE_SELECT}
       WHERE scope = $1 AND status = 'active' AND (${likeClauses.join(' OR ')})
       ORDER BY updated_at DESC
-       LIMIT $${terms.length + 2}`,
+       LIMIT 500`,
      params,
    );
-    if (rows.length > 0) await bumpUseCount(rows.map((r) => r.id));
-    return rows.map(rowToExperience);
+    const ranked = rankRows(rows, terms, now()).slice(0, max);
+    if (ranked.length > 0) await bumpUseCount(ranked.map((entry) => entry.row.id));
+    const hits = ranked.map((entry) => rowToExperience(entry.row));
+    void emitExperienceRecalled({ query: text, hits, scope });
+    return hits;
  }

  async function bumpUseCount(ids) {
@@ -255,8 +314,62 @@ export async function createPgExperienceService(options = {}) {
    ];
  }

-  async function reflect() {
-    return { ok: false, reason: 'not_implemented' };
+  async function reflect({
+    scope = 'global',
+    minGroupSize = DEFAULT_REFLECT_MIN_GROUP_SIZE,
+    dryRun = false,
+  } = {}) {
+    const { rows } = await pool.query(
+      `${PG_EXPERIENCE_SELECT}
+       WHERE scope = $1
+       ORDER BY updated_at DESC
+       LIMIT 2000`,
+      [scope],
+    );
+    const plans = planExperienceReflection(rows, { minGroupSize });
+    if (dryRun) {
+      return {
+        ok: true,
+        dryRun: true,
+        scope,
+        groupsEligible: plans.length,
+        wouldArchive: plans.reduce((sum, plan) => sum + plan.sourceIds.length, 0),
+        previews: plans.map((plan) => ({
+          fingerprint: plan.fingerprint,
+          sourceCount: plan.sourceIds.length,
+          title: plan.payload.title,
+        })),
+      };
+    }
+
+    let created = 0;
+    let archived = 0;
+    const patterns = [];
+    for (const plan of plans) {
+      const saved = await record({
+        ...plan.payload,
+        scope,
+        source: 'reflect',
+      });
+      await pool.query(
+        `UPDATE h5_experience
+         SET status = 'archived', parent_experience_id = $1::uuid, updated_at = $2
+         WHERE id = ANY($3::uuid[])`,
+        [saved.id, now(), plan.sourceIds],
+      );
+      created += 1;
+      archived += plan.sourceIds.length;
+      patterns.push(saved);
+    }
+
+    return {
+      ok: true,
+      dryRun: false,
+      scope,
+      created,
+      archived,
+      patterns,
+    };
  }

  async function close() {