开源模型应用落地-FastAPI-助力模型交互-进阶篇（一）

一、前言

   FastAPI 的高级用法可以为开发人员带来许多好处。它能帮助实现更复杂的路由逻辑和参数处理，使应用程序能够处理各种不同的请求场景，提高应用程序的灵活性和可扩展性。

    在数据验证和转换方面，高级用法提供了更精细和准确的控制，确保输入数据的质量和安全性。它还能更高效地处理异步操作，提升应用程序的性能和响应速度，特别是在处理大量并发请求时优势明显。

    此外，高级用法还有助于更好地整合数据库操作、实现数据的持久化和查询优化，以及实现更严格的认证和授权机制，保护应用程序的敏感数据和功能。总之,掌握 FastAPI 的高级用法可以帮助开发人员构建出功能更强大、性能更卓越、安全可靠的 Web 应用程序。

    本篇学习FastAPI的生命周期事件，示例均在开源模型应用落地-FastAPI-助力模型交互-WebSocket篇（二）基础上进行扩展，建议有需要的老铁们，先去学习。

二、术语

2.1. Lifespan Events（生命周期事件）

    通过生命周期事件，可以更好地管理应用的整个生命周期中的资源和操作，确保资源的正确初始化和释放，提高应用的性能、可靠性和可维护性。

    Lifespan Events主要有以下作用：

  1. 资源初始化与释放：可以在应用启动时执行一些初始化操作，例如创建数据库连接池、加载共享的机器学习模型等需要在整个应用中使用且可在请求间共享的资源。在应用关闭时，执行清理和释放资源的操作，例如关闭数据库连接、释放内存或其他相关资源。
  2. 避免不必要的操作：如果某些资源的初始化成本较高（如加载大型模型），使用 Lifespan Events 可以避免在每次请求时都进行初始化，仅在应用启动后且接收请求之前执行一次。同时，也可以防止在一些不需要处理实际请求的情况下（如运行简单的自动化测试）进行不必要的资源加载，从而提高性能和效率。
  3. 分离启动和关闭逻辑：将与应用启动和关闭相关的逻辑集中在一个地方进行管理，使代码更加清晰和可维护。
     

三、前置条件

3.1. 创建虚拟环境&安装依赖

  增加Google Search以及langchainhub的依赖包

conda create -n fastapi_test python=3.10
conda activate fastapi_test
pip install fastapi websockets uvicorn transformers==4.32.0 accelerate tiktoken einops transformers_stream_generator==0.0.4 scipy

3.2. 下载Qwen-1_8B-Chat模型

huggingface：

https://huggingface.co/Qwen/Qwen-1_8B-Chaticon-default.png?t=N7T8https://huggingface.co/Qwen/Qwen-1_8B-Chat

​魔搭：

魔搭社区汇聚各领域最先进的机器学习模型，提供模型探索体验、推理、训练、部署和应用的一站式服务。icon-default.png?t=N7T8https://modelscope.cn/models/qwen/Qwen-1_8B-Chat

四、技术实现

4.1. startup & shutdown event

# -*- coding: utf-8 -*-
import tracebackfrom transformers import AutoTokenizer, AutoModelForCausalLM
from transformers import GenerationConfigimport torch
import uvicornfrom typing import Annotated
from fastapi import (Depends,FastAPI,WebSocket,WebSocketException,WebSocketDisconnect,status,
)model_path = "E:/model/qwen-1_8b-chat"class ConnectionManager:def __init__(self):self.active_connections: list[WebSocket] = []async def connect(self, websocket: WebSocket):await websocket.accept()self.active_connections.append(websocket)def disconnect(self, websocket: WebSocket):self.active_connections.remove(websocket)async def send_personal_message(self, message: str, websocket: WebSocket):await websocket.send_text(message)async def broadcast(self, message: str):for connection in self.active_connections:await connection.send_text(message)manager = ConnectionManager()app = FastAPI()async def authenticate(websocket: WebSocket,userid: str,secret: str,
):if userid is None or secret is None:raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION)print(f'userid: {userid},secret: {secret}')if '12345' == userid and 'xxxxxxxxxxxxxxxxxxxxxxxxxx' == secret:return 'pass'else:return 'fail'async def chat(query):position = 0try:for response in model.chat_stream(tokenizer, query, history = None):result = response[position:]position = len(response)yield resultexcept Exception:traceback.print_exc()@app.websocket("/ws")
async def websocket_endpoint(*,websocket: WebSocket,userid: str,permission: Annotated[str, Depends(authenticate)],):await manager.connect(websocket)try:while True:text = await websocket.receive_text()if 'fail' == permission:await manager.send_personal_message(f"authentication failed", websocket)else:if text is not None and len(text) > 0:async for msg in chat(text):await manager.send_personal_message(msg, websocket)except WebSocketDisconnect:manager.disconnect(websocket)print(f"Client #{userid} left the chat")await manager.broadcast(f"Client #{userid} left the chat")def loadTokenizer():tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)return tokenizerdef loadModel(config):model = AutoModelForCausalLM.from_pretrained(model_path, device_map="cpu", trust_remote_code=True).eval()model.generation_config = configreturn model@app.on_event("startup")
async def startup_event():global model,tokenizerconfig = GenerationConfig.from_pretrained(model_path, trust_remote_code=True, top_p=0.9, temperature=0.45,repetition_penalty=1.1, do_sample=True, max_new_tokens=8192)tokenizer = loadTokenizer()model = loadModel(config)@app.on_event("shutdown")
def shutdown_event():torch.cuda.empty_cache()if __name__ == '__main__':uvicorn.run(app, host='0.0.0.0',port=7777)

调用结果：

用户输入：你好

模型输出：你好！有什么我能帮助你的吗？

说明：

  1. 在startup事件函数中加载模型资源
  2. 在shutdown时间函数中释放资源
  3. startup & shutdown event已过期，后面可能会被移除，建议使用lifespan event代替

4.2. lifespan event

import traceback
from contextlib import asynccontextmanagerfrom transformers import AutoTokenizer, AutoModelForCausalLM
from transformers import GenerationConfigimport torch
import uvicornfrom typing import Annotated
from fastapi import (Depends,FastAPI,WebSocket,WebSocketException,WebSocketDisconnect,status,
)model_path = "E:/model/qwen-1_8b-chat"class ConnectionManager:def __init__(self):self.active_connections: list[WebSocket] = []async def connect(self, websocket: WebSocket):await websocket.accept()self.active_connections.append(websocket)def disconnect(self, websocket: WebSocket):self.active_connections.remove(websocket)async def send_personal_message(self, message: str, websocket: WebSocket):await websocket.send_text(message)async def broadcast(self, message: str):for connection in self.active_connections:await connection.send_text(message)manager = ConnectionManager()def loadTokenizer():tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)return tokenizerdef loadModel(config):model = AutoModelForCausalLM.from_pretrained(model_path, device_map="cpu", trust_remote_code=True).eval()model.generation_config = configreturn model@asynccontextmanager
async def lifespan(app: FastAPI):# 加载模型global model, tokenizerconfig = GenerationConfig.from_pretrained(model_path, trust_remote_code=True, top_p=0.9, temperature=0.45,repetition_penalty=1.1, do_sample=True, max_new_tokens=8192)tokenizer = loadTokenizer()model = loadModel(config)yield# 释放资源torch.cuda.empty_cache()app = FastAPI(lifespan=lifespan)async def authenticate(websocket: WebSocket,userid: str,secret: str,
):if userid is None or secret is None:raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION)print(f'userid: {userid},secret: {secret}')if '12345' == userid and 'xxxxxxxxxxxxxxxxxxxxxxxxxx' == secret:return 'pass'else:return 'fail'async def chat(query):position = 0try:for response in model.chat_stream(tokenizer, query, history = None):result = response[position:]position = len(response)yield resultexcept Exception:traceback.print_exc()@app.websocket("/ws")
async def websocket_endpoint(*,websocket: WebSocket,userid: str,permission: Annotated[str, Depends(authenticate)],):await manager.connect(websocket)try:while True:text = await websocket.receive_text()if 'fail' == permission:await manager.send_personal_message(f"authentication failed", websocket)else:if text is not None and len(text) > 0:async for msg in chat(text):await manager.send_personal_message(msg, websocket)except WebSocketDisconnect:manager.disconnect(websocket)print(f"Client #{userid} left the chat")await manager.broadcast(f"Client #{userid} left the chat")if __name__ == '__main__':uvicorn.run(app, host='0.0.0.0',port=7777)

调用结果：

没有输出警告信息

用户输入：你好，广州有什么好玩的地方推荐？

模型输出：广州有很多值得一去的景点，比如白云山、长隆野生动物园、陈家祠、珠江夜游等。此外，你还可以去逛逛上下九步行街，品尝当地的美食，或者参观广州塔等高楼大厦。

五、附带说明

5.1. 测试界面

<!DOCTYPE html>
<html><head><title>Chat</title></head><body><h1>WebSocket Chat</h1><form action="" onsubmit="sendMessage(event)"><label>USERID: <input type="text" id="userid" autocomplete="off" value="12345"/></label><label>SECRET: <input type="text" id="secret" autocomplete="off" value="xxxxxxxxxxxxxxxxxxxxxxxxxx"/></label><br/><button onclick="connect(event)">Connect</button><hr><label>Message: <input type="text" id="messageText" autocomplete="off"/></label><button>Send</button></form><ul id='messages'></ul><script>var ws = null;function connect(event) {var userid = document.getElementById("userid")var secret = document.getElementById("secret")ws = new WebSocket("ws://localhost:7777/ws?userid="+userid.value+"&secret=" + secret.value);ws.onmessage = function(event) {var messages = document.getElementById('messages')var message = document.createElement('li')var content = document.createTextNode(event.data)message.appendChild(content)messages.appendChild(message)};event.preventDefault()}function sendMessage(event) {var input = document.getElementById("messageText")ws.send(input.value)input.value = ''event.preventDefault()}</script></body>
</html>

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/bicheng/42936.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Leetcode 3209. Number of Subarrays With AND Value of K

Leetcode 3209. Number of Subarrays With AND Value of K

Leetcode 3209. Number of Subarrays With AND Value of K 1. 解题思路2. 代码实现 题目链接&#xff1a;3209. Number of Subarrays With AND Value of K 1. 解题思路 这一题的话整体上是一个滑动窗口的思路&#xff0c;我们维护一个滑动窗口&#xff0c;确保其每一个窗口都…
阅读更多...
『大模型笔记』你需要的不是智能体，而是一个适合 AI 的工作流

『大模型笔记』你需要的不是智能体，而是一个适合 AI 的工作流

你需要的不是智能体，而是一个适合 AI 的工作流 文章目录 一. 你需要的不是智能体，而是一个适合 AI 的工作流1. 不要将 AI 的解决方案局限在人类现有的解决方案上2. 不必完全依赖 AI 做决策，而是让 AI 辅助做决策或者做简单的决策3. 结合不同领域的 AI 模型或者工具，设计合适…
阅读更多...
RedHat运维-Linux文本操作基础-SED基础

RedHat运维-Linux文本操作基础-SED基础

1. 打印出/etc/passwd的第12行的命令为_______________________________________&#xff1b; 2. 打印出/etc/passwd的第12到第18行的命令为________________________________________&#xff1b; 3. 打印出/etc/passwd的总行数的命令为_____________________________________…
阅读更多...
低代码研发项目管理流程优化：提效与创新的双重驱动

低代码研发项目管理流程优化：提效与创新的双重驱动

随着信息技术的迅猛发展&#xff0c;软件项目的规模和复杂度日益增加&#xff0c;传统的软件开发方式已经难以满足快速迭代和高效交付的需求。在这一背景下&#xff0c;低代码平台应运而生&#xff0c;以其高效、灵活、易用的特点&#xff0c;迅速成为软件行业的新宠。然而&…
阅读更多...
运行pip出现UnicodeDecodeError: ‘ascii‘ codec can‘t decode

运行pip出现UnicodeDecodeError: ‘ascii‘ codec can‘t decode

错误: UnicodeDecodeError: ascii codec cant decode byte 0xe2 in position 1025: ordinal not in range(128) ERROR: Exception: Traceback (most recent call last):File "/usr/local/lib/python3.6/dist-packages/pip/_internal/cli/base_command.py", line 22…
阅读更多...
C语言下结构体、共用体、枚举类型的讲解

C语言下结构体、共用体、枚举类型的讲解

主要内容 结构体结构体数组结构体指针包含结构体的结构链表链表相关操作共用体枚举类型 结构体 结构体的类型的概念 结构体实现步骤 结构体变量的声明 struct struct 结构体名{ 数据类型 成员名1; 数据类型 成员名2; ..…
阅读更多...
PostgreSQL的pg_bulkload工具

PostgreSQL的pg_bulkload工具

PostgreSQL的pg_bulkload工具 pg_bulkload 是一个针对 PostgreSQL 提供高性能批量数据加载的工具。相较于内置的 COPY 命令&#xff0c;pg_bulkload 更加灵活并且在许多情况下性能更高。它支持数据的强制加载、数据过滤、数据转换以及错误处理等多种功能&#xff0c;非常适合需…
阅读更多...
qt hasPendingDatagrams() 函数

qt hasPendingDatagrams() 函数

hasPendingDatagrams 是 Qt 框架中 QUdpSocket 类的一个方法&#xff0c;用于检查是否有待处理的数据报到达。在 UDP 通信中&#xff0c;数据以数据报的形式发送&#xff0c;而 QUdpSocket 类提供了用于接收和处理这些数据报的功能。 功能描述 hasPendingDatagrams() 方法用于…
阅读更多...
从数据到洞察：DataOps加速AI模型开发的秘密实践大公开！

从数据到洞察：DataOps加速AI模型开发的秘密实践大公开！

作者 | 代立冬&#xff0c;白鲸开源科技联合创始人&CTO 引言 在AI驱动的商业世界中&#xff0c;DataOps作为连接数据与洞察的桥梁&#xff0c;正迅速成为企业数据战略的核心。 在WOT全球技术创新大会2024北京站&#xff0c;白鲸开源联合创始人&CTO 代立冬 在「大数据…
阅读更多...
严重的OpenSSH漏洞威胁数百万Linux系统

严重的OpenSSH漏洞威胁数百万Linux系统

Qualys威胁研究部门(TRU)发现了OpenSSH服务器 (sshd) 中的一个严重漏洞&#xff0c;可能影响全球超过 1400 万个Linux系统。该漏洞被指定为 CVE-2024-6387&#xff0c;允许在基于 glibc 的 Linux 系统上以 root 权限进行远程未经身份验证的代码执行 (RCE)。 此漏洞源于信号处理…
阅读更多...
自己写个简单的vite插件

自己写个简单的vite插件

需求&#xff1a;根据使用环境显示对应的标题和icon 先在根目录建个plugins/vite-plugin-title-html.ts 文件内容如下: /*** 替换html里面的标题和icon*/ type HtmlTemplate {title?: string,icon?: string } export default function vitePluginHtmlTitle({ title, icon…
阅读更多...
Python 处理Excel 文件， openpyxl 库的使用：

Python 处理Excel 文件， openpyxl 库的使用：

下载&#xff1a; pip install openpyxl 基本使用&#xff1a; 新建一个Excel 工作簿&#xff1a; 使用openpyxl 需要先导入一个Workbook 类&#xff0c; 使用它可以创建一个Workbook<工作簿>对象&#xff0c; 也就是创建一个Excel表文件&#xff0c; web.active 可用来…
阅读更多...
智能体重秤pcba方案

智能体重秤pcba方案

智能体重秤应用系统由硬件和软件两部分组成。硬件是指微控制器、扩展存储器、扩展输入输出设备等。软件是各种工作过程的通用名称。硬件和软件只有紧密协调&#xff0c;才能提高系统的性价比。从硬件设计开始&#xff0c;应考虑相应软件的设计方法&#xff0c;软件的设计是基于…
阅读更多...
代码随想录算法训练营：26/60

代码随想录算法训练营：26/60

非科班学习算法day26 | LeetCode491:非递减子序列 &#xff0c;Leetcode46:全排列 &#xff0c;Leetcode47:全排列|| 介绍 包含LC的两道题目&#xff0c;还有相应概念的补充。 相关图解和更多版本&#xff1a; 代码随想录 (programmercarl.com)https://programmercarl.com/…
阅读更多...
5款好用公司监控软件分享|管理者必看

5款好用公司监控软件分享|管理者必看

当今社会&#xff0c;企业数据安全和员工工作效率成为了管理者不可忽视的重要议题。 选择合适的公司监控软件&#xff0c;不仅有助于提升管理效率&#xff0c;还能有效保障企业信息安全。 下面小编将为您分享五款备受好评的公司监控软件&#xff0c;助力管理者更好地管理企业…
阅读更多...
C# Winform权限、用户和菜单开发的顺序和注意点

C# Winform权限、用户和菜单开发的顺序和注意点

在C# Winform应用程序中&#xff0c;开发权限、用户和菜单功能通常遵循一定的顺序和注意点&#xff0c;以确保功能的连贯性和安全性。下面是一个推荐的开发流程及其注意事项&#xff1a; 开发流程 1. 数据库设计 用户表&#xff1a;存储用户基本信息&#xff0c;如用户名、密…
阅读更多...
vue使用HMAC-SHA256签名算法

vue使用HMAC-SHA256签名算法

在 Vue.js 应用中生成签名算法通常涉及以下几个步骤&#xff1a; 收集数据&#xff1a;获取需要签名的数据。整理数据&#xff1a;根据协议或需求对数据进行排序、拼接、编码等处理。计算签名&#xff1a;使用密钥和算法&#xff08;如 HMAC-SHA256&#xff09;计算签名。附加…
阅读更多...
微服务通信方式详解

微服务通信方式详解

引言 随着互联网应用的不断发展和用户需求的多样化&#xff0c;传统的单体架构已经无法满足现代应用的灵活性和扩展性需求。微服务架构因其模块化、松耦合、易于扩展和部署等优势&#xff0c;逐渐成为现代软件开发的重要趋势。在微服务架构中&#xff0c;各个服务相互独立、自…
阅读更多...
公司可以拿监控辞退员工吗？有什么法律依据？

公司可以拿监控辞退员工吗？有什么法律依据？

李经理&#xff1a;小张&#xff0c;我听说最近人力资源部打算使用我们新安装的安企神软件来监控员工的工作行为&#xff0c;以提高工作效率和确保公司信息安全。不过&#xff0c;我有点担心这会不会触及法律红线&#xff0c;比如如果我们发现某位员工严重违反公司规定&#xf…
阅读更多...
virtualbox和docker的区别和优缺点以及如何选择

virtualbox和docker的区别和优缺点以及如何选择

使用 Docker 和 VirtualBox 各有优缺点&#xff0c;具体取决于你的需求和使用场景。以下是两种方法的详细对比&#xff1a; Docker 使用 CentOS 7 优点 轻量级&#xff1a; Docker 容器共享主机操作系统内核&#xff0c;启动速度快&#xff0c;占用资源少。适合开发、测试和部…
阅读更多...
最新文章

Copyright @ 2022~2023