autowzry-agent

QMobaAgent 项目架构文档

用途: Agent快速了解项目结构，节省token 更新日期: 2025-11-24 重要变更: 项目已重构为标准Python包结构

---

1. 项目结构树

autowzry-agent/
├── qmoba_agent/         # 主包目录
│   ├── __init__.py      # 包初始化（导出常用类）
│   ├── config/          # 配置管理
│   │   ├── __init__.py
│   │   └── config.py    # 配置类定义
│   ├── core/            # 核心模块
│   │   ├── __init__.py
│   │   ├── lightweight_dqn.py  # 轻量级DQN（88K参数）
│   │   └── trainer.py   # DQN训练器
│   ├── data/            # 数据处理
│   │   ├── __init__.py
│   │   ├── data_manager.py  # 数据管理（收集+标记）
│   │   └── training_buffer.py  # 训练数据加载
│   ├── environment/     # 环境交互
│   │   ├── __init__.py
│   │   ├── action_space.py # 动作空间定义
│   │   ├── compatibility.py # 兼容层（依赖注入中心）
│   │   └── game_state.py   # 游戏状态检测
│   ├── utils/           # 工具函数
│   │   ├── __init__.py
│   │   ├── image_processing.py  # 图像预处理
│   │   └── video_reader.py  # 视频读取
│   └── autowzry_lite/   # autowzry精简版
│       └── autowzry/
│           ├── __init__.py
│           ├── version.py
│           └── wzry.py
├── scripts/             # 执行脚本
│   ├── __init__.py
│   ├── collect_from_video.py   # 从视频收集数据
│   ├── collect_from_device.py  # 从设备收集数据
│   ├── label_data.py    # 标记数据（action+state）
│   ├── train.py         # 训练模型
│   ├── battle.py        # 模型测试与对战（支持离线/在线）
│   ├── check_buffer.py  # Buffer数据验证（调试用）
│   ├── extract_data.py  # HDF5帧范围提取
│   └── test_pipeline.py # 端到端自动化测试（开发用）
├── config/              # 配置文件目录（YAML文件）
│   └── agent.config.yaml  # 主配置文件
├── docs/                # 文档
│   ├── guides/          # 使用指南
│   │   ├── README_FOR_AGENT.md  # Agent快速上手
│   │   ├── quickstart.md        # 用户快速开始
│   │   └── AGENT_COLLABORATION_RULES.md  # Agent协作规范
│   ├── design/          # 架构设计
│   │   └── ARCHITECTURE.md      # 本文件
│   └── logs/            # 开发日志
│       └── development_log.md
├── workspace/           # 运行时数据（.gitignore）
│   ├── episodes/        # HDF5数据存储
│   ├── checkpoints/     # 模型检查点
│   ├── videos/          # 视频素材
│   ├── buffer_check/    # Buffer检查输出
│   └── test_pipeline/   # 端到端测试输出
├── pyproject.toml       # 项目配置（包元数据、依赖）
└── README.md            # 项目主页

重要说明：

• 所有核心代码都在qmoba_agent/包目录下
• 使用前需要运行pip install -e .安装包
• 所有导入使用from qmoba_agent.xxx import yyy格式
• scripts中的脚本不再需要sys.path操作

---

2. 核心文件说明

qmoba_agent/config/config.py

依赖: yaml 类:

• Config: autowzry-agent配置类
  • load(yaml_path) -> Config: 从YAML加载
  • from_yaml(yaml_path) -> Config: 从YAML加载（推荐）
  • default() -> Config: 获取默认配置
  • 属性:
    • data_dir, enable_training
    • use_autowzry_lite: 是否使用本地autowzry_lite库（默认：true，false则使用已安装的autowzry）
    • autowzry_config: autowzry库的配置文件路径（用于设备连接）
    • buffer_capacity
    • train_resolution: 训练分辨率 [height, width, channels]
    • use_last_state: 是否使用laststate+state合并模式（默认：true）
    • enabled_states: 启用的状态列表（默认：[‘in_battle’, ‘dead’, ‘alive’, ‘kill’, ‘assist’]）
    • enabled_actions: 启用的动作列表（默认：[‘move’]）
    • training: num_epochs, batch_size, learning_rate, gamma, samples_per_epoch
      • use_target_network: 是否使用目标网络（默认：true）
      • target_update_epochs: 目标网络更新间隔（默认：5）
      • use_amp: 是否使用混合精度训练（默认：mini模式自动开启）
      • device: 训练设备（默认：’auto’，可选：’cpu’, ‘cuda’, ‘cuda:0’, ‘cuda:1’, etc.）
    • checkpoint: dir, save_interval, resume_model
  • 重要：data_dir 和 checkpoint.dir 用于配置文件中的默认路径，所有脚本的 –output 参数接受完整路径（绝对或相对）

core/lightweight_dqn.py

依赖: torch, torch.nn 类:

• LightweightDQN(nn.Module): DQN模型
  • __init__(input_channels, input_height, input_width, num_actions, mode):
    • input_channels: 输入通道数（由buffer决定，use_last_state=false时为3，true时为6）
    • input_height, input_width: 输入分辨率（由config.train_resolution决定）
    • num_actions: 动作空间维度（由action_space决定）
    • mode: 模型规模（’mini’/’full’/’high’）
  • forward(x) -> q_values: 前向传播，输出Q值
  • save(path): 保存模型和超参数
  • get_params_breakdown() -> dict: 获取参数分布统计
  • 详细设计原理: 见 docs/design/MODEL_DESIGN.md

core/trainer.py

依赖: torch, torch.optim, matplotlib, numpy, data.training_buffer 类:

• Trainer: DQN训练器
  • __init__(compatibility_layer, config):
    • 初始化模型、优化器、训练历史
    • 自动加载resume_model（如果配置）
    • 计算samples_per_epoch（默认为buffer_size）
    • 可选初始化Target Network（use_target_network=true）
    • 可选初始化AMP（use_amp=true且有CUDA）
    • 初始化时同步目标网络状态（如果加载了resume_model）
    • device初始化：config.device=’auto’时自动选择cuda/cpu，否则使用指定设备
  • train(): 训练模型
    • 使用DataLoader遍历数据
    • 每个epoch训练多个batch
    • 支持AMP混合精度训练（autocast + scaler）
    • 记录epoch平均loss
    • 定期保存检查点（save_interval）
    • 定期更新Target Network（target_update_epochs）
    • 训练结束时保存final_model.pth
  • _compute_loss(batch) -> loss: 计算DQN loss（输入已是tensor）
    • 使用Target Network计算max_next_q（提高训练稳定性）
    • 计算action reward（基于模型输出和action_rewards配置）
    • 总奖励 = state_reward + action_reward
  • save_model(path): 保存模型
  • load_model(path): 加载模型
  • plot_training_history(save_dir): 绘制训练曲线（对数坐标）并保存txt数据
  • 属性:
    • self.history: {‘epoch’: [], ‘loss’: []}
    • self.model: 主网络（训练中更新的网络）
    • self.target_model: 目标网络（用于稳定训练，每target_update_epochs同步一次）
    • self.samples_per_epoch: 每个epoch使用的样本数
    • self.use_sampling: 是否需要每个epoch重新采样
    • self.action_slices: 动作切片索引
    • self.action_rewards_tensors: 动作奖励张量（GPU计算用）
    • self.use_amp: 是否使用AMP
    • self.scaler: GradScaler实例（AMP启用时）

data/data_manager.py

依赖: h5py, numpy, cv2, environment.compatibility, environment.game_state, environment.action_space 类:

• StreamingRecorder: 流式HDF5记录器（边记录边写入磁盘，避免内存爆炸）
  • __init__(filepath): 初始化流式记录器，创建HDF5文件
    • 如果文件已存在，自动备份到 .backup（覆盖旧备份）
  • append_frame(frame_data: Dict[str, Any]): 追加一帧数据到HDF5文件（立即写入磁盘）
    • frame_data支持任意字段，常用：image, timestamp, q_values
    • 每次调用立即写入，内存占用仅为单帧大小
  • finalize() -> str: 完成记录，更新文件属性，返回文件路径
  • get_frame_count() -> int: 获取已记录的帧数
• DataManager: 数据管理器
  • __init__(compatibility_layer, action_space, game_state): 初始化（不再接受data_dir参数）
  • create_streaming_recorder(filepath) -> StreamingRecorder: 创建流式记录器（接受完整路径）
  • collect(filepath, max_frames, interval) -> filepath: 收集数据（使用流式写入，接受完整路径）
  • load_dataset(filepath, keys, frame_range) -> Tuple[List[Dict], List[str], Dict]: 加载HDF5数据
    • 返回三元组：(frames_data, frame_names, file_attrs)
  • save_dataset(frames, frame_names, filepath, attrs): 保存到HDF5（支持任意字段）
  • label(filepath, overwrite=False): 标记数据（添加state_/action_字段）
  • add_attrs(filepath, attrs): 添加自定义属性到HDF5文件
  • info(filepath) -> Dict: 获取文件信息
  • 数据格式: HDF5包含image, timestamp, state_, action_（battle.py录制时还包含q_values）
  • 重要：所有方法接受完整路径（filepath），不再使用filename+data_dir模式

data/training_buffer.py

依赖: h5py, torch, torch.utils.data, numpy, os, datetime, random 类:

• TrainingBuffer: 训练数据加载器
  • __init__(data_manager=None, action_space=None, game_state=None, capacity=100000, config=None)
  • load(filepath, max_frames=None): 从HDF5加载数据到buffer
    • 使用 data_manager.load_dataset() 获取 (frames_data, frame_names, file_attrs)
    • 从HDF5读取state_和action_字段
    • 使用game_state.sum_reward和action_space.sum_reward计算奖励
    • 当use_last_state=true时：
      • 读取frames[i-1]作为laststate（边界和in_battle检查）
      • 预处理后合并：state = laststate + current_state，next_state = current_state + next_state
      • 通道数从3变为6（或1变为2，取决于train_resolution）
    • 样本中添加 frame_name 字段用于调试追踪
  • sample_indices(sample_size) -> List[int]: 随机采样索引
  • get_dataloader(batch_size, shuffle, device, sample_size) -> DataLoader: 获取PyTorch DataLoader
    • sample_size=None: 使用所有数据
    • sample_size<buffer_size: 随机采样指定数量
  • save_to_hdf5(filepath): 保存所有buffer样本到HDF5（用于调试验证）
    • 动态检测所有字段（state, next_state, frame_name, move, reward等）
    • 自动适配未来新增字段
    • 使用gzip压缩大型数组
  • __len__() -> int: 总样本数
• BufferDataset(Dataset): PyTorch Dataset包装器
  • __init__(samples, device): 初始化
  • __getitem__(idx) -> dict: 返回tensor格式的样本（已在device上）
    • 字符串字段（如frame_name）保持不变
    • 数值字段转换为tensor

environment/compatibility.py

依赖: numpy, cv2, autowzry, utils.video_reader 类:

• CompatibilityLayer: 兼容层（依赖注入中心）
  • __init__(config, mode='online'/'offline', video_reader=None, connect=True)
  • 初始化模块:
    • action_module: ActionSpace实例
    • game_state_module: GameState实例
    • data_manager: DataManager实例
    • training_buffer: TrainingBuffer实例（enable_training=True时）
  • 画面获取:
    • get_screen() -> np.ndarray: 获取当前画面
  • 状态检测:
    • is_in_battle(screen) -> bool: 检测是否在战斗中
    • is_dead(screen) -> bool: 检测死亡
    • detect_kill_count(screen) -> int: 检测击杀数
    • detect_assist_count(screen) -> int: 检测助攻数
  • 动作执行 (仅online):
    • execute_move(cmd): 执行移动
    • execute_attack(cmd): 执行攻击
  • 离线模式:
    • 离线模式也初始化autowzry（connect=False）
    • 可以调用检测函数进行数据标记

environment/action_space.py

依赖: environment.compatibility 类:

• ActionSpace: 动作空间定义
  • __init__(compatibility_layer, enabled_actions=None): 初始化
    • enabled_actions: 启用的动作列表（如 [‘move’, ‘attack’]）
  • get_action_keys() -> List[str]: 获取激活的action名称
  • get_action_dims() -> Dict[str, int]: 获取每个action的维度
  • get_total_dim() -> int: 获取总维度
  • get_frame_action(frame1, frame2) -> Dict[str, List[float]]: 获取当前帧的动作字典
    • 参数: frame1(当前帧), frame2(下一帧)
    • 返回: 如 {‘move’: [0,0,0,1], ‘attack’: [0]*10}
  • execute_action(cmd): 执行动作（委托给兼容层）
  • get_action_slices() -> List[Tuple[int, int]]: 获取每个action在模型输出中的索引范围
  • get_action_rewards_list() -> List[List[float]]: 获取每个action的rewards数组（按enabled_actions顺序）
  • 属性:
    • actions: 动作空间定义（如 {‘move’: 4, ‘attack’: 10}）
    • action_rewards: 动作奖励权重（如 {‘move’: [0.01, 0.01, 0.01, 0.1]}），所有移动都有基础奖励0.01（和alive状态一致），向右移动10倍奖励，与state_rewards单位统一

environment/game_state.py

依赖: environment.compatibility 类:

• GameState: 游戏状态检测和得分计算
  • __init__(compatibility_layer, enabled_states=None): 初始化
    • enabled_states: 启用的状态列表（如 [‘in_battle’, ‘dead’, ‘alive’, ‘kill’, ‘assist’]）
  • get_frame_state(frame1, frame2) -> dict: 获取当前帧的状态字典
    • 参数: frame1(当前帧), frame2(下一帧)
    • 返回: 如 {‘in_battle’: True, ‘dead’: False, ‘alive’: True, ‘kill’: False, ‘assist’: False}
    • kill/assist通过比较两帧的计数判断
  • sum_reward(state_dict) -> float: 根据状态字典计算总得分
  • 属性:
    • enabled_states: 启用的状态列表
    • state_scores: 状态得分配置（如 {‘dead’: -2, ‘alive’: 0.01, ‘kill’: 1, ‘assist’: 1, ‘in_battle’: 0}）

utils/image_processing.py

依赖: cv2, numpy 函数:

• resize_frame(frame, width, height)
• to_grayscale(frame)
• normalize(frame): 0-255 -> 0-1
• preprocess_frame(frame, width, height, grayscale, resize=True)
• stack_frames(frames) 类:
• FrameBuffer:
  • __init__(stack_size, frame_shape)
  • add_frame(frame) -> stacked: 添加并返回堆叠帧
  • 支持(H,W)和(H,W,C)两种形状

utils/video_reader.py

依赖: cv2 类:

• VideoReader:
  • __init__(video_path, frame_interval=0, target_fps=None)
  • __iter__(), __next__() -> (idx, frame)
  • 支持跳帧和FPS控制

---

3. 数据类型与流转

数据流转过程

画面数据 (RGB图像)
  -> CompatibilityLayer.capture_screen()
  -> DataManager.collect()
  -> HDF5文件 (未标记)
  -> DataManager.label()
  -> HDF5文件 (已标记)
  -> TrainingBuffer.load()
  -> TrainingBuffer.get_dataloader()
  -> Trainer.train_step()
  -> 模型

说明：

• 画面来源由CompatibilityLayer决定（视频文件或设备画面）
• 数据流转过程与模式（离线/在线）无关
• 标记过程添加动作和奖励标签

HDF5文件结构

标准数据文件（collect + label生成）：

episode_xxx.hdf5
├── attrs (文件级属性)
│   ├── num_frames: 总帧数
│   ├── created_at: 创建时间
│   ├── labeled_at: 标记时间（可选）
│   ├── original_file: 原始文件（提取时）
│   ├── extracted_range: 提取范围（提取时）
│   └── ...
├── frame_000000/
│   ├── image: (540, 960, 3) RGB图像
│   ├── timestamp: 时间戳
│   ├── state_in_battle: bool（标记后）
│   ├── state_dead: bool（标记后）
│   ├── state_alive: bool（标记后）
│   ├── state_kill: bool（标记后）
│   ├── state_assist: bool（标记后）
│   ├── action_move: [上, 下, 左, 右] 得分（标记后）
│   └── action_attack: [目标1, ..., 目标10] 得分（标记后）
├── frame_000001/
│   └── ...
└── ...

battle.py录制文件（使用–record-interval和–output生成）：

battle_record.hdf5
├── attrs (文件级属性)
│   ├── num_frames: 总帧数
│   ├── created_at: 创建时间
│   ├── finished_at: 完成时间
│   ├── recording_mode: 'streaming'
│   └── ...
├── frame_000000/
│   ├── image: (540, 960, 3) RGB图像
│   ├── timestamp: 时间戳
│   └── q_values: (total_dim,) 模型输出的Q值
├── frame_000001/
│   └── ...
└── ...

说明：

• 帧名称格式：frame_XXXXXX（6位数字）
• 帧编号可能不连续（提取后）
• 标记前只有 image 和 timestamp
• 标记后添加 state_* 和 action_* 字段
• battle.py录制时包含 q_values（模型输出，可用于分析决策过程）
• reward 不存储，由 buffer 动态计算

TrainingBuffer样本格式

{
    'state': (C, 540, 960),      # 当前帧图像，C=3或6（取决于use_last_state）
    'next_state': (C, 540, 960), # 下一帧图像，C=3或6
    'frame_name': str,            # 帧名称（如'frame_000042'），用于调试追踪
    'move': [4],                  # 移动动作得分 [上, 下, 左, 右]
    'reward': float               # 总奖励值（状态奖励 + 动作奖励）
}

说明：

• state 和 next_state 是相邻帧（单帧模式）或合并帧（双帧模式）
• use_last_state=false时：C=3（单帧RGB），state=frames[i]，next_state=frames[i+1]
• use_last_state=true时：C=6（双帧合并），state=frames[i-1]+frames[i]，next_state=frames[i]+frames[i+1]
• frame_name 标识当前帧在HDF5文件中的名称，便于追踪异常样本
• move 是当前帧的动作标签（从HDF5的action_move读取）
• reward 由buffer动态计算（仅使用game_state.sum_reward，表示下一帧的状态奖励）
• action reward 在训练时由trainer根据模型输出的Q值计算（查表action_rewards）