270 lines
6.8 KiB
Plaintext
270 lines
6.8 KiB
Plaintext
import { ChatDataAdapter, SSEChunk } from './chatDataAdapter'
|
|
import { MsgItem } from '@/uni_modules/uni-ai-x/types'
|
|
|
|
/**
|
|
* SSE数据处理器
|
|
* 专门用于处理Server-Sent Events流式数据
|
|
*/
|
|
export class SSEDataProcessor {
|
|
|
|
// 存储累积的SSE数据块
|
|
private sseChunks: SSEChunk[] = []
|
|
|
|
// 当前会话ID
|
|
private chatId: string = ''
|
|
|
|
// 消息映射表，用于累积文本内容
|
|
private messageMap = new Map<string, MsgItem>()
|
|
|
|
constructor(chatId: string) {
|
|
this.chatId = chatId
|
|
}
|
|
|
|
/**
|
|
* 处理单个SSE数据块
|
|
* @param chunk SSE数据块
|
|
* @returns 处理后的MsgItem数组
|
|
*/
|
|
processChunk(chunk: SSEChunk): MsgItem[] {
|
|
// 添加到累积列表
|
|
this.sseChunks.push(chunk)
|
|
|
|
if (!ChatDataAdapter.isRenderableSSEChunk(chunk)) {
|
|
return this.getAllMessages()
|
|
}
|
|
|
|
// 处理数据块
|
|
this.processMessage(chunk)
|
|
|
|
// 返回当前所有消息
|
|
return this.getAllMessages()
|
|
}
|
|
|
|
/**
|
|
* 处理消息累积
|
|
* @param chunk SSE数据块
|
|
*/
|
|
private processMessage(chunk: SSEChunk): void {
|
|
if (!ChatDataAdapter.isRenderableSSEChunk(chunk)) {
|
|
return
|
|
}
|
|
|
|
const messageId = chunk.data.id
|
|
const text = chunk.data.text || ''
|
|
const isFinished = chunk.data.finished
|
|
|
|
// console.log(`Processing message ${messageId}: text="${text}", finished=${isFinished}`)
|
|
|
|
if (!this.messageMap.has(messageId)) {
|
|
// 创建新消息
|
|
// console.log(`Creating new message for ID: ${messageId}`)
|
|
const msgItem = ChatDataAdapter.convertSSEChunkToMsgItem(chunk, this.chatId)
|
|
this.messageMap.set(messageId, msgItem)
|
|
// console.log(`New message created:`, msgItem)
|
|
} else {
|
|
// 累积文本内容，使用更智能的去重逻辑
|
|
// console.log(`Updating existing message for ID: ${messageId}`)
|
|
const existingMsg = this.messageMap.get(messageId)!
|
|
// console.log(`Previous body: "${existingMsg.body}"`)
|
|
|
|
// 智能去重：检查新文本是否是现有内容的子串或重复
|
|
if (this.shouldAppendText(existingMsg.body, text)) {
|
|
existingMsg.body += text
|
|
// console.log(`New body: "${existingMsg.body}"`)
|
|
} else {
|
|
// console.log(`Text skipped (duplicate or redundant): "${text}"`)
|
|
}
|
|
|
|
// 更新状态
|
|
if (isFinished) {
|
|
existingMsg.state = 3 // 完成状态
|
|
console.log(`Message ${messageId} marked as finished`)
|
|
}
|
|
}
|
|
|
|
// console.log(`Current message map size: ${this.messageMap.size}`)
|
|
// this.messageMap.forEach((msg, id) => {
|
|
// console.log(`Message ${id}: body="${msg.body}", state=${msg.state}`)
|
|
// })
|
|
}
|
|
|
|
/**
|
|
* 智能判断是否应该追加文本
|
|
* @param existingBody 现有文本内容
|
|
* @param newText 新文本内容
|
|
* @returns 是否应该追加
|
|
*/
|
|
private shouldAppendText(existingBody: string, newText: string): boolean {
|
|
// 如果新文本为空，不追加
|
|
if (!newText || newText.trim().length === 0) {
|
|
return false
|
|
}
|
|
|
|
// 如果现有内容已经包含新文本，不追加
|
|
if (existingBody.includes(newText)) {
|
|
return false
|
|
}
|
|
|
|
// 如果新文本是现有内容的子串，不追加
|
|
if (newText.length < existingBody.length && existingBody.includes(newText)) {
|
|
return false
|
|
}
|
|
|
|
// 检查是否是重复的标点符号或空格
|
|
const lastChar = existingBody.charAt(existingBody.length - 1)
|
|
const firstChar = newText.charAt(0)
|
|
|
|
// 避免重复的标点符号
|
|
if (lastChar === firstChar && '.,!?;:'.includes(lastChar)) {
|
|
return false
|
|
}
|
|
|
|
// 避免重复的空格
|
|
if (lastChar === ' ' && firstChar === ' ') {
|
|
return false
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
/**
|
|
* 获取所有消息
|
|
* @returns MsgItem数组
|
|
*/
|
|
getAllMessages(): MsgItem[] {
|
|
const result: MsgItem[] = []
|
|
// console.log(`Getting all messages from map (size: ${this.messageMap.size})`)
|
|
|
|
this.messageMap.forEach((value, key) => {
|
|
// console.log(`Adding message ${key}:`, value)
|
|
result.push(value)
|
|
})
|
|
|
|
// console.log(`Returning ${result.length} messages:`, result)
|
|
return result
|
|
}
|
|
|
|
/**
|
|
* 获取指定ID的消息
|
|
* @param messageId 消息ID
|
|
* @returns MsgItem或null
|
|
*/
|
|
getMessage(messageId: string): MsgItem | null {
|
|
return this.messageMap.get(messageId) || null
|
|
}
|
|
|
|
/**
|
|
* 清除所有数据
|
|
*/
|
|
clear(): void {
|
|
this.sseChunks = []
|
|
this.messageMap.clear()
|
|
}
|
|
|
|
/**
|
|
* 获取累积的SSE数据块
|
|
* @returns SSE数据块数组
|
|
*/
|
|
getSSEChunks(): SSEChunk[] {
|
|
return [...this.sseChunks]
|
|
}
|
|
|
|
/**
|
|
* 检查是否有未完成的消息
|
|
* @returns 是否有未完成的消息
|
|
*/
|
|
hasIncompleteMessages(): boolean {
|
|
let hasIncomplete = false
|
|
this.messageMap.forEach((msg) => {
|
|
if (msg.state === 2) { // 未完成状态
|
|
hasIncomplete = true
|
|
}
|
|
})
|
|
return hasIncomplete
|
|
}
|
|
|
|
/**
|
|
* 获取未完成的消息数量
|
|
* @returns 未完成消息数量
|
|
*/
|
|
getIncompleteMessageCount(): number {
|
|
let count = 0
|
|
this.messageMap.forEach((msg) => {
|
|
if (msg.state === 2) { // 未完成状态
|
|
count++
|
|
}
|
|
})
|
|
return count
|
|
}
|
|
}
|
|
|
|
/**
|
|
* 解析SSE数据字符串
|
|
* @param sseString SSE原始字符串
|
|
* @returns SSEChunk数组
|
|
*/
|
|
export function parseSSEString(sseString: string): SSEChunk[] {
|
|
const chunks: SSEChunk[] = []
|
|
|
|
// 按行分割
|
|
const lines = sseString.split('\n')
|
|
|
|
for (const line of lines) {
|
|
const trimmedLine = line.trim()
|
|
|
|
// 跳过空行和注释行
|
|
if (!trimmedLine || trimmedLine.startsWith(':')) {
|
|
continue
|
|
}
|
|
|
|
// 检查是否是data行
|
|
if (trimmedLine.startsWith('data:')) {
|
|
const dataContent = trimmedLine.substring(5).trim()
|
|
|
|
// 跳过[DONE]标记
|
|
if (dataContent === '[DONE]') {
|
|
continue
|
|
}
|
|
|
|
try {
|
|
// 解析JSON数据
|
|
const chunk: SSEChunk = JSON.parse(dataContent)
|
|
chunks.push(chunk)
|
|
} catch (error) {
|
|
console.error('解析SSE数据失败:', error, '原始数据:', dataContent)
|
|
}
|
|
}
|
|
}
|
|
|
|
return chunks
|
|
}
|
|
|
|
/**
|
|
* 处理完整的SSE响应
|
|
* @param sseString 完整的SSE响应字符串
|
|
* @param chatId 聊天ID
|
|
* @returns 处理后的MsgItem数组
|
|
*/
|
|
export function processCompleteSSEResponse(sseString: string, chatId: string): MsgItem[] {
|
|
const chunks = parseSSEString(sseString)
|
|
const processor = new SSEDataProcessor(chatId)
|
|
|
|
chunks.forEach(chunk => {
|
|
processor.processChunk(chunk)
|
|
})
|
|
|
|
return processor.getAllMessages()
|
|
}
|
|
|
|
/**
|
|
* 创建SSE数据处理器实例
|
|
* @param chatId 聊天ID
|
|
* @returns SSEDataProcessor实例
|
|
*/
|
|
export function createSSEProcessor(chatId: string): SSEDataProcessor {
|
|
return new SSEDataProcessor(chatId)
|
|
}
|
|
|
|
// 重新导出SSEChunk类型
|
|
export { SSEChunk } from './chatDataAdapter'
|