Files
crm-ai/bridge/src/agent.js
T

510 lines
25 KiB
JavaScript
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
'use strict';
const { createLLM } = require('./llm');
const { createLibreDeskClient } = require('./libredesk');
const { createTools, resolveSiteProfile } = require('./tools');
const { buildSystemPrompt } = require('./prompts');
const { createVerification } = require('./verification');
const { createRetriever } = require('./rag');
const { createMetrics } = require('./metrics');
const { auditAnswer } = require('./audit');
const { markdownToHtml } = require('./markdown');
const { guardReply } = require('./output-guard');
const { appendMissingProductLinks } = require('./reply-links');
const { quickClassify, detectEmotion, INTENTS } = require('./intent');
const { createContextManager } = require('./context');
const { createReranker } = require('./reranker');
const { detectInjection, markToolResult, INJECTION_CAUTION_PROMPT } = require('./security');
const { hasBuyingIntent, missingLeadContacts, detectLanguage, appendLeadFollowup, alreadyAsksForLead } = require('./lead-intent');
const { createLlmCache } = require('./llm-cache');
const { checkCitation } = require('./citation-checker');
const { createFlywheelWriter } = require('./flywheel');
const { languageMandate } = require('./language');
const { createEscalation } = require('./escalation');
const { createExperimentManager } = require('./experiments');
const { createTraceWriter } = require('./trace-writer');
const { extractLeadFields } = require('./lead-extract');
const crypto = require('crypto');
const prom = require('./prom');
const MAX_TOOL_STEPS = 4;
function stripHtml(html) {
return String(html || '')
.replace(/<br\s*\/?>/gi, '\n')
.replace(/<[^>]+>/g, ' ')
.replace(/&nbsp;/g, ' ')
.replace(/&amp;/g, '&')
.replace(/&lt;/g, '<')
.replace(/&gt;/g, '>')
.replace(/&quot;/g, '"')
.replace(/\s+/g, ' ')
.trim();
}
function createAgent(cfg, deps = {}) {
const llm = deps.llm || createLLM(cfg);
const client = deps.client || createLibreDeskClient(cfg);
const verification = deps.verification || createVerification(cfg);
const retriever = deps.retriever || (cfg.ragPgUrl ? createRetriever(cfg) : null);
const metrics = deps.metrics || createMetrics(cfg);
const contextManager = deps.contextManager || createContextManager();
const llmCache = deps.llmCache || createLlmCache(cfg);
const escalation = deps.escalation || createEscalation({ client, handoffUserId: cfg.handoffUserId || 1 });
const experiments = deps.experiments || createExperimentManager(cfg);
const traceWriter = deps.traceWriter || createTraceWriter(metrics.pool, cfg.traceEnabled);
const flywheel = deps.flywheel || createFlywheelWriter({
pool: metrics.pool,
callLLM: async (messages, system, maxTokens) => {
const msgs = system ? [{ role: 'system', content: system }, ...messages] : messages;
const res = await llm.chat({ messages: msgs, maxTokens: maxTokens || 300 });
return { text: res.content || '' };
},
});
const reranker = deps.reranker || createReranker(llm);
const tools = deps.tools || createTools(cfg, { verification, retriever, client, reranker });
const leadStore = deps.leadStore || (tools && tools.leadStore) || null;
const byName = new Map(tools.map((t) => [t.name, t]));
const toolDefs = tools.map((t) => ({
type: 'function',
function: { name: t.name, description: t.description, parameters: t.parameters },
}));
const locks = new Map();
async function identityFor(conv) {
const contact = conv.contact || {};
// 方案 A 保留 LibreDesk JWT+OTP:JWT 登录联系人 type=contact,视为已验证。
if (contact.type === 'contact' && contact.email) {
return { verified: true, email: String(contact.email).trim().toLowerCase() };
}
const email = await verification.getVerifiedEmail(conv.uuid);
return { verified: Boolean(email), email: email || '' };
}
async function buildMessages(conv, currentMessage) {
const history = await client.getMessages(conv.uuid, 60);
const sorted = history
.slice()
.sort((a, b) => Number(a.id) - Number(b.id))
.filter((m) => m && !m.private && m.type !== 'activity' && m.sender_type)
.map((m) => {
const text = (m.text_content || stripHtml(m.content)).trim();
if (!text) return null;
return { role: m.sender_type === 'contact' ? 'user' : 'assistant', content: text };
})
.filter(Boolean);
// 确保当前消息在最后(webhook 可能先于 history 可见)
const currentText = (currentMessage.text_content || stripHtml(currentMessage.content)).trim();
if (currentText && (!sorted.length || sorted[sorted.length - 1].content !== currentText || sorted[sorted.length - 1].role !== 'user')) {
sorted.push({ role: 'user', content: currentText });
} else if (currentText && sorted[sorted.length - 1].role !== 'user') {
sorted.push({ role: 'user', content: currentText });
}
return contextManager.compressHistory(conv.uuid, sorted, llm);
}
async function runOnce(conv, webhookMessage) {
if (!llm.enabled) throw new Error('LLM is not configured');
const startedAt = Date.now();
const stats = { llmCalls: 0, promptTokens: 0, completionTokens: 0, totalTokens: 0, toolCalls: [], toolResults: [] };
const traceSteps = [];
const traceId = `${conv.uuid}:${webhookMessage.uuid || ''}`;
const profile = resolveSiteProfile(cfg, { inboxId: String(conv.inbox_id || '') });
const variant = experiments.getVariant(conv.uuid);
const identity = await identityFor(conv);
const ctx = {
verified: identity.verified,
email: identity.email,
inboxId: String(conv.inbox_id || ''),
conversationUuid: conv.uuid,
siteProfile: profile,
};
const currentText = (webhookMessage.text_content || stripHtml(webhookMessage.content)).trim();
ctx.language = detectLanguage(currentText);
const messages = await buildMessages(conv, webhookMessage);
if (!messages.length) return null;
const injection = detectInjection(currentText);
let systemContent = buildSystemPrompt({
verified: ctx.verified,
email: ctx.email,
site: profile.site,
channel: conv.channel || 'livechat',
}) + (injection.level === 'high' ? INJECTION_CAUTION_PROMPT : '');
systemContent += languageMandate(currentText);
// 购买意向:LLM 之前先给硬指令,LLM 之后还有程序化兜底
let leadState = null;
if (leadStore) {
try { leadState = await leadStore.get(conv.uuid); } catch { leadState = null; }
}
const preSlots = (leadState && leadState.slots) || {};
const buyingIntent = hasBuyingIntent(currentText);
const preMissing = missingLeadContacts(preSlots, ctx.email);
if (buyingIntent && preMissing.length) {
systemContent += `\n\nLEAD COMPLETION (hard rule): the customer shows buying intent but we still miss ${preMissing.join(', ')}. At the END of your reply, ask naturally for the missing piece(s) in the customer's language. Do not end without asking.`;
}
messages.unshift({ role: 'system', content: systemContent });
if (injection.level === 'high') {
console.log(JSON.stringify({ ts: new Date().toISOString(), agent: 'injection_detected', conversation: conv.uuid, matches: injection.matches.length }));
}
// 预检索:政策问题先查 KB,商品问题先查商品,避免模型跳过工具直接编。
const intent = quickClassify(currentText);
const emotion = detectEmotion(currentText);
const emotionStrategy = escalation.applyEmotionStrategy(emotion);
const maxSteps = emotionStrategy.maxToolTurns || MAX_TOOL_STEPS;
if (emotion !== 'neutral') {
const toneLine = emotion === 'angry'
? 'TONE: The customer is angry. Acknowledge their frustration, be empathetic, and escalate/hand off if needed.'
: emotion === 'frustrated'
? 'TONE: The customer is frustrated. Be patient and reassuring.'
: 'TONE: The customer is urgent. Be direct and concise.';
messages[0].content += `\n\n${toneLine}`;
}
let preResult = null;
try {
let preTool = null;
if (intent === INTENTS.POLICY_FAQ) preTool = byName.get('search_kb');
else if (intent === INTENTS.PRODUCT_SEARCH) preTool = byName.get('search_products');
if (preTool) {
const started = Date.now();
preResult = await preTool.handler({ query: currentText, limit: 5 }, ctx);
stats.toolCalls.push({ name: preTool.name, ms: Date.now() - started, ok: true, forced: true });
stats.toolResults.push(preResult);
console.log(JSON.stringify({ ts: new Date().toISOString(), agent: 'tool', tool: preTool.name, forced: true, ms: Date.now() - started, conversation: conv.uuid, ok: true }));
messages.splice(1, 0, {
role: 'system',
content: `Pre-retrieved tool result for this customer message (${preTool.name}). Use it as the primary source; if it is insufficient, call tools again.\n${markToolResult(preTool.name, JSON.stringify(preResult).slice(0, 6000))}`,
});
}
} catch (err) {
console.log(JSON.stringify({ ts: new Date().toISOString(), agent: 'pre_tool_error', conversation: conv.uuid, error: err.message }));
}
const lang = /[\u3400-\u9fff]/.test(currentText) ? 'zh' : 'en';
const systemHash = crypto.createHash('sha256').update(systemContent).digest('hex').slice(0, 24);
let kbFingerprint = '';
if (preResult && Array.isArray(preResult.results)) {
kbFingerprint = crypto.createHash('sha256').update(preResult.results.map((h) => `${h.question}|${(h.source_urls || []).join(',')}`).join('\n')).digest('hex').slice(0, 24);
}
let finalContent = '';
if (intent === INTENTS.POLICY_FAQ && llmCache.enabled) {
const cached = await llmCache.get({
query: currentText,
intent,
verifiedScope: ctx.verified ? 'verified' : 'anonymous',
lang,
kbFingerprint,
systemHash,
});
if (cached) {
const outbound = /<(p|br|a|ul|ol|strong|div)\b/i.test(cached) ? cached : markdownToHtml(cached);
await client.sendMessage(conv.uuid, outbound, { echoId: `bridge-ai-${webhookMessage.uuid}` });
await metrics.record({
conversationUuid: conv.uuid, inboxId: conv.inbox_id, channel: conv.channel || 'livechat',
verified: ctx.verified, model: llm.model, latencyMs: Date.now() - startedAt,
llmCalls: 0, promptTokens: 0, completionTokens: 0, totalTokens: 0,
toolCalls: stats.toolCalls, audit: { ...auditAnswer({ finalContent: cached, toolResults: stats.toolResults, toolCalls: stats.toolCalls }), cache_hit: true },
replyLength: cached.length,
});
prom.inc('bridge_ai_cache_hits_total');
console.log(JSON.stringify({ ts: new Date().toISOString(), agent: 'cache_hit', conversation: conv.uuid, length: cached.length }));
return cached;
}
}
for (let step = 0; step < maxSteps; step += 1) {
prom.inc('bridge_ai_llm_calls_total');
const llmStarted = Date.now();
const response = await llm.chat({
messages,
tools: toolDefs,
temperature: (variant.overriddenParams && variant.overriddenParams.temperature) ?? 0.3,
maxTokens: (variant.overriddenParams && variant.overriddenParams.maxTokens) ?? 1200,
});
stats.llmCalls += 1;
traceSteps.push({ step_type: 'llm', turn: step, model: llm.model, prompt_tokens: response.usage?.prompt_tokens, completion_tokens: response.usage?.completion_tokens, duration_ms: Date.now() - llmStarted, finish_reason: response.finishReason || null });
if (response.usage) {
stats.promptTokens += Number(response.usage.prompt_tokens) || 0;
stats.completionTokens += Number(response.usage.completion_tokens) || 0;
stats.totalTokens += Number(response.usage.total_tokens) || 0;
}
// 兜底(旧 bridge 同款):product_search 意图下首轮 LLM 一个工具都没调时,
// 丢弃它凭记忆编的回答,强制查一次真实商品结果再让它重答,避免幻觉商品/链接进入回复。
if (!(response.toolCalls && response.toolCalls.length) && step === 0 && intent === INTENTS.PRODUCT_SEARCH && !stats.toolCalls.length) {
const searchTool = byName.get('search_products');
if (searchTool) {
const started = Date.now();
try {
const forced = await searchTool.handler({ query: currentText, limit: 5 }, ctx);
stats.toolCalls.push({ name: 'search_products', ms: Date.now() - started, ok: true, forced: true });
stats.toolResults.push(forced);
traceSteps.push({ step_type: 'tool', turn: step, tool_name: 'search_products', duration_ms: Date.now() - started });
messages.push({ role: 'assistant', content: '', tool_calls: [{ id: 'forced-search-products', type: 'function', function: { name: 'search_products', arguments: JSON.stringify({ query: currentText, limit: 5 }) } }] });
messages.push({ role: 'tool', tool_call_id: 'forced-search-products', content: markToolResult('search_products', JSON.stringify(forced).slice(0, 6000)) });
console.log(JSON.stringify({ ts: new Date().toISOString(), agent: 'forced_search_products', conversation: conv.uuid, ok: true }));
continue;
} catch (err) {
console.log(JSON.stringify({ ts: new Date().toISOString(), agent: 'forced_search_products', conversation: conv.uuid, ok: false, error: err.message }));
}
}
}
if (response.toolCalls && response.toolCalls.length) {
messages.push({ role: 'assistant', content: response.content || '', tool_calls: response.toolCalls });
for (const call of response.toolCalls) {
const name = call.function && call.function.name;
const tool = byName.get(name);
let result;
const started = Date.now();
try {
if (!tool) throw new Error(`unknown tool: ${name}`);
const args = call.function.arguments ? JSON.parse(call.function.arguments) : {};
// verify_email_code 成功后,后续工具要立刻看到 verified/email 变化
const latest = await identityFor(conv);
ctx.verified = latest.verified;
ctx.email = latest.email;
result = await tool.handler(args, ctx);
} catch (err) {
result = { error: err.code || err.message || 'tool_failed', message: err.message || String(err) };
}
prom.inc('bridge_ai_tool_calls_total', { tool: name, ok: !(result && result.error) });
traceSteps.push({ step_type: 'tool', turn: step, tool_name: name, duration_ms: Date.now() - started });
stats.toolCalls.push({ name, ms: Date.now() - started, ok: !(result && result.error) });
stats.toolResults.push(result);
console.log(JSON.stringify({
ts: new Date().toISOString(),
agent: 'tool',
tool: name,
ms: Date.now() - started,
conversation: conv.uuid,
verified: ctx.verified,
ok: !(result && result.error),
}));
messages.push({ role: 'tool', tool_call_id: call.id, content: markToolResult(name, JSON.stringify(result).slice(0, 6000)) });
}
continue;
}
finalContent = String(response.content || '').trim();
break;
}
if (!finalContent) {
finalContent = '抱歉,我这边暂时无法处理,请您稍后再试。';
}
// 确定性链接兜底(旧 bridge buildProductListText 同款):LLM 漏写工具返回的真实商品/类目 URL 时,
// 由代码在回复末尾补齐,保证客户至少拿到真实可点链接(放在护栏前,补的链接同样过白名单校验)。
try {
const backfill = appendMissingProductLinks({
content: finalContent,
toolResults: stats.toolResults,
lang: detectLanguage(currentText),
html: /<(p|br|a|ul|ol|strong|div)\b/i.test(finalContent),
});
if (backfill.appended) {
finalContent = backfill.content;
console.log(JSON.stringify({ ts: new Date().toISOString(), agent: 'link_backfill', conversation: conv.uuid, appended: backfill.appended }));
}
} catch (err) {
console.error(JSON.stringify({ ts: new Date().toISOString(), warn: `link backfill failed: ${err.message}` }));
}
// collect_info 兜底:消息含明显 lead 信息但 LLM 没调 collect_info 时,补抽槽位
if (buyingIntent && leadStore && !(stats.toolCalls || []).some((t) => t.name === 'collect_info')) {
try {
const latest = await leadStore.get(conv.uuid);
if (!latest || !latest.opportunityId) {
const fields = await extractLeadFields(llm, currentText);
if (fields) {
const state = latest ? { ...latest } : { slots: {}, createdFingerprints: [], opportunityId: null, stage: 'NEW' };
state.slots = { ...(state.slots || {}), ...fields };
if (leadStore.saveState) await leadStore.saveState(conv.uuid, state);
else await leadStore.save(conv.uuid, state.slots);
console.log(JSON.stringify({ ts: new Date().toISOString(), agent: 'lead_extract_fallback', conversation: conv.uuid, fields: Object.keys(fields) }));
}
}
} catch (err) {
console.error(JSON.stringify({ ts: new Date().toISOString(), warn: `lead extract fallback failed: ${err.message}` }));
}
}
// post-LLM 兜底:LLM 没问公司/联系方式时,代码强制追加多语言追问
let leadFollowup = false;
try {
if (buyingIntent && leadStore) {
const latest = await leadStore.get(conv.uuid);
const latestSlots = (latest && latest.slots) || {};
const missing = missingLeadContacts(latestSlots, ctx.email);
if (missing.length && !alreadyAsksForLead(finalContent)) {
const lang = detectLanguage(currentText);
finalContent = appendLeadFollowup(finalContent, lang);
leadFollowup = true;
console.log(JSON.stringify({ ts: new Date().toISOString(), agent: 'lead_followup', conversation: conv.uuid, lang, missing }));
}
}
} catch (err) {
console.error(JSON.stringify({ ts: new Date().toISOString(), warn: `lead followup failed: ${err.message}` }));
}
// 发送前硬护栏:伪造下单/邮件/订单号、无工具证据的库存价格、非工具 URL
const guard = guardReply({
content: finalContent,
toolResults: stats.toolResults,
toolCalls: stats.toolCalls,
lang: detectLanguage(currentText),
});
if (guard.blocked) prom.inc('bridge_ai_guard_blocked_total', { reason: (guard.reasons || []).join('|') || 'unknown' });
if (guard.blocked || guard.removed_urls.length) {
console.log(JSON.stringify({
ts: new Date().toISOString(),
agent: 'output_guard',
conversation: conv.uuid,
blocked: guard.blocked,
rewritten: guard.rewritten,
reasons: guard.reasons,
removed_urls: guard.removed_urls,
evidence_missing: guard.evidence_missing,
}));
}
finalContent = guard.content;
// 引用一致性检查(语义/数字覆盖)+ 飞轮
const kbHits = [];
const kbSources = [];
(stats.toolCalls || []).forEach((call, i) => {
if (call.name !== 'search_kb') return;
const result = stats.toolResults[i];
for (const hit of (result && result.results) || []) {
kbHits.push(hit);
for (const u of hit.source_urls || []) kbSources.push(u);
}
});
const citation = checkCitation(finalContent, kbHits, intent, [...new Set(kbSources)]);
// 三级升级 + 会话状态
let escalationLevel = 0;
let escalationReason = null;
try {
const latestState = (leadStore && await leadStore.get(conv.uuid)) || {};
const state = {
...latestState,
messagesProcessed: (Number(latestState.messagesProcessed) || 0) + 1,
consecutiveFails: Number(latestState.consecutiveFails) || 0,
};
const topScore = kbHits.length ? Number(kbHits[0].score) || 0 : 0;
const kbConfidence = topScore >= 0.7 ? 'high' : topScore >= 0.55 ? 'medium' : topScore > 0 ? 'low' : 'none';
const decision = escalation.shouldEscalate(
latestState,
intent,
kbConfidence,
Number(latestState.consecutiveFails) || 0,
kbHits.length
);
escalationLevel = decision.level;
prom.inc('bridge_ai_escalations_total', { level: decision.level });
escalationReason = decision.reason;
if (decision.level === 1 && !/handoff|human|人工/i.test(finalContent)) {
finalContent = escalation.appendEscalationOption(finalContent, currentText);
} else if (decision.level === 2) {
await escalation.escalateToHuman({ convUuid: conv.uuid, trigger: decision.reason, history: messages.slice(-6) });
const zh = /[\u4e00-\u9fff]/.test(currentText);
const line = zh ? '\n\n已为您转接人工同事,请稍候。' : '\n\nI have transferred you to a human colleague.';
if (!finalContent.includes(line.trim())) finalContent += line;
}
state.consecutiveFails = (kbConfidence === 'low' || kbConfidence === 'none' || citation.hallucination_flag === true) ? state.consecutiveFails + 1 : 0;
state.kbConfidence = kbConfidence;
if (leadStore && leadStore.saveState) await leadStore.saveState(conv.uuid, state);
else if (leadStore && leadStore.save) await leadStore.save(conv.uuid, state.slots || {});
} catch (err) {
console.error(JSON.stringify({ ts: new Date().toISOString(), warn: `escalation failed: ${err.message}` }));
}
const alreadyHtml = /<(p|br|a|ul|ol|strong|div)\b/i.test(finalContent);
const outboundContent = alreadyHtml ? finalContent : markdownToHtml(finalContent);
await client.sendMessage(conv.uuid, outboundContent, { echoId: `bridge-ai-${webhookMessage.uuid}` });
if (intent === INTENTS.POLICY_FAQ && llmCache.enabled) {
await llmCache.store({
query: currentText,
intent,
verifiedScope: ctx.verified ? 'verified' : 'anonymous',
lang,
kbFingerprint,
systemHash,
reply: finalContent,
});
}
if (citation.hallucination_flag === true || (citation.citation_score !== null && citation.citation_score < 0.5)) {
try {
const sampleId = await flywheel.captureSample({
convId: conv.uuid,
query: currentText,
reply: finalContent,
intent,
kbHits: kbHits.slice(0, 10),
signal: citation.hallucination_flag === true ? 'hallucination' : 'low_citation',
citationScore: citation.citation_score,
});
if (sampleId) {
setImmediate(async () => {
try {
const review = await flywheel.preLabel({ query: currentText, reply: finalContent, kbHits });
if (review) await flywheel.updateReview(sampleId, review);
} catch { /* noop */ }
});
}
await client.sendPrivateNote(conv.uuid, `[⚠️ Citation Alert] score=${citation.citation_score}, severity=${citation.severity || 'low'}. This reply may need review.`).catch(() => {});
console.log(JSON.stringify({ ts: new Date().toISOString(), agent: 'citation_alert', conversation: conv.uuid, citation_score: citation.citation_score, severity: citation.severity }));
} catch (err) {
console.error(JSON.stringify({ ts: new Date().toISOString(), warn: `citation/flywheel failed: ${err.message}` }));
}
}
const audit = auditAnswer({ finalContent, toolResults: stats.toolResults, toolCalls: stats.toolCalls });
audit.injection_level = injection.level;
audit.escalation_level = escalationLevel;
audit.escalation_reason = escalationReason;
audit.experiment = { id: variant.experiment_id, variant: variant.variant };
audit.citation_score = citation.citation_score;
audit.hallucination_flag = citation.hallucination_flag;
audit.uncovered_entities = citation.uncovered_entities;
audit.lead_followup = leadFollowup;
audit.output_guard = { blocked: guard.blocked, rewritten: guard.rewritten, reasons: guard.reasons, removed_urls: guard.removed_urls, evidence_missing: guard.evidence_missing };
await metrics.record({
conversationUuid: conv.uuid,
inboxId: conv.inbox_id,
channel: conv.channel || 'livechat',
verified: ctx.verified,
model: llm.model,
latencyMs: Date.now() - startedAt,
llmCalls: stats.llmCalls,
promptTokens: stats.promptTokens,
completionTokens: stats.completionTokens,
totalTokens: stats.totalTokens,
toolCalls: stats.toolCalls,
audit,
replyLength: finalContent.length,
});
prom.inc('bridge_ai_runs_total');
await traceWriter.writeSteps(traceSteps, traceId, conv.uuid).catch(() => {});
console.log(JSON.stringify({ ts: new Date().toISOString(), agent: 'replied', conversation: conv.uuid, length: finalContent.length, audit }));
return finalContent;
}
async function run(conv, webhookMessage) {
const key = String(conv.uuid);
const previous = locks.get(key) || Promise.resolve();
const current = previous
.catch(() => {})
.then(() => runOnce(conv, webhookMessage));
locks.set(key, current);
try {
return await current;
} finally {
if (locks.get(key) === current) locks.delete(key);
}
}
return { run, tools: byName, toolDefs };
}
module.exports = { createAgent };