当前位置：首页 > news >正文

python3写一个异步流式 http 接口服务调用大模型(async, stream, sanic)---6.2

news 2025/7/27 21:44:31

python3写一个异步流式 http 接口服务调用大模型(async, stream, sanic)

python3写一个异步流式 http 接口服务调用大模型(async, stream, sanic)，异步适合处理I/O密集型操作(文件/网络请求)，特别地，调用大模型等待时间特别长。记录一下！

一、Sanic介绍

Sanic是一个Python3的web服务器和web框架，旨在快速运行。它允许使用Python 3.5中添加的async/await语法。

安装

pip install sanic

二、异步接口代码实现

# !/usr/bin/python
# -*- coding: utf-8 -*-
# @time    : 2025/07/19 18:34
# @author  : Mo
# @function: async http
import json
import traceback
import time
import sys
import ospath_sys = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
sys.path.append(path_sys)
print(path_sys)from sanic.views import HTTPMethodView
from sanic.log import logger
from sanic import response
from sanic import Sanic
import asyncioapp = Sanic("add_async")
app.config.RESPONSE_TIMEOUT = 96   # 超时时间配置
rsp_kwargs = {"content_type": "text/event-stream","headers": {"Cache-Control": "no-cache","Connection": "keep-alive","X-Accel-Buffering": "no"}}async def n_range_stream(n=None, **kwargs):"""  异步程序xxx   """await asyncio.sleep(1)n = n or 10for i in range(n):await asyncio.sleep(0.01)yield json.dumps({"index": i}, ensure_ascii=False) + "\n"class textStreamView(HTTPMethodView):def __init__(self) -> None:super().__init__()async def post(self, request):"""   简单流式输出  """if "application/json" in request.content_type:data = request.jsonelse:data = request.formasync def stream_generator(response):# 调用异步流式响应函数async for content in n_range_stream(**data):# 将每个响应块发送给客户端# await asyncio.sleep(0.01)await response.write(content)# 返回流式响应return response.ResponseStream(stream_generator, **rsp_kwargs)app.add_route(textStreamView.as_view(),uri="/text_stream",methods=["POST", "GET"],version=1,name="add")# ### 测试异步, async
# import time
# time_start = time.time()
# resp = asyncio.run(n_range_stream(a=2, b=3))
# print(resp)
# time_end = time.time()
# print(time_end - time_start)if __name__ == '__main__':app.run(host="0.0.0.0",port=8032,workers=1,access_log=True)