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

news2025/3/26 7:01:51

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 OpenAI

client = 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 OpenAI

client = 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.content
            if 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 forever

if __name__ == "__main__":
    asyncio.run(main())