ChatGLM3-6B部署教程Airflow定时任务触发模型批量文档处理1. 项目概述今天给大家分享一个实用方案如何将强大的ChatGLM3-6B模型与Airflow调度系统结合实现自动化批量文档处理。这个方案特别适合需要定期处理大量文档的企业场景比如每日报告生成、周报分析、月度数据汇总等。传统的AI模型调用往往需要手动操作效率低下且容易出错。通过本教程你将学会搭建一个完全自动化的智能文档处理流水线让AI模型按计划自动工作彻底解放人力。核心价值自动化处理设定好时间模型自动运行无需人工干预批量处理能力一次性处理成百上千份文档效率提升显著私有化部署所有数据在本地处理确保商业机密安全稳定可靠基于成熟的开源组件长期运行无忧2. 环境准备与依赖安装2.1 系统要求在开始部署前请确保你的服务器满足以下要求操作系统Ubuntu 20.04 或 CentOS 7GPUNVIDIA RTX 4090D 或同等级别显卡至少24GB显存内存32GB RAM 或更高存储至少50GB可用空间用于模型文件和文档存储2.2 基础环境配置首先安装必要的系统依赖# 更新系统包 sudo apt update sudo apt upgrade -y # 安装Python环境 sudo apt install python3.9 python3.9-venv python3.9-dev -y # 安装CUDA工具包如果尚未安装 wget https://developer.download.nvidia.com/compute/cuda/11.8.0/local_installers/cuda_11.8.0_520.61.05_linux.run sudo sh cuda_11.8.0_520.61.05_linux.run # 配置Python虚拟环境 python3.9 -m venv ~/glm3-airflow-env source ~/glm3-airflow-env/bin/activate2.3 Python依赖安装创建requirements.txt文件并安装依赖# requirements.txt transformers4.40.2 torch2.1.2cu118 streamlit1.28.1 apache-airflow2.7.1 cryptography41.0.4 pandas2.0.3 python-dotenv1.0.0 accelerate0.24.1 sentencepiece0.1.99 protobuf3.20.3安装命令pip install -r requirements.txt3. ChatGLM3-6B模型部署3.1 模型下载与配置下载ChatGLM3-6B-32k模型文件from transformers import AutoModel, AutoTokenizer model_path THUDM/chatglm3-6b-32k local_path /opt/models/chatglm3-6b-32k # 下载模型如果尚未下载 tokenizer AutoTokenizer.from_pretrained(model_path, trust_remote_codeTrue) model AutoModel.from_pretrained( model_path, trust_remote_codeTrue, device_mapauto, torch_dtypetorch.float16 ) # 保存到本地 model.save_pretrained(local_path) tokenizer.save_pretrained(local_path)3.2 创建模型推理服务建立简单的模型调用接口# model_service.py import torch from transformers import AutoModel, AutoTokenizer import streamlit as st st.cache_resource def load_model(): 加载模型并缓存避免重复加载 model_path /opt/models/chatglm3-6b-32k tokenizer AutoTokenizer.from_pretrained( model_path, trust_remote_codeTrue ) model AutoModel.from_pretrained( model_path, trust_remote_codeTrue, device_mapauto, torch_dtypetorch.float16 ).eval() return model, tokenizer def process_document(document_text, max_length8192): 处理单个文档 model, tokenizer load_model() # 构建提示词 prompt f请分析以下文档内容并生成摘要\n\n{document_text} # 生成响应 response, history model.chat( tokenizer, prompt, history[], max_lengthmax_length, temperature0.7 ) return response4. Airflow环境搭建4.1 Airflow初始化配置Airflow工作环境# 设置Airflow家目录 export AIRFLOW_HOME~/airflow # 初始化数据库 airflow db init # 创建管理员用户 airflow users create \ --username admin \ --firstname Admin \ --lastname User \ --role Admin \ --email adminexample.com \ --password admin123 # 启动Web服务器 airflow webserver --port 8080 -D # 启动调度器 airflow scheduler -D4.2 配置Airflow连接创建模型服务连接# 在Airflow UI中添加连接 # Connection ID: chatglm3_service # Connection Type: HTTP # Host: http://localhost:8501 # Extra: {Content-Type: application/json}5. 构建文档处理DAG5.1 创建批量处理DAG# ~/airflow/dags/document_processing_dag.py from datetime import datetime, timedelta from airflow import DAG from airflow.operators.python import PythonOperator from airflow.providers.http.operators.http import SimpleHttpOperator import pandas as pd import os default_args { owner: document_team, depends_on_past: False, email_on_failure: True, email: [adminexample.com], retries: 3, retry_delay: timedelta(minutes5) } def process_documents(**kwargs): 处理文档目录中的所有文件 input_dir /data/documents/input output_dir /data/documents/processed # 确保输出目录存在 os.makedirs(output_dir, exist_okTrue) processed_files [] for filename in os.listdir(input_dir): if filename.endswith(.txt): filepath os.path.join(input_dir, filename) try: # 读取文档内容 with open(filepath, r, encodingutf-8) as f: content f.read() # 调用模型处理这里简化了实际调用 summary process_with_chatglm3(content) # 保存处理结果 output_path os.path.join(output_dir, fprocessed_{filename}) with open(output_path, w, encodingutf-8) as f: f.write(summary) processed_files.append(filename) except Exception as e: print(f处理文件 {filename} 时出错: {str(e)}) return processed_files def process_with_chatglm3(content): 调用ChatGLM3模型处理内容 # 这里是简化的示例实际需要实现模型调用逻辑 # 可以使用HTTP请求或直接导入模型模块 return f摘要已处理内容长度{len(content)} 字符 # 定义DAG dag DAG( daily_document_processing, default_argsdefault_args, description每日文档批量处理任务, schedule_interval0 2 * * *, # 每天凌晨2点运行 start_datedatetime(2024, 1, 1), catchupFalse, tags[document, chatglm3, batch_processing] ) # 定义任务 process_task PythonOperator( task_idprocess_documents, python_callableprocess_documents, dagdag, ) # 可以添加更多任务比如结果通知、清理等5.2 高级功能动态文档发现增强版文档处理函数支持多种文件格式def enhanced_document_processor(**kwargs): 增强版文档处理器 supported_extensions [.txt, .pdf, .docx, .md] input_dir /data/documents/input for filename in os.listdir(input_dir): file_ext os.path.splitext(filename)[1].lower() if file_ext in supported_extensions: filepath os.path.join(input_dir, filename) process_single_file(filepath, file_ext) def process_single_file(filepath, file_ext): 处理单个文件 try: if file_ext .txt: content read_text_file(filepath) elif file_ext .pdf: content read_pdf_file(filepath) elif file_ext .docx: content read_docx_file(filepath) else: content read_text_file(filepath) # 调用模型处理 result process_with_chatglm3(content) # 保存结果 save_processing_result(filepath, result) except Exception as e: log_processing_error(filepath, str(e)) def read_pdf_file(filepath): 读取PDF文件内容 # 需要安装PyPDF2: pip install PyPDF2 import PyPDF2 content with open(filepath, rb) as f: reader PyPDF2.PdfReader(f) for page in reader.pages: content page.extract_text() \n return content6. 实战演示配置定时任务6.1 创建多种调度策略根据业务需求可以配置不同的调度频率# 多个DAG示例按小时、天、周调度 hourly_dag DAG( hourly_document_processing, schedule_interval0 * * * *, # 每小时 # ...其他参数 ) daily_dag DAG( daily_document_processing, schedule_interval0 2 * * *, # 每天凌晨2点 # ...其他参数 ) weekly_dag DAG( weekly_report_processing, schedule_interval0 3 * * 0, # 每周日凌晨3点 # ...其他参数 )6.2 监控与告警配置设置任务监控和失败告警# 在DAG中添加回调函数 def alert_on_failure(context): 任务失败时发送告警 task_instance context[task_instance] error_message context[exception] # 发送邮件实际项目中可集成短信、钉钉等 send_email( toadminexample.com, subjectfAirflow任务失败: {task_instance.task_id}, contentf任务执行失败错误信息: {error_message} ) # 在任务中配置失败回调 process_task PythonOperator( task_idprocess_documents, python_callableprocess_documents, on_failure_callbackalert_on_failure, dagdag, )7. 性能优化与最佳实践7.1 模型加载优化使用Singleton模式确保模型只加载一次class ModelSingleton: _instance None _model None _tokenizer None classmethod def get_model(cls): if cls._instance is None: cls._instance cls() return cls._model, cls._tokenizer def __init__(self): if self._model is None: model_path /opt/models/chatglm3-6b-32k self._tokenizer AutoTokenizer.from_pretrained( model_path, trust_remote_codeTrue ) self._model AutoModel.from_pretrained( model_path, trust_remote_codeTrue, device_mapauto, torch_dtypetorch.float16 ).eval()7.2 批量处理优化实现文档批量处理减少模型调用次数def process_documents_batch(documents, batch_size5): 批量处理文档提高效率 results [] for i in range(0, len(documents), batch_size): batch documents[i:ibatch_size] batch_results process_batch_with_model(batch) results.extend(batch_results) return results def process_batch_with_model(documents_batch): 使用模型处理批量文档 # 将多个文档组合成一个提示词 combined_prompt 请分析以下多个文档并生成摘要\n\n for i, doc in enumerate(documents_batch): combined_prompt f文档{i1}:\n{doc}\n\n # 调用模型 response call_model(combined_prompt) # 解析批量结果需要根据模型输出格式调整 return parse_batch_response(response, len(documents_batch))8. 常见问题与解决方案8.1 内存管理问题问题处理大量文档时内存不足解决方案def process_large_documents(filepath): 处理大文件的内存友好方法 chunk_size 1000 # 每次处理1000个字符 with open(filepath, r, encodingutf-8) as f: while True: chunk f.read(chunk_size) if not chunk: break # 处理当前块 process_chunk(chunk) # 手动释放内存 del chunk import gc gc.collect()8.2 模型响应超时问题模型处理时间过长导致任务超时解决方案# 设置超时机制 import signal from contextlib import contextmanager class TimeoutException(Exception): pass contextmanager def time_limit(seconds): 执行时间限制 def signal_handler(signum, frame): raise TimeoutException(处理超时) signal.signal(signal.SIGALRM, signal_handler) signal.alarm(seconds) try: yield finally: signal.alarm(0) # 使用示例 try: with time_limit(300): # 5分钟超时 result process_with_chatglm3(content) except TimeoutException: result 处理超时请重试或减少内容长度9. 总结通过本教程我们成功搭建了一个基于ChatGLM3-6B和Airflow的自动化文档处理系统。这个方案具有以下优势核心价值完全自动化设定好调度策略后系统会自动处理文档无需人工干预处理能力强支持批量处理大量文档显著提升工作效率灵活可扩展可以根据业务需求调整处理逻辑和调度频率企业级稳定基于成熟的开源组件确保长期稳定运行实际应用场景企业每日报告自动生成与摘要周报月报的智能分析与汇总客户反馈的定期处理与分析技术文档的自动化整理与分类下一步建议根据实际业务需求调整文档处理逻辑配置更详细的监控和告警机制考虑添加文档处理结果的可视化展示探索更多AI模型的应用场景这个方案不仅适用于文档处理还可以扩展到其他需要定期AI处理的业务场景为企业智能化转型提供有力支撑。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。