Python----大模型（基于Fastapi+streamlit的機器人對話）

一、準備工作

1.1、魔搭社區下載大模型

通義千問2.5-7B-Instruct · 模型庫

from modelscope.hub.snapshot_download import snapshot_download
llm_model_dir = snapshot_download('Qwen/Qwen2.5-7B-Instruct',cache_dir='models')

1.2、啟動vllm大模型?

python -m vllm.entrypoints.openai.api_server --port 10222 --model /home/AI_big_model/models/Qwen/Qwen2.5-7B-Instruct --served-model-name Qwen2.5-7B-Instruct

二、Fastapi后端

????????FastAPI 是一個用于構建API的現代化、快速（高性能）的Web框架，使用Python 3.7+的標準類型提示。它的性能媲美Node.js和Go，是基于Python的框架中最快的之一。

主要特點：

????????高性能：與Starlette、Pydantic等框架深度集成，性能優異。

????????簡潔明了：基于類型提示（Type Hints），使得代碼更加簡潔且具備良好的可讀性。

????????自動生成文檔：自動生成Swagger UI和ReDoc文檔，方便開發者查看和測試 API。

????????異步支持：原生支持Python的async和await，適合處理異步任務。

pip install fastapi

pip install uvicorn?

# 導入必要的庫
from fastapi import FastAPI, Body  # FastAPI框架和Body請求體
from openai import AsyncOpenAI  # OpenAI異步客戶端
from typing import List  # 類型提示
from fastapi.responses import StreamingResponse  # 流式響應# 初始化FastAPI應用
app = FastAPI()# 初始化openai的客戶端
api_key = "EMPTY"  # 空API密鑰（因為使用本地部署的模型）
base_url = "http://127.0.0.1:10222/v1"  # 本地部署的模型API地址
aclient = AsyncOpenAI(api_key=api_key, base_url=base_url)  # 創建異步客戶端實例# 初始化對話列表（全局變量）
messages = []# 定義路由，實現接口對接
@app.post("/chat")
async def chat(query: str = Body(default='你是誰？', description="用戶輸入"),  # 用戶輸入的查詢文本sys_prompt: str = Body("你是一個有用的助手。", description="系統提示詞"),  # 系統角色設定history: List = Body([], description="歷史對話"),  # 歷史對話記錄history_len: int = Body(1, description="保留歷史對話的輪數"),  # 保留的歷史對話輪數temperature: float = Body(0.5, description="LLM采樣溫度"),  # 生成文本的隨機性控制top_p: float = Body(0.5, description="LLM采樣概率"),  # 核采樣概率閾值max_tokens: int = Body(default=1024, description="LLM最大token數量")  # 生成的最大token數
):  global messages  # 使用全局的messages列表# 控制歷史記錄長度（只保留指定輪數的對話）if history_len > 0:history = history[-2 * history_len:]  # 每輪對話包含用戶和AI兩條記錄，所以乘以2# 清空消息列表（每次請求都重新構建）messages.clear()# 添加系統提示詞messages.append({"role": "system", "content": sys_prompt})# 在message中添加歷史記錄messages.extend(history)# 在message中添加用戶當前的查詢messages.append({"role": "user", "content": query})# 發送請求到本地部署的模型response = await aclient.chat.completions.create(model="Qwen2.5-7B-Instruct",  # 使用的模型名稱messages=messages,  # 完整的對話上下文max_tokens=max_tokens,  # 最大token數temperature=temperature,  # 溫度參數top_p=top_p,  # 核采樣參數stream=True  # 啟用流式輸出)# 定義生成響應的異步生成器函數async def generate_response():# 遍歷流式響應的每個chunkasync for chunk in response:chunk_msg = chunk.choices[0].delta.content  # 獲取當前chunk的文本內容if chunk_msg:  # 如果有內容則yieldyield chunk_msg# 返回流式響應，媒體類型為純文本return StreamingResponse(generate_response(), media_type="text/plain")# 主程序入口
if __name__ == "__main__":import uvicorn# 啟動FastAPI應用# 參數說明：# "fastapi_bot:app" - 要運行的模塊和應用實例# host="0.0.0.0" - 監聽所有網絡接口# port=6066 - 服務端口# log_level="info" - 日志級別# reload=True - 開發模式下自動重載uvicorn.run("fastapi_bot:app", host="0.0.0.0", port=6066, log_level="info", reload=True)

三、streamlit界面設計

????????Streamlit 是一個非常方便的 Python 庫，用來快速構建數據驅動的 Web 應用。在這個項目中，Streamlit 將用于展示聊天界面并與后端進行交互。

API鏈接：API Reference - Streamlit Docs

pip install streamlit==1.39.0?

導入依賴

import streamlit as st
import requests

?頁面設置

st.set_page_config(page_title="ChatBot", page_icon="🤖", layout="centered")
st.title("🤖 聊天機器人")

?側邊欄配置

with st.sidebar:st.title("ChatBot")sys_prompt = st.text_input("系統提示詞：", value="你是一個有用的助手")history_len=st.slider("保留歷史對話的數量",min_value=-1,max_value=10,value=1,step=1)temperature=st.slider("temperature",min_value=0.01,max_value=2.0,value=0.5,step=0.01)top_p=st.slider("top_p",min_value=0.01,max_value=2.0,value=0.5,step=0.01)max_tokens=st.slider("max_tokens",min_value=256,max_value=4096,value=1024,step=8)stream=st.checkbox("stream",value=True)

?聊天歷史記錄管理

if "history" not in st.session_state:st.session_state.history = []

for messgae in st.session_state.history:with st.chat_message(messgae["role"]):st.markdown(messgae["content"])

?清空聊天記錄

def clear_chat_history():st.session_state.history = []

st.button("清空聊天歷史", on_click=clear_chat_history)

接受用戶輸入并顯示消息?

if prompt := st.chat_input("請輸入你的內容"):with st.chat_message('user'):st.markdown(prompt)

與后端?Fastapi通信

    response = requests.post('http://127.0.0.1:6066/chat', json=data, stream=True)if response.status_code==200:chunks = ""assistant_placeholder=st.chat_message("assistant")assistant_text=assistant_placeholder.markdown("")if stream:  # 流式輸出for chunk in response.iter_content(chunk_size=None, decode_unicode=True):# 處理響應的內容，并累加起來chunks += chunk# 實時顯示和更新助手的消息assistant_text.markdown(chunks)else:for chunk in response.iter_content(chunk_size=None, decode_unicode=True):chunks += chunkassistant_text.markdown(chunks)

四、Fastapi+streamlit

# 導入必要的庫
import streamlit as st  # 用于構建Web應用的庫
import requests  # 用于發送HTTP請求# 設置頁面配置
st.set_page_config(page_title="ChatBot",  # 頁面標題page_icon="🤖",  # 頁面圖標layout="centered"  # 布局方式
)
st.title("🤖 聊天機器人")  # 主標題# 定義清空聊天歷史的函數
def clear_chat_history():st.session_state.history = []  # 重置聊天歷史為空列表# 側邊欄設置
with st.sidebar:st.title("ChatBot")  # 側邊欄標題# 系統提示詞輸入框sys_prompt = st.text_input("系統提示詞：", value="你是一個有用的助手")# 歷史對話輪數滑塊history_len = st.slider("保留歷史對話的數量",min_value=-1,  # -1表示保留全部歷史max_value=10,value=1,step=1)# 溫度參數滑塊temperature = st.slider("temperature",min_value=0.01,max_value=2.0,value=0.5,step=0.01)# top_p參數滑塊top_p = st.slider("top_p",min_value=0.01,max_value=2.0,value=0.5,step=0.01)# 最大token數滑塊max_tokens = st.slider("max_tokens",min_value=256,max_value=4096,value=1024,step=8)# 是否流式輸出的復選框stream = st.checkbox("stream", value=True)# 清空聊天歷史按鈕st.button("清空聊天歷史", on_click=clear_chat_history)# 初始化聊天歷史
if "history" not in st.session_state:st.session_state.history = []  # 如果不存在則創建空列表# 顯示歷史消息
for messgae in st.session_state.history:with st.chat_message(messgae["role"]):  # 根據角色創建消息氣泡st.markdown(messgae["content"])  # 顯示消息內容# 獲取用戶輸入
if prompt := st.chat_input("請輸入你的內容"):# 顯示用戶消息with st.chat_message('user'):st.markdown(prompt)# 準備請求數據data = {"query": prompt,  # 用戶輸入"sys_prompt": sys_prompt,  # 系統提示詞"history_len": history_len,  # 歷史對話輪數"history": st.session_state.history,  # 歷史對話"temperature": temperature,  # 溫度參數"top_p": top_p,  # top_p參數"max_tokens": max_tokens  # 最大token數}# 發送POST請求到FastAPI服務response = requests.post('http://127.0.0.1:6066/chat',  # FastAPI服務地址json=data,  # 請求體數據stream=True  # 啟用流式接收)# 處理成功響應if response.status_code == 200:chunks = ""  # 初始化內容容器# 創建助手消息占位符assistant_placeholder = st.chat_message("assistant")assistant_text = assistant_placeholder.markdown("")  # 初始空內容if stream:  # 流式輸出模式for chunk in response.iter_content(chunk_size=None, decode_unicode=True):# 處理響應的內容，并累加起來chunks += chunk# 實時顯示和更新助手的消息assistant_text.markdown(chunks)else:  # 非流式模式for chunk in response.iter_content(chunk_size=None, decode_unicode=True):chunks += chunkassistant_text.markdown(chunks)# 將對話添加到歷史記錄st.session_state.history.append({"role": "user", "content": prompt})st.session_state.history.append({"role": "assistant", "content": chunks})