Ubuntu
767 lines
28 KiB
Python
767 lines
28 KiB
Python
"""
|
|
斗地主 AI 适配器实现
|
|
支持规则AI和DeepSeek大模型AI
|
|
"""
|
|
import os
|
|
import json
|
|
import asyncio
|
|
import aiohttp
|
|
from typing import Dict, List, Any, Optional
|
|
from collections import Counter
|
|
|
|
import sys
|
|
import os as os_module
|
|
sys.path.insert(0, os_module.path.dirname(os_module.path.dirname(os_module.path.abspath(__file__))))
|
|
|
|
from engine.ai_adapter import AIAdapter, RuleBasedAdapter, LLMAdapter
|
|
from doudizhu.game_logic import (
|
|
Card, identify_card_type, can_beat, evaluate_hand_strength,
|
|
find_cards_by_ids, Deck, CardType
|
|
)
|
|
|
|
# 加载斗地主模型配置文件（支持 DeepSeek 等外部模型）
|
|
def _load_doudizhu_config():
|
|
config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'model_config.json')
|
|
if os.path.exists(config_path):
|
|
with open(config_path, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
return {}
|
|
|
|
DOUDIZHU_CONFIG = _load_doudizhu_config()
|
|
|
|
|
|
class DoudizhuRuleAI(RuleBasedAdapter):
|
|
"""
|
|
斗地主基于规则的AI
|
|
从现有JS代码迁移
|
|
"""
|
|
|
|
def __init__(self, config: Optional[Dict] = None):
|
|
super().__init__(config or {})
|
|
self.ai_type = 'doudizhu_rule'
|
|
|
|
def get_name(self) -> str:
|
|
return "斗地主规则AI"
|
|
|
|
def want_to_be_lord(self, cards: List[Card]) -> bool:
|
|
"""决定是否叫地主"""
|
|
score = evaluate_hand_strength(cards)
|
|
return score >= 8 # 分数>=8才当地主
|
|
|
|
def find_all_valid_plays(self, cards: List[Card], desktop_type: Optional[Dict]) -> List[Dict]:
|
|
"""找出所有可出的牌型"""
|
|
valid_plays = []
|
|
n = len(cards)
|
|
|
|
# 辅助函数：生成所有组合
|
|
def get_combinations(arr, k):
|
|
if k == 0:
|
|
return [[]]
|
|
if len(arr) < k:
|
|
return []
|
|
result = []
|
|
for i in range(len(arr) - k + 1):
|
|
sub_combos = get_combinations(arr[i + 1:], k - 1)
|
|
for sub in sub_combos:
|
|
result.append([arr[i]] + sub)
|
|
return result
|
|
|
|
# 尝试不同数量的牌（限制在合理范围内）
|
|
for k in range(1, min(n + 1, 9)): # 最多8张
|
|
combos = get_combinations(cards, k)
|
|
for combo in combos:
|
|
card_type = identify_card_type(combo)
|
|
if card_type and can_beat(desktop_type, card_type):
|
|
valid_plays.append({
|
|
'cards': combo,
|
|
'type': card_type,
|
|
'card_ids': [c.id for c in combo]
|
|
})
|
|
|
|
# 炸弹单独处理（可以管任何非炸弹）
|
|
value_count = {}
|
|
for c in cards:
|
|
value_count[c.value] = value_count.get(c.value, 0) + 1
|
|
|
|
for value, count in value_count.items():
|
|
if count == 4:
|
|
bomb_cards = [c for c in cards if c.value == value]
|
|
bomb_type = identify_card_type(bomb_cards)
|
|
if bomb_type and can_beat(desktop_type, bomb_type):
|
|
# 去重
|
|
bomb_ids = set(c.id for c in bomb_cards)
|
|
is_duplicate = any(
|
|
set(c.id for c in play['cards']) == bomb_ids
|
|
for play in valid_plays
|
|
)
|
|
if not is_duplicate:
|
|
valid_plays.append({
|
|
'cards': bomb_cards,
|
|
'type': bomb_type,
|
|
'card_ids': [c.id for c in bomb_cards]
|
|
})
|
|
|
|
# 王炸
|
|
jokers = [c for c in cards if c.is_joker]
|
|
if len(jokers) == 2:
|
|
joker_type = identify_card_type(jokers)
|
|
if joker_type and can_beat(desktop_type, joker_type):
|
|
valid_plays.append({
|
|
'cards': jokers,
|
|
'type': joker_type,
|
|
'card_ids': [c.id for c in jokers]
|
|
})
|
|
|
|
return valid_plays
|
|
|
|
# ───────────────────────── 辅助函数 ─────────────────────────
|
|
|
|
def _estimate_min_rounds(self, cards: List[Card]) -> int:
|
|
"""估算最少需要几轮出完（简化版）"""
|
|
if not cards:
|
|
return 0
|
|
n = len(cards)
|
|
value_count = Counter(c.value for c in cards)
|
|
|
|
rounds = 0
|
|
# 王炸
|
|
if sum(1 for c in cards if c.is_joker) == 2:
|
|
rounds += 1
|
|
n -= 2
|
|
|
|
# 炸弹
|
|
for v, c in list(value_count.items()):
|
|
if c == 4:
|
|
rounds += 1
|
|
n -= 4
|
|
value_count[v] = 0
|
|
|
|
# 飞机（连续三张）
|
|
triples = sorted([v for v, c in value_count.items() if c >= 3 and v <= 14])
|
|
if len(triples) >= 2:
|
|
streak = 1
|
|
for i in range(1, len(triples)):
|
|
if triples[i] == triples[i - 1] + 1:
|
|
streak += 1
|
|
else:
|
|
break
|
|
if streak >= 2:
|
|
rounds += 1
|
|
n -= streak * 3
|
|
for v in triples[:streak]:
|
|
value_count[v] -= 3
|
|
|
|
# 连对
|
|
pairs = sorted([v for v, c in value_count.items() if c >= 2 and v <= 14])
|
|
if len(pairs) >= 3:
|
|
streak = 1
|
|
for i in range(1, len(pairs)):
|
|
if pairs[i] == pairs[i - 1] + 1:
|
|
streak += 1
|
|
else:
|
|
break
|
|
if streak >= 3:
|
|
rounds += 1
|
|
n -= streak * 2
|
|
for v in pairs[:streak]:
|
|
value_count[v] -= 2
|
|
|
|
# 顺子
|
|
unique = sorted([v for v, c in value_count.items() if c > 0 and v <= 14])
|
|
if len(unique) >= 5:
|
|
streak = 1
|
|
max_streak = 1
|
|
for i in range(1, len(unique)):
|
|
if unique[i] == unique[i - 1] + 1:
|
|
streak += 1
|
|
max_streak = max(max_streak, streak)
|
|
else:
|
|
streak = 1
|
|
if max_streak >= 5:
|
|
rounds += 1
|
|
n -= max_streak
|
|
|
|
# 剩余牌按最紧凑估算
|
|
if n > 0:
|
|
rounds += max(1, (n + 1) // 2)
|
|
return rounds
|
|
|
|
def _calculate_break_cost(self, cards: List[Card], play_cards: List[Card]) -> float:
|
|
"""计算拆牌代价：出了这些牌后，是否拆了原本的组合"""
|
|
play_values = {c.value for c in play_cards}
|
|
all_values = [c.value for c in cards]
|
|
cost = 0.0
|
|
|
|
# 检查是否拆了顺子
|
|
sorted_unique = sorted(set(v for v in all_values if v <= 14))
|
|
if len(sorted_unique) >= 5:
|
|
# 找最长连续序列
|
|
streaks = []
|
|
current = [sorted_unique[0]]
|
|
for v in sorted_unique[1:]:
|
|
if v == current[-1] + 1:
|
|
current.append(v)
|
|
else:
|
|
if len(current) >= 5:
|
|
streaks.append(list(current))
|
|
current = [v]
|
|
if len(current) >= 5:
|
|
streaks.append(current)
|
|
|
|
for streak in streaks:
|
|
removed = play_values & set(streak)
|
|
if removed and len(removed) < len(streak):
|
|
cost += len(removed) * 2.0
|
|
|
|
# 检查是否拆了连对
|
|
value_count = Counter(all_values)
|
|
pair_values = sorted([v for v, c in value_count.items() if c >= 2 and v <= 14])
|
|
if len(pair_values) >= 3:
|
|
current = [pair_values[0]]
|
|
for v in pair_values[1:]:
|
|
if v == current[-1] + 1:
|
|
current.append(v)
|
|
else:
|
|
if len(current) >= 3:
|
|
removed = play_values & set(current)
|
|
if removed and len(removed) < len(current):
|
|
cost += len(removed) * 3.0
|
|
current = [v]
|
|
if len(current) >= 3:
|
|
removed = play_values & set(current)
|
|
if removed and len(removed) < len(current):
|
|
cost += len(removed) * 3.0
|
|
|
|
# 检查是否拆了飞机（连续三张）
|
|
triple_values = sorted([v for v, c in value_count.items() if c >= 3 and v <= 14])
|
|
if len(triple_values) >= 2:
|
|
current = [triple_values[0]]
|
|
for v in triple_values[1:]:
|
|
if v == current[-1] + 1:
|
|
current.append(v)
|
|
else:
|
|
if len(current) >= 2:
|
|
removed = play_values & set(current)
|
|
if removed and len(removed) < len(current):
|
|
cost += len(removed) * 4.0
|
|
current = [v]
|
|
if len(current) >= 2:
|
|
removed = play_values & set(current)
|
|
if removed and len(removed) < len(current):
|
|
cost += len(removed) * 4.0
|
|
|
|
return cost
|
|
|
|
def _evaluate_remaining_shape(self, cards: List[Card]) -> float:
|
|
"""评估剩余手牌的牌型质量（越高越好）"""
|
|
if not cards:
|
|
return 100.0
|
|
quality = 0.0
|
|
value_count = Counter(c.value for c in cards)
|
|
|
|
# 炸弹/王炸
|
|
for c in value_count.values():
|
|
if c == 4:
|
|
quality += 10.0
|
|
jokers = [c for c in cards if c.is_joker]
|
|
if len(jokers) == 2:
|
|
quality += 20.0
|
|
|
|
# 2 和王
|
|
for c in cards:
|
|
if c.value == 15:
|
|
quality += 3.0
|
|
elif c.value >= 16:
|
|
quality += 5.0
|
|
|
|
# 顺子潜力
|
|
unique = sorted(set(v for v in value_count.keys() if v <= 14))
|
|
streaks = 0
|
|
cur = 0
|
|
for i, v in enumerate(unique):
|
|
if i == 0 or v == unique[i - 1] + 1:
|
|
cur += 1
|
|
else:
|
|
if cur >= 3:
|
|
streaks += cur
|
|
cur = 1
|
|
if cur >= 3:
|
|
streaks += cur
|
|
quality += streaks * 0.8
|
|
|
|
# 对子/三张
|
|
pairs = sum(1 for c in value_count.values() if c >= 2)
|
|
triples = sum(1 for c in value_count.values() if c >= 3)
|
|
quality += pairs * 1.5 + triples * 3.0
|
|
|
|
return quality
|
|
|
|
def _evaluate_play(self, play: Dict, cards: List[Card],
|
|
is_lord: bool, teammate_cards: int,
|
|
desktop_type: Optional[Dict]) -> float:
|
|
"""评估一个出牌方案的得分，越高越好"""
|
|
play_card_ids = {c.id for c in play['cards']}
|
|
remaining = [c for c in cards if c.id not in play_card_ids]
|
|
remaining_count = len(remaining)
|
|
play_count = len(play['cards'])
|
|
play_type = play['type']['type']
|
|
play_value = play['type']['value']
|
|
|
|
score = 0.0
|
|
|
|
# 1. 能出完 → 最高优先级
|
|
if remaining_count == 0:
|
|
return 20000.0
|
|
|
|
# 2. 出得多比出得少好（减少轮数）
|
|
score += play_count * 5.0
|
|
|
|
# 3. 剩余轮数估算：轮数越少越好
|
|
my_rounds = self._estimate_min_rounds(remaining)
|
|
score -= my_rounds * 30.0
|
|
|
|
# 4. 牌值越小越好（保留大牌）
|
|
score -= play_value * 0.8
|
|
|
|
# 5. 拆牌代价
|
|
break_cost = self._calculate_break_cost(cards, play['cards'])
|
|
score -= break_cost * 8.0
|
|
|
|
# 6. 剩余牌质量
|
|
remaining_quality = self._evaluate_remaining_shape(remaining)
|
|
score += remaining_quality * 1.5
|
|
|
|
# 7. 炸弹特殊处理
|
|
if play_type in ('bomb', 'joker_bomb'):
|
|
if remaining_count <= 2:
|
|
score += 800 # 快出完了，果断炸
|
|
elif teammate_cards <= 2 and not is_lord:
|
|
score += 800 # 队友快走完了，炸地主
|
|
elif not is_lord and not desktop_type:
|
|
score -= 300 # 农民且没人出牌，不要主动炸
|
|
else:
|
|
score -= 100 # 一般情况，不优先炸
|
|
|
|
# 8. 角色策略
|
|
if is_lord:
|
|
# 地主：优先出对方难接的长牌型
|
|
if play_type == 'straight' and play_count >= 5:
|
|
score += play_count * 3.0
|
|
elif play_type == 'double_straight':
|
|
score += play_count * 2.5
|
|
elif play_type == 'triple_straight':
|
|
score += play_count * 3.5
|
|
elif play_type in ('triple_with_single', 'triple_with_pair'):
|
|
score += 5.0
|
|
else:
|
|
# 农民：队友牌少时让牌，桌面有牌时要帮管
|
|
if teammate_cards <= 3:
|
|
if not desktop_type:
|
|
score -= 300 # 不要主动抢节奏
|
|
elif desktop_type:
|
|
score += 150 # 队友快走完了，积极管牌
|
|
|
|
# 9. 管牌时：尽量用刚好能管的最小组合，优先同类型
|
|
if desktop_type:
|
|
desktop_type_str = desktop_type['type']
|
|
if play_type == desktop_type_str:
|
|
score += 25 # 同类型优先，不浪费牌型
|
|
elif play_type in ('bomb', 'joker_bomb'):
|
|
# 用炸弹管，前面已处理
|
|
pass
|
|
else:
|
|
score -= 10 # 跨类型管牌（如用对子管单张），轻微惩罚
|
|
|
|
return score
|
|
|
|
def choose_best_play(self, valid_plays: List[Dict], cards: List[Card],
|
|
is_lord: bool, teammate_cards: int,
|
|
desktop_type: Optional[Dict] = None) -> Optional[Dict]:
|
|
"""选择最佳出牌（多维度评分）"""
|
|
if not valid_plays:
|
|
return None
|
|
|
|
# 炸弹/王炸单独收集，最后决策
|
|
bombs = [p for p in valid_plays if p['type']['type'] in ('bomb', 'joker_bomb')]
|
|
non_bombs = [p for p in valid_plays if p['type']['type'] not in ('bomb', 'joker_bomb')]
|
|
|
|
best = None
|
|
best_score = -float('inf')
|
|
|
|
# 先评估非炸弹
|
|
for play in non_bombs:
|
|
score = self._evaluate_play(play, cards, is_lord, teammate_cards, desktop_type)
|
|
if score > best_score:
|
|
best_score = score
|
|
best = play
|
|
|
|
# 炸弹决策：只在必要时才考虑
|
|
if bombs:
|
|
# 判断是否该出炸弹
|
|
should_bomb = False
|
|
remaining_count = len(cards)
|
|
|
|
if remaining_count <= 2:
|
|
should_bomb = True
|
|
elif teammate_cards <= 2 and not is_lord:
|
|
should_bomb = True
|
|
elif desktop_type:
|
|
# 桌面牌很大，没有非炸弹能管，必须炸
|
|
if not best or best_score < 0:
|
|
should_bomb = True
|
|
|
|
if should_bomb:
|
|
# 选最小的炸弹
|
|
bombs.sort(key=lambda p: p['type']['value'])
|
|
bomb = bombs[0]
|
|
bomb_score = self._evaluate_play(bomb, cards, is_lord, teammate_cards, desktop_type)
|
|
if bomb_score > best_score:
|
|
best = bomb
|
|
|
|
return best
|
|
|
|
async def make_decision_calling(self, cards: List[Card]) -> Dict[str, Any]:
|
|
"""叫地主决策"""
|
|
want = self.want_to_be_lord(cards)
|
|
return {
|
|
'action': 'call' if want else 'pass',
|
|
'data': {'want': want},
|
|
'reasoning': f'手牌强度评分: {evaluate_hand_strength(cards)}',
|
|
'confidence': 0.8 if want else 0.9
|
|
}
|
|
|
|
async def make_decision_playing(self, cards: List[Card],
|
|
desktop_type: Optional[Dict],
|
|
is_lord: bool,
|
|
teammate_cards: int) -> Dict[str, Any]:
|
|
"""出牌决策（优化版）"""
|
|
valid_plays = self.find_all_valid_plays(cards, desktop_type)
|
|
|
|
if not valid_plays:
|
|
return {
|
|
'action': 'pass',
|
|
'data': {},
|
|
'reasoning': '没有能管上的牌',
|
|
'confidence': 1.0
|
|
}
|
|
|
|
# 农民让牌逻辑：队友快走完了且自己不能一轮出完，不要主动出牌
|
|
if not is_lord and not desktop_type and teammate_cards <= 3:
|
|
my_rounds = self._estimate_min_rounds(cards)
|
|
if my_rounds > 1:
|
|
return {
|
|
'action': 'pass',
|
|
'data': {},
|
|
'reasoning': f'队友剩{teammate_cards}张牌，让队友先走',
|
|
'confidence': 0.9
|
|
}
|
|
|
|
best_play = self.choose_best_play(valid_plays, cards, is_lord, teammate_cards, desktop_type)
|
|
|
|
if best_play:
|
|
play_name = best_play['type']['name']
|
|
remaining = len(cards) - len(best_play['cards'])
|
|
reasoning = f'出 {play_name}，剩余{remaining}张牌'
|
|
|
|
# 根据决策补充理由
|
|
if best_play['type']['type'] in ('bomb', 'joker_bomb'):
|
|
reasoning = f'炸弹！{reasoning}'
|
|
elif not is_lord and teammate_cards <= 3:
|
|
reasoning = f'帮队友管牌：{reasoning}'
|
|
elif is_lord and best_play['type']['type'] in ('straight', 'double_straight', 'triple_straight'):
|
|
reasoning = f'地主出长牌型逼拆：{reasoning}'
|
|
|
|
return {
|
|
'action': 'play',
|
|
'data': {'card_ids': best_play['card_ids']},
|
|
'reasoning': reasoning,
|
|
'confidence': 0.88
|
|
}
|
|
|
|
return {
|
|
'action': 'pass',
|
|
'data': {},
|
|
'reasoning': '选择不出',
|
|
'confidence': 0.7
|
|
}
|
|
|
|
async def make_decision(self, game_state: Dict[str, Any],
|
|
available_actions: List[str],
|
|
player_context: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""AI 做出决策"""
|
|
start_time = asyncio.get_event_loop().time()
|
|
|
|
# 解析上下文
|
|
phase = game_state.get('phase') or game_state.get('data', {}).get('phase')
|
|
cards_data = player_context.get('cards', [])
|
|
cards = [Card(**c) for c in cards_data]
|
|
|
|
is_lord = player_context.get('is_lord', False)
|
|
teammate_cards = player_context.get('teammate_cards', 17)
|
|
|
|
# desktop_cards 可能在 game_state['doudizhu'] 或 game_state['data'] 中
|
|
desktop_type = None
|
|
if 'doudizhu' in game_state:
|
|
desktop_type = game_state['doudizhu'].get('desktop_cards')
|
|
elif 'data' in game_state:
|
|
desktop_type = game_state['data'].get('desktop_cards')
|
|
|
|
# 决策
|
|
if 'call' in available_actions or ('calling' in str(phase) and 'pass' in available_actions):
|
|
result = await self.make_decision_calling(cards)
|
|
else:
|
|
result = await self.make_decision_playing(cards, desktop_type, is_lord, teammate_cards)
|
|
|
|
# 记录调用时间
|
|
latency = (asyncio.get_event_loop().time() - start_time) * 1000
|
|
self.record_call(latency)
|
|
|
|
return result
|
|
|
|
|
|
class DoudizhuDeepSeekAI(LLMAdapter):
|
|
"""
|
|
斗地主 DeepSeek 大模型 AI
|
|
调用 DeepSeek API 做决策
|
|
"""
|
|
|
|
def __init__(self, api_key: str = None, config: Optional[Dict] = None):
|
|
file_config = DOUDIZHU_CONFIG.get('deepseek', {})
|
|
config = config or {}
|
|
# 优先级：传入参数 > 配置文件 > 环境变量 > 硬编码默认值
|
|
final_api_key = api_key or file_config.get('api_key') or os.getenv('DEEPSEEK_API_KEY', '')
|
|
final_model = config.get('model') or file_config.get('model') or 'deepseek-chat'
|
|
final_api_url = config.get('api_url') or file_config.get('api_url') or 'https://api.deepseek.com/v1/chat/completions'
|
|
super().__init__(
|
|
model=final_model,
|
|
api_key=final_api_key,
|
|
api_url=final_api_url,
|
|
config=config
|
|
)
|
|
self.ai_type = 'doudizhu_deepseek'
|
|
|
|
def get_name(self) -> str:
|
|
return f"斗地主DeepSeek({self.model})"
|
|
|
|
def _build_prompt_calling(self, cards: List[Card]) -> str:
|
|
"""构建叫地主提示词"""
|
|
cards_str = ', '.join([f"{c.suit}{c.rank}" for c in cards])
|
|
score = evaluate_hand_strength(cards)
|
|
|
|
return f"""你是一个斗地主高手。请根据你的手牌决定是否叫地主。
|
|
|
|
你的手牌（共17张）:
|
|
{cards_str}
|
|
|
|
手牌强度评分: {score}/30 (分数越高手牌越好)
|
|
评分标准：
|
|
- K=1分, A=2分, 2=3分, 王=4分
|
|
- 炸弹额外+6分, 王炸额外+10分
|
|
|
|
规则：
|
|
- 如果叫地主，你将获得3张底牌，总共20张牌
|
|
- 地主需要对抗两个农民
|
|
- 地主先出牌
|
|
|
|
请决定是否叫地主，回答格式：
|
|
{{"call": true/false, "reason": "你的理由"}}
|
|
|
|
只输出JSON，不要其他内容。"""
|
|
|
|
def _build_prompt_playing(self, cards: List[Card], desktop_type: Optional[Dict],
|
|
is_lord: bool, game_history: List[Dict]) -> str:
|
|
"""构建出牌提示词"""
|
|
cards_str = ', '.join([f"{c.suit}{c.rank}({c.id})" for c in cards])
|
|
|
|
desktop_str = "无（你可以出任意牌）"
|
|
if desktop_type:
|
|
# desktop_type 中不包含 cards 字段（为了JSON序列化）
|
|
desktop_name = desktop_type.get('name', '未知')
|
|
desktop_value = desktop_type.get('value', 0)
|
|
desktop_str = f"{desktop_name} (牌值: {desktop_value})"
|
|
|
|
history_str = ""
|
|
if game_history:
|
|
for h in game_history[-5:]: # 最近5步
|
|
action = h.get('action', '')
|
|
player = h.get('player_id', '')[:8]
|
|
history_str += f"- {player}: {action}\n"
|
|
|
|
role = "地主" if is_lord else "农民"
|
|
|
|
return f"""你是一个斗地主高手。请根据当前局面选择最佳出牌策略。
|
|
|
|
你的角色: {role}
|
|
你的手牌（共{len(cards)}张）:
|
|
{cards_str}
|
|
|
|
桌面上的牌: {desktop_str}
|
|
|
|
最近的操作记录:
|
|
{history_str if history_str else '无'}
|
|
|
|
牌型说明：
|
|
- 单张：任意一张牌
|
|
- 对子：两张相同牌值的牌
|
|
- 三张：三张相同牌值的牌
|
|
- 三带一：三张 + 一张单牌
|
|
- 三带二：三张 + 一对
|
|
- 顺子：5张或以上连续单牌（如3-4-5-6-7）
|
|
- 炸弹：四张相同牌值的牌
|
|
- 王炸：大王+小王
|
|
|
|
规则：
|
|
- 如果桌面上没有牌或你是最后出牌的，可以出任意合法牌型
|
|
- 否则必须出能管上桌面牌型的牌（同类型且更大，或炸弹/王炸）
|
|
- 如果无法或不想出牌，可以选择"不出"
|
|
|
|
请决定你的行动，回答格式：
|
|
{{"action": "play"/"pass", "card_ids": [牌的ID列表], "reason": "你的理由"}}
|
|
|
|
注意：
|
|
- action为"pass"时，card_ids为空数组
|
|
- 只能从你的手牌中选择牌的ID
|
|
- 只输出JSON，不要其他内容
|
|
|
|
你的决策："""
|
|
|
|
async def _call_deepseek(self, prompt: str) -> Dict[str, Any]:
|
|
"""调用 DeepSeek API"""
|
|
if not self.api_key:
|
|
raise ValueError("DeepSeek API Key 未配置")
|
|
|
|
headers = {
|
|
'Authorization': f'Bearer {self.api_key}',
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
payload = {
|
|
'model': self.model,
|
|
'messages': [
|
|
{'role': 'system', 'content': '你是一个斗地主游戏AI助手。请根据提示做出最佳决策，只输出JSON格式。'},
|
|
{'role': 'user', 'content': prompt}
|
|
],
|
|
'temperature': 0.3,
|
|
'max_tokens': 500
|
|
}
|
|
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(
|
|
self.api_url,
|
|
headers=headers,
|
|
json=payload,
|
|
timeout=aiohttp.ClientTimeout(total=self.timeout)
|
|
) as response:
|
|
if response.status != 200:
|
|
error_text = await response.text()
|
|
raise Exception(f"API Error: {response.status} - {error_text}")
|
|
|
|
data = await response.json()
|
|
content = data['choices'][0]['message']['content']
|
|
|
|
# 解析 JSON
|
|
try:
|
|
# 尝试直接解析
|
|
result = json.loads(content)
|
|
except json.JSONDecodeError:
|
|
# 尝试从文本中提取 JSON
|
|
import re
|
|
json_match = re.search(r'\{[^}]+\}', content)
|
|
if json_match:
|
|
result = json.loads(json_match.group())
|
|
else:
|
|
raise Exception(f"无法解析响应: {content}")
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
print(f"DeepSeek API 调用失败: {e}")
|
|
# 失败时返回默认决策
|
|
return {'action': 'pass', 'reason': f'API调用失败: {str(e)}'}
|
|
|
|
async def make_decision(self, game_state: Dict[str, Any],
|
|
available_actions: List[str],
|
|
player_context: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""AI 做出决策"""
|
|
import time
|
|
start_time = time.time()
|
|
|
|
# 解析上下文
|
|
phase = game_state.get('phase') or game_state.get('data', {}).get('phase')
|
|
cards_data = player_context.get('cards', [])
|
|
cards = [Card(**c) for c in cards_data]
|
|
|
|
is_lord = player_context.get('is_lord', False)
|
|
game_history = game_state.get('history', [])
|
|
|
|
# desktop_cards 可能在 game_state['doudizhu'] 或 game_state['data'] 中
|
|
desktop_type = None
|
|
if 'doudizhu' in game_state:
|
|
desktop_type = game_state['doudizhu'].get('desktop_cards')
|
|
elif 'data' in game_state:
|
|
desktop_type = game_state['data'].get('desktop_cards')
|
|
|
|
try:
|
|
if 'call' in available_actions or ('calling' in str(phase) and 'pass' in available_actions):
|
|
# 叫地主决策
|
|
prompt = self._build_prompt_calling(cards)
|
|
result = await self._call_deepseek(prompt)
|
|
|
|
want = result.get('call', False)
|
|
reason = result.get('reason', '大模型决策')
|
|
|
|
action_result = {
|
|
'action': 'call' if want else 'pass',
|
|
'data': {'want': want},
|
|
'reasoning': reason,
|
|
'confidence': 0.75
|
|
}
|
|
else:
|
|
# 出牌决策
|
|
prompt = self._build_prompt_playing(cards, desktop_type, is_lord, game_history)
|
|
result = await self._call_deepseek(prompt)
|
|
|
|
action = result.get('action', 'pass')
|
|
card_ids = result.get('card_ids', []) if action == 'play' else []
|
|
reason = result.get('reason', '大模型决策')
|
|
|
|
action_result = {
|
|
'action': action,
|
|
'data': {'card_ids': card_ids} if card_ids else {},
|
|
'reasoning': reason,
|
|
'confidence': 0.75
|
|
}
|
|
|
|
except Exception as e:
|
|
# 出错时回退到规则AI
|
|
print(f"DeepSeek AI 出错，回退到规则AI: {e}")
|
|
rule_ai = DoudizhuRuleAI()
|
|
action_result = await rule_ai.make_decision(game_state, available_actions, player_context)
|
|
|
|
# 记录调用时间
|
|
latency = (time.time() - start_time) * 1000
|
|
self.record_call(latency)
|
|
|
|
return action_result
|
|
|
|
|
|
def create_doudizhu_ai(ai_type: str, **kwargs) -> AIAdapter:
|
|
"""
|
|
创建斗地主AI的工厂函数
|
|
|
|
Args:
|
|
ai_type: 'rule' 或 'deepseek'
|
|
**kwargs: 额外配置
|
|
|
|
Returns:
|
|
AIAdapter 实例
|
|
"""
|
|
if ai_type == 'rule':
|
|
return DoudizhuRuleAI(kwargs.get('config'))
|
|
elif ai_type == 'deepseek':
|
|
api_key = kwargs.get('api_key') or os.getenv('DEEPSEEK_API_KEY')
|
|
return DoudizhuDeepSeekAI(api_key=api_key, config=kwargs.get('config'))
|
|
else:
|
|
raise ValueError(f"未知的AI类型: {ai_type}")
|