官方api调用有点慢，只实现了一些基础功能。
import pygame
import random
import requests
import time
from openai import OpenAI# 初始化参数
WIDTH, HEIGHT = 400, 400
GRID_SIZE = 20
client = OpenAI(api_key="sk-2af2*********", base_url="https://api.deepseek.com")# 在文件顶部添加导入
import openai
import threading
from queue import Queue# 在全局参数部分修改
CONVERSATION_HISTORY = [] # 重命名为对话历史
MAX_HISTORY = 10 # 增加历史记录容量def get_direction_from_api(snake, food, direction, conversation_history):"""修改提示生成部分"""head = snake[0]# 新增方向元组定义directions = [(0, -1), (0, 1), (-1, 0), (1, 0)] # 上、下、左、右# 构建障碍物检测提示obstacles = []for distance in range(1, 6):check_pos = (head[0] + direction[0] * distance * GRID_SIZE,head[1] + direction[1] * distance * GRID_SIZE)if (check_pos in snake[1:] or not (0 <= check_pos[0] < WIDTH and 0 <= check_pos[1] < HEIGHT)):obstacles.append(str(distance))# 构建食物检测提示food_dir = ""if food[0] > head[0]: food_dir = "right"elif food[0] < head[0]: food_dir = "left"elif food[1] > head[1]: food_dir = "down"elif food[1] < head[1]: food_dir = "up"# 修复处：处理初始静止状态current_dir = "未移动" if direction == (0,0) else ['up','down','left','right'][directions.index(direction)]prompt = f"当前方向：{current_dir}。"prompt += f"前方障碍在距离：{','.join(obstacles)}格。" if obstacles else "前方无近距离障碍。"prompt += f"食物位于你的{food_dir}方，距离X轴{abs(food[0]-head[0])//GRID_SIZE}格，Y轴{abs(food[1]-head[1])//GRID_SIZE}格。"# 添加碰撞历史到提示if conversation_history:prompt += f"\n最近{MAX_HISTORY}条对话记录：\n" + "\n".join([f"{i+1}. {msg}" for i, msg in enumerate(conversation_history[-MAX_HISTORY:])])try:response = client.chat.completions.create(model="deepseek-reasoner",messages=[{"role": "user","content": prompt + "\n请用单个方向词（up/down/left/right）回答"}],timeout=300)print(f"[API请求] {prompt}")print(f"[API响应] {response}")choice = response.choices[0].message.content.strip().lower()return {"up": (0, -1), "down": (0, 1),"left": (-1, 0), "right": (1, 0)}.get(choice, direction)except openai.APIError as e:print(f"OpenAI API错误: {e}")return directiondef reset_game():"""重置游戏状态"""return {'snake': [(WIDTH//2, HEIGHT//2)],'food': (random.randrange(0, WIDTH, GRID_SIZE), random.randrange(0, HEIGHT, GRID_SIZE)),'direction': (0, 0),'last_api_call': time.time(),'growth_pending': 0}# 在import区域新增
import threading
from queue import Queue# 在全局参数新增
direction_queue = Queue()
api_requesting = False # API请求状态标志def get_direction_async(snake, food, direction, conversation_history):"""异步获取方向"""global api_requestingtry:new_dir = get_direction_from_api(snake, food, direction, conversation_history)direction_queue.put(new_dir)finally:api_requesting = Falsedef game_loop():pygame.init()screen = pygame.display.set_mode((WIDTH, HEIGHT))clock = pygame.time.Clock()game_state = reset_game()collision_history = []global api_requesting # 新增全局变量声明while True:# 事件处理for event in pygame.event.get():if event.type == pygame.QUIT:pygame.quit()return# 修改API调用部分为异步if not api_requesting and (time.time() - game_state['last_api_call'] > 0.5):api_requesting = Truethreading.Thread(target=get_direction_async, args=(game_state['snake'], game_state['food'],game_state['direction'],CONVERSATION_HISTORY[-MAX_HISTORY:] if CONVERSATION_HISTORY else [] # 修复参数)).start()game_state['last_api_call'] = time.time()# 处理方向队列中的新方向# 修改后的移动逻辑if not direction_queue.empty():game_state['direction'] = direction_queue.get()# 执行单步移动new_head = (game_state['snake'][0][0] + game_state['direction'][0] * GRID_SIZE,game_state['snake'][0][1] + game_state['direction'][1] * GRID_SIZE)# 碰撞检测if (new_head in game_state['snake'] or not (0 <= new_head[0] < WIDTH and 0 <= new_head[1] < HEIGHT)):print("撞到障碍物！停止移动")# 停止移动并保持当前状态game_state['direction'] = (0, 0)game_state['last_api_call'] = time.time()continue # 跳过后续移动逻辑# 移动蛇并处理食物game_state['snake'].insert(0, new_head)if new_head == game_state['food']:game_state['food'] = (random.randrange(0, WIDTH, GRID_SIZE),random.randrange(0, HEIGHT, GRID_SIZE))while game_state['food'] in game_state['snake']:game_state['food'] = (random.randrange(0, WIDTH, GRID_SIZE),random.randrange(0, HEIGHT, GRID_SIZE))game_state['growth_pending'] += 1else:if game_state['growth_pending'] > 0:game_state['growth_pending'] -= 1else:game_state['snake'].pop()# 移动后立即停止并重置方向game_state['direction'] = (0, 0)game_state['last_api_call'] = time.time()# 保持API调用持续进行（原0.5秒间隔限制移除）if not api_requesting:api_requesting = Truethreading.Thread(target=get_direction_async, args=(game_state['snake'], game_state['food'],game_state['direction'],CONVERSATION_HISTORY[-MAX_HISTORY:] if CONVERSATION_HISTORY else [] # 修复参数)).start()# 绘制界面screen.fill((0,0,0))for segment in game_state['snake']: # 修复snake引用pygame.draw.rect(screen, (0,255,0), (segment[0], segment[1], GRID_SIZE-1, GRID_SIZE-1))pygame.draw.rect(screen, (255,0,0),(game_state['food'][0], game_state['food'][1], GRID_SIZE-1, GRID_SIZE-1)) # 修复food引用pygame.display.flip()clock.tick(10)if __name__ == "__main__":game_loop()
