点【关注+收藏】实时获取最新的实战代码案例

效果图

基于火山方舟&豆包API的全屏实时聊天应用

本文档提供完整的实时聊天应用实现方案，包含全屏居中布局、WebSocket实时通信、豆包API集成（基于火山方舟平台），且已移除Token验证，可直接部署使用。

一、项目结构说明

volcano_ark_chat/          # 项目根目录
├── static/                # 静态资源目录（前端页面）
│   └── chat.html          # 全屏聊天界面（核心前端文件）
├── .env                   # 环境变量配置（API密钥等敏感信息）
├── main.py                # 后端核心逻辑（FastAPI+WebSocket+API调用）
└── README.md              # 项目说明文档（本文档）

二、文件内容详解

1. 环境变量配置文件：.env

存储火山方舟API的密钥、URL和模型信息，避免硬编码敏感数据。

# 火山方舟平台API配置（替换为你的实际信息）
VOLCANO_ARK_API_URL=https://ark.cn-beijing.volces.com/api/v3/chat/completions
VOLCANO_ARK_API_KEY=73556896-33e7-4304-9989-c608509397f4ysp  # 你的API密钥
VOLCANO_ARK_MODEL=doubao-seed-1-6-250615                  # 豆包模型名

2. 后端核心逻辑：main.py

基于FastAPI实现WebSocket实时通信，集成豆包API调用，支持消息广播和历史记录。

import os
import json
import asyncio
import aiohttp
from datetime import datetime
from dotenv import load_dotenv
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from fastapi.responses import HTMLResponse

# 加载环境变量
load_dotenv()
app = FastAPI(title="火山方舟实时聊天应用", version="1.0")

# 挂载静态文件（前端页面）
app.mount("/static", StaticFiles(directory="static"), name="static")
templates = Jinja2Templates(directory="static")

# 全局变量：存储消息历史（内存存储，生产环境可替换为Redis）
message_history = []
MAX_HISTORY_LENGTH = 100# 最大历史消息数（避免内存溢出）

# 火山方舟API配置（从.env读取）
ARK_API_URL = os.getenv("VOLCANO_ARK_API_URL")
ARK_API_KEY = os.getenv("VOLCANO_ARK_API_KEY")
ARK_MODEL = os.getenv("VOLCANO_ARK_MODEL")


# ------------------------------
# WebSocket连接管理器（管理在线用户）
# ------------------------------
classConnectionManager:
def__init__(self):
        self.active_connections: list[WebSocket] = []  # 存储所有活跃连接

asyncdefconnect(self, websocket: WebSocket):
"""新用户连接"""
await websocket.accept()
        self.active_connections.append(websocket)
# 发送历史消息给新连接用户
await self.send_history(websocket)

defdisconnect(self, websocket: WebSocket):
"""用户断开连接"""
if websocket in self.active_connections:
            self.active_connections.remove(websocket)

asyncdefbroadcast(self, message: dict):
"""广播消息给所有在线用户"""
# 先将消息加入历史记录
global message_history
        message_history.append(message)
# 截断历史记录（保持最大长度）
if len(message_history) > MAX_HISTORY_LENGTH:
            message_history = message_history[-MAX_HISTORY_LENGTH:]

# 广播消息（JSON字符串格式）
for connection in self.active_connections:
await connection.send_text(json.dumps({
"type": "new_message",
"message": message
            }))

asyncdefsend_history(self, websocket: WebSocket):
"""发送历史消息给指定用户"""
await websocket.send_text(json.dumps({
"type": "history_messages",
"messages": message_history
        }))


# 初始化连接管理器
manager = ConnectionManager()


# ------------------------------
# 豆包API调用（异步）
# ------------------------------
asyncdefcall_doubao_api(user_query: str) -> str:
"""调用火山方舟豆包API，获取AI回复"""
ifnot ARK_API_KEY ornot ARK_API_URL:
return"错误：未配置火山方舟API密钥或URL"

# API请求参数（符合火山方舟格式）
    payload = {
"model": ARK_MODEL,
"messages": [
            {"role": "system", "content": "你是一个友好的智能助手，回答简洁准确"},
            {"role": "user", "content": user_query}
        ],
"temperature": 0.7,
"max_tokens": 1000
    }

# 请求头（包含认证信息）
    headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {ARK_API_KEY}"
    }

try:
# 异步发送API请求（避免阻塞WebSocket）
asyncwith aiohttp.ClientSession() as session:
asyncwith session.post(
                ARK_API_URL,
                headers=headers,
                json=payload,
                timeout=15# 15秒超时
            ) as response:
if response.status == 200:
                    result = await response.json()
return result["choices"][0]["message"]["content"].strip()
else:
                    error_text = await response.text()
returnf"AI请求失败（{response.status}）：{error_text[:100]}"
except Exception as e:
returnf"AI调用异常：{str(e)[:80]}"


# ------------------------------
# 路由定义
# ------------------------------
@app.get("/", response_class=HTMLResponse)
asyncdefindex():
"""根路径：返回全屏聊天页面"""
return templates.TemplateResponse("chat.html", {"request": {}})


@app.websocket("/ws")
asyncdefwebsocket_endpoint(websocket: WebSocket):
"""WebSocket实时通信端点（核心功能）"""
# 1. 建立连接
await manager.connect(websocket)

try:
# 2. 循环监听客户端消息
whileTrue:
# 接收客户端发送的JSON消息
            data = await websocket.receive_text()
            client_msg = json.loads(data)

# 验证消息格式（必须包含type、username、message）
ifnot all(key in client_msg for key in ["type", "username", "message"]):
await websocket.send_text(json.dumps({
"type": "error",
"message": {"content": "消息格式错误，请重试"}
                }))
continue

# 处理"发送消息"请求
if client_msg["type"] == "send_message":
                username = client_msg["username"].strip() or"匿名用户"
                content = client_msg["content"].strip()
                timestamp = datetime.now().strftime("%H:%M:%S")  # 消息时间戳

# 情况1：普通用户消息（直接广播）
ifnot content.startswith("@豆包"):
                    user_message = {
"username": username,
"content": content,
"timestamp": timestamp,
"is_ai": False# 标记为非AI消息
                    }
await manager.broadcast(user_message)

# 情况2：AI交互消息（调用豆包API后广播）
else:
# 1. 先广播用户的AI请求（告知其他用户）
                    ai_request_msg = {
"username": username,
"content": content,
"timestamp": timestamp,
"is_ai": False
                    }
await manager.broadcast(ai_request_msg)

# 2. 提取用户真实查询（去掉"@豆包"前缀）
                    user_query = content.replace("@豆包", "").strip()
ifnot user_query:
                        user_query = "你好，我想和你聊天"

# 3. 异步调用豆包API（不阻塞WebSocket循环）
                    ai_response = await call_doubao_api(user_query)

# 4. 广播AI回复
                    ai_message = {
"username": "豆包AI",
"content": ai_response,
"timestamp": datetime.now().strftime("%H:%M:%S"),
"is_ai": True# 标记为AI消息（前端区分样式）
                    }
await manager.broadcast(ai_message)

# 3. 处理客户端断开连接
except WebSocketDisconnect:
        manager.disconnect(websocket)
except Exception as e:
# 捕获未知异常，避免服务崩溃
        print(f"WebSocket异常：{str(e)}")
        manager.disconnect(websocket)
await websocket.close(code=1011)  # 服务器内部错误


# ------------------------------
# 启动服务（直接运行main.py即可）
# ------------------------------
if __name__ == "__main__":
import uvicorn
    uvicorn.run(
        app="main:app",
        host="0.0.0.0",
        port=8000,
        reload=True# 开发模式热重载（生产环境关闭）
    )

3. 前端全屏聊天界面：static/chat.html

全屏布局，对话框居中占据大部分空间，支持响应式适配，区分用户/AI/系统消息样式。