feat(stream): 添加子代理消息流增量处理功能
- 在 runtime 模块中新增 handle_stream_delta 方法，支持子代理消息流的增量和结束事件处理 - 为流消息新增 stream_message_id，用于标识和管理消息流状态 - 修改 ConversationContext 初始化，加入 stream_message_id 互斥锁字段 - 更新 web 端 Chat 组件，添加对流增量内容的条件渲染，避免空字符串渲染 - 在 useChat 钩子内增加对 stream_delta 和 stream_end 类型消息的识别和处理逻辑 - 实现流增量消息的累积更新，合并多次流增量内容和推理文本 - 处理流结束消息时的无操作逻辑，确保消息流完整性 - 对 assistant_response 消息进行替换更新，修正流消息的最终呈现内容
This commit is contained in:
parent
fc7df67474
commit
b67848180b
@ -11,6 +11,7 @@ use crate::agent::{AgentLoop, AgentRuntimeConfig, EmittedMessageHandler, Persist
|
||||
use crate::bus::ChatMessage;
|
||||
use crate::bus::message::{OutboundMessage, OutboundEventKind};
|
||||
use crate::bus::MessageBus;
|
||||
use crate::providers::StreamDelta;
|
||||
use crate::config::{LLMProviderConfig, SubagentsConfig};
|
||||
use crate::storage::{ConversationRepository, SessionStore};
|
||||
use crate::tools::{ToolContext, ToolRegistry};
|
||||
@ -107,6 +108,7 @@ struct SubAgentEmitter {
|
||||
metadata: HashMap<String, String>,
|
||||
store: Arc<SessionStore>,
|
||||
sub_session_id: String,
|
||||
stream_message_id: std::sync::Mutex<Option<String>>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
@ -159,6 +161,41 @@ impl EmittedMessageHandler for SubAgentEmitter {
|
||||
self.persist_todo_write_result(&message);
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_stream_delta(&self, delta: &StreamDelta) {
|
||||
let message_id = {
|
||||
let mut guard = self.stream_message_id.lock().unwrap();
|
||||
guard.get_or_insert_with(|| uuid::Uuid::new_v4().to_string()).clone()
|
||||
};
|
||||
|
||||
let outbound = if delta.content.is_empty() && delta.reasoning_content.is_none() {
|
||||
OutboundMessage::stream_end(
|
||||
&self.channel_name,
|
||||
&self.chat_id,
|
||||
None,
|
||||
&message_id,
|
||||
self.metadata.clone(),
|
||||
)
|
||||
} else {
|
||||
OutboundMessage::stream_delta(
|
||||
&self.channel_name,
|
||||
&self.chat_id,
|
||||
None,
|
||||
&message_id,
|
||||
&delta.content,
|
||||
delta.reasoning_content.clone(),
|
||||
self.metadata.clone(),
|
||||
)
|
||||
};
|
||||
|
||||
if let Err(error) = self.bus.publish_outbound(outbound).await {
|
||||
tracing::error!(error = %error, channel = %self.channel_name, "Failed to publish sub-agent stream delta");
|
||||
}
|
||||
}
|
||||
|
||||
async fn set_stream_message_id(&self, id: &str) {
|
||||
*self.stream_message_id.lock().unwrap() = Some(id.to_string());
|
||||
}
|
||||
}
|
||||
|
||||
impl SubAgentEmitter {
|
||||
@ -316,6 +353,7 @@ impl DefaultSubAgentRuntime {
|
||||
metadata,
|
||||
store: self.store.clone(),
|
||||
sub_session_id: session.session_id.clone(),
|
||||
stream_message_id: std::sync::Mutex::new(None),
|
||||
},
|
||||
self.conversation_repository.clone(),
|
||||
session.session_id.clone(),
|
||||
|
||||
@ -684,6 +684,7 @@ export function MessageBubble({ message, onNavigateToSubAgent, showThinking = tr
|
||||
<ThinkingSection content={message.reasoningContent} />
|
||||
)}
|
||||
{/* AI 和工具消息使用 Markdown 渲染 */}
|
||||
{message.content.trim() && (
|
||||
<div className="markdown-content text-sm leading-relaxed">
|
||||
<ReactMarkdown
|
||||
remarkPlugins={[remarkGfm]}
|
||||
@ -763,6 +764,7 @@ export function MessageBubble({ message, onNavigateToSubAgent, showThinking = tr
|
||||
{message.content}
|
||||
</ReactMarkdown>
|
||||
</div>
|
||||
)}
|
||||
</>)}
|
||||
{message.attachments && message.attachments.length > 0 && (
|
||||
<div className="mt-2 space-y-2">
|
||||
|
||||
@ -28,6 +28,7 @@ import type {
|
||||
Channel,
|
||||
ChannelList,
|
||||
StreamDelta,
|
||||
StreamEnd,
|
||||
} from '../types/protocol'
|
||||
|
||||
// 简化后的层级状态
|
||||
@ -187,6 +188,9 @@ export function useChat(): UseChatReturn {
|
||||
|| message.type === 'tool_pending' || message.type === 'assistant_response') {
|
||||
return (message as ToolCall | ToolResult | ToolPending | AssistantResponse).subagent_task_id
|
||||
}
|
||||
if (message.type === 'stream_delta' || message.type === 'stream_end') {
|
||||
return (message as StreamDelta | StreamEnd).subagent_task_id
|
||||
}
|
||||
return undefined
|
||||
}
|
||||
|
||||
@ -258,6 +262,18 @@ export function useChat(): UseChatReturn {
|
||||
subagentTaskId: msg.subagent_task_id,
|
||||
}
|
||||
}
|
||||
case 'stream_delta': {
|
||||
const msg = message as StreamDelta
|
||||
return {
|
||||
id: msg.id,
|
||||
role: 'assistant' as const,
|
||||
content: msg.delta,
|
||||
timestamp: Math.floor(Date.now() / 1000),
|
||||
type: 'message' as const,
|
||||
subagentTaskId: msg.subagent_task_id,
|
||||
reasoningContent: msg.reasoning_delta,
|
||||
}
|
||||
}
|
||||
case 'error': {
|
||||
return {
|
||||
id: generateMessageId(),
|
||||
@ -272,15 +288,48 @@ export function useChat(): UseChatReturn {
|
||||
}
|
||||
}
|
||||
|
||||
// Append a server message to the sub-agent view
|
||||
// Append a server message to the sub-agent view (with streaming delta accumulation)
|
||||
const appendToSubAgentViewMessage = (message: WsOutbound) => {
|
||||
// stream_delta: accumulate into existing message by ID, or create new
|
||||
if (message.type === 'stream_delta') {
|
||||
const msg = message as StreamDelta
|
||||
setSubAgentView((prev) => {
|
||||
if (!prev) return prev
|
||||
const existingIdx = prev.messages.findIndex(m => m.id === msg.id && m.type === 'message')
|
||||
if (existingIdx >= 0) {
|
||||
const updated = [...prev.messages]
|
||||
const existing = updated[existingIdx]
|
||||
updated[existingIdx] = {
|
||||
...existing,
|
||||
content: existing.content + msg.delta,
|
||||
reasoningContent: msg.reasoning_delta
|
||||
? (existing.reasoningContent || '') + msg.reasoning_delta
|
||||
: existing.reasoningContent,
|
||||
}
|
||||
return { ...prev, messages: updated }
|
||||
}
|
||||
const chatMsg = serverMessageToChatMessage(message)
|
||||
return chatMsg ? { ...prev, messages: [...prev.messages, chatMsg] } : prev
|
||||
})
|
||||
return
|
||||
}
|
||||
// stream_end: no-op, assistant_response will replace
|
||||
if (message.type === 'stream_end') return
|
||||
// Other messages: assistant_response replaces streamed message by ID
|
||||
const chatMsg = serverMessageToChatMessage(message)
|
||||
if (chatMsg) {
|
||||
setSubAgentView((prev) =>
|
||||
prev
|
||||
? { ...prev, messages: [...prev.messages, chatMsg] }
|
||||
: prev
|
||||
)
|
||||
setSubAgentView((prev) => {
|
||||
if (!prev) return prev
|
||||
if (message.type === 'assistant_response') {
|
||||
const existingIdx = prev.messages.findIndex(m => m.id === chatMsg.id && m.type === 'message')
|
||||
if (existingIdx >= 0) {
|
||||
const updated = [...prev.messages]
|
||||
updated[existingIdx] = chatMsg
|
||||
return { ...prev, messages: updated }
|
||||
}
|
||||
}
|
||||
return { ...prev, messages: [...prev.messages, chatMsg] }
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user