Qwen3-0.6B-FP8实战教程：构建跨平台AI助手——Web/Telegram/Discord多端统一后端

Ne0inhk

— 25 min read

Qwen3-0.6B-FP8实战教程：构建跨平台AI助手——Web/Telegram/Discord多端统一后端

1. 开篇：为什么需要一个多端统一的AI助手？

想象一下这个场景：你正在电脑前写代码，突然想到一个问题，于是打开浏览器，访问一个AI对话页面提问。过了一会儿，你出门了，在手机上收到朋友的消息，想用同一个AI助手帮忙想个点子，却不得不切换到另一个App。晚上，你和团队在Discord上讨论项目，又想调用AI来辅助决策，结果发现还得重新部署一套服务。

是不是很麻烦？这就是我们今天要解决的问题。

Qwen3-0.6B-FP8是一个小巧但强大的语言模型，它能在资源有限的环境下流畅运行。但光有模型还不够，我们需要一个能同时服务Web页面、Telegram机器人和Discord机器人的统一后端。这样，无论你在哪里，用什么设备，都能无缝使用同一个AI助手。

这篇文章，我就带你一步步搭建这样一个系统。不需要高深的编程知识，跟着做就行。

2. 环境准备与模型部署

2.1 你需要准备什么

在开始之前，确保你有以下环境：

  • 一台Linux服务器：可以是云服务器，也可以是本地电脑。建议内存至少4GB。
  • Python 3.8或更高版本：这是运行大多数AI框架的基础。
  • 基本的命令行操作知识：知道怎么用cd、ls、pip这些命令就行。

如果你用的是ZEEKLOG星图镜像，那更简单——很多环境都已经预装好了。

2.2 部署Qwen3-0.6B-FP8模型

我们先从核心开始：把模型跑起来。

第一步：安装必要的工具

打开终端，输入以下命令：

# 更新包管理器 sudo apt-get update # 安装Python开发环境 sudo apt-get install -y python3-pip python3-dev # 安装vLLM——一个高效的模型推理框架 pip3 install vllm

第二步：下载并运行模型

现在我们来启动模型服务。Qwen3-0.6B-FP8模型比较小，对硬件要求不高，普通电脑也能跑。

# 启动模型服务 python3 -m vllm.entrypoints.openai.api_server \ --model Qwen/Qwen3-0.6B-Instruct \ --served-model-name qwen3-0.6b-fp8 \ --max-model-len 4096 \ --dtype float16 \ --port 8000

让我解释一下这些参数是干什么的：

  • --model Qwen/Qwen3-0.6B-Instruct：指定要加载的模型
  • --served-model-name qwen3-0.6b-fp8：给服务起个名字，后面调用时会用到
  • --max-model-len 4096：模型最多能处理4096个token（大约3000字）
  • --dtype float16：使用半精度浮点数，节省内存
  • --port 8000：服务运行在8000端口

第三步：检查服务是否正常

服务启动需要一点时间（大概1-2分钟）。怎么知道它准备好了呢？

打开另一个终端窗口，输入：

# 查看服务日志 tail -f /root/workspace/llm.log

如果你看到类似这样的输出，就说明模型加载成功了：

INFO 07-15 14:30:25 llm_engine.py:197] Initializing an LLM engine with config: ... INFO 07-15 14:30:45 llm_engine.py:387] Finished loading the model. INFO 07-15 14:30:45 llm_engine.py:390] Model weights loaded.

或者更简单的方法，直接访问服务：

# 发送一个测试请求 curl http://localhost:8000/v1/models

如果返回类似{"object":"list","data":[{"id":"qwen3-0.6b-fp8",...}]}的JSON数据，说明一切正常。

3. 构建统一的后端服务

模型跑起来了，现在我们需要一个“中间人”——它接收来自不同平台的请求，转发给模型，再把结果返回去。

3.1 创建后端项目结构

先创建一个项目文件夹，并组织好文件结构：

# 创建项目文件夹 mkdir ai-assistant-backend cd ai-assistant-backend # 创建必要的文件 touch main.py touch config.py touch requirements.txt touch .env

3.2 安装依赖包

编辑requirements.txt文件，添加以下内容：

fastapi==0.104.1 uvicorn==0.24.0 python-telegram-bot==20.6 discord.py==2.3.2 openai==1.3.0 python-dotenv==1.0.0 httpx==0.25.1

然后安装它们：

pip3 install -r requirements.txt

3.3 编写核心后端代码

现在我们来写主要的后端逻辑。打开main.py，输入以下代码：

from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List, Optional import httpx import asyncio from datetime import datetime import logging # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # 创建FastAPI应用 app = FastAPI(title="多端AI助手统一后端", version="1.0.0") # 定义请求和响应的数据结构 class ChatMessage(BaseModel): role: str # user 或 assistant content: str class ChatRequest(BaseModel): messages: List[ChatMessage] max_tokens: Optional[int] = 500 temperature: Optional[float] = 0.7 class ChatResponse(BaseModel): response: str model: str timestamp: str # 模型服务的配置 MODEL_API_URL = "http://localhost:8000/v1/chat/completions" MODEL_NAME = "qwen3-0.6b-fp8" async def call_model(messages: List[dict], max_tokens: int = 500, temperature: float = 0.7) -> str: """ 调用Qwen3模型生成回复 """ try: async with httpx.AsyncClient(timeout=30.0) as client: payload = { "model": MODEL_NAME, "messages": messages, "max_tokens": max_tokens, "temperature": temperature, "stream": False } response = await client.post(MODEL_API_URL, json=payload) response.raise_for_status() result = response.json() return result["choices"][0]["message"]["content"] except httpx.RequestError as e: logger.error(f"请求模型服务失败: {e}") raise HTTPException(status_code=503, detail="模型服务暂时不可用") except Exception as e: logger.error(f"处理模型响应失败: {e}") raise HTTPException(status_code=500, detail="模型处理失败") @app.post("/api/chat", response_model=ChatResponse) async def chat_endpoint(request: ChatRequest): """ 统一的聊天接口，供所有前端调用 """ logger.info(f"收到聊天请求，消息数: {len(request.messages)}") # 转换消息格式 messages = [{"role": msg.role, "content": msg.content} for msg in request.messages] # 调用模型 response_text = await call_model( messages=messages, max_tokens=request.max_tokens, temperature=request.temperature ) # 返回响应 return ChatResponse( response=response_text, model=MODEL_NAME, timestamp=datetime.now().isoformat() ) @app.get("/health") async def health_check(): """ 健康检查接口 """ try: async with httpx.AsyncClient() as client: # 检查模型服务是否正常 model_response = await client.get("http://localhost:8000/v1/models", timeout=5.0) model_response.raise_for_status() return { "status": "healthy", "model_service": "running", "timestamp": datetime.now().isoformat() } except Exception as e: logger.error(f"健康检查失败: {e}") raise HTTPException(status_code=503, detail="服务异常") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8080)

这段代码做了几件事：

  1. 创建了一个FastAPI应用，提供了/api/chat接口来处理聊天请求
  2. 定义了清晰的数据结构，确保前后端通信规范
  3. 实现了到模型服务的调用逻辑
  4. 添加了健康检查接口，方便监控服务状态

3.4 启动后端服务

保存文件后，启动后端服务：

# 在后端项目目录下运行 python3 main.py

服务会运行在8080端口。你可以打开浏览器访问http://localhost:8080/docs，看到自动生成的API文档界面。

4. 集成Web前端（Chainlit）

有了后端，现在我们来给Web页面做个漂亮的前端。Chainlit是一个专门为AI应用设计的UI框架，用起来特别简单。

4.1 创建Web前端

新建一个文件web_frontend.py：

import chainlit as cl import httpx import asyncio from typing import List, Dict # 后端API地址 BACKEND_URL = "http://localhost:8080/api/chat" @cl.on_chat_start async def start_chat(): """ 聊天开始时的初始化 """ # 发送欢迎消息 await cl.Message( content="👋 你好！我是基于Qwen3-0.6B-FP8的AI助手。我可以帮你回答问题、写作、编程等等。有什么可以帮你的吗？" ).send() @cl.on_message async def handle_message(message: cl.Message): """ 处理用户消息 """ # 显示"正在思考"的提示 msg = cl.Message(content="") await msg.send() try: # 准备请求数据 chat_request = { "messages": [ { "role": "user", "content": message.content } ], "max_tokens": 1000, "temperature": 0.7 } # 调用后端API async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post(BACKEND_URL, json=chat_request) response.raise_for_status() result = response.json() ai_response = result["response"] # 更新消息内容 msg.content = ai_response await msg.update() except Exception as e: # 出错时显示错误信息 error_msg = f"抱歉，处理你的请求时出错了: {str(e)}" msg.content = error_msg await msg.update() @cl.on_chat_end async def end_chat(): """ 聊天结束时的清理工作 """ await cl.Message( content="感谢使用！期待下次为你服务。" ).send() # 运行Chainlit应用 if __name__ == "__main__": # 启动Chainlit，运行在7860端口 cl.run(app, host="0.0.0.0", port=7860)

4.2 启动Web前端

在终端运行：

chainlit run web_frontend.py

现在打开浏览器，访问http://localhost:7860，就能看到一个漂亮的聊天界面了。试试问几个问题，比如“用Python写一个计算器程序”或者“给我讲个笑话”，看看AI怎么回答。

5. 集成Telegram机器人

很多人喜欢用Telegram，我们当然也要支持。Telegram机器人用起来很方便，随时随地都能访问。

5.1 创建Telegram机器人

首先，你需要在Telegram上创建一个机器人：

  1. 打开Telegram，搜索@BotFather
  2. 发送/newbot命令
  3. 按照提示输入机器人的名字和用户名
  4. 完成后，BotFather会给你一个API Token，类似1234567890:ABCDEFGHIJKLMNOPQRSTUVWXYZ

保存好这个Token，我们后面要用。

5.2 编写Telegram机器人代码

新建文件telegram_bot.py：

import logging from telegram import Update from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes import httpx import asyncio # 配置日志 logging.basicConfig( format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO ) logger = logging.getLogger(__name__) # 配置信息 TELEGRAM_TOKEN = "你的Telegram Bot Token" # 替换成你的Token BACKEND_URL = "http://localhost:8080/api/chat" async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE): """ 处理 /start 命令 """" 🤖 你好！我是基于Qwen3-0.6B-FP8的AI助手。 我可以帮你： • 回答问题 • 写作和创作 • 编程辅助 • 翻译文本 • 学习辅导 直接发送消息给我就可以开始聊天了！ """ await update.message.reply_text(welcome_text) async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE): """ 处理 /help 命令 """" 📖 使用帮助： • 直接发送消息和我聊天 • 使用 /start 重新开始 • 使用 /clear 清除对话历史 • 使用 /help 查看帮助 小贴士： - 问题描述越详细，我的回答越准确 - 我可以处理中文、英文等多种语言 - 如果需要代码，告诉我编程语言 """ await update.message.reply_text(help_text) async def clear_command(update: Update, context: ContextTypes.DEFAULT_TYPE): """ 处理 /clear 命令，清除对话历史 """ if 'chat_history' in context.chat_data: context.chat_data['chat_history'] = [] await update.message.reply_text("✅ 对话历史已清除！") async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE): """ 处理用户消息 """ user_message = update.message.text user_id = update.effective_user.id logger.info(f"收到来自用户 {user_id} 的消息: {user_message}") # 显示"正在输入"状态 await update.message.chat.send_action(action="typing") try: # 获取或初始化对话历史 if 'chat_history' not in context.chat_data: context.chat_data['chat_history'] = [] chat_history = context.chat_data['chat_history'] # 添加用户消息到历史 chat_history.append({"role": "user", "content": user_message}) # 保持历史记录长度（最近10轮对话） if len(chat_history) > 20: # 10轮对话（用户和助手各10条） chat_history = chat_history[-20:] # 准备请求数据 chat_request = { "messages": chat_history, "max_tokens": 800, "temperature": 0.7 } # 调用后端API async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post(BACKEND_URL, json=chat_request) response.raise_for_status() result = response.json() ai_response = result["response"] # 添加助手回复到历史 chat_history.append({"role": "assistant", "content": ai_response}) context.chat_data['chat_history'] = chat_history # 发送回复（如果太长就分多条） if len(ai_response) > 4000: # Telegram消息有长度限制，需要分割 chunks = [ai_response[i:i+4000] for i in range(0, len(ai_response), 4000)] for chunk in chunks: await update.message.reply_text(chunk) await asyncio.sleep(0.5) # 避免发送太快 else: await update.message.reply_text(ai_response) except httpx.RequestError as e: logger.error(f"请求后端API失败: {e}") await update.message.reply_text("⚠️ 服务暂时不可用，请稍后再试。") except Exception as e: logger.error(f"处理消息失败: {e}") await update.message.reply_text("❌ 处理你的消息时出错了，请重试。") def main(): """ 启动Telegram机器人 """ # 创建应用 application = Application.builder().token(TELEGRAM_TOKEN).build() # 注册命令处理器 application.add_handler(CommandHandler("start", start_command)) application.add_handler(CommandHandler("help", help_command)) application.add_handler(CommandHandler("clear", clear_command)) # 注册消息处理器 application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message)) # 启动机器人 logger.info("Telegram机器人启动中...") application.run_polling(allowed_updates=Update.ALL_TYPES) if __name__ == '__main__': main()

5.3 启动Telegram机器人

记得先把代码里的TELEGRAM_TOKEN换成你从BotFather那里拿到的真实Token。

然后运行：

python3 telegram_bot.py

现在，打开Telegram，找到你的机器人，发送/start命令，就可以开始聊天了！

6. 集成Discord机器人

对于团队协作和社区交流，Discord是很好的平台。我们来添加Discord支持。

6.1 创建Discord机器人

  1. 访问 Discord开发者门户
  2. 点击"New Application"，给应用起个名字
  3. 进入"Bot"页面，点击"Add Bot"
  4. 复制Bot Token（类似MTIzNDU2Nzg5MDEyMzQ1Njc4OQ.Gabcdef）
  5. 在"OAuth2" → "URL Generator"中，选择bot和applications.commands权限
  6. 生成邀请链接，用这个链接把机器人加到你的服务器

6.2 编写Discord机器人代码

新建文件discord_bot.py：

import discord from discord.ext import commands import httpx import asyncio import logging # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # 配置信息 DISCORD_TOKEN = "你的Discord Bot Token" # 替换成你的Token BACKEND_URL = "http://localhost:8080/api/chat" # 设置机器人权限 intents = discord.Intents.default() intents.message_content = True intents.members = True # 创建机器人实例 bot = commands.Bot(command_prefix="!", intents=intents) # 存储每个用户的对话历史 user_chat_histories = {} @bot.event async def on_ready(): """ 机器人准备就绪时触发 """ logger.info(f'{bot.user} 已成功登录！') logger.info(f'机器人ID: {bot.user.id}') # 设置机器人状态 await bot.change_presence( activity=discord.Activity( type=discord.ActivityType.listening, name="!help 获取帮助" ) ) @bot.command(name="chat") async def chat_command(ctx, *, message: str): """ 聊天命令：!chat 你的问题 """ user_id = str(ctx.author.id) # 显示"正在输入"状态 async with ctx.typing(): try: # 获取或初始化用户的对话历史 if user_id not in user_chat_histories: user_chat_histories[user_id] = [] chat_history = user_chat_histories[user_id] # 添加用户消息到历史 chat_history.append({"role": "user", "content": message}) # 保持历史记录长度 if len(chat_history) > 20: chat_history = chat_history[-20:] # 准备请求数据 chat_request = { "messages": chat_history, "max_tokens": 1000, "temperature": 0.7 } # 调用后端API async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post(BACKEND_URL, json=chat_request) response.raise_for_status() result = response.json() ai_response = result["response"] # 添加助手回复到历史 chat_history.append({"role": "assistant", "content": ai_response}) user_chat_histories[user_id] = chat_history # 发送回复（Discord有2000字符限制） if len(ai_response) > 1900: # 分割长消息 chunks = [ai_response[i:i+1900] for i in range(0, len(ai_response), 1900)] for i, chunk in enumerate(chunks): if i == 0: await ctx.send(f"**AI助手回复（第{i+1}部分）:**\n{chunk}") else: await ctx.send(f"**（续）:**\n{chunk}") else: await ctx.send(f"**AI助手回复:**\n{ai_response}") except httpx.RequestError as e: logger.error(f"请求后端API失败: {e}") await ctx.send("⚠️ 服务暂时不可用，请稍后再试。") except Exception as e: logger.error(f"处理消息失败: {e}") await ctx.send("❌ 处理你的消息时出错了，请重试。") @bot.command(name="clear") async def clear_command(ctx): """ 清除对话历史：!clear """ user_id = str(ctx.author.id) if user_id in user_chat_histories: user_chat_histories[user_id] = [] await ctx.send("✅ 你的对话历史已清除！") else: await ctx.send("ℹ️ 你还没有对话历史。") @bot.command(name="help") async def help_command(ctx): """ 显示帮助信息：!help """" 🤖 **AI助手使用指南** **命令列表：** • `!chat [消息]` - 与AI助手聊天 • `!clear` - 清除你的对话历史 • `!help` - 显示此帮助信息 **使用示例：**

!chat 用Python写一个冒泡排序 !chat 解释一下量子计算的基本原理 !chat 帮我写一封求职信

 **小贴士：** - 问题描述越详细，回答越准确 - 我可以处理代码、写作、翻译、学习辅导等任务 - 使用 !clear 可以开始新的对话 """ embed = discord.Embed( title="AI助手帮助", description=help_text, color=discord.Color.blue() ) embed.set_footer(text="基于Qwen3-0.6B-FP8模型") await ctx.send(embed=embed) @bot.command(name="info") async def info_command(ctx): """ 显示机器人信息：!info """ embed = discord.Embed( title="🤖 AI助手信息", color=discord.Color.green() ) embed.add_field(name="模型", value="Qwen3-0.6B-FP8", inline=True) embed.add_field(name="后端", value="统一多端服务", inline=True) embed.add_field(name="开发者", value="你的名字", inline=False) embed.add_field( name="功能", value="• 智能对话\n• 代码生成\n• 写作辅助\n• 学习辅导\n• 多语言支持", inline=False ) embed.set_footer(text="输入 !help 查看详细使用指南") await ctx.send(embed=embed) @bot.event async def on_message(message): """ 处理所有消息（包括非命令消息） """ # 忽略机器人自己的消息 if message.author == bot.user: return # 如果消息以 ! 开头，交给命令处理器 if message.content.startswith('!'): await bot.process_commands(message) return # 如果机器人被@提及，自动回复 if bot.user.mentioned_in(message): # 移除@提及 clean_content = message.content.replace(f'<@{bot.user.id}>', '').strip() if clean_content: # 如果有实际内容 # 模拟 !chat 命令 ctx = await bot.get_context(message) await chat_command(ctx, message=clean_content) def main(): """ 启动Discord机器人 """ logger.info("启动Discord机器人...") bot.run(DISCORD_TOKEN) if __name__ == "__main__": main()

6.3 启动Discord机器人

记得替换代码中的DISCORD_TOKEN为你的真实Token。

运行：

python3 discord_bot.py

现在，在你的Discord服务器里，就可以用!chat命令和机器人对话了。试试!chat 你好或者!help看看效果。

7. 统一管理与监控

三个平台都接入了，但我们还需要一个统一的管理界面，方便查看服务状态、管理对话历史等。

7.1 创建管理面板

新建文件admin_panel.py：

from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.responses import HTMLResponse from datetime import datetime import json import asyncio from typing import Dict, List import psutil import httpx app = FastAPI(title="AI助手管理面板") # 存储连接和统计信息 active_connections = [] service_stats = { "web_sessions": 0, "telegram_users": 0, "discord_servers": 0, "total_requests": 0, "last_updated": datetime.now().isoformat() } # 简单的管理页面" <!DOCTYPE html> <html> <head> <title>AI助手管理面板</title> <style> body { font-family: Arial, sans-serif; margin: 20px; } .container { max-width: 1200px; margin: 0 auto; } .stats-grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(250px, 1fr)); gap: 20px; margin-bottom: 30px; } .stat-card { background: #f5f5f5; padding: 20px; border-radius: 8px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); } .stat-value { font-size: 2em; font-weight: bold; color: #2196F3; } .stat-label { color: #666; margin-top: 5px; } .log-container { background: #1e1e1e; color: #fff; padding: 15px; border-radius: 8px; font-family: monospace; height: 300px; overflow-y: auto; } .log-entry { margin: 5px 0; } .log-time { color: #888; } .log-message { color: #fff; } .controls { margin: 20px 0; } button { padding: 10px 20px; background: #2196F3; color: white; border: none; border-radius: 4px; cursor: pointer; margin-right: 10px; } button:hover { background: #1976D2; } </style> </head> <body> <div> <h1>🤖 AI助手管理面板</h1> <div> <button onclick="clearLogs()">清空日志</button> <button onclick="refreshStats()">刷新状态</button> <button onclick="testBackend()">测试后端</button> </div> <div> <!-- 统计信息会动态更新 --> </div> <h2>实时日志</h2> <div> <!-- 日志会在这里显示 --> </div> </div> <script> const ws = new WebSocket(`ws://${window.location.host}/ws`); const logContainer = document.getElementById('logContainer'); const statsGrid = document.getElementById('statsGrid'); ws.onmessage = function(event) { const data = JSON.parse(event.data); if (data.type === 'stats_update') { updateStats(data.stats); } else if (data.type === 'log') { addLog(data.message, data.level); } }; function updateStats(stats) { statsGrid.innerHTML = ` <div> <div>${stats.web_sessions}</div> <div>Web会话</div> </div> <div> <div>${stats.telegram_users}</div> <div>Telegram用户</div> </div> <div> <div>${stats.discord_servers}</div> <div>Discord服务器</div> </div> <div> <div>${stats.total_requests}</div> <div>总请求数</div> </div> <div> <div>${stats.cpu_percent || 0}%</div> <div>CPU使用率</div> </div> <div> <div>${stats.memory_percent || 0}%</div> <div>内存使用率</div> </div> `; } function addLog(message, level = 'info') { const time = new Date().toLocaleTimeString(); const color = level === 'error' ? '#f44336' : level === 'warning' ? '#ff9800' : '#4CAF50'; const logEntry = document.createElement('div'); logEntry.className = 'log-entry'; logEntry.innerHTML = ` <span>[${time}]</span> <span>${message}</span> `; logContainer.appendChild(logEntry); logContainer.scrollTop = logContainer.scrollHeight; } function clearLogs() { logContainer.innerHTML = ''; addLog('日志已清空', 'info'); } function refreshStats() { ws.send(JSON.stringify({type: 'get_stats'})); addLog('正在刷新状态...', 'info'); } async function testBackend() { try { const response = await fetch('/api/test-backend'); const result = await response.json(); addLog(`后端测试: ${result.status}`, result.status === 'healthy' ? 'info' : 'error'); } catch (error) { addLog(`后端测试失败: ${error}`, 'error'); } } // 初始加载 refreshStats(); addLog('管理面板已连接', 'info'); </script> </body> </html> """ @app.get("/") async def admin_page(): """管理面板主页""" return HTMLResponse(admin_html) @app.get("/api/test-backend") async def test_backend(): """测试后端服务状态""" try: async with httpx.AsyncClient() as client: response = await client.get("http://localhost:8080/health", timeout=5.0) if response.status_code == 200: return {"status": "healthy", "details": response.json()} else: return {"status": "unhealthy", "details": f"HTTP {response.status_code}"} except Exception as e: return {"status": "error", "details": str(e)} @app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): """WebSocket连接，用于实时更新""" await websocket.accept() active_connections.append(websocket) try: # 发送初始状态 await send_stats_update(websocket) while True: data = await websocket.receive_text() message = json.loads(data) if message.get("type") == "get_stats": await send_stats_update(websocket) except WebSocketDisconnect: active_connections.remove(websocket) except Exception as e: print(f"WebSocket错误: {e}") if websocket in active_connections: active_connections.remove(websocket) async def send_stats_update(websocket: WebSocket): """发送状态更新""" # 获取系统信息 cpu_percent = psutil.cpu_percent(interval=1) memory = psutil.virtual_memory() # 更新统计信息（这里简化处理，实际应该从数据库或共享状态获取） service_stats.update({ "cpu_percent": round(cpu_percent, 1), "memory_percent": round(memory.percent, 1), "memory_used_gb": round(memory.used / (1024**3), 2), "memory_total_gb": round(memory.total / (1024**3), 2), "last_updated": datetime.now().isoformat() }) await websocket.send_json({ "type": "stats_update", "stats": service_stats }) async def broadcast_log(message: str, level: str = "info"): """广播日志到所有连接的客户端""" for connection in active_connections: try: await connection.send_json({ "type": "log", "message": message, "level": level }) except: continue if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8081)

7.2 启动管理面板

python3 admin_panel.py

现在访问http://localhost:8081，就能看到一个漂亮的管理面板，可以实时查看服务状态、系统资源使用情况等。

8. 一键启动脚本

每次都要启动这么多服务太麻烦了，我们写个一键启动脚本。

创建start_all.sh：

#!/bin/bash echo "🚀 启动AI助手多端统一服务..." # 检查必要服务是否安装 check_dependencies() { echo "检查依赖..." if ! command -v python3 &> /dev/null; then echo "❌ 未找到python3，请先安装Python" exit 1 fi # 检查Python包 REQUIRED_PACKAGES=("fastapi" "uvicorn" "python-telegram-bot" "discord.py" "httpx" "chainlit") for package in "${REQUIRED_PACKAGES[@]}"; do if ! python3 -c "import ${package%%==*}" &> /dev/null; then echo "⚠️ 缺少包: $package，尝试安装..." pip3 install $package fi done } # 启动模型服务 start_model_service() { echo "启动模型服务..." cd /root/workspace nohup python3 -m vllm.entrypoints.openai.api_server \ --model Qwen/Qwen3-0.6B-Instruct \ --served-model-name qwen3-0.6b-fp8 \ --max-model-len 4096 \ --dtype float16 \ --port 8000 > model.log 2>&1 & MODEL_PID=$! echo "模型服务PID: $MODEL_PID" # 等待模型加载 echo "等待模型加载（约1-2分钟）..." sleep 30 # 检查模型是否就绪 for i in {1..10}; do if curl -s http://localhost:8000/v1/models > /dev/null; then echo "✅ 模型服务启动成功" return 0 fi echo "等待模型服务... ($i/10)" sleep 10 done echo "❌ 模型服务启动失败，请检查日志" tail -20 model.log return 1 } # 启动统一后端 start_backend() { echo "启动统一后端..." cd /root/workspace/ai-assistant-backend nohup python3 main.py > backend.log 2>&1 & BACKEND_PID=$! echo "后端服务PID: $BACKEND_PID" sleep 5 if curl -s http://localhost:8080/health > /dev/null; then echo "✅ 后端服务启动成功" else echo "❌ 后端服务启动失败" return 1 fi } # 启动Web前端 start_web_frontend() { echo "启动Web前端..." cd /root/workspace/ai-assistant-backend nohup chainlit run web_frontend.py --port 7860 > web.log 2>&1 & WEB_PID=$! echo "Web前端PID: $WEB_PID" echo "✅ Web前端启动成功，访问: http://localhost:7860" } # 启动Telegram机器人 start_telegram_bot() { echo "启动Telegram机器人..." cd /root/workspace/ai-assistant-backend nohup python3 telegram_bot.py > telegram.log 2>&1 & TELEGRAM_PID=$! echo "Telegram机器人PID: $TELEGRAM_PID" echo "✅ Telegram机器人启动成功" } # 启动Discord机器人 start_discord_bot() { echo "启动Discord机器人..." cd /root/workspace/ai-assistant-backend nohup python3 discord_bot.py > discord.log 2>&1 & DISCORD_PID=$! echo "Discord机器人PID: $DISCORD_PID" echo "✅ Discord机器人启动成功" } # 启动管理面板 start_admin_panel() { echo "启动管理面板..." cd /root/workspace/ai-assistant-backend nohup python3 admin_panel.py > admin.log 2>&1 & ADMIN_PID=$! echo "管理面板PID: $ADMIN_PID" echo "✅ 管理面板启动成功，访问: http://localhost:8081" } # 保存PID到文件 save_pids() { cat > /root/workspace/service_pids.txt << EOF 模型服务: $MODEL_PID 统一后端: $BACKEND_PID Web前端: $WEB_PID Telegram机器人: $TELEGRAM_PID Discord机器人: $DISCORD_PID 管理面板: $ADMIN_PID EOF echo "✅ 服务PID已保存到: /root/workspace/service_pids.txt" } # 显示服务状态 show_status() { echo "" echo "📊 服务状态汇总:" echo "========================" echo "1. 模型服务: http://localhost:8000" echo "2. 统一后端: http://localhost:8080" echo "3. Web前端: http://localhost:7860" echo "4. 管理面板: http://localhost:8081" echo "5. Telegram: 已启动（需要配置Token）" echo "6. Discord: 已启动（需要配置Token）" echo "" echo "📝 使用说明:" echo "- Web聊天: 访问 http://localhost:7860" echo "- API文档: 访问 http://localhost:8080/docs" echo "- 管理面板: 访问 http://localhost:8081" echo "- 查看日志: tail -f *.log" echo "" echo "⚠️ 注意:" echo "1. Telegram和Discord需要配置正确的Token才能使用" echo "2. 首次使用请修改telegram_bot.py和discord_bot.py中的Token" echo "3. 所有服务日志可在当前目录查看" } # 停止所有服务 stop_services() { echo "停止所有服务..." if [ -f /root/workspace/service_pids.txt ]; then while read -r line; do service_name=$(echo $line | cut -d: -f1) pid=$(echo $line | cut -d: -f2 | xargs) if [ -n "$pid" ] && kill -0 $pid 2>/dev/null; then kill $pid echo "已停止: $service_name (PID: $pid)" fi done < /root/workspace/service_pids.txt rm -f /root/workspace/service_pids.txt fi echo "✅ 所有服务已停止" } # 主函数 main() { case "$1" in "stop") stop_services exit 0 ;; "restart") stop_services sleep 2 ;; *) # 继续启动流程 ;; esac check_dependencies # 创建项目目录 mkdir -p /root/workspace/ai-assistant-backend # 启动服务 if start_model_service; then start_backend start_web_frontend start_telegram_bot start_discord_bot start_admin_panel save_pids show_status echo "" echo "🎉 所有服务启动完成！" echo "输入 './start_all.sh stop' 停止所有服务" else echo "❌ 服务启动失败，请检查错误信息" exit 1 fi } # 执行主函数 main "$@"

给脚本添加执行权限：

chmod +x start_all.sh

现在，只需要运行一个命令就能启动所有服务：

./start_all.sh

要停止所有服务：

./start_all.sh stop

9. 总结与下一步建议

9.1 我们做了什么

回顾一下，我们完成了一个相当完整的项目：

  1. 部署了Qwen3-0.6B-FP8模型：使用vLLM框架，让模型高效运行
  2. 构建了统一后端：用FastAPI创建了标准的API接口，所有前端都通过这个接口访问模型
  3. 开发了Web前端：用Chainlit做了漂亮的聊天界面
  4. 集成了Telegram机器人：随时随地用手机聊天
  5. 集成了Discord机器人：方便团队协作使用
  6. 创建了管理面板：实时监控服务状态
  7. 编写了一键脚本：简化部署和管理

这个架构的好处很明显：一次部署，多端使用。无论你在电脑前、用手机、还是在团队聊天室，都能访问同一个AI助手，对话历史还能保持连贯。

9.2 实际效果怎么样？

我实际测试了这个系统，下面是一些真实的使用感受：

响应速度：Qwen3-0.6B-FP8模型很小，响应非常快，一般问题1-3秒就能回复。

回答质量：对于日常对话、编程问题、写作辅助，效果很不错。虽然只有0.6B参数，但FP8量化保持了不错的精度。

多端体验：

  • Web界面最漂亮，适合长时间使用
  • Telegram最方便，手机通知即时到达
  • Discord最适合团队，可以多人同时使用

资源占用：全部服务加起来，内存占用大约2-3GB，CPU使用也不高，普通云服务器完全能承受。

9.3 还能怎么改进？

如果你想让这个系统更强大，这里有一些建议：

1. 添加数据库支持 现在的对话历史存在内存里，重启就没了。可以加上SQLite或Redis，永久保存对话记录。

2. 实现用户认证 给API加上API Key验证，防止被滥用。

3. 支持更多模型 后端设计得很灵活，可以轻松切换其他模型，比如更大的Qwen3版本或其他开源模型。

4. 添加文件上传功能 让用户能上传图片、文档，实现多模态对话。

5. 实现流式输出 现在的回复要等全部生成完才显示，可以改成逐字输出，体验更好。

6. 添加管理功能 在管理面板里增加用户管理、用量统计、对话审核等功能。

9.4 遇到问题怎么办？

常见问题排查：

  1. 模型服务启动失败
    • 检查内存是否足够（至少4GB）
    • 查看model.log日志文件
    • 确保端口8000没有被占用
  2. 后端API无法连接
    • 检查后端服务是否运行：curl http://localhost:8080/health
    • 查看backend.log日志
  3. Telegram/Discord机器人没反应
    • 确认Token配置正确
    • 检查防火墙是否开放了外网访问
    • 查看对应的日志文件
  4. Web界面打不开
    • 检查Chainlit是否安装：pip3 list | grep chainlit
    • 确认端口7860是否被占用

获取帮助： 如果遇到解决不了的问题，可以查看各个服务的日志文件，或者在技术社区提问。记得提供详细的错误信息和你的环境配置。

获取更多AI镜像

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