如何构建实时评估更新的Chess应用？FastAPI+React架构设计

问题背景 #

我正在开发一款国际象棋应用，核心流程如下：

• 前端采用Chess.js库解析从Chess.com下载的PGN文件，存储moveList和moveIndex
• 当moveIndex变化时，前端将当前局面的FEN字符串发送至运行持久化Stockfish引擎的FastAPI端点
• 后端通过Python-Chess处理引擎输出，再将评估结果返回至前端

需要实现类似Chess.com的持续更新评估功能，让前端UI能实时刷新。我找到了Python-Chess的无限分析代码，但不确定如何适配FastAPI（FastAPI自带事件循环，无需调用asyncio.run）：

import asyncio
import chess
import chess.engine

async def main() -> None:
    transport, engine = await chess.engine.popen_uci("/usr/bin/stockfish")

    with await engine.analysis(chess.Board()) as analysis:
        async for info in analysis:
            print(info.get("score"), info.get("pv"))

            # Arbitrary stop condition.
            if info.get("seldepth", 0) > 20:
                break

    await engine.quit()

asyncio.run(main())

高层级代码结构设计 #

一、前端结构 #

核心状态与逻辑

• 维护三个核心状态：
  • currentFen：随moveIndex更新的当前局面FEN字符串
  • evaluation：存储实时返回的评估分数、最佳走法（PV）、分析深度等数据
  • isAnalyzing：标记是否正在分析，避免重复发起请求
• 使用WebSocket替代HTTP请求：HTTP是单次响应模式，无法实现持续推送评估结果，WebSocket长连接可实时传输增量分析数据
• UI绑定：将evaluation状态与界面元素绑定，每次收到新数据立即刷新分数、最佳走法等展示内容

前端示例代码

// 建立WebSocket连接
const ws = new WebSocket(`ws://${window.location.host}/ws/analysis`);

// 接收后端推送的评估数据，更新UI
ws.onmessage = (event) => {
    const evaluation = JSON.parse(event.data);
    document.getElementById('score-display').textContent = evaluation.score;
    document.getElementById('pv-display').textContent = evaluation.pv.join(' ');
    document.getElementById('depth-display').textContent = `深度: ${evaluation.depth}`;
};

// 当moveIndex变化时发送新FEN
function triggerAnalysis(newFen) {
    ws.send(newFen);
}

// 页面卸载时关闭连接
window.addEventListener('beforeunload', () => ws.close());

二、后端结构 #

核心组件

1. 持久化Stockfish引擎实例：全局维护一个引擎连接，避免每次请求重启引擎（启动开销大）
2. WebSocket端点：处理前端的FEN接收、评估结果推送、分析任务终止
3. 分析任务管理器：跟踪当前活跃的分析任务，收到新FEN时立即取消旧任务，避免资源浪费

后端关键代码实现

全局引擎初始化

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import chess
import chess.engine
import asyncio

app = FastAPI()

# 全局持久化Stockfish引擎实例与锁（引擎非线程安全）
engine: chess.engine.UciProtocol = None
engine_lock = asyncio.Lock()

@app.on_event("startup")
async def startup_engine():
    global engine
    # 替换为你的Stockfish引擎路径
    transport, engine = await chess.engine.popen_uci("/usr/bin/stockfish")

@app.on_event("shutdown")
async def shutdown_engine():
    global engine
    await engine.quit()

WebSocket实时分析端点

class AnalysisTaskManager:
    def __init__(self):
        self.current_task: asyncio.Task | None = None

task_manager = AnalysisTaskManager()

@app.websocket("/ws/analysis")
async def handle_analysis(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            # 接收前端发送的FEN字符串
            fen = await websocket.receive_text()
            board = chess.Board(fen)

            # 取消之前的分析任务
            if task_manager.current_task and not task_manager.current_task.done():
                task_manager.current_task.cancel()

            # 启动新的分析任务
            async def run_analysis():
                try:
                    async with engine_lock:
                        with await engine.analysis(board) as analysis:
                            async for info in analysis:
                                # 提取关键评估数据
                                score = str(info.get("score")) if info.get("score") else None
                                pv = [move.uci() for move in info.get("pv")] if info.get("pv") else []
                                depth = info.get("depth")
                                seldepth = info.get("seldepth")

                                # 推送数据至前端
                                await websocket.send_json({
                                    "score": score,
                                    "pv": pv,
                                    "depth": depth,
                                    "seldepth": seldepth
                                })

                                # 可选：设置最大分析深度，避免无限运行
                                if seldepth and seldepth > 20:
                                    break
                except asyncio.CancelledError:
                    # 任务被取消时正常退出
                    pass

            task_manager.current_task = asyncio.create_task(run_analysis())
    except WebSocketDisconnect:
        # 前端断开连接时清理任务
        if task_manager.current_task and not task_manager.current_task.done():
            task_manager.current_task.cancel()

后端注意事项

• 引擎锁：Stockfish非线程安全，必须用engine_lock确保同一时间只有一个分析任务占用引擎
• 任务取消：每次收到新FEN时必须终止旧任务，避免多任务抢占引擎资源
• 异常处理：覆盖WebSocket断开、任务取消、引擎异常等场景，保证后端稳定性

内容的提问来源于stack exchange，提问作者thuang