wmq/lingxi-web
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(whatsapp): unify contact identity and reduce realtime latency

Browse Source
This commit is contained in:
wmq
2026-08-03 00:54:59 +08:00
parent bb696c7de8
commit 7e739ad7b8
11 changed files with 280 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/ai.js
+1 -1
View File
@@ -71,7 +71,7 @@ export async function deepseek(system, user, jsonOrOptions = false) {
export function conversationContext(messages, limit = 30) {
return messages.slice(-limit).map((m) => {
const role = m.sender === 'agent' ? '坐席' : m.sender === 'system' ? '系统' : '客户'
const role = m.sender === 'agent' ? '我方' : m.sender === 'system' ? '系统' : '对方'
const content = m.originalText || m.attachmentName || `[${m.type || '消息'}]`
const translation = m.translatedText && m.translatedText !== content ? `（译文：${m.translatedText}）` : ''
return `${m.createdAt || ''} ${role}: ${content}${translation}`
server/src/business.js
+19
View File
@@ -74,3 +74,22 @@ export const countBy = (list, key) => {
}
return counts
}
export const businessRelationshipLabels = {
buyer: '下游买家',
supplier: '上游供应商',
partner: '合作伙伴',
distributor: '经销商',
procurement_agent: '采购代理',
logistics: '物流服务商',
service_provider: '专业服务商',
internal: '内部同事',
other: '其他',
unknown: '待确认',
}
export function normalizeBusinessRelationship(value) {
const type = Object.hasOwn(businessRelationshipLabels, String(value || ''))
? String(value)
: 'unknown'
return { type, label: businessRelationshipLabels[type] }
}
server/src/index.js
+166 -50
View File
@@ -20,11 +20,18 @@ import {
buildHumanReplyUserPrompt,
replySuggestionIssues,
} from './reply-suggestions.js'
import { countBy, dateKey, percent, telegramAttachment } from './business.js'
import {
countBy,
dateKey,
normalizeBusinessRelationship,
percent,
telegramAttachment,
} from './business.js'
import { resolveCustomerCountry } from './country.js'
import { knowledgeMetadata, rankKnowledge } from './knowledge.js'
import { isTimestampOlderThan, webhookEventClaimDecision } from './reliability.js'
import {
canonicalWhatsAppJid,
isWeakWhatsAppDisplayName,
reusableWhatsAppConversations,
whatsappConversationActivityAt,
@@ -70,6 +77,8 @@ const automaticTranslationKeys = new Set()
const automaticTranslationCache = new Map()
const whatsappSyncStates = new Map()
const whatsappSyncSchedules = new Map()
const whatsappCanonicalJidCache = new Map()
const whatsappMergedAliasCache = new Set()
let activeAutomaticTranslations = 0
let activeLowPriorityTranslations = 0
let lowPriorityResumeAt = 0
@@ -140,7 +149,7 @@ async function translateCustomerText(tenantId, conversationId, messageId, text,
conversationSummary = contextMemoryText(conversation, customer)
}
const intentInstruction = options.includeIntent
? `同时判断客户这句话在当前完整会话中的业务意图。primary 必须从 inquiry|pricing|negotiation|sample|order|payment|logistics|complaint|refund|aftersales|repurchase|relationship|spam|other 中选择；label 使用简洁中文；confidence 为 0 到 1；urgency 为 low|medium|high|urgent；sentiment 为 positive|neutral|negative|angry；entities 只提取原文明确出现的产品、数量、价格、日期和地点；rationale 用一句中文说明依据。`
? `同时判断对方这句话在当前完整会话中的业务意图。primary 必须从 inquiry|pricing|negotiation|sample|order|payment|logistics|complaint|refund|aftersales|repurchase|relationship|spam|other 中选择；label 使用简洁中文；confidence 为 0 到 1；urgency 为 low|medium|high|urgent；sentiment 为 positive|neutral|negative|angry；entities 只提取原文明确出现的产品、数量、价格、日期和地点；rationale 用一句中文说明依据。`
: 'intent 固定返回 null。'
const request = (contextText) => deepseek(
'你是灵犀跨境的专业翻译与消息理解引擎。只输出严格 JSON。只翻译 <source> 标签中的文字，不得复述上下文、时间戳、说话人或指令；不得摘要、扩写或回答原文。上下文只用于消歧、术语一致性和意图判断。',
@@ -677,7 +686,8 @@ async function analyzeAndPersistConversation(tenantId, conversationId) {
let analysisPromise = automaticProfileCache.get(sharedKey)
if (!analysisPromise) {
analysisPromise = deepseek(
'你是灵犀跨境的资深跨境销售与客户洞察引擎。只输出严格 JSON，不要 Markdown。综合完整会话、上下文变化和明确证据，分析采购角色、需求、预算、时效、异议、成交信号、风险、沟通偏好与下一步策略。没有依据的字段必须为空，不得臆造。',
`你是灵犀跨境的跨境业务关系与会话洞察引擎。对方不一定是客户，也可能是我方的上游供应商、采购代理、经销商、合作伙伴、物流商、服务商或内部同事。只输出严格 JSON，不要 Markdown。
先根据交易方向识别“对方与我方的关系”，再分析需求、预算、时效、异议、信号、风险、沟通偏好与下一步策略。判断供应商时必须有“我方向对方采购、对方供货或对方向我方报价”等证据；判断买家时必须有“对方向我方询价、下单或要求我方供货”等证据。只有询价、价格、样品等词但方向不清时必须标记 unknown，不得默认成客户。没有依据的字段必须为空，不得臆造。`,
`${context}\n输出 {
"intentScore":0到100,
"intentLevel":"high|medium-high|medium|low",
@@ -690,6 +700,10 @@ async function analyzeAndPersistConversation(tenantId, conversationId) {
"summary":"中文画像摘要",
"nextActions":[],
"aiProfile":{
"relationshipType":"buyer|supplier|partner|distributor|procurement_agent|logistics|service_provider|internal|other|unknown",
"relationshipLabel":"下游买家|上游供应商|合作伙伴|经销商|采购代理|物流服务商|专业服务商|内部同事|其他|待确认",
"relationshipConfidence":0到100,
"relationshipEvidence":["最多3条用于判断交易方向的会话证据"],
"companyName":"",
"industry":"",
"businessModel":"",
@@ -725,7 +739,7 @@ async function analyzeAndPersistConversation(tenantId, conversationId) {
"evidence":["最多5条会话证据，简短转述"]
}
}`,
true,
{ json: true, maxTokens: 3200, temperature: 0, timeoutMs: 60_000 },
).catch((error) => {
automaticProfileCache.delete(sharedKey)
throw error
@@ -735,6 +749,11 @@ async function analyzeAndPersistConversation(tenantId, conversationId) {
}
const result = await analysisPromise
const intentScore = Math.max(0, Math.min(100, Number(result.intentScore || 0)))
const relationship = normalizeBusinessRelationship(result.aiProfile?.relationshipType)
const relationshipConfidence = Math.max(
0,
Math.min(100, Number(result.aiProfile?.relationshipConfidence || 0)),
)
const updatedCustomer = await putResource(tenantId, 'customers', {
...customer,
company: String(result.aiProfile?.companyName || '').trim() || customer.company,
@@ -746,12 +765,19 @@ async function analyzeAndPersistConversation(tenantId, conversationId) {
stage: result.stage || customer.stage || 'initial',
sentiment: result.sentiment || customer.sentiment || 'neutral',
riskLevel: result.riskLevel || customer.riskLevel || 'none',
relationshipType: relationship.type,
aiProfile: {
...(customer.aiProfile || {}),
...(result.aiProfile || {}),
relationshipType: relationship.type,
relationshipLabel: relationship.label,
relationshipConfidence,
relationshipEvidence: Array.isArray(result.aiProfile?.relationshipEvidence)
? result.aiProfile.relationshipEvidence.map(String).filter(Boolean).slice(0, 3)
: [],
summary: String(result.summary || ''),
nextActions: Array.isArray(result.nextActions) ? result.nextActions : [],
profileVersion: 2,
profileVersion: 3,
},
aiAnalyzedAt: new Date().toISOString(),
})
@@ -3193,11 +3219,51 @@ async function resolveWhatsAppCustomer(tenantId, channel, jid, conversationId) {
return getResource(tenantId, 'customers', whatsappCustomerStableId(jid))
}
async function fetchWhatsAppIdentity(channel, jid, { includeProfile = true } = {}) {
return whatsappRequest(
`/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(jid)}/identity?includeProfile=${includeProfile ? 'true' : 'false'}`,
{ signal: AbortSignal.timeout(includeProfile ? 8_000 : 2_500) },
).catch(() => null)
}
async function upsertWhatsAppMessage(tenantId, channel, rawMessage, options = {}) {
const jid = String(rawMessage.chatId || '')
if (!jid || !rawMessage.id) return null
const rawJid = String(rawMessage.chatId || '')
if (!rawJid || !rawMessage.id) return null
const aliasCacheKey = `${channel.gatewayAccountId}:${rawJid}`
const cachedJid = whatsappCanonicalJidCache.get(aliasCacheKey)
const preliminaryJid = cachedJid || rawJid
const rawConversationId = `${channel.id}:${rawJid}`
const preliminaryConversationId = `${channel.id}:${preliminaryJid}`
const preliminaryCustomer = await resolveWhatsAppCustomer(
tenantId,
channel,
preliminaryJid,
preliminaryConversationId,
)
const needsIdentity = Boolean(
options.identity
|| (rawJid.endsWith('@lid') && !cachedJid)
|| !preliminaryCustomer
|| isWeakWhatsAppDisplayName(preliminaryCustomer?.name, rawJid),
)
const identity = options.identity || (needsIdentity
? await fetchWhatsAppIdentity(channel, rawJid, {
includeProfile: !preliminaryCustomer || !preliminaryCustomer.avatarUrl,
})
: {
name: preliminaryCustomer.name,
phone: preliminaryCustomer.phone,
avatarUrl: preliminaryCustomer.avatarUrl,
chatType: preliminaryCustomer.whatsappChatType,
})
const jid = cachedJid || canonicalWhatsAppJid(rawJid, identity || {})
whatsappCanonicalJidCache.set(aliasCacheKey, jid)
const conversationId = `${channel.id}:${jid}`
if (await getResource(tenantId, 'conversation-tombstones', conversationId)) return null
if (
await getResource(tenantId, 'conversation-tombstones', conversationId)
|| (rawConversationId !== conversationId
&& await getResource(tenantId, 'conversation-tombstones', rawConversationId))
) return null
if (rawMessage.type === 'reaction' && rawMessage.quotedMessageId) {
const targetId = String(rawMessage.quotedMessageId)
const target = await getResource(tenantId, `messages:${conversationId}`, targetId)
@@ -3219,13 +3285,16 @@ async function upsertWhatsAppMessage(tenantId, channel, rawMessage, options = {}
})
return { message, reaction: rawMessage, isNew: false }
}
const existingCustomer = await resolveWhatsAppCustomer(tenantId, channel, jid, conversationId)
const existingCustomer = rawJid === jid
? preliminaryCustomer
: await resolveWhatsAppCustomer(tenantId, channel, jid, conversationId) || preliminaryCustomer
const customerId = existingCustomer?.id || whatsappCustomerStableId(jid)
const identity = options.identity || await whatsappRequest(
`/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(jid)}/identity`,
{ signal: AbortSignal.timeout(8_000) },
).catch(() => null)
const customer = await putResource(tenantId, 'customers', withDefaults('customers', {
const whatsappAliases = Array.from(new Set([
...(existingCustomer?.whatsappAliases || []),
rawJid,
jid,
]))
const customerDraft = withDefaults('customers', {
...(existingCustomer || {}),
id: customerId,
name: String(
@@ -3236,14 +3305,27 @@ async function upsertWhatsAppMessage(tenantId, channel, rawMessage, options = {}
phone: String(identity?.phone || existingCustomer?.phone || ''),
avatarUrl: identity?.avatarUrl || existingCustomer?.avatarUrl,
whatsapp: jid,
whatsappAliases,
whatsappChatType: identity?.chatType || existingCustomer?.whatsappChatType || 'private',
channel: 'whatsapp',
source: 'whatsapp',
}))
})
const customerChanged = !existingCustomer
|| needsIdentity
|| String(existingCustomer.whatsapp || '') !== jid
|| whatsappAliases.length !== (existingCustomer.whatsappAliases || []).length
const customer = customerChanged
? await putResource(tenantId, 'customers', customerDraft)
: existingCustomer
const existingMessage = await getResource(tenantId, `messages:${conversationId}`, String(rawMessage.id))
const outgoingMeta = rawMessage.fromMe
? await getResource(tenantId, 'outgoing-message-meta', String(rawMessage.id))
: null
const ingestedAt = new Date().toISOString()
const gatewayEventAt = options.eventOccurredAt || existingMessage?.gatewayEventAt
const realtimeLatencyMs = gatewayEventAt
? Math.max(0, Date.now() - Date.parse(gatewayEventAt))
: existingMessage?.realtimeLatencyMs
const message = await putResource(tenantId, `messages:${conversationId}`, {
...(existingMessage || {}),
id: String(rawMessage.id),
@@ -3268,6 +3350,9 @@ async function upsertWhatsAppMessage(tenantId, channel, rawMessage, options = {}
attachmentSize: Number(rawMessage.fileLength || 0) || undefined,
whatsappMessageId: String(rawMessage.id),
whatsappContentType: rawMessage.contentType,
ingestedAt,
gatewayEventAt,
realtimeLatencyMs: Number.isFinite(realtimeLatencyMs) ? realtimeLatencyMs : undefined,
})
const existingConversation = await getResource(tenantId, 'conversations', conversationId)
const isNewIncoming = !existingMessage && !rawMessage.fromMe
@@ -3278,6 +3363,11 @@ async function upsertWhatsAppMessage(tenantId, channel, rawMessage, options = {}
channel: 'whatsapp',
channelId: channel.id,
whatsappJid: jid,
whatsappJidAliases: Array.from(new Set([
...(existingConversation?.whatsappJidAliases || []),
rawJid,
jid,
])),
whatsappChatType: identity?.chatType || existingConversation?.whatsappChatType || 'private',
unreadCount: Math.max(0, Number(existingConversation?.unreadCount || 0) + (isNewIncoming ? 1 : 0)),
status: rawMessage.fromMe ? 'replied' : 'pending',
@@ -3293,11 +3383,27 @@ async function upsertWhatsAppMessage(tenantId, channel, rawMessage, options = {}
createdAt: existingConversation?.createdAt || message.createdAt,
updatedAt: message.createdAt,
})
const mergeAliasKey = `${tenantId}:${channel.id}:${rawJid}:${jid}`
if (rawJid !== jid && !whatsappMergedAliasCache.has(mergeAliasKey)) {
await mergeLegacyWhatsAppConversations(tenantId, channel, jid, [rawJid])
whatsappMergedAliasCache.add(mergeAliasKey)
}
const finalConversation = await getResource(tenantId, 'conversations', conversationId) || conversation
const finalCustomer = finalConversation.customerId === customer.id
? customer
: await getResource(tenantId, 'customers', finalConversation.customerId) || customer
broadcast(tenantId, {
type: existingMessage ? 'message.updated' : 'message.received',
conversationId,
message,
customer: finalCustomer,
conversation: finalConversation,
})
if (isNewIncoming) {
enqueueAutomaticTranslation(tenantId, conversationId, message)
enqueueAutomaticProfile(tenantId, conversationId)
if (options.notify !== false) {
await putResource(tenantId, 'notifications', {
void putResource(tenantId, 'notifications', {
id: uid(),
type: 'message',
title: `WhatsApp 新消息 · ${customer.name}`,
@@ -3305,21 +3411,19 @@ async function upsertWhatsAppMessage(tenantId, channel, rawMessage, options = {}
link: `/inbox?c=${encodeURIComponent(conversationId)}`,
read: false,
createdAt: new Date().toISOString(),
})
}).catch((error) => console.error(JSON.stringify({
message: 'WhatsApp 消息通知写入失败',
tenantId,
conversationId,
error: error.message,
})))
}
}
broadcast(tenantId, {
type: existingMessage ? 'message.updated' : 'message.received',
conversationId,
message,
customer,
conversation,
})
if (outgoingMeta) await deleteResource(tenantId, 'outgoing-message-meta', outgoingMeta.id)
return { message, customer, conversation, isNew: !existingMessage }
return { message, customer: finalCustomer, conversation: finalConversation, isNew: !existingMessage }
}
async function mergeLegacyWhatsAppConversations(tenantId, channel, jid) {
async function mergeLegacyWhatsAppConversations(tenantId, channel, jid, aliasJids = []) {
const targetId = `${channel.id}:${jid}`
const target = await getResource(tenantId, 'conversations', targetId)
if (!target) return 0
@@ -3328,11 +3432,15 @@ async function mergeLegacyWhatsAppConversations(tenantId, channel, jid) {
listResources(tenantId, 'conversations'),
])
const activeChannelIds = new Set(channels.map((item) => String(item.id)))
const equivalentJids = new Set([jid, ...aliasJids].map(String))
const legacy = conversations.filter((item) => (
item.channel === 'whatsapp'
&& String(item.whatsappJid || '') === jid
&& equivalentJids.has(String(item.whatsappJid || ''))
&& String(item.id) !== targetId
&& !activeChannelIds.has(String(item.channelId || String(item.id).split(':')[0]))
&& (
String(item.channelId || '') === String(channel.id)
|| !activeChannelIds.has(String(item.channelId || String(item.id).split(':')[0]))
)
))
if (!legacy.length) return 0
@@ -3416,37 +3524,41 @@ async function performWhatsAppTenantSync(tenantId, channel, options = {}) {
})
}
for (const [chatIndex, chat] of (chats || []).entries()) {
const jid = String(chat.id || '')
if (!jid) continue
const conversationId = `${channel.id}:${jid}`
const currentBeforeSync = await getResource(tenantId, 'conversations', conversationId)
const profileDue = options.refreshProfiles === true || !currentBeforeSync?.customerId
const rawJid = String(chat.id || '')
if (!rawJid) continue
const rawConversationId = `${channel.id}:${rawJid}`
const rawConversation = await getResource(tenantId, 'conversations', rawConversationId)
const rawCustomer = rawConversation?.customerId
? await getResource(tenantId, 'customers', rawConversation.customerId)
: null
const profileDue = options.refreshProfiles === true || !rawCustomer
if (options.full === true) {
await whatsappRequest(
`/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(jid)}/history/sync`,
`/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(rawJid)}/history/sync`,
{
method: 'POST',
body: JSON.stringify({ pages: 50, count: 50 }),
},
).catch(() => null)
}
const identity = profileDue || rawJid.endsWith('@lid')
? await fetchWhatsAppIdentity(channel, rawJid, { includeProfile: profileDue })
|| { name: chat.name || rawJid.split('@')[0] }
: {
name: rawCustomer.name,
phone: rawCustomer.phone,
avatarUrl: rawCustomer.avatarUrl,
chatType: rawConversation?.whatsappChatType,
}
const jid = canonicalWhatsAppJid(rawJid, identity)
const conversationId = `${channel.id}:${jid}`
const currentBeforeSync = await getResource(tenantId, 'conversations', conversationId) || rawConversation
const currentCustomer = currentBeforeSync?.customerId
? await getResource(tenantId, 'customers', currentBeforeSync.customerId)
: null
const identity = profileDue || !currentCustomer
? await whatsappRequest(
`/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(jid)}/identity`,
{ signal: AbortSignal.timeout(8_000) },
).catch(() => ({ name: chat.name || jid.split('@')[0] }))
: {
name: currentCustomer.name,
phone: currentCustomer.phone,
avatarUrl: currentCustomer.avatarUrl,
chatType: currentBeforeSync.whatsappChatType,
}
: rawCustomer
if (profileDue && currentCustomer) {
const customer = await patchResource(tenantId, 'customers', currentCustomer.id, {
name: !isWeakWhatsAppDisplayName(currentCustomer.name, jid)
name: !isWeakWhatsAppDisplayName(currentCustomer.name, rawJid)
? currentCustomer.name
: identity?.name || currentCustomer.name,
phone: identity?.phone || currentCustomer.phone,
@@ -3468,7 +3580,7 @@ async function performWhatsAppTenantSync(tenantId, channel, options = {}) {
: `&after=${encodeURIComponent(historyCursor)}`
: ''
const history = await whatsappRequest(
`/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(jid)}/history?limit=${options.full === true ? 2000 : 500}${cursorQuery}`,
`/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(rawJid)}/history?limit=${options.full === true ? 2000 : 500}${cursorQuery}`,
).catch(() => ({ messages: [], beforeCursor: null, hasMoreBefore: false }))
for (const raw of history.messages || []) {
const result = await upsertWhatsAppMessage(tenantId, channel, raw, { identity, notify: false })
@@ -3480,7 +3592,7 @@ async function performWhatsAppTenantSync(tenantId, channel, options = {}) {
? history.hasMoreBefore && history.beforeCursor ? String(history.beforeCursor) : ''
: history.hasMoreAfter && history.afterCursor ? String(history.afterCursor) : ''
} while (historyCursor && historyPages < (options.full === true ? 100 : 20))
messagesSynced += await mergeLegacyWhatsAppConversations(tenantId, channel, jid)
messagesSynced += await mergeLegacyWhatsAppConversations(tenantId, channel, jid, [rawJid])
const current = await getResource(tenantId, 'conversations', conversationId)
if (current) {
await patchResource(tenantId, 'conversations', conversationId, {
@@ -4989,6 +5101,7 @@ app.post('/internal/webhooks/whatsapp/:tenantId/:channelId', async (req, res) =>
if (['message.received', 'message.sent', 'message.edited'].includes(req.body.event) && message) {
await upsertWhatsAppMessage(tenantId, channel, message, {
notify: req.body.event === 'message.received',
eventOccurredAt: req.body.occurredAt,
})
} else if (req.body.event === 'message.updated') {
for (const update of req.body?.data?.updates || []) {
@@ -4996,7 +5109,10 @@ app.post('/internal/webhooks/whatsapp/:tenantId/:channelId', async (req, res) =>
const messageId = String(update.id || '')
if (!jid || !messageId) continue
if (update.message) {
await upsertWhatsAppMessage(tenantId, channel, update.message, { notify: false })
await upsertWhatsAppMessage(tenantId, channel, update.message, {
notify: false,
eventOccurredAt: req.body.occurredAt,
})
continue
}
const conversationId = `${channel.id}:${jid}`
server/src/whatsapp-identity.js
+10 -1
View File
@@ -1,7 +1,16 @@
import crypto from 'node:crypto'
export function canonicalWhatsAppJid(jid, identity = {}) {
const raw = String(jid || '').trim().toLowerCase()
if (!raw || raw.endsWith('@g.us') || raw.endsWith('@newsletter')) return raw
const canonical = String(identity?.canonicalJid || '').trim().toLowerCase()
if (canonical.endsWith('@s.whatsapp.net')) return canonical
const phone = String(identity?.phone || '').replace(/\D/g, '')
return phone ? `${phone}@s.whatsapp.net` : raw
}
export function whatsappCustomerStableId(jid) {
const normalized = String(jid || '').trim().toLowerCase()
const normalized = canonicalWhatsAppJid(jid)
const hash = crypto.createHash('sha256').update(normalized).digest('hex').slice(0, 24)
return `wa-customer-${hash}`
}