From 272c24da539434439b71de36072f00e9ed585a3a Mon Sep 17 00:00:00 2001 From: shaohuzhang1 Date: Wed, 22 Jul 2026 18:31:16 +0800 Subject: [PATCH] feat: Add questions to conversation records and adapt messages to the new version of conversation flow --- ...d_messages_chatrecord_question_and_more.py | 30 ++ apps/application/models/application_chat.py | 6 + .../serializers/application_chat_record.py | 3 + apps/application/serializers/common.py | 4 + apps/application/workflow/i_node.py | 2 +- .../workflow/message/aggregator/__init__.py | 12 + .../message/aggregator/aggregation_manager.py | 59 ++++ .../message/aggregator/aggregator_factory.py | 53 +++ .../message/aggregator/content_aggregator.py | 45 +++ .../message/aggregator/impl/__init__.py | 12 + .../aggregator/impl/reasoning_aggregator.py | 44 +++ .../aggregator/impl/text_aggregator.py | 44 +++ .../aggregator/impl/tool_aggregator.py | 53 +++ .../workflow/message/struct/content.py | 26 ++ .../workflow/message/struct/form_content.py | 8 + .../message/struct/reasoning_content.py | 5 + .../workflow/message/struct/text_content.py | 5 + .../workflow/message/struct/tool_content.py | 7 + .../nodes/ai_chat_node/ai_chat_node.py | 16 +- apps/application/workflow/workflow_manage.py | 8 +- apps/chat/serializers/chat.py | 44 ++- .../conversation/chat-panel/index.vue | 308 +++++++++++++----- .../shared/use-message-pagination.ts | 30 +- .../conversation/content-list/index.vue | 20 +- .../conversation/content/items/question.vue | 22 +- ui/src/components/conversation/index.vue | 72 +--- 26 files changed, 723 insertions(+), 215 deletions(-) create mode 100644 apps/application/migrations/0016_chatrecord_messages_chatrecord_question_and_more.py create mode 100644 apps/application/workflow/message/aggregator/__init__.py create mode 100644 apps/application/workflow/message/aggregator/aggregation_manager.py create mode 100644 apps/application/workflow/message/aggregator/aggregator_factory.py create mode 100644 apps/application/workflow/message/aggregator/content_aggregator.py create mode 100644 apps/application/workflow/message/aggregator/impl/__init__.py create mode 100644 apps/application/workflow/message/aggregator/impl/reasoning_aggregator.py create mode 100644 apps/application/workflow/message/aggregator/impl/text_aggregator.py create mode 100644 apps/application/workflow/message/aggregator/impl/tool_aggregator.py diff --git a/apps/application/migrations/0016_chatrecord_messages_chatrecord_question_and_more.py b/apps/application/migrations/0016_chatrecord_messages_chatrecord_question_and_more.py new file mode 100644 index 00000000000..bf835deb315 --- /dev/null +++ b/apps/application/migrations/0016_chatrecord_messages_chatrecord_question_and_more.py @@ -0,0 +1,30 @@ +# Generated by Django 5.2.14 on 2026-07-22 08:33 + +import common.encoder.encoder +import django.contrib.postgres.fields +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('application', '0015_chat_execute_type'), + ] + + operations = [ + migrations.AddField( + model_name='chatrecord', + name='messages', + field=django.contrib.postgres.fields.ArrayField(base_field=models.JSONField(), default=list, size=None, verbose_name='响应message'), + ), + migrations.AddField( + model_name='chatrecord', + name='question', + field=models.JSONField(default=dict, encoder=common.encoder.encoder.SystemEncoder, verbose_name='用户的消息'), + ), + migrations.AddField( + model_name='chatrecord', + name='version', + field=models.IntegerField(default=1, verbose_name='版本号'), + ), + ] diff --git a/apps/application/models/application_chat.py b/apps/application/models/application_chat.py index 4b4f7979994..16fb6d432ea 100644 --- a/apps/application/models/application_chat.py +++ b/apps/application/models/application_chat.py @@ -117,6 +117,12 @@ class ChatRecord(AppModelMixin): index = models.IntegerField(verbose_name="对话下标") source = models.JSONField(verbose_name="来源", default=dict) ip_address = models.CharField(max_length=128, verbose_name="ip地址", default='') + version = models.IntegerField(verbose_name="版本号", default=1) + question = models.JSONField(verbose_name="用户的消息", default=dict, encoder=SystemEncoder) + messages = ArrayField(verbose_name="响应message", + base_field=models.JSONField() + , default=list) + workflow_context = models.JSONField(verbose_name="工作流上下文", default=dict, null=True, blank=True) def get_human_message(self): diff --git a/apps/application/serializers/application_chat_record.py b/apps/application/serializers/application_chat_record.py index a2e79ccbc8f..9a8743926a5 100644 --- a/apps/application/serializers/application_chat_record.py +++ b/apps/application/serializers/application_chat_record.py @@ -53,6 +53,9 @@ class Meta: "answer_text_list", "create_time", "update_time", + "version", + "question", + "messages" ] diff --git a/apps/application/serializers/common.py b/apps/application/serializers/common.py index 96357ea187f..688fe2661c9 100644 --- a/apps/application/serializers/common.py +++ b/apps/application/serializers/common.py @@ -347,6 +347,8 @@ def append_chat_record(self, chat_record: ChatRecord): 'run_time': chat_record.run_time, 'source': chat_record.source, 'ip_address': chat_record.ip_address or '', + 'question': chat_record.question, + 'messages': chat_record.messages, 'index': chat_record.index}, defaults={ "vote_status": chat_record.vote_status, @@ -361,6 +363,8 @@ def append_chat_record(self, chat_record: ChatRecord): 'run_time': chat_record.run_time, 'index': chat_record.index, 'source': chat_record.source, + 'question': chat_record.question, + 'messages': chat_record.messages, 'ip_address': chat_record.ip_address or '', }) ChatCountSerializer(data={'chat_id': self.chat_id}).update_chat() diff --git a/apps/application/workflow/i_node.py b/apps/application/workflow/i_node.py index 2f40dcdbabe..b9906956548 100644 --- a/apps/application/workflow/i_node.py +++ b/apps/application/workflow/i_node.py @@ -122,7 +122,7 @@ def complete(self, status, anchors=None, error=None, signal: Optional[Signal] = if anchors is None: anchors = [self.success_anchor() if status == Status.SUCCESS else self.fail_anchor()] self._dispatch(anchors) - self.workflow_manage.assertion_end() + self.workflow_manage.assertion_end(error) def _dispatch(self, anchors): """ diff --git a/apps/application/workflow/message/aggregator/__init__.py b/apps/application/workflow/message/aggregator/__init__.py new file mode 100644 index 00000000000..82b549143e2 --- /dev/null +++ b/apps/application/workflow/message/aggregator/__init__.py @@ -0,0 +1,12 @@ +# coding=utf-8 +""" + @project: MaxKB + @file： __init__.py + @date：2026/7/22 16:24 + @desc: 内容聚合器模块 +""" +from application.workflow.message.aggregator.content_aggregator import ContentAggregator +from application.workflow.message.aggregator.aggregator_factory import AggregatorFactory +from application.workflow.message.aggregator.aggregation_manager import AggregationManager + +__all__ = ['ContentAggregator', 'AggregatorFactory', 'AggregationManager'] diff --git a/apps/application/workflow/message/aggregator/aggregation_manager.py b/apps/application/workflow/message/aggregator/aggregation_manager.py new file mode 100644 index 00000000000..149345c9788 --- /dev/null +++ b/apps/application/workflow/message/aggregator/aggregation_manager.py @@ -0,0 +1,59 @@ +# coding=utf-8 +""" + @project: MaxKB + @file： aggregation_manager.py + @date：2026/7/22 16:24 + @desc: 聚合管理器 +""" +from typing import Dict, List + +from application.workflow.message.struct.content import Content +from application.workflow.message.aggregator.aggregator_factory import AggregatorFactory + + +class AggregationManager: + """ + 聚合管理器 + 管理内容块的聚合，将相同id和类型的内容合并 + """ + + def __init__(self): + self._key_to_index: Dict[str, int] = {} + self._contents: List[Content] = [] + + @property + def contents(self) -> List[Content]: + """获取聚合后的内容列表""" + return self._contents + + def aggregate(self, chunk: Content) -> None: + """ + 聚合内容块 + + @param chunk: 内容块 + """ + key = f"{chunk.id}_{chunk.type.value if hasattr(chunk.type, 'value') else chunk.type}" + + idx = self._key_to_index.get(key) + if idx is None: + # 新key + self._key_to_index[key] = len(self._contents) + self._contents.append(chunk) + else: + # 已存在，聚合 + prev = self._contents[idx] + aggregator = AggregatorFactory.get_aggregator(type(prev)) + self._contents[idx] = aggregator.aggregate(prev, chunk) + + def clear(self) -> None: + """清空聚合器""" + self._contents.clear() + self._key_to_index.clear() + + def get_contents(self) -> List[Dict]: + """ + 获取所有聚合后的内容（字典格式） + + @return: 内容字典列表 + """ + return [content.to_dict() for content in self._contents] diff --git a/apps/application/workflow/message/aggregator/aggregator_factory.py b/apps/application/workflow/message/aggregator/aggregator_factory.py new file mode 100644 index 00000000000..18a96461447 --- /dev/null +++ b/apps/application/workflow/message/aggregator/aggregator_factory.py @@ -0,0 +1,53 @@ +# coding=utf-8 +""" + @project: MaxKB + @file： aggregator_factory.py + @date：2026/7/22 16:24 + @desc: 聚合器工厂 +""" +from typing import Dict, Type, Optional + +from application.workflow.message.struct.content import Content +from application.workflow.message.struct.text_content import TextContent +from application.workflow.message.struct.reasoning_content import ReasoningContent +from application.workflow.message.struct.tool_content import ToolContent +from application.workflow.message.aggregator.content_aggregator import ContentAggregator +from application.workflow.message.aggregator.impl.text_aggregator import TextAggregator +from application.workflow.message.aggregator.impl.reasoning_aggregator import ReasoningAggregator +from application.workflow.message.aggregator.impl.tool_aggregator import ToolAggregator + + +class AggregatorFactory: + """ + 聚合器工厂 + 根据内容类型获取对应的聚合器 + """ + _aggregators: Dict[Type[Content], ContentAggregator] = { + TextContent: TextAggregator(), + ReasoningContent: ReasoningAggregator(), + ToolContent: ToolAggregator(), + } + + @classmethod + def get_aggregator(cls, content_class: Type[Content]) -> ContentAggregator: + """ + 获取聚合器 + + @param content_class: 内容类型 + @return: 聚合器实例 + @raises ValueError: 如果找不到对应的聚合器 + """ + aggregator = cls._aggregators.get(content_class) + if aggregator is None: + raise ValueError(f"No aggregator found for class: {content_class.__name__}") + return aggregator + + @classmethod + def get_aggregator_optional(cls, content_class: Type[Content]) -> Optional[ContentAggregator]: + """ + 获取聚合器（可选） + + @param content_class: 内容类型 + @return: 聚合器实例或None + """ + return cls._aggregators.get(content_class) diff --git a/apps/application/workflow/message/aggregator/content_aggregator.py b/apps/application/workflow/message/aggregator/content_aggregator.py new file mode 100644 index 00000000000..fb3def1f997 --- /dev/null +++ b/apps/application/workflow/message/aggregator/content_aggregator.py @@ -0,0 +1,45 @@ +# coding=utf-8 +""" + @project: MaxKB + @file： content_aggregator.py + @date：2026/7/22 16:24 + @desc: 内容聚合器接口 +""" +from abc import ABC, abstractmethod +from typing import TypeVar, Generic + +from application.workflow.message.struct.content import Content + +T = TypeVar('T', bound=Content) + + +class ContentAggregator(ABC, Generic[T]): + """ + 内容聚合器接口 + 用于合并相同类型的流式内容块 + """ + + @abstractmethod + def aggregate(self, prev: T, chunk: T) -> T: + """ + 聚合两个内容块 + + @param prev: 之前的内容 + @param chunk: 新的内容块 + @return: 合并后的内容 + """ + pass + + def merge_base_fields(self, prev: T, chunk: T, result: T) -> None: + """ + 合并基础字段 + + @param prev: 之前的内容 + @param chunk: 新的内容块 + @param result: 结果对象 + """ + result.id = chunk.id if chunk.id else prev.id + result.status = chunk.status if chunk.status else prev.status + result.node_info = chunk.node_info if chunk.node_info else prev.node_info + result.position = chunk.position if chunk.position else prev.position + result.extra = chunk.extra if chunk.extra else prev.extra diff --git a/apps/application/workflow/message/aggregator/impl/__init__.py b/apps/application/workflow/message/aggregator/impl/__init__.py new file mode 100644 index 00000000000..5b6b3cfe902 --- /dev/null +++ b/apps/application/workflow/message/aggregator/impl/__init__.py @@ -0,0 +1,12 @@ +# coding=utf-8 +""" + @project: MaxKB + @file： __init__.py + @date：2026/7/22 16:24 + @desc: 聚合器实现模块 +""" +from application.workflow.message.aggregator.impl.text_aggregator import TextAggregator +from application.workflow.message.aggregator.impl.reasoning_aggregator import ReasoningAggregator +from application.workflow.message.aggregator.impl.tool_aggregator import ToolAggregator + +__all__ = ['TextAggregator', 'ReasoningAggregator', 'ToolAggregator'] diff --git a/apps/application/workflow/message/aggregator/impl/reasoning_aggregator.py b/apps/application/workflow/message/aggregator/impl/reasoning_aggregator.py new file mode 100644 index 00000000000..c278f6a6887 --- /dev/null +++ b/apps/application/workflow/message/aggregator/impl/reasoning_aggregator.py @@ -0,0 +1,44 @@ +# coding=utf-8 +""" + @project: MaxKB + @file： reasoning_aggregator.py + @date：2026/7/22 16:24 + @desc: ReasoningContent 聚合器 +""" +from application.workflow.message.aggregator.content_aggregator import ContentAggregator +from application.workflow.message.struct.reasoning_content import ReasoningContent + + +class ReasoningAggregator(ContentAggregator[ReasoningContent]): + """ + 推理内容聚合器 + 用于合并流式推理内容块 + """ + + def aggregate(self, prev: ReasoningContent, chunk: ReasoningContent) -> ReasoningContent: + """ + 聚合推理内容 + + @param prev: 之前的内容 + @param chunk: 新的内容块 + @return: 合并后的内容 + """ + if prev is None: + return chunk + + # 合并 content + prev_content = prev.content if prev.content else "" + chunk_content = chunk.content if chunk.content else "" + merged_content = prev_content + chunk_content + + # 合并 status: 优先使用 chunk 的，否则使用 prev 的 + merged_status = chunk.status if chunk.status else prev.status + + # 合并基础字段 + merged_id = chunk.id if chunk.id else prev.id + merged_node_info = chunk.node_info if chunk.node_info else prev.node_info + merged_position = chunk.position if chunk.position else prev.position + + result = ReasoningContent(merged_id, merged_content, merged_status, merged_node_info, merged_position) + + return result diff --git a/apps/application/workflow/message/aggregator/impl/text_aggregator.py b/apps/application/workflow/message/aggregator/impl/text_aggregator.py new file mode 100644 index 00000000000..af519bad38e --- /dev/null +++ b/apps/application/workflow/message/aggregator/impl/text_aggregator.py @@ -0,0 +1,44 @@ +# coding=utf-8 +""" + @project: MaxKB + @file： text_aggregator.py + @date：2026/7/22 16:24 + @desc: TextContent 聚合器 +""" +from application.workflow.message.aggregator.content_aggregator import ContentAggregator +from application.workflow.message.struct.text_content import TextContent + + +class TextAggregator(ContentAggregator[TextContent]): + """ + 文本内容聚合器 + 用于合并流式文本内容块 + """ + + def aggregate(self, prev: TextContent, chunk: TextContent) -> TextContent: + """ + 聚合文本内容 + + @param prev: 之前的内容 + @param chunk: 新的内容块 + @return: 合并后的内容 + """ + if prev is None: + return chunk + + # 合并 content + prev_content = prev.content if prev.content else "" + chunk_content = chunk.content if chunk.content else "" + merged_content = prev_content + chunk_content + + # 合并 status: 优先使用 chunk 的，否则使用 prev 的 + merged_status = chunk.status if chunk.status else prev.status + + # 合并基础字段 + merged_id = chunk.id if chunk.id else prev.id + merged_node_info = chunk.node_info if chunk.node_info else prev.node_info + merged_position = chunk.position if chunk.position else prev.position + + result = TextContent(merged_id, merged_content, merged_status, merged_node_info, merged_position) + + return result diff --git a/apps/application/workflow/message/aggregator/impl/tool_aggregator.py b/apps/application/workflow/message/aggregator/impl/tool_aggregator.py new file mode 100644 index 00000000000..029ddc0224b --- /dev/null +++ b/apps/application/workflow/message/aggregator/impl/tool_aggregator.py @@ -0,0 +1,53 @@ +# coding=utf-8 +""" + @project: MaxKB + @file： tool_aggregator.py + @date：2026/7/22 16:24 + @desc: ToolContent 聚合器 +""" +from application.workflow.message.aggregator.content_aggregator import ContentAggregator +from application.workflow.message.struct.tool_content import ToolContent + + +class ToolAggregator(ContentAggregator[ToolContent]): + """ + 工具内容聚合器 + 用于合并流式工具调用内容块 + """ + + def aggregate(self, prev: ToolContent, chunk: ToolContent) -> ToolContent: + """ + 聚合工具内容 + + @param prev: 之前的内容 + @param chunk: 新的内容块 + @return: 合并后的内容 + """ + if prev is None: + return chunk + + # 合并 content (tool_name) + prev_content = prev.content if prev.content else "" + chunk_content = chunk.content if chunk.content else "" + merged_content = chunk_content if chunk_content else prev_content + + # 合并 arguments + prev_arguments = prev.arguments if prev.arguments else "" + chunk_arguments = chunk.arguments if chunk.arguments else "" + merged_arguments = prev_arguments + chunk_arguments + + # 合并 result + prev_result = prev.result if prev.result else "" + chunk_result = chunk.result if chunk.result else "" + merged_result = prev_result + chunk_result + + # 合并基础字段 + merged_id = chunk.id if chunk.id else prev.id + merged_status = chunk.status if chunk.status else prev.status + merged_node_info = chunk.node_info if chunk.node_info else prev.node_info + merged_position = chunk.position if chunk.position else prev.position + + result = ToolContent(merged_id, merged_content, merged_arguments, merged_result, + merged_status, merged_node_info, merged_position) + + return result diff --git a/apps/application/workflow/message/struct/content.py b/apps/application/workflow/message/struct/content.py index 2763c2093e0..ed57f6bd8a3 100644 --- a/apps/application/workflow/message/struct/content.py +++ b/apps/application/workflow/message/struct/content.py @@ -19,6 +19,13 @@ def __init__(self, _id: str, name: str, status: Status): self.name = name self.status = status + def to_dict(self): + return { + 'id': self.id, + 'name': self.name, + 'status': self.status.value if hasattr(self.status, 'value') else str(self.status), + } + class Position: def __init__(self, _id: str, index: Optional[int] = None, children: Optional['Position'] = None): @@ -26,6 +33,13 @@ def __init__(self, _id: str, index: Optional[int] = None, children: Optional['Po self.index = index self.children = children + def to_dict(self): + return { + 'id': self.id, + 'index': self.index, + 'children': self.children.to_dict() if self.children else None, + } + class Content: def __init__(self, _id, status: Status, _type: ContentType, node_info: NodeInfo, position: Position, **kwargs): @@ -35,3 +49,15 @@ def __init__(self, _id, status: Status, _type: ContentType, node_info: NodeInfo, self.node_info = node_info self.position = position self.extra = kwargs + + def to_dict(self): + result = { + 'id': self.id, + 'type': self.type.value if hasattr(self.type, 'value') else str(self.type), + 'status': self.status.value if hasattr(self.status, 'value') else str(self.status), + 'node_info': self.node_info.to_dict() if self.node_info else None, + 'position': self.position.to_dict() if self.position else None, + } + if self.extra: + result.update(self.extra) + return result diff --git a/apps/application/workflow/message/struct/form_content.py b/apps/application/workflow/message/struct/form_content.py index 1cc5e27dff0..080837d409c 100644 --- a/apps/application/workflow/message/struct/form_content.py +++ b/apps/application/workflow/message/struct/form_content.py @@ -22,3 +22,11 @@ def __init__(self, _id, form_field_list: List[Dict], form_content_format: str, self.is_submit = is_submit self.form_data = form_data or {} super().__init__(_id, status, ContentType.FORM, node_info, position, **kwargs) + + def to_dict(self): + result = super().to_dict() + result['form_field_list'] = self.form_field_list + result['form_content_format'] = self.form_content_format + result['is_submit'] = self.is_submit + result['form_data'] = self.form_data + return result diff --git a/apps/application/workflow/message/struct/reasoning_content.py b/apps/application/workflow/message/struct/reasoning_content.py index 6a5129f4e24..e7903d2e188 100644 --- a/apps/application/workflow/message/struct/reasoning_content.py +++ b/apps/application/workflow/message/struct/reasoning_content.py @@ -15,3 +15,8 @@ class ReasoningContent(Content): def __init__(self, _id, content: str, status: Status, node_info: NodeInfo, position: Position, **kwargs): self.content = content super().__init__(_id, status, ContentType.REASONING, node_info, position, **kwargs) + + def to_dict(self): + result = super().to_dict() + result['content'] = self.content + return result diff --git a/apps/application/workflow/message/struct/text_content.py b/apps/application/workflow/message/struct/text_content.py index c2f49f677b7..a2691778ac4 100644 --- a/apps/application/workflow/message/struct/text_content.py +++ b/apps/application/workflow/message/struct/text_content.py @@ -15,3 +15,8 @@ class TextContent(Content): def __init__(self, _id, content: str, status: Status, node_info: NodeInfo, position: Position, **kwargs): self.content = content super().__init__(_id, status, ContentType.TEXT, node_info, position, **kwargs) + + def to_dict(self): + result = super().to_dict() + result['content'] = self.content + return result diff --git a/apps/application/workflow/message/struct/tool_content.py b/apps/application/workflow/message/struct/tool_content.py index e873a545411..3f89ea6d04b 100644 --- a/apps/application/workflow/message/struct/tool_content.py +++ b/apps/application/workflow/message/struct/tool_content.py @@ -18,3 +18,10 @@ def __init__(self, _id, tool_name: str, arguments: str, result: str, status: Sta self.arguments = arguments self.result = result super().__init__(_id, status, ContentType.TOOL, node_info, position, **kwargs) + + def to_dict(self): + result = super().to_dict() + result['content'] = self.content + result['arguments'] = self.arguments + result['result'] = self.result + return result diff --git a/apps/application/workflow/nodes/ai_chat_node/ai_chat_node.py b/apps/application/workflow/nodes/ai_chat_node/ai_chat_node.py index d7d2720e00b..bbb92d24512 100644 --- a/apps/application/workflow/nodes/ai_chat_node/ai_chat_node.py +++ b/apps/application/workflow/nodes/ai_chat_node/ai_chat_node.py @@ -245,7 +245,7 @@ def execute(self): mcp_source, mcp_servers, mcp_tool_id, mcp_tool_ids, tool_ids, application_ids, skill_tool_ids, mcp_output_enable, chat_model, SystemMessage(system), message_list, history_message, - question, chat_id, workspace_id, workflow_type, reasoning_content_id, text_content_id, + question, chat_id, workspace_id, workflow_type, reasoning_content_id, text_content_id, is_result, ) if not mcp_handled: message_list_with_system = [SystemMessage(system)] + message_list @@ -256,12 +256,7 @@ def execute(self): reasoning_content_id, text_content_id) else: r = chat_model.invoke(message_list_with_system) - self._invoke_response(r, chat_model, message_list_with_system, question.content) - - if is_result: - answer = self.get_context('answer') - node_info = NodeInfo(self.get_node_id(), self.get_node_name(), Status.SUCCESS) - self.write(TextContent(text_content_id, answer, Status.SUCCESS, node_info, Position(self.get_node_id()))) + self._invoke_response(r, chat_model, message_list_with_system, question.content, is_result, text_content_id) def _generate_prompt_question(self, prompt, model, vision, image_list, video_list): images = [] @@ -327,7 +322,7 @@ def _stream_response(self, response, chat_model, message_list, question, self._write_final_context(chat_model, message_list, question, answer, reasoning_content) - def _invoke_response(self, response, chat_model, message_list, question): + def _invoke_response(self, response, chat_model, message_list, question, is_result=False, text_content_id=None): model_setting = self.get_context('model_setting') or {} reasoning = Reasoning( model_setting.get('reasoning_content_start', ''), @@ -344,6 +339,9 @@ def _invoke_response(self, response, chat_model, message_list, question): reasoning_result_end.get('reasoning_content') or '' ) self._write_final_context(chat_model, message_list, question, content, reasoning_content) + if is_result: + node_info = NodeInfo(self.get_node_id(), self.get_node_name(), Status.SUCCESS) + self.write(TextContent(text_content_id, content, Status.SUCCESS, node_info, Position(self.get_node_id()))) def _write_final_context(self, chat_model, message_list, question, answer, reasoning_content): message_tokens = chat_model.get_num_tokens_from_messages(message_list) @@ -359,7 +357,7 @@ def _handle_mcp( tool_ids, application_ids, skill_tool_ids, mcp_output_enable, chat_model, system_prompt, message_list, history_message, question, chat_id, workspace_id, workflow_type, - reasoning_content_id, text_content_id, + reasoning_content_id, text_content_id, is_result=False, ): mcp_servers_config = {} diff --git a/apps/application/workflow/workflow_manage.py b/apps/application/workflow/workflow_manage.py index 157aa5001b2..fd989bca779 100644 --- a/apps/application/workflow/workflow_manage.py +++ b/apps/application/workflow/workflow_manage.py @@ -102,14 +102,14 @@ def next_nodes(self, nodes: Optional[List[Node]]): for inst in instances: self._run_async(inst) - def assertion_end(self): + def assertion_end(self, error=None): with self._lock: if self.done: return if not self.is_end(): return self.done = True # 锁内抢占,保证只有一个线程能往下走 - self.end() # 回调放到锁外,避免回调里再触碰本 manage 造成重入/死锁 + self.end(error) # 回调放到锁外,避免回调里再触碰本 manage 造成重入/死锁 def is_end(self): """ @@ -166,12 +166,12 @@ def write(self, message: Content): """ self.call_back.on_next(self, message) - def end(self): + def end(self, error=None): """ 工作流输出结束的时候调用 @return: None """ - self.call_back.on_complete(self, None) + self.call_back.on_complete(self, error) def get_parameters(self): """ diff --git a/apps/chat/serializers/chat.py b/apps/chat/serializers/chat.py index 898190cf214..4aa6401ac94 100644 --- a/apps/chat/serializers/chat.py +++ b/apps/chat/serializers/chat.py @@ -27,6 +27,7 @@ from application.flow.common import Answer from application.flow.tools import to_stream_response_simple from application.workflow.common import WorkflowType, new_instance +from application.workflow.message.aggregator import AggregationManager from application.workflow.workflow_manage import WorkflowManage, CallBack from application.workflow.nodes import get_start_node from application.workflow.message.struct.text_content import TextContent @@ -77,7 +78,7 @@ def is_valid(self, *, raise_exception=False): class ChatMessageSerializers(serializers.Serializer): - message = serializers.CharField(required=True, label=_("User Questions")) + message = serializers.DictField(required=True, label=_("User Questions")) stream = serializers.BooleanField(required=True, label=_("Is the answer in streaming mode")) re_chat = serializers.BooleanField(required=True, label=_("Do you want to reply again")) @@ -94,10 +95,6 @@ class ChatMessageSerializers(serializers.Serializer): label=_("Node parameters")) form_data = serializers.DictField(required=False, label=_("Global variables")) - image_list = serializers.ListField(required=False, label=_("picture")) - document_list = serializers.ListField(required=False, label=_("document")) - audio_list = serializers.ListField(required=False, label=_("Audio")) - other_list = serializers.ListField(required=False, label=_("Other")) child_node = serializers.DictField(required=False, allow_null=True, label=_("Child Nodes")) @@ -340,7 +337,8 @@ def is_valid_application_simple(self, *, chat_info: ChatInfo, raise_exception=Fa return chat_info def chat_simple(self, chat_info: ChatInfo, instance, base_to_response): - message = instance.get('message') + message_dict = instance.get('message') + message = message_dict.get('content', '') if isinstance(message_dict, dict) else message_dict re_chat = instance.get('re_chat') stream = instance.get('stream') chat_user_id = self.data.get('chat_user_id') @@ -397,7 +395,8 @@ def get_chat_record(chat_info, chat_record_id): def chat_work_flow(self, chat_info: ChatInfo, instance: dict, base_to_response): import queue - message = instance.get('message') + message_dict = instance.get('message') + message = message_dict.get('content', '') if isinstance(message_dict, dict) else message_dict re_chat = instance.get('re_chat') stream = instance.get('stream') chat_user_id = self.data.get("chat_user_id") @@ -405,11 +404,11 @@ def chat_work_flow(self, chat_info: ChatInfo, instance: dict, base_to_response): ip_address = self.data.get('ip_address') source = self.data.get('source') form_data = instance.get('form_data') - image_list = instance.get('image_list') - video_list = instance.get('video_list') - document_list = instance.get('document_list') - audio_list = instance.get('audio_list') - other_list = instance.get('other_list') + image_list = message_dict.get('image_list', []) if isinstance(message_dict, dict) else [] + video_list = message_dict.get('video_list', []) if isinstance(message_dict, dict) else [] + document_list = message_dict.get('document_list', []) if isinstance(message_dict, dict) else [] + audio_list = message_dict.get('audio_list', []) if isinstance(message_dict, dict) else [] + other_list = message_dict.get('other_list', []) if isinstance(message_dict, dict) else [] workspace_id = chat_info.application.workspace_id chat_record_id = instance.get('chat_record_id') position = instance.get('position') @@ -453,13 +452,12 @@ def chat_work_flow(self, chat_info: ChatInfo, instance: dict, base_to_response): } result_queue = queue.Queue() - answer_text = '' - reasoning_text = '' + + aggregation = AggregationManager() def on_next(wf_manage, content): - nonlocal answer_text, reasoning_text + aggregation.aggregate(content) if isinstance(content, TextContent): - answer_text += content.content result_queue.put(('chunk', { 'content': [{ 'id': content.id, @@ -468,7 +466,6 @@ def on_next(wf_manage, content): }] })) elif isinstance(content, ReasoningContent): - reasoning_text += content.content result_queue.put(('chunk', { 'content': [{ 'id': content.id, @@ -515,9 +512,8 @@ def position_to_dict(pos): def on_complete(wf_manage, error): if error: result_queue.put(('error', error)) - else: - self._save_chat_record(chat_info, chat_info.chat_id, chat_record_id_str, - message, answer_text, reasoning_text, wf_manage) + self._save_chat_record(chat_info, chat_info.chat_id, chat_record_id_str, + message, '', '', wf_manage, message_dict, aggregation.get_contents()) result_queue.put(('done', None)) call_back = CallBack(on_next, on_complete) @@ -580,10 +576,10 @@ def generate(): if msg_type == 'error': raise data return base_to_response.to_block_response( - chat_info.chat_id, chat_record_id_str, answer_text, True, 0, 0) + chat_info.chat_id, chat_record_id_str, '', True, 0, 0) def _save_chat_record(self, chat_info, chat_id, chat_record_id, question, answer, - reasoning_content, wf_manage): + reasoning_content, wf_manage, question_data=None, messages=None): context = wf_manage.context message_tokens = sum( v.get('message_tokens', 0) for v in context.values() if isinstance(v, dict) and 'message_tokens' in v) @@ -606,7 +602,9 @@ def _save_chat_record(self, chat_info, chat_id, chat_record_id, question, answer index=len(chat_info.chat_record_list) + 1, ip_address=chat_info.ip_address, source=chat_info.source, - workflow_context=dict(context) if context else None + workflow_context=dict(context) if context else None, + question=question_data if question_data else {}, + messages=messages ) chat_info.append_chat_record(chat_record) chat_info.set_cache() diff --git a/ui/src/components/conversation/chat-panel/index.vue b/ui/src/components/conversation/chat-panel/index.vue index cb92ca3d128..5bde97a2086 100644 --- a/ui/src/components/conversation/chat-panel/index.vue +++ b/ui/src/components/conversation/chat-panel/index.vue @@ -3,7 +3,12 @@
@@ -49,11 +54,7 @@ @mouseenter="showDelete = f.url || ''" @mouseleave="showDelete = ''" > -
+
@@ -104,11 +105,7 @@ {{ f.name }}
-
+
@@ -142,11 +139,7 @@ {{ f.name }}
-
+
@@ -164,11 +157,7 @@ @mouseenter="showDelete = f.url || ''" @mouseleave="showDelete = ''" > -
+
@@ -212,14 +201,12 @@ style="display: none" @change="handleFileSelect" /> - + - + @@ -271,7 +253,7 @@ const props = withDefaults( const emit = defineEmits<{ toggle: [] - send: [text: string, files?: any[]] + refresh: [chatId: string] stop: [] }>() @@ -289,7 +271,9 @@ const imageExts = ['jpg', 'jpeg', 'png', 'gif', 'bmp'] const documentExts = ['pdf', 'docx', 'txt', 'xls', 'xlsx', 'md', 'html', 'csv'] const videoExts = ['mp4', 'avi', 'mkv', 'mov', 'flv', 'wmv'] const audioExts = ['mp3', 'wav', 'ogg', 'aac', 'm4a'] -const acceptList = [...imageExts, ...documentExts, ...videoExts, ...audioExts].map(e => `.${e}`).join(',') +const acceptList = [...imageExts, ...documentExts, ...videoExts, ...audioExts] + .map((e) => `.${e}`) + .join(',') const getExt = (name: string) => name.split('.').pop()?.toLowerCase() || '' const isImage = (name: string) => imageExts.includes(getExt(name)) const isDocument = (name: string) => documentExts.includes(getExt(name)) @@ -297,8 +281,14 @@ const isAudio = (name: string) => audioExts.includes(getExt(name)) const isVideo = (name: string) => videoExts.includes(getExt(name)) interface FileItem { - uid: number; name: string; size: number; raw: File - url?: string; file_id?: string; previewUrl?: string; uploading?: boolean + uid: number + name: string + size: number + raw: File + url?: string + file_id?: string + previewUrl?: string + uploading?: boolean } const fileList = ref([]) @@ -312,13 +302,15 @@ const placeholder = computed(() => { return '输入消息...' }) -const canSend = computed(() => (question.value.content.trim() || fileList.value.length > 0) && !streamLoading.value) +const canSend = computed( + () => (question.value.content.trim() || fileList.value.length > 0) && !streamLoading.value, +) // 分类文件列表 -const imageFiles = computed(() => fileList.value.filter(f => isImage(f.name))) -const documentFiles = computed(() => fileList.value.filter(f => isDocument(f.name))) -const audioFiles = computed(() => fileList.value.filter(f => isAudio(f.name))) -const videoFiles = computed(() => fileList.value.filter(f => isVideo(f.name))) +const imageFiles = computed(() => fileList.value.filter((f) => isImage(f.name))) +const documentFiles = computed(() => fileList.value.filter((f) => isDocument(f.name))) +const audioFiles = computed(() => fileList.value.filter((f) => isAudio(f.name))) +const videoFiles = computed(() => fileList.value.filter((f) => isVideo(f.name))) const getFileIcon = (name: string) => { const ext = getExt(name) @@ -338,7 +330,10 @@ const getFileIcon = (name: string) => { aac: 'https://cdn.jsdelivr.net/npm/@element-plus/icons-vue@2.3.1/dist/svg/headset.svg', m4a: 'https://cdn.jsdelivr.net/npm/@element-plus/icons-vue@2.3.1/dist/svg/headset.svg', } - return iconMap[ext] || 'https://cdn.jsdelivr.net/npm/@element-plus/icons-vue@2.3.1/dist/svg/document.svg' + return ( + iconMap[ext] || + 'https://cdn.jsdelivr.net/npm/@element-plus/icons-vue@2.3.1/dist/svg/document.svg' + ) } const validateFile = (file: File): boolean => { @@ -380,7 +375,7 @@ const addFile = (file: File) => { uploadPromises.value.push(uploadPromise) uploadPromise.finally(() => { - uploadPromises.value = uploadPromises.value.filter(p => p !== uploadPromise) + uploadPromises.value = uploadPromises.value.filter((p) => p !== uploadPromise) }) } @@ -391,7 +386,9 @@ const removeFile = (index: number) => { } const clearFiles = () => { - fileList.value.forEach(f => { if (f.previewUrl) URL.revokeObjectURL(f.previewUrl) }) + fileList.value.forEach((f) => { + if (f.previewUrl) URL.revokeObjectURL(f.previewUrl) + }) fileList.value = [] } @@ -416,9 +413,12 @@ const handleDrop = (e: DragEvent) => { } const categorize = () => { - const images: any[] = []; const documents: any[] = [] - const audio: any[] = []; const video: any[] = []; const files: any[] = [] - fileList.value.forEach(f => { + const images: any[] = [] + const documents: any[] = [] + const audio: any[] = [] + const video: any[] = [] + const files: any[] = [] + fileList.value.forEach((f) => { const entry = { url: f.url || '', file_id: f.file_id || '', name: f.name } if (isImage(f.name)) images.push(entry) else if (documentExts.includes(getExt(f.name))) documents.push(entry) @@ -435,18 +435,83 @@ const send = async () => { if (uploadPromises.value.length) await Promise.all(uploadPromises.value) const text = question.value.content.trim() - const media = categorize() - - let content = text - if (!content) { - const types = [ - media.images.length && '图片', media.documents.length && '文档', - media.audio.length && '音频', media.video.length && '视频', media.files.length && '文件', - ].filter(Boolean) - content = types.length > 1 ? '文件消息' : `${types[0]}消息` + + // 如果没有当前对话，创建新对话 + if (!currentChatId.value) { + const id = await store.openChat(store.applicationId.value) + currentChatId.value = id + await store.loadConversations() } - emit('send', content, fileList.value.length ? fileList.value : undefined) + const cid = currentChatId.value + + // 判断是否是第一条消息，如果是则更新对话的 abstract + const isFirstMessage = messages.value.length === 0 + if (isFirstMessage && text) { + const abstract = text.substring(0, 256) + await store.renameChat(cid, abstract) + } + + // 分类文件 + const images = fileList.value + .filter((f) => isImage(f.name)) + .map((f) => ({ url: f.url, file_id: f.file_id, name: f.name })) + const documents = fileList.value + .filter((f) => isDocument(f.name)) + .map((f) => ({ url: f.url, file_id: f.file_id, name: f.name })) + const audio = fileList.value + .filter((f) => isAudio(f.name)) + .map((f) => ({ url: f.url, file_id: f.file_id, name: f.name })) + const video = fileList.value + .filter((f) => isVideo(f.name)) + .map((f) => ({ url: f.url, file_id: f.file_id, name: f.name })) + const other = fileList.value + .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 + 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: '', + }) + + store.pushMessage(store.createAnswerMessage()) + const aiMsg = messages.value[messages.value.length - 1] + + // 构建 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 } + + store.startStream({ + cid, + request: () => store.chat(cid, payload), + onStream: (chunk: any) => { + store.appendChunk(aiMsg, chunk) + scrollToBottom() + }, + onFinish: () => { + aiMsg.write_ed = true + emit('refresh', cid) + }, + onFailure: () => { + aiMsg.write_ed = true + }, + }) + question.value.content = '' clearFiles() } @@ -467,12 +532,15 @@ const handleKeydown = (e: KeyboardEvent) => { } } else { // 如果同时按下ctrl/shift/cmd/opt +enter，则会换行 - const textarea = inputRef.value?.$el?.querySelector('.el-textarea__inner') as HTMLTextAreaElement + const textarea = inputRef.value?.$el?.querySelector( + '.el-textarea__inner', + ) as HTMLTextAreaElement if (textarea) { const startPos = textarea.selectionStart const endPos = textarea.selectionEnd e.preventDefault() - question.value.content = question.value.content.slice(0, startPos) + '\n' + question.value.content.slice(endPos) + question.value.content = + question.value.content.slice(0, startPos) + '\n' + question.value.content.slice(endPos) nextTick(() => { textarea.setSelectionRange(startPos + 1, startPos + 1) }) @@ -486,15 +554,20 @@ const onScroll = () => { if (!el || el.scrollTop > 60) return } -const scrollToBottom = () => { scroll?.forceBottom() } +const scrollToBottom = () => { + scroll?.forceBottom() +} onMounted(() => { if (msgBoxRef.value) scroll = new Scroll(msgBoxRef.value) }) -watch(() => messages.value.length, () => { - nextTick(() => scroll?.scrollBottom()) -}) +watch( + () => messages.value.length, + () => { + nextTick(() => scroll?.scrollBottom()) + }, +) defineExpose({ scrollToBottom }) @@ -523,38 +596,103 @@ defineExpose({ scrollToBottom }) } .header-btn { - width: 32px; height: 32px; border: none; background: transparent; - cursor: pointer; display: flex; align-items: center; justify-content: center; - border-radius: 6px; color: var(--t2, #606266); flex-shrink: 0; + width: 32px; + height: 32px; + border: none; + background: transparent; + cursor: pointer; + display: flex; + align-items: center; + justify-content: center; + border-radius: 6px; + color: var(--t2, #606266); + flex-shrink: 0; +} +.header-btn:hover { + background: rgba(0, 0, 0, 0.05); } -.header-btn:hover { background: rgba(0, 0, 0, 0.05); } .header-info { - display: flex; align-items: center; gap: 8px; flex: 1; min-width: 0; + display: flex; + align-items: center; + gap: 8px; + flex: 1; + min-width: 0; +} +.header-icon { + width: 24px; + height: 24px; + border-radius: 6px; + object-fit: cover; + flex-shrink: 0; +} +.header-text { + display: flex; + flex-direction: column; + min-width: 0; +} +.header-app-name { + font-size: 11px; + font-weight: 400; + color: var(--t3, #909399); + line-height: 1.2; +} +.header-title { + font-size: 13px; + font-weight: 500; + color: var(--t1, #303133); + white-space: nowrap; + overflow: hidden; + text-overflow: ellipsis; } -.header-icon { width: 24px; height: 24px; border-radius: 6px; object-fit: cover; flex-shrink: 0; } -.header-text { display: flex; flex-direction: column; min-width: 0; } -.header-app-name { font-size: 11px; font-weight: 400; color: var(--t3, #909399); line-height: 1.2; } -.header-title { font-size: 13px; font-weight: 500; color: var(--t1, #303133); white-space: nowrap; overflow: hidden; text-overflow: ellipsis; } .panel-messages { - flex: 1; min-height: 0; overflow-y: auto; padding: 14px 12px; - scrollbar-width: thin; display: flex; flex-direction: column; align-items: center; + flex: 1; + min-height: 0; + overflow-y: auto; + padding: 14px 12px; + scrollbar-width: thin; + display: flex; + flex-direction: column; + align-items: center; } .welcome { - flex: 1; display: flex; flex-direction: column; align-items: center; - justify-content: center; text-align: center; padding: 32px 16px; + flex: 1; + display: flex; + flex-direction: column; + align-items: center; + justify-content: center; + text-align: center; + padding: 32px 16px; +} +.welcome-title { + font-size: 18px; + font-weight: 600; + color: var(--t1, #303133); + margin-bottom: 8px; +} +.welcome-sub { + font-size: 14px; + color: var(--t3, #909399); } -.welcome-title { font-size: 18px; font-weight: 600; color: var(--t1, #303133); margin-bottom: 8px; } -.welcome-sub { font-size: 14px; color: var(--t3, #909399); } .msg-row { - display: flex; flex-direction: column; gap: 6px; margin-bottom: 10px; - max-width: 680px; width: 100%; box-sizing: border-box; + display: flex; + flex-direction: column; + gap: 6px; + margin-bottom: 10px; + max-width: 680px; + width: 100%; + box-sizing: border-box; +} +.msg-row.user { + align-items: flex-end; + flex-direction: row-reverse; +} +.msg-row.assistant { + align-items: flex-start; } -.msg-row.user { align-items: flex-end; flex-direction: row-reverse; } -.msg-row.assistant { align-items: flex-start; } .panel-input { flex-shrink: 0; @@ -573,7 +711,9 @@ defineExpose({ scrollToBottom }) border-radius: 8px; border: 1px solid #ffffff; box-sizing: border-box; - transition: border-color 0.2s, box-shadow 0.2s; + transition: + border-color 0.2s, + box-shadow 0.2s; } .operate-textarea:focus-within { diff --git a/ui/src/components/conversation/common/use-chat-store/shared/use-message-pagination.ts b/ui/src/components/conversation/common/use-chat-store/shared/use-message-pagination.ts index 8083444a553..04686d1672d 100644 --- a/ui/src/components/conversation/common/use-chat-store/shared/use-message-pagination.ts +++ b/ui/src/components/conversation/common/use-chat-store/shared/use-message-pagination.ts @@ -17,16 +17,28 @@ export function useMessagePagination(opts: { const records = res.data?.records || [] const result: ChatMessage[] = [] records.forEach((record: any) => { - result.push({ - id: record.id, - role: 'USER', - content: [{ type: 'QUESTION', content: record.problem_text }], - }) - if (record.answer_text_list) { + // 将 question 和 messages 平铺成一个数组 + const contentList: any[] = [] + + // 添加 question + if (record.question) { + contentList.push(record.question) + } else if (record.problem_text) { + contentList.push({ type: 'QUESTION', content: record.problem_text }) + } + + // 添加 messages + if (record.messages && Array.isArray(record.messages)) { + contentList.push(...record.messages) + } else if (record.answer_text_list) { + contentList.push(...record.answer_text_list.flat()) + } + + if (contentList.length > 0) { result.push({ - id: record.id + '-answer', - role: 'ASSISTANT', - content: record.answer_text_list.flat(), + id: record.id, + role: 'USER', + content: contentList, write_ed: true, }) } diff --git a/ui/src/components/conversation/content-list/index.vue b/ui/src/components/conversation/content-list/index.vue index 8a733c61f0e..2fdbc6f473c 100644 --- a/ui/src/components/conversation/content-list/index.vue +++ b/ui/src/components/conversation/content-list/index.vue @@ -1,7 +1,9 @@ + + diff --git a/ui/src/components/conversation/content/items/question.vue b/ui/src/components/conversation/content/items/question.vue index 0738401a719..08abf1edaaa 100644 --- a/ui/src/components/conversation/content/items/question.vue +++ b/ui/src/components/conversation/content/items/question.vue @@ -1,33 +1,33 @@