python使用StreamingResponse进行流式传输，一直出现socket.send() raised exception，该怎么解决？

• Z时代
• 2024-02-08
• 分类：IT

问题描述

在python环境下使用fastAPI编写了一个流式响应的接口，在多次调接口的时候后台会不断报错socket.send() raised exception,一直循环报错，导致接口无法使用。

问题出现的环境背景及自己尝试过哪些方法

尝试过加错误处理检查是否与客户端连接断开，但问题仍然存在。

问chatGPT说
在使用 asyncio 进行流式传输时可能遇到的一个问题，即在发送大量数据或者在短时间内建立新的连接时，可能会触发一个无限循环的异常。

如何能在发生这个错误时使得服务能继续使用，或者这个错误是什么导致的呢该如何解决？

完整api代码

app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=['*'],
    allow_credentials=True,
    allow_methods=['*'],
    allow_headers=['*'],
)
class Message(BaseModel):
    role: str
    content: str
class ChatBody(BaseModel):
    messages: List[Message]
    model: str
    stream: Optional[bool] = True
    max_tokens: Optional[int]
    temperature: Optional[float]
    top_p: Optional[float]
@app.post("/v1/chat/completions")
async def completions(body: ChatBody, request: Request, background_tasks: BackgroundTasks):
    background_tasks.add_task(torch_gc)
    question = body.messages[-1]
    if question.role == 'user':
        question = question.content
    else:
        raise HTTPException(status.HTTP_400_BAD_REQUEST, "No Question Found")
    history = []
    user_question = ''
    for message in body.messages:
        if message.role == 'system':
            history.append((message.content, "OK"))
        if message.role == 'user':
            user_question = message.content
        elif message.role == 'assistant':
            assistant_answer = message.content
            history.append((user_question, assistant_answer))
    if body.stream:
        async def eval_llm():
            first = True
            for response in context.model.do_chat_stream(
                context.model, context.tokenizer, question, history, {
                    "temperature": body.temperature,
                    "top_p": body.top_p,
                    "max_tokens": body.max_tokens,
                }):
                try:
                    yield response
                except BrokenPipeError as e:
                    print(f"Client disconnected: {e}")
                    break
                except Exception as e:
                    print(f"An exception occurred while sending data: {e}")
                    break
        return StreamingResponse(
            eval_llm(),
            media_type="text/plain",
            headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
        )

do_chat_stream函数代码

def do_chat_stream(model, tokenizer, question, history, model_args = None):
    model_args = init_model_args(model_args)
    sends = 0
    #根据question和history计算聊天回答，循环迭代模型的输出每次迭代获取回答sends变量记录已发送的回答长度
    for response, _ in model.stream_chat(
            tokenizer, question, history,
            temperature=model_args['temperature'],
            top_p=model_args['top_p'],
            max_length=max(2048, model_args['max_tokens'])):
        ret = response[sends:]
        # 修复表情符号的输出问题
        if "\uFFFD" == ret[-1:]:
            continue
        sends = len(response)
        yield ret

你期待的结果是什么？实际看到的错误信息又是什么？

报错信息：
socket.send() raised exception.
socket.send() raised exception.
socket.send() raised exception.
socket.send() raised exception.
socket.send() raised exception.
socket.send() raised exception.
socket.send() raised exception.
socket.send() raised exception.

回答：

多加一些异常捕获，看看哪里的问题

if body.stream:
    async def eval_llm():
        first = True
        for response in context.model.do_chat_stream(
            context.model, context.tokenizer, question, history, {
                "temperature": body.temperature,
                "top_p": body.top_p,
                "max_tokens": body.max_tokens,
            }):
            try:
                yield response
            except (BrokenPipeError, ConnectionResetError) as e:
                print(f"Client disconnected: {e}")
                break
            except Exception as e:
                print(f"An exception occurred while sending data: {e}")
                break
    return StreamingResponse(
        eval_llm(),
        media_type="text/plain",
        headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
    )

以上是 python使用StreamingResponse进行流式传输，一直出现socket.send() raised exception，该怎么解决？ 的全部内容， 来源链接： utcz.com/p/938892.html