GLM-4-9B-Chat-1M代码实例：构建CLI命令行工具支持批量处理TXT/MD/LOG文件

1. 项目概述

今天给大家分享一个实用工具开发案例：基于GLM-4-9B-Chat-1M大模型，构建一个命令行工具，能够批量处理各种文本文件。无论你是需要分析日志文件、整理Markdown文档，还是处理大量文本数据，这个工具都能帮你自动化完成。

GLM-4-9B-Chat-1M是智谱AI推出的开源大模型，最大特点是支持100万tokens的超长上下文。这意味着它可以一次性处理整本小说、大型代码库或者数百页的文档，而不会丢失上下文信息。

通过4-bit量化技术，这个9B参数的模型只需要8GB左右显存就能运行，真正实现了在消费级硬件上运行大模型的可能性。我们的CLI工具就是基于这个强大能力开发的。

2. 环境准备与安装

2.1 系统要求

在开始之前，请确保你的系统满足以下要求：

• Python 3.8或更高版本
• 至少8GB显存（推荐12GB以上）
• 10GB可用磁盘空间
• Linux或Windows系统（建议Linux）

2.2 安装依赖

创建并激活Python虚拟环境：

python -m venv glm-cli-env
source glm-cli-env/bin/activate  # Linux/Mac
# 或者
glm-cli-env\Scripts\activate  # Windows

安装所需依赖包：

pip install torch transformers bitsandbytes accelerate
pip install python-dotenv tqdm

2.3 下载模型

你可以从Hugging Face下载GLM-4-9B-Chat-1M模型：

# 使用git lfs下载模型
git lfs install
git clone https://huggingface.co/THUDM/glm-4-9b-chat-1m

或者使用transformers库自动下载：

from transformers import AutoModel, AutoTokenizer

model_name = "THUDM/glm-4-9b-chat-1m"
tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
model = AutoModel.from_pretrained(model_name, trust_remote_code=True)

3. CLI工具核心代码实现

3.1 基础工具类设计

首先创建一个基础处理类，负责加载模型和处理基本交互：

import os
import torch
from transformers import AutoModel, AutoTokenizer
from tqdm import tqdm
import argparse

class GLMTextProcessor:
    def __init__(self, model_path="THUDM/glm-4-9b-chat-1m"):
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        print(f"使用设备: {self.device}")
        
        # 加载tokenizer和模型
        self.tokenizer = AutoTokenizer.from_pretrained(
            model_path, 
            trust_remote_code=True
        )
        
        self.model = AutoModel.from_pretrained(
            model_path,
            trust_remote_code=True,
            torch_dtype=torch.float16,
            device_map="auto",
            load_in_4bit=True  # 使用4-bit量化减少显存占用
        ).eval()
        
        print("模型加载完成!")

    def process_text(self, text, prompt_template="请分析以下文本: {}"):
        """处理单段文本"""
        full_prompt = prompt_template.format(text)
        
        try:
            response, history = self.model.chat(
                self.tokenizer, 
                full_prompt, 
                history=[]
            )
            return response
        except Exception as e:
            return f"处理失败: {str(e)}"

3.2 批量文件处理功能

接下来实现批量处理文件的功能：

class BatchFileProcessor(GLMTextProcessor):
    def __init__(self, model_path="THUDM/glm-4-9b-chat-1m"):
        super().__init__(model_path)
    
    def process_file(self, file_path, output_dir, prompt_template):
        """处理单个文件"""
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                content = f.read()
            
            # 如果文件太大，分批处理
            if len(content) > 500000:  # 约50万字符
                results = self._process_large_file(content, prompt_template)
            else:
                results = self.process_text(content, prompt_template)
            
            # 保存结果
            output_path = self._get_output_path(file_path, output_dir)
            with open(output_path, 'w', encoding='utf-8') as f:
                f.write(results)
            
            return True
            
        except Exception as e:
            print(f"处理文件 {file_path} 时出错: {str(e)}")
            return False
    
    def _process_large_file(self, content, prompt_template, chunk_size=200000):
        """处理大文件，分块处理"""
        chunks = [content[i:i+chunk_size] for i in range(0, len(content), chunk_size)]
        results = []
        
        for i, chunk in enumerate(tqdm(chunks, desc="处理文本块")):
            result = self.process_text(chunk, prompt_template)
            results.append(f"=== 块 {i+1} 结果 ===\n{result}\n")
        
        return "\n".join(results)
    
    def _get_output_path(self, input_path, output_dir):
        """生成输出文件路径"""
        base_name = os.path.basename(input_path)
        name, ext = os.path.splitext(base_name)
        output_name = f"{name}_processed{ext}"
        return os.path.join(output_dir, output_name)
    
    def process_directory(self, input_dir, output_dir, file_extensions=['.txt', '.md', '.log']):
        """处理整个目录的文件"""
        if not os.path.exists(output_dir):
            os.makedirs(output_dir)
        
        processed_count = 0
        failed_count = 0
        
        # 收集所有符合条件的文件
        all_files = []
        for ext in file_extensions:
            all_files.extend([f for f in os.listdir(input_dir) if f.endswith(ext)])
        
        print(f"找到 {len(all_files)} 个待处理文件")
        
        # 处理每个文件
        for filename in tqdm(all_files, desc="处理文件"):
            file_path = os.path.join(input_dir, filename)
            success = self.process_file(file_path, output_dir, self.prompt_template)
            
            if success:
                processed_count += 1
            else:
                failed_count += 1
        
        return processed_count, failed_count

3.3 命令行界面实现

创建命令行接口，让工具更易用：

def main():
    parser = argparse.ArgumentParser(description='GLM-4-9B文本批量处理工具')
    parser.add_argument('--input', '-i', required=True, help='输入文件或目录路径')
    parser.add_argument('--output', '-o', required=True, help='输出目录路径')
    parser.add_argument('--prompt', '-p', default='请分析以下文本: {}', 
                       help='提示词模板，{}会被替换为文本内容')
    parser.add_argument('--extensions', '-e', nargs='+', 
                       default=['.txt', '.md', '.log'],
                       help='要处理的文件扩展名')
    
    args = parser.parse_args()
    
    # 初始化处理器
    print("初始化GLM模型...")
    processor = BatchFileProcessor()
    processor.prompt_template = args.prompt
    
    # 处理单个文件或整个目录
    if os.path.isfile(args.input):
        print(f"处理单个文件: {args.input}")
        success = processor.process_file(args.input, args.output, args.prompt)
        if success:
            print("文件处理完成!")
        else:
            print("文件处理失败!")
    else:
        print(f"处理目录: {args.input}")
        processed, failed = processor.process_directory(
            args.input, args.output, args.extensions
        )
        print(f"处理完成! 成功: {processed}, 失败: {failed}")

if __name__ == "__main__":
    main()

4. 实用功能扩展

4.1 预设提示词模板

为了让工具更实用，我们可以预设一些常用的提示词模板：

class PromptTemplates:
    @staticmethod
    def get_summary_template():
        return "请用中文总结以下文本的核心内容，列出3-5个主要要点: {}"
    
    @staticmethod
    def get_translation_template(target_lang="英文"):
        return f"请将以下文本翻译成{target_lang}，保持专业术语准确: {}"
    
    @staticmethod
    def get_code_analysis_template():
        return "请分析以下代码，指出可能的问题和改进建议: {}"
    
    @staticmethod
    def get_sentiment_analysis_template():
        return "请分析以下文本的情感倾向（正面/负面/中性）并说明理由: {}"
    
    @staticmethod
    def get_qa_template(question):
        return f"请基于以下文本回答问题: {question}\n\n文本内容: {{}}"

4.2 配置文件支持

添加配置文件支持，让用户可以保存常用设置：

import json
import yaml

class ConfigManager:
    def __init__(self, config_path="config.yaml"):
        self.config_path = config_path
        self.default_config = {
            "model_path": "THUDM/glm-4-9b-chat-1m",
            "default_prompt": "请分析以下文本: {}",
            "supported_extensions": [".txt", ".md", ".log", ".json"],
            "chunk_size": 200000,
            "max_retries": 3
        }
    
    def load_config(self):
        if os.path.exists(self.config_path):
            try:
                with open(self.config_path, 'r', encoding='utf-8') as f:
                    if self.config_path.endswith('.json'):
                        return json.load(f)
                    else:
                        return yaml.safe_load(f)
            except:
                print("配置文件损坏，使用默认配置")
                return self.default_config
        return self.default_config
    
    def save_config(self, config):
        try:
            with open(self.config_path, 'w', encoding='utf-8') as f:
                if self.config_path.endswith('.json'):
                    json.dump(config, f, indent=2, ensure_ascii=False)
                else:
                    yaml.safe_dump(config, f, allow_unicode=True)
            return True
        except Exception as e:
            print(f"保存配置失败: {str(e)}")
            return False

5. 完整使用示例

5.1 基本使用方法

保存上面的代码为glm_cli_tool.py，然后可以通过命令行使用：

# 处理单个文件
python glm_cli_tool.py -i input.txt -o outputs/ -p "请总结以下文本: {}"

# 处理整个目录
python glm_cli_tool.py -i ./documents/ -o ./processed/ -e .txt .md

# 使用预设模板
python glm_cli_tool.py -i code.log -o results/ -p "$(python -c "from glm_cli_tool import PromptTemplates; print(PromptTemplates.get_code_analysis_template())")"

5.2 实际应用场景

场景1：日志文件分析

# 分析服务器日志中的错误信息
python glm_cli_tool.py -i server.log -o analysis/ -p "请从以下日志中提取所有错误信息并分类: {}"

场景2：文档批量总结

# 批量总结Markdown文档
python glm_cli_tool.py -i ./markdown_files/ -o ./summaries/ -p "请用200字总结以下文档: {}"

场景3：代码审查助手

# 分析代码质量
python glm_cli_tool.py -i source_code.py -o review/ -p "请检查以下代码的质量，指出潜在问题和改进建议: {}"

5.3 高级功能：自定义处理流程

你可以扩展工具来支持更复杂的处理流程：

def advanced_processing_pipeline(processor, input_path, output_path):
    """高级处理流水线示例"""
    # 第一步：总结文本
    summary = processor.process_text(input_path, PromptTemplates.get_summary_template())
    
    # 第二步：情感分析
    sentiment = processor.process_text(input_path, PromptTemplates.get_sentiment_analysis_template())
    
    # 第三步：提取关键信息
    keywords = processor.process_text(input_path, "请从以下文本中提取5-10个关键词: {}")
    
    # 保存综合结果
    combined_result = f"""综合分析报告:
    
摘要:
{summary}

情感分析:
{sentiment}

关键词:
{keywords}
"""
    
    with open(output_path, 'w', encoding='utf-8') as f:
        f.write(combined_result)

6. 性能优化建议

6.1 内存管理

处理大文件时，注意内存使用：

def memory_efficient_processing(self, file_path, chunk_size=100000):
    """内存友好的大文件处理"""
    results = []
    
    with open(file_path, 'r', encoding='utf-8') as f:
        chunk = []
        current_size = 0
        
        for line in f:
            chunk.append(line)
            current_size += len(line)
            
            if current_size >= chunk_size:
                text_chunk = ''.join(chunk)
                result = self.process_text(text_chunk)
                results.append(result)
                
                # 重置
                chunk = []
                current_size = 0
        
        # 处理最后一块
        if chunk:
            text_chunk = ''.join(chunk)
            result = self.process_text(text_chunk)
            results.append(result)
    
    return "\n".join(results)

6.2 批量处理优化

使用多线程加速批量处理：

from concurrent.futures import ThreadPoolExecutor, as_completed

def parallel_process_directory(self, input_dir, output_dir, max_workers=2):
    """并行处理目录中的文件"""
    files = [f for f in os.listdir(input_dir) 
             if any(f.endswith(ext) for ext in self.supported_extensions)]
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {}
        for file in files:
            input_path = os.path.join(input_dir, file)
            future = executor.submit(self.process_file, input_path, output_dir)
            futures[future] = file
        
        # 等待所有任务完成
        for future in tqdm(as_completed(futures), total=len(futures), desc="并行处理"):
            file = futures[future]
            try:
                future.result()
            except Exception as e:
                print(f"处理文件 {file} 时出错: {str(e)}")

7. 总结

通过这个GLM-4-9B-Chat-1M命令行工具，我们实现了一个强大的文本处理解决方案。这个工具的优势在于：

• 超长上下文处理：能够处理长达100万tokens的文本，适合分析大型文档
• 批量处理能力：支持一次性处理整个目录的文件，提高工作效率
• 灵活可配置：通过提示词模板支持各种处理任务
• 本地化部署：所有数据处理都在本地完成，保证数据安全
• 资源高效：4-bit量化技术让大模型在消费级硬件上运行成为可能

无论是处理技术文档、分析日志文件，还是进行文本总结和翻译，这个工具都能提供强大的支持。你可以根据自己的需求进一步扩展功能，比如添加更多的预设模板、支持更多文件格式，或者集成到现有的工作流程中。

---

获取更多AI镜像

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