diff --git a/backend/package/yuxi/services/agent_run_service.py b/backend/package/yuxi/services/agent_run_service.py
index 169588db..9ed14bf7 100644
--- a/backend/package/yuxi/services/agent_run_service.py
+++ b/backend/package/yuxi/services/agent_run_service.py
@@ -17,6 +17,7 @@ from yuxi.repositories.agent_repository import AgentRepository
from yuxi.repositories.agent_run_repository import TERMINAL_RUN_STATUSES, AgentRunRepository
from yuxi.repositories.conversation_repository import ConversationRepository
from yuxi.services.run_queue_service import (
+ build_run_event_envelope,
get_arq_pool,
get_last_run_stream_seq,
list_run_stream_events,
@@ -273,14 +274,13 @@ async def stream_agent_run_events(
if terminal_seq in {"", "0-0"}:
terminal_seq = None
yield _format_sse(
- {
- "schema_version": 1,
- "run_id": run_id,
- "thread_id": run.thread_id,
- "event": "end",
- "payload": {"status": run.status},
- "created_at": utc_now_naive().isoformat(),
- },
+ build_run_event_envelope(
+ run_id=run_id,
+ thread_id=run.thread_id,
+ event_type="end",
+ payload={"status": run.status},
+ created_at=utc_now_naive().isoformat(),
+ ),
event="end",
event_id=terminal_seq,
)
diff --git a/backend/package/yuxi/services/run_queue_service.py b/backend/package/yuxi/services/run_queue_service.py
index e6c13d35..df93fe32 100644
--- a/backend/package/yuxi/services/run_queue_service.py
+++ b/backend/package/yuxi/services/run_queue_service.py
@@ -55,6 +55,32 @@ def normalize_after_seq(after_seq: str | None) -> str:
return "0-0"
+def build_run_event_envelope(
+ *,
+ run_id: str,
+ event_type: str,
+ payload: dict | None = None,
+ thread_id: str | None = None,
+ created_at: str | None = None,
+) -> dict:
+ return {
+ "schema_version": 1,
+ "run_id": run_id,
+ "thread_id": thread_id,
+ "event": event_type,
+ "payload": payload or {},
+ "created_at": created_at or datetime.now(tz=UTC).isoformat(),
+ }
+
+
+def _payload_thread_id(payload: dict | None) -> str | None:
+ chunk = payload.get("chunk") if isinstance(payload, dict) else None
+ if not isinstance(chunk, dict):
+ return None
+ thread_id = chunk.get("thread_id")
+ return thread_id.strip() if isinstance(thread_id, str) and thread_id.strip() else None
+
+
async def get_redis_client():
global _redis_client
if _redis_client is not None:
@@ -164,14 +190,14 @@ async def append_run_stream_event(run_id: str, event_type: str, payload: dict, *
key = _event_stream_key(run_id)
now = datetime.now(tz=UTC)
now_ms = int(now.timestamp() * 1000)
- envelope = {
- "schema_version": 1,
- "run_id": run_id,
- "thread_id": thread_id,
- "event": event_type,
- "payload": payload or {},
- "created_at": now.isoformat(),
- }
+ event_thread_id = thread_id or _payload_thread_id(payload)
+ envelope = build_run_event_envelope(
+ run_id=run_id,
+ event_type=event_type,
+ payload=payload or {},
+ thread_id=event_thread_id,
+ created_at=now.isoformat(),
+ )
fields = {
"event_type": event_type,
"payload": json.dumps(envelope, ensure_ascii=False),
diff --git a/backend/test/unit/services/test_run_queue_service.py b/backend/test/unit/services/test_run_queue_service.py
index ecccc417..1366fdee 100644
--- a/backend/test/unit/services/test_run_queue_service.py
+++ b/backend/test/unit/services/test_run_queue_service.py
@@ -97,7 +97,11 @@ async def test_run_stream_event_roundtrip(monkeypatch: pytest.MonkeyPatch):
run_id = "run-1"
seq1 = await run_queue_service.append_run_stream_event(run_id, "loading", {"items": [1]})
- seq2 = await run_queue_service.append_run_stream_event(run_id, "finished", {"chunk": {"status": "finished"}})
+ seq2 = await run_queue_service.append_run_stream_event(
+ run_id,
+ "finished",
+ {"chunk": {"status": "finished", "thread_id": "child-thread"}},
+ )
assert seq1 < seq2
@@ -106,6 +110,7 @@ async def test_run_stream_event_roundtrip(monkeypatch: pytest.MonkeyPatch):
assert events[0]["payload"]["schema_version"] == 1
assert events[0]["payload"]["run_id"] == run_id
assert events[0]["payload"]["payload"] == {"items": [1]}
+ assert events[1]["payload"]["thread_id"] == "child-thread"
next_events = await run_queue_service.list_run_stream_events(run_id, after_seq=seq1, limit=100)
assert len(next_events) == 1
diff --git a/docs/develop-guides/roadmap.md b/docs/develop-guides/roadmap.md
index ea615c0e..4230f5fc 100644
--- a/docs/develop-guides/roadmap.md
+++ b/docs/develop-guides/roadmap.md
@@ -67,7 +67,7 @@
- 收敛用户身份命名:原业务登录标识统一改为 `uid`,Agent/LangGraph runtime、conversation、agent_run、sandbox 路径和前端用户态均使用字符串 `uid`;`user_id` 仅保留给外部响应中的数值 `users.id` 或真实外键场景。
- 工作区知识库分类显示:知识库侧边栏按创建者分组为“我的知识库”和“共享知识库”,自己创建的知识库显示在“我的知识库”下,非自己创建的显示在“共享知识库”下;`knowledge_bases` 表新增 `created_by` 字段记录创建者 uid。
- 聊天附件新增 MinIO tmp 临时上传、可选 PDF/图片解析、确认后加入线程附件的流程;前端改为弹窗内上传、解析与确认。
-- 标准化 Agent run/SSE 执行链路:run 创建时持久化输入消息并提交后入队,worker 统一写入 Redis Stream envelope,SSE 输出 `event/data/id`、心跳注释、`Last-Event-ID` 回放和终止 `end` 事件;前端强制使用 run API 并支持 ask_user_question 中断后以 resume run 恢复。
+- 标准化 Agent run/SSE 执行链路:run 创建时持久化输入消息并提交后入队,worker 统一写入 Redis Stream envelope,SSE 输出 `event/data/id`、心跳注释、`Last-Event-ID` 回放和终止 `end` 事件;前端强制使用 run API 并支持 ask_user_question 中断后以 resume run 恢复;事件 envelope 构造收敛到统一 helper,前端优先使用 envelope 一级 `thread_id` 路由。
- 收敛后端模块边界:文档解析从 `plugins.parser` 移动到 `knowledge.parser`,内容审查从 `plugins.guard` 移动到 `services.guard`。
- 收敛文件服务边界:文件预览判断抽为独立服务,Viewer 文件系统的 workspace 分支复用用户 workspace 服务,线程运行时上下文解析从泛化 `filesystem_service` 拆出为 agent runtime helper。
- 升级 DeepAgents 到 0.6.7 并适配新版文件系统协议:SubAgentMiddleware 改为显式 subagent spec,Skills prompt 补齐新版占位符;sandbox/skills backend 复用新版 `ReadResult`、`GlobResult`、`GrepResult` 等协议类型,文件权限在 backend 层明确区分 skills、uploads、outputs 与 workspace,保留最小 `CustomCompositeBackend` 以避免非 route glob 误扫其他 route;Agent 上下文压缩改为复用 DeepAgents SummarizationMiddleware,历史摘要与大工具结果统一 offload 到 outputs。
diff --git a/web/src/components/AgentChatComponent.vue b/web/src/components/AgentChatComponent.vue
index 92cce254..6482d4d7 100644
--- a/web/src/components/AgentChatComponent.vue
+++ b/web/src/components/AgentChatComponent.vue
@@ -45,10 +45,7 @@
-
+
{
const rows = conversations.value.map((conv, index) => ({
type: 'conversation',
key: conv.status === 'streaming' ? 'ongoing-conversation' : `history-${index}`,
- conv
+ conv,
+ displayItems: getConversationDisplayItems(conv)
}))
if (currentThreadConfigNotice.value) {
@@ -1769,27 +1766,10 @@ const handleResizingChange = (isResizingState, clientX = 0) => {
}
// ==================== HELPER FUNCTIONS ====================
-const extractAssistantMessageBody = (message) => {
- let content = typeof message?.content === 'string' ? message.content.trim() : ''
- let reasoningContent = message?.additional_kwargs?.reasoning_content || ''
-
- if (!reasoningContent && content) {
- const thinkRegex = /(.*?)<\/think>|(.*?)$/s
- const thinkMatch = content.match(thinkRegex)
-
- if (thinkMatch) {
- reasoningContent = (thinkMatch[1] || thinkMatch[2] || '').trim()
- content = content.replace(thinkMatch[0], '').trim()
- }
- }
-
- return { content, reasoningContent }
-}
-
const hasVisibleAssistantBody = (message) => {
if (!message || message.type !== 'ai') return true
- const { content, reasoningContent } = extractAssistantMessageBody(message)
+ const { content, reasoningContent } = MessageProcessor.parseAssistantMessageBody(message)
return Boolean(
content ||
reasoningContent ||
@@ -1800,19 +1780,8 @@ const hasVisibleAssistantBody = (message) => {
}
const getMessageToolCalls = (message) => {
- if (!Array.isArray(message?.tool_calls)) return []
-
- return message.tool_calls
- .filter((toolCall) => {
- return (
- toolCall &&
- (toolCall.id || toolCall.name || toolCall.function?.name) &&
- (toolCall.args !== undefined ||
- toolCall.function?.arguments !== undefined ||
- toolCall.tool_call_result !== undefined)
- )
- })
- .map((toolCall) => {
+ return normalizeToolCalls(message?.tool_calls, {
+ mapToolCall: (toolCall) => {
const subagentRun = toolCall.id ? currentSubagentRunById.value.get(String(toolCall.id)) : null
if (!subagentRun) return toolCall
@@ -1821,7 +1790,8 @@ const getMessageToolCalls = (message) => {
subagent_run: subagentRun,
display_label: subagentRun.subagent_name || subagentRun.subagent_type || undefined
}
- })
+ }
+ })
}
// 将 AI 消息拆成“正文块”和“工具块”,再跨消息合并相邻工具块。
diff --git a/web/src/components/AgentMessageComponent.vue b/web/src/components/AgentMessageComponent.vue
index 62b54870..4b7bc2fb 100644
--- a/web/src/components/AgentMessageComponent.vue
+++ b/web/src/components/AgentMessageComponent.vue
@@ -148,6 +148,7 @@ import { storeToRefs } from 'pinia'
import { MessageProcessor } from '@/utils/messageProcessor'
import { normalizeAttachmentPreviews } from '@/utils/file_utils'
import { buildMentionDisplayLabels } from '@/utils/mention_utils'
+import { normalizeToolCalls } from '@/components/ToolCallingResult/toolRegistry'
const props = defineProps({
// 消息角色:'user'|'assistant'|'sent'|'received'
@@ -281,50 +282,13 @@ const messageSources = computed(() => {
return { knowledgeChunks: [], webSources: [] }
})
-// 过滤有效的工具调用
-const validToolCalls = computed(() => {
- if (!props.message.tool_calls || !Array.isArray(props.message.tool_calls)) {
- return []
- }
-
- return props.message.tool_calls.filter((toolCall) => {
- // 过滤掉无效的工具调用
- return (
- toolCall &&
- (toolCall.id || toolCall.name || toolCall.function?.name) &&
- (toolCall.args !== undefined ||
- toolCall.function?.arguments !== undefined ||
- toolCall.tool_call_result !== undefined)
- )
- })
-})
+const validToolCalls = computed(() => normalizeToolCalls(props.message.tool_calls))
const parsedData = computed(() => {
- // Start with default values from the prop to avoid mutation.
- let content = props.message.content.trim() || ''
- let reasoning_content = props.message.additional_kwargs?.reasoning_content || ''
-
- if (reasoning_content) {
- return {
- content,
- reasoning_content
- }
- }
-
- // Regex to find ... or an unclosed ... at the end of the string.
- const thinkRegex = /(.*?)<\/think>|(.*?)$/s
- const thinkMatch = content.match(thinkRegex)
-
- if (thinkMatch) {
- // The captured reasoning is in either group 1 (closed tag) or 2 (unclosed tag).
- reasoning_content = (thinkMatch[1] || thinkMatch[2] || '').trim()
- // Remove the entire matched block from the original content.
- content = content.replace(thinkMatch[0], '').trim()
- }
-
+ const { content, reasoningContent } = MessageProcessor.parseAssistantMessageBody(props.message)
return {
content,
- reasoning_content
+ reasoning_content: reasoningContent
}
})
diff --git a/web/src/components/ToolCallingResult/ToolCallRenderer.vue b/web/src/components/ToolCallingResult/ToolCallRenderer.vue
index e5a5863e..0a11473b 100644
--- a/web/src/components/ToolCallingResult/ToolCallRenderer.vue
+++ b/web/src/components/ToolCallingResult/ToolCallRenderer.vue
@@ -41,7 +41,7 @@ import MysqlDescribeTableTool from './tools/MysqlDescribeTableTool.vue'
import MysqlListTablesTool from './tools/MysqlListTablesTool.vue'
import AskUserQuestionTool from './tools/AskUserQuestionTool.vue'
import ExecuteTool from './tools/ExecuteTool.vue'
-import { getToolCallId, HIDDEN_TOOL_CALL_IDS } from './toolRegistry'
+import { getToolCallId, isHiddenToolCall } from './toolRegistry'
const props = defineProps({
toolCall: {
@@ -91,7 +91,7 @@ const TOOL_RENDERERS = {
}
const currentRenderer = computed(() => TOOL_RENDERERS[toolId.value] || null)
-const isHidden = computed(() => HIDDEN_TOOL_CALL_IDS.includes(toolId.value))
+const isHidden = computed(() => isHiddenToolCall(props.toolCall))
const toolRendererRef = ref(null)
const refreshGraph = () => {
diff --git a/web/src/components/ToolCallingResult/toolRegistry.js b/web/src/components/ToolCallingResult/toolRegistry.js
index b1014e03..14325e1f 100644
--- a/web/src/components/ToolCallingResult/toolRegistry.js
+++ b/web/src/components/ToolCallingResult/toolRegistry.js
@@ -53,4 +53,27 @@ export const HIDDEN_TOOL_CALL_IDS = ['present_artifacts']
export const getToolCallId = (toolCall) => toolCall?.name || toolCall?.function?.name || ''
+export const isHiddenToolCall = (toolCall) => HIDDEN_TOOL_CALL_IDS.includes(getToolCallId(toolCall))
+
+export const isValidToolCall = (toolCall) => {
+ return Boolean(
+ toolCall &&
+ (toolCall.id || toolCall.name || toolCall.function?.name) &&
+ (toolCall.args !== undefined ||
+ toolCall.function?.arguments !== undefined ||
+ toolCall.tool_call_result !== undefined)
+ )
+}
+
+export const normalizeToolCalls = (toolCalls, { includeHidden = false, mapToolCall } = {}) => {
+ if (!Array.isArray(toolCalls)) return []
+
+ return toolCalls
+ .filter((toolCall) => {
+ if (!isValidToolCall(toolCall)) return false
+ return includeHidden || !isHiddenToolCall(toolCall)
+ })
+ .map((toolCall) => (mapToolCall ? mapToolCall(toolCall) : toolCall))
+}
+
export const getToolIcon = (toolId) => TOOL_ICON_MAP[toolId] || null
diff --git a/web/src/components/ToolCallsGroupComponent.vue b/web/src/components/ToolCallsGroupComponent.vue
index b4289bb0..7fe5e863 100644
--- a/web/src/components/ToolCallsGroupComponent.vue
+++ b/web/src/components/ToolCallsGroupComponent.vue
@@ -42,7 +42,7 @@
import { computed, ref, watch } from 'vue'
import { ChevronDown, ChevronRight, Wrench } from 'lucide-vue-next'
import { ToolCallRenderer } from '@/components/ToolCallingResult'
-import { getToolCallId, HIDDEN_TOOL_CALL_IDS } from '@/components/ToolCallingResult/toolRegistry'
+import { getToolCallId, normalizeToolCalls } from '@/components/ToolCallingResult/toolRegistry'
const props = defineProps({
toolCalls: {
@@ -55,20 +55,7 @@ const props = defineProps({
}
})
-const normalizedToolCalls = computed(() => {
- return (props.toolCalls || []).filter((toolCall) => {
- const toolId = getToolCallId(toolCall)
-
- return (
- toolCall &&
- !HIDDEN_TOOL_CALL_IDS.includes(toolId) &&
- (toolCall.id || toolCall.name || toolCall.function?.name) &&
- (toolCall.args !== undefined ||
- toolCall.function?.arguments !== undefined ||
- toolCall.tool_call_result !== undefined)
- )
- })
-})
+const normalizedToolCalls = computed(() => normalizeToolCalls(props.toolCalls))
const shouldCollapseToolCalls = computed(() => normalizedToolCalls.value.length > 0)
const areToolCallsExpanded = ref(false)
diff --git a/web/src/composables/useAgentRunStream.js b/web/src/composables/useAgentRunStream.js
index 93d952d5..62291e4d 100644
--- a/web/src/composables/useAgentRunStream.js
+++ b/web/src/composables/useAgentRunStream.js
@@ -55,9 +55,9 @@ const getThreadIdFromObject = (value) => {
const resolveChunkThreadId = ({ envelope, payload, chunk, fallbackThreadId }) => {
return (
- getThreadIdFromObject(chunk) ||
- getThreadIdFromObject(payload) ||
getThreadIdFromObject(envelope) ||
+ getThreadIdFromObject(payload) ||
+ getThreadIdFromObject(chunk) ||
fallbackThreadId
)
}
diff --git a/web/src/utils/__tests__/messageProcessor.spec.js b/web/src/utils/__tests__/messageProcessor.spec.js
index f573ccff..e29850df 100644
--- a/web/src/utils/__tests__/messageProcessor.spec.js
+++ b/web/src/utils/__tests__/messageProcessor.spec.js
@@ -109,6 +109,12 @@ const run = () => {
assert.equal(conversations[0].messages.at(-1).isLast, true)
assert.equal(conversations[0].status, 'finished')
+ const assistantBody = MessageProcessor.parseAssistantMessageBody({
+ type: 'ai',
+ content: '推理过程最终答案'
+ })
+ assert.deepEqual(assistantBody, { content: '最终答案', reasoningContent: '推理过程' })
+
console.log('messageProcessor extractKnowledgeChunksFromConversation: all assertions passed')
}
diff --git a/web/src/utils/messageProcessor.js b/web/src/utils/messageProcessor.js
index 3f51f5d4..df9e9ebf 100644
--- a/web/src/utils/messageProcessor.js
+++ b/web/src/utils/messageProcessor.js
@@ -283,6 +283,28 @@ export class MessageProcessor {
}
}
+ /**
+ * 解析助手消息正文与推理内容,保持渲染和列表拆分使用同一套规则。
+ * @param {Object} message - AI 消息对象
+ * @returns {{content: string, reasoningContent: string}}
+ */
+ static parseAssistantMessageBody(message) {
+ let content = typeof message?.content === 'string' ? message.content.trim() : ''
+ let reasoningContent = message?.additional_kwargs?.reasoning_content || ''
+
+ if (!reasoningContent && content) {
+ const thinkRegex = /(.*?)<\/think>|(.*?)$/s
+ const thinkMatch = content.match(thinkRegex)
+
+ if (thinkMatch) {
+ reasoningContent = (thinkMatch[1] || thinkMatch[2] || '').trim()
+ content = content.replace(thinkMatch[0], '').trim()
+ }
+ }
+
+ return { content, reasoningContent }
+ }
+
/**
* 合并消息块
* @param {Array} chunks - 消息块数组
@@ -398,96 +420,6 @@ export class MessageProcessor {
}
}
- /**
- * 处理流式响应数据块
- * @param {Object} data - 响应数据
- * @param {Object} onGoingConv - 进行中的对话对象
- * @param {Object} state - 状态对象
- * @param {Function} getAgentHistory - 获取历史记录函数
- * @param {Function} handleError - 错误处理函数
- */
- static async processResponseChunk(data, onGoingConv, state, getAgentHistory, handleError) {
- try {
- switch (data.status) {
- case 'init':
- // 代表服务端收到请求并返回第一个响应
- state.waitingServerResponse = false
- onGoingConv.msgChunks[data.request_id] = [data.msg]
- break
-
- case 'loading':
- if (data.msg.id) {
- if (!onGoingConv.msgChunks[data.msg.id]) {
- onGoingConv.msgChunks[data.msg.id] = []
- }
- onGoingConv.msgChunks[data.msg.id].push(data.msg)
- }
- break
-
- case 'error':
- console.error('流式处理出错:', data.message)
- handleError(new Error(data.message), 'stream')
- break
-
- case 'finished':
- await getAgentHistory()
- break
-
- default:
- console.warn('未知的响应状态:', data.status)
- }
- } catch (error) {
- handleError(error, 'stream')
- }
- }
-
- /**
- * 处理流式响应
- * @param {Response} response - 响应对象
- * @param {Function} processChunk - 处理块的函数
- * @param {Function} scrollToBottom - 滚动到底部函数
- * @param {Function} handleError - 错误处理函数
- */
- static async handleStreamResponse(response, processChunk, scrollToBottom, handleError) {
- try {
- const reader = response.body.getReader()
- let buffer = ''
- const decoder = new TextDecoder()
-
- while (true) {
- const { done, value } = await reader.read()
- if (done) break
-
- buffer += decoder.decode(value, { stream: true })
- const lines = buffer.split('\n')
- buffer = lines.pop() || '' // 保留最后一行可能不完整的内容
-
- for (const line of lines) {
- if (line.trim()) {
- try {
- const data = JSON.parse(line.trim())
- await processChunk(data)
- } catch (e) {
- console.debug('解析JSON出错:', e.message)
- }
- }
- }
- await scrollToBottom()
- }
-
- // 处理缓冲区中可能剩余的内容
- if (buffer.trim()) {
- try {
- const data = JSON.parse(buffer.trim())
- await processChunk(data)
- } catch {
- console.warn('最终缓冲区内容无法解析:', buffer)
- }
- }
- } catch (error) {
- handleError(error, 'stream')
- }
- }
}
export default MessageProcessor