通义千问1.8B-GPTQ-Int4代码实例：Python调用vLLM API与Chainlit集成详解

本文介绍了如何在星图GPU平台上自动化部署通义千问1.5-1.8B-Chat-GPTQ-Int4镜像，并详细讲解了通过Python调用vLLM API与Chainlit框架集成，快速构建一个交互式Web对话应用。该方案能帮助开发者便捷地将大语言模型转化为实际可用的智能对话助手。

江卓尔

207人浏览 · 2026-03-28 05:23:35

江卓尔 · 2026-03-28 05:23:35 发布

通义千问1.8B-GPTQ-Int4代码实例：Python调用vLLM API与Chainlit集成详解

1. 引言：从模型部署到交互式应用

你刚刚在服务器上成功部署了通义千问1.8B的GPTQ-Int4量化模型，看着日志里显示“模型加载成功”的字样，心里一阵激动。但接下来呢？怎么才能让这个模型真正“活”起来，变成一个可以对话、可以测试、可以展示给别人的应用？

这就是我们今天要解决的问题。很多朋友在部署完模型后，往往卡在“下一步怎么用”这个环节。模型跑起来了，但只能通过命令行测试，或者写个简单的Python脚本调用，离一个完整的应用还有距离。

本文将带你一步步实现从基础API调用到完整前端集成的全过程。我们会用Python编写简洁的代码，通过vLLM提供的API接口调用模型，再用Chainlit构建一个美观的Web界面。整个过程就像搭积木一样简单，即使你是Python新手，也能跟着做出来。

学完本文，你将掌握：

如何用Python代码调用vLLM部署的模型
如何构建一个完整的对话应用前端
如何调试和优化你的模型调用
如何将技术部署转化为实际可用的产品

让我们开始吧，把那个在后台默默运行的模型，变成一个可以对话的智能助手。

2. 环境准备与模型确认

在开始编写代码之前，我们需要先确认几件事情。这就像做菜前要准备好食材和厨具一样，准备工作做得好，后面才会顺利。

2.1 确认模型服务状态

首先，确保你的模型已经通过vLLM成功部署。打开终端，运行以下命令查看服务状态：

# 查看模型服务日志
cat /root/workspace/llm.log

你应该能看到类似这样的输出：

INFO 07-15 14:30:25 llm_engine.py:72] Initializing an LLM engine...
INFO 07-15 14:30:30 llm_engine.py:150] Loading model weights...
INFO 07-15 14:31:15 llm_engine.py:180] Model loaded successfully.
INFO 07-15 14:31:16 api_server.py:150] Starting API server on http://0.0.0.0:8000

关键信息是最后两行：

Model loaded successfully. - 模型加载成功
Starting API server on http://0.0.0.0:8000 - API服务在8000端口启动

如果看到这些信息，恭喜你，模型服务已经正常运行了。

2.2 安装必要的Python包

接下来，我们需要安装几个Python包。打开一个新的终端窗口，或者在你打算编写代码的环境中，执行以下命令：

# 安装requests库，用于发送HTTP请求
pip install requests

# 安装chainlit库，用于构建Web界面
pip install chainlit

# 如果需要，也可以安装openai库（vLLM兼容OpenAI API格式）
pip install openai

这些包都很小，安装很快。requests是我们调用API的基础工具，chainlit是构建界面的框架，openai库是可选的，但它提供了更符合习惯的调用方式。

2.3 了解vLLM的API接口

vLLM提供了一个兼容OpenAI格式的API接口，这意味着我们可以用类似调用ChatGPT API的方式来调用我们自己的模型。主要用到的接口有两个：

聊天补全接口 - /v1/chat/completions
- 用于对话场景，支持多轮对话
- 输入格式是消息列表（system、user、assistant角色）
补全接口 - /v1/completions
- 用于单轮文本生成
- 输入是简单的提示文本

我们今天主要使用第一个接口，因为它更适合构建对话应用。

准备工作就绪，接下来我们开始编写真正的代码。

3. 基础API调用：让模型开口说话

现在进入实战环节。我们将从最简单的代码开始，逐步构建完整的应用。别担心代码复杂，我会把每一步都解释清楚。

3.1 最简化的API调用示例

我们先写一个最简单的Python脚本，测试模型是否能正常响应。创建一个新文件，命名为 test_api.py：

import requests
import json

# vLLM API服务器的地址
API_URL = "http://localhost:8000/v1/chat/completions"

# 准备请求数据
payload = {
    "model": "Qwen1.5-1.8B-Chat-GPTQ-Int4",  # 模型名称，与部署时一致
    "messages": [
        {
            "role": "user",  # 用户角色
            "content": "你好，请介绍一下你自己"  # 用户消息
        }
    ],
    "max_tokens": 100,  # 最大生成token数
    "temperature": 0.7,  # 温度参数，控制随机性
    "stream": False  # 是否流式输出
}

# 发送请求
try:
    response = requests.post(API_URL, json=payload)
    response.raise_for_status()  # 检查请求是否成功
    
    # 解析响应
    result = response.json()
    
    # 提取模型回复
    reply = result["choices"][0]["message"]["content"]
    print("模型回复：")
    print(reply)
    
except requests.exceptions.RequestException as e:
    print(f"请求失败：{e}")
except KeyError as e:
    print(f"解析响应失败：{e}")
    print(f"完整响应：{response.text}")

保存文件后，在终端运行：

python test_api.py

如果一切正常，你应该能看到模型的自我介绍。这个简单的脚本验证了几个关键点：

API服务可访问
模型能正常响应
数据传输格式正确

3.2 添加错误处理和超时设置

实际应用中，网络可能不稳定，服务可能暂时不可用。我们需要增强代码的健壮性：

import requests
import json
import time

def call_model_with_retry(prompt, max_retries=3, retry_delay=2):
    """
    带重试机制的模型调用函数
    
    参数：
    prompt: 用户输入的问题
    max_retries: 最大重试次数
    retry_delay: 重试间隔（秒）
    """
    API_URL = "http://localhost:8000/v1/chat/completions"
    
    payload = {
        "model": "Qwen1.5-1.8B-Chat-GPTQ-Int4",
        "messages": [{"role": "user", "content": prompt}],
        "max_tokens": 200,
        "temperature": 0.7,
        "stream": False
    }
    
    headers = {
        "Content-Type": "application/json"
    }
    
    for attempt in range(max_retries):
        try:
            # 设置超时，避免长时间等待
            response = requests.post(
                API_URL, 
                json=payload, 
                headers=headers,
                timeout=30  # 30秒超时
            )
            
            if response.status_code == 200:
                result = response.json()
                return result["choices"][0]["message"]["content"]
            else:
                print(f"请求失败，状态码：{response.status_code}")
                print(f"响应内容：{response.text}")
                
        except requests.exceptions.Timeout:
            print(f"第{attempt + 1}次请求超时")
        except requests.exceptions.ConnectionError:
            print(f"第{attempt + 1}次连接失败")
        except Exception as e:
            print(f"第{attempt + 1}次请求异常：{e}")
        
        # 如果不是最后一次尝试，等待后重试
        if attempt < max_retries - 1:
            print(f"等待{retry_delay}秒后重试...")
            time.sleep(retry_delay)
    
    return "抱歉，模型暂时无法响应，请稍后再试。"

# 测试函数
if __name__ == "__main__":
    test_prompts = [
        "你好，你是谁？",
        "Python是什么？",
        "写一个简单的Hello World程序"
    ]
    
    for prompt in test_prompts:
        print(f"\n用户：{prompt}")
        print("-" * 40)
        reply = call_model_with_retry(prompt)
        print(f"模型：{reply}")
        time.sleep(1)  # 避免请求过快

这个版本增加了重试机制和超时设置，在实际应用中更加可靠。你可以根据需要调整重试次数和超时时间。

3.3 支持多轮对话

真正的对话应用需要记住上下文。让我们升级代码，支持连续对话：

import requests
import json

class QwenChatClient:
    """通义千问聊天客户端"""
    
    def __init__(self, api_url="http://localhost:8000/v1/chat/completions"):
        self.api_url = api_url
        self.model_name = "Qwen1.5-1.8B-Chat-GPTQ-Int4"
        self.conversation_history = []  # 存储对话历史
        
    def add_system_message(self, system_prompt):
        """添加系统提示，设定助手的行为"""
        self.conversation_history.append({
            "role": "system",
            "content": system_prompt
        })
    
    def chat(self, user_message, max_tokens=200, temperature=0.7):
        """发送消息并获取回复"""
        
        # 添加用户消息到历史
        self.conversation_history.append({
            "role": "user",
            "content": user_message
        })
        
        # 准备请求
        payload = {
            "model": self.model_name,
            "messages": self.conversation_history,
            "max_tokens": max_tokens,
            "temperature": temperature,
            "stream": False
        }
        
        try:
            response = requests.post(
                self.api_url,
                json=payload,
                headers={"Content-Type": "application/json"},
                timeout=30
            )
            
            if response.status_code == 200:
                result = response.json()
                assistant_reply = result["choices"][0]["message"]["content"]
                
                # 添加助手回复到历史
                self.conversation_history.append({
                    "role": "assistant",
                    "content": assistant_reply
                })
                
                return assistant_reply
            else:
                return f"请求失败：{response.status_code}"
                
        except Exception as e:
            return f"调用出错：{str(e)}"
    
    def clear_history(self):
        """清空对话历史"""
        self.conversation_history = []
    
    def get_history(self):
        """获取对话历史"""
        return self.conversation_history.copy()

# 使用示例
if __name__ == "__main__":
    # 创建客户端
    client = QwenChatClient()
    
    # 设置系统提示（可选）
    client.add_system_message("你是一个有帮助的AI助手，用中文回答问题。")
    
    # 开始对话
    print("开始对话（输入'退出'结束）")
    print("=" * 50)
    
    while True:
        user_input = input("\n你：")
        
        if user_input.lower() in ["退出", "exit", "quit"]:
            print("对话结束")
            break
        
        # 获取回复
        reply = client.chat(user_input)
        print(f"\n助手：{reply}")
        
        # 可选：查看当前对话历史
        # history = client.get_history()
        # print(f"历史记录：{history}")

这个类封装了完整的对话逻辑，支持：

多轮对话（自动维护历史）
系统提示设置
历史记录管理
错误处理

现在我们已经有了一个功能完整的Python客户端。接下来，让我们给它加上一个漂亮的界面。

4. Chainlit集成：构建Web对话界面

有了能工作的API客户端，现在是时候给它穿上"外衣"了。Chainlit是一个专门为AI应用设计的Python框架，可以快速构建交互式Web界面。它比写HTML/CSS/JavaScript简单得多。

4.1 创建基本的Chainlit应用

首先，创建一个新文件 app.py，这是我们的主应用文件：

import chainlit as cl
import requests
import json
from typing import Optional

# 配置vLLM API地址
VLLM_API_URL = "http://localhost:8000/v1/chat/completions"
MODEL_NAME = "Qwen1.5-1.8B-Chat-GPTQ-Int4"

class ChatManager:
    """聊天管理器，处理对话逻辑"""
    
    def __init__(self):
        self.conversation_history = []
        self.system_prompt = """你是一个有帮助的AI助手，用中文回答问题。
        请保持回答简洁、准确、有帮助。如果不知道答案，请诚实说明。"""
        
        # 初始化时添加系统提示
        self.conversation_history.append({
            "role": "system",
            "content": self.system_prompt
        })
    
    def send_message(self, user_message: str) -> str:
        """发送消息到模型并获取回复"""
        
        # 添加用户消息
        self.conversation_history.append({
            "role": "user",
            "content": user_message
        })
        
        # 准备请求数据
        payload = {
            "model": MODEL_NAME,
            "messages": self.conversation_history,
            "max_tokens": 300,
            "temperature": 0.7,
            "stream": False
        }
        
        headers = {
            "Content-Type": "application/json"
        }
        
        try:
            # 发送请求
            response = requests.post(
                VLLM_API_URL,
                json=payload,
                headers=headers,
                timeout=60  # 生成可能需要时间
            )
            
            if response.status_code == 200:
                result = response.json()
                assistant_reply = result["choices"][0]["message"]["content"]
                
                # 添加助手回复到历史
                self.conversation_history.append({
                    "role": "assistant",
                    "content": assistant_reply
                })
                
                return assistant_reply
            else:
                error_msg = f"API请求失败，状态码：{response.status_code}"
                return error_msg
                
        except requests.exceptions.Timeout:
            return "请求超时，请稍后重试"
        except requests.exceptions.ConnectionError:
            return "无法连接到模型服务，请检查服务是否运行"
        except Exception as e:
            return f"发生错误：{str(e)}"
    
    def clear_history(self):
        """清空对话历史（保留系统提示）"""
        self.conversation_history = [{
            "role": "system",
            "content": self.system_prompt
        }]
    
    def get_history_count(self):
        """获取对话轮数（不包括系统提示）"""
        return len(self.conversation_history) - 1

# 创建全局聊天管理器实例
chat_manager = ChatManager()

@cl.on_chat_start
async def on_chat_start():
    """聊天开始时的初始化"""
    
    # 发送欢迎消息
    welcome_msg = """👋 你好！我是基于通义千问1.8B模型构建的AI助手。
    
我可以帮助你：
- 回答各种问题
- 协助编程和调试
- 进行创意写作
- 翻译和总结文本

请直接在下方输入你的问题，我会尽力帮助你！"""
    
    await cl.Message(content=welcome_msg).send()
    
    # 设置聊天设置
    settings = {
        "model": MODEL_NAME,
        "temperature": 0.7,
        "max_tokens": 300
    }
    
    await cl.ChatSettings(settings).send()

@cl.on_message
async def on_message(message: cl.Message):
    """处理用户消息"""
    
    # 显示用户消息
    user_msg = cl.Message(content=message.content)
    await user_msg.send()
    
    # 检查是否是特殊命令
    if message.content.strip() == "/clear":
        chat_manager.clear_history()
        await cl.Message(content="✅ 对话历史已清空").send()
        return
    
    if message.content.strip() == "/history":
        history_count = chat_manager.get_history_count()
        await cl.Message(content=f"📊 当前对话轮数：{history_count}").send()
        return
    
    # 显示"正在思考"提示
    thinking_msg = cl.Message(content="正在思考...")
    await thinking_msg.send()
    
    # 调用模型获取回复
    reply = chat_manager.send_message(message.content)
    
    # 更新消息内容
    thinking_msg.content = reply
    await thinking_msg.update()

@cl.on_stop
def on_stop():
    """聊天结束时清理"""
    print("聊天会话结束")

# 应用配置
cl.instrument()

这个文件创建了一个完整的Chainlit应用。让我解释一下关键部分：

ChatManager类：封装了所有与模型API交互的逻辑
@cl.on_chat_start：用户开始聊天时触发，发送欢迎消息
@cl.on_message：用户发送消息时触发，处理消息并调用模型
@cl.on_stop：聊天结束时触发，可以在这里做清理工作

4.2 添加配置文件和运行应用

Chainlit需要一个配置文件来定制应用的外观和行为。创建 chainlit.md 文件：

# 欢迎使用通义千问聊天助手

这是一个基于通义千问1.8B模型构建的智能对话应用。

## 功能特点
- 🚀 快速响应
- 💬 多轮对话
- 🔧 代码辅助
- 📚 知识问答

## 使用方法
1. 直接在下方输入框输入问题
2. 按回车或点击发送按钮
3. 等待模型回复

## 可用命令
- `/clear` - 清空对话历史
- `/history` - 查看对话轮数

## 技术栈
- 后端：vLLM + 通义千问1.8B-GPTQ-Int4
- 前端：Chainlit
- 部署：Python + FastAPI

开始你的对话吧！

再创建一个 chainlit.yaml 配置文件：

# Chainlit 配置文件
chainlit:
  # 应用名称
  name: "通义千问聊天助手"
  
  # 描述
  description: "基于通义千问1.8B模型的智能对话应用"
  
  # 作者
  author: "你的名字"
  
  # 版本
  version: "1.0.0"
  
  # 网站链接
  website: "https://example.com"
  
  # 图标
  icon: "🤖"
  
  # 默认展开侧边栏
  default_expand_messages: false
  
  # 默认折叠侧边栏
  default_collapse_messages: false
  
  # 隐藏侧边栏
  hide_sidebar: false
  
  # 主题
  theme:
    primary_color: "#4f46e5"
    background_color: "#ffffff"
    text_color: "#000000"
  
  # 功能开关
  features:
    # 显示上传按钮
    upload: true
    # 显示设置按钮
    settings: true
    # 显示历史按钮
    history: true
  
  # 消息配置
  messages:
    # 最大消息数
    max_size_mb: 10
    # 最大消息长度
    max_length: 10000
  
  # 用户会话
  user_session:
    # 会话过期时间（秒）
    expiration: 3600
  
  # 安全性
  security:
    # 允许的文件类型
    allowed_mime_types:
      - "text/plain"
      - "application/pdf"
      - "image/png"
      - "image/jpeg"

现在，让我们运行应用。在终端中执行：

# 运行Chainlit应用
chainlit run app.py

第一次运行时会提示你创建配置文件，按回车确认即可。应用启动后，会自动打开浏览器，访问 http://localhost:8000（注意：这是Chainlit的默认端口，不要和vLLM的8000端口冲突，如果冲突可以修改）。

4.3 高级功能：文件上传和流式输出

让我们给应用添加更多实用功能。修改 app.py，增加文件处理和流式输出：

import chainlit as cl
import requests
import json
import tempfile
import os
from typing import Optional, List
from pathlib import Path

# ... 之前的ChatManager类保持不变 ...

@cl.on_chat_start
async def on_chat_start():
    """聊天开始时的初始化"""
    
    # 发送欢迎消息
    welcome_msg = cl.Message(content="")
    await welcome_msg.send()
    
    # 逐步显示欢迎消息（模拟打字效果）
    welcome_text = """👋 你好！我是基于通义千问1.8B模型构建的AI助手。
    
我可以帮助你：
- 📝 回答各种问题
- 💻 协助编程和调试  
- ✍️ 进行创意写作
- 🔍 翻译和总结文本
- 📄 处理上传的文本文件

请直接在下方输入你的问题，或者上传文件让我分析！"""
    
    # 逐字显示效果
    for i in range(0, len(welcome_text) + 1, 5):
        welcome_msg.content = welcome_text[:i]
        await welcome_msg.update()
        await cl.sleep(0.01)  # 短暂延迟
    
    # 设置聊天设置
    settings = {
        "model": MODEL_NAME,
        "temperature": 0.7,
        "max_tokens": 500,
        "stream": True  # 启用流式输出
    }
    
    await cl.ChatSettings(settings).send()

@cl.on_message
async def on_message(message: cl.Message):
    """处理用户消息"""
    
    # 处理文件上传
    if message.elements:
        for element in message.elements:
            if element.type == "file":
                # 读取文件内容
                file_path = element.path
                try:
                    with open(file_path, 'r', encoding='utf-8') as f:
                        file_content = f.read()
                    
                    # 添加到用户消息中
                    message.content = f"请分析以下文件内容：\n\n{file_content}\n\n{message.content}"
                    
                    await cl.Message(content=f"📄 已读取文件：{element.name}").send()
                    
                except Exception as e:
                    await cl.Message(content=f"❌ 读取文件失败：{str(e)}").send()
                    return
    
    # 显示用户消息
    user_msg = cl.Message(content=message.content)
    await user_msg.send()
    
    # 检查特殊命令
    if message.content.strip() == "/clear":
        chat_manager.clear_history()
        await cl.Message(content="✅ 对话历史已清空").send()
        return
    
    if message.content.strip() == "/history":
        history_count = chat_manager.get_history_count()
        await cl.Message(content=f"📊 当前对话轮数：{history_count}").send()
        return
    
    if message.content.strip() == "/help":
        help_text = """可用命令：
        /clear - 清空对话历史
        /history - 查看对话轮数
        /help - 显示帮助信息
        
        支持功能：
        • 文本对话
        • 文件上传（txt、pdf、图片）
        • 流式响应
        • 多轮对话"""
        await cl.Message(content=help_text).send()
        return
    
    # 创建流式响应消息
    msg = cl.Message(content="")
    await msg.send()
    
    # 准备请求数据（流式）
    chat_manager.conversation_history.append({
        "role": "user",
        "content": message.content
    })
    
    payload = {
        "model": MODEL_NAME,
        "messages": chat_manager.conversation_history,
        "max_tokens": 500,
        "temperature": 0.7,
        "stream": True  # 启用流式
    }
    
    headers = {
        "Content-Type": "application/json"
    }
    
    try:
        # 发送流式请求
        response = requests.post(
            VLLM_API_URL,
            json=payload,
            headers=headers,
            stream=True,
            timeout=60
        )
        
        if response.status_code == 200:
            full_response = ""
            
            # 处理流式响应
            for line in response.iter_lines():
                if line:
                    line = line.decode('utf-8')
                    
                    # 跳过SSE格式的前缀
                    if line.startswith("data: "):
                        data = line[6:]  # 去掉"data: "前缀
                        
                        if data == "[DONE]":
                            break
                        
                        try:
                            json_data = json.loads(data)
                            if "choices" in json_data and len(json_data["choices"]) > 0:
                                delta = json_data["choices"][0].get("delta", {})
                                if "content" in delta:
                                    content = delta["content"]
                                    full_response += content
                                    
                                    # 逐步更新消息
                                    msg.content = full_response
                                    await msg.update()
                        except json.JSONDecodeError:
                            continue
            
            # 添加助手回复到历史
            chat_manager.conversation_history.append({
                "role": "assistant",
                "content": full_response
            })
            
        else:
            msg.content = f"请求失败：{response.status_code}"
            await msg.update()
            
    except Exception as e:
        msg.content = f"发生错误：{str(e)}"
        await msg.update()

@cl.on_stop
def on_stop():
    """聊天结束时清理"""
    print("聊天会话结束")
    # 可以在这里添加清理代码，如保存对话记录等

# 文件上传处理器
@cl.step(type="tool")
async def process_file(file_path: str):
    """处理上传的文件"""
    
    step = cl.context.current_step
    step.name = "文件处理"
    
    try:
        # 获取文件信息
        file_name = Path(file_path).name
        file_size = os.path.getsize(file_path)
        
        step.output = f"正在处理文件：{file_name} ({file_size}字节)"
        
        # 这里可以添加具体的文件处理逻辑
        # 例如：文本提取、格式转换等
        
        return {
            "status": "success",
            "file_name": file_name,
            "file_size": file_size
        }
        
    except Exception as e:
        step.output = f"文件处理失败：{str(e)}"
        return {
            "status": "error",
            "error": str(e)
        }

# 运行配置
if __name__ == "__main__":
    # 这行代码让Chainlit运行应用
    pass

这个增强版本添加了：

文件上传支持：用户可以上传文本文件，模型会分析文件内容
流式输出：模型回复逐字显示，体验更好
更多命令：增加了 /help 命令
打字效果：欢迎消息逐字显示
文件处理工具：专门处理上传的文件

4.4 自定义界面样式

如果你想要进一步美化界面，可以创建自定义CSS文件。创建 chainlit.css：

/* 自定义Chainlit样式 */

/* 主容器 */
.cl-container {
    max-width: 1200px;
    margin: 0 auto;
}

/* 消息气泡 */
.cl-message {
    border-radius: 12px;
    margin: 12px 0;
    padding: 16px;
    box-shadow: 0 2px 8px rgba(0, 0, 0, 0.1);
}

/* 用户消息 */
.cl-message-user {
    background-color: #e3f2fd;
    border-left: 4px solid #2196f3;
}

/* 助手消息 */
.cl-message-assistant {
    background-color: #f5f5f5;
    border-left: 4px solid #4caf50;
}

/* 系统消息 */
.cl-message-system {
    background-color: #fff3e0;
    border-left: 4px solid #ff9800;
}

/* 输入框 */
.cl-input {
    border-radius: 8px;
    border: 2px solid #e0e0e0;
    padding: 12px;
    font-size: 16px;
}

.cl-input:focus {
    border-color: #4f46e5;
    box-shadow: 0 0 0 3px rgba(79, 70, 229, 0.1);
    outline: none;
}

/* 按钮 */
.cl-button {
    background-color: #4f46e5;
    color: white;
    border-radius: 8px;
    padding: 10px 20px;
    border: none;
    cursor: pointer;
    font-weight: 500;
    transition: background-color 0.2s;
}

.cl-button:hover {
    background-color: #4338ca;
}

.cl-button:active {
    background-color: #3730a3;
}

/* 上传区域 */
.cl-upload-area {
    border: 2px dashed #cbd5e1;
    border-radius: 8px;
    padding: 32px;
    text-align: center;
    background-color: #f8fafc;
    transition: border-color 0.2s;
}

.cl-upload-area:hover {
    border-color: #4f46e5;
    background-color: #f1f5f9;
}

/* 侧边栏 */
.cl-sidebar {
    background-color: #f8fafc;
    border-right: 1px solid #e2e8f0;
}

/* 标题 */
.cl-header {
    background: linear-gradient(135deg, #4f46e5 0%, #7c3aed 100%);
    color: white;
    padding: 20px;
}

.cl-header h1 {
    margin: 0;
    font-size: 24px;
    font-weight: 600;
}

/* 加载动画 */
.cl-loading {
    display: inline-block;
    width: 20px;
    height: 20px;
    border: 3px solid #f3f3f3;
    border-top: 3px solid #4f46e5;
    border-radius: 50%;
    animation: spin 1s linear infinite;
}

@keyframes spin {
    0% { transform: rotate(0deg); }
    100% { transform: rotate(360deg); }
}

/* 代码块 */
.cl-code-block {
    background-color: #1e293b;
    color: #e2e8f0;
    border-radius: 8px;
    padding: 16px;
    font-family: 'Courier New', monospace;
    overflow-x: auto;
}

/* 响应式设计 */
@media (max-width: 768px) {
    .cl-container {
        padding: 10px;
    }
    
    .cl-message {
        padding: 12px;
        margin: 8px 0;
    }
    
    .cl-input {
        font-size: 14px;
        padding: 10px;
    }
}

/* 暗色模式支持 */
@media (prefers-color-scheme: dark) {
    .cl-message-user {
        background-color: #1e3a8a;
        color: #e2e8f0;
    }
    
    .cl-message-assistant {
        background-color: #374151;
        color: #e2e8f0;
    }
    
    .cl-input {
        background-color: #374151;
        color: #e2e8f0;
        border-color: #4b5563;
    }
}

然后在 chainlit.yaml 中引用这个CSS文件：

# 在chainlit.yaml中添加
ui:
  # 自定义CSS文件
  stylesheet: "chainlit.css"
  
  # 自定义JavaScript文件（可选）
  # script: "custom.js"

现在你的应用就有了自定义的样式，看起来更加专业和美观。

5. 部署优化与实用技巧

应用已经可以运行了，但在实际使用中，我们还需要考虑性能、稳定性和用户体验。下面是一些实用的优化技巧。

5.1 性能优化：缓存和批处理

当有多个用户同时使用时，我们需要优化性能。修改 ChatManager 类：

import hashlib
import time
from functools import lru_cache
from concurrent.futures import ThreadPoolExecutor

class OptimizedChatManager:
    """优化后的聊天管理器"""
    
    def __init__(self, max_workers=4):
        self.conversation_history = []
        self.system_prompt = """你是一个有帮助的AI助手，用中文回答问题。"""
        
        self.conversation_history.append({
            "role": "system",
            "content": self.system_prompt
        })
        
        # 创建线程池处理并发请求
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
        
        # 请求统计
        self.request_count = 0
        self.total_response_time = 0
        
    @lru_cache(maxsize=100)
    def _get_response_cache(self, prompt_hash: str, history_hash: str):
        """缓存响应结果"""
        # 这里实际应该从缓存存储（如Redis）获取
        # 为了简单，我们只实现内存缓存
        return None
    
    def _set_response_cache(self, prompt_hash: str, history_hash: str, response: str):
        """设置缓存"""
        # 这里实际应该存储到缓存系统
        pass
    
    def _generate_cache_key(self, user_message: str) -> tuple:
        """生成缓存键"""
        # 基于用户消息和最近的历史生成哈希
        recent_history = self.conversation_history[-5:]  # 只考虑最近5轮
        history_str = json.dumps(recent_history, sort_keys=True)
        
        prompt_hash = hashlib.md5(user_message.encode()).hexdigest()[:8]
        history_hash = hashlib.md5(history_str.encode()).hexdigest()[:8]
        
        return (prompt_hash, history_hash)
    
    def send_message_optimized(self, user_message: str) -> str:
        """优化版的发送消息方法"""
        
        start_time = time.time()
        
        # 检查缓存
        cache_key = self._generate_cache_key(user_message)
        cached_response = self._get_response_cache(*cache_key)
        
        if cached_response:
            print(f"使用缓存响应：{cache_key}")
            return cached_response
        
        # 添加用户消息
        self.conversation_history.append({
            "role": "user",
            "content": user_message
        })
        
        # 准备请求数据
        payload = {
            "model": MODEL_NAME,
            "messages": self.conversation_history[-10:],  # 只发送最近10条消息
            "max_tokens": 300,
            "temperature": 0.7,
            "stream": False
        }
        
        try:
            # 使用线程池异步发送请求
            future = self.executor.submit(
                requests.post,
                VLLM_API_URL,
                json=payload,
                headers={"Content-Type": "application/json"},
                timeout=45
            )
            
            response = future.result(timeout=50)
            
            if response.status_code == 200:
                result = response.json()
                assistant_reply = result["choices"][0]["message"]["content"]
                
                # 添加到历史
                self.conversation_history.append({
                    "role": "assistant",
                    "content": assistant_reply
                })
                
                # 更新统计
                self.request_count += 1
                response_time = time.time() - start_time
                self.total_response_time += response_time
                
                print(f"请求完成，耗时：{response_time:.2f}秒")
                print(f"平均响应时间：{self.total_response_time/self.request_count:.2f}秒")
                
                # 缓存结果
                self._set_response_cache(*cache_key, assistant_reply)
                
                return assistant_reply
            else:
                return f"请求失败：{response.status_code}"
                
        except Exception as e:
            return f"调用出错：{str(e)}"
    
    def get_stats(self):
        """获取统计信息"""
        avg_time = (self.total_response_time / self.request_count 
                   if self.request_count > 0 else 0)
        
        return {
            "total_requests": self.request_count,
            "avg_response_time": f"{avg_time:.2f}秒",
            "history_length": len(self.conversation_history) - 1
        }
    
    def clear_history(self):
        """清空历史"""
        self.conversation_history = [{
            "role": "system",
            "content": self.system_prompt
        }]

这个优化版本添加了：

响应缓存：对相同的问题缓存响应，减少API调用
线程池：支持并发请求处理
历史截断：只发送最近的消息，减少传输数据量
性能统计：记录响应时间和请求次数

5.2 错误处理和降级策略

在实际部署中，模型服务可能不稳定。我们需要完善的错误处理：

class RobustChatManager:
    """健壮性更强的聊天管理器"""
    
    def __init__(self, fallback_responses=None):
        self.conversation_history = []
        self.system_prompt = """你是一个有帮助的AI助手。"""
        
        self.conversation_history.append({
            "role": "system",
            "content": self.system_prompt
        })
        
        # 备用回复（当模型不可用时使用）
        self.fallback_responses = fallback_responses or [
            "我目前正在思考中，请稍后再试。",
            "系统正在处理您的请求，请耐心等待。",
            "这个问题很有趣，让我再想想。",
            "抱歉，我暂时无法回答这个问题。",
            "请尝试重新提问或稍后再试。"
        ]
        
        # 错误计数
        self.error_count = 0
        self.last_error_time = None
        
    def send_message_robust(self, user_message: str, max_retries=2) -> str:
        """带重试和降级的发送消息方法"""
        
        for attempt in range(max_retries + 1):
            try:
                # 检查错误状态
                if self._should_use_fallback():
                    return self._get_fallback_response(user_message)
                
                # 正常请求
                response = self._call_model_api(user_message)
                
                # 重置错误计数
                if self.error_count > 0:
                    self.error_count = 0
                    self.last_error_time = None
                    print("模型服务恢复正常")
                
                return response
                
            except requests.exceptions.ConnectionError as e:
                self._handle_error("连接错误", e, attempt, max_retries)
                
            except requests.exceptions.Timeout as e:
                self._handle_error("请求超时", e, attempt, max_retries)
                
            except requests.exceptions.HTTPError as e:
                self._handle_error(f"HTTP错误 {e.response.status_code}", e, attempt, max_retries)
                
            except json.JSONDecodeError as e:
                self._handle_error("响应解析错误", e, attempt, max_retries)
                
            except Exception as e:
                self._handle_error(f"未知错误: {type(e).__name__}", e, attempt, max_retries)
        
        # 所有重试都失败，使用备用回复
        return self._get_fallback_response(user_message)
    
    def _call_model_api(self, user_message: str) -> str:
        """调用模型API"""
        # 添加用户消息
        self.conversation_history.append({
            "role": "user",
            "content": user_message
        })
        
        payload = {
            "model": MODEL_NAME,
            "messages": self.conversation_history[-6:],  # 最近6条
            "max_tokens": 250,
            "temperature": 0.7,
            "stream": False
        }
        
        response = requests.post(
            VLLM_API_URL,
            json=payload,
            headers={"Content-Type": "application/json"},
            timeout=30
        )
        
        response.raise_for_status()
        result = response.json()
        
        assistant_reply = result["choices"][0]["message"]["content"]
        
        # 添加到历史
        self.conversation_history.append({
            "role": "assistant",
            "content": assistant_reply
        })
        
        return assistant_reply
    
    def _handle_error(self, error_type: str, error: Exception, 
                     attempt: int, max_retries: int):
        """处理错误"""
        self.error_count += 1
        self.last_error_time = time.time()
        
        print(f"{error_type}: {str(error)}")
        
        if attempt < max_retries:
            retry_delay = 2 ** attempt  # 指数退避
            print(f"第{attempt + 1}次重试，等待{retry_delay}秒...")
            time.sleep(retry_delay)
        else:
            print(f"已达到最大重试次数({max_retries})，使用备用回复")
    
    def _should_use_fallback(self) -> bool:
        """判断是否应该使用备用回复"""
        # 如果最近错误太多，暂时使用备用回复
        if self.error_count >= 3:
            # 检查是否在冷却期（5分钟内）
            if self.last_error_time and (time.time() - self.last_error_time) < 300:
                return True
            else:
                # 超过5分钟，重置错误计数
                self.error_count = 0
                self.last_error_time = None
        
        return False
    
    def _get_fallback_response(self, user_message: str) -> str:
        """获取备用回复"""
        # 基于用户消息选择备用回复
        message_hash = hash(user_message) % len(self.fallback_responses)
        response = self.fallback_responses[message_hash]
        
        # 添加到历史（标记为备用回复）
        self.conversation_history.append({
            "role": "user",
            "content": user_message
        })
        self.conversation_history.append({
            "role": "assistant",
            "content": f"[备用回复] {response}"
        })
        
        return response
    
    def get_health_status(self):
        """获取健康状态"""
        status = "健康"
        if self.error_count >= 3:
            status = "降级运行"
        elif self.error_count > 0:
            status = "部分异常"
        
        return {
            "status": status,
            "error_count": self.error_count,
            "last_error_time": self.last_error_time,
            "conversation_length": len(self.conversation_history) - 1
        }

这个版本提供了：

自动重试：网络错误时自动重试
指数退避：重试间隔逐渐增加
降级策略：模型不可用时使用备用回复
健康检查：监控服务状态
错误恢复：一段时间后自动恢复

5.3 监控和日志

在生产环境中，监控和日志非常重要：

import logging
from datetime import datetime

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('chat_app.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

class MonitoredChatManager:
    """带监控的聊天管理器"""
    
    def __init__(self):
        self.conversation_history = []
        self.system_prompt = """你是一个有帮助的AI助手。"""
        
        self.conversation_history.append({
            "role": "system",
            "content": self.system_prompt
        })
        
        # 监控数据
        self.metrics = {
            "total_requests": 0,
            "successful_requests": 0,
            "failed_requests": 0,
            "total_response_time": 0,
            "cache_hits": 0,
            "cache_misses": 0
        }
        
        logger.info("聊天管理器初始化完成")
    
    def send_message_with_monitoring(self, user_message: str) -> str:
        """带监控的发送消息方法"""
        
        self.metrics["total_requests"] += 1
        start_time = time.time()
        
        logger.info(f"收到用户消息: {user_message[:50]}...")
        
        try:
            # 添加用户消息
            self.conversation_history.append({
                "role": "user",
                "content": user_message
            })
            
            payload = {
                "model": MODEL_NAME,
                "messages": self.conversation_history[-8:],
                "max_tokens": 300,
                "temperature": 0.7,
                "stream": False
            }
            
            logger.debug(f"发送请求到API: {payload}")
            
            response = requests.post(
                VLLM_API_URL,
                json=payload,
                headers={"Content-Type": "application/json"},
                timeout=30
            )
            
            response_time = time.time() - start_time
            self.metrics["total_response_time"] += response_time
            
            if response.status_code == 200:
                self.metrics["successful_requests"] += 1
                
                result = response.json()
                assistant_reply = result["choices"][0]["message"]["content"]
                
                # 添加到历史
                self.conversation_history.append({
                    "role": "assistant",
                    "content": assistant_reply
                })
                
                logger.info(f"请求成功，响应时间: {response_time:.2f}秒")
                logger.debug(f"模型回复: {assistant_reply[:100]}...")
                
                # 记录详细指标
                self._record_metrics({
                    "user_message_length": len(user_message),
                    "assistant_reply_length": len(assistant_reply),
                    "response_time": response_time,
                    "timestamp": datetime.now().isoformat()
                })
                
                return assistant_reply
                
            else:
                self.metrics["failed_requests"] += 1
                error_msg = f"API返回错误: {response.status_code}"
                logger.error(f"{error_msg}, 响应: {response.text}")
                
                return error_msg
                
        except Exception as e:
            self.metrics["failed_requests"] += 1
            logger.error(f"请求异常: {str(e)}", exc_info=True)
            
            return f"请求异常: {str(e)}"
    
    def _record_metrics(self, metrics: dict):
        """记录详细指标"""
        # 这里可以将指标保存到数据库或文件
        metrics_file = "chat_metrics.json"
        
        try:
            # 读取现有指标
            if os.path.exists(metrics_file):
                with open(metrics_file, 'r', encoding='utf-8') as f:
                    all_metrics = json.load(f)
            else:
                all_metrics = []
            
            # 添加新指标
            all_metrics.append(metrics)
            
            # 只保留最近1000条记录
            if len(all_metrics) > 1000:
                all_metrics = all_metrics[-1000:]
            
            # 保存到文件
            with open(metrics_file, 'w', encoding='utf-8') as f:
                json.dump(all_metrics, f, ensure_ascii=False, indent=2)
                
        except Exception as e:
            logger.error(f"记录指标失败: {str(e)}")
    
    def get_summary_metrics(self):
        """获取汇总指标"""
        avg_response_time = (
            self.metrics["total_response_time"] / self.metrics["successful_requests"]
            if self.metrics["successful_requests"] > 0 else 0
        )
        
        success_rate = (
            self.metrics["successful_requests"] / self.metrics["total_requests"] * 100
            if self.metrics["total_requests"] > 0 else 0
        )
        
        return {
            "总请求数": self.metrics["total_requests"],
            "成功请求": self.metrics["successful_requests"],
            "失败请求": self.metrics["failed_requests"],
            "成功率": f"{success_rate:.1f}%",
            "平均响应时间": f"{avg_response_time:.2f}秒",
            "缓存命中": self.metrics["cache_hits"],
            "缓存未命中": self.metrics["cache_misses"],
            "当前对话轮数": len(self.conversation_history) - 1
        }
    
    def export_conversation(self, filename=None):
        """导出对话记录"""
        if filename is None:
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            filename = f"conversation_{timestamp}.json"
        
        try:
            with open(filename, 'w', encoding='utf-8') as f:
                json.dump(self.conversation_history, f, ensure_ascii=False, indent=2)
            
            logger.info(f"对话记录已导出到: {filename}")
            return True
            
        except Exception as e:
            logger.error(f"导出对话记录失败: {str(e)}")
            return False

这个监控版本提供了：

详细日志：记录所有重要事件
性能指标：跟踪响应时间、成功率等
对话导出：保存对话记录到文件
错误追踪：记录异常堆栈信息
指标持久化：将指标保存到文件供分析

6. 总结与下一步建议

6.1 本文内容回顾

通过本文的学习，我们完成了一个完整的通义千问1.8B模型应用开发流程。让我们回顾一下关键步骤：

模型部署确认：首先确保vLLM服务正常运行，模型加载成功
基础API调用：使用Python的requests库调用vLLM的API接口
对话逻辑封装：创建ChatManager类管理对话历史和API调用
Web界面构建：使用Chainlit框架创建交互式聊天界面
功能增强：添加文件上传、流式输出、自定义样式等高级功能
性能优化：实现缓存、并发处理、错误恢复等生产级特性
监控日志：添加完整的监控和日志系统

整个过程中，我们始终关注代码的实用性、健壮性和可维护性。每个版本都在前一个基础上增加新功能，你可以根据自己的需求选择合适的版本。

6.2 实际应用建议

在实际部署和使用时，我有几个建议：

对于个人使用或小规模测试：

使用基础版本或增强版本即可
关注功能的完整性和用户体验
定期备份对话记录

对于团队使用或生产环境：

使用优化版本或监控版本
考虑添加用户认证和权限控制
部署到云服务器，配置域名和SSL证书
设置自动备份和监控告警

性能调优建议：

调整vLLM参数：根据你的硬件配置调整--max-model-len、--gpu-memory-utilization等参数
启用批处理：vLLM支持请求批处理，可以显著提高吞吐量
使用量化模型：像我们使用的GPTQ-Int4量化版本，可以在保持性能的同时减少内存占用
配置反向代理：使用Nginx等反向代理处理并发连接

6.3 扩展功能思路

如果你想让这个应用更强大，可以考虑以下扩展方向：

功能扩展：

多模型支持：让用户可以选择不同的模型
对话分享：生成对话链接，方便分享
API接口：提供REST API供其他应用调用
插件系统：支持自定义插件扩展功能
知识库集成：连接外部知识库提供更准确的回答

技术优化：

数据库存储：使用SQLite或PostgreSQL存储对话历史
Redis缓存：用Redis替代内存缓存，支持分布式部署
异步处理：使用asyncio提高并发性能
Docker部署：容器化部署，方便迁移和扩展
CI/CD流水线：自动化测试和部署

用户体验改进：

主题切换：支持亮色/暗色主题
快捷键支持：键盘快捷键提高操作效率
语音输入：集成语音识别功能
多语言界面：支持中英文切换
移动端适配：优化手机端使用体验

6.4 常见问题解决

在实际使用中，你可能会遇到一些问题。这里是一些常见问题的解决方法：

问题1：模型响应慢

检查服务器资源使用情况（CPU、内存、GPU）
调整vLLM的--max-num-batched-tokens参数
考虑升级硬件或使用更小的模型

问题2：内存不足

使用量化版本模型（如GPTQ-Int4）
减少--max-model-len参数值
增加服务器内存或使用内存交换

问题3：API调用失败

检查vLLM服务是否正常运行
确认端口是否正确（默认8000）
查看vLLM日志排查错误

问题4：Chainlit界面无法访问

检查Chainlit是否在运行
确认端口是否被占用（默认8000）
查看Chainlit日志获取错误信息

问题5：对话历史丢失

实现对话持久化存储
定期备份对话记录
使用数据库替代内存存储

6.5 学习资源推荐

如果你想深入学习相关技术，我推荐以下资源：

vLLM相关：

官方文档：https://docs.vllm.ai/
GitHub仓库：https://github.com/vllm-project/vllm
论文：《Efficient Memory Management for Large Language Model Serving》

Chainlit相关：

官方文档：https://docs.chainlit.io/
示例项目：https://github.com/Chainlit/chainlit
社区讨论：https://discord.gg/chainlit

通义千问相关：

官方GitHub：https://github.com/QwenLM/Qwen1.5
模型卡片：https://huggingface.co/Qwen
技术报告：https://arxiv.org/abs/2309.16609

Python Web开发：

FastAPI官方文档：https://fastapi.tiangolo.com/
Flask官方文档：https://flask.palletsprojects.com/
Django官方文档：https://docs.djangoproject.com/

6.6 最后的建议

开发AI应用是一个持续迭代的过程。不要试图一开始就做出完美的系统，而是从最小可行产品（MVP）开始，然后根据用户反馈不断改进。

记住几个关键原则：

用户第一：始终关注用户体验和实际需求
渐进增强：先实现核心功能，再逐步添加高级特性
监控度量：没有度量就没有改进，要持续监控系统表现
安全可靠：确保系统稳定安全，处理好边界情况
保持学习：AI技术发展很快，要持续学习新技术

希望本文能帮助你快速上手通义千问模型的集成开发。如果你在实践过程中遇到问题，或者有新的想法和改进，欢迎分享和交流。技术之路，我们一起前行。

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。

DeepSeek技术社区

欢迎加入DeepSeek 技术社区。在这里，你可以找到志同道合的朋友，共同探索AI技术的奥秘。

更多推荐

DeepSeek-V4 复杂指令执行失败排查：为什么你的 RAG 管道吞掉了嵌套 JSON？

DeepSeek技术社区

DeepSeek Golang SDK 接入实战：多租户 API 网关的配额与熔断设计

DeepSeek技术社区

DeepSeek 成本看板搭建实战：如何从 per-token 粒度优化推理账单

DeepSeek技术社区

所有评论(0)

查看更多评论

江卓尔

@weixin_32098457

已为社区贡献19条内容

通义千问1.8B-GPTQ-Int4代码实例：Python调用vLLM API与Chainlit集成详解

江卓尔

通义千问1.8B-GPTQ-Int4代码实例：Python调用vLLM API与Chainlit集成详解

1. 引言：从模型部署到交互式应用

2. 环境准备与模型确认

2.1 确认模型服务状态

2.2 安装必要的Python包

2.3 了解vLLM的API接口

3. 基础API调用：让模型开口说话

3.1 最简化的API调用示例

3.2 添加错误处理和超时设置

3.3 支持多轮对话

4. Chainlit集成：构建Web对话界面

4.1 创建基本的Chainlit应用

4.2 添加配置文件和运行应用

4.3 高级功能：文件上传和流式输出

4.4 自定义界面样式

5. 部署优化与实用技巧

5.1 性能优化：缓存和批处理

5.2 错误处理和降级策略

5.3 监控和日志

6. 总结与下一步建议

6.1 本文内容回顾

6.2 实际应用建议

6.3 扩展功能思路

6.4 常见问题解决

6.5 学习资源推荐

6.6 最后的建议

所有评论(0)

温馨提示：您尚未绑定手机号

江卓尔