企业官网建设流程全解析

镇江做网站公司,在线教育培训平台定制,怎么查询自己注册的商标,做网站宽度和长度布局使用 FastAPI 实现 Coze 流式聊天 SSE 接口 在开发 AI 助手或聊天应用时#xff0c;我们通常希望服务端能够 实时向前端推送消息#xff0c;让用户看到逐字打字效果。本文演示如何使用 FastAPI Coze Python SDK#xff08;cozepy#xff09; 实现 流式聊天 SSE 接口…使用 FastAPI 实现 Coze 流式聊天 SSE 接口在开发 AI 助手或聊天应用时我们通常希望服务端能够实时向前端推送消息让用户看到逐字打字效果。本文演示如何使用FastAPI Coze Python SDKcozepy实现流式聊天 SSE 接口并提供curl测试方法。功能特点流式输出前端可以实时接收聊天增量消息。SSE 格式便于浏览器或 Go/Node 前端解析。兼容不同版本 Coze SDK处理可能缺失的异常类。可直接使用curl测试无需前端即可验证接口。技术栈Python 3.10FastAPIuvicornASGI 服务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应用# appFastAPI(titleCoze Stream Chat API)# # 全局配置与Coze客户端初始化# COZE_API_TOKENos.getenv(COZE_API_TOKEN,你的默认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:# 构建 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))# 调用流式接口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})}\n\n# 聊天完成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})}\n\nyielddata: [DONE]\n\nexceptExceptionase:yieldfdata:{json.dumps({type:error,message:str(e)})}\n\nreturnStreamingResponse(generate_stream(),media_typetext/event-stream,headers{Cache-Control:no-cache,Connection:keep-alive,Access-Control-Allow-Origin:*})exceptCozeAuthErrorase:raiseHTTPException(status_code401,detailf认证失败{str(e)})exceptCozeAPIErrorase:raiseHTTPException(status_code502,detailfCoze API错误{str(e)})exceptExceptionase:raiseHTTPException(status_code500,detailf服务器错误{str(e)})# # 启动服务# if__name____main__:importuvicorn uvicorn.run(app,host0.0.0.0,port8000)使用方法安装依赖pipinstallfastapi uvicorn cozepy设置环境变量可选exportCOZE_API_TOKEN你的CozeToken启动服务python main.py服务将监听http://0.0.0.0:8000。使用curl测试接口你可以使用curl来实时查看 SSE 流# 测试Python服务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: [{\type\:\text\,\text\:\你好\}] } ] }http://localhost:8000/api/coze-chat参数说明-N/--no-buffer禁用输出缓存实时显示流式数据。-X POST发送 POST 请求。-d传递 JSON 请求体。执行后你会看到类似以下输出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]前端示例实时渲染打字机效果dividchat/divscriptconstchatDivdocument.getElementById(chat);constevtSourcenewEventSource(http://localhost:8000/api/coze-chat);evtSource.onmessage(e){if(e.data[DONE]){console.log(聊天结束);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(连接错误或关闭);/script效果消息逐字符显示模拟 AI 打字机输出。总结通过 FastAPI 可以快速实现 Coze 流式聊天接口。SSE 格式让前端无需轮询即可接收消息增量。使用curl或前端 JS 均可实时验证流式输出。可扩展为 AI 聊天助手、客服机器人或协作工具。