Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 7 additions & 19 deletions apps/chat/serializers/chat.py
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,6 @@
from models_provider.tools import get_model_instance_by_model_workspace_id
from system_manage.models.chat_user_token_quota import ChatUserTokenQuota

# 「Chat 行尚未查询」的哨兵,区别于「查询过但不存在(None)」
_CHAT_UNSET = object()


Expand Down Expand Up @@ -200,13 +199,10 @@ def ensure_chat_row(self, question, asker):
self._chat_cache = chat
return chat

def save_chat_record(self, chat_record_id, question, asker, new_record):
"""插入一条占位 ChatRecord(workflow 完成后由 update_chat_record 回填)。"""
chat_id = self.data.get("chat_id")
q_text = question.get("content", "") if isinstance(question, dict) else question
self.ensure_chat_row(q_text, asker)
defaults = {
"chat_id": chat_id,
def get_defaults_record(self, question):
"""构造一条占位 ChatRecord 的字段(workflow 完成后由 update_chat_record 回填)。"""
return {
"chat_id": self.data.get("chat_id"),
"problem_text": "",
"answer_text": "",
"details": {},
Expand All @@ -222,12 +218,6 @@ def save_chat_record(self, chat_record_id, question, asker, new_record):
"question": question,
"messages": [],
}
if new_record:
# 全新记录:直接插入,省掉 update_or_create 的一次 SELECT
ChatRecord(id=chat_record_id, **defaults).save(force_insert=True)
else:
# Form 提交 / 重答:复用同一 chat_record_id
QuerySet(ChatRecord).update_or_create(id=chat_record_id, defaults=defaults)

@staticmethod
def _usage_from_context(workflow_context):
Expand Down Expand Up @@ -288,6 +278,9 @@ def chat_work_flow(self, application, instance: dict, base_to_response):
workflow = new_instance(work_flow, WorkflowType.APPLICATION)

chat_record_id_str = str(uuid.uuid7()) if chat_record_id is None else str(chat_record_id)
self.ensure_chat_row(message, chat_user)
if chat_record_id is None:
ChatRecord(id=chat_record_id_str, **self.get_defaults_record(message_dict)).save(force_insert=True)

parameters = {
"history_chat_record": history_chat_record,
Expand Down Expand Up @@ -317,12 +310,10 @@ def chat_work_flow(self, application, instance: dict, base_to_response):

result_queue = queue.Queue()
aggregation = AggregationManager()
self.save_chat_record(chat_record_id_str, message_dict, chat_user, new_record=chat_record_id is None)

def on_next(wf_manage, content):
aggregation.aggregate(content)
block = content.to_dict()
# 持久化(resume)与 live 队列都放裸内容块,格式化统一交给消费端的 base_to_response
get_message_queue().produce(chat_record_id_str, block)
result_queue.put(("chunk", block))

Expand All @@ -337,9 +328,7 @@ def on_complete(wf_manage, error):
)
messages = aggregation.get_contents()
self.update_chat_record(chat_user_id, chat_record_id_str, wf_manage.context, messages)
# 计数统计放到内容落库之后再更新(挪出进对话前的关键路径)
ChatCountSerializer(data={"chat_id": chat_id}).update_chat()
# 定稿后把本轮记录追加进历史缓存(question 已在建占位时确定,messages 为聚合结果)
ChatHistory(chat_id).append(
ChatRecord(
id=chat_record_id_str,
Expand Down Expand Up @@ -403,7 +392,6 @@ def generate():
yield "data: [DONE]\n\n"
break
if msg_type == "chunk":
# data 是裸内容块(content.to_dict()),格式化交给 base_to_response
frame = base_to_response.to_stream(chat_id, chat_record_id_str, data)
if frame is not None:
yield "data: " + frame + "\n\n"
Expand Down
2 changes: 1 addition & 1 deletion apps/common/handle/impl/response/system_to_response.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ def to_stream(self, chat_id, chat_record_id, block: dict):
{
"chat_id": str(chat_id),
"chat_record_id": str(chat_record_id),
"content": [block],
"content": [{**block, "chat_id": str(chat_id), "chat_record_id": str(chat_record_id)}],
},
ensure_ascii=False,
)
Expand Down
152 changes: 96 additions & 56 deletions ui/src/components/conversation/chat-panel/index.vue
Original file line number Diff line number Diff line change
Expand Up @@ -236,7 +236,7 @@
</template>

<script setup lang="ts">
import { ref, computed, nextTick, watch, onMounted, reactive } from 'vue'
import { ref, computed, nextTick, watch, onMounted, reactive, provide } from 'vue'
import { CircleCloseFilled, Paperclip, Promotion, VideoPause } from '@element-plus/icons-vue'
import ContentList from '../content-list/index.vue'
import Loading from '../loading/index.vue'
Expand Down Expand Up @@ -364,10 +364,8 @@ const addFile = (file: File) => {

const uploadPromise = (async () => {
try {
let cid = currentChatId.value
if (!cid) {
cid = store.newChat()
}
// 惰性取当前 chat_id(幂等),保证文件 source_id 与后续发送的 chat_id 一致
const cid = store.getChatId()
const result = await store.uploadFile(file, cid)
item.url = result.url
const parts = result.url.split('/')
Expand Down Expand Up @@ -446,19 +444,15 @@ const send = async () => {

const text = question.value.content.trim()

// 没有当前对话时本地生成草稿 chat_id(后端首发时 open-if-missing)
if (!currentChatId.value) {
store.newChat()
}

const cid = currentChatId.value
// 惰性取 chat_id(无则前端生成,后端首发时 open-if-missing)
const cid = store.getChatId()

// 首条消息:乐观更新草稿标题;落库交给流结束后的后台 renameChat
// 首条消息:此刻才把会话 push 进左侧列表(新建时不 push),标题用 question 内容乐观更新;
// 落库交给流结束后的后台 renameChat。
const isFirstMessage = messages.value.length === 0
const abstract = text.substring(0, 256)
if (isFirstMessage && text) {
const conv = store.conversations.value.find((c: any) => c.id === cid)
if (conv) conv.abstract = abstract
if (isFirstMessage && !store.conversations.value.some((c: any) => c.id === cid)) {
store.conversations.value.unshift({ id: cid, abstract: text ? abstract : '新对话' })
}

// 分类文件
Expand All @@ -478,54 +472,27 @@ const send = async () => {
.filter((f) => !isImage(f.name) && !isDocument(f.name) && !isAudio(f.name) && !isVideo(f.name))
.map((f) => ({ url: f.url, file_id: f.file_id, name: f.name }))

// 构建 question content
// 构建 question content(USER 气泡展示用)与 API message(发后端用)
const questionContent: any = { type: 'QUESTION', content: text }
if (images.length) questionContent.image_list = images
if (documents.length) questionContent.document_list = documents
if (audio.length) questionContent.audio_list = audio
if (video.length) questionContent.video_list = video
if (other.length) questionContent.other_list = other

store.pushMessage({
role: 'USER',
content: [questionContent],
id: nanoid(),
})

store.pushMessage(store.createAnswerMessage())
const aiMsg = messages.value[messages.value.length - 1]
if (!aiMsg) return

// 构建 API payload
const message: any = { content: text, type: 'QUESTION' }
if (images.length) message.image_list = images
if (documents.length) message.document_list = documents
if (audio.length) message.audio_list = audio
if (video.length) message.video_list = video
if (other.length) message.other_list = other

const payload: any = { message, stream: true, re_chat: false }

loading.value = true
store.startStream({
cid,
request: () => store.chat(cid, payload),
onStream: (chunk) => {
store.appendChunk(aiMsg, chunk)
scrollToBottom()
},
if (images.length) questionContent.image_list = message.image_list = images
if (documents.length) questionContent.document_list = message.document_list = documents
if (audio.length) questionContent.audio_list = message.audio_list = audio
if (video.length) questionContent.video_list = message.video_list = video
if (other.length) questionContent.other_list = message.other_list = other

// 委托通用 sendMessage:新问答(新增 USER 气泡 + 新 answer 消息),
// 首条消息落库后把真实标题写回后端交给 onFinish。
sendMessage({
message,
newQuestion: true,
questionContent,
reChat: false,
onFinish: () => {
scrollToBottom()
loading.value = false
// 首条消息落库后,把真实标题写回后端(此时会话已被 open-if-missing 建好)
if (isFirstMessage && text) {
store.renameChat(cid, abstract).catch(() => {})
}
},
onFailure: () => {
scrollToBottom()
loading.value = false
},
})
question.value.content = ''
clearFiles()
Expand Down Expand Up @@ -592,6 +559,79 @@ watch(
},
)

// 新建/切换会话时清空暂存文件与输入,避免文件的 source_id 与新 chat_id 对不上
watch(
() => store.composerResetSignal.value,
() => {
clearFiles()
question.value.content = ''
},
)

// ── 通用发起对话(注入给内容组件) ──────────────────────────
// 表单提交、以及未来任何“能发起对话”的内容组件都通过它发起/续跑一次对话。
// 调用方只需显式声明 newQuestion:是否新增一个 question(新 USER 气泡 + 新 answer 消息),
// 否则续跑聚合回已存在的那条 assistant 消息(后端复用同一 chat_record)。
interface SendMessageOptions {
message: any // 后端 message dict {content,type,image_list?,...}
newQuestion: boolean
questionContent?: any // newQuestion=true 时 USER 气泡展示的 content 项,缺省用 message
reChat?: boolean // 默认 false;新引擎下仅影响知识节点“换答案”去重
formData?: Record<string, any>
position?: any
chatRecordId?: string | null
chunkId?: string | null
onFinish?: () => void // 收尾钩子(如首条消息落库改标题),在内置结束逻辑后调用
onFailure?: () => void
}

const sendMessage = (opts: SendMessageOptions) => {
const cid = store.getChatId()

// 定位流式写入目标消息
let target: any
if (opts.newQuestion) {
store.pushMessage({ role: 'USER', content: [opts.questionContent ?? opts.message], id: nanoid() })
store.pushMessage(store.createAnswerMessage())
target = messages.value[messages.value.length - 1]
} else {
// 续跑:reload 后按 `${chat_record_id}_ASSISTANT` 命中;实时流中 answer 消息 id 为 '' → 回退最后一条
target =
messages.value.find((m: any) => m.id === `${opts.chatRecordId}_ASSISTANT`) ||
messages.value[messages.value.length - 1]
}
if (!target) return

const payload: any = { message: opts.message, stream: true, re_chat: opts.reChat ?? false }
if (opts.formData) payload.form_data = opts.formData
if (opts.position !== undefined) payload.position = opts.position
if (opts.chatRecordId) payload.chat_record_id = opts.chatRecordId
if (opts.chunkId) payload.chunk_id = opts.chunkId

loading.value = true
store.startStream({
cid,
request: () => store.chat(cid, payload),
onStream: (chunk: any) => {
store.appendChunk(target, chunk)
scrollToBottom()
},
onFinish: () => {
target.write_ed = true
loading.value = false
scrollToBottom()
opts.onFinish?.()
},
onFailure: () => {
target.write_ed = true
loading.value = false
opts.onFailure?.()
},
})
}

provide('sendMessage', sendMessage)

defineExpose({ scrollToBottom })
</script>

Expand Down
48 changes: 16 additions & 32 deletions ui/src/components/conversation/common/use-chat-store/chat/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,9 @@ import type { ChatMessage } from '../../types'
// ── 共享状态(单例) ─────────────────────────────────────
const appInfo = ref<{ name: string; icon: string } | null>(null)
const currentChatId = ref('')
// composer 重置信号:新建/切换会话时自增,chat-panel 监听后清空暂存文件与输入,
// 避免上传文件(source_id 绑旧 chat_id)在切上下文后与新 chat_id 对不上。
const composerResetSignal = ref(0)

// ── 会话 CRUD ─────────────────────────────────────────
const { conversations, loadConversations, loadMore } = useConversationCrud({
Expand Down Expand Up @@ -78,16 +81,13 @@ export function useChatStore() {
})

// ── 会话操作 ─────────────────────────────────────────
// 本地新建:前端直接生成 chat_id(草稿),不预先请求后端 open;
// 首次发消息 / 上传时后端会按该 id 现开会话(open-if-missing)。
const newChat = (): string => {
const id = crypto.randomUUID()
currentChatId.value = id
resetMsgState()
if (!conversations.value.some((c) => c.id === id)) {
conversations.value.unshift({ id, abstract: '新建对话' })
}
return id
// 惰性获取当前 chat_id:已有则原样返回;为空才前端生成一个并赋值。
// 幂等——反复调用返回同一个 id,从而上传文件的 source_id 与后续发送的 chat_id 始终一致。
// 不请求后端 open,不往左侧列表 push(直到首次发消息才入列表,见 chat-panel send)。
const getChatId = (): string => {
if (currentChatId.value) return currentChatId.value
currentChatId.value = crypto.randomUUID()
return currentChatId.value
}

const openChat = async (appId?: string) => {
Expand All @@ -114,26 +114,6 @@ export function useChatStore() {
return { url: res, name: file.name }
}

// ── 发送消息 ─────────────────────────────────────────
const sendMessage = (cid: string, payload: any, aiMsg: ChatMessage) => {
loading.value = true
streamManager.startStream({
cid,
request: () => chat(cid, payload),
onStream: (chunk) => {
appendChunk(aiMsg, chunk)
},
onFinish: () => {
aiMsg.write_ed = true
loading.value = false
},
onFailure: () => {
aiMsg.write_ed = true
loading.value = false
}
})
}

// ── 切换对话 ─────────────────────────────────────────
const switchConversation = async (cid: string) => {
await streamManager.switchConversation({
Expand Down Expand Up @@ -169,7 +149,7 @@ export function useChatStore() {
// 会话
loadConversations,
loadMore,
newChat,
getChatId,
openChat,
deleteChat,
renameChat,
Expand All @@ -183,13 +163,17 @@ export function useChatStore() {
appendChunk,
// 流式
chat,
sendMessage,
startStream: streamManager.startStream,
switchConversation,
closeStream: streamManager.closeStream,
stopWorkflow: (cid: string) => streamManager.stopWorkflow(cid, 'chat'),
cancelWorkflow: (cid: string) => streamManager.cancelWorkflow(cid, 'chat'),
// 文件
uploadFile,
// composer 重置
composerResetSignal,
resetComposer: () => {
composerResetSignal.value++
},
}
}
Loading
Loading