当前位置：首页 > news >正文

gradio快速部署大模型进行问答

news 2026/6/12 7:22:09

一、模型准备

1. 下载模型：Qwen3-4B-Instruct-2507

import os # 1. 优先设置环境变量（必须在导入transformers之前！） os.environ["HUGGINGFACE_HUB_CACHE"] = r"D:\plc\cache" os.environ["HF_ENDPOINT"] = "https://hf-mirror.com" # 补充额外配置，确保镜像源生效 os.environ["TRANSFORMERS_OFFLINE"] = "0" # 关闭离线模式 os.environ["HF_HUB_OFFLINE"] = "0" # 2. 环境变量设置完成后，再导入transformers库 from transformers import AutoModelForCausalLM, AutoTokenizer model_name = "Qwen/Qwen3-4B-Instruct-2507" # load the tokenizer and the model tokenizer = AutoTokenizer.from_pretrained( model_name, trust_remote_code=True, # Qwen模型需要开启该参数 mirror="https://hf-mirror.com" # 显式指定镜像源，双重保障 ) model = AutoModelForCausalLM.from_pretrained( model_name, torch_dtype="auto", device_map="auto", trust_remote_code=True, # Qwen模型需要开启该参数 mirror="https://hf-mirror.com" # 显式指定镜像源 ) # prepare the model input prompt = "Give me a short introduction to large language model." messages = [ {"role": "user", "content": prompt} ] text = tokenizer.apply_chat_template( messages, tokenize=False, add_generation_prompt=True, ) model_inputs = tokenizer([text], return_tensors="pt").to(model.device) # conduct text completion generated_ids = model.generate( **model_inputs, max_new_tokens=16384, do_sample=True, # 增加采样参数，让输出更自然 temperature=0.7 # 温度参数，控制输出随机性 ) output_ids = generated_ids[0][len(model_inputs.input_ids[0]):].tolist() content = tokenizer.decode(output_ids, skip_special_tokens=True) print("content:", content)

2. 正常下载时，显示以下内容：

二、基于 FastAPI 的大模型后端服务（流式输出）

1. 代码如下：

import os import torch import uvicorn import threading from fastapi import FastAPI, HTTPException from fastapi.responses import StreamingResponse from pydantic import BaseModel from typing import List, Dict, Any, Generator # 1. 环境变量配置 os.environ["HUGGINGFACE_HUB_CACHE"] = r"D:\plc\cache" os.environ["HF_ENDPOINT"] = "https://hf-mirror.com" os.environ["TRANSFORMERS_OFFLINE"] = "0" os.environ["HF_HUB_OFFLINE"] = "0" # 2. 导入模型相关库 from transformers import AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer # 3. 加载模型和分词器 model_name = "Qwen/Qwen3-4B-Instruct-2507" print("正在加载Qwen3-4B模型...") tokenizer = AutoTokenizer.from_pretrained( model_name, trust_remote_code=True, mirror="https://hf-mirror.com" ) # 补充pad_token_id if tokenizer.pad_token_id is None: tokenizer.pad_token_id = tokenizer.eos_token_id model = AutoModelForCausalLM.from_pretrained( model_name, dtype=torch.bfloat16 if torch.cuda.is_available() else torch.float32, device_map="cuda:0" if torch.cuda.is_available() else "cpu", trust_remote_code=True, mirror="https://hf-mirror.com", load_in_4bit=False, # 显存不足可改为True low_cpu_mem_usage=True ) print("模型加载完成！") # 4. 创建FastAPI应用 app = FastAPI(title="本地大模型流式服务") # 定义请求体格式 class ChatRequest(BaseModel): messages: List[Dict[str, str]] # 5. 流式聊天接口（核心修复：同步遍历streamer） @app.post("/chat/stream") def chat_stream(request: ChatRequest) -> StreamingResponse: # 验证输入 if not request.messages: raise HTTPException(status_code=400, detail="messages列表不能为空") # 构建输入文本 text = tokenizer.apply_chat_template( request.messages, tokenize=False, add_generation_prompt=True, ) # 编码输入 model_inputs = tokenizer([text], return_tensors="pt").to(model.device) input_ids = model_inputs.input_ids attention_mask = model_inputs.attention_mask # 创建流式输出器（设置超时时间） streamer = TextIteratorStreamer( tokenizer, skip_prompt=True, skip_special_tokens=True, timeout=120.0, decode_kwargs={"skip_special_tokens": True} ) # 定义生成器函数（核心：同步遍历streamer，生成SSE格式数据） def generate_stream() -> Generator[str, None, None]: try: # 定义模型生成函数（在后台线程运行） def run_generate(): model.generate( input_ids=input_ids, attention_mask=attention_mask, streamer=streamer, max_new_tokens=2048, do_sample=True, temperature=0.7, top_p=0.85, repetition_penalty=1.1, eos_token_id=tokenizer.eos_token_id, pad_token_id=tokenizer.pad_token_id, max_length=None, num_return_sequences=1 ) # 启动后台线程执行模型生成 generate_thread = threading.Thread(target=run_generate) generate_thread.start() # 同步遍历streamer（关键修复：替换async for为普通for） for token in streamer: if token: # 按SSE格式返回token yield f"data: {token}\n\n" # 等待生成线程结束 generate_thread.join() # 发送结束标记 yield "data: [DONE]\n\n" except Exception as e: # 发送错误信息 yield f"data: ERROR:{str(e)}\n\n" # 返回流式响应（SSE格式） return StreamingResponse( generate_stream(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", # 禁用nginx缓冲，确保实时输出 "Connection": "keep-alive" } ) # 6. 启动服务 if __name__ == "__main__": uvicorn.run( app, host="0.0.0.0", port=8001, reload=False, workers=1 )

2. 模型成功加载后，显示：

三、基于 Gradio 的前端交互界面（流式输出）

1. 代码如下：

import gradio as gr import requests import time # API流式服务地址 API_STREAM_URL = "http://localhost:8001/chat/stream" # 工控大模型系统提示词 SYSTEM_PROMPT = """ 你是一名PLC代码专家，请参考给出的标准功能块信息和仿照示例代码,生成符合IEC61131-3标准和plcopen标准的功能代码,要求代码尽可能详细，不要使用while循环。 示例代码: PROGRAM PLC_PRG VAR i :INT; mcPower :MC_Power; mcMoveAbs :MC_MoveAbsolute; mcReset :MC_Reset; mcReadActPos :MC_ReadActualTorque; MoveExe: BOOL; END_VAR IF DIO THEN MoveExe := TRUE; END_IF mcPower( Axis:= Axis1, Enable:= , bRegulatorOn:= , bDriveStart:= , Status=> , bRegulatorRealState=> , bDriveStartRealState=> , Busy=> , Error=> , ErrorID=> ); 输出时请使用代码块格式 """ # 流式调用API的生成器函数 def call_chat_api_stream(message, history): """ 流式接收后端响应，逐字更新界面 :param message: 当前用户输入 :param history: Gradio历史对话 [(用户1, 助手1), ...] :return: 生成器，逐次返回更新后的历史 """ # 构建完整的messages列表 messages = [{"role": "system", "content": SYSTEM_PROMPT}] for user_msg, assistant_msg in history: if user_msg: messages.append({"role": "user", "content": user_msg}) if assistant_msg: messages.append({"role": "assistant", "content": assistant_msg}) messages.append({"role": "user", "content": message}) try: # 发送流式POST请求 response = requests.post( API_STREAM_URL, json={"messages": messages}, stream=True, timeout=120 ) if response.status_code != 200: history.append((message, f"请求失败：状态码{response.status_code}，{response.text}")) yield history return # 逐行接收流式数据 full_response = "" history.append((message, "")) # 先添加空的助手回复，后续逐字更新 for line in response.iter_lines(): if not line: continue # 解析SSE格式数据 line = line.decode("utf-8") if line.startswith("data: "): content = line[6:].strip() # 处理结束标记 if content == "[DONE]": break # 处理错误信息 if content.startswith("ERROR:"): history[-1] = (message, f"生成错误：{content[6:]}") yield history return # 逐字追加内容并更新界面 full_response += content history[-1] = (message, full_response) yield history # 实时返回更新后的历史 time.sleep(0.01) # 轻微延迟，避免界面刷新过快 except requests.exceptions.Timeout: history.append((message, "请求超时（超过120秒），请简化问题重试")) yield history except requests.exceptions.ConnectionError: history.append((message, f"无法连接到后端服务：{API_STREAM_URL}")) yield history except Exception as e: history.append((message, f"调用出错：{str(e)}")) yield history # 创建Gradio界面（适配低版本Gradio的流式输出） with gr.Blocks(title="工控大模型（流式版）") as demo: gr.Markdown( """ # 工控大模型 基于Qwen模型微调，支持流式输出的PLC代码生成工具 """ ) # 聊天界面 chatbot = gr.Chatbot( label="问答对话（流式输出）", height=500, bubble_full_width=False, avatar_images=(None, "https://img.alicdn.com/imgextra/i4/O1CN01nqE6sX1rH6H99Q759_!!6000000005599-2-tps/200-200.png") ) # 输入框 msg = gr.Textbox( label="请输入你的问题", placeholder="例如：编写一段西门子S7-1200的电机启停PLC程序", lines=3 ) # 按钮区域 with gr.Row(): submit_btn = gr.Button("发送", variant="primary") clear_btn = gr.Button("清空对话") # 绑定事件（适配流式生成器） submit_btn.click( fn=call_chat_api_stream, inputs=[msg, chatbot], outputs=chatbot, queue=True # 开启队列，支持流式更新 ).then( fn=lambda: "", outputs=msg # 发送后清空输入框 ) msg.submit( # 回车发送 fn=call_chat_api_stream, inputs=[msg, chatbot], outputs=chatbot, queue=True ).then( fn=lambda: "", outputs=msg ) clear_btn.click( fn=lambda: [], outputs=chatbot ).then( fn=lambda: "", outputs=msg ) # 启动前端（适配低版本Gradio的队列配置） if __name__ == "__main__": # 核心修复：移除concurrency_count参数，适配低版本Gradio demo.queue() # 仅开启队列即可，无需额外参数 demo.launch( server_name="0.0.0.0", server_port=7860, share=False, inbrowser=True )

2. 成功运行后，打开url：工控大模型（流式版）

3. 项目文件组织架构：

查看全文

http://www.gsyq.cn/news/88534.html