4、感知模块设计：传感器抽象、数据采集接口、多模态输入处理（文本、图像、音频）、数据预处理流水线。

好，我们开始进入智能体架构里最接地气、也最容易踩坑的一章——感知模块。

说白了，感知模块就是智能体的「五官」。没有它，你的智能体就是个闭着眼睛、捂着耳朵、不说话的空壳。我见过不少团队，一上来就猛搞推理引擎、知识图谱，结果发现数据进都进不来，或者进来的全是脏数据。嗯，那感觉就像你盖好了豪华别墅，结果发现没装水管。

这一章，我会带着你把感知模块的四个核心部分拆开揉碎：传感器抽象、数据采集接口、多模态输入处理、数据预处理流水线。

4.1 传感器抽象：给智能体装个「通用五官」

先问一个问题：你的智能体要从哪里获取信息？

可能是摄像头、麦克风、键盘输入、API 接口、文件上传……如果每接入一种新设备就重写一遍逻辑，那项目很快就会变成一团乱麻。

所以我们需要一个传感器抽象层。它的作用很简单：把千奇百怪的物理或逻辑传感器，统一成一个标准接口。

我个人习惯这样设计基础接口：

from abc import ABC, abstractmethod
from typing import Any, Dict

class BaseSensor(ABC):
    """所有传感器的基类"""
    
    @abstractmethod
    def initialize(self, config: Dict[str, Any]) -> None:
        """初始化传感器，比如打开摄像头、建立连接"""
        pass
    
    @abstractmethod
    def read(self) -> Any:
        """读取一次数据，返回原始格式"""
        pass
    
    @abstractmethod
    def calibrate(self) -> None:
        """校准传感器，比如调整白平衡、降噪"""
        pass
    
    @abstractmethod
    def shutdown(self) -> None:
        """安全关闭，释放资源"""
        pass

有了这个基类，你就能派生出各种具体传感器：

传感器类型	数据源	输出格式	典型初始化参数
CameraSensor	USB摄像头 / IP摄像头	numpy.ndarray (RGB)	device_id, resolution, fps
MicrophoneSensor	系统麦克风 / 音频流	numpy.ndarray (PCM)	sample_rate, channels, chunk_size
TextInputSensor	键盘 / 文件 / API	str	source_type, encoding
FileSensor	本地文件 / 网络文件	bytes	path, poll_interval

4.2 数据采集接口：统一的数据「水龙头」

传感器抽象解决了「怎么读」的问题，但数据采集接口解决的是「怎么统一管理」的问题。

你想想看，一个智能体可能同时挂着 5 个摄像头、3 个麦克风、还有一堆文本输入源。如果每个传感器各自为政，数据格式、时间戳、采样频率全都不一样，那下游处理模块根本没法干活。

所以我们需要一个数据采集管理器：

class DataAcquisitionManager:
    def __init__(self):
        self._sensors: Dict[str, BaseSensor] = {}
        self._buffer: asyncio.Queue = asyncio.Queue()
    
    def register_sensor(self, name: str, sensor: BaseSensor, config: dict):
        sensor.initialize(config)
        self._sensors[name] = sensor
    
    async def start_collection(self):
        """启动所有传感器的数据采集循环"""
        tasks = [self._collect_loop(name, sensor) 
                 for name, sensor in self._sensors.items()]
        await asyncio.gather(*tasks)
    
    async def _collect_loop(self, name: str, sensor: BaseSensor):
        while True:
            raw_data = await sensor.read()
            # 统一包装成标准数据包
            packet = {
                "sensor_id": name,
                "timestamp": time.time(),
                "data": raw_data,
                "modality": self._detect_modality(sensor)
            }
            await self._buffer.put(packet)
            await asyncio.sleep(0)  # 让出控制权

这里的关键是统一数据包格式。不管什么传感器，出来的数据都带上 sensor_id、timestamp、modality 三个字段。这样下游处理时，你只需要根据 modality 分发，不用关心数据是从哪来的。

4.3 多模态输入处理：文本、图像、音频

数据采集上来之后，不同模态的数据需要不同的处理方式。这部分我直接给你画个图，一目了然：

每个模态的处理细节，我挑重点说：

文本处理

• 分词：中文用 jieba 或 HanLP，英文用 spaCy 或 NLTK。我个人习惯在智能体场景里直接用 tokenizer 级别的分词，省去词性标注这些不必要的开销。
• 编码：统一转 UTF-8，处理 BOM 头。我遇到过从 Windows 传过来的 GBK 文件，没处理直接崩了。
• 向量化：轻量场景用 TF-IDF，语义场景用 Sentence-BERT 或 OpenAI Embeddings。

图像处理

• 解码：OpenCV 的 imdecode 比 imread 更灵活，支持从内存直接解码。
• 缩放：统一到 224x224 或 512x512，保持宽高比，用 padding 补全。
• 归一化：像素值缩放到 [0,1] 或 [-1,1]，具体看下游模型要求。

音频处理

• 重采样：统一到 16kHz，这是大多数语音模型的标准采样率。
• 降噪：简单的用谱减法，效果好的用 RNNoise 或 WebRTC 的降噪模块。
• 分帧：帧长 25ms，帧移 10ms，提取 MFCC 或 Fbank 特征。

4.4 数据预处理流水线：让数据「干干净净」进模型

数据预处理，说白了就是「洗菜」。菜不洗干净，炒出来的菜没法吃。数据不预处理，模型训练出来就是个废物。

我推荐用流水线模式来组织预处理流程。每个处理步骤是一个独立的节点，可以插拔、复用、测试。

class PreprocessingPipeline:
    def __init__(self):
        self._stages: List[ProcessingStage] = []
    
    def add_stage(self, stage: ProcessingStage):
        self._stages.append(stage)
    
    async def process(self, data_packet: dict) -> dict:
        result = data_packet
        for stage in self._stages:
            result = await stage.execute(result)
            if result is None:
                # 某个阶段判定数据无效，直接丢弃
                return None
        return result

class ProcessingStage(ABC):
    @abstractmethod
    async def execute(self, packet: dict) -> dict:
        pass

一个典型的预处理流水线包含这些阶段：

阶段	作用	示例操作
1. 格式校验	检查数据包结构是否完整	验证 sensor_id、timestamp、data 字段存在
2. 数据清洗	去除无效或异常数据	丢弃空帧、静音段、乱码文本
3. 模态转换	统一数据格式	图像转 RGB、音频转 float32
4. 特征提取	从原始数据中提取有用特征	文本向量化、图像特征图、音频 MFCC
5. 标准化	缩放到模型期望的范围	Z-score 归一化、Min-Max 缩放
6. 缓存	避免重复计算	LRU 缓存最近 1000 条处理结果

最后，别忘了给流水线加一个监控和告警机制。我习惯在每个阶段记录处理耗时、成功/失败次数、数据量大小。一旦某个阶段的失败率超过 5%，立刻告警。这样你就能在用户发现问题之前，先把问题解决掉。