Go + Wails 3 打造极致流畅、高可靠的 AI 流式对话系统设计方案

┌──────────────────────────────────────────────────────────────┐
│  大模型/后台服务 (LLM Provider / Backend Engine)               │
│    高频、微秒级 Token 流                                      │
└──────────────────────┬───────────────────────────────────────┘
                       │
                       ▼
┌──────────────────────────────────────────────────────────────┐
│  Go Backend (Wails 服务端)                                    │
│                                                              │
│  ┌─────────────────────────────────────────────────────────┐ │
│  │ Attempt Caching (以 StreamID 为联合键的真理缓存)          │ │
│  │   · 为每次生成请求（包含重新生成）创建唯一的 StreamID (UUID)  │ │
│  │   · 全量 assembled 数据均以此 ID 隔离缓存，防止覆盖竞态      │ │
│  └───────────────┬─────────────────────────────────────────┘ │
│                  │                                            │
│  ┌───────────────▼─────────────────────────────────────────┐ │
│  │ StreamDispatcher (定时合批平滑分发器)                    │ │
│  │   · 30ms 定时 Ticker + 50 帧容量阈值限制                   │ │
│  │   · 将高频 Token 帧打包成数组形式发送，极大缓解 IPC 压力      │ │
│  └───────────────┬─────────────────────────────────────────┘ │
└──────────────────┼──────────────────────────────────────────┘
                   │ Wails IPC (Events.Emit)
                   ▼
┌──────────────────────────────────────────────────────────────┐
│  Frontend Client (Wails 网页前端)                              │
│                                                               │
│  ┌─────────────────────────────────────────────────────────┐ │
│  │ Reassembly Buffer (滑动窗口重组缓冲区)                    │ │
│  │   · 根据 chunkIndex 组装，自愈因异步并发带来的时序空洞   │ │
│  └───────────────┬─────────────────────────────────────────┘ │
│                  │                                            │
│  ┌───────────────▼─────────────────────────────────────────┐ │
│  │ requestAnimationFrame Scheduler (16ms 增量批量更新)       │ │
│  │   · 一帧内最多触发一次 setState，配合屏幕刷新率防抖        │ │
│  └───────────────┬─────────────────────────────────────────┘ │
│                  │                                            │
│  ┌───────────────▼─────────────────────────────────────────┐ │
│  │ Post-Stream Sync (后置对齐) & Active Patching (补丁通道)   │ │
│  │   · 流结束：同步清空缓冲区 -> 异步拉取 GetFullMessage()      │ │
│  │   · 补丁：监听 'chat-patch' 通道，跨周期就地更新指定消息   │ │
│  └─────────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘

// 消息帧类型枚举定义
enum ChunkType {
    CHUNK_TEXT      = 0;  // 聊天对话正文
    CHUNK_THINKING  = 1;  // 模型深度思考过程
    CHUNK_TOOL_CALL = 2;  // 工具调用明细（包含状态、入参、出参）
    CHUNK_NODE_HOP  = 3;  // 多 Agent 协作跳转拓扑图节点
    CHUNK_MEDIA     = 4;  // 多媒体流附加卡片
}

type StreamPayload struct {
	StreamID   string            `json:"streamId"`
	ChunkIndex int32             `json:"chunkIndex"`
	Type       int32             `json:"type"` // 对应 ChunkType 枚举
	Content    string            `json:"content"`
	// 可选字段，当 Type 对应 CHUNK_TOOL_CALL 或 CHUNK_MEDIA 时，携带强类型结构体
	ToolCall   *ToolCallDetail   `json:"toolCall,omitempty"`
	Media      *MediaDetail      `json:"media,omitempty"`
}

const applyPayloadToMessage = (msg: ChatMessage, pl: StreamPayload) => {
    switch (pl.type) {
        case 0: // CHUNK_TEXT
            msg.content += pl.chunk;
            break;
        case 1: // CHUNK_THINKING
            msg.thinking = (msg.thinking || '') + pl.chunk;
            break;
        case 2: // CHUNK_TOOL_CALL
            msg.toolCalls = mergeToolCalls(msg.toolCalls || [], pl.toolCall);
            break;
        case 3: // CHUNK_NODE_HOP
            msg.nodeHops = appendNodeHop(msg.nodeHops || [], pl.nodeHop);
            break;
        case 4: // CHUNK_MEDIA
            msg.media = appendMedia(msg.media || [], pl.media);
            break;
    }
};

type StreamDispatcher struct {
	buffer   []StreamPayload
	mu       sync.Mutex
	interval time.Duration
	maxBatch int
}

func (d *StreamDispatcher) Enqueue(payload StreamPayload) {
	d.mu.Lock()
	d.buffer = append(d.buffer, payload)
	needFlush := len(d.buffer) >= d.maxBatch
	d.mu.Unlock()

	if needFlush {
		d.flush()
	}
}

// flush 方法负责在 Ticker 到期或 buffer 满时，将平滑数组打包通过 Wails 广播出去

const chunkBufferRef = useRef<Record<number, StreamPayload>>({});
const nextChunkIndexRef = useRef<number>(1);
const pendingPayloadsRef = useRef<StreamPayload[]>([]);

const onChunkReceived = (payload: StreamPayload) => {
    const idx = payload.chunkIndex;
    if (idx > 0) {
        chunkBufferRef.current[idx] = payload;
    }

    // 提取无空洞的连续帧
    const readyPayloads: StreamPayload[] = [];
    while (chunkBufferRef.current[nextChunkIndexRef.current]) {
        const nextIdx = nextChunkIndexRef.current;
        readyPayloads.push(chunkBufferRef.current[nextIdx]);
        delete chunkBufferRef.current[nextIdx];
        nextChunkIndexRef.current++;
    }

    if (readyPayloads.length > 0) {
        pendingPayloadsRef.current.push(...readyPayloads);
        triggerScheduler();
    }
};

const rafPendingRef = useRef<boolean>(false);

const triggerScheduler = () => {
    if (rafPendingRef.current) return;
    rafPendingRef.current = true;

    requestAnimationFrame(() => {
        rafPendingRef.current = false;
        const toRender = pendingPayloadsRef.current;
        pendingPayloadsRef.current = [];

        if (toRender.length === 0) return;

        setMessages((prevMsgs) => {
            // 一次性应用 toRender 里的所有增量数据，并在一个 VDOM Tick 内渲染完毕
            return applyPayloadsToState(prevMsgs, toRender);
        });
    });
};

export const ChatMessageItem = React.memo<ChatMessageItemProps>(({
    msg,
    index,
    // ...
}) => {
    // 组件渲染逻辑
}, (prevProps, nextProps) => {
    return (
        prevProps.index === nextProps.index &&
        prevProps.debugMode === nextProps.debugMode &&
        prevProps.copiedIndex === nextProps.copiedIndex &&
        prevProps.msg.content === nextProps.msg.content &&
        prevProps.msg.thinking === nextProps.msg.thinking &&
        prevProps.msg._synced === nextProps.msg._synced
        // ... 其他关键渲染字段的比对
    );
});

type ChatMessagePatch = {
    sessionId: string;
    messageId: number;   // 定位具体的消息轮次
    field: 'content' | 'toolCalls' | 'media';
    action: 'set' | 'append' | 'merge';
    value: any;
};

func TestChatCorePostStreamSync(t *testing.T) {
	// 1. 初始化 Mock 生成流，设置预期的 streamID
	streamID := "test-sync-stream-uuid"

	// 2. 模拟 LLM 异步生成并在后台向本地缓存塞入已组装好的真理消息
	expectedMessage := &AssembledMessage{
		Content:  "Hello, this is a streaming test reply.",
		Thinking: "Reasoning: checking pipeline.",
		MaxSeq:   5,
	}
	svc.lastMessages.Store(streamID, expectedMessage)

	// 3. 模拟前端调用 GetFullMessage 接口进行后置数据同步
	assembled, err := svc.GetFullMessage(streamID)
	
	// 4. 断言真理源内容正确无误，保障隔离性
	assert.NoError(t, err)
	assert.Equal(t, expectedMessage.Content, assembled.Content)
	assert.Equal(t, expectedMessage.Thinking, assembled.Thinking)
}

describe('Chat Reassembly Buffer', () => {
    it('should reassemble out-of-order chunks in correct sequence', () => {
        const buffer: Record<number, any> = {};
        let nextExpectedIndex = 1;
        const readyPayloads: any[] = [];

        // 模拟重组解析器
        const receiveChunk = (idx: number, payload: any) => {
            buffer[idx] = payload;
            while (buffer[nextExpectedIndex]) {
                readyPayloads.push(buffer[nextExpectedIndex]);
                delete buffer[nextExpectedIndex];
                nextExpectedIndex++;
            }
        };

        // 1. 模拟 2 号 Chunk 先于 1 号 Chunk 到达（典型乱序）
        receiveChunk(2, { chunkIndex: 2, content: " World" });
        expect(readyPayloads.length).toBe(0); // 1 号未到，重组挂起

        // 2. 1 号 Chunk 延迟到达
        receiveChunk(1, { chunkIndex: 1, content: "Hello" });
        expect(readyPayloads.length).toBe(2); // 1 号到达后，依次重组吐出
        expect(readyPayloads[0].content).toBe("Hello");
        expect(readyPayloads[1].content).toBe(" World");
    });
});