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})