★ FEATURED ARTICLE 1. 痛点场景为什么需要AI自动办公助手在日常办公场景中我们经常面临重复性工作的困扰每天早上需要查看邮件汇总、整理日程安排、生成数据报表、发送提醒通知等固定流程。这些任务虽然规则明确但耗时耗力且容易因人为因素出错。特别是对于需要跨系统操作的任务传统自动化方案往往需要复杂的API对接和脚本编写维护成本较高。AI Agent技术的出现为这类问题提供了新的解决方案。通过大语言模型的理解和推理能力AI Agent可以模拟人类操作流程自动完成多步骤任务。比如Gemini Spark这类智能体框架能够理解自然语言指令结合工具调用能力实现真正的一句话需求全自动执行。本文将围绕Gemini Spark框架详细拆解如何构建一个每天早上5点自动执行办公任务的AI Agent。从环境搭建到任务编排从权限配置到错误处理提供完整的实操指南。无论你是想提升个人工作效率的开发者还是为企业寻求自动化解决方案的技术负责人都能从中获得可直接复用的技术方案。2. AI Agent核心概念与技术选型2.1 什么是AI AgentAI 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.04Python版本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-core1.2.0 pip install python-crontab3.0.0 pip install requests2.28.0 pip install pandas1.5.0 pip install openpyxl3.0.0 # 办公自动化相关依赖 pip install exchangelib4.0.0 # 邮件处理 pip install google-api-python-client2.0.0 # Google服务 pip install schedule1.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_emailcompany.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_keyself.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(fAI客户端初始化失败: {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( usernameself.config[email][username], passwordself.config[email][password] ) config Configuration( serverself.config[email][server], credentialscredentials ) return Account( primary_smtp_addressself.config[email][username], configconfig, autodiscoverFalse, access_typedelegate ) 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(hourssince_hours) unread_emails self.account.inbox.filter( datetime_received__gtesince_time, is_readFalse ) 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__containsemail_subject) for email in target_emails: email.is_read True email.save() return True except Exception as e: self.logger.error(f标记邮件为已读失败: {e}) return False4.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, credentialscreds) except Exception as e: self.logger.error(f日历服务初始化失败: {e}) raise def get_today_events(self) - list: 获取今天的日程安排 try: today_start datetime.now().replace(hour0, minute0, second0, microsecond0) today_end today_start timedelta(days1) events_result self.service.events().list( calendarIdprimary, timeMintoday_start.isoformat() Z, timeMaxtoday_end.isoformat() Z, singleEventsTrue, orderBystartTime ).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(minutesduration_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( calendarIdprimary, bodyevent ).execute() self.logger.info(f提醒事件创建成功: {event.get(htmlLink)}) return True except Exception as e: self.logger.error(f创建提醒事件失败: {e}) return False5. 定时任务系统实现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_time05: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(userTrue) # 清理可能存在的旧任务 cron.remove_all(commentai_office_agent) job cron.new( commandfcd {os.getcwd()} python run_daily_task.py, commentai_office_agent ) job.setall(minute, hour, *, *, *) cron.write() self.logger.info(Cron备份任务设置成功) except Exception as e: self.logger.warning(fCron备份设置失败: {e}) def start_scheduler(self): 启动调度器 self.running True self.scheduler_thread threading.Thread(targetself._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(timeout10) 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( levellogging.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, encodingutf-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_okTrue) Path(data).mkdir(exist_okTrue) agent AIOfficeAgent() agent.start()6.2 系统服务化部署创建系统服务配置文件支持后台运行# ai-office-agent.service [Unit] DescriptionAI Office Automation Agent Afternetwork.target [Service] Typesimple Userubuntu WorkingDirectory/opt/ai-office-agent ExecStart/opt/ai-office-agent/venv/bin/python main.py Restartalways RestartSec10 StandardOutputjournal StandardErrorjournal [Install] WantedBymulti-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-agent7. 监控与错误处理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_asciiFalse) def setup_advanced_logging(config: Dict[str, Any]): 设置高级日志配置 log_dir Path(config[storage][logs_path]) log_dir.mkdir(exist_okTrue) # 创建不同的日志处理器 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_logger7.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 decorator8. 安全与权限管理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}) raise8.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_asciiFalse)})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_workersself.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(interval0.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 True9.2 缓存策略实现减少重复的AI调用和API请求# src/optimization
阅读完成 · 觉得有帮助？
