Godot游戏接入Deepseek

功能说明

在现在AI盛行的环境下，在游戏中接入AI是一个很常见的需求，但是由于Godot使用的gds无法使用python等语言现成的OpenAI库。所以相关需求都需要自行编写。我公布出一个已经验证可行的写法，方便需要的人使用。

前置说明

需自行购买DeepSeek官方token，购买DeepSeek平台Token

定价策略可查看链接

使用流式输出

流式输出是当前LLM较有特色的功能。AI生成的内容往往需要较长的时间，使用流式输出可以以最快的速度让用户看到输出结果。这大大优化了用户的体验。

主要代码

可直接复制并在Godot中新建一个deepseek_chat_stream.gd文件并粘贴保存后使用。

代码

gdscript

@tool
class_name DeepSeekChatStream
extends Node

## 用于向deepseek发送请求并获取流式返回的节点

## deepseek Token，在开放平台获取
@export var ds_token: String = ''
## 系统角色prompt，用于预设人设
@export_multiline var prompt: String = ""
## 深度思考
@export var use_thinking: bool = false
## 温度值，越高输出越随机，默认为1
@export_range(0.0, 2.0, 0.1) var temperature: float = 1.0
## 为正数时降低模型重复相同内容的可能性
@export_range(-2.0, 2.0, 0.1) var frequency_penalty: float = 0
## 为正数时增加模型谈论新主题的可能性
@export_range(-2.0, 2.0, 0.1) var presence_penalty: float = 0
## 最大输出长度，deepseek-chat模型，最大8K，deepseek-reasoner模型，最大64K
@export var max_tokens: int = 4096

## 返回正文
signal message(msg: String)
## 返回正思考内容
signal think(msg: String)
## 返回结束
signal generate_finish

## 发送请求的http客户端
@onready var http_client: HTTPClient = HTTPClient.new()

var generatting: bool = false

## 发送请求
func post_message(msg: String):
	# 准备请求数据
	var headers = [
		"Accept: application/json",
		"Authorization: Bearer %s" % ds_token,
		"Content-Type: application/json"
	]

	var request_body = JSON.stringify({
		"messages": [
			{
				"content": prompt,
				"role": "system"
			},
			{
				"content": msg,
				"role": "user"
			}
		],
		"model": "deepseek-reasoner" if use_thinking else "deepseek-chat",
		"frequency_penalty": frequency_penalty,
		"max_tokens": max_tokens,
		"presence_penalty": presence_penalty,
		"response_format": {
			"type": "text"
		},
		"stream": true,
		"stream_options": null,
		"temperature": temperature,
		"top_p": 1,
		"tools": null,
		"tool_choice": "none",
		"logprobs": false,
		"top_logprobs": null
	})

	var connect_err = http_client.connect_to_host("https://api.deepseek.com")
	generatting = true
	if connect_err != OK:
		push_error("连接服务器失败: " + error_string(connect_err))
		return
	while http_client.get_status() == HTTPClient.STATUS_CONNECTING or http_client.get_status() == HTTPClient.STATUS_RESOLVING:
		http_client.poll()
		#print("Connecting...")
		await get_tree().process_frame
	# 发送POST请求
	var err = http_client.request(HTTPClient.METHOD_POST, "/chat/completions", headers, request_body)
	if err != OK:
		push_error("请求发送失败: " + error_string(err))
		return
	while http_client.get_status() == HTTPClient.STATUS_REQUESTING:
		http_client.poll()
		await get_tree().process_frame

	if http_client.has_response():
		headers = http_client.get_response_headers_as_dictionary()

		while http_client.get_status() == HTTPClient.STATUS_BODY:
			http_client.poll()
			var chunk = http_client.read_response_body_chunk()
			if chunk.size() == 0:
				await get_tree().process_frame
			else:
				var chunk_string = chunk.get_string_from_utf8()
				var data_array = chunk_string.split("\n")
				for data_string in data_array:
					if data_string.begins_with("data: "):
						data_string = data_string.replace("data: ", "")
						if data_string == "[DONE]":
							continue
						var json = JSON.new()
						var parse_err = json.parse(data_string)
						if parse_err != OK:
							push_error("JSON解析错误: " + json.get_error_message())
							push_error(data_string)
							return

						var data = json.get_data()
						if data and data.has("choices"):
							var choices := data["choices"] as Array
							var delta = choices[0]["delta"]
							if use_thinking and delta.has("reasoning_content") and delta.get("reasoning_content") != null:
								think.emit(delta["reasoning_content"])
							else:
								message.emit(delta["content"])
							if choices[0].has("finish_reason") and choices[0].get("finish_reason") == "stop":
								generatting = false
								generate_finish.emit()
						else:
							generatting = false
							print(data)
							push_error("无效的响应结构")
## 中断请求
func close():
	generatting = false
	http_client.close()

代码重点说明

继承自Node类，所以可以在场景树中添加任意个流式输出节点，并且可以分别方便的在检查器中设置token和prompt。
内部使用HTTPClient类，它是Godot中基础的网络请求类。使用它的原因是HTTPRequest类无法处理流式响应，需要更底层的HTTPClient类。
使用节点的post_message方法发送请求问题。
use_thinking参数可以控制是否使用深度思考，即使用的模型为deepseek-reasoner还是deepseek-chat。
如果开启了深度思考，则可以监听think信号获得思考内容。
无论是否开启深度思考，正文内容都会通过message信号返回。
think信号和message信号返回的内容都是片段，如果使用label等，可以按以下示例使用。
链接处理信号应在发送信息之前。

使用示例

代码

gdscript

extends Control
# deepseek流式输出节点
@onready var deep_seek_chat_stream: DeepSeekChatStream = %DeepSeekChatStream
# label输出展示节点
@onready var output_content: Label = %OutputContent

func _ready():
    deep_seek_chat_stream.ds_token = "some_token_you_get_from_deepseek"
    deep_seek_chat_stream.prompt = "你是一个AI智能助手。"
    # 链接message事件，think也可以按此处理
    if not deep_seek_chat_stream.message.is_connected(_on_message):
        deep_seek_chat_stream.message.connect(_on_message)

    # 链接generate_finish事件
    if not deep_seek_chat_stream.generate_finish.is_connected(_on_generate_finish):
        deep_seek_chat_stream.generate_finish.connect(_on_generate_finish)

# 发送请求接口
func _handle_question(question: String):
    # 清空原内容
    output_content.text = ""
    # 编写问题
    var message = "世界上最高的山是哪座？"
    # 发送请求
    deep_seek_chat_stream.post_message(message)

func _on_message(message: String):
    output_content.text += message

func _on_generate_finish():
    print("输出结果：")
    print(output_content.text)

使用非流式输出

非流式输出的场景比较少，通常是希望在后台处理的任务，并一次性返回。

优点是只需要处理一次结束事件即可。

主要代码

可直接复制并在Godot中新建一个deepseek_chat.gd文件并粘贴保存后使用。

代码

gdscript

@tool
class_name DeepSeekChat
extends Node

## 非流式DeepSeek生成内容节点

## deepseek Token，在开放平台获取
@export var ds_token: String = ''
## 系统角色prompt，用于预设人设
@export_multiline var prompt: String = ""
## 深度思考
@export var use_thinking: bool = false
## 温度值，越高输出越随机，默认为1
@export_range(0.0, 2.0, 0.1) var temperature: float = 1.0
## 为正数时降低模型重复相同内容的可能性
@export_range(-2.0, 2.0, 0.1) var frequency_penalty: float = 0
## 为正数时增加模型谈论新主题的可能性
@export_range(-2.0, 2.0, 0.1) var presence_penalty: float = 0
## 最大输出长度，deepseek-chat模型，最大8K，deepseek-reasoner模型，最大64K
@export var max_tokens: int = 4096
## 输出内容的类型，如果为JSON，则必须在prompt中带有JSON字样，并带有示例
@export_enum("text", "json_object") var response_format: String = "text"

## 生成结束信号
signal generate_finish(msg: String, think_msg: String)

## 发送请求的HTTPRequest节点
var http_request: HTTPRequest = null
## 是否在生成中，可以使用此字段防止重复请求
var generatting: bool = false

func _ready() -> void:
	var node = HTTPRequest.new()
	add_child(node)
	http_request = node

## 发送请求
func post_message(msg: String):
	# 准备请求数据
	var headers = [
		"Accept: application/json",
		"Authorization: Bearer %s" % ds_token,
		"Content-Type: application/json"
	]

	var request_body = JSON.stringify({
		"messages": [
			{
				"content": prompt,
				"role": "system"
			},
			{
				"content": msg,
				"role": "user"
			}
		],
		"model": "deepseek-reasoner" if use_thinking else "deepseek-chat",
		"frequency_penalty": frequency_penalty,
		"max_tokens": max_tokens,
		"presence_penalty": presence_penalty,
		"response_format": {
			"type": response_format
		},
		"stream": false,
		"stream_options": null,
		"temperature": temperature,
		"top_p": 1,
		"tools": null,
		"tool_choice": "none",
		"logprobs": false,
		"top_logprobs": null
	})
	if not http_request.request_completed.is_connected(_http_request_completed):
		http_request.request_completed.connect(_http_request_completed)

	# 发送POST请求
	var err = http_request.request( "https://api.deepseek.com/chat/completions", headers, HTTPClient.METHOD_POST, request_body)
	generatting = true
	if err != OK:
		push_error("请求发送失败: " + str(err))
		return

func _http_request_completed(_result, _response_code, _headers, body: PackedByteArray):
	generatting = false
	var json = JSON.new()
	var err = json.parse(body.get_string_from_utf8())
	if err != OK:
		push_error("JSON解析错误: " + json.get_error_message())
		push_error(body.get_string_from_utf8())
		return

	var data = json.get_data()
	if data and data.has("choices"):
		var choices := data["choices"] as Array
		var think_msg = choices[0]["message"].get("reasoning_content", "")
		generate_finish.emit(choices[0]["message"]["content"], think_msg)
	else:
		print(data)
		push_error("无效的响应结构")

## 结束请求
func close():
	if http_request:
		http_request.cancel_request()
		generatting = false

代码重点说明

继承自Node类，所以可以在场景树中添加任意个输出节点，并且可以分别方便的在检查器中设置token和prompt。
内部使用了HTTPRequest类，相较流式输出的写法逻辑和流程简单了一些。
使用节点的post_message方法发送请求问题。
use_thinking参数可以控制是否使用深度思考，即使用的模型为deepseek-reasoner还是deepseek-chat。
监听信号generate_finish结束后返回的数据。第一个参数为返回的正文。第二个参数为思考的内容，如果未开启深度思考，则会返回空字符串。
链接处理信号应在发送信息之前。
response_format详见下方JSON输出小节

使用示例

代码

gdscript

extends Control
# deepseek非流式输出节点
@onready var deep_seek_chat: DeepSeekChat = %DeepSeekChat
# label输出展示节点
@onready var output_content: Label = %OutputContent

func _ready():
    deep_seek_chat.ds_token = "some_token_you_get_from_deepseek"
    deep_seek_chat.prompt = "你是一个AI智能助手。"

    # 链接generate_finish事件
    if not deep_seek_chat.generate_finish.is_connected(_on_generate_finish):
        deep_seek_chat.generate_finish.connect(_on_generate_finish)

# 发送请求接口
func _handle_question(question: String):
    # 清空原内容
    output_content.text = ""
    # 编写问题
    var message = "世界上最高的山是哪座？"
    # 发送请求
    deep_seek_chat.post_message(message)


func _on_generate_finish(msg: String, think_msg: String):
    print("输出结果：")
    print(output_content.text)

JSON输出

相较于流式节点，非流式添加了一个response_format参数，当此参数选择为json_object时，模型会尽量返回一个json字符串。

但是prompt中必须包含json字样以及每个参数的说明以及一个示例输出。

以下为官网示例prompt

官网示例prompt

The user will provide some exam text. 
Please parse the "question" and "answer" and output them in JSON format. 

EXAMPLE INPUT: 
Which is the highest mountain in the world? Mount Everest.

EXAMPLE JSON OUTPUT:
{
    "question": "Which is the highest mountain in the world?",
    "answer": "Mount Everest"
}

本节点在生成结束会将json以字符串的方式作为msg参数返回。用户需要自行解析msg。

另外，官方明确说明，该功能在使用时，有小概率不会返回任何数据。所以用户需要特殊处理msg为空字符串的情况。此非bug。

再另外，官方建议应保证max_token值应足够大，防止返回的json字符串超过最大大小被截取，导致解析错误。

Godot游戏接入Deepseek ​

功能说明 ​

前置说明 ​

使用流式输出 ​

主要代码 ​

代码重点说明 ​

使用示例 ​

使用非流式输出 ​

主要代码 ​

代码重点说明 ​

使用示例 ​

JSON输出 ​

Godot游戏接入Deepseek

功能说明

前置说明

使用流式输出

主要代码

代码重点说明

使用示例

使用非流式输出

主要代码

代码重点说明

使用示例

JSON输出