feat: Add questions to conversation records and adapt messages to the new version of conversation flow

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='版本号'),
+        ),
+    ]

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):

apps/application/serializers/application_chat_record.py

@@ -53,6 +53,9 @@ class Meta:
             "answer_text_list",
             "create_time",
             "update_time",
+            "version",
+            "question",
+            "messages"
         ]
 
 

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()

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):
         """

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']

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]

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)

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

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']

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

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