/** * 消息处理工具类 */ export class MessageProcessor { /** * 将工具结果与消息合并 * @param {Array} msgs - 消息数组 * @returns {Array} 合并后的消息数组 */ static convertToolResultToMessages(msgs) { const toolResponseMap = new Map() // 构建工具响应映射 for (const item of msgs) { if (item.type === 'tool') { // 使用多种可能的ID字段来匹配工具调用 const toolCallId = item.tool_call_id || item.id if (toolCallId) { toolResponseMap.set(toolCallId, item) } } } // 合并工具调用和响应 const convertedMsgs = msgs.map((item) => { if (item.type === 'ai' && item.tool_calls && item.tool_calls.length > 0) { return { ...item, tool_calls: item.tool_calls.map((toolCall) => { const toolResponse = toolResponseMap.get(toolCall.id) return { ...toolCall, tool_call_result: toolResponse || null } }) } } return item }) return convertedMsgs } /** * 将服务器历史记录转换为对话格式 * @param {Array} serverHistory - 服务器历史记录 * @returns {Array} 对话数组 */ static convertServerHistoryToMessages(serverHistory) { // Filter out standalone 'tool' messages since tool results are already in AI messages' tool_calls // Backend new storage: tool results are embedded in AI messages' tool_calls array with tool_call_result field const filteredHistory = serverHistory.filter((item) => item.type !== 'tool') // 按照对话分组 const conversations = [] let currentConv = null for (const item of filteredHistory) { if (item.type === 'human') { // Start new conversation, finalize previous one if (currentConv) { // Find the last AI message and mark it as final for (let i = currentConv.messages.length - 1; i >= 0; i--) { if (currentConv.messages[i].type === 'ai') { currentConv.messages[i].isLast = true currentConv.status = 'finished' break } } } currentConv = { messages: [item], status: 'loading' } conversations.push(currentConv) } else if (item.type === 'ai' && currentConv) { currentConv.messages.push(item) } } // Mark the last conversation as finished if (currentConv && currentConv.messages.length > 0) { // Find the last AI message and mark it as final for (let i = currentConv.messages.length - 1; i >= 0; i--) { if (currentConv.messages[i].type === 'ai') { currentConv.messages[i].isLast = true currentConv.status = 'finished' break } } } return conversations } /** * 提取一轮对话中所有知识库检索块 * @param {Object} conv - 单轮对话 * @param {Array} databases - 知识库列表 * @returns {Array} 归一化后的检索块 */ static extractKnowledgeChunksFromConversation(conv, databases = []) { if (!conv || !Array.isArray(conv.messages) || conv.messages.length === 0) return [] const databaseNames = new Set( (databases || []).map((db) => db?.name).filter((name) => typeof name === 'string' && name.trim()) ) if (databaseNames.size === 0) return [] const normalizedChunks = [] const dedupSet = new Set() const appendChunk = (chunk, kbName) => { if (!chunk || typeof chunk !== 'object') return const content = typeof chunk.content === 'string' ? chunk.content.trim() : '' if (!content) return const metadata = chunk.metadata && typeof chunk.metadata === 'object' ? chunk.metadata : {} const dedupKey = metadata.chunk_id && typeof metadata.chunk_id === 'string' ? `${kbName}::${metadata.chunk_id}` : `${kbName}::${content}` if (dedupSet.has(dedupKey)) return dedupSet.add(dedupKey) const score = typeof chunk.score === 'number' ? chunk.score : null normalizedChunks.push({ kb_name: kbName, content, score, metadata: { source: metadata.source || '', file_id: metadata.file_id || '', chunk_id: metadata.chunk_id || '', chunk_index: metadata.chunk_index, }, }) } const parseToolResultContent = (content) => { if (Array.isArray(content)) return content if (content && typeof content === 'object') return content if (typeof content === 'string') { try { return JSON.parse(content) } catch { return null } } return null } for (const msg of conv.messages) { if (!msg || msg.type !== 'ai' || !Array.isArray(msg.tool_calls)) continue for (const toolCall of msg.tool_calls) { const kbName = toolCall?.name || toolCall?.function?.name if (!databaseNames.has(kbName)) continue const content = toolCall?.tool_call_result?.content const parsed = parseToolResultContent(content) if (!parsed) continue // Milvus / Dify: 直接是 chunks 数组 if (Array.isArray(parsed)) { for (const chunk of parsed) appendChunk(chunk, kbName) continue } // LightRAG: 结果为对象,chunks 在 data.chunks 下 const lightragChunks = parsed?.data?.chunks if (Array.isArray(lightragChunks)) { for (const chunk of lightragChunks) appendChunk(chunk, kbName) } } } normalizedChunks.sort((a, b) => { const scoreA = typeof a.score === 'number' ? a.score : Number.NEGATIVE_INFINITY const scoreB = typeof b.score === 'number' ? b.score : Number.NEGATIVE_INFINITY return scoreB - scoreA }) return normalizedChunks } /** * 提取一轮对话中的网络搜索来源 * @param {Object} conv - 单轮对话 * @returns {Array} 归一化后的网络来源 */ static extractWebSourcesFromConversation(conv) { if (!conv || !Array.isArray(conv.messages) || conv.messages.length === 0) return [] const webSources = [] const dedupSet = new Set() const parseToolResultContent = (content) => { if (Array.isArray(content)) return content if (content && typeof content === 'object') return content if (typeof content === 'string') { try { return JSON.parse(content) } catch { return null } } return null } for (const msg of conv.messages) { if (!msg || msg.type !== 'ai' || !Array.isArray(msg.tool_calls)) continue for (const toolCall of msg.tool_calls) { const toolName = (toolCall?.name || toolCall?.function?.name || '').toLowerCase() if (!toolName.includes('tavily_search')) continue const content = toolCall?.tool_call_result?.content const parsed = parseToolResultContent(content) const results = Array.isArray(parsed?.results) ? parsed.results : [] if (results.length === 0) continue for (const item of results) { const title = typeof item?.title === 'string' ? item.title.trim() : '' const url = typeof item?.url === 'string' ? item.url.trim() : '' if (!title || !url) continue if (dedupSet.has(url)) continue dedupSet.add(url) webSources.push({ tool_name: toolCall?.name || toolCall?.function?.name || '网络搜索', title, url, score: typeof item?.score === 'number' ? item.score : null, content: typeof item?.content === 'string' ? item.content : '', published_date: typeof item?.published_date === 'string' ? item.published_date : '', }) } } } webSources.sort((a, b) => { const scoreA = typeof a.score === 'number' ? a.score : Number.NEGATIVE_INFINITY const scoreB = typeof b.score === 'number' ? b.score : Number.NEGATIVE_INFINITY return scoreB - scoreA }) return webSources } /** * 提取单个消息中的来源 * @param {Object} message - 消息对象 * @param {Array} databases - 知识库列表 * @returns {{knowledgeChunks: Array, webSources: Array}} */ static extractSourcesFromMessage(message, databases = []) { if (!message || message.type !== 'ai') return { knowledgeChunks: [], webSources: [] } // 复用提取逻辑,通过构建临时对话对象 const mockConv = { messages: [message] } return { knowledgeChunks: MessageProcessor.extractKnowledgeChunksFromConversation(mockConv, databases), webSources: MessageProcessor.extractWebSourcesFromConversation(mockConv), } } /** * 提取一轮对话中的全部来源(知识库+网络搜索) * @param {Object} conv - 单轮对话 * @param {Array} databases - 知识库列表 * @returns {{knowledgeChunks: Array, webSources: Array}} */ static extractSourcesFromConversation(conv, databases = []) { return { knowledgeChunks: MessageProcessor.extractKnowledgeChunksFromConversation(conv, databases), webSources: MessageProcessor.extractWebSourcesFromConversation(conv), } } /** * 合并消息块 * @param {Array} chunks - 消息块数组 * @returns {Object|null} 合并后的消息 */ static mergeMessageChunk(chunks) { if (chunks.length === 0) return null // 深拷贝第一个chunk作为结果 const result = JSON.parse(JSON.stringify(chunks[0])) // 处理用户消息的内容格式 - 确保显示纯文本 if (result.type === 'human' || result.role === 'user') { // 如果content是数组格式(LangChain多模态消息),提取文本部分 if (Array.isArray(result.content)) { const textPart = result.content.find((item) => item.type === 'text') result.content = textPart ? textPart.text : '' } else { result.content = result.content || '' } } else { result.content = result.content || '' } // 合并后续chunks for (let i = 1; i < chunks.length; i++) { const chunk = chunks[i] // 合并内容 if (chunk.content) { result.content += chunk.content } // 合并reasoning_content if (chunk.reasoning_content) { if (!result.reasoning_content) { result.reasoning_content = '' } result.reasoning_content += chunk.reasoning_content } // 合并additional_kwargs中的reasoning_content if (chunk.additional_kwargs?.reasoning_content) { if (!result.additional_kwargs) result.additional_kwargs = {} if (!result.additional_kwargs.reasoning_content) { result.additional_kwargs.reasoning_content = '' } result.additional_kwargs.reasoning_content += chunk.additional_kwargs.reasoning_content } // 合并tool_calls (处理新的数据结构) MessageProcessor._mergeToolCalls(result, chunk) } // 处理AIMessageChunk类型 if (result.type === 'AIMessageChunk') { result.type = 'ai' } return result } /** * 合并工具调用 * @private * @param {Object} result - 结果对象 * @param {Object} chunk - 当前块 */ static _mergeToolCalls(result, chunk) { if (chunk.tool_call_chunks && chunk.tool_call_chunks.length > 0) { // 确保 result 有 tool_calls 数组 if (!result.tool_calls) result.tool_calls = [] for (const toolCallChunk of chunk.tool_call_chunks) { // 使用 index 来标识工具调用(因为可能有多个工具调用) const existingToolCallIndex = result.tool_calls.findIndex( (t) => t.index === toolCallChunk.index ) if (existingToolCallIndex !== -1) { // 合并相同index的tool call const existingToolCall = result.tool_calls[existingToolCallIndex] // 更新名称和ID(如果存在) if (toolCallChunk.name && !existingToolCall.function?.name) { if (!existingToolCall.function) existingToolCall.function = {} existingToolCall.function.name = toolCallChunk.name } if (toolCallChunk.id && !existingToolCall.id) { existingToolCall.id = toolCallChunk.id } // 合并参数 if (toolCallChunk.args) { if (!existingToolCall.function) existingToolCall.function = {} if (!existingToolCall.function.arguments) existingToolCall.function.arguments = '' existingToolCall.function.arguments += toolCallChunk.args } } else { // 添加新的tool call const newToolCall = { index: toolCallChunk.index, id: toolCallChunk.id, function: { name: toolCallChunk.name || null, arguments: toolCallChunk.args || '' } } result.tool_calls.push(newToolCall) } } } } /** * 处理流式响应数据块 * @param {Object} data - 响应数据 * @param {Object} onGoingConv - 进行中的对话对象 * @param {Object} state - 状态对象 * @param {Function} getAgentHistory - 获取历史记录函数 * @param {Function} handleError - 错误处理函数 */ static async processResponseChunk(data, onGoingConv, state, getAgentHistory, handleError) { try { switch (data.status) { case 'init': // 代表服务端收到请求并返回第一个响应 state.waitingServerResponse = false onGoingConv.msgChunks[data.request_id] = [data.msg] break case 'loading': if (data.msg.id) { if (!onGoingConv.msgChunks[data.msg.id]) { onGoingConv.msgChunks[data.msg.id] = [] } onGoingConv.msgChunks[data.msg.id].push(data.msg) } break case 'error': console.error('流式处理出错:', data.message) handleError(new Error(data.message), 'stream') break case 'finished': await getAgentHistory() break default: console.warn('未知的响应状态:', data.status) } } catch (error) { handleError(error, 'stream') } } /** * 处理流式响应 * @param {Response} response - 响应对象 * @param {Function} processChunk - 处理块的函数 * @param {Function} scrollToBottom - 滚动到底部函数 * @param {Function} handleError - 错误处理函数 */ static async handleStreamResponse(response, processChunk, scrollToBottom, handleError) { try { const reader = response.body.getReader() let buffer = '' const decoder = new TextDecoder() while (true) { const { done, value } = await reader.read() if (done) break buffer += decoder.decode(value, { stream: true }) const lines = buffer.split('\n') buffer = lines.pop() || '' // 保留最后一行可能不完整的内容 for (const line of lines) { if (line.trim()) { try { const data = JSON.parse(line.trim()) await processChunk(data) } catch (e) { console.debug('解析JSON出错:', e.message) } } } await scrollToBottom() } // 处理缓冲区中可能剩余的内容 if (buffer.trim()) { try { const data = JSON.parse(buffer.trim()) await processChunk(data) } catch { console.warn('最终缓冲区内容无法解析:', buffer) } } } catch (error) { handleError(error, 'stream') } } } export default MessageProcessor