Build Your Own AI Butler - A Scheduled Agent That Runs Itself!

┌─────────────────────────────────────────────────────┐
│                   AI Butler                          │
├─────────────────────────────────────────────────────┤
│  Scheduler Layer（调度层）                           │
│  - cron 定时触发                                     │
│  - 检查是否该运行                                    │
│  - 管理运行状态                                      │
├─────────────────────────────────────────────────────┤
│  Memory Layer（记忆层）                              │
│  - 保存历史任务记录                                  │
│  - 保存上下文信息                                    │
│  - 跨任务保持连贯性                                  │
├─────────────────────────────────────────────────────┤
│  Agent Layer（智能体层）                              │
│  - 理解任务目标                                      │
│  - 制定执行计划                                      │
│  - 调用工具完成任务                                  │
├─────────────────────────────────────────────────────┤
│  Tools Layer（工具层）                               │
│  - 搜索网页                                          │
│  - 发送邮件/消息                                    │
│  - 读写文件                                          │
│  - 执行代码                                          │
└─────────────────────────────────────────────────────┘

import schedule
import time
import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class Scheduler:
    def __init__(self, agent):
        self.agent = agent
        self.running = False

    def job_morning_brief(self):
        """每天早上 8 点运行：行业动态 + 任务清单"""
        logger.info("🕗 开始执行早间简报任务")
        try:
            result = self.agent.run([
                "搜索今天的 AI 行业最新动态",
                "整理昨天的工作日志摘要",
                "生成今天的任务清单，优先级从高到低"
            ])
            self.send_notification(result)
            logger.info("✅ 早间简报任务完成")
        except Exception as e:
            logger.error(f"❌ 任务执行失败: {e}")
            self.send_error_alert(str(e))

    def job_daily_report(self):
        """每天下午 6 点运行：生成工作日报"""
        logger.info("🕕 开始执行工作日报任务")
        try:
            result = self.agent.run([
                "整理今天完成的所有工作",
                "统计时间分配",
                "生成明天的计划建议"
            ])
            self.save_report(result, date.today().isoformat())
            logger.info("✅ 工作日报任务完成")
        except Exception as e:
            logger.error(f"❌ 任务执行失败: {e}")

    def send_notification(self, content):
        """发送飞书通知"""
        # 飞书 webhook 或者 API
        pass

    def send_error_alert(self, error_message):
        """发送错误告警"""
        pass

    def save_report(self, content, date_str):
        """保存报告到本地"""
        with open(f"reports/{date_str}.md", "w") as f:
            f.write(content)

    def start(self):
        """启动调度器"""
        # 早上 8 点
        schedule.every().day.at("08:00").do(self.job_morning_brief)

        # 下午 6 点
        schedule.every().day.at("18:00").do(self.job_daily_report)

        # 每小时的健康检查
        schedule.every().hour.do(self.health_check)

        logger.info("⏰ 调度器已启动")
        logger.info("📅 任务安排：")
        logger.info("  - 08:00 早间简报")
        logger.info("  - 18:00 工作日报")
        logger.info("  - 每小时健康检查")

        while True:
            schedule.run_pending()
            time.sleep(60)  # 每分钟检查一次

    def health_check(self):
        """健康检查"""
        import psutil
        memory = psutil.virtual_memory()
        if memory.percent > 90:
            logger.warning(f"⚠️ 内存使用率过高: {memory.percent}%")

import json
import os
from datetime import datetime

class Memory:
    """记忆层：保存历史任务和上下文"""

    def __init__(self, memory_dir="memory"):
        self.memory_dir = memory_dir
        os.makedirs(memory_dir, exist_ok=True)
        self.short_term = []  # 短期记忆：最近的任务
        self.long_term = self._load_long_term()  # 长期记忆：总结的经验

    def _load_long_term(self):
        """加载长期记忆"""
        long_term_file = os.path.join(self.memory_dir, "long_term.json")
        if os.path.exists(long_term_file):
            with open(long_term_file, "r") as f:
                return json.load(f)
        return {
            "user_preferences": {},
            "repeated_tasks": [],
            "learned_patterns": [],
            "important_facts": []
        }

    def save_long_term(self):
        """保存长期记忆"""
        long_term_file = os.path.join(self.memory_dir, "long_term.json")
        with open(long_term_file, "w") as f:
            json.dump(self.long_term, f, indent=2, ensure_ascii=False)

    def add_task_record(self, task: str, result: str, success: bool):
        """记录任务执行"""
        record = {
            "timestamp": datetime.now().isoformat(),
            "task": task,
            "result_length": len(result),
            "success": success
        }
        self.short_term.append(record)

        # 超过 100 条记录，做一次总结，把重要信息移到长期记忆
        if len(self.short_term) > 100:
            self._consolidate_memory()

    def _consolidate_memory(self):
        """整合记忆：把短期记忆的重要信息移到长期记忆"""
        # 分析最近的任务模式
        recent_tasks = self.short_term[-50:]
        success_rate = sum(1 for t in recent_tasks if t["success"]) / len(recent_tasks)

        # 如果某个任务反复成功，记住它
        task_counts = {}
        for task in recent_tasks:
            # 简化任务描述作为 key
            key = task["task"][:50]
            task_counts[key] = task_counts.get(key, 0) + 1

        repeated = [k for k, v in task_counts.items() if v >= 3]
        if repeated:
            self.long_term["repeated_tasks"].extend(repeated[:10])

        # 清理短期记忆
        self.short_term = self.short_term[-50:]  # 只保留最近 50 条

        self.save_long_term()

    def get_context(self) -> str:
        """获取当前上下文，用于给 Agent 参考"""
        context_parts = []

        # 最近的任务
        if self.short_term:
            recent = self.short_term[-5:]
            context_parts.append("最近完成的任务：")
            for task in recent:
                status = "✅" if task["success"] else "❌"
                context_parts.append(
                    f"  {status} {task['timestamp'][:10]}: {task['task'][:80]}"
                )

        # 用户的偏好
        if self.long_term.get("user_preferences"):
            prefs = self.long_term["user_preferences"]
            context_parts.append(f"\n你的偏好：{prefs}")

        # 重要的facts
        if self.long_term.get("important_facts"):
            facts = self.long_term["important_facts"]
            context_parts.append(f"\n重要事实：{facts}")

        return "\n".join(context_parts)

import openai
from typing import List

class AIAgent:
    """AI Agent：理解任务、制定计划、执行操作"""

    def __init__(self, memory: Memory, tools: list):
        self.client = openai.OpenAI()
        self.memory = memory
        self.tools = tools
        self.system_prompt = self._build_system_prompt()

    def _build_system_prompt(self) -> str:
        return """你是我的个人 AI Butler，帮助我处理日常任务。

你的职责：
1. 搜索和整理信息
2. 生成报告和摘要
3. 提醒重要事项
4. 自动化日常任务

你有以下工具可以使用：
- search_web(query): 搜索网页获取信息
- send_message(to, content): 发送消息
- save_file(path, content): 保存文件
- read_file(path): 读取文件

工作原则：
1. 每一步都要清楚你在做什么
2. 如果不确定，先搜索再行动
3. 重要的操作要有确认
4. 完成后简洁汇报结果

记住我的偏好：
- 报告要简洁，突出重点
- 不要说废话
- 如果任务复杂，分解成小步骤"""

    def run(self, tasks: List[str]) -> str:
        """执行任务列表"""
        results = []

        for task in tasks:
            result = self._execute_task(task)
            results.append(result)

            # 记录到记忆
            self.memory.add_task_record(task, result, success=True)

        return "\n\n".join(results)

    def _execute_task(self, task: str) -> str:
        """执行单个任务"""
        # 获取上下文
        context = self.memory.get_context()

        # 构建消息
        messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "system", "content": f"当前上下文：\n{context}"},
            {"role": "user", "content": f"任务：{task}\n\n请完成这个任务，并汇报结果。"}
        ]

        # 调用 LLM
        response = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            tools=self._get_tools_spec(),
            tool_choice="auto"
        )

        # 处理工具调用
        full_response = self._handle_tool_calls(response)

        return full_response

    def _get_tools_spec(self):
        """获取工具定义"""
        return [
            {
                "type": "function",
                "function": {
                    "name": "search_web",
                    "description": "搜索网页获取信息",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "query": {"type": "string", "description": "搜索关键词"}
                        },
                        "required": ["query"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "send_message",
                    "description": "发送消息给用户",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "to": {"type": "string", "description": "收件人"},
                            "content": {"type": "string", "description": "消息内容"}
                        },
                        "required": ["to", "content"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "save_file",
                    "description": "保存内容到文件",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "path": {"type": "string", "description": "文件路径"},
                            "content": {"type": "string", "description": "文件内容"}
                        },
                        "required": ["path", "content"]
                    }
                }
            }
        ]

    def _handle_tool_calls(self, response) -> str:
        """处理工具调用"""
        message = response.choices[0].message

        if not message.tool_calls:
            return message.content

        # 执行工具调用
        tool_results = []
        for tool_call in message.tool_calls:
            tool_name = tool_call.function.name
            tool_args = json.loads(tool_call.function.arguments)

            # 根据工具名称调用不同的处理函数
            if tool_name == "search_web":
                result = self._search_web(tool_args["query"])
            elif tool_name == "send_message":
                result = self._send_message(tool_args["to"], tool_args["content"])
            elif tool_name == "save_file":
                result = self._save_file(tool_args["path"], tool_args["content"])
            else:
                result = f"Unknown tool: {tool_name}"

            tool_results.append({
                "tool_call_id": tool_call.id,
                "tool_name": tool_name,
                "result": result
            })

        # 把工具结果返回给 LLM，让它继续
        messages = [
            {"role": "user", "content": f"工具执行结果：{tool_results}"}
        ]

        follow_up = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages
        )

        return follow_up.choices[0].message.content

    def _search_web(self, query: str) -> str:
        """搜索网页"""
        # 这里可以接入真实的搜索引擎 API
        # 比如 Google Custom Search、Bing API，或者 DuckDuckGo
        return f"[模拟搜索结果] 关于 '{query}' 的信息：xxx"

    def _send_message(self, to: str, content: str) -> str:
        """发送消息"""
        # 这里可以接入飞书、企业微信、Slack 等消息渠道
        return f"消息已发送给 {to}"

    def _save_file(self, path: str, content: str) -> str:
        """保存文件"""
        os.makedirs(os.path.dirname(path), exist_ok=True)
        with open(path, "w") as f:
            f.write(content)
        return f"文件已保存到 {path}"

#!/usr/bin/env python3
"""
AI Butler - 个人 AI 助手
启动脚本
"""

import logging
from memory import Memory
from agent import AIAgent
from scheduler import Scheduler

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

def main():
    logger.info("🚀 启动 AI Butler...")

    # 初始化各层
    memory = Memory()
    tools = ["search_web", "send_message", "save_file"]
    agent = AIAgent(memory=memory, tools=tools)
    scheduler = Scheduler(agent)

    # 启动调度器
    scheduler.start()

if __name__ == "__main__":
    main()

# 在服务器上后台运行
nohup python ai_butler.py > ai_butler.log 2>&1 &

# 或者用 systemd 管理（更可靠）
# /etc/systemd/system/ai-butler.service

[Unit]
Description=AI Butler - Personal AI Assistant
After=network.target

[Service]
Type=simple
User=pi
WorkingDirectory=/home/pi/ai-butler
ExecStart=/usr/bin/python3 ai_butler.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

🤖 AI Butler 早间简报 (2026-05-07)

📰 今日 AI 行业动态：
1. OpenAI 发布 GPT-5，性能提升 40%
2. Meta 开源 LLaMA-4，7B 参数跑出 70B 效果
3. 斯坦福发布 AI Safety 报告，提出新框架

📋 昨日工作摘要：
- 完成了 PPT Agent v2.0 的架构设计
- 修复了 RAG 检索的 bug
- 参加了产品评审会议

✅ 今日任务清单（按优先级）：
1. [高] 代码 review：PPT Agent 的 Agent 层实现
2. [高] 测试：RAG 系统的混合检索功能
3. [中] 文档：更新技术架构文档
4. [低] 优化：LLM 调用成本分析

⏰ 当前时间：08:00，祝你今天高效工作！

import pytz

tz = pytz.timezone('Asia/Shanghai')
schedule.every().day.at("08:00", tz).do(self.job_morning_brief)

def _execute_task_with_retry(self, task, max_retries=3):
    for attempt in range(max_retries):
        try:
            return self._execute_task(task)
        except Exception as e:
            if attempt < max_retries - 1:
                logger.warning(f"任务执行失败，重试中... ({attempt + 1}/{max_retries})")
                time.sleep(2 ** attempt)  # 指数退避
            else:
                logger.error(f"任务执行失败，已重试 {max_retries} 次")
                raise