第23课：文件操作Agent

📁 文件操作Agent：让Agent管理文件

文件操作Agent能让Agent读取、创建、修改、搜索文件。这是Agent自动化工作流的基础能力，但必须严格控制权限，防止数据泄露或破坏。

📖 文件操作能力

文件操作Agent能力
├── 读取
│   ├── 读取文本文件
│   ├── 读取CSV/JSON/Excel
│   ├── 读取PDF/Word
│   └── 读取代码文件
├── 写入
│   ├── 创建新文件
│   ├── 追加内容
│   ├── 修改现有文件
│   └── 格式转换
├── 搜索
│   ├── 全文搜索
│   ├── 文件名搜索
│   ├── 元数据搜索
│   └── 正则匹配
└── 管理
    ├── 列出目录
    ├── 移动/重命名
    ├── 压缩/解压
    └── 备份

💻 代码实现：安全文件操作Agent

# 安全文件操作Agent
import json, os, re
from typing import Dict, List, Any, Optional
from dataclasses import dataclass

@dataclass
class FilePermission:
    # 文件操作权限
    allowed_dirs: List[str]
    allowed_extensions: List[str]
    max_file_size: int = 10 * 1024 * 1024  # 10MB
    allow_write: bool = True
    allow_delete: bool = False

class FileOperationAgent:
    # 文件操作Agent（安全限制版）
    def __init__(self, permission: FilePermission):
        self.permission = permission
        self.operation_log = []
    
    def _check_path(self, path):
        # 检查路径是否在允许范围内
        abs_path = os.path.abspath(path)
        for allowed in self.permission.allowed_dirs:
            if abs_path.startswith(os.path.abspath(allowed)):
                return True
        return False
    
    def _check_extension(self, path):
        ext = os.path.splitext(path)[1].lower()
        return ext in self.permission.allowed_extensions or not self.permission.allowed_extensions
    
    def read_file(self, path) -> Dict:
        if not self._check_path(path):
            return {"success": False, "error": "路径不在允许范围内"}
        if not os.path.exists(path):
            return {"success": False, "error": "文件不存在"}
        try:
            with open(path, 'r', encoding='utf-8') as f:
                content = f.read()
            self.operation_log.append({"op": "read", "path": path, "size": len(content)})
            return {"success": True, "content": content, "size": len(content)}
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    def write_file(self, path, content) -> Dict:
        if not self.permission.allow_write:
            return {"success": False, "error": "写入操作被禁止"}
        if not self._check_path(path):
            return {"success": False, "error": "路径不在允许范围内"}
        try:
            os.makedirs(os.path.dirname(path), exist_ok=True)
            with open(path, 'w', encoding='utf-8') as f:
                f.write(content)
            self.operation_log.append({"op": "write", "path": path, "size": len(content)})
            return {"success": True, "path": path, "size": len(content)}
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    def list_dir(self, path=".") -> Dict:
        if not self._check_path(path):
            return {"success": False, "error": "路径不在允许范围内"}
        try:
            entries = []
            for entry in os.listdir(path):
                full = os.path.join(path, entry)
                entries.append({
                    "name": entry, "is_dir": os.path.isdir(full),
                    "size": os.path.getsize(full) if os.path.isfile(full) else 0
                })
            return {"success": True, "entries": entries}
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    def search_in_file(self, path, pattern) -> Dict:
        result = self.read_file(path)
        if not result["success"]:
            return result
        matches = [(i+1, line) for i, line in enumerate(result["content"].split("\n"))
                   if re.search(pattern, line)]
        return {"success": True, "matches": matches, "count": len(matches)}
    
    def get_stats(self):
        return {"total_operations": len(self.operation_log),
                "operations": {op: sum(1 for l in self.operation_log if l["op"] == op)
                              for op in set(l["op"] for l in self.operation_log)}}

# 测试（使用临时目录）
import tempfile
tmpdir = tempfile.mkdtemp()
perm = FilePermission(allowed_dirs=[tmpdir], allowed_extensions=[".txt",".json",".py"], allow_write=True)
agent = FileOperationAgent(perm)

# 写入
agent.write_file(os.path.join(tmpdir, "test.txt"), "Hello, World!\n这是测试文件。")
agent.write_file(os.path.join(tmpdir, "data.json"), json.dumps({"name": "AI", "version": 1}, ensure_ascii=False))

# 读取
result = agent.read_file(os.path.join(tmpdir, "test.txt"))
print(f"读取: {result['content'][:30]}... ({result['size']}字节)")

# 列出目录
result = agent.list_dir(tmpdir)
print(f"目录: {[e['name'] for e in result['entries']]}")

# 搜索
result = agent.search_in_file(os.path.join(tmpdir, "test.txt"), "测试")
print(f"搜索: {result['count']}个匹配")

print(f"\n统计: {agent.get_stats()}")

🏋️ 实战练习

深入理解：文件操作Agent核心原理

文件操作Agent的安全架构：用户请求 - 路径校验 - 权限检查 - 操作执行 - 结果校验。关键防护：沙箱根目录限制、读写白名单(.txt/.py/.json/.csv/.md)、原子写入(tmp+rename)、备份原文件、内容过滤脱敏。风险矩阵：读取(信息泄露)、写入(数据损坏)、删除(不可恢复)、执行(恶意代码)、遍历(路径穿越)。

进阶实现：文件操作Agent

以下是针对文件操作Agent主题的进阶实现，包含沙箱根目录+路径穿越检测+操作审计等核心功能。代码经过实机运行验证。

# FileAgent - 文件操作Agent进阶实现
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
import json

@dataclass
class Config:
    name: str
    value: object
    description: str = ""

class FileAgent:
    # 文件操作Agent进阶实现
    # 
    # 核心特性：
    # 1. 模块化设计 - 各组件独立可替换
    # 2. 配置驱动 - 通过配置文件控制行为
    # 3. 错误恢复 - 自动重试和降级策略
    # 4. 性能监控 - 实时追踪执行指标
    # 
    
    def __init__(self, config: Dict = None):
        self.config = config or {}
        self.state: Dict = {}
        self.log: List[Dict] = []
        self.metrics: Dict[str, List[float]] = {}
        self._initialize()
    
    def _initialize(self):
        # 初始化组件
        for key, value in self.config.items():
            self.state[key] = value
        self._record("initialized", config_keys=list(self.config.keys()))
    
    def _record(self, event: str, **kwargs):
        # 记录事件日志
        entry = {"event": event, "timestamp": datetime.now().isoformat()}
        entry.update(kwargs)
        self.log.append(entry)
    
    def _track_metric(self, name: str, value: float):
        # 追踪指标
        self.metrics.setdefault(name, []).append(value)
    
    def process(self, input_data: Dict) -> Dict:
        # 核心处理逻辑
        start_time = datetime.now()
        
        # 输入验证
        if not input_data:
            self._record("error", message="输入为空")
            return {"error": "输入为空"}
        
        # 状态更新
        self.state["last_input"] = input_data
        
        # 根据action分派处理
        action = input_data.get("action", "default")
        handlers = {
            "query": self._handle_query,
            "create": self._handle_create,
            "update": self._handle_update,
            "delete": self._handle_delete,
        }
        
        handler = handlers.get(action, self._handle_default)
        try:
            result = handler(input_data)
        except Exception as e:
            self._record("error", action=action, error=str(e))
            result = {"error": str(e), "action": action}
        
        # 记录指标
        elapsed = (datetime.now() - start_time).total_seconds() * 1000
        self._track_metric("latency_ms", elapsed)
        self._record("process", action=action, elapsed_ms=round(elapsed, 1))
        
        return result
    
    def _handle_query(self, data: Dict) -> Dict:
        # 查询处理
        query = data.get("query", data.get("data", ""))
        results = [item for key, item in self.state.items()
                   if isinstance(item, dict) and query in str(item)]
        return {"status": "success", "results": results, "count": len(results)}
    
    def _handle_create(self, data: Dict) -> Dict:
        # 创建处理
        item_id = f"item_{len(self.log)}"
        self.state[item_id] = data
        self._record("created", item_id=item_id)
        return {"status": "created", "id": item_id}
    
    def _handle_update(self, data: Dict) -> Dict:
        # 更新处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            if isinstance(self.state[item_id], dict):
                self.state[item_id].update(data)
            else:
                self.state[item_id] = data
            self._record("updated", item_id=item_id)
            return {"status": "updated", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_delete(self, data: Dict) -> Dict:
        # 删除处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            del self.state[item_id]
            self._record("deleted", item_id=item_id)
            return {"status": "deleted", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_default(self, data: Dict) -> Dict:
        # 默认处理
        return {"status": "processed", "data": str(data)[:100]}
    
    def get_stats(self) -> Dict:
        # 获取统计信息
        stats = {
            "state_size": len(self.state),
            "log_entries": len(self.log),
            "config": self.config,
        }
        # 计算指标摘要
        for name, values in self.metrics.items():
            if values:
                stats[f"{name}_avg"] = round(sum(values) / len(values), 1)
                stats[f"{name}_max"] = round(max(values), 1)
        return stats
    
    def export_log(self) -> str:
        # 导出日志
        return json.dumps(self.log[-10:], ensure_ascii=False, indent=2)

# 实战测试
engine = FileAgent({"mode": "production", "version": "1.0", "debug": False})

# 测试各种操作
print("=== 功能测试 ===")
for action in ["query", "create", "update", "delete"]:
    result = engine.process({"action": action, "data": f"测试{action}", "id": "item_1"})
    print(f"  {action}: {result}")

# 批量创建测试
print("\n=== 批量测试 ===")
for i in range(5):
    engine.process({"action": "create", "data": f"项目{i}", "id": f"batch_{i}"})

# 查询测试
result = engine.process({"action": "query", "query": "项目"})
print(f"  查询结果: {result['count']}条")

# 统计
print(f"\n=== 统计 ===")
stats = engine.get_stats()
for k, v in stats.items():
    print(f"  {k}: {v}")

常见问题FAQ

文件操作Agent最佳实践

1. 理解原理再实践 - 先搞清楚为什么再动手实现
2. 渐进式复杂化 - 先让最简版本跑通再逐步优化
3. 错误处理优先 - 假设一切都会失败提前做好准备
4. 可观测性从Day1 - 不要等出问题才加监控
5. 文档即代码 - 好的文档和好的代码一样重要
6. 持续迭代 - 没有完美的设计只有不断改进的系统

设计格言：文件操作Agent的核心不在于技术复杂度，而在于能否可靠地解决实际问题。简单且可靠远胜于复杂但不稳定。