SeqGPT-560M实战教程：将提取结果自动写入MySQL/Excel，支持定时批处理

SeqGPT-560M实战教程：将提取结果自动写入MySQL/Excel，支持定时批处理

1. 这不是聊天机器人，而是一台“信息榨汁机”

你有没有遇到过这样的场景：
每天收到几十份PDF简历，要手动复制姓名、公司、职位、电话；
销售团队发来上百条会议纪要，需要从中扒出客户名称、签约意向、金额和时间节点；
法务部堆积着未整理的合同扫描件，关键条款散落在段落里，没人有时间逐字核对。

这些都不是“对话”问题，而是结构化信息提取问题。
SeqGPT-560M 不是另一个会讲段子的大模型，它是一台专为“榨取信息”而生的轻量级工业级工具——模型参数仅5.6亿，却在双路RTX 4090上跑出<200ms单次推理速度，且不生成一句废话。

它的核心设计哲学就一句话：不编、不猜、不绕，只准提。
不用“请帮我找出……”，不用“能否提取以下内容？”，你只需要说“姓名, 公司, 职位, 手机号”，它就老老实实把对应字段填进表格里，像一台精准校准过的数控机床。

本教程不讲原理推导，不调参、不训微调，只聚焦一件事：
把你刚提取出来的那几行结构化数据，自动存进MySQL表里；
同时生成一份带格式的Excel文件，每天早上9点准时发到你邮箱；
支持批量处理整文件夹的TXT/LOG/MD文本，一次塞入500份，后台静默跑完。

全程无需改一行模型代码，所有功能都通过配置+轻量脚本实现。下面我们就从零开始搭起这条“信息流水线”。

2. 环境准备与一键部署（5分钟搞定）

2.1 硬件与系统要求

注意：本教程默认你已成功运行官方Streamlit界面（即能打开http://localhost:8501并完成单次提取）。若尚未部署，请先执行官方仓库中的./deploy.sh脚本完成基础安装。

2.2 安装配套工具链（终端中逐行执行）

# 激活项目虚拟环境（假设你已在项目根目录） source venv/bin/activate # 安装数据库与办公文件支持包 pip install PyMySQL openpyxl pandas schedule python-dotenv # 创建必要目录结构 mkdir -p data/input data/output logs config touch config/db_config.env config/email_config.env

2.3 配置MySQL连接（真实可用，非示例）

在config/db_config.env中填入你的内网数据库信息：

DB_HOST=192.168.1.100 DB_PORT=3306 DB_NAME=info_extraction_db DB_USER=seqgpt_app DB_PASSWORD=YourSecurePass2024!

小贴士：首次使用前，请在MySQL中手动建库建表（SQL语句见下节），避免程序因权限或表缺失崩溃。

3. 结构化结果自动写入MySQL（零SQL手写）

3.1 创建目标数据表（只需执行一次）

用MySQL客户端连接后，执行以下建表语句。表名extracted_entities，字段完全适配SeqGPT-560M输出格式：

CREATE TABLE IF NOT EXISTS extracted_entities ( id BIGINT PRIMARY KEY AUTO_INCREMENT, task_id VARCHAR(32) NOT NULL COMMENT '批次唯一标识', input_text_hash CHAR(32) NOT NULL COMMENT '原文MD5，防重复入库', extracted_at DATETIME DEFAULT CURRENT_TIMESTAMP, source_file VARCHAR(255) DEFAULT NULL COMMENT '原始文件名（如resume_001.txt）', -- 动态字段区（按你定义的标签自动生成列） name VARCHAR(128) DEFAULT NULL, company VARCHAR(255) DEFAULT NULL, position VARCHAR(128) DEFAULT NULL, phone VARCHAR(32) DEFAULT NULL, email VARCHAR(128) DEFAULT NULL, amount DECIMAL(12,2) DEFAULT NULL, date DATE DEFAULT NULL, notes TEXT DEFAULT NULL, -- 元数据 model_version VARCHAR(20) DEFAULT 'SeqGPT-560M-v1.2', inference_time_ms INT DEFAULT 0, INDEX idx_task_id (task_id), INDEX idx_hash (input_text_hash), INDEX idx_date (extracted_at) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

为什么字段名用小写+下划线？因为SeqGPT-560M的输出JSON键名默认小写（如{"name": "张三", "company": "XX科技"}），直接映射省去转换逻辑，也避免大小写敏感引发的BUG。

3.2 编写入库脚本（scripts/save_to_mysql.py）

# scripts/save_to_mysql.py import json import os import pymysql from dotenv import load_dotenv from pathlib import Path # 加载数据库配置 load_dotenv("config/db_config.env") def save_to_mysql(extracted_data: dict, task_id: str, source_file: str = None): """ 将SeqGPT-560M返回的字典结果写入MySQL extracted_data: {"name": "张三", "company": "XX科技", ...} task_id: 批次ID，建议用 datetime.now().strftime("%Y%m%d_%H%M%S") source_file: 原始文件路径（可选） """ conn = pymysql.connect( host=os.getenv("DB_HOST"), port=int(os.getenv("DB_PORT", "3306")), user=os.getenv("DB_USER"), password=os.getenv("DB_PASSWORD"), database=os.getenv("DB_NAME"), charset="utf8mb4", autocommit=True ) try: with conn.cursor() as cursor: # 构建动态INSERT语句（安全拼接字段与值） fields = ["task_id", "input_text_hash", "source_file"] values = [task_id, extract_hash(extracted_data), source_file or "manual_input"] # 自动添加用户定义字段（跳过内置元字段） for key, value in extracted_data.items(): if key not in ["task_id", "input_text_hash", "source_file", "model_version"]: fields.append(key) # 对字符串做安全转义，数值/None保持原样 if isinstance(value, str): values.append(cursor.escape(value)) else: values.append(value) placeholders = ", ".join(["%s"] * len(values)) field_names = ", ".join([f"`{f}`" for f in fields]) sql = f"INSERT INTO extracted_entities ({field_names}) VALUES ({placeholders})" cursor.execute(sql, values) print(f" 已写入MySQL：{len(extracted_data)}个字段，ID={cursor.lastrowid}") except Exception as e: print(f"❌ MySQL写入失败：{e}") finally: conn.close() def extract_hash(data: dict) -> str: """生成输入内容简易哈希（避免全量MD5计算开销）""" import hashlib text = "".join(str(v) for v in data.values() if v is not None) return hashlib.md5(text.encode()).hexdigest()[:32] if __name__ == "__main__": # 示例调用（实际由主程序传入） sample = {"name": "李四", "company": "智算科技", "phone": "138****1234"} save_to_mysql(sample, "20240520_143022", "resume_007.txt")

3.3 在Streamlit中集成入库动作

打开app.py（或你启动Web界面的主文件），找到提取按钮触发逻辑，在st.button("开始精准提取")之后插入：

# 假设 extract_result 是你调用模型后得到的字典 if st.button("开始精准提取"): with st.spinner("正在提取并保存..."): extract_result = seqgpt_inference(input_text, target_labels) # 新增：自动保存至MySQL from scripts.save_to_mysql import save_to_mysql import datetime task_id = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") save_to_mysql(extract_result, task_id, uploaded_file.name if uploaded_file else None) # 同时显示结果 st.json(extract_result)

效果验证：提交一次后，登录MySQL执行SELECT * FROM extracted_entities ORDER BY id DESC LIMIT 1;，即可看到刚入库的记录。

4. Excel导出与定时邮件发送（全自动闭环）

4.1 生成带格式Excel（含表头、自动列宽、时间水印）

创建scripts/export_to_excel.py：

# scripts/export_to_excel.py import pandas as pd from openpyxl import Workbook from openpyxl.styles import Font, PatternFill, Alignment, Border, Side from openpyxl.utils import get_column_letter from datetime import datetime import os def export_to_excel(data_list: list, filename: str): """ 将多条提取结果导出为Excel（自动适配字段） data_list: [{"name":"张三","company":"XX"}, ...] filename: 输出路径，如 "output/20240520_extract.xlsx" """ if not data_list: print(" 无数据可导出") return # 转为DataFrame，自动对齐所有字段 df = pd.DataFrame(data_list) # 创建工作簿 wb = Workbook() ws = wb.active ws.title = "提取结果" # 写入表头（加粗+背景色） header_font = Font(bold=True, color="FFFFFF") header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid") border = Border( left=Side(style='thin'), right=Side(style='thin'), top=Side(style='thin'), bottom=Side(style='thin') ) for col_num, column_title in enumerate(df.columns, 1): cell = ws.cell(row=1, column=col_num, value=column_title) cell.font = header_font cell.fill = header_fill cell.alignment = Alignment(horizontal="center", vertical="center") cell.border = border # 写入数据行 for row_num, row_data in enumerate(df.values, 2): for col_num, value in enumerate(row_data, 1): cell = ws.cell(row=row_num, column=col_num, value=value) cell.border = border cell.alignment = Alignment(vertical="center") # 自动列宽（中文字符按2个英文宽度算） for column in ws.columns: max_length = 0 column_letter = get_column_letter(column[0].column) for cell in column: try: if cell.value: length = len(str(cell.value)) # 中文字符加权 cn_chars = sum(1 for c in str(cell.value) if '\u4e00' <= c <= '\u9fff') adjusted = length + cn_chars if adjusted > max_length: max_length = adjusted except: pass adjusted_width = min(max_length + 2, 50) # 限制最大宽度 ws.column_dimensions[column_letter].width = adjusted_width # 添加页脚（生成时间） footer_row = len(df) + 2 ws.cell(row=footer_row, column=1, value=f"生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") ws.cell(row=footer_row, column=1).font = Font(italic=True, size=10, color="888888") # 保存 os.makedirs(os.path.dirname(filename), exist_ok=True) wb.save(filename) print(f" Excel已生成：{filename}") # 示例调用 if __name__ == "__main__": test_data = [ {"name": "王五", "company": "云智未来", "position": "算法工程师", "phone": "159****5678"}, {"name": "赵六", "company": "深瞳数科", "position": "产品经理", "email": "zhaoliu@deepvision.ai"} ] export_to_excel(test_data, "output/test_export.xlsx")

4.2 配置邮箱发送（企业微信/Outlook/SMTP通用）

在config/email_config.env中填写：

EMAIL_SMTP_SERVER=smtp.exmail.qq.com EMAIL_SMTP_PORT=465 EMAIL_SENDER=ai@yourcompany.com EMAIL_PASSWORD=your_app_password_here # 注意：不是邮箱密码，是SMTP专用密码 EMAIL_RECIPIENTS=hr@yourcompany.com,ops@yourcompany.com

创建scripts/send_email.py：

# scripts/send_email.py import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from email.mime.base import MIMEBase from email import encoders from dotenv import load_dotenv import os load_dotenv("config/email_config.env") def send_excel_report(excel_path: str, subject: str = "【SeqGPT】今日信息提取报告"): msg = MIMEMultipart() msg["From"] = os.getenv("EMAIL_SENDER") msg["To"] = os.getenv("EMAIL_RECIPIENTS") msg["Subject"] = subject body = f""" 您好， 附件为 {os.path.basename(excel_path)}，包含今日自动提取的结构化信息。 数据已同步写入MySQL表 `extracted_entities`。 —— SeqGPT-560M 自动化流水线 """ msg.attach(MIMEText(body, "plain", "utf-8")) # 附加Excel with open(excel_path, "rb") as f: part = MIMEBase("application", "vnd.openxmlformats-officedocument.spreadsheetml.sheet") part.set_payload(f.read()) encoders.encode_base64(part) part.add_header( "Content-Disposition", f"attachment; filename= {os.path.basename(excel_path)}", filename=os.path.basename(excel_path) ) msg.attach(part) # 发送 try: server = smtplib.SMTP_SSL(os.getenv("EMAIL_SMTP_SERVER"), int(os.getenv("EMAIL_SMTP_PORT"))) server.login(os.getenv("EMAIL_SENDER"), os.getenv("EMAIL_PASSWORD")) server.send_message(msg) server.quit() print(f" 邮件已发送至 {os.getenv('EMAIL_RECIPIENTS')}") except Exception as e: print(f"❌ 邮件发送失败：{e}") if __name__ == "__main__": send_excel_report("output/20240520_extract.xlsx")

4.3 设置每日定时任务（Linux Cron）

编辑定时任务：

crontab -e

添加以下行（每天上午9:05执行）：

5 9 * * * cd /path/to/your/seqgpt-project && source venv/bin/activate && python scripts/run_daily_batch.py >> logs/batch.log 2>&1

创建scripts/run_daily_batch.py：

# scripts/run_daily_batch.py import os import glob import json from datetime import datetime from scripts.save_to_mysql import save_to_mysql from scripts.export_to_excel import export_to_excel from scripts.send_email import send_excel_report def run_batch(): input_dir = "data/input" output_dir = "data/output" today = datetime.now().strftime("%Y%m%d") # 1. 查找今日待处理文件（支持 .txt .log .md） files = glob.glob(os.path.join(input_dir, f"*.{ext}") for ext in ["txt", "log", "md"]) all_files = [f for sublist in files for f in sublist] if not all_files: print(" 今日无待处理文件") return results = [] task_id = f"{today}_{datetime.now().strftime('%H%M%S')}" # 2. 逐个处理（此处调用你的实际提取函数） for file_path in all_files: try: with open(file_path, "r", encoding="utf-8") as f: text = f.read()[:5000] # 限长防OOM # 模拟调用SeqGPT（替换为你的真实调用） # extracted = seqgpt_inference(text, ["name", "company", "phone"]) extracted = {"name": "模拟姓名", "company": "模拟公司", "phone": "138****0000"} # 仅演示 # 写入MySQL save_to_mysql(extracted, task_id, os.path.basename(file_path)) # 收集结果 extracted["source_file"] = os.path.basename(file_path) results.append(extracted) except Exception as e: print(f"❌ 处理失败 {file_path}：{e}") # 3. 导出Excel excel_path = os.path.join(output_dir, f"{today}_extract.xlsx") export_to_excel(results, excel_path) # 4. 发送邮件 send_excel_report(excel_path, f"【SeqGPT】{today}信息提取日报") if __name__ == "__main__": run_batch()

验证方式：手动运行python scripts/run_daily_batch.py，检查data/output/下是否生成Excel，logs/batch.log是否有成功日志，邮箱是否收到附件。

5. 批量文本处理实战技巧（避坑指南）

5.1 如何让500份简历“一锅端”不出错？

• 分块读取：单个TXT超过1MB时，用with open(..., buffering=8192)控制内存；
• 错误隔离：每份文件try-except独立包裹，一份失败不影响其余；
• 进度可见：在Streamlit中加st.progress()，或终端打印Processing 123/500...；
• 命名规范：要求输入文件名为HR_20240520_张三_前端工程师.txt，程序可自动解析张三为name字段，减少模型负担。

5.2 提取字段不一致？试试“柔性字段映射”

SeqGPT-560M有时会把“联系电话”识别为tel，把“入职时间”识别为start_date。在入库前加一层映射：

# 字段标准化映射表 FIELD_MAPPING = { "tel": "phone", "mobile": "phone", "contact": "phone", "hire_date": "date", "start_date": "date", "joined": "date" } def normalize_keys(data: dict) -> dict: return {FIELD_MAPPING.get(k, k): v for k, v in data.items()}

5.3 MySQL写入变慢？开启批量INSERT

当单次处理超100条记录时，改用executemany：

# 替换 save_to_mysql.py 中的单条INSERT sql = "INSERT INTO extracted_entities (...) VALUES (...)" cursor.executemany(sql, all_values_list) # all_values_list 是二维列表

6. 总结：你已经拥有一条完整的信息自动化产线

回顾我们走过的路：
🔹不是调模型，而是接业务——从Streamlit界面出发，不碰训练代码，专注数据流转；
🔹不是存文件，而是建资产——每一条提取结果，自动成为MySQL里的可查、可关联、可分析的数据资产；
🔹不是手动点，而是自动跑——定时任务+邮件通知，真正实现“下班前扔进去，上班后收报表”；
🔹不是单点解，而是可扩展——Excel导出模块可轻松换成飞书多维表格API、钉钉机器人推送，甚至对接BI看板。

你不需要成为大模型专家，也能让SeqGPT-560M在你公司的内网里，日复一日稳定地产出结构化价值。

下一步，你可以：
→ 把data/input目录挂载为NAS共享文件夹，让业务部门直接拖文件进来；
→ 在MySQL上加视图，比如CREATE VIEW hr_candidates AS SELECT * FROM extracted_entities WHERE position LIKE '%算法%'；
→ 用Grafana连MySQL，做出“今日提取量趋势图”挂在部门大屏上。

技术的价值，从来不在参数多大、效果多炫，而在于——它是否真的帮你省下了那3小时的人工复制粘贴。

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。