AI结合微服务的落地应用

1**.**需求

   应客户需求，需要图像高清、抠图、扩图、文生图、图生图、换脸、模特换装、智能客服、以图搜图等AI相关的服务应用。

   通过分析客户需求，使用微服务开发基于AI各种算法使用的应用系统，系统包括：接口调用，同步、异步任务调度，图像算法实现，自然语言模型使用，非结构化数据存储，结构化数据存储，消息队列，网关管理等功能。

2.整体架构设计

2.1 系统架构图

3.开发环境配置

3.1使用丹摩平台的GPU服务器，链接：https://www.damodel.com/

4090的服务器现在有优惠，创建实例后：

3.2通过终端进入系统，安装pytorch

pip3 install torch torchvision torchaudio

在丹摩平台安装完成

4.图像高清环境准备

4.1模型下载

提前下载好模型，上传到丹摩平台

模型有两处，第一处：gfpgan/weights

第二处：server_router/high_resolution_fix/weights

4.2 算法验证

5.分割模型环境准备

5.1 提前下载好分割模型sam_hq_vit_l.pth

注意：vit模型有4个，通过反复试验，sam_hq_vit_l.pth模型满足使用我的使用要求

5.2 分割模型调用代码：

# 设置模型路径
sam_model =”sam_hq_vit_l.pth”
# 转化矩形为np数组
# sam_boxs数据格式为一个矩形框的点坐标，例如：[[45, 260, 515, 470], [310, 228, 424, 296]]
sam_array = json.loads(sam_boxs)
# 模型准备
sam = sam_model_registry[model_type](checkpoint=sam_model)
sam.to(device=device)
predictor = SamPredictor(sam)
hq_token_only = False

# 将图像数据读入opencv
cv2_image = cv2.imdecode(np.fromstring(data, np.uint8), cv2.IMREAD_COLOR)
predictor.set_image(cv2_image)

# multi box input
input_box = torch.tensor(sam_array, device=predictor.device)
transformed_box = predictor.transform.apply_boxes_torch(input_box, cv2_image.shape[:2])
input_point, input_label = None, None

# 通过模型预测执行
masks, scores, logits = predictor.predict_torch(
    point_coords=input_point,
    point_labels=input_label,
    boxes=transformed_box,
    multimask_output=False,
    hq_token_only=hq_token_only,
)
masks = masks.squeeze(1).cpu().numpy()
scores = scores.squeeze(1).cpu().numpy()
input_box = input_box.cpu().numpy()

# 生成识别区域图片
fn = output_image(cv2_image, masks, scores, input_box, logits)

5.3分割模型验证

6.FastApi构建AI应用

6.1项目目录结构

6.2 功能

1、在server.py里设置路由，按照功能设置路由

import uvicorn
from fastapi import FastAPI

from starlette.middleware.cors import CORSMiddleware

from common.log import Loggers
from common.utils import CustomResponse

# 引入路由
from server_router.high_resolution_fix.server import app as high_resolution_fix
from server_router.segment_anything_model.server import app as segment_anything_model
from server_router.stable_diffusion.server import app as stable_diffusion
from server_router.object_storage_service.server import app as object_storage_service
from server_router.img_task.server import app as img_task
from server_router.img_test.server import app as img_test

Loggers.init_config()
# 创建服务
app = FastAPI(default_response_class=CustomResponse)

# 添加CORS中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 允许所有源
    allow_credentials=True,
    allow_methods=["*"],  # 允许所有方法
    allow_headers=["*"],  # 允许所有头
)

# 路由挂载
app.include_router(high_resolution_fix, prefix='/watermelon/hrf', tags=['高清', '高清修复'])
app.include_router(segment_anything_model, prefix='/watermelon/sam', tags=['分割', '抠图、局部识别'])
app.include_router(stable_diffusion, prefix='/watermelon/sd', tags=['AI生成图像', '文生图、图生图'])
app.include_router(object_storage_service, prefix='/watermelon/oss', tags=['对象存储', '读取、写入文件'])
app.include_router(img_test, prefix='/watermelon/img_test', tags=['测试', '测试'])
app.include_router(img_task, prefix='/watermelon/img_task', tags=['调度任务', '任务状态、执行、结果'])

if __name__ == '__main__':
    uvicorn.run(app=app, host='0.0.0.0', port=8081)

2、路由对应模块

每个功能目录下的内容包括utils.py、server.py、model.py、init.py，以img_test目录举例：

utils.py里写入功能逻辑，用于接口的调用

def func_test1(db: Session, code: str):
    print(code)
    log.info(code)

server.py里写入具体接口，用于和前端交互

@app.post("/test1")
async def test1(data: Test1Model = Body(..., embed=False)):
    log.info("/test1")
    local_file = f'E:\\pic3.jpg'
    minio_client.upload_local_file("test1", local_file, "pic3.jpg")

    return CustomResponse(
        content={
            "code": status.HTTP_200_OK,
            "data": data.json()
        },
        msg=get_code_message(status.HTTP_200_OK),
        error=None,
        status=status.HTTP_200_OK

model.py里写入数据模型定义

class Test1Model(BaseModel):
    key1: str
    key2: str

6.3 FastApi启动

在项目目录下启动

nohup uvicorn server:app --host 0.0.0.0 --port 8080 --workers 10 --reload > log.txt 2>&1 &