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

WebSocket 使用

news 2025/11/6 8:38:35

下面是一个完整的 Python WebSocket 服务端代码，使用 websockets 库实现，配合你之前提供的前端 HTML 页面，可以实现一个简单的 AI 对话系统（此处 AI 用一个模拟函数代替，你可以替换成调用真实大模型 API，如通义千问、OpenAI 等）。

功能说明

启动 WebSocket 服务在 ws://localhost:8765
接收用户消息（JSON 格式：{"type": "user", "content": "你好"}）
模拟 AI 回复（你可以替换为真实 AI 调用）
返回格式：{"type": "ai", "content": "你好！我是AI。"}
支持多客户端连接

前端适配说明

请将你之前 HTML 中的 WebSocket 地址改为：

const ws = new WebSocket('ws://localhost:8765');

<!DOCTYPE html>
<html lang="zh-CN">
<head><meta charset="UTF-8" /><meta name="viewport" content="width=device-width, initial-scale=1.0"/><title>AI 对话 (WebSocket + 自动重连)</title><style>body {font-family: 'Segoe UI', Arial, sans-serif;max-width: 800px;margin: 20px auto;padding: 20px;background-color: #fafafa;}h2 {text-align: center;color: #333;}#status {text-align: center;padding: 8px;margin-bottom: 10px;font-size: 0.95em;color: #555;border-radius: 4px;}#chatBox {height: 400px;overflow-y: auto;border: 1px solid #ddd;padding: 12px;background: white;border-radius: 8px;box-shadow: 0 1px 3px rgba(0,0,0,0.1);margin-bottom: 12px;}.user-msg {text-align: right;color: #007bff;margin: 8px 0;}.ai-msg {text-align: left;color: #28a745;margin: 8px 0;}.system-msg {text-align: center;color: #6c757d;font-style: italic;margin: 6px 0;}#inputArea {display: flex;gap: 8px;}#messageInput {flex: 1;padding: 10px;border: 1px solid #ccc;border-radius: 4px;font-size: 1em;}#sendBtn {padding: 10px 20px;background-color: #007bff;color: white;border: none;border-radius: 4px;cursor: pointer;font-size: 1em;}#sendBtn:hover {background-color: #0056b3;}#sendBtn:disabled {background-color: #ccc;cursor: not-allowed;}</style>
</head>
<body><h2>🤖 AI 智能对话</h2><div id="status">正在连接到 AI 服务...</div><div id="chatBox"></div><div id="inputArea"><input type="text" id="messageInput" placeholder="输入你的问题，按回车发送..." autocomplete="off" /><button id="sendBtn" onclick="sendMessage()">发送</button></div><script>// WebSocket 管理let ws = null;let reconnectAttempts = 0;const MAX_RECONNECT_ATTEMPTS = 5;const RECONNECT_DELAY = 2000; // 2秒const chatBox = document.getElementById('chatBox');const statusEl = document.getElementById('status');const messageInput = document.getElementById('messageInput');const sendBtn = document.getElementById('sendBtn');function updateStatus(text, isError = false) {statusEl.textContent = text;statusEl.style.color = isError ? 'red' : '#555';}function addMessage(sender, text, role) {const msgDiv = document.createElement('div');msgDiv.className = `${role}-msg`;if (role === 'system') {msgDiv.classList.add('system-msg');msgDiv.textContent = text;} else {msgDiv.innerHTML = `<strong>${sender}:</strong> ${text}`;}chatBox.appendChild(msgDiv);chatBox.scrollTop = chatBox.scrollHeight;}function connectWebSocket() {ws = new WebSocket('ws://localhost:8765');ws.onopen = () => {updateStatus('✅ 已连接到 AI 服务');addMessage('系统', '连接成功，可以开始对话了！', 'system');reconnectAttempts = 0;sendBtn.disabled = false;};ws.onmessage = (event) => {try {const data = JSON.parse(event.data);if (data.type === 'ai') {addMessage('AI', data.content, 'ai');} else if (data.type === 'error') {addMessage('系统', '⚠️ ' + data.content, 'system');}} catch (e) {console.error('消息解析失败:', e);}};ws.onerror = (error) => {console.error('WebSocket 错误:', error);};ws.onclose = () => {sendBtn.disabled = true;if (reconnectAttempts < MAX_RECONNECT_ATTEMPTS) {updateStatus(`连接断开，第 ${reconnectAttempts + 1} 次尝试重连...`);setTimeout(() => {reconnectAttempts++;connectWebSocket();}, RECONNECT_DELAY);} else {updateStatus('❌ 连接失败，请刷新页面重试', true);}};}function sendMessage() {const msg = messageInput.value.trim();if (!msg) return;// ✅ 关键：检查连接状态if (!ws || ws.readyState !== WebSocket.OPEN) {alert('尚未连接到 AI 服务，请稍等或刷新页面。');return;}addMessage('你', msg, 'user');ws.send(JSON.stringify({ type: 'user', content: msg }));messageInput.value = '';}// 回车发送messageInput.addEventListener('keypress', (e) => {if (e.key === 'Enter') {sendMessage();}});// 初始化连接connectWebSocket();</script></body>
</html>

后端使用Python（server.py）
安装依赖

pip install websockets

import asyncio
import websockets
import json
import logging# 可选：接入真实 AI（这里先用模拟回复）
def mock_ai_response(user_input: str) -> str:"""模拟 AI 回复。你可以替换为调用真实模型，如 DashScope、OpenAI 等"""if "你好" in user_input:return "你好！很高兴见到你 😊"elif "名字" in user_input:return "我是 AI 助手，你可以叫我小智。"elif "时间" in user_input:from datetime import datetimereturn f"现在是 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}。"else:return f"你说的是：“{user_input}”？我还在学习中，请多多指教！"# 存储所有连接的客户端（可选，用于广播）
connected_clients = set()async def handle_client(websocket):# 添加客户端connected_clients.add(websocket)print(f"新客户端连接: {websocket.remote_address}")try:async for message in websocket:try:data = json.loads(message)if data.get("type") == "user":user_msg = data.get("content", "").strip()print(f"收到用户消息: {user_msg}")# 调用 AI 生成回复（这里用模拟函数）ai_reply = mock_ai_response(user_msg)# 构造回复response = {"type": "ai","content": ai_reply}# 发送回复await websocket.send(json.dumps(response))else:# 未知消息类型await websocket.send(json.dumps({"type": "error","content": "未知消息类型"}))except json.JSONDecodeError:await websocket.send(json.dumps({"type": "error","content": "消息格式错误，请发送 JSON"}))except websockets.exceptions.ConnectionClosed:print(f"客户端断开连接: {websocket.remote_address}")finally:connected_clients.discard(websocket)async def main():print("WebSocket 服务器启动中... 地址: ws://localhost:8765")server = await websockets.serve(handle_client, "localhost", 8765)await server.wait_closed()if __name__ == "__main__":# 可选：设置日志logging.basicConfig(level=logging.INFO)asyncio.run(main())

启动步骤

保存服务端代码为 server.py
安装依赖：pip install websockets
运行服务端：
```
python server.py
```
打开你的 HTML 文件（建议用本地 HTTP 服务器打开，避免 file:// 协议限制，可用 python -m http.server 8000 启一个）
在浏览器中访问 http://localhost:8000，即可与 AI 对话

查看全文

http://www.dtcms.com/a/573559.html