Ascend+FastAPI+ Uvicorn 实现推理

1、FastAPI & Uvicorn

FastAPI

FastAPI 是一个用于构建 API 的现代、快速（高性能）的 Python Web 框架，专为在 Python 中构建 RESTful API 而设计，具有以下特点：
高性能

• 性能卓越：FastAPI 的性能可与 NodeJS 和 Go 等语言的高性能 Web 框架相媲美，是最快的 Python Web 框架之一。这得益于其底层的 Starlette 框架和异步编程模型，能够充分利用 Python 异步生态系统，提供卓越的性能和吞吐量。
• 异步支持：FastAPI 支持基于 Python 3.7 及以上版本的异步编程，使得处理 IO 密集型任务更加高效，非常适合高并发的场景。

from fastapi import FastAPI
app = FastAPI()

@app.get("/")
async def read_root():
    return {"Hello": "World"}

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

在这个例子中，我们定义了两个路由：一个根路由 / 和一个带参数的路由 /items/{item_id}。通过装饰器 @app.get()，我们指定了 HTTP 方法和路由路径。函数参数将自动从查询参数、路径参数、请求头等中提取，并进行数据验证。

Uvicorn

Uvicorn 是一个基于 ASGI（Asynchronous Server Gateway Interface）的轻量级异步服务器，用于运行 Python Web 应用程序，特别是那些基于 ASGI 的现代异步框架，如 FastAPI、Starlette 等。以下是关于 Uvicorn 的详细介绍：
基本功能

• 异步支持：Uvicorn 是一个异步服务器，能够充分利用 Python 的异步编程特性。它使用 Python 的 asyncio 库来处理并发请求，使得应用程序能够更高效地处理大量并发连接，特别适合处理 IO 密集型任务，如网络请求、数据库操作等。
• 高性能：Uvicorn 的性能非常出色，它能够快速地处理请求和响应，提供高吞吐量和低延迟的服务。在许多基准测试中，Uvicorn 的性能表现与一些传统的同步服务器（如 Gunicorn）相比具有显著优势，尤其是在处理大量并发请求时。

FastAPI + Uvicorn 的结合

• 无缝集成：Uvicorn 与 FastAPI 结合使用非常方便。FastAPI 生成的 ASGI 应用可以直接由 Uvicorn 运行，无需额外的适配。这种组合充分利用了 FastAPI 的开发友好性和 Uvicorn 的高性能，使得开发和部署异步 Web 应用变得非常简单。
• 生产环境：在生产环境中，Uvicorn 可以与 Nginx 等反向代理服务器配合使用，提供更稳定、更安全的服务。Nginx 可以处理静态文件、负载均衡、SSL/TLS 加密等任务，而 Uvicorn 专注于处理动态请求。

假设有一个简单的 FastAPI 应用 app.py：

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def read_root():
    return {"Hello": "World"}

可以通过以下命令使用 Uvicorn 启动该应用：
bash复制uvicorn app:app --host 0.0.0.0 --port 8000
访问 http://localhost:8000 即可看到返回的 JSON 响应。

2、Ascend上跑推理

方案设计
具体的模型通过transformers 库的from_pretrained接口从xxx-config配置文章引入。

model = AutoModel.from_pretrained("xxx-config", torch_dtype="auto").to("npu:0")，

引入。然后就是，以FastAPI + Uvicorn启动一个web服务，通过generate_chat_output运行推理，获取推理结果。
整体方案代码参考如下：

import torch
import torch_npu
import uvicorn
from typing import List
from fastapi import FastAPI
from contextlib import asynccontextmanager
from transformers import AutoTokenizer, AutoModel
from fastapi.middleware.cors import CORSMiddleware

# 设置卡
torch.npu.set_device("npu:0")

# 配置tokenizer 
tokenizer = AutoTokenizer.from_pretrained("xxx-config")

# 获取模型
model = AutoModel.from_pretrained("xxx-config", torch_dtype="auto").to("npu:0")

@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    if torch_npu.npu.is_available():
        torch_npu.npu.empty_cache()

# 实例化api服务器
app = FastAPI(lifespan=lifespan)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
)

# 推理实现
@app.post("/llm/chat")
async def generate_chat_output(item: Item):
    try:
        texts = [t.replace("\n", " ") for t in item.sentences]
        encoded_inputs = tokenizer(texts, truncation=True, return_tensors="pt", padding=True, max_length=512).to("npu:0")
        with torch.no_grad():
            model_output = model(**encoded_inputs)
    except Exception as e:
        traceback.print_exc()

    return sequence_embeddings.tolist()

# uvicorn提供服务化
if __name__ == '__main__':
    uvicorn.run(app, host="0.0.0.0", port=8080)

构建测试
测试文件test.py如下：

import httpx
data = {
    "sentences": ["中国GDP为129.43万亿元，GDP增速为5.25%，人均GDP为8.94万元/人，人均GDP增速为5.40%，其中第一产业占6.90%，第二产业占36.80%，第三产业占56.30%"]
}
def request_url():
    with httpx.Client() as client:
        res = client.post("http://127.0.0.1:1025/llm/chat", json=data)
        print(res.json())

运行

python test.py

客户端和服务器均正常。