disknext/CLAUDE.md

DiskNext Server 后端开发规范

设计哲学

本项目遵循以下核心设计原则，所有代码实现都应符合这些哲学思想：

1. 严格的类型安全与显式优于隐式

• 拒绝泛型（dict/list），必须具体类型（dict[str, str]）
• 拒绝 Any 类型，除非真的无法确定（需要 TODO 标记）
• Python 3.10+ 语法强制（int | None 而非 Optional[int]）
• 所有参数、返回值、变量都必须有类型注解
• 引用任何属性前必须先查看类实现确认存在

原因：编译时发现错误比运行时发现好，显式类型让 IDE 和人都能理解代码

2. 充血模型与运行时状态管理

• 充血模型：SQLModel实例同时包含数据库字段和业务方法（如Character.handle_message()）
• 运行时状态：使用 model_post_init()初始化运行时属性（以 _开头），不存储到数据库
• ClassVar管理跨实例状态：如MQTTManager._client、MQTTManager._message_task，用于全局单例或共享状态管理
• 严禁动态添加属性：所有实例属性必须在 model_post_init()或类定义中声明，避免SQLAlchemy冲突
• 会话参数化：运行时方法接收 session: AsyncSession参数，不存储session引用

原因：充血模型让业务逻辑内聚，所有相关行为都在实体类中。运行时状态通过 model_post_init()明确声明，保持类型安全

3. 单一真相来源（Single Source of Truth）

• 代码规范只有本文档（AGENTS.md），删除所有其他会造成混淆的规范文档
• SQLModelBase 定义了 model_config，子类不要重复定义
• ServerConfig 直接存属性（port），而非嵌套结构（server.port）
• 配置从数据库读取，而非 YAML 文件

原因：多个数据源会导致不一致，维护地狱。一个权威来源，其他地方引用它

4. 异步优先，IO绝不阻塞

• 所有 IO 操作必须是 async/await
• 禁止创建新的事件循环（使用 FastAPI 管理的）
• 同步库必须用 to_thread 或 Celery Worker 隔离
• 数据库、HTTP、文件操作都是异步（AsyncSession, aiohttp, aiofiles）

原因：现代 Python 服务器必须能处理高并发，阻塞 IO 是性能杀手

5. 组合优于继承，聚合优于散布

• 使用组合而非继承：Character组合LLM/TTS/RVA，而非继承它们
• Mixin模式用于横切关注点：如AioHttpClientSessionMixin添加HTTP客户端能力
• 联表继承仅在数据库多态时使用：OpenAICompatibleLLM → DouBaoLLM（数据库层面的多态）
• 相关逻辑聚合在实体类中：避免分散到多个辅助类，保持业务逻辑内聚

原因：继承是强耦合，组合给你灵活性。充血模型让相关行为聚合在一起，易于理解和维护

6. 目录结构即 API 结构

• URL 路径 /api/v1/ota 必须对应 root_endpoint/api/v1/ota/init.py
• 禁止独立 .py 文件，只允许 init.py 和子目录
• 每个目录的 init.py 负责定义该层级的路由

原因：代码组织和 URL 结构一致，降低认知负担。看 URL 就知道文件在哪

7. 错误快速失败，而非隐藏

• 禁止返回 None 来表示错误（会隐藏问题）
• 必须抛出异常，让 FastAPI 捕获并统一返回 500
• 端点不要自己处理异常（除非有特殊逻辑）

原因：隐藏的错误比显式的崩溃更可怕。让系统在问题发生时立即暴露

8. 模块内聚，跨模块松耦合

• 一个模块的文件合并（character.py 合并了 5 个文件）
• 通过 init.py 明确导出 API，隐藏内部实现
• 依赖注入而非全局变量（SessionDep, ServerConfigDep）

原因：相关代码应该在一起（减少跳转），但模块间通过接口通信（降低耦合）

9. 约定优于配置

• SQLModelBase 自动配置 use_attribute_docstrings
• TableBase 自动设置 table=True
• 字段用 docstring 描述，而非 Field(description=...)

原因：95% 的情况都用默认配置，特殊情况再覆盖。减少样板代码

10. 清晰的所有权和生命周期

• 传输层独立管理（MQTT连接、UDP套接字由transport模块管理，HTTP端点由root_endpoint管理）
• ClientDevice 拥有业务逻辑状态（输入输出队列、Character运行时实例）
• Character 在 model_post_init 中初始化运行时属性（会话历史、内存引用、消息队列）
• ClassVar 管理全局单例和共享状态（如 MQTTManager._client 存储全局MQTT连接实例、VerificationCode._verification_codes 存储TTL缓存）
• 使用 @asynccontextmanager 管理资源生命周期（如 Character.init()、ClientDevice.init()）
• 断开连接时明确清理所有相关资源（取消任务、关闭连接、保存状态）

原因：谁创建谁负责销毁。资源泄漏是难以调试的问题，必须有清晰的生命周期管理。传输层与业务层解耦，降低系统复杂度

11. 文档即代码

• Docstring 不是可选的，复杂逻辑必须写
• 类型注解本身就是文档
• 字段描述用 docstring（自动生成 API 文档）

原因：好的代码自己会说话，但复杂的逻辑需要解释。文档和代码在一起，不会过时

12. 渐进式重构，保留历史参考

• deprecated/ 目录保留旧代码
• REFACTOR_DOCUMENTATION.md 详细记录变更
• 数据库迁移用 Alembic，可回滚

原因：大重构不能一步到位。保留旧代码作为参考，记录设计决策，允许回退

---

代码规范

基础格式规范

• 所有的代码文件必须使用UTF-8编码
• 所有的代码文件必须使用4个空格缩进，不允许使用Tab缩进
• PR/commit时不要有任何语法错误（红线）
• 文件末尾必须有一个换行符
• 使用PyCharm默认的代码规范（变量命名，类命名，换行，空格，注释）（在默认情况下不要出现黄线，明显是linter的错误的除外）

类型注解规范

• 所有的类型注解都必须使用Python 3.10+的简化语法
  • 例如：使用 dict[str, Any] | None 而不是 Optional[Dict[str, Any]]
  • 用字符串表示可空的类型标注时，不能用 "TypeName" | None（这是语法错误），必须使用 'TypeName | None'（用单引号包裹整体类型）
• 使用内置类型而非typing模块：
  • 使用 type[ClassName] 而不是 Type[ClassName]
  • 使用 list[int] 而不是 List[int]
  • 使用 dict[str, Any] 而不是 Dict[str, Any]
  • 使用 tuple[int, str] 而不是 Tuple[int, str]
  • 使用 set[str] 而不是 Set[str]
  • Python 3.9+ 支持直接使用内置类型作为泛型，无需从typing导入
• 参数、类变量、实例变量等必须有类型注解，函数返回必须要注明类型
• 所有的类型注解都必须是明确的类型，不能使用 Any 或 object，除非确实无法确定类型，需要明确使用todo标注，以便后期研究类型
• 所有的类型注解都必须是具体的类型，不能使用泛型（如 List、Dict、Tuple、Set、Union 等），必须使用具体的类型（如 list[int]、dict[str, Any]、tuple[int, str]、set[str]、int | str 等）
• 所有的类型注解都必须是导入的类型，不能使用字符串表示类型（如 def func(param: 'CustomType') -> 'ReturnType':），除非是前向引用（即类型在当前作用域中还未定义）

异步编程规范

• 使用FastAPI管理的事件循环，不要再新建任何事件循环，不论是在任何线程或任何子进程中
• IO操作必须使用协程，不涉及任何CPU密集或IO的操作必须不使用协程，按需使用to_thread线程或Celery Worker
• 所有的数据库操作必须使用异步数据库驱动（如SQLModel的AsyncSession），不允许使用同步数据库驱动
• 所有的HTTP请求必须使用异步HTTP客户端（如aiohttp），不允许使用同步HTTP客户端
• 所有的文件操作必须使用异步文件操作库（如aiofiles），不允许使用同步文件操作
• 所有的子进程操作必须使用异步子进程库（如anyio），不允许使用同步子进程库
• 所有的第三方库调用必须使用异步版本，不允许使用同步版本，如果没有同步版本，视cpu负载情况使用to_thread线程或Celery Worker
• 所有的高cpu阻塞操作必须使用to_thread线程或Celery Worker，不允许在协程中直接调用高cpu阻塞操作

函数与参数规范

• 一个方法最多五个参数，多了考虑拆分方法或合并参数（SQLModel），不要简单的用tuple或dict

代码格式规范

• 容器类型定义：元组、字典、列表定义时，若定义只用了一行，则最后一个元素后面一律不加逗号，否则一律加逗号
• 括号换行：括号要么不换行，要么换行且用下面的形式写（一行最多一个变量，以逗号和换行分割）

示例代码

from loguru import logger as l

from api_client_models import (
    AgentModelsRequest,
    ReportRequest,
    SaveMemoryRequest,
    UserInfoResponse,
)

async def lookup_user_info(
        session: AsyncSession,
        user_id: int,
        short_name: str,
        data_1_with_a_long_name: dict[str, Any] | None,
        data_2_with_a_even_longer_name: CustomType
) -> UserInfoResponse:
    user = await User.get(session, User.id == user_id)
    new_dict = { user_id, short_name }
    l.debug(f"查到的数据: {new_dict}")
    result = UserInfoResponse(
        user.id,
        user.a_long_attribute,
        data_1_with_a_long_name,
        data_2_with_a_even_longer_name,
    )

    return result

文档与注释规范

• 复杂的类或函数（无法从名字推断明确的操作，如 handle_connection()）一律要写docstring，采用reStructuredText风格

字符串处理规范

• 引号使用：单引号 ' 用于给电脑看的文本（字典的键），双引号 " 用于给人看的文本（面向用户的提示，面向开发者的注释、log信息等）
• 字符串格式化：所有的字符串都用f-string格式化，不要使用 % 或 .format()
• 多行字符串：多行字符串使用"""或'''，"""给人看(如docstring)，'''给电脑看（如SQL语句或HTML内容）

命名规范

• 除非专有名词，代码中不要出现任何拼音变量名，所有变量名必须是英文
• 所有的变量名、函数名、方法名、参数名都必须使用蛇形命名法（snake_case）
• 所有的类名都必须使用帕斯卡命名法（PascalCase）
• 所有的常量名都必须使用全大写蛇形命名法（UPPER_SNAKE_CASE）
• 所有的私有变量、私有方法都必须使用单下划线前缀（_private_var）
• 所有的非常私有变量、非常私有方法都必须使用双下划线前缀（__very_private_var）
• 所有的布尔变量都必须使用is_、has_、can_、should_等前缀命名，且变量名必须是形容词或动词短语（如 is_valid, has_data, can_execute, should_retry）

异常处理规范

• 所有的异常都必须被捕获，且要有明确的处理逻辑
• 如果出现错误，不要return None，这样会造成隐藏的不易发现的错误，必须明确抛出异常

日志处理规范

• 所有的日志都必须用 from loguru import logger as l 处理，不要使用print
• 所有的日志都必须有明确的上下文，且要有明确的日志级别

框架使用规范

• 使用SQLModel，而不是Pydantic或SQLAlchemy。
• 使用SQLModel时禁止从future导入annotations
• 使用SQLModel时禁止从future导入annotations
• 使用SQLModel时禁止从future导入annotations
• 使用FastAPI，而不是Flask或Django
• 使用Aiohttp，而不是Requests
• 使用Aiofiles，而不是内置的open

import os as sync_os
from aiofiles import os, open
...
async with open('file.txt', 'r') as f:
    content = await f.read()
...
path = sync_os.path(...)

• 使用Anyio，而不是内置的subprocess
• 使用Loguru，而不是内置的logging：from loguru import logger as l
• 使用Celery，而不是内置的multiprocessing
• 使用GitHub Desktop，而不是直接在文件系统操作
• 使用PyCharm，而不是其他IDE

JSON处理规范

永远使用SQLModel/Pydantic的内置序列化方法，不要手动调用json库：

1. 模型序列化（推荐）

# ✅ 正确：使用 model_dump_json()
from sqlmodels.character.messages import TextEchoMessage

message = TextEchoMessage(text="你好", msg_id="123")
json_str: str = message.model_dump_json()  # 直接返回JSON字符串

# ✅ 正确：如果需要字典
message_dict: dict = message.model_dump()

2. 原始JSON处理（仅在必要时）

如果需要处理不是SQLModel/Pydantic的原始JSON数据：

# ✅ 正确：使用 orjson
import orjson

# 解析JSON
data: dict = orjson.loads(json_bytes)  # 接受 bytes 或 str

# 生成JSON
json_bytes: bytes = orjson.dumps(data)  # 返回 bytes
json_str: str = orjson.dumps(data).decode('utf-8')  # 需要str时decode

3. 错误示例

# ❌ 错误：使用标准库json
import json
json_str = json.dumps(message.model_dump())

# ❌ 错误：手动调用orjson序列化模型
import orjson
json_str = orjson.dumps(message.model_dump()).decode('utf-8')

# ✅ 正确：直接用模型方法
json_str = message.model_dump_json()

4. 消息序列化示例

# ✅ 正确：使用模型内置序列化
json_str: str = message.model_dump_json()
# 通过MQTT或HTTP发送
await mqtt_client.publish(topic, json_str)

# ❌ 错误：手动序列化
json_str = orjson.dumps(message.model_dump()).decode('utf-8')

配置说明：

• SQLModel/Pydantic v2 内部可配置使用 orjson 作为序列化器
• model_dump_json() 比手动序列化更快且类型安全
• 标准库 json 已被项目全面禁用

安全规范

• 禁止硬编码敏感信息：API密钥、数据库密码、JWT密钥等必须使用环境变量或安全配置管理
• 使用环境变量：通过 os.getenv() 或配置文件读取敏感信息
• SQL注入防护：使用SQLModel/SQLAlchemy的参数化查询，禁止字符串拼接SQL
• JWT认证：WebSocket和敏感端点必须验证JWT token
• CORS配置：生产环境必须配置正确的CORS策略

错误示例：

# ❌ 严重错误：硬编码API密钥
api_key: str = "bce-v3/ALTAK-xcRfCL5sLloSNKpU3z9xX/..."

正确示例：

# ✅ 正确：从环境变量读取
import os
api_key: str = os.getenv("BAIDU_API_KEY")
if not api_key:
    raise ValueError("BAIDU_API_KEY environment variable not set")

AI编码规范

• 如果有条件，inline completion插件使用GitHub Copilot，而不是JetBrains自带的
• 如果让AI直接编码，使用Gemini 2.5 Pro及以上, Claude 3.7 Sonnet Thinking及以上，而不是GPT系列模型，DeepSeek，豆包，文心一言等
• 使用AI生成代码时，提示词必须带上这个代码规范
• 在实现任何功能前，必须先看看有没有现成的解决方案，比如pypi包，不要重复造轮子

SQLModel规范

• 使用字段后面的"""..."""（docstring）而不是参数description="..."来写字段描述

示例：

class User(SQLModel, table=True):
    model_config = ConfigDict(use_attribute_docstrings=True)

    id: int = Field(default=None, primary_key=True, description="用户ID")  # 错误示范
    name: str = Field(description="用户名")  # 错误示范
    email: str = Field(unique=True)  # 正确示范
    """用户邮箱"""

• Field使用原则：
  • 只有default参数时：直接赋值，不使用Field
  • 有其他参数时（如ge, le, foreign_key, unique等）：必须使用Field

# 错误示范
class CharacterConfig(SQLModelBase):
    vad_frame_ms: int = Field(default=20)  # 只有default，不应该用Field
    """VAD帧大小（毫秒）"""

    vad_aggressiveness: int = Field(default=2, ge=0, le=3)  # 正确，有ge/le约束
    """VAD激进程度（0-3）"""

# 正确示范
class CharacterConfig(SQLModelBase):
    vad_frame_ms: int = 20  # 只有default，直接赋值
    """VAD帧大小（毫秒）"""

    vad_aggressiveness: int = Field(default=2, ge=0, le=3)  # 有约束，使用Field
    """VAD激进程度（0-3）"""

    llm_id: UUID = Field(..., foreign_key='llm.id')  # 有foreign_key，使用Field
    """LLM配置ID"""

• 不要重复定义model_config：SQLModelBase已经定义了 model_config = ConfigDict(use_attribute_docstrings=True, validate_by_name=True)，所有继承自SQLModelBase的类都会自动继承这个配置，不需要重复定义

# 错误示范
class OTARequest(SQLModelBase):
    model_config = ConfigDict(use_attribute_docstrings=True)  # 不必要的重复

    application: OTAApplicationInfo
    """应用信息"""

# 正确示范
class OTARequest(SQLModelBase):
    application: OTAApplicationInfo
    """应用信息"""

• 请一律使用TableBase系列类提供的crud，不要试图直接操作session

• 所有的SQLModel类都必须继承自SQLModelBase或其子类，请不要直接继承SQLModel

• 所有的存数据库的类必须继承自TableBase的子类，并且table=True

• 所有的不存数据库的类必须继承自SQLModelBase

• 定义枚举类请使用 StrEnum 而不是 (str, Enum)

• TableBase CRUD方法返回值规范：

  • 必须使用返回值：save() 和 update() 方法调用后，session中的所有对象都会过期（expired），必须使用返回值，而不是继续使用原对象

# ✅ 正确：使用返回值
device = ClientDevice(...)
device = await device.save(session)
return device

# ❌ 错误：不使用返回值（device对象已过期）
device = ClientDevice(...)
await device.save(session)
return device  # 此时device已过期，访问属性会报错

# ✅ 正确：update也要使用返回值
device = await device.update(session, update_data)
return device

# ❌ 错误：不使用返回值
await device.update(session, update_data)
return device  # 此时device已过期

原因：save() 和 update() 内部调用了 session.commit()，commit后SQLAlchemy会让所有对象过期以确保数据一致性。这些方法内部会调用 session.refresh(self) 来刷新对象，并返回刷新后的实例。

队列驱动架构规范

使用场景：需要解耦生产者和消费者时使用 asyncio.Queue

设计原则：

• 队列在 model_post_init() 中初始化（运行时属性）
• 提供清晰的 put/get 接口供外部调用
• 后台任务持续处理队列（使用循环）
• 使用 asyncio.CancelledError 优雅退出

示例 - Character消息处理：

import asyncio
from sqlmodel import Field
from sqlmodel_ext import SQLModelBase, UUIDTableBaseMixin

class CharacterBase(SQLModelBase):
    name: str
    """角色名称"""

class Character(CharacterBase, UUIDTableBaseMixin):
    """充血模型：包含数据和业务逻辑"""

    # ==================== 运行时属性（在model_post_init初始化） ====================
    _input_queue: asyncio.Queue
    """消息输入队列（由ClientDevice写入）"""

    _output_queue: asyncio.Queue
    """消息输出队列（由ClientDevice读取）"""

    _processor_task: asyncio.Task | None
    """后台处理任务"""

    def model_post_init(self, __context) -> None:
        """初始化运行时属性"""
        self._input_queue = asyncio.Queue()
        self._output_queue = asyncio.Queue()
        self._processor_task = None

    # ==================== 公共接口 ====================
    async def put_input_message(self, msg: str | bytes) -> None:
        """写入输入队列（由ClientDevice调用）"""
        await self._input_queue.put(msg)

    async def get_output_message(self) -> str | bytes:
        """读取输出队列（由ClientDevice调用）"""
        return await self._output_queue.get()

    # ==================== 后台任务 ====================
    async def _message_processor_loop(self, session: AsyncSession) -> None:
        """后台任务：持续处理输入队列"""
        try:
            while True:
                msg = await self._input_queue.get()

                # 处理消息
                response = await self._process_message(msg, session)

                # 输出结果
                await self._output_queue.put(response)
        except asyncio.CancelledError:
            # 优雅退出
            pass
        finally:
            # 清理资源
            pass

    async def _process_message(self, msg: str | bytes, session: AsyncSession) -> str:
        """处理单条消息"""
        # 业务逻辑
        return f"处理结果: {msg}"

    # ==================== 生命周期管理 ====================
    @asynccontextmanager
    async def init(self, session: AsyncSession):
        """初始化并启动后台任务"""
        # 启动后台任务
        self._processor_task = asyncio.create_task(
            self._message_processor_loop(session)
        )

        try:
            yield
        finally:
            # 取消后台任务
            if self._processor_task:
                self._processor_task.cancel()
                try:
                    await self._processor_task
                except asyncio.CancelledError:
                    pass

使用示例：

# 在ClientDevice中使用
async with character.init(session):
    # 发送消息
    await character.put_input_message("你好")

    # 接收响应
    response = await character.get_output_message()
    # 通过MQTT或UDP发送（具体取决于消息类型）
    if isinstance(response, bytes):
        await udp_client.send(response)  # 音频数据
    else:
        await mqtt_client.publish(topic, response)  # 控制消息

关键点：

• ✅ 队列明确标记为 _ 开头（私有属性）
• ✅ 提供公共的 put/get 方法（封装队列操作）
• ✅ 后台任务使用 asyncio.create_task() 启动
• ✅ 使用 @asynccontextmanager 确保资源清理
• ✅ 捕获 CancelledError 优雅退出

联表继承（Joined Table Inheritance）规范

当需要实现多态的数据库模型时（如ASR、TTS、Tool等），使用联表继承模式。项目提供了通用工具简化实现：

基本结构：

1. Base类：只包含字段，不继承TableBase（无表）
2. 抽象父类：继承Base + UUIDTableBase + ABC，有自己的表
3. SubclassIdMixin：为子类提供外键指向父表的主键
4. AutoPolymorphicIdentityMixin：自动生成polymorphic_identity

完整示例：

from abc import ABC, abstractmethod
from uuid import UUID
from sqlmodel import Field
from sqlmodel_ext import (
    SQLModelBase,
    UUIDTableBaseMixin,
    create_subclass_id_mixin,
    AutoPolymorphicIdentityMixin,
)

# 1. 定义Base类（只有字段，无表）
class ASRBase(SQLModelBase):
    name: str
    """配置名称"""

    base_url: str
    """服务地址"""

    language: str = "zh"
    """语言代码：'zh' | 'en' | 'auto'"""

# 2. 定义抽象父类（有表）
class ASR(
    ASRBase,
    UUIDTableBaseMixin,
    ABC,
    polymorphic_on='__polymorphic_name',
    polymorphic_abstract=True
):
    """ASR配置的抽象基类"""

    __polymorphic_name: str
    """多态类型鉴别器，用于区分不同的ASR子类"""

    @abstractmethod
    async def transcribe(self, pcm_data: bytes) -> str:
        """转录音频为文本"""
        pass

# 3. 为第二层子类创建ID Mixin
ASRSubclassIdMixin = create_subclass_id_mixin('asr')

# 4. 创建第二层抽象类（如果需要多级继承）
class FunASR(
    ASRSubclassIdMixin,
    ASR,
    AutoPolymorphicIdentityMixin,
    polymorphic_abstract=True
):
    """FunASR的抽象基类"""
    # polymorphic_identity 会自动设置为 'funasr'

    model_size: str = "medium"
    """模型大小"""

# 5. 创建具体实现类
class FunASRLocal(FunASR, table=True):
    """FunASR本地部署版本"""
    # polymorphic_identity 会自动设置为 'funasr.funasrlocal'

    model_path: str
    """本地模型路径"""

    async def transcribe(self, pcm_data: bytes) -> str:
        # 具体实现
        pass

class FunASRCloud(FunASR, table=True):
    """FunASR云端版本"""
    # polymorphic_identity 会自动设置为 'funasr.funasrcloud'

    api_key: str
    """云端API密钥"""

    async def transcribe(self, pcm_data: bytes) -> str:
        # 具体实现
        pass

工具说明：

1. create_subclass_id_mixin(parent_table_name: str)

   • 动态创建SubclassIdMixin类
   • 提供 id: UUID = Field(foreign_key='parent_table.id', primary_key=True)
   • 必须放在继承列表的第一位，确保通过MRO覆盖UUIDTableBase的id

2. AutoPolymorphicIdentityMixin

   • 自动根据类名生成polymorphic_identity
   • 格式：{parent_identity}.{classname_lowercase}
   • 如果没有父类identity，直接使用类名小写

注意事项：

• SubclassIdMixin必须放在继承列表第一位（MRO优先级）
• AutoPolymorphicIdentityMixin放在靠后位置（在ABC之前）
• 如果需要手动指定polymorphic_identity，可以在类定义时传入参数：

  class CustomASR(ASR, table=True, polymorphic_identity='my_custom_asr'):
      pass
  

---

__init__.py 模块组织规范

__init__.py 是Python包的入口文件，本项目根据不同场景使用不同的组织模式。

模式一：模块导出模式（SQLModel/业务模块）

用途：从子模块收集和重新导出公共API

特征：

• 使用相对导入 from .xxx import ...
• 平铺导出所有需要外部访问的类/函数/常量
• 可选：使用分组注释说明导出内容的类型
• 可选但推荐：添加模块级 docstring 说明架构和历史变更

基础示例：

# sqlmodels/user/__init__.py
from .user import (
    User,
    UserSourceMethodEnum,
    UserInfoResponse,
    UserLoginOrRegisterRequest,
    UserJWTPayload,
)
from .speaker_info import (
    SpeakerInfo,
    SpeakerInfoBase,
    SpeakerData,
    SpeakerIdentificationResult,
    SexEnum,
    SpeakerSourceTypeEnum,
    SpeakerInfoExtractionException,
)

分组注释示例（当导出内容较多时）：

# sqlmodels/character/__init__.py
"""Character模块 - 统一导出"""

from .character import (
    # 数据库模型
    Character,
    UserCharacterLinkModel,
    CharacterBase,

    # 消息和常量
    Messages,
    DEFAULT_SUMMARY_PROMPT,
)

from .memories.text.text import TextMemory
from .messages import CharacterOutputMessageBase, TextEchoMessage, AudioPlaybackEndMessage

文档化导出示例（⭐ 可选但推荐）：

注意：文档化导出并不是必须的，对于任何包都是可选但推荐的。当模块架构复杂或有重要历史变更时，建议使用此模式。

# sqlmodels/character/asr/__init__.py
"""
ASR（Automatic Speech Recognition）配置模块

使用联表继承（Joined Table Inheritance）实现多态ASR配置。

架构：
    ASR (抽象基类，有表)
    └── WebSocketASR (WebSocket连接到外部ASR Service) [推荐]

历史：
    - FunASR, ExperimentalASR 已移除（使用HTTP API，已被WebSocketASR替代）
"""
from .base import ASR, ASRBase, ASRException
from .websocket_asr import WebSocketASR, WebSocketASRBase

# WebSocket消息定义
from .websocket_messages import (
    MessageTypeEnum,
    ErrorMessage,
    EndOfStreamMessage,
    TranscriptionInfo,
    TranscriptionResponse,
)

适用场景：

• sqlmodels/ 下的所有子包
• utils/ 等工具模块
• 所有业务逻辑模块

推荐添加文档化的场景：

• 使用联表继承的配置模块（ASR, TTS, LLM, Tool）
• 有历史演进需要记录的模块
• 架构复杂需要说明的模块

---

模式二：FastAPI路由聚合模式（路由中间节点）

用途：构建层级化的API路由结构

特征：

• 从 fastapi 导入 APIRouter
• 创建带 prefix 的 router
• 通过 router.include_router() 包含子路由
• 导出 router 供上层包含

示例：

# root_endpoint/__init__.py
from fastapi import APIRouter
from .api import router as api_router

router = APIRouter()
router.include_router(api_router)

# root_endpoint/api/__init__.py
from fastapi import APIRouter
from .v1 import router as v1_router

router = APIRouter(prefix="/api")
router.include_router(v1_router)

# root_endpoint/api/v1/__init__.py
from fastapi import APIRouter
from .admin import router as admin_router
from .character import router as character_router
from .device import router as device_router
from .llm import router as llm_router
from .ota import router as ota_router
from .rva import router as rva_router
from .tts import router as tts_router
from .user import router as user_router
from .verification_code import router as verification_code_router

router = APIRouter(prefix="/v1")
router.include_router(admin_router)
router.include_router(character_router)
router.include_router(device_router)
router.include_router(llm_router)
router.include_router(ota_router)
router.include_router(rva_router)
router.include_router(tts_router)
router.include_router(user_router)
router.include_router(verification_code_router)

路由层次：

root_endpoint/           → /
└── api/                 → /api
    └── v1/              → /api/v1
        ├── admin/       → /api/v1/admin
        ├── character/   → /api/v1/character
        ├── device/      → /api/v1/device
        ├── llm/         → /api/v1/llm
        ├── ota/         → /api/v1/ota
        ├── rva/         → /api/v1/rva
        ├── tts/         → /api/v1/tts
        ├── user/        → /api/v1/user
        └── verification_code/  → /api/v1/verification_code

适用场景：

• root_endpoint/ 及其所有非叶子子目录
• 所有需要路由聚合的中间层节点

---

模式三：FastAPI端点实现模式（路由叶子节点）

用途：在 __init__.py 中直接定义端点处理函数

特征：

• 创建带 prefix 的 APIRouter
• 导入依赖项和数据模型
• 使用 @router.post/get/websocket 装饰器定义端点
• 包含详细的 docstring 说明端点行为

示例：

# root_endpoint/api/v1/user/__init__.py
from fastapi import APIRouter
from loguru import logger as l

from dependencies import SessionDep, CurrentActiveUserDep, ServerConfigDep
from sqlmodels.user import UserLoginOrRegisterRequest, UserSourceMethodEnum
from sqlmodels import User

router = APIRouter(prefix="/user")

@router.post('')
async def register_or_login(
        session: SessionDep,
        user: CurrentActiveUserDep,
        server_config: ServerConfigDep,
        request: UserLoginOrRegisterRequest
) -> ...:
    """
    用户注册或登录端点

    如果用户不存在则注册，存在则登录。
    """
    source: UserSourceMethodEnum = request.source

    if server_config.require_captcha:
        # 验证码逻辑
        pass

    # 生成并返回令牌
    pass

FastStream MQTT订阅器示例：

# stream/subscribers/device_input.py
"""
设备输入订阅器

订阅Stream: devices:input
功能：验证设备输入并转发到角色处理Stream
"""
from typing import Any
from uuid import UUID

from faststream import FastStream
from loguru import logger as l
from sqlmodel.ext.asyncio.session import AsyncSession

from dependencies import SessionDep
from sqlmodels.character import Character
from sqlmodels.client_device import ClientDevice
from sqlmodels.database_connection import DatabaseManager

from ..app import CHARACTERS_PROCESSING_STREAM, DEVICES_INPUT_STREAM, broker


@broker.subscriber(DEVICES_INPUT_STREAM, group="device-input-processor")
async def handle_device_input(msg: dict[str, Any]) -> None:
    """
    处理设备输入消息并转发到角色处理Stream

    流程：
    1. 验证消息格式（device_id, character_id必填）
    2. PostgreSQL-first验证ClientDevice和Character存在性
    3. 转发到characters:processing Stream

    消息格式：
    - 上行（MQTT/UDP）: {"device_id": "uuid", "character_id": "uuid", "type": "audio"|"text", "data": ...}
    - 下行（FastStream）: 转发到characters:processing Stream

    认证：
    - MQTT: 通过Hello握手建立会话
    - UDP: 通过SessionInfo验证序列号

    错误处理：
    - 验证失败: 记录日志并清理Redis状态
    - 设备不存在: 拒绝处理
    - 角色不存在: 拒绝处理
    """
    try:
        # 验证消息格式
        device_id_str: str | None = msg.get('device_id')
        character_id_str: str | None = msg.get('character_id')

        if not device_id_str or not character_id_str:
            l.warning(f"消息格式无效: {msg}")
            return

        device_id = UUID(device_id_str)
        character_id = UUID(character_id_str)

        # 验证设备和角色存在
        async with DatabaseManager.session() as session:
            device = await ClientDevice.get(
                session,
                ClientDevice.id == device_id,
                fetch_mode='one_or_none'
            )
            if not device:
                l.warning(f"设备不存在: {device_id}")
                return

            character = await Character.get(
                session,
                Character.id == character_id,
                fetch_mode='one_or_none'
            )
            if not character:
                l.warning(f"角色不存在: {character_id}")
                return

        # 转发到处理Stream
        await broker.publish(
            msg,
            stream=CHARACTERS_PROCESSING_STREAM
        )
        l.debug(f"已转发消息到处理队列: device={device_id}, character={character_id}")

    except ValueError as e:
        l.error(f"消息解析错误: {e}")
    except Exception as e:
        l.exception(f"处理设备输入异常: {e}")

适用场景：

• root_endpoint/api/v1/ 下的所有叶子端点目录
• 所有需要定义具体端点的目录

---

命名约定

• 相对导入：一律使用 from .xxx import ...，不使用 from package.xxx import ...
• 子路由命名：统一格式 from .子目录 import router as 子目录_router
• 类型导出：导出类的同时导出相关的枚举、异常、DTO、常量

示例：

# ✅ 正确
from .base import Tool, ToolResponse, ToolTypeEnum, ToolException

# ❌ 错误
from sqlmodels.character.llm.openai_compatibles.tools.base import Tool

---

导出粒度原则

应该导出的内容：

• ✅ 需要被外部模块使用的类、函数、常量
• ✅ 数据库模型及其配套的枚举、异常、DTO
• ✅ 抽象基类和具体实现类
• ✅ 公共工具函数和常量

不应该导出的内容：

• ❌ 以 _ 开头的私有对象
• ❌ 只在包内部使用的辅助函数
• ❌ 实现细节（如内部使用的临时类）

关于 __all__ 的规定：

禁止使用 __all__。__init__.py 应该只负责显式导出必要的公共内容，私有内容（以 _ 开头）不允许在 __init__.py 中导出。通过显式的 from .xxx import ... 语句已经清楚表明了导出意图，使用 __all__ 会造成冗余和维护负担。

错误示例：

# ❌ 错误：使用 __all__
from .base import Tool, _validate_tool_name

__all__ = ['Tool']  # 不要使用 __all__

正确示例：

# ✅ 正确：只导出公共API，不使用 __all__
from .tool import Tool, ToolResponse, ToolTypeEnum
from .function import Function, FunctionParam
from .exceptions import ToolException

# _validate_tool_name() 是私有函数，不在这里导入，自然不会导出

原因：

1. 显式导入已经明确了导出意图，__all__ 是冗余的
2. 私有内容（_xxx）不应该出现在 __init__.py 中
3. 减少维护成本（不需要同时维护导入列表和 __all__ 列表）
4. 符合"显式优于隐式"的设计哲学

---

FastAPI路由结构规则

层级规则：

每层 __init__.py 的职责：

1. 创建本层 APIRouter（设置 prefix）
2. 包含所有直接子路由
3. 导出 router 供上层包含

完整示例：

# 第一层：root_endpoint/__init__.py
from fastapi import APIRouter
from .api import router as api_router

router = APIRouter()
router.include_router(api_router)

# 第二层：root_endpoint/api/__init__.py
from fastapi import APIRouter
from .v1 import router as v1_router

router = APIRouter(prefix="/api")
router.include_router(v1_router)

# 第三层：root_endpoint/api/v1/__init__.py
from fastapi import APIRouter
from .user import router as user_router
from .character import router as character_router

router = APIRouter(prefix="/v1")
router.include_router(user_router)
router.include_router(character_router)

# 第四层：root_endpoint/api/v1/user/__init__.py
from fastapi import APIRouter

router = APIRouter(prefix="/user")

@router.post('/login')
async def login(...):
    pass

URL与目录对应关系：

URL路径	文件路径	prefix设置
/	root_endpoint/__init__.py	无
/api	root_endpoint/api/__init__.py	prefix="/api"
/api/v1	root_endpoint/api/v1/__init__.py	prefix="/v1"
/api/v1/user	root_endpoint/api/v1/user/__init__.py	prefix="/user"
/api/v1/user/login	端点装饰器 @router.post('/login')	-

关键原则：

• ✅ URL结构 = 目录结构（设计哲学第6条）
• ✅ prefix只写当前层的路径部分，不包含父路径
• ✅ 中间节点负责聚合，叶子节点负责实现
• ❌ 禁止在 root_endpoint/ 下创建独立的 .py 文件（只允许 __init__.py 和子目录）

---

导入顺序规范（符合PEP 8和Black）

所有 __init__.py 文件必须遵循以下导入顺序：

# 1. 标准库导入
from typing import Any
from uuid import UUID

# 2. 第三方库导入
from fastapi import APIRouter
from loguru import logger as l
from sqlmodel import Field

# 3. 本地应用导入（从项目根目录的包开始）
from dependencies import SessionDep
from sqlmodels.user import User
from sqlmodel_ext import UUIDTableBaseMixin

# 4. 相对导入（同包内的模块）
from .base import BaseClass
from .submodule import SubClass

分组规则：

• 每组之间用一个空行分隔
• 同组内按字母顺序排序（from 语句按模块名排序）
• 相对导入永远放在最后

错误示例：

# ❌ 混乱的导入顺序
from .base import BaseClass
from fastapi import APIRouter
from sqlmodels.user import User
from typing import Any
from .submodule import SubClass

正确示例：

# ✅ 清晰的导入顺序
from typing import Any

from fastapi import APIRouter

from sqlmodels.user import User

from .base import BaseClass
from .submodule import SubClass

---

特殊注意事项

1. 所有Python包目录都必须有 __init__.py

   • 即使是空文件也要创建
   • 空文件可以只包含一个 pass 或留空
   • 确保包可以被正确导入

2. 避免循环导入

   • 顶层 __init__.py（如 sqlmodels/__init__.py）可以导出常用的顶层类
   • 子模块的 __init__.py 只导出本包内容
   • 禁止向上导入（from ..parent import）

3. 类型提示的导出

   • 如果导出泛型类型变量（如 T），需要一并导出
   • 例如：from .table_base import TableBase, UUIDTableBase, T

4. FastAPI端点的docstring必须包含

   • 端点功能描述
   • 认证方式说明
   • 请求/响应格式规范
   • 异常处理说明

示例：

@router.post("/character")
async def create_character(
        session: SessionDep,
        current_user: CurrentActiveUserDep,
        character_data: CharacterBase
):
    """
    创建角色端点

    功能：创建新的Character配置

    认证：
    - JWT token in Authorization header
    - 验证用户存在且激活

    请求体：
    - CharacterBase模型（JSON格式）

    响应：
    - 完整的Character对象（包含ID）

    错误处理：
    - HTTPException 400: 请求数据无效
    - HTTPException 401: 未授权
    - HTTPException 500: 服务器错误
    """
    pass

---

使用模式决策树

如何选择使用哪个模式？

是 __init__.py 文件吗？
├─ 否 → 不适用本规范
└─ 是
   ├─ 是 root_endpoint/ 下的文件吗？
   │  ├─ 是
   │  │  ├─ 包含子路由吗（有 router.include_router()）？
   │  │  │  ├─ 是 → 模式二：FastAPI路由聚合
   │  │  │  └─ 否 → 模式三：FastAPI端点实现
   │  │  └─ ...
   │  └─ 否 → 模式一：模块导出（考虑添加文档化说明）
   └─ ...

快速参考表：

目录类型	使用模式	核心特征
SQLModel业务包	模式一	from .xxx import ... 平铺导出
配置类子包	模式一（建议文档化）	包含架构docstring + 导出
FastAPI中间节点	模式二	router.include_router() 聚合
FastAPI叶子节点	模式三	@router.post/websocket 实现

---

注意：此规范会持续更新，对此文件有任何建议修改可以发起PR，没有在规范里提到的都没有硬性要求，可以参考PEP 8