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
12 changes: 7 additions & 5 deletions apps/application/workflow/content_type.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,12 @@
# coding=utf-8
"""
@project: MaxKB
@Author:虎虎虎
@file: content_type.py
@date:2026/6/30 15:57
@desc:
@project: MaxKB
@Author:虎虎虎
@file: content_type.py
@date:2026/6/30 15:57
@desc:
"""

from enum import Enum


Expand All @@ -17,3 +18,4 @@ class ContentType(Enum):
CONTINUE = "CONTINUE"
BREAK = "BREAK"
FORM = "FORM"
PROGRESS = "PROGRESS"
17 changes: 11 additions & 6 deletions apps/application/workflow/message/aggregator/aggregator_factory.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,16 @@
# coding=utf-8
"""
@project: MaxKB
@file: aggregator_factory.py
@date:2026/7/22 16:24
@desc: 聚合器工厂
@project: MaxKB
@file: aggregator_factory.py
@date:2026/7/22 16:24
@desc: 聚合器工厂
"""

from typing import Dict, Type, Optional

from application.workflow.message.aggregator.impl import ProgressAggregator
from application.workflow.message.struct.content import Content
from application.workflow.message.struct.progress_content import ProgressContent
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
Expand All @@ -22,17 +25,19 @@ class AggregatorFactory:
聚合器工厂
根据内容类型获取对应的聚合器
"""

_aggregators: Dict[Type[Content], ContentAggregator] = {
TextContent: TextAggregator(),
ReasoningContent: ReasoningAggregator(),
ToolContent: ToolAggregator(),
ProgressContent: ProgressAggregator(),
}

@classmethod
def get_aggregator(cls, content_class: Type[Content]) -> ContentAggregator:
"""
获取聚合器

@param content_class: 内容类型
@return: 聚合器实例
@raises ValueError: 如果找不到对应的聚合器
Expand All @@ -46,7 +51,7 @@ def get_aggregator(cls, content_class: Type[Content]) -> ContentAggregator:
def get_aggregator_optional(cls, content_class: Type[Content]) -> Optional[ContentAggregator]:
"""
获取聚合器(可选)

@param content_class: 内容类型
@return: 聚合器实例或None
"""
Expand Down
12 changes: 7 additions & 5 deletions apps/application/workflow/message/aggregator/impl/__init__.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,14 @@
# coding=utf-8
"""
@project: MaxKB
@file: __init__.py
@date:2026/7/22 16:24
@desc: 聚合器实现模块
@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
from application.workflow.message.aggregator.impl.progress_aggregator import ProgressAggregator

__all__ = ['TextAggregator', 'ReasoningAggregator', 'ToolAggregator']
__all__ = ["TextAggregator", "ReasoningAggregator", "ToolAggregator", "ProgressAggregator"]
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
# coding=utf-8
"""
@project: MaxKB
@file: progress_aggregator.py
@date:2026/7/22 16:24
@desc: ReasoningContent 聚合器
"""

from application.workflow.message.aggregator.content_aggregator import ContentAggregator
from application.workflow.message.struct.progress_content import ProgressContent


class ProgressAggregator(ContentAggregator[ProgressContent]):
"""
推理内容聚合器
用于合并流式推理内容块
"""

def aggregate(self, prev: ProgressContent, chunk: ProgressContent) -> ProgressContent:
"""
聚合推理内容

@param prev: 之前的内容
@param chunk: 新的内容块
@return: 合并后的内容
"""
if prev is None:
return chunk

# 合并 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 = ProgressContent(merged_id, merged_status, merged_node_info, merged_position)

return result
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@

class ProgressContent(Content):
def __init__(self, _id, status: Status, node_info: NodeInfo, position: Position, **kwargs):
super().__init__(_id, status, ContentType.REASONING, node_info, position, **kwargs)
super().__init__(_id, status, ContentType.PROGRESS, node_info, position, **kwargs)

def to_dict(self):
result = super().to_dict()
Expand Down
13 changes: 12 additions & 1 deletion ui/src/components/conversation/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -104,10 +104,21 @@ const FORM = (prev: any, chunk: any) => {
}
}

const PROGRESS = (prev: any = {}, chunk: any) => {
return {
type: 'PROGRESS',
id: chunk.id ?? prev.id,
status: chunk.status ?? prev.status,
workflowRunId: chunk.workflowRunId ?? prev.workflowRunId,
extra: chunk.extra ?? prev.extra
}
}

export const aggregators: Record<string, (prev: any, chunk: any) => any> = {
TEXT,
REASONING,
FAILURE,
TOOL,
FORM
FORM,
PROGRESS
}
Loading