• 后端
• 即时通讯
• API设计

过滤器（Filters）是 aiogram 路由器（Router）将 Telegram 更新（Update）路由到对应处理器（Handler）的核心机制。本文以官方文档《Filtering events》为主线，结合仓库源码逐一讲解 aiogram 内置的 Command、MagicData、CallbackData、ChatMemberUpdated、Exception 过滤器，并深入说明如何使用 Magic Filter（F）做声明式事件匹配、如何编写自己的过滤器，以及如何用“与 / 或 / 取反”三种方式组合过滤器。读完本文，你将能熟练地把任意 Telegram 事件精确、高效地路由到自己的处理器中。

过滤器与事件路由的基本原理

在 aiogram 中，过滤器解决的核心问题是：这条 Update 该交给哪个 Handler 处理？

官方文档明确指出两条基本规则：

• 过滤器的存在，是为了把更新路由到特定的处理器；
• 处理器的查找总是在“第一组过滤器全部通过的处理器”处停止，也就是说匹配到第一个满足条件的手处理器后，后续处理器不会再被检查；
• 默认情况下，所有处理器的过滤器集合为空，因此所有更新都会进入“第一个过滤器集合为空”的处理器——这也是为什么“兜底”处理器通常直接写成@router.message()不带任何过滤器。

从源码看，这条规则由 aiogram/dispatcher/event/handler.py 中的HandlerObject.check落实：

async def check(self, *args, **kwargs): if not self.filters: return True, kwargs for event_filter in self.filters: check = await event_filter.call(*args, **kwargs) if not check: return False, kwargs if isinstance(check, dict): kwargs.update(check) return True, kwargs

这里揭示了一个关键机制：过滤器不仅可以返回True/False，还可以返回dict。如果返回字典，其中的键值对会被合并到kwargs中，传递给后续过滤器以及最终的处理器——这正是Command过滤器把command对象、CallbackQueryFilter把callback_data对象注入处理器参数的实现基础。

在 aiogram/dispatcher/router.py 中，Router为 message、callback_query、chat_member 等几十种事件类型各维护一个TelegramEventObserver（事件观察者），处理器通过@router.<event_type>(<filters...>)装饰器注册，注册时传入的过滤器列表最终被包装为FilterObject并挂到HandlerObject.filters上。

内置过滤器一览

aiogram 提供了一批开箱即用的内置过滤器，全部从 aiogram/filters/init.py 导出：

此外，F（MagicFilter）是贯穿所有过滤场景的“瑞士军刀”，它与and_f、or_f、invert_f一起承担了组合表达式的职责。以下逐类详解。

Magic Filters：用F声明式匹配事件属性

Magic Filter 是 aiogram 核心团队维护的外部包magic-filter，随 aiogram 一起安装，也可以脱离 aiogram 单独从 PyPI 使用。注意：aiogram 对magic-filter有一个小型扩展（.as_()方法），因此推荐从aiogram导入F，而不是从magic_filter包导入。

在 aiogram/init.py 中可以看到：

F = MagicFilter()

MagicFilter对象是可调用的，它会“记住”你访问的属性链以及最终施加的操作，并在需要时用单个对象参数进行求值。例如先构造属性链F.foo.bar.baz，再叠加操作F.foo.bar.baz == 'spam'，最后通过(F.foo.bar.baz == 'spam').resolve(obj)对具体对象求值。

支持的魔法操作

官方文档（docs/dispatcher/filters/magic_filters.rst）列出了全部可用操作，并附等价 lambda 便于理解：

存在或不为 None（默认行为）

F.photo # 等价于 lambda message: message.photo

等于 / 不等于

F.text == 'hello' # lambda message: message.text == 'hello' F.from_user.id == 42 # lambda message: message.from_user.id == 42 F.text != 'spam' # lambda message: message.text != 'spam'

属于某个集合（in_方法或@运算符）

F.from_user.id.in_({42, 1000, 123123}) F.data.in_({'foo', 'bar', 'baz'})

包含子串

F.text.contains('foo') # lambda message: 'foo' in message.text

字符串前缀 / 后缀（仅适用于文本属性）

F.text.startswith('foo') # lambda message: message.text.startswith('foo') F.text.endswith('bar') # lambda message: message.text.startswith('bar')

正则匹配

F.text.regexp(r'Hello, .+') # lambda message: re.match(r'Hello, .+', message.text)

自定义函数（func）

F.chat.func(lambda chat: chat.id == -42)

取反（位取反~）——任意操作都可以被反转：

~F.text # lambda message: not message.text ~F.text.startswith('spam') # lambda message: not message.text.startswith('spam')

组合（按位与&/ 按位或|）

(F.from_user.id == 42) & (F.text == 'admin') F.text.startswith('a') | F.text.endswith('b') (F.from_user.id.in_({42, 777, 911})) & (F.text.startswith('!') | F.text.startswith('/')) & F.text.contains('ban')

字符串属性修饰（大小写、长度，仅限字符串属性）

F.text.lower() == 'test' # lambda message: message.text.lower() == 'test' F.text.upper().in_({'FOO', 'BAR'}) # lambda message: message.text.upper() in {'FOO', 'BAR'} F.text.len() == 5 # lambda message: len(message.text) == 5

把过滤结果作为处理器参数：.as_()

这是 magic-filter 本身不具备、由 aiogram 扩展的功能，扩展实现在 aiogram/utils/magic_filter.py：

class AsFilterResultOperation(BaseOperation): def resolve(self, value, initial_value): if value is None or (isinstance(value, Iterable) and not value): return None return {self.name: value} class MagicFilter(_MagicFilter): def as_(self, name: str): return self._extend(AsFilterResultOperation(name=name))

用法示例（官方文档原例）：

from aiogram import F @router.message(F.text.regexp(r"^(\d+)$").as_("digits")) async def any_digits_handler(message: Message, digits: Match[str]): await message.answer(html.quote(str(digits)))

as_("digits")会把正则匹配结果以digits为名注入处理器关键字参数。

在 aiogram 中的典型用法

@router.message(F.text == 'hello') @router.inline_query(F.data == 'button:1') @router.message(F.text.startswith('foo')) @router.message(F.content_type.in_({'text', 'sticker'})) @router.message(F.text.regexp(r'\d+'))

F几乎可以访问事件对象的任意属性链，覆盖绝大多数“检查某个事件属性”的场景。需要提醒的是：若直接使用magic_filter包提供的F而不用 aiogram 扩展，在 aiogram/dispatcher/event/handler.py 的FilterObject初始化中会触发一条Recommendation警告，提示改用from aiogram import F。

Command 过滤器：命令解析与匹配

Command过滤器专门处理带文本的命令消息，其实现位于 aiogram/filters/command.py。

五种用法

官方文档（docs/dispatcher/filters/command.rst）给出的用法：

1. 匹配单个命令：Command("start")
2. 按正则匹配：Command(re.compile(r"item_(\d+)"))
3. 同时匹配多个变体：Command("item", re.compile(r"item_(\d+)"))
4. 在公开群里处理发给其他 Bot 的命令：Command("command", ignore_mention=True)
5. 用BotCommand对象作为命令参考：Command(BotCommand(command="command", description="My awesome command"))

⚠️ 注意：命令名不能包含空格或任何空白字符。

构造参数与源码细节

从 aiogram/filters/command.py 的Command.__init__可以看到完整参数：

• prefix：命令前缀，默认"/"。前缀始终是单个字符，但这里可以传一组允许的前缀，例如"/!"表示同时支持/和!开头的命令；
• ignore_case：忽略大小写（对正则不生效，正则请使用 flags）；
• ignore_mention：忽略 Bot 提及。默认情况下 Bot 不能处理发给其他 Bot 的命令（带@其他bot提及会校验失败）；
• magic：在所有检查完成后，用 Magic Filter 进一步校验CommandObject；
• 至少需要指定一个命令（字符串、已编译正则或BotCommand），否则抛出ValueError。

测试用例（tests/test_filters/test_command.py）验证了这些行为，例如/test@tbot在ignore_mention=False时被拒绝、/test@tbot some args可以匹配、Command(commands=["test"], prefix="/!")能同时接受/test与!test等。

CommandObject：注入处理器的命令对象

过滤器通过后，一个 CommandObject 会以command为参数名传入处理器，其字段包括：

• prefix：命令前缀；
• command：不含前缀和提及的命令本体；
• mention：提及（若有）；
• args：命令参数（若存在，如/start foo中的foo）；
• regexp_match：若命令以正则形式声明，这里保存匹配结果；
• magic_result：若配置了magic校验，这里保存校验结果。

同时提供mentioned属性（是否有提及）和text属性（从对象重新生成原始文本）。

CommandStart：专门处理/start

CommandStart继承自Command（aiogram/filters/command.py），除命令解析外额外支持deep-link（深度链接）校验，参数为deep_link（True要求必须携带参数、False要求不得携带参数、None不校验）和deep_link_encoded（参数是否经 Base64 编码，配合aiogram.utils.deep_linking的解码逻辑使用）。

允许的事件类型

Command过滤器仅适用于message与edited_message两类更新。

MagicData 过滤器：匹配上下文数据

MagicData用于结合上下文数据（通常来自中间件）过滤事件。典型用法（官方文档原例）：

from aiogram.filters import MagicData @router.message(MagicData(F.event.from_user.id == F.config.admin_id)) async def handler(message: Message, config): ...

注意：这里F.config的config需要由中间件（Middleware）注入到处理流程的上下文中。

其实现非常简短（aiogram/filters/magic_data.py）：

async def __call__(self, event, *args, **kwargs): return self.magic_data.resolve( AttrDict({"event": event, **dict(enumerate(args)), **kwargs}), )

也就是说，MagicData会把事件对象、位置参数和关键字参数打包成一个AttrDict，其中事件以event为键，之后用 Magic Filter 对整包数据进行求值。正因如此，你可以在F.event.from_user.id与F.config.admin_id之间做跨数据源的比较。

MagicData支持的更新类型很广：message、edited_message、channel_post、edited_channel_post、inline_query、chosen_inline_result、callback_query、shipping_query、pre_checkout_query、poll、poll_answer、my_chat_member、chat_member、chat_join_request、error。

CallbackData：回调数据工厂与过滤

CallbackData是一个基于 Pydantic 的类工厂（aiogram/filters/callback_data.py），用于结构化生成和解析 callback_data 字符串，避免手写容易出错的"my:demo:42"式字符串。

定义与生成

from aiogram.filters.callback_data import CallbackData class MyCallback(CallbackData, prefix="my"): foo: str bar: int cb1 = MyCallback(foo="demo", bar=42) cb1.pack() # 返回 'my:demo:42' cb1.unpack('my:demo:42') # 返回 <MyCallback(foo="demo", bar=42)>

类关键字prefix是必填的，同时可通过sep指定分隔符（默认:）。源码中__init_subclass__会强制要求 prefix，并拒绝“分隔符出现在前缀中”的情况。pack()内部会做两件重要的事：

• 检查每个值编码后是否包含分隔符，若包含则抛ValueError；
• 检查最终字符串长度，Telegram Bot API 的 callback_data 上限为 64 字节（源码常量MAX_CALLBACK_LENGTH: int = 64），超长会抛ValueError。

放入内联键盘

在InlineKeyboardButton中必须显式pack()为字符串：

InlineKeyboardButton( text="demo", callback_data=MyCallback(foo="demo", bar="42").pack(), )

而如果使用InlineKeyboardBuilder（详见 docs/utils/keyboard.rst），可以直接传入 CallbackData 实例，builder 会自动完成打包：

builder = InlineKeyboardBuilder() builder.button(text="demo", callback_data=MyCallback(foo="demo", bar="42"))

用.filter()过滤回调

@router.callback_query(MyCallback.filter(F.foo == "demo")) async def my_callback_foo(query: CallbackQuery, callback_data: MyCallback): await query.answer(...) print("bar =", callback_data.bar)

MyCallback.filter(rule)生成CallbackQueryFilter，其实现（aiogram/filters/callback_data.py）逻辑为：先用callback_data.unpack(query.data)解析，解析失败（TypeError/ValueError）则过滤不通过；随后用可选规则rule对解析出的实例求值，通过后把callback_data注入处理器。

综合示例：管理动作

class Action(str, Enum): ban = "ban" kick = "kick" warn = "warn" class AdminAction(CallbackData, prefix="adm"): action: Action chat_id: int user_id: int # 处理器内构建键盘 builder = InlineKeyboardBuilder() for action in Action: builder.button( text=action.value.title(), callback_data=AdminAction(action=action, chat_id=chat_id, user_id=user_id), ) await bot.send_message( chat_id=admins_chat, text=f"What do you want to do with {html.quote(name)}", reply_markup=builder.as_markup(), ) # 过滤并按 action 分支处理 @router.callback_query(AdminAction.filter(F.action == Action.ban)) async def ban_user(query: CallbackQuery, callback_data: AdminAction, bot: Bot): await bot.ban_chat_member( chat_id=callback_data.chat_id, user_id=callback_data.user_id, ... )

已知限制：字段类型白名单

CallbackData的字段只允许以下类型（及其子类）：

• str
• int
• bool
• float
• Decimal（from decimal import Decimal）
• Fraction（from fractions import Fraction）
• UUID（from uuid import UUID）
• Enum（from enum import Enum，仅限字符串枚举）
• IntEnum（from enum import IntEnum，仅限整数枚举）

注意：整数枚举必须是IntEnum的子类，因为字符串形式的 callback_data 在解析回整数时存在歧义（见 docs/dispatcher/filters/callback_data.rst 的“Known limitations”）。

从源码 aiogram/filters/callback_data.py 的_encode_value可看到编码规则：None编码为空串、Enum取其.value、UUID取其十六进制、bool转为0/1，其余基础类型直接字符串化。

ChatMemberUpdated 过滤器：成员状态变化

ChatMemberUpdatedFilter用于处理群成员状态变化事件（aiogram/filters/chat_member_updated.py）。最基本也是最常用的两个场景：

from aiogram.filters import IS_MEMBER, IS_NOT_MEMBER, ChatMemberUpdatedFilter @router.chat_member(ChatMemberUpdatedFilter(IS_MEMBER >> IS_NOT_MEMBER)) async def on_user_leave(event: ChatMemberUpdated): ... @router.chat_member(ChatMemberUpdatedFilter(IS_NOT_MEMBER >> IS_MEMBER)) async def on_user_join(event: ChatMemberUpdated): ...

状态（Statuses）

状态还可以用+（is_member == True）或-（is_member == False）前缀扩展is_member标志，例如+RESTRICTED表示“处于受限状态且 is_member 为真”，-RESTRICTED反之。

状态组（Status Groups）

状态之间可以用位或|合并：

转换（Transitions）

转换用位移位运算符>>与<<定义：>>时旧状态在左、新状态在右（<<相反）。方向可以用~反转。预置的三个转换：

⚠️ 注意：如果你用|定义了状态组，在使用移位运算符之前必须加上括号，因为运算符优先级不同。例如必须写(MEMBER | RESTRICTED | LEFT | KICKED) >> ADMINISTRATOR。

底层实现要点

从源码看，状态被实现为_MemberStatusMarker，支持+/-（__pos__/__neg__）、|（__or__）、>>/<<（__rshift__/__lshift__）等运算符；ChatMemberUpdatedFilter.__call__会取出old_chat_member与new_chat_member：

• 若规则是状态/状态组，只对new成员做检查；
• 若规则是转换，则同时对old和new做检查（rule.check(old=old, new=new)）。

预置常量的定义同样在 aiogram/filters/chat_member_updated.py：

CREATOR = _MemberStatusMarker("creator") ADMINISTRATOR = _MemberStatusMarker("administrator") MEMBER = _MemberStatusMarker("member") RESTRICTED = _MemberStatusMarker("restricted") LEFT = _MemberStatusMarker("left") KICKED = _MemberStatusMarker("kicked") IS_MEMBER = CREATOR | ADMINISTRATOR | MEMBER | +RESTRICTED IS_ADMIN = CREATOR | ADMINISTRATOR IS_NOT_MEMBER = LEFT | KICKED | -RESTRICTED JOIN_TRANSITION = IS_NOT_MEMBER >> IS_MEMBER LEAVE_TRANSITION = ~JOIN_TRANSITION PROMOTED_TRANSITION = (MEMBER | RESTRICTED | LEFT | KICKED) >> ADMINISTRATOR

允许的事件类型

该过滤器适用于my_chat_member与chat_member两类更新。

Exception 过滤器：按类型或消息捕获错误

异常过滤器用于在错误处理器中按异常特征分流，官方文档称其“对处理文本消息中的错误很有帮助”（docs/dispatcher/filters/exception.rst）。

• ExceptionTypeFilter(*exceptions)：按异常类型匹配，传入一个或多个异常类，实现为isinstance(event.exception, self.exceptions)；
• ExceptionMessageFilter(pattern)：按异常消息正则匹配（字符串会被编译为正则），匹配通过时把re.Match结果以match_exception注入处理器。

from aiogram.filters import ExceptionTypeFilter, ExceptionMessageFilter @router.error(ExceptionTypeFilter(TelegramBadRequest)) async def handle_bad_request(error_event: ErrorEvent): ... @router.error(ExceptionMessageFilter(r"message is not modified")) async def handle_not_modified(error_event: ErrorEvent): ...

两个过滤器的实现都在 aiogram/filters/exception.py，分别检查ErrorEvent.exception的isinstance与re.match。允许的更新类型仅error（即router.error(...)/router.errors(...)）。

编写自己的过滤器

官方文档列出了合法的过滤器形态，它们都可以作为@router.<event>(...)的参数：

• 异步函数：async def my_filter(*args, **kwargs): pass
• 同步函数：def my_filter(*args, **kwargs): pass
• 匿名函数：lambda event: True
• 任意可 await 对象
• Filter 的子类
• MagicFilter 的实例

过滤器的返回值必须是bool或dict。若返回字典，其结果会传播给后续过滤器与处理器，作为关键字参数（这正是上文HandlerObject.check中kwargs.update(check)的行为）。

Filter 基类

自定义过滤器的推荐方式是继承 Filter。基类定义了两个关键成员：

• __call__(*args, **kwargs) -> bool | dict[str, Any]：抽象方法，接收事件并返回布尔值或字典，必须在子类中实现；
• update_handler_flags(flags)：可选的钩子方法，用于通过该过滤器向处理器扩展 flags（例如Command.update_handler_flags会把自身追加到flags["commands"]列表，供设置 bot commands 使用）；
• 此外基类实现了__invert__，因此自定义过滤器天然支持~MyFilter()取反（内部委托给invert_f）。

完整示例：文本过滤器

官方文档给出的示例（examples/own_filter.py）：

from aiogram import Router from aiogram.filters import Filter from aiogram.types import Message router = Router() class MyFilter(Filter): def __init__(self, my_text: str) -> None: self.my_text = my_text async def __call__(self, message: Message) -> bool: return message.text == self.my_text @router.message(MyFilter("hello")) async def my_handler(message: Message) -> None: ...

如果你想在过滤器中注入更多数据，只需让__call__返回dict，例如：

class TextLenFilter(Filter): async def __call__(self, message: Message) -> dict[str, int]: return {"text_len": len(message.text or "")} @router.message(TextLenFilter()) async def handler(message: Message, text_len: int): ...

组合过滤器：与 / 或 / 取反

官方文档将过滤器组合分为两种方式。

推荐方式：直接并列 + 重复注册 + 取反

多个过滤器并列 = “与”关系：

@<router>.message(F.text.startswith("show"), F.text.endswith("example"))

同一个处理器注册多次 = “或”关系（两种条件任一满足即可触发）：

@<router>.message(F.text == "hi") @<router>.message(CommandStart())

~取反：例如已有IsAdmin过滤器，想匹配“非管理员”：

@<router>.message(~IsAdmin())

另一种方式：and_f/or_f/invert_f

从aiogram.filters模块导入专用组合函数：

from aiogram.filters import and_f, or_f, invert_f and_f(F.text.startswith("show"), F.text.endswith("example")) or_f(F.text(text="hi"), CommandStart()) invert_f(IsAdmin()) and_f(<A>, or_f(<B>, <C>)) # 嵌套组合

组合的底层实现

这些函数定义在 aiogram/filters/logic.py：

• _AndFilter：顺序调用所有子过滤器，任一返回假值则整体失败；若有多个返回字典则合并，全部通过时返回合并后的字典（若无则返回True）；
• _OrFilter：顺序调用，命中第一个真值即返回该结果（若结果是字典则返回字典）；
• _InvertFilter：对目标结果取反（not bool(...)）。

~Filter()运算符（在 aiogram/filters/base.py 的__invert__中）最终也委托给invert_f。测试文件 tests/test_filters/test_logic.py 覆盖了这些组合的布尔/字典返回值语义，以及and_f/or_f/invert_f与~生成的过滤器类型。

过滤器工作机制小结

把全链路串起来，一次事件的分发大致如下（依据 aiogram/dispatcher/router.py 与 aiogram/dispatcher/event/handler.py）：

1. 更新到达后，Dispatcher按更新类型找到Router上对应的TelegramEventObserver；
2. Router.propagate_event先检查观察者的根过滤器，再调用observer.trigger，逐个尝试已注册的HandlerObject；
3. 每个HandlerObject.check按注册顺序调用其FilterObject列表：过滤器依次执行，全部通过（返回真值或字典）则该处理器被选中，字典内容合并进上下文；任一失败则尝试下一个处理器；
4. 过滤器可以返回dict为后续过滤器和处理器注入数据（如command、callback_data、match_exception）；
5. 若全部处理器都不匹配，事件被视为未处理（UNHANDLED）。

理解这一点后，你就能用“内置过滤器 +F表达式 + 自定义Filter子类 + 组合函数”自由搭建精确的事件路由策略。本文涉及的内置过滤器官方文档均可在 docs/dispatcher/filters/ 目录下找到完整说明，相关单元测试则集中在 tests/test_filters/ 中，可作为深入阅读与验证的起点。

• 后端
• 即时通讯
• API设计