基于Gemini Spark的AI自动办公助手开发实战指南

1. 痛点场景：为什么需要AI自动办公助手

在日常办公场景中，我们经常面临重复性工作的困扰：每天早上需要查看邮件汇总、整理日程安排、生成数据报表、发送提醒通知等固定流程。这些任务虽然规则明确，但耗时耗力且容易因人为因素出错。特别是对于需要跨系统操作的任务，传统自动化方案往往需要复杂的API对接和脚本编写，维护成本较高。

AI Agent技术的出现为这类问题提供了新的解决方案。通过大语言模型的理解和推理能力，AI Agent可以模拟人类操作流程，自动完成多步骤任务。比如Gemini Spark这类智能体框架，能够理解自然语言指令，结合工具调用能力，实现真正的"一句话需求，全自动执行"。

本文将围绕Gemini Spark框架，详细拆解如何构建一个每天早上5点自动执行办公任务的AI Agent。从环境搭建到任务编排，从权限配置到错误处理，提供完整的实操指南。无论你是想提升个人工作效率的开发者，还是为企业寻求自动化解决方案的技术负责人，都能从中获得可直接复用的技术方案。

2. AI Agent核心概念与技术选型

2.1 什么是AI Agent

AI Agent（智能体）是指能够感知环境、进行决策并执行动作的智能系统。与传统自动化脚本相比，AI Agent的核心优势在于：

• 语义理解能力：能够理解自然语言描述的任务需求，无需精确的程序化指令
• 动态决策能力：根据环境变化调整执行策略，处理预期外的异常情况
• 工具使用能力：可以调用外部API、操作软件界面、处理文件数据等
• 学习适应能力：通过反馈机制不断优化任务执行效果

在办公自动化场景中，AI Agent可以替代人工完成邮件处理、数据整理、报告生成、日程管理等重复性工作。

2.2 Gemini Spark框架优势

Gemini Spark是基于大语言模型的AI Agent开发框架，相比其他方案具有以下特点：

• 多模型支持：可以灵活切换不同的底层大语言模型，平衡成本与效果
• 工具生态丰富：内置邮件、日历、文档、数据库等常用办公工具集成
• 任务编排灵活：支持复杂工作流的图形化设计和代码配置
• 权限管理完善：提供细粒度的访问控制，确保企业级安全要求
• 监控日志全面：详细记录任务执行过程，便于排查问题和优化效果

2.3 技术架构概述

典型的AI自动办公系统包含以下核心组件：

任务调度层 → AI决策层 → 工具执行层 → 结果反馈层

• 任务调度层：负责定时触发任务执行，支持cron表达式配置
• AI决策层：基于大语言模型解析任务需求，生成执行计划
• 工具执行层：调用具体的API或界面操作完成实际工作
• 结果反馈层：收集执行结果，生成报告并处理异常情况

3. 环境准备与基础配置

3.1 系统要求与依赖安装

确保你的开发环境满足以下要求：

• 操作系统：Windows 10/11, macOS 10.14+, Ubuntu 18.04+
• Python版本：3.8-3.11（推荐3.9）
• 内存要求：至少8GB RAM，推荐16GB
• 网络环境：稳定的互联网连接，用于访问AI模型API

安装核心依赖包：

# 创建虚拟环境 python -m venv ai_agent_env source ai_agent_env/bin/activate # Linux/macOS # ai_agent_env\Scripts\activate # Windows # 安装Gemini Spark核心包 pip install gemini-spark-core>=1.2.0 pip install python-crontab>=3.0.0 pip install requests>=2.28.0 pip install pandas>=1.5.0 pip install openpyxl>=3.0.0 # 办公自动化相关依赖 pip install exchangelib>=4.0.0 # 邮件处理 pip install google-api-python-client>=2.0.0 # Google服务 pip install schedule>=1.0.0 # 任务调度

3.2 API密钥配置

创建配置文件config.yaml，存储各类服务的访问凭证：

# config.yaml ai_provider: gemini: api_key: "your_gemini_api_key_here" model: "gemini-pro" temperature: 0.1 email: server: "outlook.office365.com" username: "your_email@company.com" password: "your_app_password" mailbox: "INBOX" calendar: provider: "google" # 或 "outlook" credentials_file: "credentials/calendar.json" storage: data_path: "./data" logs_path: "./logs"

重要安全提示：配置文件应加入.gitignore，敏感信息使用环境变量或密钥管理服务。

3.3 项目结构设计

建立清晰的项目目录结构：

ai_office_agent/ ├── src/ │ ├── agents/ # AI Agent核心逻辑 │ ├── tools/ # 工具函数库 │ ├── workflows/ # 工作流定义 │ └── utils/ # 工具函数 ├── config/ # 配置文件 ├── data/ # 数据文件 ├── logs/ # 日志文件 ├── tests/ # 测试用例 └── requirements.txt # 依赖列表

4. 核心Agent开发实战

4.1 基础Agent类实现

创建基础Agent类，封装与大模型交互的核心逻辑：

# src/agents/base_agent.py import logging from abc import ABC, abstractmethod from typing import List, Dict, Any import google.generativeai as genai class BaseAgent(ABC): def __init__(self, config: Dict[str, Any]): self.config = config self.logger = logging.getLogger(self.__class__.__name__) self.setup_ai_client() def setup_ai_client(self): """初始化AI客户端""" try: genai.configure(api_key=self.config['ai_provider']['gemini']['api_key']) self.model = genai.GenerativeModel( self.config['ai_provider']['gemini']['model'] ) self.logger.info("AI客户端初始化成功") except Exception as e: self.logger.error(f"AI客户端初始化失败: {e}") raise async def generate_plan(self, task_description: str) -> Dict[str, Any]: """基于任务描述生成执行计划""" prompt = f""" 请为以下办公任务制定详细的执行计划： 任务：{task_description} 请按步骤列出具体操作，包括： 1. 需要访问哪些系统或数据源 2. 每个步骤的具体操作内容 3. 预期的输出结果格式 4. 可能遇到的异常及处理方案 以JSON格式返回： {{ "steps": [ {{ "step_number": 1, "action": "具体操作描述", "tools_needed": ["工具名称"], "expected_output": "预期结果", "error_handling": "异常处理方案" }} ], "estimated_time": "预计耗时", "prerequisites": ["前提条件"] }} """ try: response = self.model.generate_content(prompt) plan = self._parse_response(response.text) return plan except Exception as e: self.logger.error(f"生成执行计划失败: {e}") return {"error": str(e)} def _parse_response(self, response_text: str) -> Dict[str, Any]: """解析AI返回的文本为结构化数据""" # 实现JSON解析逻辑，处理可能的格式错误 import json import re # 提取JSON部分 json_match = re.search(r'\{.*\}', response_text, re.DOTALL) if json_match: try: return json.loads(json_match.group()) except json.JSONDecodeError: self.logger.warning("JSON解析失败，使用备用解析方案") # 备用解析逻辑 return self._fallback_parse(response_text)

4.2 邮件处理工具实现

开发邮件读取和分析工具：

# src/tools/email_tool.py from exchangelib import Credentials, Account, Configuration, Message from exchangelib.items import Mailbox import pandas as pd from datetime import datetime, timedelta import logging class EmailTool: def __init__(self, config: Dict[str, Any]): self.config = config self.logger = logging.getLogger(__name__) self.account = self._setup_account() def _setup_account(self): """设置Exchange账户连接""" try: credentials = Credentials( username=self.config['email']['username'], password=self.config['email']['password'] ) config = Configuration( server=self.config['email']['server'], credentials=credentials ) return Account( primary_smtp_address=self.config['email']['username'], config=config, autodiscover=False, access_type='delegate' ) except Exception as e: self.logger.error(f"邮箱连接失败: {e}") raise def get_unread_emails(self, since_hours: int = 24) -> pd.DataFrame: """获取指定时间范围内的未读邮件""" try: since_time = datetime.now() - timedelta(hours=since_hours) unread_emails = self.account.inbox.filter( datetime_received__gte=since_time, is_read=False ) emails_data = [] for email in unread_emails: emails_data.append({ 'subject': email.subject, 'sender': email.sender.email_address, 'received_time': email.datetime_received, 'importance': email.importance, 'has_attachments': email.has_attachments, 'body_preview': email.text_body[:200] if email.text_body else "" }) return pd.DataFrame(emails_data) except Exception as e: self.logger.error(f"获取未读邮件失败: {e}") return pd.DataFrame() def mark_as_read(self, email_subject: str) -> bool: """标记特定主题的邮件为已读""" try: target_emails = self.account.inbox.filter(subject__contains=email_subject) for email in target_emails: email.is_read = True email.save() return True except Exception as e: self.logger.error(f"标记邮件为已读失败: {e}") return False

4.3 日程管理工具

实现日历事件的读取和创建功能：

# src/tools/calendar_tool.py from datetime import datetime, timedelta from google.oauth2.credentials import Credentials from googleapiclient.discovery import build from google.auth.transport.requests import Request import pickle import os.path import logging class CalendarTool: def __init__(self, config: Dict[str, Any]): self.config = config self.logger = logging.getLogger(__name__) self.service = self._setup_calendar_service() def _setup_calendar_service(self): """设置Google日历服务""" try: creds = None token_file = self.config['calendar']['credentials_file'] if os.path.exists(token_file): with open(token_file, 'rb') as token: creds = pickle.load(token) if not creds or not creds.valid: if creds and creds.expired and creds.refresh_token: creds.refresh(Request()) else: # 这里需要实现OAuth流程 raise Exception("请先完成OAuth认证流程") with open(token_file, 'wb') as token: pickle.dump(creds, token) return build('calendar', 'v3', credentials=creds) except Exception as e: self.logger.error(f"日历服务初始化失败: {e}") raise def get_today_events(self) -> list: """获取今天的日程安排""" try: today_start = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) today_end = today_start + timedelta(days=1) events_result = self.service.events().list( calendarId='primary', timeMin=today_start.isoformat() + 'Z', timeMax=today_end.isoformat() + 'Z', singleEvents=True, orderBy='startTime' ).execute() events = events_result.get('items', []) return events except Exception as e: self.logger.error(f"获取今日日程失败: {e}") return [] def create_reminder(self, title: str, start_time: datetime, duration_minutes: int = 30) -> bool: """创建提醒事件""" try: end_time = start_time + timedelta(minutes=duration_minutes) event = { 'summary': title, 'start': { 'dateTime': start_time.isoformat(), 'timeZone': 'Asia/Shanghai', }, 'end': { 'dateTime': end_time.isoformat(), 'timeZone': 'Asia/Shanghai', }, 'reminders': { 'useDefault': True, }, } event = self.service.events().insert( calendarId='primary', body=event ).execute() self.logger.info(f"提醒事件创建成功: {event.get('htmlLink')}") return True except Exception as e: self.logger.error(f"创建提醒事件失败: {e}") return False

5. 定时任务系统实现

5.1 基于cron的调度机制

实现可靠的定时任务调度系统：

# src/scheduler/task_scheduler.py import schedule import time import threading from datetime import datetime import logging from crontab import CronTab class TaskScheduler: def __init__(self, config: Dict[str, Any]): self.config = config self.logger = logging.getLogger(__name__) self.running = False self.scheduler_thread = None def setup_daily_task(self, task_func, task_time="05:00"): """设置每日定时任务""" try: # 使用schedule库设置每日任务 schedule.every().day.at(task_time).do(task_func) self.logger.info(f"每日任务设置成功: {task_time}") # 同时设置cron任务作为备用方案 self._setup_cron_backup(task_time) except Exception as e: self.logger.error(f"定时任务设置失败: {e}") def _setup_cron_backup(self, task_time: str): """设置cron任务作为备用调度方案""" try: hour, minute = map(int, task_time.split(':')) cron = CronTab(user=True) # 清理可能存在的旧任务 cron.remove_all(comment='ai_office_agent') job = cron.new( command=f'cd {os.getcwd()} && python run_daily_task.py', comment='ai_office_agent' ) job.setall(minute, hour, '*', '*', '*') cron.write() self.logger.info("Cron备份任务设置成功") except Exception as e: self.logger.warning(f"Cron备份设置失败: {e}") def start_scheduler(self): """启动调度器""" self.running = True self.scheduler_thread = threading.Thread(target=self._scheduler_loop) self.scheduler_thread.daemon = True self.scheduler_thread.start() self.logger.info("任务调度器已启动") def _scheduler_loop(self): """调度器主循环""" while self.running: try: schedule.run_pending() time.sleep(60) # 每分钟检查一次 except Exception as e: self.logger.error(f"调度器循环异常: {e}") time.sleep(300) # 异常时等待5分钟 def stop_scheduler(self): """停止调度器""" self.running = False if self.scheduler_thread: self.scheduler_thread.join(timeout=10) self.logger.info("任务调度器已停止")

5.2 每日办公任务工作流

实现具体的每日自动办公流程：

# src/workflows/daily_office_workflow.py import asyncio from datetime import datetime import pandas as pd from src.agents.base_agent import BaseAgent from src.tools.email_tool import EmailTool from src.tools.calendar_tool import CalendarTool import logging class DailyOfficeWorkflow: def __init__(self, config: Dict[str, Any]): self.config = config self.logger = logging.getLogger(__name__) self.agent = BaseAgent(config) self.email_tool = EmailTool(config) self.calendar_tool = CalendarTool(config) async def execute_daily_tasks(self): """执行每日办公任务序列""" task_start_time = datetime.now() self.logger.info(f"开始执行每日办公任务: {task_start_time}") try: # 任务1: 邮件汇总分析 email_report = await self._process_daily_emails() # 任务2: 日程安排检查 schedule_report = await self._check_daily_schedule() # 任务3: 生成综合日报 daily_report = await self._generate_daily_report( email_report, schedule_report ) # 任务4: 发送结果通知 await self._send_notification(daily_report) task_end_time = datetime.now() duration = (task_end_time - task_start_time).total_seconds() self.logger.info( f"每日办公任务执行完成，耗时: {duration:.2f}秒" ) return { "status": "success", "execution_time": duration, "reports_generated": True } except Exception as e: self.logger.error(f"每日任务执行失败: {e}") return { "status": "error", "error": str(e), "execution_time": 0 } async def _process_daily_emails(self) -> Dict[str, Any]: """处理每日邮件汇总""" self.logger.info("开始处理每日邮件...") # 获取过去24小时的未读邮件 unread_emails = self.email_tool.get_unread_emails(24) if unread_emails.empty: self.logger.info("过去24小时无未读邮件") return {"email_count": 0, "important_emails": []} # 使用AI分析邮件重要性 email_analysis = await self._analyze_email_importance(unread_emails) # 标记已处理的重要邮件 important_count = await self._process_important_emails(email_analysis) return { "email_count": len(unread_emails), "important_emails": important_count, "analysis": email_analysis } async def _analyze_email_importance(self, emails_df: pd.DataFrame) -> Dict[str, Any]: """使用AI分析邮件重要性""" email_samples = emails_df.head(5).to_dict('records') analysis_prompt = f""" 请分析以下邮件的重要性，识别需要立即关注的邮件： 邮件样本: {email_samples} 请按紧急程度分类，并建议处理优先级。 返回JSON格式： {{ "urgent_emails": [ {{ "subject": "邮件主题", "reason": "紧急原因", "suggested_action": "建议操作" }} ], "important_emails": [ {{ "subject": "邮件主题", "priority": "高/中/低" }} ], "summary": "整体分析摘要" }} """ try: response = self.agent.model.generate_content(analysis_prompt) return self.agent._parse_response(response.text) except Exception as e: self.logger.error(f"邮件分析失败: {e}") return {"error": str(e)}

6. 系统集成与部署

6.1 主程序入口实现

创建系统主入口文件：

# main.py import asyncio import logging import signal import sys import yaml from pathlib import Path from src.scheduler.task_scheduler import TaskScheduler from src.workflows.daily_office_workflow import DailyOfficeWorkflow # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('logs/ai_agent.log'), logging.StreamHandler(sys.stdout) ] ) class AIOfficeAgent: def __init__(self, config_path: str = "config/config.yaml"): self.config = self._load_config(config_path) self.logger = logging.getLogger(__name__) self.scheduler = TaskScheduler(self.config) self.workflow = DailyOfficeWorkflow(self.config) # 设置信号处理 signal.signal(signal.SIGINT, self._signal_handler) signal.signal(signal.SIGTERM, self._signal_handler) def _load_config(self, config_path: str) -> Dict[str, Any]: """加载配置文件""" try: with open(config_path, 'r', encoding='utf-8') as f: return yaml.safe_load(f) except Exception as e: logging.error(f"配置文件加载失败: {e}") sys.exit(1) def _signal_handler(self, signum, frame): """信号处理函数""" self.logger.info(f"收到信号 {signum}，开始优雅关闭...") self.stop() async def run_daily_task(self): """执行每日任务""" self.logger.info("触发每日任务执行") result = await self.workflow.execute_daily_tasks() self.logger.info(f"任务执行结果: {result}") def start(self): """启动AI办公助手""" self.logger.info("启动AI自动办公助手系统") # 设置定时任务 self.scheduler.setup_daily_task( lambda: asyncio.run(self.run_daily_task()), self.config.get('schedule', {}).get('daily_time', '05:00') ) # 启动调度器 self.scheduler.start_scheduler() self.logger.info("系统启动完成，等待定时任务执行...") # 保持主线程运行 try: while True: signal.pause() except KeyboardInterrupt: self.stop() def stop(self): """停止系统""" self.logger.info("正在停止系统...") self.scheduler.stop_scheduler() self.logger.info("系统已停止") sys.exit(0) if __name__ == "__main__": # 创建必要的目录 Path("logs").mkdir(exist_ok=True) Path("data").mkdir(exist_ok=True) agent = AIOfficeAgent() agent.start()

6.2 系统服务化部署

创建系统服务配置文件，支持后台运行：

# ai-office-agent.service [Unit] Description=AI Office Automation Agent After=network.target [Service] Type=simple User=ubuntu WorkingDirectory=/opt/ai-office-agent ExecStart=/opt/ai-office-agent/venv/bin/python main.py Restart=always RestartSec=10 StandardOutput=journal StandardError=journal [Install] WantedBy=multi-user.target

创建部署脚本：

#!/bin/bash # deploy.sh echo "开始部署AI办公助手系统..." # 检查Python环境 if ! command -v python3.9 &> /dev/null; then echo "错误: 需要Python 3.9" exit 1 fi # 创建部署目录 sudo mkdir -p /opt/ai-office-agent sudo chown $USER:$USER /opt/ai-office-agent # 复制项目文件 cp -r . /opt/ai-office-agent/ # 创建虚拟环境 cd /opt/ai-office-agent python3.9 -m venv venv source venv/bin/activate # 安装依赖 pip install -r requirements.txt # 设置配置文件 if [ ! -f config/config.yaml ]; then cp config/config.example.yaml config/config.yaml echo "请编辑 config/config.yaml 文件配置API密钥" fi # 设置系统服务 sudo cp deployment/ai-office-agent.service /etc/systemd/system/ sudo systemctl daemon-reload sudo systemctl enable ai-office-agent.service echo "部署完成！" echo "请配置 config/config.yaml 后运行: sudo systemctl start ai-office-agent"

7. 监控与错误处理

7.1 完善的日志系统

实现分级日志记录和监控：

# src/utils/logging_utils.py import logging import json from datetime import datetime from pathlib import Path class JSONFormatter(logging.Formatter): """JSON格式日志格式化器""" def format(self, record): log_entry = { "timestamp": datetime.now().isoformat(), "level": record.levelname, "logger": record.name, "message": record.getMessage(), "module": record.module, "function": record.funcName, "line": record.lineno } if record.exc_info: log_entry["exception"] = self.formatException(record.exc_info) return json.dumps(log_entry, ensure_ascii=False) def setup_advanced_logging(config: Dict[str, Any]): """设置高级日志配置""" log_dir = Path(config['storage']['logs_path']) log_dir.mkdir(exist_ok=True) # 创建不同的日志处理器 json_handler = logging.FileHandler(log_dir / 'agent.json.log') json_handler.setFormatter(JSONFormatter()) debug_handler = logging.FileHandler(log_dir / 'agent.debug.log') debug_handler.setLevel(logging.DEBUG) # 设置根日志记录器 root_logger = logging.getLogger() root_logger.setLevel(logging.INFO) for handler in [json_handler, debug_handler]: root_logger.addHandler(handler) return root_logger

7.2 错误处理与重试机制

实现健壮的错误处理策略：

# src/utils/error_handling.py import asyncio from functools import wraps from typing import Type, Tuple import logging class RetryStrategy: def __init__(self, max_retries: int = 3, base_delay: float = 1.0): self.max_retries = max_retries self.base_delay = base_delay async def execute_with_retry(self, func, *args, **kwargs): """带重试的执行策略""" last_exception = None for attempt in range(self.max_retries + 1): try: if asyncio.iscoroutinefunction(func): return await func(*args, **kwargs) else: return func(*args, **kwargs) except Exception as e: last_exception = e if attempt < self.max_retries: delay = self.base_delay * (2 ** attempt) # 指数退避 logging.warning( f"操作失败，{delay}秒后重试 (尝试 {attempt + 1}/{self.max_retries + 1}): {e}" ) await asyncio.sleep(delay) else: logging.error(f"操作最终失败 after {self.max_retries + 1} 次尝试: {e}") raise last_exception raise last_exception def retry_on_exception(retry_strategy: RetryStrategy, exceptions: Tuple[Type[Exception]] = (Exception,)): """重试装饰器""" def decorator(func): @wraps(func) async def async_wrapper(*args, **kwargs): return await retry_strategy.execute_with_retry(func, *args, **kwargs) @wraps(func) def sync_wrapper(*args, **kwargs): return retry_strategy.execute_with_retry(func, *args, **kwargs) return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper return decorator

8. 安全与权限管理

8.1 敏感信息保护

实现安全的密钥管理方案：

# src/security/credential_manager.py import os import keyring from cryptography.fernet import Fernet import base64 import logging class CredentialManager: def __init__(self, service_name: str = "ai_office_agent"): self.service_name = service_name self.logger = logging.getLogger(__name__) self._setup_encryption_key() def _setup_encryption_key(self): """设置加密密钥""" key_env_var = "AI_AGENT_ENCRYPTION_KEY" key = os.getenv(key_env_var) if not key: # 生成新密钥并提示用户保存 key = Fernet.generate_key().decode() self.logger.warning( f"请设置环境变量 {key_env_var}={key} " f"并妥善保存此密钥！" ) self.fernet = Fernet(key.encode()) def store_credential(self, credential_name: str, value: str): """安全存储凭据""" try: encrypted_value = self.fernet.encrypt(value.encode()) keyring.set_password( self.service_name, credential_name, base64.b64encode(encrypted_value).decode() ) self.logger.info(f"凭据 {credential_name} 已安全存储") except Exception as e: self.logger.error(f"凭据存储失败: {e}") raise def get_credential(self, credential_name: str) -> str: """获取存储的凭据""" try: encrypted_b64 = keyring.get_password(self.service_name, credential_name) if not encrypted_b64: raise ValueError(f"凭据 {credential_name} 不存在") encrypted_value = base64.b64decode(encrypted_b64) decrypted_value = self.fernet.decrypt(encrypted_value) return decrypted_value.decode() except Exception as e: self.logger.error(f"凭据获取失败: {e}") raise

8.2 访问控制与审计

实现操作审计和权限验证：

# src/security/access_controller.py import json from datetime import datetime from typing import List, Set import logging class AccessController: def __init__(self, config: Dict[str, Any]): self.config = config self.logger = logging.getLogger(__name__) self.allowed_operations = self._load_allowed_operations() self.audit_log = [] def _load_allowed_operations(self) -> Set[str]: """加载允许的操作列表""" return { "read_emails", "send_emails", "read_calendar", "create_events", "read_files", "write_reports" } def check_permission(self, operation: str, context: Dict[str, Any]) -> bool: """检查操作权限""" if operation not in self.allowed_operations: self._log_audit("PERMISSION_DENIED", operation, context, "操作不在允许列表中") return False # 检查时间限制（避免非工作时间执行敏感操作） if not self._check_time_restriction(operation): self._log_audit("TIME_RESTRICTION", operation, context, "非允许时间段") return False self._log_audit("PERMISSION_GRANTED", operation, context, "权限检查通过") return True def _check_time_restriction(self, operation: str) -> bool: """检查时间限制""" current_hour = datetime.now().hour sensitive_operations = {"send_emails", "create_events"} if operation in sensitive_operations: # 敏感操作只在工作时间允许（9-18点） return 9 <= current_hour <= 18 return True def _log_audit(self, action: str, operation: str, context: Dict[str, Any], reason: str): """记录审计日志""" audit_entry = { "timestamp": datetime.now().isoformat(), "action": action, "operation": operation, "context": context, "reason": reason, "user": "ai_agent" # 在实际系统中替换为实际用户标识 } self.audit_log.append(audit_entry) self.logger.info(f"审计日志: {json.dumps(audit_entry, ensure_ascii=False)}")

9. 性能优化与最佳实践

9.1 资源使用优化

优化AI模型调用和资源使用：

# src/optimization/resource_manager.py import asyncio from concurrent.futures import ThreadPoolExecutor import psutil import logging from typing import List, Callable class ResourceManager: def __init__(self, max_workers: int = None): self.max_workers = max_workers or min(32, (psutil.cpu_count() or 1) + 4) self.thread_pool = ThreadPoolExecutor(max_workers=self.max_workers) self.logger = logging.getLogger(__name__) self._setup_resource_monitoring() def _setup_resource_monitoring(self): """设置资源监控""" self.memory_threshold = 0.8 # 80%内存使用阈值 self.cpu_threshold = 0.7 # 70%CPU使用阈值 async def execute_with_throttling(self, tasks: List[Callable], max_concurrent: int = 5): """带限流的批量任务执行""" semaphore = asyncio.Semaphore(max_concurrent) async def bounded_task(task): async with semaphore: # 检查系统资源 if not self._check_system_resources(): await asyncio.sleep(1) # 资源紧张时等待 if asyncio.iscoroutinefunction(task): return await task() else: loop = asyncio.get_event_loop() return await loop.run_in_executor(self.thread_pool, task) return await asyncio.gather(*[bounded_task(task) for task in tasks]) def _check_system_resources(self) -> bool: """检查系统资源使用情况""" memory_usage = psutil.virtual_memory().percent / 100 cpu_usage = psutil.cpu_percent(interval=0.1) / 100 if memory_usage > self.memory_threshold or cpu_usage > self.cpu_threshold: self.logger.warning( f"系统资源紧张: 内存{memory_usage:.1%}, CPU{cpu_usage:.1%}" ) return False return True

9.2 缓存策略实现

减少重复的AI调用和API请求：

# src/optimization