From afd83e508454a53ec709614a7234eb56526f3677 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=82=96=E6=B3=BD=E6=B6=9B?= Date: Tue, 24 Feb 2026 18:03:47 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=A2=9E=E5=BC=BAAgent=E6=96=87?= =?UTF-8?q?=E4=BB=B6=E8=83=BD=E5=8A=9B=E5=B9=B6=E8=A1=A5=E5=85=85=E6=B5=81?= =?UTF-8?q?=E5=BC=8F=E8=B0=83=E8=AF=95=E6=97=A5=E5=BF=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/agents/chatbot/graph.py | 2 +- web/src/components/AgentChatComponent.vue | 69 +++++++++++++-- web/src/composables/useAgentStreamHandler.js | 55 +++++++++++- .../utils/__tests__/messageProcessor.spec.js | 86 +++++++++++++++++++ 4 files changed, 201 insertions(+), 11 deletions(-) create mode 100644 web/src/utils/__tests__/messageProcessor.spec.js diff --git a/src/agents/chatbot/graph.py b/src/agents/chatbot/graph.py index 32223288..633aade3 100644 --- a/src/agents/chatbot/graph.py +++ b/src/agents/chatbot/graph.py @@ -19,7 +19,7 @@ def _create_fs_backend(rt): class ChatbotAgent(BaseAgent): name = "智能体助手" description = "基础的对话机器人，可以回答问题，可在配置中启用需要的工具。" - capabilities = ["file_upload"] # 支持文件上传功能 + capabilities = ["file_upload", "files"] # 支持文件上传功能 def __init__(self, **kwargs): super().__init__(**kwargs) diff --git a/web/src/components/AgentChatComponent.vue b/web/src/components/AgentChatComponent.vue index 39157311..4ab81c15 100644 --- a/web/src/components/AgentChatComponent.vue +++ b/web/src/components/AgentChatComponent.vue @@ -337,6 +337,10 @@ const supportsFiles = computed(() => { return capabilities.includes('files') }) +const currentCapabilities = computed(() => { + return currentAgent.value?.capabilities || [] +}) + // AgentState 相关计算属性 const currentAgentState = computed(() => { return currentChatId.value ? getThreadState(currentChatId.value)?.agentState || null : null @@ -949,6 +953,13 @@ const handleSendMessage = async ({ image } = {}) => { threadState.streamAbortController = new AbortController() try { + console.log('[AgentStateDebug][before_send]', { + threadId, + currentAgentId: currentAgentId.value, + capabilities: currentCapabilities.value, + supportsTodo: supportsTodo.value, + supportsFiles: supportsFiles.value + }) const response = await sendMessage({ agentId: currentAgentId.value, threadId: threadId, @@ -967,10 +978,20 @@ const handleSendMessage = async ({ image } = {}) => { } threadState.isStreaming = false } finally { + console.log('[AgentStateDebug][send_finally_start]', { + threadId, + currentAgentId: currentAgentId.value, + supportsTodo: supportsTodo.value, + supportsFiles: supportsFiles.value + }) threadState.streamAbortController = null // 异步加载历史记录，保持当前消息显示直到历史记录加载完成 fetchThreadMessages({ agentId: currentAgentId.value, threadId: threadId, delay: 500 }).finally( () => { + console.log('[AgentStateDebug][history_refreshed_after_send]', { + threadId, + currentAgentId: currentAgentId.value + }) // 历史记录加载完成后，安全地清空当前进行中的对话 resetOnGoingConv(threadId) scrollController.scrollToBottom() @@ -1019,6 +1040,14 @@ const handleApprovalWithStream = async (approved) => { } try { + console.log('[AgentStateDebug][before_resume_send]', { + threadId, + currentAgentId: currentAgentId.value, + approved, + capabilities: currentCapabilities.value, + supportsTodo: supportsTodo.value, + supportsFiles: supportsFiles.value + }) // 使用审批 composable 处理审批 const response = await handleApproval( approved, @@ -1044,6 +1073,12 @@ const handleApprovalWithStream = async (approved) => { } } finally { console.log('🔄 [STREAM] Cleaning up streaming state') + console.log('[AgentStateDebug][resume_finally_start]', { + threadId, + currentAgentId: currentAgentId.value, + supportsTodo: supportsTodo.value, + supportsFiles: supportsFiles.value + }) if (threadState) { threadState.isStreaming = false threadState.streamAbortController = null @@ -1052,6 +1087,10 @@ const handleApprovalWithStream = async (approved) => { // 异步加载历史记录，保持当前消息显示直到历史记录加载完成 fetchThreadMessages({ agentId: currentAgentId.value, threadId: threadId, delay: 500 }).finally( () => { + console.log('[AgentStateDebug][history_refreshed_after_resume]', { + threadId, + currentAgentId: currentAgentId.value + }) // 历史记录加载完成后，安全地清空当前进行中的对话 resetOnGoingConv(threadId) scrollController.scrollToBottom() @@ -1110,14 +1149,15 @@ const handleAgentStateRefresh = async (threadId = null) => { if (!currentAgentId.value) return // 优先使用传入的 threadId，否则使用当前的 currentChatId let chatId = threadId || currentChatId.value - console.log( - '[handleAgentStateRefresh] input threadId:', - threadId, - 'currentChatId:', - currentChatId.value, - 'final chatId:', - chatId - ) + console.log('[AgentStateDebug][manual_refresh_start]', { + inputThreadId: threadId, + currentChatId: currentChatId.value, + finalChatId: chatId, + currentAgentId: currentAgentId.value, + capabilities: currentCapabilities.value, + supportsTodo: supportsTodo.value, + supportsFiles: supportsFiles.value + }) if (!chatId) return await fetchAgentState(currentAgentId.value, chatId) } @@ -1245,6 +1285,19 @@ onMounted(async () => { scrollController.enableAutoScroll() }) +watch( + [currentAgentId, currentCapabilities, supportsTodo, supportsFiles], + ([agentId, capabilities, todo, files]) => { + console.log('[AgentStateDebug][capabilities_changed]', { + currentAgentId: agentId, + capabilities, + supportsTodo: todo, + supportsFiles: files + }) + }, + { immediate: true } +) + watch( currentAgentId, async (newAgentId, oldAgentId) => { diff --git a/web/src/composables/useAgentStreamHandler.js b/web/src/composables/useAgentStreamHandler.js index f51413cf..6e857c54 100644 --- a/web/src/composables/useAgentStreamHandler.js +++ b/web/src/composables/useAgentStreamHandler.js @@ -69,6 +69,7 @@ export function useAgentStreamHandler({ supportsTodo, supportsFiles }) { + const debugPrefix = '[AgentStateDebug]' /** * Process a single stream chunk based on its status * @param {Object} chunk - The parsed JSON chunk @@ -110,17 +111,39 @@ export function useAgentStreamHandler({ return true case 'human_approval_required': + console.log(`${debugPrefix}[approval_required]`, { + threadId, + currentAgentId: unref(currentAgentId) + }) // 使用审批 composable 处理审批请求 return processApprovalInStream(chunk, threadId, unref(currentAgentId)) case 'agent_state': + console.log(`${debugPrefix}[agent_state_chunk]`, { + threadId, + supportsTodo: unref(supportsTodo), + supportsFiles: unref(supportsFiles), + currentAgentId: unref(currentAgentId), + hasAgentState: !!chunk.agent_state, + todoCount: Array.isArray(chunk.agent_state?.todos) ? chunk.agent_state.todos.length : 0, + fileKeys: chunk.agent_state?.files ? Object.keys(chunk.agent_state.files) : [] + }) if ((unref(supportsTodo) || unref(supportsFiles)) && chunk.agent_state) { - console.log('[AgentState]', { + console.log(`${debugPrefix}[agent_state_apply]`, { threadId, todos: chunk.agent_state?.todos || [], files: chunk.agent_state?.files || [] }) threadState.agentState = chunk.agent_state + } else { + console.warn(`${debugPrefix}[agent_state_skip]`, { + reason: 'capability_gate_or_empty_state', + supportsTodo: unref(supportsTodo), + supportsFiles: unref(supportsFiles), + hasAgentState: !!chunk.agent_state, + currentAgentId: unref(currentAgentId), + threadId + }) } return false @@ -128,6 +151,13 @@ export function useAgentStreamHandler({ // 先标记流式结束，但保持消息显示直到历史记录加载完成 if (threadState) { threadState.isStreaming = false + console.log(`${debugPrefix}[finished]`, { + threadId, + currentAgentId: unref(currentAgentId), + hasThreadAgentState: !!threadState.agentState, + supportsTodo: unref(supportsTodo), + supportsFiles: unref(supportsFiles) + }) if ((unref(supportsTodo) || unref(supportsFiles)) && threadState.agentState) { console.log( `[AgentState|Final] ${new Date().toLocaleTimeString()}.${new Date().getMilliseconds()}`, @@ -143,7 +173,11 @@ export function useAgentStreamHandler({ case 'interrupted': // 中断状态，刷新消息历史 - console.warn('[Interrupted] case') + console.warn(`${debugPrefix}[interrupted]`, { + threadId, + message: chunkMessage, + currentAgentId: unref(currentAgentId) + }) if (threadState) { threadState.isStreaming = false } @@ -164,10 +198,27 @@ export function useAgentStreamHandler({ * @param {Function} [onChunk] - Optional callback for each chunk (e.g. for logging) */ const handleAgentResponse = async (response, threadId, onChunk = null) => { + console.log(`${debugPrefix}[stream_start]`, { + threadId, + currentAgentId: unref(currentAgentId), + supportsTodo: unref(supportsTodo), + supportsFiles: unref(supportsFiles) + }) await processStreamResponse(response, (chunk) => { + if (chunk?.status && chunk.status !== 'loading') { + console.log(`${debugPrefix}[chunk_status]`, { + threadId, + status: chunk.status, + requestId: chunk.request_id + }) + } if (onChunk) onChunk(chunk) return handleStreamChunk(chunk, threadId) }) + console.log(`${debugPrefix}[stream_end]`, { + threadId, + currentAgentId: unref(currentAgentId) + }) } return { diff --git a/web/src/utils/__tests__/messageProcessor.spec.js b/web/src/utils/__tests__/messageProcessor.spec.js new file mode 100644 index 00000000..0d6f456d --- /dev/null +++ b/web/src/utils/__tests__/messageProcessor.spec.js @@ -0,0 +1,86 @@ +import assert from 'node:assert/strict' + +import { MessageProcessor } from '../messageProcessor.js' + +const databases = [{ name: '财税库' }, { name: 'DifyKB' }, { name: 'LightGraphKB' }] + +const run = () => { + const conv = { + messages: [ + { + type: 'ai', + tool_calls: [ + { + name: '财税库', + tool_call_result: { + content: JSON.stringify([ + { + content: 'A', + score: 0.9, + metadata: { source: 'doc-a', chunk_id: 'c1', file_id: 'f1', chunk_index: 1 }, + }, + { + content: 'A', + score: 0.8, + metadata: { source: 'doc-a', chunk_id: 'c1', file_id: 'f1', chunk_index: 1 }, + }, + ]), + }, + }, + { + name: 'LightGraphKB', + tool_call_result: { + content: JSON.stringify({ + data: { + chunks: [ + { + content: 'B', + score: 0.4, + metadata: { source: 'doc-b', chunk_id: 'c2', file_id: 'f2', chunk_index: 2 }, + }, + ], + }, + }), + }, + }, + { + name: 'not_kb_tool', + tool_call_result: { + content: JSON.stringify([{ content: 'X', score: 0.99, metadata: { chunk_id: 'cx' } }]), + }, + }, + { + name: 'DifyKB', + tool_call_result: { content: 'not-json' }, + }, + ], + }, + ], + } + + const chunks = MessageProcessor.extractKnowledgeChunksFromConversation(conv, databases) + + // 1. Milvus/Dify 数组提取 + assert.equal(chunks.some((c) => c.content === 'A' && c.kb_name === '财税库'), true) + + // 2. LightRAG data.chunks 提取 + assert.equal(chunks.some((c) => c.content === 'B' && c.kb_name === 'LightGraphKB'), true) + + // 3. 非知识库工具忽略 + assert.equal(chunks.some((c) => c.content === 'X'), false) + + // 4. 非法 JSON 自动跳过 + assert.equal(chunks.some((c) => c.kb_name === 'DifyKB'), false) + + // 5. 去重生效（chunk_id=c1 仅一条） + assert.equal(chunks.filter((c) => c.metadata?.chunk_id === 'c1').length, 1) + + // 6. 分数排序（A 0.9 在 B 0.4 前） + const idxA = chunks.findIndex((c) => c.content === 'A') + const idxB = chunks.findIndex((c) => c.content === 'B') + assert.equal(idxA < idxB, true) + + console.log('messageProcessor extractKnowledgeChunksFromConversation: all assertions passed') +} + +run()