|
|
@@ -142,6 +142,58 @@ class AgentWorkbenchDb {
|
|
|
updated_at TEXT NOT NULL,
|
|
|
UNIQUE(conversation_id, property_id)
|
|
|
);
|
|
|
+ CREATE TABLE IF NOT EXISTS customer_memory_items (
|
|
|
+ id TEXT PRIMARY KEY,
|
|
|
+ conversation_id TEXT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
|
|
|
+ memory_key TEXT NOT NULL,
|
|
|
+ type TEXT NOT NULL DEFAULT 'fact' CHECK(type IN ('fact','preference','constraint','event','hypothesis')),
|
|
|
+ content TEXT NOT NULL,
|
|
|
+ confidence REAL NOT NULL DEFAULT 1,
|
|
|
+ importance REAL NOT NULL DEFAULT 0.5,
|
|
|
+ status TEXT NOT NULL DEFAULT 'active' CHECK(status IN ('active','superseded','rejected')),
|
|
|
+ source_message_ids_json TEXT NOT NULL DEFAULT '[]',
|
|
|
+ direction TEXT NOT NULL DEFAULT 'customer_to_agent',
|
|
|
+ created_by TEXT NOT NULL DEFAULT 'rule',
|
|
|
+ expires_at TEXT,
|
|
|
+ created_at TEXT NOT NULL,
|
|
|
+ updated_at TEXT NOT NULL,
|
|
|
+ UNIQUE(conversation_id, memory_key)
|
|
|
+ );
|
|
|
+ CREATE TABLE IF NOT EXISTS customer_memory_snapshots (
|
|
|
+ id TEXT PRIMARY KEY,
|
|
|
+ conversation_id TEXT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
|
|
|
+ version INTEGER NOT NULL,
|
|
|
+ compact_text TEXT NOT NULL DEFAULT '',
|
|
|
+ content_hash TEXT NOT NULL,
|
|
|
+ created_at TEXT NOT NULL,
|
|
|
+ UNIQUE(conversation_id, version)
|
|
|
+ );
|
|
|
+ CREATE TABLE IF NOT EXISTS memory_extraction_jobs (
|
|
|
+ id TEXT PRIMARY KEY,
|
|
|
+ conversation_id TEXT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
|
|
|
+ message_id TEXT NOT NULL REFERENCES messages(id) ON DELETE CASCADE,
|
|
|
+ profile_updates_json TEXT NOT NULL DEFAULT '{}',
|
|
|
+ status TEXT NOT NULL DEFAULT 'pending' CHECK(status IN ('pending','processing','completed','failed')),
|
|
|
+ attempts INTEGER NOT NULL DEFAULT 0,
|
|
|
+ max_attempts INTEGER NOT NULL DEFAULT 3,
|
|
|
+ next_attempt_at TEXT NOT NULL,
|
|
|
+ error TEXT,
|
|
|
+ result_json TEXT NOT NULL DEFAULT '{}',
|
|
|
+ created_at TEXT NOT NULL,
|
|
|
+ updated_at TEXT NOT NULL,
|
|
|
+ completed_at TEXT,
|
|
|
+ UNIQUE(conversation_id, message_id)
|
|
|
+ );
|
|
|
+ CREATE TABLE IF NOT EXISTS customer_memory_revisions (
|
|
|
+ id TEXT PRIMARY KEY,
|
|
|
+ memory_id TEXT NOT NULL REFERENCES customer_memory_items(id) ON DELETE CASCADE,
|
|
|
+ conversation_id TEXT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
|
|
|
+ memory_key TEXT NOT NULL,
|
|
|
+ reason TEXT NOT NULL,
|
|
|
+ previous_json TEXT NOT NULL,
|
|
|
+ next_json TEXT NOT NULL,
|
|
|
+ created_at TEXT NOT NULL
|
|
|
+ );
|
|
|
CREATE TABLE IF NOT EXISTS drafts (
|
|
|
id TEXT PRIMARY KEY,
|
|
|
conversation_id TEXT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
|
|
|
@@ -180,8 +232,14 @@ class AgentWorkbenchDb {
|
|
|
CREATE INDEX IF NOT EXISTS idx_customer_tasks_conversation ON customer_tasks(conversation_id, status, updated_at DESC);
|
|
|
CREATE INDEX IF NOT EXISTS idx_customer_alerts_conversation ON customer_alerts(conversation_id, status, updated_at DESC);
|
|
|
CREATE INDEX IF NOT EXISTS idx_customer_recommendations_conversation ON customer_recommendations(conversation_id, status, last_recommended_at DESC);
|
|
|
+ CREATE INDEX IF NOT EXISTS idx_customer_memory_conversation ON customer_memory_items(conversation_id, status, importance DESC, updated_at DESC);
|
|
|
+ CREATE INDEX IF NOT EXISTS idx_customer_memory_snapshots ON customer_memory_snapshots(conversation_id, version DESC);
|
|
|
+ CREATE INDEX IF NOT EXISTS idx_memory_extraction_jobs_pending ON memory_extraction_jobs(status, next_attempt_at, created_at);
|
|
|
+ CREATE INDEX IF NOT EXISTS idx_customer_memory_revisions ON customer_memory_revisions(memory_id, created_at DESC);
|
|
|
CREATE INDEX IF NOT EXISTS idx_audit_created ON audit_logs(created_at DESC);
|
|
|
`);
|
|
|
+ this.db.prepare(`UPDATE memory_extraction_jobs SET status='pending',next_attempt_at=?,updated_at=? WHERE status='processing'`)
|
|
|
+ .run(now(), now());
|
|
|
this.ensureColumn('customer_tasks', 'business_key', "business_key TEXT NOT NULL DEFAULT ''");
|
|
|
this.ensureColumn('customer_tasks', 'managed_by', "managed_by TEXT NOT NULL DEFAULT 'agent'");
|
|
|
this.ensureColumn('customer_tasks', 'evidence_json', "evidence_json TEXT NOT NULL DEFAULT '[]'");
|
|
|
@@ -409,6 +467,214 @@ class AgentWorkbenchDb {
|
|
|
return this.getProfile(conversationId);
|
|
|
}
|
|
|
|
|
|
+ upsertCustomerMemory(conversationId, item = {}) {
|
|
|
+ const memoryKey = String(item.memoryKey || item.memory_key || '').trim().slice(0, 240);
|
|
|
+ const content = String(item.content || '').trim().slice(0, 1200);
|
|
|
+ if (!memoryKey || !content) return null;
|
|
|
+ const type = ['fact', 'preference', 'constraint', 'event', 'hypothesis'].includes(item.type) ? item.type : 'fact';
|
|
|
+ const sourceMessageIds = [...new Set((item.sourceMessageIds || item.source_message_ids || []).map(String).filter(Boolean))].slice(-20);
|
|
|
+ const existing = this.db.prepare('SELECT * FROM customer_memory_items WHERE conversation_id=? AND memory_key=?').get(conversationId, memoryKey);
|
|
|
+ const timestamp = now();
|
|
|
+ if (existing) {
|
|
|
+ const mergedSources = [...new Set([...parse(existing.source_message_ids_json, []), ...sourceMessageIds])].slice(-20);
|
|
|
+ const nextState = {
|
|
|
+ ...existing,
|
|
|
+ type,
|
|
|
+ content,
|
|
|
+ confidence: Math.max(0, Math.min(1, Number(item.confidence ?? existing.confidence ?? 1))),
|
|
|
+ importance: Math.max(0, Math.min(1, Number(item.importance ?? existing.importance ?? 0.5))),
|
|
|
+ status: 'active',
|
|
|
+ source_message_ids_json: json(mergedSources),
|
|
|
+ direction: String(item.direction || existing.direction || 'customer_to_agent'),
|
|
|
+ created_by: String(item.createdBy || item.created_by || existing.created_by || 'rule'),
|
|
|
+ expires_at: item.expiresAt || item.expires_at || null,
|
|
|
+ };
|
|
|
+ if (['type', 'content', 'confidence', 'importance', 'status', 'source_message_ids_json', 'expires_at'].some(key => String(existing[key] ?? '') !== String(nextState[key] ?? ''))) {
|
|
|
+ this.recordCustomerMemoryRevision(existing, nextState, item.revisionReason || (existing.content !== content ? 'superseded_by_new_evidence' : 'evidence_or_metadata_updated'));
|
|
|
+ }
|
|
|
+ this.db.prepare(`UPDATE customer_memory_items SET type=?,content=?,confidence=?,importance=?,status='active',
|
|
|
+ source_message_ids_json=?,direction=?,created_by=?,expires_at=?,updated_at=? WHERE id=?`)
|
|
|
+ .run(nextState.type, nextState.content, nextState.confidence, nextState.importance, nextState.source_message_ids_json,
|
|
|
+ nextState.direction, nextState.created_by, nextState.expires_at, timestamp, existing.id);
|
|
|
+ return this.getCustomerMemory(existing.id);
|
|
|
+ }
|
|
|
+ const id = makeId('memory');
|
|
|
+ this.db.prepare(`INSERT INTO customer_memory_items(id,conversation_id,memory_key,type,content,confidence,importance,status,
|
|
|
+ source_message_ids_json,direction,created_by,expires_at,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)`)
|
|
|
+ .run(id, conversationId, memoryKey, type, content, Math.max(0, Math.min(1, Number(item.confidence ?? 1))),
|
|
|
+ Math.max(0, Math.min(1, Number(item.importance ?? 0.5))), 'active', json(sourceMessageIds),
|
|
|
+ String(item.direction || 'customer_to_agent'), String(item.createdBy || item.created_by || 'rule'),
|
|
|
+ item.expiresAt || item.expires_at || null, timestamp, timestamp);
|
|
|
+ return this.getCustomerMemory(id);
|
|
|
+ }
|
|
|
+
|
|
|
+ getCustomerMemory(memoryId) {
|
|
|
+ const row = this.db.prepare('SELECT * FROM customer_memory_items WHERE id=?').get(memoryId);
|
|
|
+ return row ? { ...row, source_message_ids: parse(row.source_message_ids_json, []) } : null;
|
|
|
+ }
|
|
|
+
|
|
|
+ getCustomerMemoryByKey(conversationId, memoryKey) {
|
|
|
+ const row = this.db.prepare('SELECT * FROM customer_memory_items WHERE conversation_id=? AND memory_key=?').get(conversationId, String(memoryKey || ''));
|
|
|
+ return row ? { ...row, source_message_ids: parse(row.source_message_ids_json, []) } : null;
|
|
|
+ }
|
|
|
+
|
|
|
+ listCustomerMemories(conversationId, { status = 'active', limit = 100 } = {}) {
|
|
|
+ const rows = status
|
|
|
+ ? this.db.prepare('SELECT * FROM customer_memory_items WHERE conversation_id=? AND status=? ORDER BY importance DESC,updated_at DESC LIMIT ?').all(conversationId, status, limit)
|
|
|
+ : this.db.prepare('SELECT * FROM customer_memory_items WHERE conversation_id=? ORDER BY importance DESC,updated_at DESC LIMIT ?').all(conversationId, limit);
|
|
|
+ return rows.map(row => ({ ...row, source_message_ids: parse(row.source_message_ids_json, []) }));
|
|
|
+ }
|
|
|
+
|
|
|
+ updateCustomerMemory(memoryId, fields = {}) {
|
|
|
+ const current = this.getCustomerMemory(memoryId);
|
|
|
+ if (!current) throw new Error('客户记忆不存在');
|
|
|
+ const allowedTypes = new Set(['fact', 'preference', 'constraint', 'event', 'hypothesis']);
|
|
|
+ const allowedStatuses = new Set(['active', 'superseded', 'rejected']);
|
|
|
+ const updates = {
|
|
|
+ content: fields.content === undefined ? current.content : String(fields.content || '').trim().slice(0, 1200),
|
|
|
+ type: fields.type === undefined ? current.type : String(fields.type),
|
|
|
+ status: fields.status === undefined ? current.status : String(fields.status),
|
|
|
+ confidence: fields.confidence === undefined ? current.confidence : Math.max(0, Math.min(1, Number(fields.confidence))),
|
|
|
+ importance: fields.importance === undefined ? current.importance : Math.max(0, Math.min(1, Number(fields.importance))),
|
|
|
+ expiresAt: fields.expiresAt === undefined && fields.expires_at === undefined ? current.expires_at : (fields.expiresAt || fields.expires_at || null),
|
|
|
+ createdBy: fields.createdBy === undefined && fields.created_by === undefined ? current.created_by : String(fields.createdBy || fields.created_by || 'human'),
|
|
|
+ };
|
|
|
+ if (!updates.content) throw new Error('客户记忆内容不能为空');
|
|
|
+ if (!allowedTypes.has(updates.type)) throw new Error('不支持的客户记忆类型');
|
|
|
+ if (!allowedStatuses.has(updates.status)) throw new Error('不支持的客户记忆状态');
|
|
|
+ const nextState = {
|
|
|
+ ...current,
|
|
|
+ content: updates.content,
|
|
|
+ type: updates.type,
|
|
|
+ status: updates.status,
|
|
|
+ confidence: updates.confidence,
|
|
|
+ importance: updates.importance,
|
|
|
+ expires_at: updates.expiresAt,
|
|
|
+ created_by: updates.createdBy,
|
|
|
+ };
|
|
|
+ if (['content', 'type', 'status', 'confidence', 'importance', 'expires_at', 'created_by'].some(key => String(current[key] ?? '') !== String(nextState[key] ?? ''))) {
|
|
|
+ this.recordCustomerMemoryRevision(current, nextState, fields.revisionReason || 'memory_updated');
|
|
|
+ }
|
|
|
+ this.db.prepare(`UPDATE customer_memory_items SET content=?,type=?,status=?,confidence=?,importance=?,expires_at=?,created_by=?,updated_at=? WHERE id=?`)
|
|
|
+ .run(updates.content, updates.type, updates.status, updates.confidence, updates.importance, updates.expiresAt, updates.createdBy, now(), memoryId);
|
|
|
+ return this.getCustomerMemory(memoryId);
|
|
|
+ }
|
|
|
+
|
|
|
+ forgetCustomerMemory(memoryId) {
|
|
|
+ const current = this.getCustomerMemory(memoryId);
|
|
|
+ if (!current) throw new Error('客户记忆不存在');
|
|
|
+ this.db.prepare('DELETE FROM customer_memory_items WHERE id=?').run(memoryId);
|
|
|
+ return current;
|
|
|
+ }
|
|
|
+
|
|
|
+ expireCustomerMemories(conversationId, timestamp = now()) {
|
|
|
+ const expired = this.db.prepare(`SELECT * FROM customer_memory_items
|
|
|
+ WHERE conversation_id=? AND status='active' AND expires_at IS NOT NULL AND expires_at<>'' AND expires_at<=?`)
|
|
|
+ .all(conversationId, timestamp);
|
|
|
+ for (const row of expired) {
|
|
|
+ this.recordCustomerMemoryRevision(row, { ...row, status: 'superseded' }, 'expired');
|
|
|
+ }
|
|
|
+ if (expired.length) {
|
|
|
+ this.db.prepare(`UPDATE customer_memory_items SET status='superseded',updated_at=?
|
|
|
+ WHERE conversation_id=? AND status='active' AND expires_at IS NOT NULL AND expires_at<>'' AND expires_at<=?`)
|
|
|
+ .run(timestamp, conversationId, timestamp);
|
|
|
+ }
|
|
|
+ return expired.length;
|
|
|
+ }
|
|
|
+
|
|
|
+ recordCustomerMemoryRevision(previous, next, reason = 'memory_updated') {
|
|
|
+ if (!previous?.id) return null;
|
|
|
+ const id = makeId('memory_revision');
|
|
|
+ this.db.prepare(`INSERT INTO customer_memory_revisions(id,memory_id,conversation_id,memory_key,reason,previous_json,next_json,created_at)
|
|
|
+ VALUES(?,?,?,?,?,?,?,?)`).run(id, previous.id, previous.conversation_id, previous.memory_key, String(reason || 'memory_updated'),
|
|
|
+ json(previous), json(next || {}), now());
|
|
|
+ return id;
|
|
|
+ }
|
|
|
+
|
|
|
+ listCustomerMemoryRevisions(memoryId, limit = 100) {
|
|
|
+ return this.db.prepare('SELECT * FROM customer_memory_revisions WHERE memory_id=? ORDER BY created_at DESC LIMIT ?')
|
|
|
+ .all(memoryId, limit).map(row => ({ ...row, previous: parse(row.previous_json, {}), next: parse(row.next_json, {}) }));
|
|
|
+ }
|
|
|
+
|
|
|
+ latestMemorySnapshot(conversationId) {
|
|
|
+ return this.db.prepare('SELECT * FROM customer_memory_snapshots WHERE conversation_id=? ORDER BY version DESC LIMIT 1').get(conversationId) || null;
|
|
|
+ }
|
|
|
+
|
|
|
+ ensureMemorySnapshot(conversationId, compactText) {
|
|
|
+ const text = String(compactText || '').trim();
|
|
|
+ const contentHash = crypto.createHash('sha256').update(text).digest('hex');
|
|
|
+ const latest = this.latestMemorySnapshot(conversationId);
|
|
|
+ if (latest?.content_hash === contentHash) return latest;
|
|
|
+ const version = Number(latest?.version || 0) + 1;
|
|
|
+ const id = makeId('memory_snapshot');
|
|
|
+ this.db.prepare('INSERT INTO customer_memory_snapshots(id,conversation_id,version,compact_text,content_hash,created_at) VALUES(?,?,?,?,?,?)')
|
|
|
+ .run(id, conversationId, version, text, contentHash, now());
|
|
|
+ return this.latestMemorySnapshot(conversationId);
|
|
|
+ }
|
|
|
+
|
|
|
+ enqueueMemoryExtraction({ conversationId, messageId, profileUpdates = {}, maxAttempts = 3 }) {
|
|
|
+ const timestamp = now();
|
|
|
+ const id = makeId('memory_job');
|
|
|
+ this.db.prepare(`INSERT INTO memory_extraction_jobs(id,conversation_id,message_id,profile_updates_json,status,attempts,max_attempts,
|
|
|
+ next_attempt_at,error,result_json,created_at,updated_at) VALUES(?,?,?,?, 'pending',0,?,?,?,?,?,?)
|
|
|
+ ON CONFLICT(conversation_id,message_id) DO UPDATE SET profile_updates_json=excluded.profile_updates_json,
|
|
|
+ status=CASE WHEN memory_extraction_jobs.status='completed' THEN 'completed' ELSE 'pending' END,
|
|
|
+ next_attempt_at=excluded.next_attempt_at,error=NULL,updated_at=excluded.updated_at`)
|
|
|
+ .run(id, conversationId, messageId, json(profileUpdates || {}), Math.max(1, Math.min(10, Number(maxAttempts) || 3)),
|
|
|
+ timestamp, null, json({}), timestamp, timestamp);
|
|
|
+ return this.db.prepare('SELECT * FROM memory_extraction_jobs WHERE conversation_id=? AND message_id=?').get(conversationId, messageId);
|
|
|
+ }
|
|
|
+
|
|
|
+ claimMemoryExtractionJob(timestamp = now()) {
|
|
|
+ this.db.exec('BEGIN IMMEDIATE');
|
|
|
+ try {
|
|
|
+ const row = this.db.prepare(`SELECT * FROM memory_extraction_jobs WHERE status='pending' AND next_attempt_at<=?
|
|
|
+ ORDER BY created_at ASC LIMIT 1`).get(timestamp);
|
|
|
+ if (!row) {
|
|
|
+ this.db.exec('COMMIT');
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ this.db.prepare(`UPDATE memory_extraction_jobs SET status='processing',attempts=attempts+1,updated_at=? WHERE id=?`)
|
|
|
+ .run(timestamp, row.id);
|
|
|
+ this.db.exec('COMMIT');
|
|
|
+ const claimed = this.db.prepare('SELECT * FROM memory_extraction_jobs WHERE id=?').get(row.id);
|
|
|
+ return { ...claimed, profileUpdates: parse(claimed.profile_updates_json, {}) };
|
|
|
+ } catch (error) {
|
|
|
+ this.db.exec('ROLLBACK');
|
|
|
+ throw error;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ completeMemoryExtractionJob(jobId, result = {}) {
|
|
|
+ const timestamp = now();
|
|
|
+ this.db.prepare(`UPDATE memory_extraction_jobs SET status='completed',result_json=?,error=NULL,completed_at=?,updated_at=? WHERE id=?`)
|
|
|
+ .run(json(result || {}), timestamp, timestamp, jobId);
|
|
|
+ return this.getMemoryExtractionJob(jobId);
|
|
|
+ }
|
|
|
+
|
|
|
+ failMemoryExtractionJob(jobId, error, retryDelayMs = 1000) {
|
|
|
+ const current = this.getMemoryExtractionJob(jobId);
|
|
|
+ if (!current) return null;
|
|
|
+ const terminal = Number(current.attempts || 0) >= Number(current.max_attempts || 3);
|
|
|
+ const timestamp = now();
|
|
|
+ const nextAttemptAt = new Date(Date.now() + Math.max(100, Number(retryDelayMs) || 1000)).toISOString();
|
|
|
+ this.db.prepare(`UPDATE memory_extraction_jobs SET status=?,next_attempt_at=?,error=?,updated_at=? WHERE id=?`)
|
|
|
+ .run(terminal ? 'failed' : 'pending', nextAttemptAt, String(error?.message || error || '').slice(0, 500), timestamp, jobId);
|
|
|
+ return this.getMemoryExtractionJob(jobId);
|
|
|
+ }
|
|
|
+
|
|
|
+ getMemoryExtractionJob(jobId) {
|
|
|
+ const row = this.db.prepare('SELECT * FROM memory_extraction_jobs WHERE id=?').get(jobId);
|
|
|
+ return row ? { ...row, profileUpdates: parse(row.profile_updates_json, {}), result: parse(row.result_json, {}) } : null;
|
|
|
+ }
|
|
|
+
|
|
|
+ listMemoryExtractionJobs({ status = '', limit = 100 } = {}) {
|
|
|
+ const rows = status
|
|
|
+ ? this.db.prepare('SELECT * FROM memory_extraction_jobs WHERE status=? ORDER BY created_at DESC LIMIT ?').all(status, limit)
|
|
|
+ : this.db.prepare('SELECT * FROM memory_extraction_jobs ORDER BY created_at DESC LIMIT ?').all(limit);
|
|
|
+ return rows.map(row => ({ ...row, profileUpdates: parse(row.profile_updates_json, {}), result: parse(row.result_json, {}) }));
|
|
|
+ }
|
|
|
+
|
|
|
mergeProfileByContactId(contactId, patch = {}, tags) {
|
|
|
const conversation = this.getConversationByContactId(contactId);
|
|
|
if (!conversation) return null;
|
|
|
@@ -711,7 +977,7 @@ class AgentWorkbenchDb {
|
|
|
if (!sourcePath || !fs.existsSync(sourcePath) || path.resolve(sourcePath) === path.resolve(this.filePath)) return { imported: false, reason: 'source_missing' };
|
|
|
if (this.listConversations().length) return { imported: false, reason: 'target_not_empty' };
|
|
|
const source = new DatabaseSync(sourcePath, { readOnly: true });
|
|
|
- const tableOrder = ['settings', 'conversations', 'messages', 'customer_profiles', 'customer_tasks', 'customer_alerts', 'customer_recommendations', 'drafts', 'audit_logs', 'poll_state'];
|
|
|
+ const tableOrder = ['settings', 'conversations', 'messages', 'customer_profiles', 'customer_tasks', 'customer_alerts', 'customer_recommendations', 'customer_memory_items', 'customer_memory_snapshots', 'memory_extraction_jobs', 'customer_memory_revisions', 'drafts', 'audit_logs', 'poll_state'];
|
|
|
let rowsImported = 0;
|
|
|
this.db.exec(`
|
|
|
DROP INDEX IF EXISTS idx_customer_tasks_business_key;
|