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

深度集成DeepSeek大模型：WebSocket流式聊天实现

news 2025/8/8 18:23:34

5分钟快速接入DeepSeek大模型：WebSocket实时聊天指南

创建应用

访问DeepSeek官网
- 前往 DeepSeek官网。
- 如果还没有账号，需要先注册一个。
进入API开放平台
- 点击右上角的“API 开放平台”按钮，进入后台管理页面。
- 你可以看到默认赠送的10元免费额度。
创建API Key
- 点击“API keys”，进入Key管理页面。
- 创建一个新的API Key，并确保保存好这个Key，因为一旦离开页面就无法再次查看。

开发后端代码 (Python/Node.js)

推荐使用Python或Node.js编写后台代码。下面以Python为例进行说明。

安装OpenAI库
```
pip3 install openai
```

引入OpenAI并创建DeepSeek客户端实例

from openai import OpenAIclient = OpenAI(api_key="替换为你的key", base_url="https://api.deepseek.com")

发送请求

response = client.chat.completions.create(model="deepseek-chat",messages=[{"role": "user", "content": "详细介绍一下你自己"},],stream=False
)
print(response.choices[0].message.content)

输出结果如下：
在这里插入图片描述

流式输出结果
将请求参数 stream 设置为 True 即可实现流式输出。

response = client.chat.completions.create(model="deepseek-chat",messages=[{"role": "user", "content": "详细介绍一下你自己"},],stream=True
)
for chunk in response:print(chunk.choices[0].delta.content, end='', flush=True)

接口封装
将功能封装为WebSocket接口，实现实时聊天。

import asyncio
import json
from websockets import serve
from openai import OpenAIclient = OpenAI(api_key="替换为你的key", base_url="https://api.deepseek.com")async def chat_handler(websocket):async for message in websocket:data = json.loads(message)user_input = data.get('user_input', '')history = data.get('history', [])messages = [{"role": "system", "content": "You are a helpful assistant"}]messages.extend(history)messages.append({"role": "user", "content": user_input})response = client.chat.completions.create(model="deepseek-chat",messages=messages,stream=True)for chunk in response:content = chunk.choices[0].delta.contentif content:await websocket.send(json.dumps({'content': content}, ensure_ascii=False))async def main():async with serve(chat_handler, "localhost", 8765):await asyncio.Future()  # Run foreverif __name__ == "__main__":asyncio.run(main())