基于Aiogram 3.4实现用户间限时对话的FSM方案问题求助

针对你遇到的FSM状态冲突、用户ID传递问题，以下是几种实用的解决方案，无需使用GLOBAL策略即可实现需求：

方案一：数据库独立存储对话关联（推荐） #

放弃用FSM管理对话关系，直接通过数据库维护双用户的会话关联，逻辑更清晰且避免状态冲突。

1. 定义对话会话模型（SQLAlchemy示例） #

from sqlalchemy import Column, Integer, BigInteger, DateTime
from sqlalchemy.ext.asyncio import AsyncSession
from datetime import datetime, timedelta
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class DialogSession(Base):
    __tablename__ = "dialog_sessions"
    id = Column(Integer, primary_key=True, autoincrement=True)
    user1_id = Column(BigInteger, nullable=False, index=True)
    user2_id = Column(BigInteger, nullable=False, index=True)
    expires_at = Column(DateTime, nullable=False)

    @classmethod
    async def create(cls, db_session: AsyncSession, user1: int, user2: int, duration: int = 30):
        """创建限时会话，默认30分钟过期"""
        session = cls(
            user1_id=user1,
            user2_id=user2,
            expires_at=datetime.now() + timedelta(minutes=duration)
        )
        db_session.add(session)
        await db_session.commit()
        return session

    @classmethod
    async def get_active(cls, db_session: AsyncSession, user_id: int):
        """获取当前用户的活跃会话"""
        return await db_session.query(cls).filter(
            (cls.user1_id == user_id) | (cls.user2_id == user_id),
            cls.expires_at > datetime.now()
        ).first()

2. 内联按钮触发会话创建 #

@router.callback_query(F.data == "trigger_dialog")
async def start_dialog(callback: CallbackQuery, db_session: AsyncSession, bot: Bot):
    # 从数据库获取预先存储的对方用户ID
    user_a = callback.from_user.id
    user_b = await get_opponent_id_from_db(db_session, user_a)
    
    if not user_b:
        await callback.answer("未找到对话对象")
        return

    # 创建会话
    await DialogSession.create(db_session, user_a, user_b)
    await callback.answer("对话已启动")
    await bot.send_message(user_b, "收到对话请求，可直接回复消息")

3. 消息处理器处理对话逻辑 #

@router.message()
async def handle_dialog_message(message: Message, db_session: AsyncSession, bot: Bot):
    # 检查当前用户是否在活跃会话中
    session = await DialogSession.get_active(db_session, message.from_user.id)
    if not session:
        # 非对话状态，执行其他业务逻辑
        return

    # 获取对方用户ID
    opponent_id = session.user2_id if session.user1_id == message.from_user.id else session.user1_id
    
    # 转发消息给对方
    await bot.send_message(opponent_id, f"对方消息：{message.text}")

    # 检查会话是否过期
    if session.expires_at < datetime.now():
        await message.answer("对话已超时")
        await bot.send_message(opponent_id, "对话已超时")
        await db_session.delete(session)
        await db_session.commit()

方案二：改进FSM状态管理（复用现有FSM） #

不为用户复制状态，而是给每个用户独立设置状态并存储对方ID，利用默认USER策略保证状态隔离。

1. 定义对话状态类 #

from aiogram.fsm.state import State, StatesGroup

class DialogStates(StatesGroup):
    in_conversation = State()  # 状态值存储对方用户ID

2. 按钮触发状态设置 #

import asyncio
from aiogram.fsm.context import FSMContext

@router.callback_query(F.data == "start_dialog")
async def trigger_dialog(callback: CallbackQuery, db_session: AsyncSession, bot: Bot):
    user_a = callback.from_user.id
    user_b = await get_opponent_id_from_db(db_session, user_a)
    
    if not user_b:
        await callback.answer("无对话对象")
        return

    # 同时为双方设置状态及对方ID
    await asyncio.gather(
        # 设置用户A的状态
        bot.get_fsm_context(user_a).set_state(DialogStates.in_conversation),
        bot.get_fsm_context(user_a).update_data(opponent_id=user_b),
        # 设置用户B的状态
        bot.get_fsm_context(user_b).set_state(DialogStates.in_conversation),
        bot.get_fsm_context(user_b).update_data(opponent_id=user_a)
    )

    await callback.answer("对话已开始")
    await bot.send_message(user_b, "可回复消息进行对话")

3. 对话消息处理器 #

@router.message(StateFilter(DialogStates.in_conversation))
async def handle_conversation(message: Message, fsm_context: FSMContext, bot: Bot, db_session: AsyncSession):
    data = await fsm_context.get_data()
    opponent_id = data.get("opponent_id")
    
    if not opponent_id:
        await message.answer("对话异常，请重新发起")
        await fsm_context.clear()
        return

    # 转发消息
    await bot.send_message(opponent_id, f"对方消息：{message.text}")

    # 检查对话是否超时
    if await is_dialog_expired(db_session, message.from_user.id, opponent_id):
        await message.answer("对话已超时")
        await bot.send_message(opponent_id, "对话已超时")
        # 清除双方状态
        await asyncio.gather(
            fsm_context.clear(),
            bot.get_fsm_context(opponent_id).clear()
        )

方案三：中间件注入对话关联信息 #

通过中间件在消息处理前自动查询对话状态，将对方ID注入处理器上下文，实现业务逻辑与状态管理解耦。

1. 实现对话中间件 #

from aiogram import BaseMiddleware
from aiogram.types import Update
from typing import Callable, Dict, Any, Awaitable

class DialogMiddleware(BaseMiddleware):
    async def __call__(
        self,
        handler: Callable[[Update, Dict[str, Any]], Awaitable[Any]],
        event: Update,
        data: Dict[str, Any]
    ) -> Any:
        if event.message:
            user_id = event.message.from_user.id
            db_session = data["db_session"]
            session = await DialogSession.get_active(db_session, user_id)
            
            if session:
                # 注入对方ID和会话对象
                opponent_id = session.user2_id if session.user1_id == user_id else session.user1_id
                data["opponent_id"] = opponent_id
                data["dialog_session"] = session
        
        return await handler(event, data)

2. 注册中间件并编写处理器 #

# 为消息路由注册中间件
router.message.middleware(DialogMiddleware())

@router.message()
async def handle_message(message: Message, opponent_id: int = None, bot: Bot):
    if not opponent_id:
        # 非对话逻辑处理
        return
    
    await bot.send_message(opponent_id, f"对方消息：{message.text}")

方案四：Redis直接读取FSM状态数据（适合已用Redis存储FSM的场景） #

如果你的FSM使用Redis作为存储，可以直接通过Redis键查询其他用户的状态数据，无需依赖Aiogram的FSM上下文方法。

示例代码 #

import redis.asyncio as redis

# 初始化Redis客户端
redis_client = redis.Redis(host="localhost", port=6379, db=0)

async def get_opponent_id(bot_id: int, user_id: int) -> int | None:
    """从Redis获取对方用户ID"""
    data_key = f"fsm:{bot_id}:user:{user_id}:data"
    data = await redis_client.hgetall(data_key)
    if data:
        return int(data.get(b"opponent_id"))
    return None

@router.message()
async def handle_dialog(message: Message, bot: Bot):
    opponent_id = await get_opponent_id(bot.id, message.from_user.id)
    if opponent_id:
        await bot.send_message(opponent_id, f"对方消息：{message.text}")

方案对比 #

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