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

一个与运行 Qwen3 大语言模型的 vLLM 服务进行通信的客户端程序

news 2025/7/12 6:11:57

一个与运行 Qwen3 大语言模型的 vLLM 服务进行通信的客户端程序

flyfish

代码

import argparse
import json
import time
import requests
from typing import Iterator, Dict, Any, Optionalclass Qwen3Client:"""用于与运行Qwen3的vLLM服务通信的客户端，兼容OpenAI API格式"""def __init__(self, base_url: str = "http://localhost:8000"):"""初始化客户端Args:base_url: vLLM服务的基础URL"""self.base_url = base_urlself.headers = {"Content-Type": "application/json"}def chat_completions(self,messages: list,model: str = "Qwen/Qwen3-8B",temperature: float = 0.7,top_p: float = 0.8,top_k: int = 20,max_tokens: int = 8192,presence_penalty: float = 1.5,frequency_penalty: float = 0.0,stream: bool = False,enable_thinking: bool = False,stop: Optional[list] = None,n: int = 1,best_of: int = 1,logprobs: Optional[int] = None,echo: bool = False,suffix: Optional[str] = None,logit_bias: Optional[Dict[str, float]] = None,user: Optional[str] = None) -> Dict[str, Any] | Iterator[Dict[str, Any]]:"""发送聊天完成请求，兼容OpenAI API格式Args:messages: 消息列表，格式为[{"role": "user", "content": "..."}, ...]model: 模型名称temperature: 温度参数，控制随机性 (0.0-2.0)top_p: Top-p采样参数，控制多样性top_k: Top-k采样参数，控制候选token数量max_tokens: 最大生成token数presence_penalty: 存在惩罚系数，抑制重复话题frequency_penalty: 频率惩罚系数，抑制重复词语stream: 是否使用流式响应enable_thinking: 是否启用思考过程stop: 停止生成的token列表n: 生成的回复数量best_of: 生成多个回复并选择最佳的logprobs: 返回的log概率数量echo: 是否回显输入suffix: 生成文本的后缀logit_bias: token概率偏置user: 用户标识Returns:如果stream为False，返回完整响应字典如果stream为True，返回响应生成器"""endpoint = "/v1/chat/completions"url = f"{self.base_url}{endpoint}"payload = {"model": model,"messages": messages,"temperature": temperature,"top_p": top_p,"top_k": top_k,"max_tokens": max_tokens,"presence_penalty": presence_penalty,"frequency_penalty": frequency_penalty,"stream": stream,"chat_template_kwargs": {"enable_thinking": enable_thinking},}# 添加可选参数if stop is not None:payload["stop"] = stopif n != 1:payload["n"] = nif best_of != 1:payload["best_of"] = best_ofif logprobs is not None:payload["logprobs"] = logprobsif echo:payload["echo"] = echoif suffix is not None:payload["suffix"] = suffixif logit_bias is not None:payload["logit_bias"] = logit_biasif user is not None:payload["user"] = userresponse = requests.post(url, headers=self.headers, json=payload)if response.status_code != 200:raise Exception(f"请求失败: {response.status_code}, {response.text}")if stream:return self._parse_stream_response(response)else:return response.json()def _parse_stream_response(self, response) -> Iterator[Dict[str, Any]]:"""解析流式响应Args:response: 请求响应对象Yields:每次生成的文本块的字典"""for line in response.iter_lines():if line:# 移除SSE前缀line = line.decode('utf-8')if line.startswith('data: '):line = line[6:]# 处理特殊的结束标记if line == '[DONE]':breaktry:chunk = json.loads(line)yield chunkexcept json.JSONDecodeError as e:print(f"解析流数据时出错: {e}, 数据: {line}")def main():"""主函数，用于命令行测试客户端"""parser = argparse.ArgumentParser(description='Qwen3 vLLM客户端')group = parser.add_mutually_exclusive_group(required=True)group.add_argument('--prompt', type=str, help='用户输入提示')group.add_argument('--prompt_file', type=str, help='包含提示内容的文本文件路径')parser.add_argument('--model', type=str, default="Qwen/Qwen3-8B", help='模型名称')parser.add_argument('--max_tokens', type=int, default=8192, help='最大生成token数')parser.add_argument('--temperature', type=float, default=0.7, help='温度参数')parser.add_argument('--top_p', type=float, default=0.8, help='Top-p参数')parser.add_argument('--top_k', type=int, default=20, help='Top-k参数')parser.add_argument('--presence_penalty', type=float, default=1.5, help='存在惩罚系数')parser.add_argument('--frequency_penalty', type=float, default=0.0, help='频率惩罚系数')parser.add_argument('--stream', action='store_true', help='是否使用流式输出')parser.add_argument('--enable_thinking', action='store_true', help='是否启用思考过程')parser.add_argument('--server_url', type=str, default='http://localhost:8000', help='服务器URL')args = parser.parse_args()# 从文件或命令行获取promptif args.prompt_file:try:with open(args.prompt_file, 'r', encoding='utf-8') as f:prompt = f.read()print(f"从文件 {args.prompt_file} 读取提示内容")except Exception as e:print(f"读取文件失败: {e}")returnelse:prompt = args.promptclient = Qwen3Client(args.server_url)print(f"发送请求到 {args.server_url}")print(f"提示内容长度: {len(prompt)} 字符")messages = [{"role": "user", "content": prompt}]start_time = time.time()if args.stream:print("流式响应:")for chunk in client.chat_completions(messages=messages,model=args.model,temperature=args.temperature,top_p=args.top_p,top_k=args.top_k,max_tokens=args.max_tokens,presence_penalty=args.presence_penalty,frequency_penalty=args.frequency_penalty,stream=True,enable_thinking=args.enable_thinking):# 提取delta内容if 'choices' in chunk and len(chunk['choices']) > 0:delta = chunk['choices'][0].get('delta', {})content = delta.get('content', '')print(content, end='', flush=True)print()else:result = client.chat_completions(messages=messages,model=args.model,temperature=args.temperature,top_p=args.top_p,top_k=args.top_k,max_tokens=args.max_tokens,presence_penalty=args.presence_penalty,frequency_penalty=args.frequency_penalty,stream=False,enable_thinking=args.enable_thinking)print("完整响应:")# 提取回复内容if 'choices' in result and len(result['choices']) > 0:message = result['choices'][0].get('message', {})print(message.get('content', ''))print(f"耗时: {time.time() - start_time:.2f}秒")if __name__ == "__main__":main()

用法

1. 基础用法：直接输入短提示

python vllm_client.py --prompt "解释量子计算的基本原理"

效果：使用默认参数调用 Qwen3-8B 模型，以非流式方式输出回复。

2. 从文件读取长提示

假设有一个 document.txt 文件包含技术文档内容：

python vllm_client.py --prompt_file ./document.txt --stream

效果：

从文件读取完整内容作为提示
使用流式输出，边生成边显示结果

3. 调整生成参数（创造性文本）

python vllm_client.py \--prompt "以《星际穿越》为灵感写一篇科幻小说" \--temperature 1.0 \--max_tokens 2048 \--stream

参数说明：

temperature=1.0：增加随机性，使生成更具创造性
max_tokens=2048：允许更长的输出
--stream：实时查看生成的小说内容

4. 调整生成参数（精确问答）

python vllm_client.py \--prompt "Python中如何实现多线程编程？给出代码示例" \--temperature 0.2 \--top_p 0.9 \--presence_penalty 1.8

参数说明：

temperature=0.2：降低随机性，使回答更确定性
presence_penalty=1.8：抑制无关内容，专注于问题本身

5. 连接远程服务器

python vllm_client.py \--prompt "列出中国古代四大发明" \--server_url http://192.168.1.100:8000

效果：连接到局域网内另一台机器上运行的 vLLM 服务

6. 启用思考过程（如果模型支持）

python vllm_client.py \--prompt "如何设计一个高并发的 Web 服务器？请分步说明" \--enable_thinking \--stream

效果：模型可能会先展示思考步骤，再给出最终答案

7. 长文本生成（如论文摘要）

python vllm_client.py \--prompt_file ./research_paper.txt \--model Qwen/Qwen3-14B \--max_tokens 4096 \--temperature 0.7

参数说明：

使用更大的 Qwen3-14B 模型
允许生成更长的摘要（最多 4096 tokens）

8. 批量处理多个提示

编写一个简单的 shell 脚本循环调用客户端：

#!/bin/bash# 定义提示列表
prompts=("解释区块链技术的工作原理""简述人工智能的发展历程""对比 Python 和 Java 的优缺点"
)# 循环处理每个提示
for prompt in "${prompts[@]}"; doecho "=== 处理提示: $prompt ==="python vllm_client.py --prompt "$prompt" --streamecho "=== 完成 ==="echo
done

参数说明

参数名称	类型	默认值	是否必选	说明
互斥输入参数
`--prompt`	str	无	二选一	直接在命令行输入的提示文本（适用于短内容）
`--prompt_file`	str	无	二选一	包含提示文本的文件路径（适用于长内容，如文档、多段文字）
模型配置参数
`--model`	str	“Qwen/Qwen3-8B”	可选	调用的模型名称，需与vLLM服务端加载的模型一致（如Qwen3-14B等）
生成控制参数
`--max_tokens`	int	8192	可选	模型生成的最大token数量，限制回复长度（token≈汉字1-2个）
`--temperature`	float	0.7	可选	控制输出随机性：值越小（如0.1）输出越确定；值越大（如1.0）越灵活
`--top_p`	float	0.8	可选	核采样参数（0.0-1.0），只保留概率累积和达到该值的token，控制多样性
`--top_k`	int	20	可选	只从概率最高的k个token中采样，值越小输出越集中
`--presence_penalty`	float	1.5	可选	惩罚未在输入中出现的新内容（值越大，越抑制无关话题）
`--frequency_penalty`	float	0.0	可选	惩罚重复用词（值越大，越鼓励用词多样性，如0.5-1.0）
输出模式参数
`--stream`	开关参数	无（默认关闭）	可选	启用流式输出（边生成边显示，适合长回复实时查看）
`--enable_thinking`	开关参数	无（默认关闭）	可选	启用模型“思考过程”（若模型支持，会先展示推理步骤再给结论）
连接配置参数
`--server_url`	str	“http://localhost:8000”	可选	vLLM服务端地址（本地服务默认此值，远程服务需填对应IP:端口）