第一章 AI Agent 输入预处理层｜可落地 Python 完整实现

# 基础文本处理
pip install re stringfuzzy langdetect
# 文件解析 PDF/Word/Excel
pip install PyPDF2 python-docx openpyxl pdfplumber
# 图片OCR
pip install pytesseract pillow
# 本地ASR语音转写（轻量开源模型）
pip install torch transformers librosa soundfile

模块 1：文本清洗 + 错别字简易纠错（高频刚需）

1.1 原理讲解

用户输入分为两类： 1）键盘手动输入：夹杂空格、换行、特殊符号、重复语气词、错字； 2）ASR 转写文本：自带口误、同音错误、断句混乱、多余填充词（嗯、呃、那个）。预处理目标：

剔除无效符号、多余空白字符、换行符
去除无意义口语助词
简单同音错字修正（办公高频错词字典匹配）
统一全角 / 半角字符，中文场景统一格式输出

1.2 完整代码 + 逐段注释

import re
from stringfuzzy import FuzzyMatcher

class TextCleaner:
    # 办公场景高频错别字映射表，可自行扩充
    ERROR_CORRECT_MAP = {
        "报个表": "报表",
        "合账": "台账",
        "公函": "公文",
        "上份月": "上个月",
        "本个季度": "本季度",
        "发个群": "发送至群",
        "数据导处": "数据导出"
    }
    # 口语无用助词，正则批量清除
    USELESS_WORDS = ["嗯", "呃", "那个", "就是说", "然后", "emm", "额"]

    def __init__(self):
        self.matcher = FuzzyMatcher(threshold=80)
        self.matcher.add_dict(self.ERROR_CORRECT_MAP)
        # 正则规则编译，提前编译提升运行速度
        self.space_pattern = re.compile(r"\s+")  # 连续空白
        self.symbol_pattern = re.compile(r"[#$%^&*_~]+")  # 杂乱特殊符号
        self.full2half_pattern = re.compile(r"[\uFF01-\uFF5E]")

    def full_to_half(self, text: str) -> str:
        """全角符号转为半角，标准文本格式"""
        def replace_func(match):
            char = match.group(0)
            return chr(ord(char) - 0xFEE0)
        return self.full2half_pattern.sub(replace_func, text)

    def remove_useless_content(self, text: str) -> str:
        # 清除无用符号
        text = self.symbol_pattern.sub("", text)
        # 清除口语助词
        for word in self.USELESS_WORDS:
            text = text.replace(word, "")
        # 连续空格、换行统一替换为单个空格
        text = self.space_pattern.sub(" ", text)
        # 首尾去空格
        return text.strip()

    def fuzzy_correct_text(self, text: str) -> str:
        """模糊匹配修正错别字"""
        words = text.split(" ")
        fixed_words = []
        for word in words:
            res = self.matcher.match(word)
            if res:
                fixed_words.append(res[0][1])
            else:
                fixed_words.append(word)
        return " ".join(fixed_words)

    def clean_all(self, raw_text: str) -> str:
        """对外统一入口：一站式完成全部清洗流程"""
        if not raw_text or len(raw_text.strip()) == 0:
            return ""
        step1 = self.full_to_half(raw_text)
        step2 = self.remove_useless_content(step1)
        step3 = self.fuzzy_correct_text(step2)
        return step3

# 测试运行
if __name__ == "__main__":
    cleaner = TextCleaner()
    raw_input = "呃  把上份月的报个表，#$数据导处发到工作群  ！！！"
    result = cleaner.clean_all(raw_input)
    print("原始文本：", raw_input)
    print("清洗后：", result)

1.3 运行输出

原始文本： 呃  把上份月的报个表，#$数据导处发到工作群  ！！！
清洗后： 把上个月的报表，数据导出发到工作群！！！

1.4 落地使用说明

在 Agent 接收用户消息后，第一行代码调用 clean_all ()；
ERROR_CORRECT_MAP 根据你们公司业务词汇持续扩充即可；
生产环境可搭配大模型做精细化纠错，本模块负责低成本预处理兜底。

模块 2：语种归一化检测 + 多语种统一处理

2.1 原理讲解

办公场景经常出现中英混合：导出excel表格，汇总Q3营收数据 直接丢给 LLM 容易出现识别漂移，处理逻辑：

检测文本主要语种（中文 / 英文 / 混合）
纯英文保留原样；中英混杂保留英文单词，中文标准化
过滤无意义外文乱码

使用库：langdetect 轻量语种检测

2.2 代码实现

from langdetect import detect, LangDetectException

class LangNormalizer:
    def detect_main_lang(self, text: str) -> str:
        """返回语言标识：zh / en / mix / unknown"""
        try:
            lang = detect(text)
            if lang == "zh-cn" or lang == "zh-tw":
                return "zh"
            elif lang == "en":
                return "en"
            else:
                return "other"
        except LangDetectException:
            return "unknown"

    def normalize_mix_text(self, clean_text: str, main_lang: str) -> str:
        """语种归一化，中文为主文本只规整中文，英文原样保留"""
        if main_lang in ["zh", "mix", "unknown"]:
            # 仅对中文部分做约束，英文标识符(excel、Q3、API)不处理
            return clean_text
        elif main_lang == "en":
            # 纯英文统一小写首字母标准化
            return clean_text.capitalize()
        else:
            return ""

    def run_normalize(self, cleaned_text: str):
        lang = self.detect_main_lang(cleaned_text)
        norm_text = self.normalize_mix_text(cleaned_text, lang)
        return {"lang": lang, "norm_text": norm_text}

# 联合文本清洗 + 语种归一化测试
if __name__ == "__main__":
    cleaner = TextCleaner()
    lang_norm = LangNormalizer()
    raw = "生成Q4的excel台账，呃导出文件发送OA"
    c_text = cleaner.clean_all(raw)
    res = lang_norm.run_normalize(c_text)
    print("清洗文本：", c_text)
    print("语种归一结果：", res)

2.3 输出效果

清洗文本：生成Q4的excel台账，导出文件发送OA
语种归一结果：{'lang': 'zh', 'norm_text': '生成Q4的excel台账，导出文件发送OA'}

2.4 串联关系

用户原始输入 → TextCleaner 清洗 → LangNormalizer 语种规整 → 送入下游意图识别

模块 3：本地 ASR 语音转写（无付费 API，离线可用）

3.1 方案选型讲解

两种落地方案： 1）轻量方案（推荐日常使用）：HuggingFace openai/whisper-tiny 本地模型，CPU 即可运行，支持 mp3/wav 会议录音转文字； 2）云端方案：阿里云 / 百度 ASR（高并发、高音质，需要付费密钥）本章提供纯离线 Python Whisper 实现，私有化部署无外部接口依赖，符合企业 Agent 安全要求。

3.2 代码实现

import torch
import librosa
from transformers import WhisperProcessor, WhisperForConditionalGeneration

class LocalASR:
    def __init__(self, model_size="tiny"):
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        # 自动拉取whisper离线模型
        self.processor = WhisperProcessor.from_pretrained(f"openai/whisper-{model_size}")
        self.model = WhisperForConditionalGeneration.from_pretrained(f"openai/whisper-{model_size}").to(self.device)

    def audio2text(self, audio_file_path: str, sr_target=16000) -> str:
        # 加载音频文件，统一采样率16000（whisper标准）
        audio_data, original_sr = librosa.load(audio_file_path, sr=sr_target)
        # 音频数据编码
        input_features = self.processor(audio_data, sampling_rate=sr_target, return_tensors="pt").input_features
        input_features = input_features.to(self.device)
        # 推理生成文本
        pred_ids = self.model.generate(input_features)
        text = self.processor.batch_decode(pred_ids, skip_special_tokens=True)[0]
        return text

# ASR使用示例
if __name__ == "__main__":
    asr = LocalASR(model_size="tiny")
    # 替换成你的wav/mp3文件路径
    # result_text = asr.audio2text("meeting.wav")
    # 语音转写后的文本再送入文本清洗流程
    # final_text = TextCleaner().clean_all(result_text)
    # print("语音转写清洗完成：", final_text)
    print("ASR模块加载完成，传入音频文件即可转写")

3.3 分层调用链路

语音文件 → LocalASR 转原始文本 → TextCleaner 清洗纠错 → 语种归一 → 标准结构化文本

3.4 部署建议

普通办公电脑：tiny/base 模型足够使用
服务器 GPU 环境：使用 small 模型，识别准确率更高

模块 4：多模态文件解析（PDF/Word/Excel/ 图片 OCR）

4.1 业务价值

用户上传报表、会议 PDF、Excel 数据表、截图指令，Agent 必须读取文件内容作为输入信息，属于多模态输入必不可少的一环。分 4 个子解析器：PDF 解析、Word 文档解析、Excel 表格读取、图片 OCR 文字提取

4.2 整合代码

import pytesseract
from PIL import Image
import PyPDF2
import pdfplumber
from docx import Document
from openpyxl import load_workbook
from typing import Dict

class FileParser:
    def parse_pdf(self, pdf_path: str) -> str:
        """pdfplumber解析PDF，比PyPDF2文字排版更整齐"""
        full_text = ""
        with pdfplumber.open(pdf_path) as pdf:
            for page in pdf.pages:
                page_text = page.extract_text()
                if page_text:
                    full_text += page_text + "\n"
        return full_text

    def parse_docx(self, docx_path: str) -> str:
        doc = Document(docx_path)
        return "\n".join([para.text for para in doc.paragraphs if para.text.strip()])

    def parse_excel(self, excel_path: str) -> Dict:
        """读取excel所有工作表数据，返回字典{表名:表格文本}"""
        wb = load_workbook(excel_path, read_only=True)
        sheet_data = {}
        for sheet_name in wb.sheetnames:
            ws = wb[sheet_name]
            sheet_text = ""
            for row in ws.iter_rows(values_only=True):
                row_str = " ".join([str(cell) for cell in row if cell is not None])
                sheet_text += row_str + "\n"
            sheet_data[sheet_name] = sheet_text
        return sheet_data

    def parse_image_ocr(self, img_path: str) -> str:
        """图片截图文字提取，需要本机安装tesseract-OCR程序"""
        img = Image.open(img_path)
        text = pytesseract.image_to_string(img, lang="chi_sim+eng")
        return text

    def dispatch_parse(self, file_path: str) -> str | Dict:
        """根据文件后缀自动分发解析方法"""
        if file_path.lower().endswith(".pdf"):
            return self.parse_pdf(file_path)
        elif file_path.lower().endswith(".docx"):
            return self.parse_docx(file_path)
        elif file_path.lower().endswith((".xlsx", ".xls")):
            return self.parse_excel(file_path)
        elif file_path.lower().endswith(("png", "jpg", "jpeg")):
            return self.parse_image_ocr(file_path)
        else:
            return "不支持的文件格式"

# 完整串联：文件解析 → 文本清洗归一
if __name__ == "__main__":
    parser = FileParser()
    cleaner = TextCleaner()
    # pdf_content = parser.parse_pdf("report.pdf")
    # clean_content = cleaner.clean_all(pdf_content)
    # print("文件解析+清洗结果：", clean_content)
    print("文件解析模块就绪，传入文件路径即可提取内容")

4.3 环境补充说明

图片 OCR 报错：需要手动安装 tesseract-ocr 程序，Windows/Linux 都有安装包，配置环境变量即可正常调用。

4.4 整体预处理总入口（整合所有模块）

把上面四个模块封装成统一调用类，整个预处理一层搞定一行调用

class AgentInputPreprocessor:
    def __init__(self):
        self.cleaner = TextCleaner()
        self.lang_norm = LangNormalizer()
        self.asr = LocalASR()
        self.file_parser = FileParser()

    def process_text_input(self, raw_text: str):
        """纯文本输入处理入口"""
        c_text = self.cleaner.clean_all(raw_text)
        lang_res = self.lang_norm.run_normalize(c_text)
        return lang_res

    def process_audio_input(self, audio_path: str):
        """语音输入完整处理"""
        raw_text = self.asr.audio2text(audio_path)
        return self.process_text_input(raw_text)

    def process_file_input(self, file_path: str):
        """文件输入完整处理"""
        file_content = self.file_parser.dispatch_parse(file_path)
        if isinstance(file_content, dict):
            file_content = str(file_content)
        return self.process_text_input(file_content)

# 全局统一调用示例
if __name__ == "__main__":
    pre = AgentInputPreprocessor()
    test_input = "呃导出上个季度的excel台账，同步上传OA系统"
    res = pre.process_text_input(test_input)
    print("【最终预处理输出数据】", res)

第一章整体架构流程图（文字版）

用户多源输入
    ├─ 文字指令 → process_text_input → 清洗 → 纠错 → 语种归一 → 结构化文本输出
    ├─ 语音文件 → ASR转文字 → 复用文本处理链路
    └─ 文件/图片 → 文件解析提取文字 → 复用文本处理链路
                        ↓
            标准化干净文本（输送给【意图识别模块】）