介绍基于 Qwen3-0.6B-FP8 模型构建跨平台 AI 助手的方案。通过 vLLM 部署模型，利用 FastAPI 创建统一后端，集成 Web（Chainlit）、Telegram 及 Discord 前端。实现了多端请求转发、对话历史管理及资源监控，并提供一键启动脚本简化部署。实测显示响应速度快，资源占用低，适合中小规模应用。
想象一下这个场景：你正在电脑前写代码，突然想到一个问题，于是打开浏览器，访问一个 AI 对话页面提问。过了一会儿，你出门了，在手机上收到朋友的消息，想用同一个 AI 助手帮忙想个点子，却不得不切换到另一个 App。晚上，你和团队在 Discord 上讨论项目，又想调用 AI 来辅助决策，结果发现还得重新部署一套服务。
是不是很麻烦？这就是我们今天要解决的问题。
Qwen3-0.6B-FP8 是一个小巧但强大的语言模型，它能在资源有限的环境下流畅运行。但光有模型还不够，我们需要一个能同时服务 Web 页面、Telegram 机器人和 Discord 机器人的统一后端。这样，无论你在哪里，用什么设备，都能无缝使用同一个 AI 助手。
本文将带你一步步搭建这样一个系统。
在开始之前，确保你有以下环境：
cd、ls、pip 这些命令就行。我们先从核心开始：把模型跑起来。
第一步：安装必要的工具
打开终端，输入以下命令：
# 更新包管理器
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 数据，说明一切正常。
模型跑起来了，现在我们需要一个'中间人'——它接收来自不同平台的请求，转发给模型，再把结果返回去。
先创建一个项目文件夹，并组织好文件结构：
# 创建项目文件夹
mkdir ai-assistant-backend
cd ai-assistant-backend
# 创建必要的文件
touch main.py
touch config.py
touch requirements.txt
touch .env
编辑 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
现在我们来写主要的后端逻辑。打开 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)
这段代码做了几件事：
/api/chat 接口来处理聊天请求保存文件后，启动后端服务：
# 在后端项目目录下运行
python3 main.py
服务会运行在 8080 端口。你可以打开浏览器访问 http://localhost:8080/docs，看到自动生成的 API 文档界面。
有了后端，现在我们来给 Web 页面做个漂亮的前端。Chainlit 是一个专门为 AI 应用设计的 UI 框架，用起来特别简单。
新建一个文件 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)
在终端运行：
chainlit run web_frontend.py
现在打开浏览器，访问 http://localhost:7860，就能看到一个漂亮的聊天界面了。试试问几个问题，比如'用 Python 写一个计算器程序'或者'给我讲个笑话'，看看 AI 怎么回答。
很多人喜欢用 Telegram，我们当然也要支持。Telegram 机器人用起来很方便，随时随地都能访问。
首先，你需要在 Telegram 上创建一个机器人：
@BotFather/newbot 命令1234567890:ABCDEFGHIJKLMNOPQRSTUVWXYZ保存好这个 Token，我们后面要用。
新建文件 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 命令"""
welcome_text = """🤖 你好！我是基于 Qwen3-0.6B-FP8 的 AI 助手。
我可以帮你：
• 回答问题
• 写作和创作
• 编程辅助
• 翻译文本
• 学习辅导
直接发送消息给我就可以开始聊天了！"""
await update.message.reply_text(welcome_text)
async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""处理 /help 命令"""
help_text = """📖 使用帮助：
• 直接发送消息和我聊天
• 使用 /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()
记得先把代码里的 TELEGRAM_TOKEN 换成你从 BotFather 那里拿到的真实 Token。
然后运行：
python3 telegram_bot.py
现在，打开 Telegram，找到你的机器人，发送 /start 命令，就可以开始聊天了！
对于团队协作和社区交流，Discord 是很好的平台。我们来添加 Discord 支持。
MTIzNDU2Nzg5MDEyMzQ1Njc4OQ.Gabcdef）bot 和 applications.commands 权限新建文件 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"""
help_text = """🤖 **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="• 智能对话\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()
记得替换代码中的 DISCORD_TOKEN 为你的真实 Token。
运行：
python3 discord_bot.py
现在，在你的 Discord 服务器里，就可以用 !chat 命令和机器人对话了。试试 !chat 你好 或者 !help 看看效果。
三个平台都接入了，但我们还需要一个统一的管理界面，方便查看服务状态、管理对话历史等。
新建文件 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()
}
# 简单的管理页面
admin_html = """<!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 class="container">
<h1>🤖 AI 助手管理面板</h1>
<div class="controls">
<button onclick="clearLogs()">清空日志</button>
<button onclick="refreshStats()">刷新状态</button>
<button onclick="testBackend()">测试后端</button>
</div>
<div id="statsGrid" class="stats-grid">
<!-- 统计信息会动态更新 -->
</div>
<h2>实时日志</h2>
<div id="logContainer" class="log-container">
<!-- 日志会在这里显示 -->
</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 class="stat-card"><div class="stat-value">${stats.web_sessions}</div><div class="stat-label">Web 会话</div></div>
<div class="stat-card"><div class="stat-value">${stats.telegram_users}</div><div class="stat-label">Telegram 用户</div></div>
<div class="stat-card"><div class="stat-value">${stats.discord_servers}</div><div class="stat-label">Discord 服务器</div></div>
<div class="stat-card"><div class="stat-value">${stats.total_requests}</div><div class="stat-label">总请求数</div></div>
<div class="stat-card"><div class="stat-value">${stats.cpu_percent || 0}%</div><div class="stat-label">CPU 使用率</div></div>
<div class="stat-card"><div class="stat-value">${stats.memory_percent || 0}%</div><div class="stat-label">内存使用率</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 style="color:${color}">[${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)
python3 admin_panel.py
现在访问 http://localhost:8081，就能看到一个漂亮的管理面板，可以实时查看服务状态、系统资源使用情况等。
每次都要启动这么多服务太麻烦了，我们写个一键启动脚本。
创建 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
回顾一下，我们完成了一个相当完整的项目：
这个架构的好处很明显：一次部署，多端使用。无论你在电脑前、用手机、还是在团队聊天室，都能访问同一个 AI 助手，对话历史还能保持连贯。
该系统经过实际测试，使用感受如下：
响应速度：Qwen3-0.6B-FP8 模型很小，响应非常快，一般问题 1-3 秒就能回复。
回答质量：对于日常对话、编程问题、写作辅助，效果很不错。虽然只有 0.6B 参数，但 FP8 量化保持了不错的精度。
多端体验：
资源占用：全部服务加起来，内存占用大约 2-3GB，CPU 使用也不高，普通云服务器完全能承受。
如果你想让这个系统更强大，这里有一些建议：
1. 添加数据库支持 现在的对话历史存在内存里，重启就没了。可以加上 SQLite 或 Redis，永久保存对话记录。
2. 实现用户认证 给 API 加上 API Key 验证，防止被滥用。
3. 支持更多模型 后端设计得很灵活，可以轻松切换其他模型，比如更大的 Qwen3 版本或其他开源模型。
4. 添加文件上传功能 让用户能上传图片、文档，实现多模态对话。
5. 实现流式输出 现在的回复要等全部生成完才显示，可以改成逐字输出，体验更好。
6. 添加管理功能 在管理面板里增加用户管理、用量统计、对话审核等功能。
常见问题排查：
model.log 日志文件curl http://localhost:8080/healthbackend.log 日志pip3 list | grep chainlit获取帮助： 如果遇到解决不了的问题，可以查看各个服务的日志文件，或者在技术社区提问。记得提供详细的错误信息和你的环境配置。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online