自动化脚本开发

构建一个能够理解自然语言指令，自动规划和执行任务的Agent系统。

项目目标

输入：自然语言任务描述
处理：任务规划 → 步骤分解 → 自动执行
输出：执行结果报告

示例任务：
- "帮我整理下载文件夹，按类型分类"
- "每天早上发送天气报告到邮箱"
- "监控网站变化并发送通知"

技术架构

┌─────────────────────────────────────────────────┐
│                   用户指令                       │
└─────────────────────┬───────────────────────────┘
                      ↓
┌─────────────────────────────────────────────────┐
│              规划器 (Planner)                    │
│  分析任务 → 分解步骤 → 生成执行计划               │
└─────────────────────┬───────────────────────────┘
                      ↓
┌─────────────────────────────────────────────────┐
│              执行器 (Executor)                   │
│  调用工具 → 执行步骤 → 记录结果                   │
└─────────────────────┬───────────────────────────┘
                      ↓
┌─────────────────────────────────────────────────┐
│              检查器 (Checker)                    │
│  验证结果 → 判断成功/失败 → 决定下一步            │
└─────────────────────────────────────────────────┘

完整代码

项目结构

automation/
├── main.py             # 主程序
├── tools.py            # 自动化工具
├── agent.py            # Agent定义
└── requirements.txt

tools.py

python

"""自动化工具定义"""
import os
import shutil
import json
from datetime import datetime
from langchain_core.tools import tool
import smtplib
from email.mime.text import MIMEText
import requests

@tool
def list_files(directory: str) -> str:
    """列出目录中的文件
    
    Args:
        directory: 目录路径
        
    Returns:
        文件列表
    """
    if not os.path.exists(directory):
        return f"目录不存在: {directory}"
    
    files = []
    for item in os.listdir(directory):
        path = os.path.join(directory, item)
        if os.path.isfile(path):
            size = os.path.getsize(path)
            files.append(f"{item} ({size} bytes)")
        else:
            files.append(f"{item}/ (目录)")
    
    return "\n".join(files) if files else "目录为空"

@tool
def create_directory(path: str) -> str:
    """创建目录
    
    Args:
        path: 目录路径
        
    Returns:
        执行结果
    """
    try:
        os.makedirs(path, exist_ok=True)
        return f"已创建目录: {path}"
    except Exception as e:
        return f"创建失败: {e}"

@tool
def move_file(source: str, destination: str) -> str:
    """移动文件
    
    Args:
        source: 源文件路径
        destination: 目标路径
        
    Returns:
        执行结果
    """
    try:
        # 确保目标目录存在
        os.makedirs(os.path.dirname(destination), exist_ok=True)
        shutil.move(source, destination)
        return f"已移动: {source} → {destination}"
    except Exception as e:
        return f"移动失败: {e}"

@tool
def get_file_extension(filename: str) -> str:
    """获取文件扩展名
    
    Args:
        filename: 文件名
        
    Returns:
        扩展名（不含点）
    """
    ext = os.path.splitext(filename)[1].lower()
    return ext[1:] if ext else "unknown"

@tool
def write_file(path: str, content: str) -> str:
    """写入文件
    
    Args:
        path: 文件路径
        content: 文件内容
        
    Returns:
        执行结果
    """
    try:
        with open(path, 'w', encoding='utf-8') as f:
            f.write(content)
        return f"已写入文件: {path}"
    except Exception as e:
        return f"写入失败: {e}"

@tool
def read_file(path: str) -> str:
    """读取文件内容
    
    Args:
        path: 文件路径
        
    Returns:
        文件内容
    """
    try:
        with open(path, 'r', encoding='utf-8') as f:
            content = f.read()
        return content[:2000] + ("..." if len(content) > 2000 else "")
    except Exception as e:
        return f"读取失败: {e}"

@tool
def get_current_time() -> str:
    """获取当前时间"""
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件（模拟）
    
    Args:
        to: 收件人
        subject: 主题
        body: 正文
        
    Returns:
        执行结果
    """
    # 模拟发送，实际使用需配置SMTP
    return f"""
邮件已发送（模拟）
收件人: {to}
主题: {subject}
正文: {body[:100]}...
"""

@tool
def http_get(url: str) -> str:
    """发送HTTP GET请求
    
    Args:
        url: 请求URL
        
    Returns:
        响应内容
    """
    try:
        response = requests.get(url, timeout=10)
        return f"状态码: {response.status_code}\n内容: {response.text[:500]}"
    except Exception as e:
        return f"请求失败: {e}"

@tool
def execute_shell(command: str) -> str:
    """执行Shell命令（受限）
    
    Args:
        command: Shell命令
        
    Returns:
        执行结果
    """
    # 安全限制
    dangerous = ['rm', 'sudo', 'chmod', 'chown', '>', '>>', '|']
    if any(d in command for d in dangerous):
        return "命令包含危险操作，已拒绝执行"
    
    try:
        import subprocess
        result = subprocess.run(
            command,
            shell=True,
            capture_output=True,
            text=True,
            timeout=30
        )
        return result.stdout or result.stderr
    except Exception as e:
        return f"执行失败: {e}"

# 工具列表
AUTOMATION_TOOLS = [
    list_files,
    create_directory,
    move_file,
    get_file_extension,
    write_file,
    read_file,
    get_current_time,
    send_email,
    http_get,
    execute_shell
]

agent.py

python

"""自动化Agent"""
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain.agents import create_tool_calling_agent, AgentExecutor
from tools import AUTOMATION_TOOLS

def create_automation_agent():
    """创建自动化Agent"""
    model = ChatOpenAI(model="gpt-4o", temperature=0)
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", """你是自动化助手，负责执行用户指定的任务。

工作流程：
1. 分析任务需求
2. 制定执行计划
3. 逐步执行，每完成一步检查结果
4. 遇到问题时尝试解决或报告

注意事项：
- 操作文件前先确认路径
- 重要操作先备份
- 遇到错误要报告具体原因
- 完成后给出清晰的结果总结

可用工具：
{tool_names}
"""),
        ("user", "{input}"),
        ("placeholder", "{agent_scratchpad}")
    ])
    
    agent = create_tool_calling_agent(model, AUTOMATION_TOOLS, prompt)
    
    return AgentExecutor(
        agent=agent,
        tools=AUTOMATION_TOOLS,
        verbose=True,
        max_iterations=15,
        handle_parsing_errors=True
    )

main.py

python

"""主程序"""
from agent import create_automation_agent

def main():
    print("=" * 60)
    print("自动化脚本助手")
    print("输入'退出'结束")
    print("=" * 60)
    
    agent = create_automation_agent()
    
    # 示例任务
    examples = """
示例任务：
1. 列出当前目录文件
2. 整理下载文件夹，按类型分类
3. 创建一个日志文件
4. 发送邮件通知
"""
    print(examples)
    
    while True:
        task = input("\n任务: ").strip()
        
        if task.lower() in ['退出', 'exit', 'quit']:
            print("再见！")
            break
        
        if not task:
            continue
        
        print("\n执行中...\n")
        
        try:
            result = agent.invoke({"input": task})
            print("\n" + "=" * 60)
            print("执行结果:")
            print("=" * 60)
            print(result["output"])
        except Exception as e:
            print(f"执行出错: {e}")

if __name__ == "__main__":
    main()

运行示例

============================================================
自动化脚本助手
输入'退出'结束
============================================================

示例任务：
1. 列出当前目录文件
2. 整理下载文件夹，按类型分类
3. 创建一个日志文件
4. 发送邮件通知

任务: 列出当前目录的所有文件

执行中...

> Entering new AgentExecutor chain...

调用: list_files(".")
结果: 
README.md (1234 bytes)
src/ (目录)
package.json (567 bytes)
...

============================================================
执行结果:
============================================================
当前目录内容：

文件：
- README.md (1234 bytes)
- package.json (567 bytes)

目录：
- src/ (源代码目录)
- node_modules/ (依赖目录)

任务: 整理downloads文件夹，按文件类型分类

执行中...

> Entering new AgentExecutor chain...

思考：需要先列出downloads文件夹的内容，然后按扩展名分类
调用: list_files("downloads")
结果: 
photo.jpg
document.pdf
song.mp3
report.docx
...

思考：文件类型包括jpg、pdf、mp3、docx，需要创建分类目录
调用: create_directory("downloads/images")
调用: create_directory("downloads/documents")  
调用: create_directory("downloads/music")

思考：移动文件到对应目录
调用: move_file("downloads/photo.jpg", "downloads/images/photo.jpg")
调用: move_file("downloads/document.pdf", "downloads/documents/document.pdf")
...

============================================================
执行结果:
============================================================
已成功整理downloads文件夹！

分类结果：
- 图片文件 (images/): 3个文件
- 文档文件 (documents/): 5个文件  
- 音乐文件 (music/): 2个文件
- 其他文件: 1个文件

共处理11个文件。

安全考虑

自动化Agent有潜在风险，需要注意：

python

# 1. 限制危险命令
DANGEROUS_COMMANDS = ['rm', 'sudo', 'chmod', 'dd', 'mkfs']

# 2. 限制访问范围
ALLOWED_DIRECTORIES = ['/home/user/downloads', '/home/user/documents']

# 3. 操作确认
def confirm_action(action: str) -> bool:
    """危险操作需要确认"""
    response = input(f"确认执行: {action}? (y/n): ")
    return response.lower() == 'y'

# 4. 操作日志
def log_action(action: str, result: str):
    """记录所有操作"""
    with open('automation.log', 'a') as f:
        f.write(f"{datetime.now()}: {action} -> {result}\n")

扩展方向

定时任务 - 支持cron表达式
条件触发 - 文件变化、时间、事件
多任务并行 - 同时执行多个任务
图形界面 - 可视化任务配置

小结

本项目实现了：

自然语言任务理解
自动任务规划
文件系统操作
安全执行机制

恭喜！

你已完成所有项目实战！现在你已经掌握了：

RAG知识库构建
Agent开发
多Agent协作
自动化脚本开发

可以开始构建你自己的AI应用了！

自动化脚本开发 ​

项目目标 ​

技术架构 ​

完整代码 ​

项目结构 ​

tools.py ​

agent.py ​

main.py ​

运行示例 ​

安全考虑 ​

扩展方向 ​

小结 ​

恭喜！ ​

自动化脚本开发

项目目标

技术架构

完整代码

项目结构

tools.py

agent.py

main.py

运行示例

安全考虑

扩展方向

小结

恭喜！