購物網(wǎng)站建設(shè)項(xiàng)目可研報(bào)告,軟文推廣廣告,網(wǎng)絡(luò)機(jī)柜定制,學(xué)網(wǎng)站維護(hù)使用 Go 實(shí)現(xiàn) SSE 流式推送 打字機(jī)效果#xff08;模擬 Coze Chat#xff09; 在開發(fā)實(shí)時(shí)聊天、AI 助手或者協(xié)作應(yīng)用時(shí)#xff0c;我們經(jīng)常需要 SSE#xff08;Server-Sent Events#xff09; 實(shí)現(xiàn)服務(wù)端向前端持續(xù)推送數(shù)據(jù)。本文將分享一個(gè) Go SSE 打字機(jī)式輸出實(shí)現(xiàn) 打字機(jī)效果模擬 Coze Chat在開發(fā)實(shí)時(shí)聊天、AI 助手或者協(xié)作應(yīng)用時(shí)我們經(jīng)常需要SSEServer-Sent Events實(shí)現(xiàn)服務(wù)端向前端持續(xù)推送數(shù)據(jù)。本文將分享一個(gè)Go SSE 打字機(jī)式輸出實(shí)現(xiàn)并附上上游模擬示例、curl 測試和前端實(shí)時(shí)渲染示例。功能特點(diǎn)使用SSE推送消息流前端無需輪詢。對消息進(jìn)行逐字符打字機(jī)式輸出模擬 AI 打字效果。支持上游 SSE 模擬方便本地測試?？奢p松擴(kuò)展為真實(shí) AI 聊天接口的代理服務(wù)。技術(shù)棧Go 1.21CloudWeGo Hertz 作為 HTTP 框架resty 用于上游 SSE 請求SSE 流式推送使用hertz-contrib/sse完整示例代碼packagemainimport(bufiocontextencoding/jsonfmtmath/randstringstimegithub.com/cloudwego/hertz/pkg/appgithub.com/cloudwego/hertz/pkg/app/servergithub.com/cloudwego/hertz/pkg/common/hloggithub.com/hertz-contrib/sseresty.dev/v3)funcmain(){h:server.Default(server.WithHostPorts(:8380))h.POST(/v3/chat,CozeParseThenTypeWriter)h.GET(/upstream,MockUpstreamSSE)hlog.Info( Coze SSE parse typewriter proxy running at :8380)h.Spin()}// 核心邏輯解析 conversation.message.delta → 打字機(jī)式輸出funcCozeParseThenTypeWriter(ctx context.Context,c*app.RequestContext){// SSE Headerc.SetStatusCode(200)h:c.Response.Header h.Set(Content-Type,text/event-stream; charsetutf-8)h.Set(Cache-Control,no-cache, no-store, must-revalidate)h.Set(Pragma,no-cache)h.Set(Connection,keep-alive)h.Set(X-Accel-Buffering,no)stream:sse.NewStream(c)// Resty 上游請求client:resty.New().SetTimeout(0)resp,err:client.R().SetContext(ctx).SetDoNotParseResponse(true).SetHeader(Accept,text/event-stream).Get(http://localhost:8380/upstream)iferr!nil||resp.RawResponsenil||resp.RawResponse.Bodynil{hlog.Error(upstream connect failed)return}deferresp.RawResponse.Body.Close()// 打字機(jī)準(zhǔn)備r:rand.New(rand.NewSource(time.Now().UnixNano()))scanner:bufio.NewScanner(resp.RawResponse.Body)varcurrentEventstring// SSE 解析循環(huán)forscanner.Scan(){select{case-ctx.Done():hlog.Warn(client disconnected)returndefault:}line:scanner.Text()ifline{currentEventcontinue}ifstrings.HasPrefix(line,event:){currentEventstrings.TrimSpace(strings.TrimPrefix(line,event:))continue}ifstrings.HasPrefix(line,data:){payload:strings.TrimSpace(strings.TrimPrefix(line,data:))ifpayload[DONE]{stream.Publish(sse.Event{Data:[]byte([DONE])})return}ifcurrentEventconversation.message.delta{vardstruct{Contentstringjson:content}iferr:json.Unmarshal([]byte(payload),d);errnil{typeWriter(stream,r,d.Content)}}ifcurrentEventconversation.message.completed{stream.Publish(sse.Event{Event:conversation.message.completed,Data:[]byte({status:completed}),})}}}}// 打字機(jī)逐字符輸出functypeWriter(stream*sse.Stream,r*rand.Rand,textstring){fori,ch:rangetext{time.Sleep(getSleepDuration(r,ch))data:map[string]any{id:fmt.Sprintf(char_%d_%d,i,time.Now().UnixNano()%100000),role:assistant,type:answer,content:string(ch),created_at:time.Now().UnixMilli(),}b,_:json.Marshal(data)_stream.Publish(sse.Event{Event:conversation.message.delta,ID:fmt.Sprintf(char_%d,i),Data:b,})}}// 延遲策略funcgetSleepDuration(r*rand.Rand,chrune)time.Duration{switch{casech ||ch。||ch||ch:returntime.Duration(300r.Intn(200))*time.Millisecondcasech、||ch ||ch-||ch:||ch,:returntime.Duration(150r.Intn(100))*time.Millisecondcasech#||ch*||ch:returntime.Duration(200r.Intn(150))*time.Milliseconddefault:returntime.Duration(60r.Intn(60))*time.Millisecond}}// 上游 Mock模擬 Coze SSEfuncMockUpstreamSSE(ctx context.Context,c*app.RequestContext){c.SetStatusCode(200)c.Header(Content-Type,text/event-stream)c.Header(Cache-Control,no-cache)c.Header(Connection,keep-alive)c.Header(X-Accel-Buffering,no)c.Flush()send:func(event,datastring){c.Write([]byte(event: event data: data ,))c.Flush()}deltas:[]string{你,好,,這,是, Coze, SSE}for_,ch:rangedeltas{send(conversation.message.delta,{content:ch})time.Sleep(80*time.Millisecond)}send(conversation.message.completed,{})c.Write([]byte(data: [DONE] ))c.Flush()}使用方法啟動(dòng)服務(wù)go run main.go訪問接口上游 SSE 測試瀏覽器可直接訪問http://localhost:8380/upstream打字機(jī)代理接口POSThttp://localhost:8380/v3/chat使用curl測試 SSE# 測試上游 SSEcurl-N http://localhost:8380/upstream# 測試打字機(jī)代理 SSEcurl-N -X POST http://localhost:8380/v3/chat參數(shù)說明-N/--no-buffer禁用輸出緩存實(shí)時(shí)顯示流式數(shù)據(jù)。-X POST因?yàn)榇斫涌谑?POST。運(yùn)行后你會(huì)在終端看到類似打字機(jī)逐字符輸出event: conversation.message.delta data: {id:char_0_12345,role:assistant,type:answer,content:你,created_at:1700000000000} event: conversation.message.delta data: {id:char_1_67890,role:assistant,type:answer,content:好,created_at:1700000000050} ... event: conversation.message.completed data: {status:completed} data: [DONE]前端實(shí)時(shí)渲染打字機(jī)效果示例在前端可以使用EventSource監(jiān)聽 SSE 并動(dòng)態(tài)顯示內(nèi)容dividchat/divscriptconstchatDivdocument.getElementById(chat);constevtSourcenewEventSource(http://localhost:8380/v3/chat);evtSource.addEventListener(conversation.message.delta,e{constdataJSON.parse(e.data);chatDiv.innerHTMLdata.content;});evtSource.addEventListener(conversation.message.completed,e{console.log(消息完成);});evtSource.onopen()console.log(連接已打開);evtSource.onerror()console.log(連接錯(cuò)誤或關(guān)閉);/script效果消息逐字符顯示模擬 AI 打字機(jī)輸出。核心解析SSE Header 設(shè)置h.Set(Content-Type,text/event-stream; charsetutf-8)h.Set(Cache-Control,no-cache, no-store, must-revalidate)h.Set(Connection,keep-alive)h.Set(X-Accel-Buffering,no)保證瀏覽器或代理實(shí)時(shí)接收流式數(shù)據(jù)。打字機(jī)效果根據(jù)字符類型不同設(shè)置不同延遲casech ||ch。||ch||ch:returntime.Duration(300r.Intn(200))*time.Millisecond上游 SSE 模擬方便本地測試無需真實(shí) AI 接口即可驗(yàn)證前端打字機(jī)效果?？偨Y(jié)通過本文示例你可以快速實(shí)現(xiàn)Go SSE 服務(wù)端代理AI 聊天消息打字機(jī)式輸出上游 SSE 模擬curl 測試和前端實(shí)時(shí)渲染