架构全景：八核智能象棋大脑

第一章：环境搭建 - 象棋AI的"基础设施"

1.1 系统要求

# 最小化配置
CPU: 8核+
内存: 32GB+
存储: 500GB SSD
GPU: NVIDIA RTX 4090+ (可选，推荐)

# 推荐配置
CPU: 16核 AMD EPYC/Intel Xeon
内存: 128GB
GPU: 2× NVIDIA A100 80GB
存储: 2TB NVMe SSD

1.2 一键安装脚本

#!/bin/bash
# chess-ai-infra.sh
set -e

echo "🚀 开始部署象棋AI基础设施..."

# 1. 安装Docker和Docker Compose
curl -fsSL https://get.docker.com -o get-docker.sh
sudo sh get-docker.sh
sudo usermod -aG docker $USER
sudo systemctl enable docker
sudo systemctl start docker

# 2. 安装MindSpore GPU版本
pip install mindspore-gpu==2.2.0

# 3. 下载预训练象棋模型
wget https://models.mindspore.cn/chess/chess_transformer_v2.ckpt
wget https://models.mindspore.cn/chess/chess_analysis_model.ckpt

# 4. 创建项目目录
mkdir -p chess-ai/{data,models,config,scripts,logs}
cd chess-ai

echo "✅ 基础环境准备完成！"

1.3 Docker Compose编排文件

# docker-compose.yml
version: '3.8'

services:
  # 向量数据库 - 存储棋谱特征
  milvus-standalone:
    image: milvusdb/milvus:v2.3.0
    container_name: milvus-chess
    command: ["milvus", "run", "standalone"]
    environment:
      ETCD_ENDPOINTS: etcd:2379
      MINIO_ADDRESS: minio:9000
    ports:
      - "19530:19530"
      - "9091:9091"
    volumes:
      - ./volumes/milvus:/var/lib/milvus
    depends_on:
      - etcd
      - minio
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:9091/healthz"]
      interval: 30s
      timeout: 10s
      retries: 3

  # 分析型数据库 - 棋局统计
  clickhouse:
    image: clickhouse/clickhouse-server:23.8
    container_name: clickhouse-chess
    ports:
      - "8123:8123"  # HTTP
      - "9000:9000"  # TCP
    volumes:
      - ./volumes/clickhouse:/var/lib/clickhouse
      - ./config/clickhouse/users.xml:/etc/clickhouse-server/users.xml
      - ./config/clickhouse/config.xml:/etc/clickhouse-server/config.xml
    ulimits:
      nofile:
        soft: 262144
        hard: 262144
    command: ["--max_memory_usage=16000000000"]

  # 全文搜索引擎 - 象棋知识库
  elasticsearch:
    image: elasticsearch:8.10.2
    container_name: elasticsearch-chess
    environment:
      - discovery.type=single-node
      - "ES_JAVA_OPTS=-Xms8g -Xmx8g"
      - xpack.security.enabled=false
    ports:
      - "9200:9200"
      - "9300:9300"
    volumes:
      - ./volumes/elasticsearch:/usr/share/elasticsearch/data
    ulimits:
      memlock:
        soft: -1
        hard: -1

  # 缓存数据库 - 会话和实时数据
  redis:
    image: redis:7.2-alpine
    container_name: redis-chess
    ports:
      - "6379:6379"
    volumes:
      - ./volumes/redis:/data
      - ./config/redis/redis.conf:/usr/local/etc/redis/redis.conf
    command: redis-server /usr/local/etc/redis/redis.conf

  # 辅助服务
  etcd:
    image: quay.io/coreos/etcd:v3.5.0
    container_name: etcd
    environment:
      - ETCD_AUTO_COMPACTION_MODE=revision
      - ETCD_AUTO_COMPACTION_RETENTION=1000
      - ETCD_QUOTA_BACKEND_BYTES=4294967296
      - ETCD_SNAPSHOT_COUNT=50000
    volumes:
      - ./volumes/etcd:/etcd
    command: [
          "etcd",
          "--advertise-client-urls=http://0.0.0.0:2379",
          "--listen-client-urls=http://0.0.0.0:2379",
          "--data-dir=/etcd"
        ]

  minio:
    image: minio/minio:RELEASE.2023-08-23T10-07-06Z
    container_name: minio
    environment:
      MINIO_ACCESS_KEY: minioadmin
      MINIO_SECRET_KEY: minioadmin
    volumes:
      - ./volumes/minio:/data
    command: minio server /data --console-address ":9001"
    ports:
      - "9000:9000"
      - "9001:9001"

  # 数据导入工具
  data-importer:
    build: ./data-importer
    container_name: chess-data-importer
    depends_on:
      - milvus-standalone
      - clickhouse
      - elasticsearch
      - redis
    volumes:
      - ./data:/data
    environment:
      - MILVUS_HOST=milvus-standalone
      - CLICKHOUSE_HOST=clickhouse
      - ELASTICSEARCH_HOST=elasticsearch
      - REDIS_HOST=redis
    command: ["python", "import_all_data.py"]

  # Web界面
  chess-ui:
    build: ./web-ui
    ports:
      - "8080:80"
    depends_on:
      - api-gateway

  # API网关
  api-gateway:
    build: ./api-gateway
    ports:
      - "8000:8000"
    environment:
      - MODEL_SERVICE_HOST=mindspore-service
      - REDIS_HOST=redis
    depends_on:
      - mindspore-service
      - redis

  # MindSpore模型服务
  mindspore-service:
    build: ./mindspore-service
    ports:
      - "5000:5000"
      - "5001:5001"  # 监控端口
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
    environment:
      - MILVUS_HOST=milvus-standalone
      - CLICKHOUSE_HOST=clickhouse
      - ELASTICSEARCH_HOST=elasticsearch
      - REDIS_HOST=redis
      - CUDA_VISIBLE_DEVICES=0
    volumes:
      - ./models:/app/models
      - ./logs:/app/logs

networks:
  default:
    name: chess-ai-network
    driver: bridge

volumes:
  milvus:
  clickhouse:
  elasticsearch:
  redis:
  etcd:
  minio:

第二章：数据准备 - 构建象棋知识宇宙

2.1 数据采集脚本

# data_collector.py
import asyncio
import aiohttp
import chess
import chess.pgn
import json
from datetime import datetime
from typing import List, Dict, Any
import numpy as np
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor

@dataclass
class ChessGame:
    """象棋对局数据类"""
    game_id: str
    white_player: str
    black_player: str
    result: str
    moves: List[str]
    fen_positions: List[str]
    opening: str
    date: str
    rating_white: int
    rating_black: int
    time_control: str
    analysis: Dict[str, Any] = None
    embeddings: np.ndarray = None

class ChessDataCollector:
    """象棋数据收集器"""
    
    def __init__(self):
        self.sources = {
            "lichess": "https://lichess.org/api",
            "chess_com": "https://api.chess.com",
            "local_db": "./data/chess_games"
        }
        
    async def fetch_games_by_opening(self, opening: str, limit: int = 1000):
        """按开局获取棋局"""
        async with aiohttp.ClientSession() as session:
            # 从Lichess获取数据
            url = f"{self.sources['lichess']}/games/master"
            params = {
                "opening": opening,
                "max": limit,
                "moves": "true",
                "pgnInJson": "true"
            }
            
            async with session.get(url, params=params) as response:
                if response.status == 200:
                    data = await response.json()
                    return self._parse_games(data)
                else:
                    print(f"Failed to fetch games: {response.status}")
                    return []
    
    def _parse_games(self, raw_data: List[Dict]) -> List[ChessGame]:
        """解析原始棋局数据"""
        games = []
        
        for idx, game_data in enumerate(raw_data):
            try:
                game = ChessGame(
                    game_id=f"game_{datetime.now().strftime('%Y%m%d')}_{idx}",
                    white_player=game_data.get("white", {}).get("name", "Unknown"),
                    black_player=game_data.get("black", {}).get("name", "Unknown"),
                    result=game_data.get("result", "*"),
                    moves=game_data.get("moves", "").split(),
                    fen_positions=self._generate_fen_sequence(game_data.get("moves", "")),
                    opening=game_data.get("opening", {}).get("name", "Unknown"),
                    date=game_data.get("createdAt", ""),
                    rating_white=game_data.get("white", {}).get("rating", 1500),
                    rating_black=game_data.get("black", {}).get("rating", 1500),
                    time_control=game_data.get("clock", {}).get("initial", "300+0")
                )
                
                # 生成棋局分析
                game.analysis = self._analyze_game(game)
                
                games.append(game)
                
            except Exception as e:
                print(f"Error parsing game {idx}: {e}")
                continue
        
        return games
    
    def _generate_fen_sequence(self, moves: str) -> List[str]:
        """生成棋局FEN序列"""
        board = chess.Board()
        fen_sequence = [board.fen()]
        
        for move in moves.split():
            try:
                move_obj = board.parse_san(move) if move in board.legal_moves else None
                if move_obj:
                    board.push(move_obj)
                    fen_sequence.append(board.fen())
            except:
                continue
        
        return fen_sequence
    
    def _analyze_game(self, game: ChessGame) -> Dict[str, Any]:
        """分析棋局关键点"""
        board = chess.Board()
        
        analysis = {
            "critical_moves": [],
            "material_balance": [],
            "center_control": [],
            "king_safety": [],
            "pawn_structure": []
        }
        
        # 模拟分析每一步
        for i, move in enumerate(game.moves[:30]):  # 只分析前30步
            try:
                move_obj = board.parse_san(move)
                board.push(move_obj)
                
                # 评估关键指标
                analysis["material_balance"].append(self._calculate_material(board))
                analysis["center_control"].append(self._calculate_center_control(board))
                
                # 检测关键步
                if i > 0 and i < len(game.moves) - 1:
                    score_change = self._evaluate_move_criticality(board, move_obj)
                    if score_change > 0.5:
                        analysis["critical_moves"].append({
                            "move_number": i + 1,
                            "move": move,
                            "score_change": score_change,
                            "fen": board.fen()
                        })
                        
            except Exception as e:
                continue
        
        return analysis
    
    def _calculate_material(self, board: chess.Board) -> float:
        """计算子力价值"""
        piece_values = {
            chess.PAWN: 1,
            chess.KNIGHT: 3,
            chess.BISHOP: 3.2,
            chess.ROOK: 5,
            chess.QUEEN: 9,
            chess.KING: 0
        }
        
        white_material = 0
        black_material = 0
        
        for square in chess.SQUARES:
            piece = board.piece_at(square)
            if piece:
                value = piece_values.get(piece.piece_type, 0)
                if piece.color == chess.WHITE:
                    white_material += value
                else:
                    black_material += value
        
        return white_material - black_material  # 正值表示白优
    
    def _calculate_center_control(self, board: chess.Board) -> float:
        """计算中心控制力"""
        center_squares = [chess.D4, chess.E4, chess.D5, chess.E5]
        control_score = 0
        
        for square in center_squares:
            piece = board.piece_at(square)
            if piece:
                if piece.color == chess.WHITE:
                    control_score += 1
                else:
                    control_score -= 1
        
        return control_score
    
    def _evaluate_move_criticality(self, board: chess.Board, move: chess.Move) -> float:
        """评估步法关键性"""
        # 这里简化处理，实际应该使用引擎评估
        board.pop()  # 退回一步
        
        # 评估移动前的局面
        score_before = self._simple_evaluation(board)
        
        # 执行步法
        board.push(move)
        score_after = self._simple_evaluation(board)
        
        # 退回
        board.pop()
        
        return abs(score_after - score_before)
    
    def _simple_evaluation(self, board: chess.Board) -> float:
        """简化局面评估"""
        material = self._calculate_material(board)
        center = self._calculate_center_control(board) * 0.1
        return material + center

# 数据采集主程序
async def main():
    collector = ChessDataCollector()
    
    # 常见开局
    openings = [
        "Italian Game",
        "Sicilian Defense",
        "French Defense",
        "Ruy Lopez",
        "Queen's Gambit",
        "King's Indian Defense"
    ]
    
    all_games = []
    
    # 并行采集数据
    tasks = []
    for opening in openings:
        task = collector.fetch_games_by_opening(opening, limit=200)
        tasks.append(task)
    
    results = await asyncio.gather(*tasks)
    
    for games in results:
        all_games.extend(games)
    
    print(f"共采集 {len(all_games)} 个棋局")
    
    # 保存数据
    with open("./data/chess_games.json", "w", encoding="utf-8") as f:
        games_dict = [game.__dict__ for game in all_games]
        for game in games_dict:
            if "embeddings" in game and isinstance(game["embeddings"], np.ndarray):
                game["embeddings"] = game["embeddings"].tolist()
        json.dump(games_dict, f, ensure_ascii=False, indent=2)
    
    print("数据采集完成！")

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

2.2 数据导入脚本

# data_importer.py
import json
import numpy as np
from pymilvus import connections, Collection, CollectionSchema, FieldSchema, DataType
from clickhouse_driver import Client as ClickhouseClient
from elasticsearch import Elasticsearch
import redis
from tqdm import tqdm
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ChessDataImporter:
    """象棋数据导入器"""
    
    def __init__(self):
        # 连接Milvus
        self._connect_milvus()
        
        # 连接ClickHouse
        self._connect_clickhouse()
        
        # 连接Elasticsearch
        self._connect_elasticsearch()
        
        # 连接Redis
        self._connect_redis()
    
    def _connect_milvus(self):
        """连接Milvus"""
        try:
            connections.connect(
                alias="default",
                host="localhost",
                port="19530"
            )
            logger.info("✅ Milvus连接成功")
            
            # 定义向量字段
            fields = [
                FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
                FieldSchema(name="game_id", dtype=DataType.VARCHAR, max_length=100),
                FieldSchema(name="opening", dtype=DataType.VARCHAR, max_length=100),
                FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=768),
                FieldSchema(name="moves_count", dtype=DataType.INT32),
                FieldSchema(name="result", dtype=DataType.VARCHAR, max_length=10),
                FieldSchema(name="rating_avg", dtype=DataType.INT32),
                FieldSchema(name="critical_moves", dtype=DataType.JSON),
                FieldSchema(name="metadata", dtype=DataType.JSON)
            ]
            
            schema = CollectionSchema(fields, description="象棋棋谱向量库")
            
            # 创建集合
            self.chess_collection = Collection("chess_games_v2", schema)
            
            # 创建索引
            index_params = {
                "metric_type": "L2",
                "index_type": "IVF_FLAT",
                "params": {"nlist": 1024}
            }
            self.chess_collection.create_index("embedding", index_params)
            
        except Exception as e:
            logger.error(f"❌ Milvus连接失败: {e}")
            raise
    
    def _connect_clickhouse(self):
        """连接ClickHouse"""
        try:
            self.clickhouse_client = ClickhouseClient(
                host="localhost",
                port=9000,
                user="default",
                password="",
                database="chess_db"
            )
            
            # 创建数据库和表
            self.clickhouse_client.execute("CREATE DATABASE IF NOT EXISTS chess_db")
            
            # 棋局统计表
            self.clickhouse_client.execute('''
            CREATE TABLE IF NOT EXISTS chess_db.game_stats
            (
                game_id String,
                date Date,
                white_player String,
                black_player String,
                result Enum8('1-0' = 1, '0-1' = 2, '1/2-1/2' = 3, '*' = 4),
                moves_count UInt16,
                opening String,
                rating_white UInt16,
                rating_black UInt16,
                time_control String,
                material_balance Array(Float32),
                center_control Array(Float32),
                critical_positions Array(UInt16),
                created_at DateTime DEFAULT now()
            ) ENGINE = MergeTree()
            ORDER BY (date, opening, rating_white)
            PARTITION BY toYYYYMM(date)
            ''')
            
            # 玩家统计表
            self.clickhouse_client.execute('''
            CREATE TABLE IF NOT EXISTS chess_db.player_stats
            (
                player_name String,
                total_games UInt32,
                wins UInt32,
                losses UInt32,
                draws UInt32,
                avg_rating Float32,
                favorite_openings Array(String),
                win_rate_by_opening Map(String, Float32),
                last_updated DateTime DEFAULT now()
            ) ENGINE = MergeTree()
            ORDER BY (player_name, avg_rating)
            ''')
            
            logger.info("✅ ClickHouse连接成功")
            
        except Exception as e:
            logger.error(f"❌ ClickHouse连接失败: {e}")
            raise
    
    def _connect_elasticsearch(self):
        """连接Elasticsearch"""
        try:
            self.es = Elasticsearch(
                hosts=["http://localhost:9200"],
                timeout=30
            )
            
            # 创建象棋知识索引
            if not self.es.indices.exists(index="chess_knowledge"):
                mapping = {
                    "mappings": {
                        "properties": {
                            "title": {"type": "text", "analyzer": "ik_max_word"},
                            "content": {"type": "text", "analyzer": "ik_max_word"},
                            "category": {"type": "keyword"},
                            "tags": {"type": "keyword"},
                            "difficulty": {"type": "keyword"},
                            "rating": {"type": "float"},
                            "created_at": {"type": "date"},
                            "embedding": {
                                "type": "dense_vector",
                                "dims": 768,
                                "index": True,
                                "similarity": "cosine"
                            }
                        }
                    }
                }
                self.es.indices.create(index="chess_knowledge", body=mapping)
            
            logger.info("✅ Elasticsearch连接成功")
            
        except Exception as e:
            logger.error(f"❌ Elasticsearch连接失败: {e}")
            raise
    
    def _connect_redis(self):
        """连接Redis"""
        try:
            self.redis_client = redis.Redis(
                host="localhost",
                port=6379,
                db=0,
                decode_responses=True
            )
            logger.info("✅ Redis连接成功")
        except Exception as e:
            logger.error(f"❌ Redis连接失败: {e}")
            raise
    
    def import_games(self, file_path: str):
        """导入棋局数据"""
        logger.info(f"开始导入棋局数据: {file_path}")
        
        with open(file_path, "r", encoding="utf-8") as f:
            games = json.load(f)
        
        milvus_data = []
        clickhouse_data = []
        es_data = []
        
        for game in tqdm(games, desc="处理棋局"):
            try:
                # 准备Milvus数据
                milvus_data.append({
                    "game_id": game["game_id"],
                    "opening": game["opening"],
                    "embedding": np.array(game.get("embeddings", [0]*768), dtype=np.float32),
                    "moves_count": len(game["moves"]),
                    "result": game["result"],
                    "rating_avg": (game["rating_white"] + game["rating_black"]) // 2,
                    "critical_moves": json.dumps(game.get("analysis", {}).get("critical_moves", [])),
                    "metadata": json.dumps({
                        "white": game["white_player"],
                        "black": game["black_player"],
                        "time_control": game["time_control"]
                    })
                })
                
                # 准备ClickHouse数据
                clickhouse_data.append([
                    game["game_id"],
                    game["date"][:10],  # 只取日期部分
                    game["white_player"],
                    game["black_player"],
                    game["result"],
                    len(game["moves"]),
                    game["opening"],
                    game["rating_white"],
                    game["rating_black"],
                    game["time_control"],
                    game.get("analysis", {}).get("material_balance", []),
                    game.get("analysis", {}).get("center_control", []),
                    [m["move_number"] for m in game.get("analysis", {}).get("critical_moves", [])]
                ])
                
                # 准备Elasticsearch数据（棋局摘要）
                es_doc = {
                    "title": f"对局分析: {game['white_player']} vs {game['black_player']}",
                    "content": f"开局: {game['opening']}, 结果: {game['result']}",
                    "category": "game_analysis",
                    "tags": [game["opening"], game["white_player"], game["black_player"]],
                    "difficulty": "intermediate",
                    "rating": (game["rating_white"] + game["rating_black"]) / 2,
                    "created_at": game["date"],
                    "embedding": game.get("embeddings", [0]*768)
                }
                es_data.append(es_doc)
                
                # 批量插入
                if len(milvus_data) >= 1000:
                    self._batch_insert(milvus_data, clickhouse_data, es_data)
                    milvus_data, clickhouse_data, es_data = [], [], []
                    
            except Exception as e:
                logger.error(f"处理棋局 {game.get('game_id', 'unknown')} 失败: {e}")
                continue
        
        # 插入剩余数据
        if milvus_data:
            self._batch_insert(milvus_data, clickhouse_data, es_data)
        
        logger.info("棋局数据导入完成！")
    
    def _batch_insert(self, milvus_data, clickhouse_data, es_data):
        """批量插入数据"""
        try:
            # 插入Milvus
            if milvus_data:
                self.chess_collection.insert(milvus_data)
                self.chess_collection.flush()
            
            # 插入ClickHouse
            if clickhouse_data:
                self.clickhouse_client.execute(
                    "INSERT INTO chess_db.game_stats VALUES",
                    clickhouse_data
                )
            
            # 插入Elasticsearch
            if es_data:
                for i, doc in enumerate(es_data):
                    self.es.index(
                        index="chess_knowledge",
                        body=doc,
                        id=f"game_{i}_{len(es_data)}"
                    )
                    
        except Exception as e:
            logger.error(f"批量插入失败: {e}")
            raise

# 导入象棋知识库
def import_chess_knowledge():
    """导入象棋知识文档"""
    es = Elasticsearch(["http://localhost:9200"])
    
    knowledge_docs = [
        {
            "title": "象棋基本规则",
            "content": """
            中国象棋是由两人轮流走子，以将死对方将（帅）为胜的棋类游戏。
            棋盘由9条竖线和10条横线组成，中间有楚河汉界。
            棋子分为红黑两方，每方有16个棋子：将（帅）1、士2、象2、马2、车2、炮2、兵（卒）5。
            """,
            "category": "basic_rules",
            "tags": ["规则", "入门", "基础"],
            "difficulty": "beginner",
            "rating": 1.0
        },
        {
            "title": "常用开局策略",
            "content": """
            1. 中炮对屏风马：最常见的开局之一，攻势猛烈
            2. 飞相局：稳健型开局，注重阵地战
            3. 仙人指路：试探性开局，灵活多变
            4. 过宫炮：集中火力于一侧，寻求突破
            """,
            "category": "opening_strategy",
            "tags": ["开局", "策略", "战术"],
            "difficulty": "intermediate",
            "rating": 3.0
        },
        {
            "title": "残局技巧",
            "content": """
            1. 单马必胜单士：掌握马擒单士的固定走法
            2. 炮士必胜双士：利用炮的灵活性
            3. 车兵对车象：注意避免长兑车
            4. 马兵对士象全：需要耐心和精确计算
            """,
            "category": "endgame",
            "tags": ["残局", "技巧", "高级"],
            "difficulty": "advanced",
            "rating": 4.5
        }
    ]
    
    for i, doc in enumerate(knowledge_docs):
        es.index(index="chess_knowledge", id=f"knowledge_{i}", body=doc)
    
    print("知识库导入完成！")

if __name__ == "__main__":
    importer = ChessDataImporter()
    
    # 导入棋局数据
    importer.import_games("./data/chess_games.json")
    
    # 导入知识库
    import_chess_knowledge()

第三章：MindSpore模型开发 - 象棋AI大脑

3.1 象棋Transformer模型

# chess_transformer.py
import mindspore as ms
import mindspore.nn as nn
import mindspore.ops as ops
import mindspore.numpy as mnp
from mindspore import Tensor, Parameter
from mindspore.common.initializer import XavierUniform, Normal
import numpy as np
from typing import Tuple, Optional, List, Dict

class ChessEmbedding(nn.Cell):
    """象棋局面嵌入层"""
    
    def __init__(self, 
                 board_size: int = 90,  # 10*9棋盘
                 piece_types: int = 14,  # 7种棋子×2个颜色
                 embedding_dim: int = 512,
                 max_moves: int = 200):
        super().__init__()
        
        # 棋子类型嵌入
        self.piece_embedding = nn.Embedding(piece_types, embedding_dim)
        
        # 位置嵌入（棋盘坐标）
        self.position_embedding = nn.Embedding(board_size, embedding_dim)
        
        # 局面特征嵌入
        self.feature_embedding = nn.Dense(32, embedding_dim)
        
        # 历史着法嵌入
        self.move_embedding = nn.Embedding(max_moves, embedding_dim)
        
        # 特殊标记嵌入
        self.special_tokens = nn.Embedding(10, embedding_dim)
        
        self.layer_norm = nn.LayerNorm([embedding_dim])
        self.dropout = nn.Dropout(0.1)
        
    def construct(self, 
                  board_state: Tensor,  # [batch, 90]
                  piece_types: Tensor,   # [batch, 90]
                  position_ids: Tensor,  # [batch, 90]
                  features: Tensor,      # [batch, 32]
                  move_history: Tensor   # [batch, 20]
                 ) -> Tensor:
        
        # 棋子嵌入
        piece_emb = self.piece_embedding(piece_types)  # [batch, 90, dim]
        
        # 位置嵌入
        pos_emb = self.position_embedding(position_ids)  # [batch, 90, dim]
        
        # 特征嵌入
        feat_emb = self.feature_embedding(features).unsqueeze(1)  # [batch, 1, dim]
        
        # 历史着法嵌入
        move_emb = self.move_embedding(move_history).mean(axis=1, keep_dims=True)  # [batch, 1, dim]
        
        # 合并所有嵌入
        embeddings = piece_emb + pos_emb
        embeddings = ops.concat([embeddings, feat_emb, move_emb], axis=1)  # [batch, 92, dim]
        
        # 添加特殊标记
        batch_size = board_state.shape[0]
        special_tokens = self.special_tokens(
            Tensor(np.arange(2), dtype=ms.int32).expand_dims(0).repeat(batch_size, 0)
        )  # [batch, 2, dim]
        
        embeddings = ops.concat([special_tokens, embeddings], axis=1)  # [batch, 94, dim]
        
        return self.dropout(self.layer_norm(embeddings))

class ChessTransformerLayer(nn.Cell):
    """象棋专用Transformer层"""
    
    def __init__(self, 
                 d_model: int = 512,
                 n_heads: int = 8,
                 d_ff: int = 2048,
                 dropout: float = 0.1):
        super().__init__()
        
        # 多头注意力
        self.self_attn = nn.MultiheadAttention(
            embed_dim=d_model,
            num_heads=n_heads,
            dropout=dropout,
            batch_first=True
        )
        
        # 前馈网络
        self.ffn = nn.SequentialCell([
            nn.Dense(d_model, d_ff),
            nn.GELU(),
            nn.Dropout(dropout),
            nn.Dense(d_ff, d_model),
            nn.Dropout(dropout)
        ])
        
        # 层归一化
        self.norm1 = nn.LayerNorm([d_model])
        self.norm2 = nn.LayerNorm([d_model])
        self.dropout1 = nn.Dropout(dropout)
        self.dropout2 = nn.Dropout(dropout)
        
    def construct(self, 
                  x: Tensor,
                  attention_mask: Optional[Tensor] = None) -> Tensor:
        
        # 自注意力
        attn_output, _ = self.self_attn(x, x, x, attn_mask=attention_mask)
        x = x + self.dropout1(attn_output)
        x = self.norm1(x)
        
        # 前馈网络
        ffn_output = self.ffn(x)
        x = x + self.dropout2(ffn_output)
        x = self.norm2(x)
        
        return x

class ChessEvaluationHead(nn.Cell):
    """局面评估头"""
    
    def __init__(self, d_model: int = 512):
        super().__init__()
        
        self.evaluation_net = nn.SequentialCell([
            nn.Dense(d_model, 256),
            nn.ReLU(),
            nn.Dropout(0.1),
            nn.Dense(256, 128),
            nn.ReLU(),
            nn.Dense(128, 64),
            nn.ReLU(),
            nn.Dense(64, 1)  # 局面评分
        ])
        
        self.win_prob_net = nn.SequentialCell([
            nn.Dense(d_model, 128),
            nn.ReLU(),
            nn.Dense(128, 3),  # 红胜、黑胜、和棋概率
            nn.Softmax(axis=-1)
        ])
        
        self.best_move_net = nn.SequentialCell([
            nn.Dense(d_model, 256),
            nn.ReLU(),
            nn.Dense(256, 90 * 90)  # 所有可能的移动
        ])
        
    def construct(self, x: Tensor) -> Tuple[Tensor, Tensor, Tensor]:
        """返回局面评分、胜率、最佳着法"""
        # 取CLS标记的输出
        cls_output = x[:, 0, :]
        
        # 局面评分（[-1, 1]之间，正值表示红方优势）
        evaluation = ops.tanh(self.evaluation_net(cls_output))
        
        # 胜率预测
        win_prob = self.win_prob_net(cls_output)
        
        # 最佳着法预测
        move_logits = self.best_move_net(cls_output)
        move_logits = move_logits.view(-1, 90, 90)  # 起始位置×目标位置
        
        return evaluation, win_prob, move_logits

class ChessQAModel(nn.Cell):
    """象棋问答模型"""
    
    def __init__(self, 
                 vocab_size: int = 30000,
                 d_model: int = 512,
                 n_layers: int = 6,
                 n_heads: int = 8,
                 max_seq_len: int = 512):
        super().__init__()
        
        # 文本编码器
        self.text_embedding = nn.Embedding(vocab_size, d_model)
        self.text_position_embedding = nn.Embedding(max_seq_len, d_model)
        
        # 象棋局面编码器
        self.chess_embedding = ChessEmbedding(embedding_dim=d_model)
        
        # Transformer编码器
        self.transformer_layers = nn.CellList([
            ChessTransformerLayer(d_model, n_heads, d_ff=d_model*4)
            for _ in range(n_layers)
        ])
        
        # 问答头
        self.qa_head = nn.SequentialCell([
            nn.Dense(d_model * 2, d_model),
            nn.ReLU(),
            nn.Dropout(0.1),
            nn.Dense(d_model, vocab_size)
        ])
        
        # 局面评估头
        self.evaluation_head = ChessEvaluationHead(d_model)
        
        # 层归一化和Dropout
        self.layer_norm = nn.LayerNorm([d_model])
        self.dropout = nn.Dropout(0.1)
        
        self.d_model = d_model
        self.max_seq_len = max_seq_len
        
    def construct(self,
                  text_input: Tensor,  # [batch, seq_len]
                  board_state: Tensor,
                  piece_types: Tensor,
                  position_ids: Tensor,
                  features: Tensor,
                  move_history: Tensor,
                  attention_mask: Optional[Tensor] = None) -> Dict[str, Tensor]:
        
        batch_size = text_input.shape[0]
        
        # 文本嵌入
        text_seq_len = text_input.shape[1]
        text_positions = Tensor(np.arange(text_seq_len), dtype=ms.int32)
        text_positions = text_positions.expand_dims(0).repeat(batch_size, 0)
        
        text_emb = self.text_embedding(text_input)
        text_pos_emb = self.text_position_embedding(text_positions)
        text_combined = self.dropout(text_emb + text_pos_emb)
        
        # 象棋局面嵌入
        chess_emb = self.chess_embedding(
            board_state, piece_types, position_ids, features, move_history
        )
        
        # 合并文本和局面信息
        combined_emb = ops.concat([text_combined, chess_emb], axis=1)
        combined_emb = self.layer_norm(combined_emb)
        
        # 通过Transformer层
        for layer in self.transformer_layers:
            combined_emb = layer(combined_emb, attention_mask)
        
        # 分离文本和局面输出
        text_output = combined_emb[:, :text_seq_len, :]
        chess_output = combined_emb[:, text_seq_len:, :]
        
        # 生成回答
        answer_logits = self.qa_head(
            ops.concat([text_output[:, -1, :], chess_output[:, 0, :]], axis=-1)
        )
        
        # 局面评估
        evaluation, win_prob, move_logits = self.evaluation_head(chess_output)
        
        return {
            "answer_logits": answer_logits,
            "evaluation": evaluation,
            "win_prob": win_prob,
            "move_logits": move_logits,
            "text_output": text_output,
            "chess_output": chess_output
        }

class ChessAITrainer:
    """象棋AI训练器"""
    
    def __init__(self, model: ChessQAModel, learning_rate: float = 1e-4):
        self.model = model
        self.optimizer = nn.Adam(
            model.trainable_params(),
            learning_rate=learning_rate,
            weight_decay=1e-5
        )
        
        # 损失函数
        self.qa_loss = nn.CrossEntropyLoss()
        self.eval_loss = nn.MSELoss()
        self.move_loss = nn.CrossEntropyLoss()
        
        # 训练网络
        self.train_net = self._build_train_network()
        
    def _build_train_network(self):
        """构建训练网络"""
        
        class TrainOneStepCell(nn.Cell):
            def __init__(self, trainer):
                super().__init__()
                self.trainer = trainer
                self.optimizer = trainer.optimizer
                self.weights = self.optimizer.parameters
                self.grad = ops.GradOperation(get_by_list=True)
                
            def construct(self, *inputs):
                # 计算损失
                outputs = self.trainer.model(*inputs[:-1])
                loss = self.trainer.compute_loss(outputs, inputs[-1])
                
                # 计算梯度
                grads = self.grad(self.trainer.model, self.weights)(*inputs[:-1])
                
                # 更新参数
                self.optimizer(grads)
                
                return loss, outputs
        
        return TrainOneStepCell(self)
    
    def compute_loss(self, outputs: Dict, labels: Dict) -> Tensor:
        """计算总损失"""
        # 问答损失
        qa_loss = self.qa_loss(outputs["answer_logits"], labels["answer_labels"])
        
        # 局面评估损失
        eval_loss = self.eval_loss(outputs["evaluation"], labels["evaluation_labels"])
        
        # 最佳着法损失
        move_loss = self.move_loss(
            outputs["move_logits"].view(-1, 90*90),
            labels["move_labels"].view(-1)
        )
        
        # 组合损失
        total_loss = qa_loss + 0.5 * eval_loss + 0.3 * move_loss
        
        return total_loss
    
    def train_step(self, batch_data):
        """单步训练"""
        return self.train_net(*batch_data)

3.2 象棋局面编码器

# position_encoder.py
import chess
import numpy as np
from typing import List, Tuple, Dict
import re

class ChessPositionEncoder:
    """象棋局面编码器"""
    
    def __init__(self):
        # 棋子映射表
        self.piece_map = {
            'K': 1, 'k': 2,  # 将/帅
            'A': 3, 'a': 4,  # 士
            'B': 5, 'b': 6,  # 象
            'N': 7, 'n': 8,  # 马
            'R': 9, 'r': 10, # 车
            'C': 11, 'c': 12, # 炮
            'P': 13, 'p': 14  # 兵/卒
        }
        
        # 位置编码
        self.position_encoding = self._create_position_encoding()
        
    def _create_position_encoding(self) -> np.ndarray:
        """创建位置编码矩阵"""
        encoding = np.zeros((10, 9))
        
        # 棋盘中心权重更高
        for i in range(10):
            for j in range(9):
                # 中心权重
                center_i, center_j = 4.5, 4
                distance = np.sqrt((i - center_i)**2 + (j - center_j)**2)
                encoding[i, j] = 1.0 / (1 + distance)
        
        return encoding
    
    def encode_fen(self, fen: str) -> Dict[str, np.ndarray]:
        """编码FEN字符串"""
        # 解析FEN（简化版）
        parts = fen.split()
        board_str = parts[0]
        
        # 初始化棋盘矩阵
        board_matrix = np.zeros((10, 9), dtype=np.int32)
        piece_matrix = np.zeros((10, 9), dtype=np.int32)
        
        rows = board_str.split('/')
        for i, row in enumerate(rows):
            col = 0
            for char in row:
                if char.isdigit():
                    col += int(char)
                else:
                    piece_id = self.piece_map.get(char, 0)
                    board_matrix[i, col] = 1 if piece_id > 0 else 0
                    piece_matrix[i, col] = piece_id
                    col += 1
        
        # 提取特征
        features = self._extract_features(board_matrix, piece_matrix)
        
        return {
            "board_state": board_matrix.flatten(),
            "piece_types": piece_matrix.flatten(),
            "position_ids": np.arange(90),
            "features": features
        }
    
    def _extract_features(self, 
                         board_matrix: np.ndarray,
                         piece_matrix: np.ndarray) -> np.ndarray:
        """提取局面特征"""
        features = []
        
        # 1. 子力价值
        red_pieces, black_pieces = self._count_pieces(piece_matrix)
        features.extend(red_pieces)
        features.extend(black_pieces)
        
        # 2. 中心控制
        center_control = self._calculate_center_control(piece_matrix)
        features.extend(center_control)
        
        # 3. 兵线结构
        pawn_structure = self._analyze_pawn_structure(piece_matrix)
        features.extend(pawn_structure)
        
        # 4. 王的安全度
        king_safety = self._evaluate_king_safety(piece_matrix)
        features.extend(king_safety)
        
        # 5. 车炮位置
        rook_cannon_positions = self._evaluate_rook_cannon(piece_matrix)
        features.extend(rook_cannon_positions)
        
        return np.array(features, dtype=np.float32)
    
    def _count_pieces(self, piece_matrix: np.ndarray) -> Tuple[List[int], List[int]]:
        """统计棋子数量"""
        red_pieces = [0] * 7  # 将士象马车炮兵
        black_pieces = [0] * 7
        
        piece_counts = {
            1: (0, 0), 2: (0, 1),  # 将帅
            3: (1, 0), 4: (1, 1),  # 士
            5: (2, 0), 6: (2, 1),  # 象
            7: (3, 0), 8: (3, 1),  # 马
            9: (4, 0), 10: (4, 1), # 车
            11: (5, 0), 12: (5, 1), # 炮
            13: (6, 0), 14: (6, 1)  # 兵卒
        }
        
        for i in range(10):
            for j in range(9):
                piece_id = piece_matrix[i, j]
                if piece_id > 0:
                    idx, color = piece_counts[piece_id]
                    if color == 0:  # 红方
                        red_pieces[idx] += 1
                    else:  # 黑方
                        black_pieces[idx] += 1
        
        return red_pieces, black_pieces
    
    def _calculate_center_control(self, piece_matrix: np.ndarray) -> List[float]:
        """计算中心控制"""
        center_squares = [(4, 4), (5, 4), (4, 5), (5, 5)]
        control_score = [0.0, 0.0]  # 红方控制，黑方控制
        
        for i, j in center_squares:
            piece_id = piece_matrix[i, j]
            if piece_id > 0:
                if piece_id in [1, 3, 5, 7, 9, 11, 13]:  # 红方
                    control_score[0] += 1
                else:  # 黑方
                    control_score[1] += 1
        
        return control_score
    
    def _analyze_pawn_structure(self, piece_matrix: np.ndarray) -> List[float]:
        """分析兵线结构"""
        features = []
        
        # 红兵结构
        red_pawns = []
        for i in range(10):
            for j in range(9):
                if piece_matrix[i, j] == 13:  # 红兵
                    red_pawns.append((i, j))
        
        # 黑卒结构
        black_pawns = []
        for i in range(10):
            for j in range(9):
                if piece_matrix[i, j] == 14:  # 黑卒
                    black_pawns.append((i, j))
        
        # 兵线连接性
        red_connected = self._calculate_connectedness(red_pawns)
        black_connected = self._calculate_connectedness(black_pawns)
        
        features.extend([len(red_pawns), len(black_pawns), 
                        red_connected, black_connected])
        
        return features
    
    def _calculate_connectedness(self, positions: List[Tuple[int, int]]) -> float:
        """计算棋子连接性"""
        if len(positions) <= 1:
            return 0.0
        
        connected_pairs = 0
        for i in range(len(positions)):
            for j in range(i+1, len(positions)):
                x1, y1 = positions[i]
                x2, y2 = positions[j]
                distance = abs(x1 - x2) + abs(y1 - y2)
                if distance == 1:  # 相邻
                    connected_pairs += 1
        
        max_pairs = len(positions) * (len(positions) - 1) / 2
        return connected_pairs / max_pairs if max_pairs > 0 else 0.0
    
    def _evaluate_king_safety(self, piece_matrix: np.ndarray) -> List[float]:
        """评估王的安全度"""
        # 寻找红将和黑将位置
        red_king_pos = None
        black_king_pos = None
        
        for i in range(10):
            for j in range(9):
                if piece_matrix[i, j] == 1:  # 红将
                    red_king_pos = (i, j)
                elif piece_matrix[i, j] == 2:  # 黑将
                    black_king_pos = (i, j)
        
        # 计算防守子力
        red_defenders = self._count_defenders(piece_matrix, red_king_pos, is_red=True)
        black_defenders = self._count_defenders(piece_matrix, black_king_pos, is_red=False)
        
        return [red_defenders, black_defenders]
    
    def _count_defenders(self, 
                        piece_matrix: np.ndarray, 
                        king_pos: Tuple[int, int],
                        is_red: bool) -> float:
        """统计防守子力"""
        if king_pos is None:
            return 0.0
        
        defender_pieces = [3, 5, 7, 9, 11] if is_red else [4, 6, 8, 10, 12]
        defense_score = 0.0
        
        ki, kj = king_pos
        
        for i in range(max(0, ki-2), min(10, ki+3)):
            for j in range(max(0, kj-2), min(9, kj+3)):
                if piece_matrix[i, j] in defender_pieces:
                    distance = abs(i - ki) + abs(j - kj)
                    defense_score += 1.0 / (1 + distance)
        
        return defense_score
    
    def _evaluate_rook_cannon(self, piece_matrix: np.ndarray) -> List[float]:
        """评估车炮位置"""
        features = []
        
        # 查找车和炮
        red_rooks, red_cannons = [], []
        black_rooks, black_cannons = [], []
        
        for i in range(10):
            for j in range(9):
                piece_id = piece_matrix[i, j]
                if piece_id == 9:  # 红车
                    red_rooks.append((i, j))
                elif piece_id == 10:  # 黑车
                    black_rooks.append((i, j))
                elif piece_id == 11:  # 红炮
                    red_cannons.append((i, j))
                elif piece_id == 12:  # 黑炮
                    black_cannons.append((i, j))
        
        # 评估车的位置（控制开放线）
        red_rook_score = self._evaluate_rook_position(red_rooks, piece_matrix, is_red=True)
        black_rook_score = self._evaluate_rook_position(black_rooks, piece_matrix, is_red=False)
        
        # 评估炮的位置（有炮架）
        red_cannon_score = self._evaluate_cannon_position(red_cannons, piece_matrix, is_red=True)
        black_cannon_score = self._evaluate_cannon_position(black_cannons, piece_matrix, is_red=False)
        
        features.extend([red_rook_score, black_rook_score, 
                        red_cannon_score, black_cannon_score])
        
        return features
    
    def _evaluate_rook_position(self, 
                               rook_positions: List[Tuple[int, int]],
                               piece_matrix: np.ndarray,
                               is_red: bool) -> float:
        """评估车的位置"""
        if not rook_positions:
            return 0.0
        
        score = 0.0
        
        for ri, rj in rook_positions:
            # 检查垂直和水平线上的控制
            # 垂直线
            open_files = 0
            for i in range(10):
                if piece_matrix[i, rj] == 0:  # 空位
                    open_files += 1
            
            # 水平线
            open_ranks = 0
            for j in range(9):
                if piece_matrix[ri, j] == 0:  # 空位
                    open_ranks += 1
            
            score += (open_files + open_ranks) / (10 + 9)
        
        return score / len(rook_positions)
    
    def _evaluate_cannon_position(self,
                                 cannon_positions: List[Tuple[int, int]],
                                 piece_matrix: np.ndarray,
                                 is_red: bool) -> float:
        """评估炮的位置"""
        if not cannon_positions:
            return 0.0
        
        score = 0.0
        
        for ci, cj in cannon_positions:
            # 检查是否有炮架
            has_mount = False
            
            # 检查四个方向
            directions = [(-1, 0), (1, 0), (0, -1), (0, 1)]
            for di, dj in directions:
                ni, nj = ci + di, cj + dj
                if 0 <= ni < 10 and 0 <= nj < 9:
                    if piece_matrix[ni, nj] > 0:  # 有棋子可以作为炮架
                        has_mount = True
                        break
            
            if has_mount:
                score += 1.0
        
        return score / len(cannon_positions)