import { useState, useRef, useCallback } from 'react'; /** * 会话初始状态 * * status 枚举: * - 'idle': 未启动 * - 'running': 正在运行(流式输出中) * - 'waiting_human': 等待人工确认(仅 human_review 模式) * - 'done': 分析完成 * - 'error': 发生错误 * * phase 枚举: * - 'starting': 启动阶段(第 1 轮,协调者分配初始任务) * - 'diverging': 辩论阶段(多智能体并行输出) * - 'closing': 收敛阶段(协调者整合草稿) * - 'done': 完成 */ const initialState = { status: 'idle', sessionId: null, backendUrl: '', coordinatorMode: 'auto', phase: null, currentRound: 0, totalRounds: 0, divergeRounds: 0, reportSchema: {}, agents: [], agentStates: {}, coordinatorUpdates: [], latestDraft: {}, sharedMemory: null, backgroundRetrieval: null, finalReport: null, finalReportPartial: {}, currentFinalField: null, pendingReview: null, reviewSubmitting: false, reviewRevising: false, errors: [], timeline: [], startConfig: null, }; /** * 规范化运行时智能体配置 * * 将模板中的原始 agents 数组转换为运行时格式, * 确保每个 agent 都包含所有必需字段(id、name、dimension、expertise 等)。 * * @param {Array} agents - 原始智能体配置数组 * @param {Object} reportSchema - 报告结构定义,用于推断 relevant_fields * @returns {Array} 规范化后的智能体数组 */ function normalizeRuntimeAgents(agents = [], reportSchema = {}) { const allFields = Object.keys(reportSchema || {}); return agents.map((agent, index) => { const id = agent.id || `agent_${index + 1}`; const description = agent.description || agent.expertise || agent.dimension || ''; const skill = agent.skill || agent.expertise || description; return { id, name: agent.name || id, description, skill, dimension: agent.dimension || agent.name || id, scenario_role: agent.scenario_role || agent.expertise || description, relevant_fields: Array.isArray(agent.relevant_fields) && agent.relevant_fields.length > 0 ? agent.relevant_fields : allFields, analysis_contract: Array.isArray(agent.analysis_contract) ? agent.analysis_contract : [ '输出自然语言分析,不输出 JSON。', '充分展开材料、依据、反例、分歧和边界条件,最终收缩由协调者完成。', ], retrieval_mode: agent.retrieval_mode || 'always', allowed_source_ids: agent.allowed_source_ids || [], retrieval_instruction: agent.retrieval_instruction || '优先检索与本轮任务相关的最新新闻、公开事实和可核验线索。', }; }); } /** * 构建发送给后端的模板对象 */ function buildTemplate(config) { return { template_id: config.template_id, template_name: config.template_name || '', template_brief: config.template_brief || '', consistency_mode: config.consistency_mode || 'free', report_schema: config.report_schema || {}, report_examples: config.report_examples || {}, field_requirements: config.field_requirements || {}, coordinator_config: config.coordinator_config || {}, agent_adaptation_policy: config.agent_adaptation_policy || {}, }; } /** * 构建发送给后端的完整 session 请求体 */ function buildSessionPayload(config) { const reportSchema = config.report_schema || {}; const reviewMode = config.coordinator_mode || config.review_mode || 'auto'; return { task_context: config.task_context || '', template: buildTemplate(config), agents: normalizeRuntimeAgents(config.agents || [], reportSchema), llm_preset_id: config.llm_preset_id || '', phase_config: config.phase_config || { diverge_rounds: 3 }, review_mode: reviewMode, retrieval: config.retrieval || config.retrieve, }; } /** * 将后端 proposal 数据转换为前端 CoordinatorCard 所需格式 * * @param {Object} proposal - 后端提案对象 * @param {Object} previousDraft - 上一轮的草稿(用于合并增量更新) * @returns {Object} 前端协调者数据 */ function proposalToCoordData(proposal, previousDraft = {}) { const draftUpdates = proposal?.draft_updates || {}; return { round: proposal?.round_index, divergeRound: proposal?.diverge_round, totalDivergeRounds: proposal?.total_diverge_rounds, nextSpeakers: proposal?.next_speakers || [], focus: proposal?.focus || {}, draft: { ...previousDraft, ...draftUpdates }, draftUpdates, gaps: proposal?.gaps || [], fieldAssessment: proposal?.field_assessment || {}, sharedMemory: proposal?.shared_memory || null, reasoning: proposal?.reasoning || '', proposal, }; } /** * 解析 SSE (Server-Sent Events) 数据行 * * 格式: "data: {...}" 或 "data: [DONE]" * @param {string} line - SSE 行文本 * @returns {Object|null} 解析后的 JSON 对象,无效行返回 null */ function parseSsePayload(line) { const trimmed = line.trim(); if (!trimmed.startsWith('data:')) return null; const jsonStr = trimmed.slice(5).trim(); if (!jsonStr || jsonStr === '[DONE]') return null; return JSON.parse(jsonStr); } /** * useSession hook * * 管理整个分析会话的生命周期: * - startSession: 发送 POST /sessions/start 创建会话,然后连接 SSE 事件流 * - stopSession: 中断当前连接 * - reset: 重置所有状态 * - approveCoordinator / reviseCoordinator: 人工确认 / 修改提案 * * SSE 事件类型及处理: * - session_started: 初始化阶段、智能体、轮次 * - agent_started: 在 timeline 中插入 agent_contribution(streaming=true) * - agent_token: 追加 token 到对应 agent_contribution 的 content/thinking * - agent_done: 将对应 agent_contribution 的 streaming 标记为 false * - coordinator_started: 插入 coordinator_streaming 卡片 * - coordinator_token: 追加 coordinator 的思考过程或内容 token * - coordinator_status: 更新状态横幅 * - coordinator_proposal: 替换 coordinator_streaming 为 coordinator 完成卡片 * - human_review_required: 切换状态为 waiting_human * - final_report_token: 追加最终报告 token * - final_report_done: 插入 final_report 卡片,状态切换为 done * - error: 追加错误到 errors 数组 * * @returns {Object} 会话状态 + 操作方法 */ export function useSession() { const [state, setState] = useState(initialState); const abortControllerRef = useRef(null); const timelineRef = useRef([]); // 用 ref 保存 timeline,避免闭包过期 const configRef = useRef(null); // 保存启动时的配置引用 /** 重置所有状态 */ const reset = useCallback(() => { if (abortControllerRef.current) { abortControllerRef.current.abort(); abortControllerRef.current = null; } timelineRef.current = []; configRef.current = null; setState(initialState); }, []); /** * 从历史记录 JSON 恢复会话展示状态。 * * 后端历史接口未来会直接返回保存时的 snapshot;这里把 snapshot 写回 state 和 timelineRef, * 让用户进入页面后无需重新生成,也能直接看到 Timeline、左右侧进度和最终报告。 */ const hydrateFromSnapshot = useCallback((snapshot = {}) => { const nextTimeline = Array.isArray(snapshot.timeline) ? snapshot.timeline : []; timelineRef.current = nextTimeline; configRef.current = snapshot.config || null; setState({ ...initialState, ...snapshot, status: snapshot.status || (nextTimeline.length > 0 ? 'done' : 'idle'), phase: snapshot.phase || (nextTimeline.length > 0 ? 'done' : null), timeline: nextTimeline, errors: Array.isArray(snapshot.errors) ? snapshot.errors : [], agentStates: snapshot.agentStates || {}, agents: Array.isArray(snapshot.agents) ? snapshot.agents : [], reportSchema: snapshot.reportSchema || snapshot.config?.report_schema || {}, finalReportPartial: snapshot.finalReportPartial || snapshot.finalReport?.report || {}, pendingReview: null, reviewSubmitting: false, reviewRevising: false, }); }, []); /** 停止当前会话(中断 SSE 连接) */ const stopSession = useCallback(() => { if (abortControllerRef.current) { abortControllerRef.current.abort(); abortControllerRef.current = null; } setState(prev => ({ ...prev, status: 'idle' })); }, []); // 设置模板信息用于开启会话 const configStateChange = useCallback((config) => { setState(prev => ({ ...prev, startConfig: config })); },[]) /** * SSE 事件处理器 * * 根据 event 类型分发处理,修改 timelineRef 和 state。 * 使用 timelineRef(而非 state.timeline)进行链表式修改, * 避免在快速 token 到来时因 React 批量更新导致丢帧。 */ const handleEvent = useCallback((parsed, backendUrl) => { const { event, data } = parsed; if (!event) return; switch (event) { // ====== 会话启动 ====== case 'session_started': { const agents = data.agents || []; const agentStates = {}; agents.forEach(agent => { agentStates[agent.id] = { id: agent.id, name: agent.name, dimension: agent.dimension, status: 'idle', rounds: [], }; }); setState(prev => ({ ...prev, sessionId: prev.sessionId, backendUrl, phase: 'starting', agents, agentStates, totalRounds: (configRef.current?.phase_config?.diverge_rounds || 3) + 1, divergeRounds: configRef.current?.phase_config?.diverge_rounds || 3, reportSchema: configRef.current?.report_schema || {}, })); break; } // ====== 智能体开始输出 ====== case 'agent_started': { const agentId = data.agent_id; const round = data.round_index || 0; const phase = data.phase || 'starting'; // 检查是否已有该轮次的 round_start,没有则插入 const hasRoundStart = timelineRef.current.some( item => item.type === 'round_start' && item.round === round && item.phase === phase ); if (!hasRoundStart) { timelineRef.current = [ ...timelineRef.current, { type: 'round_start', round, phase, agentIds: [agentId] }, ]; } else { timelineRef.current = timelineRef.current.map(item => item.type === 'round_start' && item.round === round && item.phase === phase ? { ...item, agentIds: [...new Set([...(item.agentIds || []), agentId])] } : item ); } // 插入流式 agent_contribution timelineRef.current = [ ...timelineRef.current, { type: 'agent_contribution', agentId, round, phase, streaming: true, data: { round, phase, content: '', thinking: '', retrievalLog: [], retrievalCount: 0, referenceMap: {}, }, }, ]; setState(prev => { const updatedAgentStates = { ...prev.agentStates }; updatedAgentStates[agentId] = { ...(updatedAgentStates[agentId] || { id: agentId, name: data.agent_name, dimension: agentId, rounds: [] }), status: 'thinking', }; return { ...prev, phase, currentRound: Math.max(prev.currentRound, round), agentStates: updatedAgentStates, timeline: timelineRef.current, }; }); break; } // ====== 智能体完成输出 ====== case 'agent_done': { const agentId = data.agent_id; const round = data.round_index || 0; const phase = data.phase || 'starting'; const retrievalLog = data.retrieval_log || []; const roundData = { round, phase, content: data.content || '', thinking: data.thinking || '', retrievalLog, retrievalCount: retrievalLog.length, referenceMap: data.reference_map || {}, metadata: data.metadata || {}, }; // 将对应的流式条目转为完成状态 timelineRef.current = timelineRef.current.map(item => item.type === 'agent_contribution' && item.agentId === agentId && item.round === round && item.streaming ? { ...item, streaming: false, data: roundData } : item ); setState(prev => { const updatedAgentStates = { ...prev.agentStates }; const existing = updatedAgentStates[agentId] || { id: agentId, name: data.agent_name, dimension: agentId, status: 'idle', rounds: [], }; updatedAgentStates[agentId] = { ...existing, name: data.agent_name, status: 'done', rounds: [...(existing.rounds || []), roundData], }; return { ...prev, agentStates: updatedAgentStates, timeline: timelineRef.current }; }); break; } // ====== 智能体 token(流式追加) ====== case 'agent_token': { const agentId = data.agent_id; const round = data.round_index || 0; const token = data.token || ''; const tokenType = data.token_type || 'content'; // 找到对应的 streaming agent_contribution 并追加 token timelineRef.current = timelineRef.current.map(item => { if (item.type !== 'agent_contribution' || item.agentId !== agentId || item.round !== round || !item.streaming) { return item; } if (tokenType === 'thinking') { return { ...item, data: { ...item.data, thinking: (item.data.thinking || '') + token } }; } return { ...item, data: { ...item.data, content: (item.data.content || '') + token } }; }); setState(prev => ({ ...prev, timeline: timelineRef.current })); break; } // ====== 协调员开始流式输出 ====== case 'coordinator_started': { timelineRef.current = [ ...timelineRef.current, { type: 'coordinator_streaming', phase: data.phase || 'diverging', divergeRound: data.diverge_round, totalDivergeRounds: data.total_diverge_rounds, thinking: '', content: '协调者正在评估字段完成度、更新草稿并选择下一轮发言者。', statusMessage: '协调草案生成中…', currentField: null, reportPartial: {}, }, ]; setState(prev => ({ ...prev, phase: data.phase || prev.phase, timeline: timelineRef.current })); break; } // ====== 协调员 token(流式追加) ====== case 'coordinator_token': { const token = data.token || ''; const tokenType = data.token_type || 'thinking'; const phase = data.phase || 'diverging'; // 找到最后一个 coordinator_streaming 项并追加 const lastIdx = timelineRef.current.reduce((last, item, idx) => ( item.type === 'coordinator_streaming' ? idx : last ), -1); if (lastIdx === -1) { // 尚未创建,新建一个 timelineRef.current = [ ...timelineRef.current, { type: 'coordinator_streaming', phase, divergeRound: data.diverge_round, totalDivergeRounds: data.total_diverge_rounds, thinking: tokenType === 'thinking' ? token : '', content: tokenType === 'content' ? token : '', statusMessage: '', currentField: data.field || null, reportPartial: {}, }, ]; } else { timelineRef.current = timelineRef.current.map((item, idx) => { if (idx !== lastIdx || item.type !== 'coordinator_streaming') return item; return { ...item, currentField: data.field || item.currentField, thinking: tokenType === 'thinking' ? (item.thinking || '') + token : item.thinking, content: tokenType === 'content' ? (item.content || '') + token : item.content, statusMessage: '', }; }); } setState(prev => ({ ...prev, timeline: timelineRef.current })); break; } // ====== 协调员状态更新 ====== case 'coordinator_status': { const message = data.message || ''; const phase = data.phase || 'diverging'; const lastIdx = timelineRef.current.reduce((last, item, idx) => ( item.type === 'coordinator_streaming' ? idx : last ), -1); if (lastIdx === -1) { timelineRef.current = [ ...timelineRef.current, { type: 'coordinator_streaming', phase, divergeRound: data.diverge_round, totalDivergeRounds: data.total_diverge_rounds, thinking: '', content: '', statusMessage: message, currentField: null, reportPartial: {}, }, ]; } else { timelineRef.current = timelineRef.current.map((item, idx) => { if (idx !== lastIdx || item.type !== 'coordinator_streaming') return item; return { ...item, statusMessage: message }; }); } setState(prev => ({ ...prev, timeline: timelineRef.current })); break; } // ====== 协调员提案完成 ====== case 'coordinator_proposal': { setState(prev => { const coordData = proposalToCoordData(data.proposal, prev.latestDraft); let replaced = false; // 将最后一个 coordinator_streaming 替换为 coordinator 完成卡片 timelineRef.current = timelineRef.current.map((item, idx) => { if (item.type !== 'coordinator_streaming' || replaced) return item; const lastIdx = timelineRef.current.reduce((last, it, i) => it.type === 'coordinator_streaming' ? i : last, -1); if (idx !== lastIdx) return item; replaced = true; return { type: 'coordinator', data: coordData, thinking: coordData.reasoning, }; }); if (!replaced) { timelineRef.current = [...timelineRef.current, { type: 'coordinator', data: coordData, thinking: coordData.reasoning }]; } return { ...prev, status: data.revised ? 'running' : prev.status, coordinatorUpdates: [...prev.coordinatorUpdates, coordData], latestDraft: coordData.draft, sharedMemory: coordData.sharedMemory, pendingReview: data.revised ? null : prev.pendingReview, reviewRevising: false, timeline: timelineRef.current, }; }); break; } // ====== 需要人工确认(仅 human_review 模式) ====== case 'human_review_required': { setState(prev => { const coordData = proposalToCoordData(data.proposal, prev.latestDraft); return { ...prev, status: 'waiting_human', pendingReview: { ...coordData, proposal: data.proposal, }, sharedMemory: coordData.sharedMemory, reviewSubmitting: false, reviewRevising: false, }; }); break; } // ====== 最终报告生成完成 ====== case 'final_report_done': { const output = { report: data.report || {}, dimensions: data.dimensions || {}, meta: data.metadata || {}, }; timelineRef.current = [...timelineRef.current, { type: 'final_report', data: output }]; setState(prev => ({ ...prev, status: 'done', finalReport: output, finalReportPartial: output.report || {}, currentFinalField: null, phase: 'done', pendingReview: null, reviewSubmitting: false, reviewRevising: false, timeline: timelineRef.current, })); break; } // ====== 最终报告 token(流式追加) ====== case 'final_report_token': { const token = data.token || ''; const field = data.field || null; const reportPartial = data.report_partial || {}; const lastIdx = timelineRef.current.reduce((last, item, idx) => ( item.type === 'coordinator_streaming' ? idx : last ), -1); if (lastIdx === -1) { timelineRef.current = [ ...timelineRef.current, { type: 'coordinator_streaming', phase: 'final_report', thinking: '', content: token, statusMessage: '', currentField: field, reportPartial, }, ]; } else { timelineRef.current = timelineRef.current.map((item, idx) => { if (idx !== lastIdx || item.type !== 'coordinator_streaming') return item; return { ...item, phase: 'final_report', content: (item.content || '') + token, currentField: field || item.currentField, reportPartial, }; }); } setState(prev => ({ ...prev, finalReportPartial: reportPartial, currentFinalField: field, timeline: timelineRef.current, })); break; } // ====== 错误 ====== case 'error': { setState(prev => ({ ...prev, status: prev.status === 'waiting_human' ? prev.status : 'error', errors: [...prev.errors, { error: data.message || data.error || JSON.stringify(data), ...data }], })); break; } default: break; } }, []); /** * 读取 SSE 事件流 * * 使用 fetch + ReadableStream reader 逐行解析, * 支持 AbortController 中断。 * * @param {string} url - SSE 端点 URL * @param {string} backendUrl - 后端基础地址 * @param {AbortController} abortController - 中断控制器 */ const readEventStream = useCallback(async (url, backendUrl, abortController) => { const response = await fetch(url, { signal: abortController.signal }); if (!response.ok) { const text = await response.text(); throw new Error(`HTTP ${response.status}: ${text}`); } const reader = response.body.getReader(); const decoder = new TextDecoder(); let buffer = ''; while (true) { const { done, value } = await reader.read(); if (done || abortController.signal.aborted) { break; } buffer += decoder.decode(value, { stream: true }); const lines = buffer.split('\n'); buffer = lines.pop() || ''; for (const line of lines) { try { const parsed = parseSsePayload(line); if (parsed) handleEvent(parsed, backendUrl); } catch (error) { console.warn('Failed to parse SSE line:', line, error); } } } }, [handleEvent]); /** * 启动分析会话 * * 1. 发送 POST /sessions/start 创建会话 * 2. 获取 session_id * 3. 通过 GET /sessions/{id}/events 订阅 SSE 事件流 * * @param {Object} config - 页面配置(任务背景、模板、智能体等) * @param {string} backendUrl - 后端 API 地址 */ const startSession = useCallback(async (config, backendUrl) => { timelineRef.current = []; configRef.current = config; setState({ ...initialState, status: 'running', backendUrl, coordinatorMode: config.coordinator_mode || 'auto', reportSchema: config.report_schema || {}, }); if (abortControllerRef.current) { abortControllerRef.current.abort(); } const abortController = new AbortController(); abortControllerRef.current = abortController; try { const payload = buildSessionPayload(config); const response = await fetch(`${backendUrl}/sessions/start`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(payload), signal: abortController.signal, }); if (!response.ok) { const text = await response.text(); throw new Error(`HTTP ${response.status}: ${text}`); } const result = await response.json(); const sessionId = result.session_id; setState(prev => ({ ...prev, sessionId, status: result.status === 'created' ? 'running' : result.status, })); await readEventStream(`${backendUrl}/sessions/${sessionId}/events`, backendUrl, abortController); // SSE 流结束后,设置最终状态(除非已是 error/waiting_human/done) setState(prev => ({ ...prev, status: prev.status === 'error' || prev.status === 'waiting_human' || prev.status === 'done' ? prev.status : 'done', })); } catch (err) { if (err.name === 'AbortError') return; console.error('Session error:', err); setState(prev => ({ ...prev, status: 'error', errors: [...prev.errors, { error: err.message }], })); } }, [readEventStream]); /** * 人工确认协调者提案 * * 在 human_review 模式下,用户修改提案内容后提交确认。 * 发送 POST /sessions/{id}/coordinator/approve。 * * @param {Object} payload - { nextSpeakers, focus, draft, gaps } */ const approveCoordinator = useCallback(async (payload) => { if (!state.sessionId || !state.backendUrl) { throw new Error('session 未初始化'); } const baseProposal = state.pendingReview?.proposal || {}; const proposal = { ...baseProposal, next_speakers: payload.nextSpeakers, focus: payload.focus, draft_updates: payload.draft, gaps: payload.gaps, }; setState(prev => ({ ...prev, reviewSubmitting: true })); try { const response = await fetch(`${state.backendUrl}/sessions/${state.sessionId}/coordinator/approve`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ proposal }), }); if (!response.ok) { const msg = await response.text(); throw new Error(msg || `HTTP ${response.status}`); } const result = await response.json(); setState(prev => ({ ...prev, status: 'running', pendingReview: null, reviewSubmitting: false, })); } catch (err) { console.error('approve error:', err); setState(prev => ({ ...prev, reviewSubmitting: false, errors: [...prev.errors, { error: err.message }], })); } }, [state.sessionId, state.backendUrl, state.pendingReview]); /** * 发送修改指令,让协调员重新生成提案 * * @param {string} instruction - 用户的修改指令(自然语言) */ const reviseCoordinator = useCallback(async (instruction) => { if (!state.sessionId || !state.backendUrl) { throw new Error('session 未初始化'); } setState(prev => ({ ...prev, reviewRevising: true })); try { const response = await fetch(`${state.backendUrl}/sessions/${state.sessionId}/coordinator/revise`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ instruction }), }); if (!response.ok) { const msg = await response.text(); throw new Error(msg || `HTTP ${response.status}`); } // 后端会在修订完成后通过 SSE 推送新的 coordinator_proposal setState(prev => ({ ...prev, reviewRevising: false, pendingReview: null, status: 'running', })); } catch (err) { console.error('revise error:', err); setState(prev => ({ ...prev, reviewRevising: false, errors: [...prev.errors, { error: err.message }], })); } }, [state.sessionId, state.backendUrl]); return { ...state, hydrateFromSnapshot, reset, startSession, stopSession, approveCoordinator, reviseCoordinator, configStateChange, }; }