基于vllm和gradio的大模型问答

一、环境（一定要按照以下的版本，惨痛的教训）

1. vllm==0.6.1（建议使用uv pip安装）

2. gradio==5.49.1（建议使用uv pip安装）

3. python==3.12.0（建议使用conda安装虚拟环境）

二、vllm后端

1. 代码如下：（怎么下载Qwen2.5模型，参考上一篇文章）

import os import uvicorn import time from fastapi import FastAPI, HTTPException from fastapi.responses import StreamingResponse from pydantic import BaseModel from typing import List, Dict, Generator # 环境变量配置 os.environ["HUGGINGFACE_HUB_CACHE"] = r"/home/plc/cache" os.environ["HF_ENDPOINT"] = "https://hf-mirror.com" # 导入库 from vllm import LLM, SamplingParams from transformers import AutoTokenizer # 加载分词器 model_name = "Qwen/Qwen2.5-3B" tokenizer = AutoTokenizer.from_pretrained( model_name, trust_remote_code=True, mirror="https://hf-mirror.com" ) # 加载模型（优化重复控制） print("正在加载Qwen2.5模型...") llm = LLM( model=model_name, trust_remote_code=True, dtype="float16", gpu_memory_utilization=0.8, max_num_batched_tokens=4096, max_model_len=2048, max_num_seqs=2, disable_log_stats=True, download_dir=os.environ["HUGGINGFACE_HUB_CACHE"], seed=42, ) print("模型加载完成！") # FastAPI应用 app = FastAPI(title="vllm 0.6.1 大模型服务") # 请求体 class ChatRequest(BaseModel): messages: List[Dict[str, str]] # 流式聊天接口（修复重复问题） @app.post("/chat/stream") def chat_stream(request: ChatRequest) -> StreamingResponse: if not request.messages: raise HTTPException(status_code=400, detail="messages不能为空") # 构建Prompt（增加“不重复”提示） # 关键：在system prompt中明确要求“不要生成重复内容” system_prompt = request.messages[0]["content"] if request.messages[0]["role"] == "system" else "" system_prompt += "\n注意：输出内容需连贯，禁止生成重复、无意义的字符或句子。代码部分之间需要换行输出" # 重新构造messages，更新system prompt new_messages = [{"role": "system", "content": system_prompt}] + request.messages[1:] # 应用Qwen模板 prompt = tokenizer.apply_chat_template( new_messages, tokenize=False, add_generation_prompt=True, ) print(f"最终Prompt：\n{prompt}") # 通用问答最优配置（平衡稳定性+多样性+无重复） sampling_params = SamplingParams( max_tokens=800, # 3B模型无需过长输出，800足够覆盖绝大多数问答场景 temperature=0.6, # 0.6是3B模型的黄金值，兼顾稳定和少量多样性 top_p=0.9, # 配合temperature，过滤低概率词汇，提升输出质量 repetition_penalty=1.15, # 3B模型最优重复惩罚（1.1~1.2区间） skip_special_tokens=True, stop=["<<|im_end|>", "</s>", "\n\n\n"], # 保留有效停止标记，新增连续换行标记 seed=None, # 生产环境设为None，测试时可固定（如42） ) # 定义生成器（增加重复检测） def generate_stream() -> Generator[str, None, None]: try: outputs = llm.generate(prompts=[prompt], sampling_params=sampling_params) generated_text = outputs[0].outputs[0].text.strip() # 重复内容检测：如果出现连续重复字符串，截断 if "GSGSG" in generated_text: generated_text = generated_text.split("GSGSG")[0].strip() # 检测连续重复的字符（如“aaaaa”） if len(generated_text) > 10: last_char = generated_text[-1] if generated_text[-5:] == last_char*5: generated_text = generated_text[:-5].strip() # 流式返回 for char in generated_text: yield f"data: {char}\n\n" time.sleep(0.005) yield "data: [DONE]\n\n" except Exception as e: yield f"data: ERROR: {str(e)}\n\n" return StreamingResponse( generate_stream(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", "Connection": "keep-alive", "Content-Type": "text/event-stream; charset=utf-8" } ) # 启动服务 if __name__ == "__main__": uvicorn.run( app, host="0.0.0.0", port=8001, reload=False, workers=1 )

三、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.strip()}] for user_msg, assistant_msg in history: if user_msg and user_msg.strip(): messages.append({"role": "user", "content": user_msg.strip()}) if assistant_msg and assistant_msg.strip(): messages.append({"role": "assistant", "content": assistant_msg.strip()}) messages.append({"role": "user", "content": message.strip()}) try: # 发送流式POST请求（优化超时和连接配置） response = requests.post( API_STREAM_URL, json={"messages": messages}, stream=True, timeout=300, # 延长超时时间 headers={"Content-Type": "application/json"} ) if response.status_code != 200: history.append((message, f"请求失败：状态码{response.status_code}，{response.text[:200]}")) yield history return # 逐行接收流式数据（优化解析逻辑） full_response = "" history.append((message, "")) # 初始化空回复 for line in response.iter_lines(chunk_size=1, decode_unicode=True): if not line: continue # 严格解析SSE格式 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 # 逐字符追加内容 if content: full_response += content history[-1] = (message, full_response) yield history # 实时更新界面 time.sleep(0.005) # 优化刷新频率 except requests.exceptions.Timeout: history.append((message, "请求超时（超过5分钟），请简化问题重试")) yield history except requests.exceptions.ConnectionError: history.append((message, f"无法连接到后端服务：{API_STREAM_URL}\n请检查后端是否启动")) yield history except Exception as e: history.append((message, f"调用出错：{str(e)}")) yield history # 创建Gradio界面 with gr.Blocks(title="工控大模型（流式版）") as demo: gr.Markdown( """ # 工控大模型 基于Qwen模型微调，支持流式输出的PLC代码生成工具 """ ) # 聊天界面 chatbot = gr.Chatbot( label="问答对话（流式输出）", height=600, bubble_full_width=False, ) # 输入框 msg = gr.Textbox( label="请输入你的问题", placeholder="例如：编写一段西门子S7-1200的电机启停PLC程序", lines=4 ) # 按钮区域 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 ) # 启动前端 if __name__ == "__main__": demo.queue(max_size=10) # 增加队列大小 demo.launch( server_name="0.0.0.0", server_port=7860, share=False, inbrowser=True, show_error=True # 显示错误信息便于调试 )

2. 为什么使用Qwen2系列呢？因为vllm的版本偏低。