SSE流式使用

因为大模型相当于一个大型的预测机器。所以吐字是一个一个吐的。这种像水流一样的方式专业来说就是服务器单向向前端持续推送消息的机制就叫做SSE。我会用py fastapi react 去简单构建一个小demo。pip install fastapi uvicorn langchain langchain-openaifromfastapiimportFastApifromfastapi.responsesimportStreamingResponsefrompydanicimportBaseModel//fastapi处理前端的参数之后会解析成这个类。fromlangchain_core.promptsimportChatPromptTemplatefromlangchain_openaiimportChatOpenAI调用大模型的部分appFastApi()tasks{}classChatRequest(BaseModel):question:strpromptChatPromptTemplate.from_messages([(system,你是一个通用的智能助手。请认真理解用户问题并用准确、清晰、简洁的方式回答。如果你不确定答案请明确说明不要编造。),(user,{question})])llmChatOpenAI(modelgpt-4o-mini,temperature0.2,streamingTrue)chainprompt|llm 接口部分 pythonapp.post(/chat)defcreate_chat(req:ChatRequest):questionreq.question task_idstr(uuid.uuid4())tasks[task_id]{question:question,status:pending}return{task_id:task_id}app.get(/chat/stream/{task_id})defstream_chat(task_id:str):defevent_generator():tasktasks.get(task_id)ifnottask:yield:fdata:[DONE]\n\nquestiontask.get(question)task[status]runningforchunkinchain.stream({question:question}):ifhasattr(chunk,content)andchunk.content:yieldfdata:{chunk.content}\n\nyieldfdata: [DONE]\n\nreturnstreamingResponse(event_generator,media_type:text/event-stream)前端代码我这边写一些关键的// 第一次请求先拿到task_idconstsendMessageasync(){constresawaitfetch(url,{method:POST,body:JOSN.stringify({question:qus})})consttaskIdres.task_id;// 然后用eventSource去监听consteventSourcenewEventSource(${url}/${taskId})eventSource.onmessage(event){if(event.data[DONE]){eventSource.close();return}//event.data 去拼出来就行了}}