← 返回首页
🧠

LLM文件系统集成

📂 llm ⏱ 5 min 813 words

--- title: "LLM文件系统集成" description: "探讨大语言模型与文件系统的深度集成,包括文档处理、文件管理、自动分类和智能搜索" tags: ["文件系统", "文档处理", "文件管理", "自动分类", "LLM应用"] category: "llm" icon: "🧠"

LLM文件系统集成

文件系统是数据存储和管理的基础,大语言模型(LLM)正在改变我们处理文件和文档的方式。通过智能化的文档处理、自动分类、内容提取和语义搜索,LLM使文件管理变得更加高效和智能。

智能文档处理

LLM能够理解和处理各种格式的文档,从PDF、Word到电子邮件和网页内容。通过自然语言理解,LLM可以提取关键信息、生成摘要并回答文档相关的问题。

import os
from pathlib import Path
from typing import List, Dict, Any

class IntelligentDocumentProcessor:
    """智能文档处理器"""
    
    def __init__(self, llm_client):
        self.llm_client = llm_client
        self.supported_formats = ['.txt', '.md', '.pdf', '.docx', '.eml']
    
    def process_document(self, file_path: str) -> Dict[str, Any]:
        """处理文档并提取信息"""
        file_path = Path(file_path)
        
        if not file_path.exists():
            raise FileNotFoundError(f"文件不存在: {file_path}")
        
        # 读取文件内容
        content = self._read_file(file_path)
        
        # 提取文档元数据
        metadata = self._extract_metadata(content, file_path)
        
        # 分析文档内容
        analysis = self._analyze_content(content)
        
        return {
            "file_path": str(file_path),
            "metadata": metadata,
            "content_preview": content[:500] + "..." if len(content) > 500 else content,
            "analysis": analysis
        }
    
    def _read_file(self, file_path: Path) -> str:
        """读取文件内容"""
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                return f.read()
        except UnicodeDecodeError:
            # 尝试其他编码
            with open(file_path, 'r', encoding='gbk') as f:
                return f.read()
    
    def _extract_metadata(self, content: str, file_path: Path) -> Dict[str, Any]:
        """提取文档元数据"""
        prompt = f"""分析以下文档并提取元数据:

文件名:{file_path.name}
文件大小:{file_path.stat().st_size} bytes
内容预览:
{content[:1000]}

请提取:
1. 文档类型(报告、邮件、代码、配置文件等)
2. 主要主题
3. 关键实体(人名、组织、日期)
4. 语言
5. 摘要(100字以内)
"""
        
        metadata_text = self.llm_client.generate(prompt, temperature=0.2)
        return {"raw_analysis": metadata_text}
    
    def _analyze_content(self, content: str) -> Dict[str, Any]:
        """分析文档内容"""
        prompt = f"""深入分析以下文档内容:

{content[:2000]}

请提供:
1. 结构分析(章节、段落)
2. 关键观点和论点
3. 支持证据和数据
4. 结论和建议
5. 相关主题和关键词
"""
        
        analysis_text = self.llm_client.generate(prompt, temperature=0.3)
        return {"detailed_analysis": analysis_text}
    
    def batch_process(self, directory: str, pattern: str = "*") -> List[Dict]:
        """批量处理目录中的文档"""
        results = []
        path = Path(directory)
        
        for file_path in path.glob(pattern):
            if file_path.suffix.lower() in self.supported_formats:
                try:
                    result = self.process_document(str(file_path))
                    results.append(result)
                except Exception as e:
                    print(f"处理文件 {file_path} 失败: {e}")
        
        return results

# 使用示例
processor = IntelligentDocumentProcessor(llm_client)

# 处理单个文档
result = processor.process_document("report.pdf")
print(f"文档类型: {result['metadata']['raw_analysis'][:100]}...")

# 批量处理
results = processor.batch_process("./documents", "*.pdf")

智能文档处理的关键优势在于能够处理非结构化文档,提取其中的关键信息,并生成结构化的表示。这对于构建文档知识库和实现智能搜索至关重要。

自动文件分类

LLM能够根据文件内容自动分类,无需预先定义规则或训练模型。

class SmartFileClassifier:
    """智能文件分类器"""
    
    def __init__(self, llm_client):
        self.llm_client = llm_client
        self.categories = {
            "财务": ["发票", "报表", "预算", "费用"],
            "法律": ["合同", "协议", "条款", "法律"],
            "技术": ["代码", "文档", "规范", "API"],
            "人事": ["简历", "评估", "培训", "绩效"],
            "营销": ["方案", "分析", "报告", "推广"]
        }
    
    def classify_file(self, file_path: str) -> Dict[str, Any]:
        """分类单个文件"""
        processor = IntelligentDocumentProcessor(self.llm_client)
        doc_info = processor.process_document(file_path)
        
        prompt = f"""根据以下信息对文件进行分类:

文件名:{Path(file_path).name}
文档分析:
{doc_info['analysis']['detailed_analysis'][:500]}

可用分类:
{list(self.categories.keys())}

请提供:
1. 主分类
2. 置信度(0-1)
3. 次要分类(如果有)
4. 分类理由
5. 建议的标签
"""
        
        classification = self.llm_client.generate(prompt, temperature=0.2)
        
        return {
            "file_path": file_path,
            "classification": classification,
            "metadata": doc_info["metadata"]
        }
    
    def classify_batch(self, file_paths: List[str]) -> List[Dict]:
        """批量分类文件"""
        results = []
        for file_path in file_paths:
            try:
                result = self.classify_file(file_path)
                results.append(result)
            except Exception as e:
                print(f"分类文件 {file_path} 失败: {e}")
        return results
    
    def suggest_taxonomy(self, file_samples: List[str]) -> str:
        """建议文件分类体系"""
        sample_info = []
        for file_path in file_samples[:10]:  # 限制样本数量
            try:
                processor = IntelligentDocumentProcessor(self.llm_client)
                doc_info = processor.process_document(file_path)
                sample_info.append({
                    "file": Path(file_path).name,
                    "analysis": doc_info["analysis"]["detailed_analysis"][:200]
                })
            except:
                continue
        
        prompt = f"""基于以下文件样本,建议一个文件分类体系:

文件样本:
{json.dumps(sample_info, indent=2, ensure_ascii=False)}

请提供:
1. 推荐的分类层级(主分类、子分类)
2. 每个分类的定义和示例
3. 分类规则和指导原则
4. 实施建议
"""
        
        return self.llm_client.generate(prompt, temperature=0.4)

# 使用示例
classifier = SmartFileClassifier(llm_client)

# 分类单个文件
result = classifier.classify_file("contract.pdf")
print(f"分类结果: {result['classification'][:200]}...")

# 批量分类
file_list = ["report.pdf", "invoice.pdf", "code.py"]
results = classifier.classify_batch(file_list)

自动文件分类的关键优势在于能够根据内容而非仅仅文件名进行分类,适应不断变化的文档类型,并提供分类理由和置信度。

智能文件搜索

传统文件搜索基于文件名和简单关键词匹配,LLM使语义搜索成为可能。

class SemanticFileSearch:
    """语义文件搜索"""
    
    def __init__(self, llm_client, vector_db):
        self.llm_client = llm_client
        self.vector_db = vector_db
    
    def index_directory(self, directory: str):
        """为目录建立索引"""
        processor = IntelligentDocumentProcessor(self.llm_client)
        files = processor.batch_process(directory)
        
        for file_info in files:
            # 生成向量表示
            embedding = self._generate_embedding(file_info)
            
            # 存储到向量数据库
            self.vector_db.insert(
                embedding=embedding,
                metadata={
                    "file_path": file_info["file_path"],
                    "metadata": file_info["metadata"],
                    "content_preview": file_info["content_preview"]
                }
            )
    
    def search(self, query: str, top_k: int = 5) -> List[Dict]:
        """语义搜索文件"""
        # 查询向量化
        query_embedding = self._generate_embedding(query)
        
        # 向量相似度搜索
        results = self.vector_db.search(query_embedding, top_k=top_k)
        
        # 使用LLM重新排序和过滤
        reranked_results = self._rerank_results(query, results)
        
        return reranked_results
    
    def _generate_embedding(self, content) -> list:
        """生成向量表示"""
        if isinstance(content, dict):
            text = f"{content['metadata']}\n{content['content_preview']}"
        else:
            text = content
        
        # 使用LLM生成嵌入
        prompt = f"""将以下内容转换为向量表示:

{text[:1000]}

请提供:
1. 主题关键词
2. 情感倾向
3. 实体识别
4. 语义摘要
"""
        
        # 实际应用中应使用专门的嵌入模型
        return self.llm_client.generate(prompt, temperature=0.1)
    
    def _rerank_results(self, query: str, results: List[Dict]) -> List[Dict]:
        """重新排序搜索结果"""
        if not results:
            return []
        
        prompt = f"""根据查询对搜索结果进行重新排序:

查询:{query}

搜索结果:
{json.dumps(results[:10], indent=2, ensure_ascii=False)}

请评估每个结果的相关性,并按相关性排序。
返回排序后的结果列表。
"""
        
        reranked = self.llm_client.generate(prompt, temperature=0.2)
        
        # 解析LLM返回的排序结果
        # 实际实现中需要解析LLM输出
        return results  # 简化示例
    
    def contextual_search(self, query: str, context: Dict[str, Any]) -> List[Dict]:
        """上下文感知搜索"""
        context_str = json.dumps(context, ensure_ascii=False)
        
        prompt = f"""基于以下上下文进行文件搜索:

用户查询:{query}
上下文信息:{context_str}

请生成:
1. 优化的搜索查询
2. 相关的搜索过滤器
3. 预期的搜索结果类型
"""
        
        enhanced_query = self.llm_client.generate(prompt, temperature=0.3)
        
        return self.search(enhanced_query)

# 使用示例
search_engine = SemanticFileSearch(llm_client, vector_db)
search_engine.index_directory("./documents")

results = search_engine.search("最近的项目进度报告")
for result in results:
    print(f"文件: {result['file_path']}")
    print(f"相关性: {result.get('relevance_score', 'N/A')}")

智能文件搜索的关键优势在于理解查询意图,处理同义词和相关概念,并根据上下文提供更相关的搜索结果。

文件管理自动化

LLM可以自动化文件管理任务,如文件整理、备份和清理。

class AutomatedFileManager:
    """自动化文件管理器"""
    
    def __init__(self, llm_client):
        self.llm_client = llm_client
    
    def organize_files(self, source_dir: str, target_dir: str) -> Dict[str, Any]:
        """自动整理文件"""
        processor = IntelligentDocumentProcessor(self.llm_client)
        classifier = SmartFileClassifier(self.llm_client)
        
        # 扫描源目录
        files = processor.batch_process(source_dir)
        
        # 分类文件
        classified_files = []
        for file_info in files:
            classification = classifier.classify_file(file_info["file_path"])
            classified_files.append({
                **file_info,
                "classification": classification
            })
        
        # 生成整理方案
        organization_plan = self._generate_organization_plan(classified_files, target_dir)
        
        # 执行整理(可选)
        # self._execute_organization_plan(organization_plan)
        
        return {
            "total_files": len(classified_files),
            "organization_plan": organization_plan,
            "categories": self._count_categories(classified_files)
        }
    
    def _generate_organization_plan(self, files: List[Dict], target_dir: str) -> str:
        """生成文件整理方案"""
        file_summary = []
        for file in files[:20]:  # 限制数量
            file_summary.append({
                "file": Path(file["file_path"]).name,
                "classification": file["classification"][:100]
            })
        
        prompt = f"""为以下文件生成整理方案:

目标目录:{target_dir}

文件列表:
{json.dumps(file_summary, indent=2, ensure_ascii=False)}

请提供:
1. 推荐的目录结构
2. 文件移动规则
3. 命名规范建议
4. 重复文件处理策略
5. 实施步骤
"""
        
        return self.llm_client.generate(prompt, temperature=0.3)
    
    def cleanup_old_files(self, directory: str, 
                         days_old: int = 365) -> Dict[str, Any]:
        """清理旧文件"""
        processor = IntelligentDocumentProcessor(self.llm_client)
        files = processor.batch_process(directory)
        
        old_files = []
        current_time = datetime.now()
        
        for file_info in files:
            file_path = Path(file_info["file_path"])
            file_age = current_time - datetime.fromtimestamp(file_path.stat().st_mtime)
            
            if file_age.days > days_old:
                old_files.append({
                    "file": str(file_path),
                    "age_days": file_age.days,
                    "size": file_path.stat().st_size,
                    "analysis": file_info["analysis"]
                })
        
        # 生成清理建议
        if old_files:
            prompt = f"""分析以下旧文件并提供清理建议:

{json.dumps(old_files[:10], indent=2, ensure_ascii=False)}

请提供:
1. 建议删除的文件(附理由)
2. 建议归档的文件
3. 建议保留的文件
4. 风险评估
5. 备份建议
"""
            
            cleanup_advice = self.llm_client.generate(prompt, temperature=0.3)
        else:
            cleanup_advice = "没有发现需要清理的旧文件"
        
        return {
            "total_old_files": len(old_files),
            "cleanup_advice": cleanup_advice,
            "old_files": old_files
        }

# 使用示例
manager = AutomatedFileManager(llm_client)
result = manager.organize_files("./messy_files", "./organized_files")
print(f"整理了 {result['total_files']} 个文件")

文件管理自动化的关键优势在于减少手动工作量,确保文件组织的一致性,并提供智能化的决策支持。结合LLM的自然语言能力,文件管理系统可以更好地理解用户需求,提供个性化的管理方案。