鎮(zhèn)江做網(wǎng)站公司,在線教育培訓(xùn)平臺定制,怎么查詢自己注冊的商標(biāo),做網(wǎng)站寬度和長度布局使用 FastAPI 實(shí)現(xiàn) Coze 流式聊天 SSE 接口 在開發(fā) AI 助手或聊天應(yīng)用時(shí)#xff0c;我們通常希望服務(wù)端能夠 實(shí)時(shí)向前端推送消息#xff0c;讓用戶看到逐字打字效果。本文演示如何使用 FastAPI Coze Python SDK#xff08;cozepy#xff09; 實(shí)現(xiàn) 流式聊天 SSE 接口…使用 FastAPI 實(shí)現(xiàn) Coze 流式聊天 SSE 接口在開發(fā) AI 助手或聊天應(yīng)用時(shí)我們通常希望服務(wù)端能夠?qū)崟r(shí)向前端推送消息讓用戶看到逐字打字效果。本文演示如何使用FastAPI Coze Python SDKcozepy實(shí)現(xiàn)流式聊天 SSE 接口并提供curl測試方法。功能特點(diǎn)流式輸出前端可以實(shí)時(shí)接收聊天增量消息。SSE 格式便于瀏覽器或 Go/Node 前端解析。兼容不同版本 Coze SDK處理可能缺失的異常類?？芍苯邮褂胏url測試無需前端即可驗(yàn)證接口。技術(shù)棧Python 3.10FastAPIuvicornASGI 服務(wù)cozepyCoze 官方 Python SDKSSE 流式推送完整示例代碼importosfromtypingimportOptional,List,Dict,AnyfromfastapiimportFastAPI,HTTPExceptionfromfastapi.responsesimportStreamingResponsefrompydanticimportBaseModelfromcozepyimportCoze,TokenAuth,Message,ChatEventType,COZE_CN_BASE_URL# # 兼容不同版本的cozepy異常類# try:fromcozepyimportCozeAPIError,CozeAuthErrorexceptImportError:classCozeAPIError(Exception):passclassCozeAuthError(Exception):pass# # 初始化FastAPI應(yīng)用# appFastAPI(titleCoze Stream Chat API)# # 全局配置與Coze客戶端初始化# COZE_API_TOKENos.getenv(COZE_API_TOKEN,你的默認(rèn)Token)COZE_API_BASECOZE_CN_BASE_URL BOT_VERSION1756277832coze_client:Optional[Coze]Nonedefinit_coze_client():初始化Coze客戶端globalcoze_clientifcoze_client:returncoze_clienttry:coze_clientCoze(authTokenAuth(tokenCOZE_API_TOKEN),base_urlCOZE_API_BASE)returncoze_clientexceptExceptionase:raiseHTTPException(status_code500,detailfCoze客戶端初始化失敗{str(e)})init_coze_client()# # 定義請求體模型# classChatRequest(BaseModel):user_id:strbot_id:strstream:boolTrueadditional_messages:List[Dict[str,Any]]conversation_id:Optional[str]Nonebot_version:Optional[str]BOT_VERSION# # 流式聊天接口# app.post(/api/coze-chat)asyncdefcoze_chat(request:ChatRequest):try:# 構(gòu)建 Coze 消息importjson messages[]formsginrequest.additional_messages:ifmsg.get(role)userandmsg.get(content_type)text:content_listjson.loads(msg.get(content,[]))text.join([item.get(text,)foritemincontent_list])messages.append(Message.build_user_question_text(text))# 調(diào)用流式接口streamcoze_client.chat.stream(bot_idrequest.bot_id,user_idrequest.user_id,conversation_idrequest.conversation_idorNone,publish_statuspublished_online,bot_versionrequest.bot_version,auto_save_historyFalse,additional_messagesmessages)# SSE 流生成器asyncdefgenerate_stream():try:foreventinstream:ifnotevent:continue# 消息增量ifevent.eventChatEventType.CONVERSATION_MESSAGE_DELTA:contentevent.message.content.strip()ifevent.message.contentelseifcontent:yieldfdata:{json.dumps({type:delta,content:content})} # 聊天完成elifevent.eventChatEventType.CONVERSATION_CHAT_COMPLETED:usageevent.chat.usage.token_countifhasattr(event.chat,usage)else0conv_idevent.chat.conversation_idifhasattr(event.chat,conversation_id)elseyieldfdata:{json.dumps({type:completed,token_count:usage,conversation_id:conv_id})} yielddata: [DONE] exceptExceptionase:yieldfdata:{json.dumps({type:error,message:str(e)})} returnStreamingResponse(generate_stream(),media_typetext/event-stream,headers{Cache-Control:no-cache,Connection:keep-alive,Access-Control-Allow-Origin:*})exceptCozeAuthErrorase:raiseHTTPException(status_code401,detailf認(rèn)證失敗{str(e)})exceptCozeAPIErrorase:raiseHTTPException(status_code502,detailfCoze API錯(cuò)誤{str(e)})exceptExceptionase:raiseHTTPException(status_code500,detailf服務(wù)器錯(cuò)誤{str(e)})# # 啟動服務(wù)# if__name____main__:importuvicorn uvicorn.run(app,host0.0.0.0,port8000)使用方法安裝依賴pipinstallfastapi uvicorn cozepy設(shè)置環(huán)境變量可選exportCOZE_API_TOKEN你的CozeToken啟動服務(wù)python main.py服務(wù)將監(jiān)聽http://0.0.0.0:8000。使用curl測試接口你可以使用curl來實(shí)時(shí)查看 SSE 流# 測試Python服務(wù)curl-X POST -HContent-Type: application/json-d{ user_id: 123, bot_id: 7579834670624407602, stream: true, additional_messages: [ { role: user, type: question, content_type: text, content: [{ ype: ext, ext:你好}] } ] }http://localhost:8000/api/coze-chat參數(shù)說明-N/--no-buffer禁用輸出緩存實(shí)時(shí)顯示流式數(shù)據(jù)。-X POST發(fā)送 POST 請求。-d傳遞 JSON 請求體。執(zhí)行后你會看到類似以下輸出SSE 流data: {type: delta, content: 你} data: {type: delta, content: 好} data: {type: delta, content: Coze!} data: {type: completed, token_count: 12, conversation_id: conv_123} data: [DONE]前端示例實(shí)時(shí)渲染打字機(jī)效果dividchat/divscriptconstchatDivdocument.getElementById(chat);constevtSourcenewEventSource(http://localhost:8000/api/coze-chat);evtSource.onmessage(e){if(e.data[DONE]){console.log(聊天結(jié)束);return;}constdataJSON.parse(e.data);if(data.typedelta){chatDiv.innerHTMLdata.content;}elseif(data.typecompleted){console.log(聊天完成, token_count:,data.token_count);}};evtSource.onerror()console.log(連接錯(cuò)誤或關(guān)閉);/script效果消息逐字符顯示模擬 AI 打字機(jī)輸出。總結(jié)通過 FastAPI 可以快速實(shí)現(xiàn) Coze 流式聊天接口。SSE 格式讓前端無需輪詢即可接收消息增量。使用curl或前端 JS 均可實(shí)時(shí)驗(yàn)證流式輸出。可擴(kuò)展為 AI 聊天助手、客服機(jī)器人或協(xié)作工具。