Python零基础到AI Agent开发：环境搭建、核心语法与实战项目

最近在AI Agent开发领域，很多开发者反馈Python基础不扎实导致项目推进困难。吴恩达教授的AI课程结合Python实践是绝佳的学习路径，但网上资料分散不成体系。本文整合一套完整的Python入门教程，特别针对Agent开发需求，包含环境搭建、核心语法、实战案例到Agent项目衔接，新手能快速上手，有经验的开发者也能查漏补缺。

1. Python环境搭建与配置

1.1 Python安装详细步骤

Python环境是Agent开发的基础，正确的安装能避免后续很多兼容性问题。推荐使用Python 3.10及以上版本，这是大多数AI框架的兼容要求。

Windows系统安装：

1. 访问Python官网下载页面，选择Windows installer (64-bit)
2. 运行安装程序时务必勾选"Add Python to PATH"选项
3. 选择自定义安装，确保pip和IDLE等组件被选中
4. 完成安装后，打开命令提示符验证：python --version

macOS系统安装：

# 使用Homebrew安装（推荐） brew install python # 或从官网下载macOS安装包 # 验证安装 python3 --version

Linux系统安装：

# Ubuntu/Debian sudo apt update sudo apt install python3 python3-pip # CentOS/RHEL sudo yum install python3 python3-pip # 验证安装 python3 --version pip3 --version

1.2 环境变量配置详解

环境变量配置是初学者常遇到的问题，正确配置后可以在任何目录下运行Python。

Windows环境变量配置：

1. 右键"此电脑" → 属性 → 高级系统设置
2. 点击"环境变量"
3. 在系统变量中找到Path，点击编辑
4. 添加Python安装路径和Scripts路径，例如：
   • C:\Users\用户名\AppData\Local\Programs\Python\Python310\
   • C:\Users\用户名\AppData\Local\Programs\Python\Python310\Scripts\

macOS/Linux环境变量配置：

# 编辑shell配置文件（~/.bashrc, ~/.zshrc等） echo 'export PATH="$HOME/.local/bin:$PATH"' >> ~/.bashrc echo 'export PATH="/usr/local/bin:$PATH"' >> ~/.bashrc source ~/.bashrc

1.3 虚拟环境管理

虚拟环境是Python项目管理的必备技能，能有效隔离不同项目的依赖。

# 创建虚拟环境 python -m venv my_agent_env # 激活虚拟环境 # Windows: my_agent_env\Scripts\activate # macOS/Linux: source my_agent_env/bin/activate # 安装包到虚拟环境 pip install requests numpy # 退出虚拟环境 deactivate # 导出环境依赖（用于项目共享） pip freeze > requirements.txt # 从文件安装依赖 pip install -r requirements.txt

2. Python基础语法精讲

2.1 变量与数据类型

Python是动态类型语言，但理解数据类型对Agent开发至关重要。

# 基本数据类型 name = "Agent" # 字符串 age = 25 # 整数 height = 175.5 # 浮点数 is_ai = True # 布尔值 # 容器类型 fruits = ["apple", "banana", "orange"] # 列表 person = {"name": "Alice", "age": 30} # 字典 colors = {"red", "blue", "green"} # 集合 coordinates = (10, 20) # 元组 # 类型转换演示 number_str = "123" number_int = int(number_str) # 字符串转整数 number_float = float(number_str) # 字符串转浮点数 print(f"姓名: {name}, 年龄: {age}") print(f"水果列表: {fruits[0]}") # 访问列表元素 print(f"人物信息: {person['name']}") # 访问字典值

2.2 控制流与循环结构

控制流是编程逻辑的核心，Agent决策系统大量使用这些结构。

# if-elif-else条件判断 temperature = 25 if temperature > 30: print("天气炎热") elif temperature > 20: print("天气舒适") else: print("天气凉爽") # for循环遍历 tasks = ["数据收集", "模型训练", "结果分析"] for i, task in enumerate(tasks): print(f"任务{i+1}: {task}") # while循环 count = 0 while count < 5: print(f"执行第{count+1}次迭代") count += 1 # 循环控制语句 for i in range(10): if i == 3: continue # 跳过当前迭代 if i == 7: break # 退出循环 print(i)

2.3 函数定义与使用

函数是代码复用的基础，在Agent开发中用于封装特定功能。

# 基本函数定义 def calculate_bmi(weight, height): """计算BMI指数""" bmi = weight / (height ** 2) return bmi # 带默认参数的函数 def greet(name, greeting="Hello"): """打招呼函数""" return f"{greeting}, {name}!" # 可变参数函数 def process_data(*args, **kwargs): """处理可变数量参数""" print(f"位置参数: {args}") print(f"关键字参数: {kwargs}") # Lambda表达式（匿名函数） square = lambda x: x ** 2 numbers = [1, 2, 3, 4, 5] squared_numbers = list(map(lambda x: x**2, numbers)) # 函数调用示例 bmi = calculate_bmi(70, 1.75) print(f"BMI指数: {bmi:.2f}") message = greet("Agent开发者") print(message) process_data(1, 2, 3, name="Alice", age=25)

3. 面向对象编程（OOP）基础

3.1 类与对象概念

面向对象编程是构建复杂Agent系统的关键技术。

class AIAgent: """AI代理基类""" # 类属性（所有实例共享） agent_count = 0 def __init__(self, name, capability): """构造函数""" self.name = name # 实例属性 self.capability = capability self.is_active = False AIAgent.agent_count += 1 def activate(self): """激活代理""" self.is_active = True print(f"{self.name}已激活") def perform_task(self, task): """执行任务""" if self.is_active: return f"{self.name}正在执行: {task}" else: return "代理未激活，请先激活" @classmethod def get_agent_count(cls): """类方法：获取代理数量""" return cls.agent_count @staticmethod def validate_capability(capability): """静态方法：验证能力""" valid_capabilities = ["分析", "预测", "决策"] return capability in valid_capabilities # 创建对象实例 analysis_agent = AIAgent("分析助手", "分析") prediction_agent = AIAgent("预测引擎", "预测") # 调用方法 analysis_agent.activate() result = analysis_agent.perform_task("数据趋势分析") print(result) print(f"当前代理数量: {AIAgent.get_agent_count()}")

3.2 继承与多态

继承是代码复用的重要机制，在Agent系统中用于创建特化代理。

class SpecializedAgent(AIAgent): """特化代理，继承自AIAgent""" def __init__(self, name, capability, specialization): super().__init__(name, capability) self.specialization = specialization self.tools = [] def add_tool(self, tool): """添加工具""" self.tools.append(tool) print(f"已添加工具: {tool}") def perform_task(self, task): """重写父类方法""" if self.is_active: tool_list = "、".join(self.tools) if self.tools else "无" return f"{self.name}使用工具[{tool_list}]执行: {task}" else: return "代理未激活" def get_specialization_info(self): """子类特有方法""" return f"{self.name}的专业领域: {self.specialization}" # 使用继承 data_agent = SpecializedAgent("数据代理", "分析", "数据处理") data_agent.activate() data_agent.add_tool("Pandas") data_agent.add_tool("NumPy") result = data_agent.perform_task("数据清洗") print(result) print(data_agent.get_specialization_info())

4. Python高级特性

4.1 异常处理机制

健壮的异常处理是生产级Agent系统的必备特性。

import logging # 配置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') class SafeAgent: """带有异常处理的安全代理""" def __init__(self, name): self.name = name self.logger = logging.getLogger(name) def safe_divide(self, a, b): """安全的除法运算""" try: result = a / b self.logger.info(f"除法运算成功: {a} / {b} = {result}") return result except ZeroDivisionError: self.logger.error("除零错误：除数不能为零") return None except TypeError as e: self.logger.error(f"类型错误: {e}") return None finally: self.logger.info("除法运算执行完成") def process_data_safely(self, data_list): """安全处理数据""" results = [] for data in data_list: try: # 模拟数据处理 if not isinstance(data, (int, float)): raise ValueError("数据必须是数值类型") processed = data * 2 # 简单的数据处理 results.append(processed) except ValueError as e: self.logger.warning(f"数据验证失败: {e}") continue # 继续处理下一个数据 except Exception as e: self.logger.error(f"处理数据时发生未知错误: {e}") break # 严重错误，停止处理 return results # 使用示例 agent = SafeAgent("安全处理器") # 测试异常处理 print(agent.safe_divide(10, 2)) # 正常情况 print(agent.safe_divide(10, 0)) # 除零错误 print(agent.safe_divide("10", 2)) # 类型错误 # 批量数据处理 data = [1, 2, "3", 4, "invalid"] results = agent.process_data_safely(data) print(f"处理结果: {results}")

4.2 文件操作与数据持久化

Agent系统需要频繁进行文件读写和数据存储。

import json import csv import pickle class DataManager: """数据管理器类""" def __init__(self, base_path="./data"): self.base_path = base_path import os os.makedirs(base_path, exist_ok=True) def save_json(self, data, filename): """保存JSON数据""" filepath = f"{self.base_path}/{filename}" try: with open(filepath, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) print(f"JSON数据已保存到: {filepath}") except Exception as e: print(f"保存JSON失败: {e}") def load_json(self, filename): """加载JSON数据""" filepath = f"{self.base_path}/{filename}" try: with open(filepath, 'r', encoding='utf-8') as f: return json.load(f) except FileNotFoundError: print(f"文件不存在: {filepath}") return None except Exception as e: print(f"加载JSON失败: {e}") return None def save_agent_state(self, agent, filename): """使用pickle保存Agent状态""" filepath = f"{self.base_path}/{filename}" try: with open(filepath, 'wb') as f: pickle.dump(agent, f) print(f"Agent状态已保存: {filepath}") except Exception as e: print(f"保存Agent状态失败: {e}") def load_agent_state(self, filename): """加载Agent状态""" filepath = f"{self.base_path}/{filename}" try: with open(filepath, 'rb') as f: return pickle.load(f) except Exception as e: print(f"加载Agent状态失败: {e}") return None # 使用示例 manager = DataManager() # 保存配置数据 config = { "agent_name": "智能助手", "version": "1.0", "capabilities": ["分析", "预测", "决策"], "settings": { "timeout": 30, "max_retries": 3 } } manager.save_json(config, "agent_config.json") # 加载配置 loaded_config = manager.load_json("agent_config.json") print(f"加载的配置: {loaded_config}") # 模拟Agent状态保存 class SimpleAgent: def __init__(self, name): self.name = name self.history = [] agent = SimpleAgent("测试代理") agent.history.append("任务1完成") agent.history.append("任务2进行中") manager.save_agent_state(agent, "agent_state.pkl") # 加载状态 loaded_agent = manager.load_agent_state("agent_state.pkl") if loaded_agent: print(f"加载的Agent: {loaded_agent.name}") print(f"历史记录: {loaded_agent.history}")

5. 常用标准库详解

5.1 os和sys系统库

系统操作是Agent与环境交互的基础。

import os import sys import platform class SystemInspector: """系统信息检查器""" def get_system_info(self): """获取系统信息""" info = { "操作系统": platform.system(), "系统版本": platform.version(), "Python版本": platform.python_version(), "工作目录": os.getcwd(), "用户名": os.getenv('USER') or os.getenv('USERNAME'), "PATH环境变量": os.getenv('PATH', '').split(os.pathsep) } return info def explore_directory(self, path="."): """探索目录结构""" print(f"探索目录: {os.path.abspath(path)}") try: items = os.listdir(path) for item in items: item_path = os.path.join(path, item) if os.path.isfile(item_path): size = os.path.getsize(item_path) print(f"文件: {item} ({size} bytes)") elif os.path.isdir(item_path): print(f"目录: {item}/") except PermissionError: print("权限不足，无法访问该目录") except FileNotFoundError: print("目录不存在") def create_agent_workspace(self, workspace_name): """创建Agent工作空间""" base_path = f"./{workspace_name}" directories = ["data", "logs", "models", "config"] try: os.makedirs(base_path, exist_ok=True) for dir_name in directories: dir_path = os.path.join(base_path, dir_name) os.makedirs(dir_path, exist_ok=True) print(f"创建目录: {dir_path}") # 创建基础配置文件 config_content = { "workspace": workspace_name, "created": "2024-01-01", "directories": directories } config_path = os.path.join(base_path, "config", "workspace.json") with open(config_path, 'w') as f: json.dump(config_content, f, indent=2) print(f"工作空间 '{workspace_name}' 创建完成") return base_path except Exception as e: print(f"创建工作空间失败: {e}") return None # 使用示例 inspector = SystemInspector() # 获取系统信息 system_info = inspector.get_system_info() for key, value in system_info.items(): if key == "PATH环境变量": print(f"{key}: {value[:3]}...") # 只显示前3个路径 else: print(f"{key}: {value}") # 探索当前目录 inspector.explore_directory() # 创建工作空间 workspace_path = inspector.create_agent_workspace("my_agent_project")

5.2 datetime和time时间处理

时间处理在Agent调度和日志记录中至关重要。

import datetime import time from datetime import timedelta class TimeManager: """时间管理器""" def __init__(self): self.start_time = None def start_timing(self): """开始计时""" self.start_time = datetime.datetime.now() print(f"计时开始: {self.start_time}") def stop_timing(self): """结束计时并返回耗时""" if self.start_time: end_time = datetime.datetime.now() duration = end_time - self.start_time print(f"计时结束: {end_time}") print(f"总耗时: {duration}") return duration return None def schedule_task(self, task_name, delay_seconds): """调度任务""" current_time = datetime.datetime.now() scheduled_time = current_time + timedelta(seconds=delay_seconds) print(f"任务 '{task_name}' 已调度") print(f"当前时间: {current_time.strftime('%Y-%m-%d %H:%M:%S')}") print(f"执行时间: {scheduled_time.strftime('%Y-%m-%d %H:%M:%S')}") time.sleep(delay_seconds) print(f"执行任务: {task_name} - 时间: {datetime.datetime.now().strftime('%H:%M:%S')}") def format_timestamp(self, timestamp=None): """格式化时间戳""" if timestamp is None: timestamp = datetime.datetime.now() formats = { "标准格式": timestamp.strftime("%Y-%m-%d %H:%M:%S"), "文件安全格式": timestamp.strftime("%Y%m%d_%H%M%S"), "可读格式": timestamp.strftime("%A, %B %d, %Y at %I:%M %p"), "ISO格式": timestamp.isoformat() } return formats def calculate_time_differences(self, time1, time2): """计算时间差""" if isinstance(time1, str): time1 = datetime.datetime.fromisoformat(time1) if isinstance(time2, str): time2 = datetime.datetime.fromisoformat(time2) difference = time2 - time1 result = { "总秒数": difference.total_seconds(), "总天数": difference.days, "小时数": difference.total_seconds() / 3600, "详细分解": { "天": difference.days, "秒": difference.seconds % 60, "分": (difference.seconds // 60) % 60, "时": difference.seconds // 3600 } } return result # 使用示例 timer = TimeManager() # 计时功能测试 timer.start_timing() time.sleep(2) # 模拟工作负载 duration = timer.stop_timing() # 时间格式化演示 formats = timer.format_timestamp() for format_name, formatted_time in formats.items(): print(f"{format_name}: {formatted_time}") # 时间差计算 start_time = "2024-01-01 10:00:00" end_time = "2024-01-02 14:30:15" difference = timer.calculate_time_differences(start_time, end_time) print("时间差分析:") for key, value in difference.items(): print(f" {key}: {value}")

6. 第三方库集成实战

6.1 requests网络请求库

网络通信是Agent与外部服务交互的基础能力。

import requests from requests.exceptions import RequestException import json class APIAgent: """API调用代理""" def __init__(self, base_url=None, timeout=30): self.base_url = base_url self.timeout = timeout self.session = requests.Session() # 设置通用请求头 self.session.headers.update({ 'User-Agent': 'AIAgent/1.0', 'Content-Type': 'application/json' }) def make_request(self, method, endpoint, data=None, params=None): """发起HTTP请求""" url = f"{self.base_url}{endpoint}" if self.base_url else endpoint try: response = self.session.request( method=method, url=url, json=data, params=params, timeout=self.timeout ) # 检查HTTP状态码 response.raise_for_status() # 尝试解析JSON响应 try: return response.json() except json.JSONDecodeError: return response.text except RequestException as e: print(f"请求失败: {e}") return None def get_public_data(self, api_url): """获取公开API数据""" print(f"获取数据从: {api_url}") data = self.make_request('GET', api_url) if data: print("数据获取成功") return data else: print("数据获取失败") return None def post_data(self, endpoint, payload): """提交数据到API""" print(f"提交数据到: {endpoint}") response = self.make_request('POST', endpoint, data=payload) if response: print("数据提交成功") return response else: print("数据提交失败") return None # 使用示例 api_agent = APIAgent() # 测试公开API（使用JSONPlaceholder） print("=== 测试公开API ===") posts = api_agent.get_public_data('https://jsonplaceholder.typicode.com/posts') if posts and isinstance(posts, list): print(f"获取到 {len(posts)} 篇文章") for post in posts[:3]: # 显示前3篇 print(f"标题: {post['title'][:50]}...") # 模拟数据提交 print("\n=== 测试数据提交 ===") test_data = { "title": "AI Agent测试", "body": "这是由Python Agent发送的测试数据", "userId": 1 } response = api_agent.post_data('https://jsonplaceholder.typicode.com/posts', test_data) if response: print(f"服务器响应: {response}")

6.2 数据处理库基础

NumPy和Pandas是Agent数据处理的基石。

import numpy as np import pandas as pd from typing import List, Dict, Any class DataProcessingAgent: """数据处理代理""" def __init__(self): self.data_history = [] def create_sample_data(self): """创建示例数据集""" # 使用NumPy创建数值数据 np.random.seed(42) # 设置随机种子保证可重复性 # 创建模拟传感器数据 temperature_data = np.random.normal(25, 5, 100) # 平均25度，标准差5 humidity_data = np.random.normal(60, 10, 100) # 平均60%，标准差10 # 创建时间序列 dates = pd.date_range('2024-01-01', periods=100, freq='H') # 创建Pandas DataFrame df = pd.DataFrame({ 'timestamp': dates, 'temperature': temperature_data, 'humidity': humidity_data, 'location': ['Room_A'] * 50 + ['Room_B'] * 50 # 前50个Room_A，后50个Room_B }) print("示例数据集创建完成") print(f"数据形状: {df.shape}") print("\n前5行数据:") print(df.head()) return df def analyze_data(self, df): """分析数据""" print("\n=== 数据分析报告 ===") # 基本统计信息 print("基本统计信息:") print(df[['temperature', 'humidity']].describe()) # 分组统计 print("\n按位置分组统计:") group_stats = df.groupby('location').agg({ 'temperature': ['mean', 'std', 'min', 'max'], 'humidity': ['mean', 'std'] }) print(group_stats) # 数据质量检查 print("\n数据质量检查:") print(f"空值数量: {df.isnull().sum().sum()}") print(f"重复行数: {df.duplicated().sum()}") return group_stats def detect_anomalies(self, df, threshold=2): """异常值检测""" print("\n=== 异常值检测 ===") # 计算Z-score df['temp_zscore'] = np.abs((df['temperature'] - df['temperature'].mean()) / df['temperature'].std()) df['humidity_zscore'] = np.abs((df['humidity'] - df['humidity'].mean()) / df['humidity'].std()) # 检测异常值 temp_anomalies = df[df['temp_zscore'] > threshold] humidity_anomalies = df[df['humidity_zscore'] > threshold] print(f"温度异常值数量: {len(temp_anomalies)}") print(f"湿度异常值数量: {len(humidity_anomalies)}") if len(temp_anomalies) > 0: print("温度异常值详情:") print(temp_anomalies[['timestamp', 'temperature', 'temp_zscore']]) return { 'temperature_anomalies': temp_anomalies, 'humidity_anomalies': humidity_anomalies } # 使用示例 data_agent = DataProcessingAgent() # 创建和分析数据 df = data_agent.create_sample_data() stats = data_agent.analyze_data(df) anomalies = data_agent.detect_anomalies(df) # 演示数据可视化（基础版本） try: import matplotlib.pyplot as plt # 创建简单的时间序列图 plt.figure(figsize=(12, 6)) plt.subplot(2, 1, 1) plt.plot(df['timestamp'], df['temperature'], label='Temperature') plt.title('Temperature Over Time') plt.ylabel('Temperature (°C)') plt.legend() plt.subplot(2, 1, 2) plt.plot(df['timestamp'], df['humidity'], label='Humidity', color='orange') plt.title('Humidity Over Time') plt.ylabel('Humidity (%)') plt.xlabel('Time') plt.legend() plt.tight_layout() plt.savefig('sensor_data.png') print("图表已保存为 'sensor_data.png'") except ImportError: print("Matplotlib未安装，跳过图表生成")

7. Agent开发项目实战

7.1 简单任务调度Agent实现

结合前面所学知识，实现一个基础的任务调度Agent。

import threading import queue import time from abc import ABC, abstractmethod from datetime import datetime from typing import List, Dict, Callable class Task(ABC): """任务基类""" def __init__(self, task_id: str, priority: int = 1): self.task_id = task_id self.priority = priority self.status = "pending" # pending, running, completed, failed self.created_at = datetime.now() self.started_at = None self.completed_at = None self.result = None self.error = None @abstractmethod def execute(self): """执行任务的具体逻辑""" pass def to_dict(self) -> Dict: """转换为字典格式""" return { "task_id": self.task_id, "priority": self.priority, "status": self.status, "created_at": self.created_at.isoformat(), "started_at": self.started_at.isoformat() if self.started_at else None, "completed_at": self.completed_at.isoformat() if self.completed_at else None, "result": self.result, "error": self.error } class CalculationTask(Task): """计算任务""" def __init__(self, task_id: str, numbers: List[float], operation: str): super().__init__(task_id) self.numbers = numbers self.operation = operation def execute(self): """执行计算""" try: self.started_at = datetime.now() self.status = "running" if self.operation == "sum": self.result = sum(self.numbers) elif self.operation == "average": self.result = sum(self.numbers) / len(self.numbers) elif self.operation == "max": self.result = max(self.numbers) elif self.operation == "min": self.result = min(self.numbers) else: raise ValueError(f"不支持的操作: {self.operation}") self.status = "completed" self.completed_at = datetime.now() except Exception as e: self.status = "failed" self.error = str(e) self.completed_at = datetime.now() class DataProcessingTask(Task): """数据处理任务""" def __init__(self, task_id: str, data: List, processing_function: Callable): super().__init__(task_id) self.data = data self.processing_function = processing_function def execute(self): """执行数据处理""" try: self.started_at = datetime.now() self.status = "running" self.result = self.processing_function(self.data) self.status = "completed" self.completed_at = datetime.now() except Exception as e: self.status = "failed" self.error = str(e) self.completed_at = datetime.now() class TaskScheduler: """任务调度器""" def __init__(self, max_workers: int = 3): self.max_workers = max_workers self.task_queue = queue.PriorityQueue() self.completed_tasks = [] self.workers = [] self.is_running = False def add_task(self, task: Task): """添加任务到队列""" # 优先级越高，数字越小（所以用负号） self.task_queue.put((-task.priority, task)) print(f"任务已添加: {task.task_id} (优先级: {task.priority})") def worker_loop(self, worker_id: int): """工作线程循环""" while self.is_running or not self.task_queue.empty(): try: # 获取任务（阻塞，但有超时） priority, task = self.task_queue.get(timeout=1) print(f"Worker {worker_id} 开始执行任务: {task.task_id}") task.execute() self.completed_tasks.append(task) print(f"Worker {worker_id} 完成任务: {task.task_id} - 结果: {task.result}") self.task_queue.task_done() except queue.Empty: continue except Exception as e: print(f"Worker {worker_id} 执行任务时出错: {e}") def start(self): """启动调度器""" self.is_running = True self.workers = [] # 创建工作线程 for i in range(self.max_workers): worker = threading.Thread(target=self.worker_loop, args=(i,)) worker.daemon = True worker.start() self.workers.append(worker) print(f"任务调度器已启动，{self.max_workers} 个工作线程运行中") def stop(self): """停止调度器""" self.is_running = False print("正在停止任务调度器...") # 等待所有任务完成 self.task_queue.join() print("任务调度器已停止") def get_status(self) -> Dict: """获取调度器状态""" return { "is_running": self.is_running, "queue_size": self.task_queue.qsize(), "completed_tasks": len(self.completed_tasks), "active_workers": sum(1 for w in self.workers if w.is_alive()) } # 使用示例 def main(): """主函数演示任务调度""" scheduler = TaskScheduler(max_workers=2) # 创建各种任务 tasks = [ CalculationTask("calc_1", [1, 2, 3, 4, 5], "sum"), CalculationTask("calc_2", [10, 20, 30], "average"), CalculationTask("calc_3", [5, 2, 8, 1, 9], "max"), DataProcessingTask("data_1", [1, 2, 3, 4, 5], lambda x: [i**2 for i in x]), DataProcessingTask("data_2", ["hello", "world"], lambda x: [s.upper() for s in x]) ] # 设置不同优先级 tasks[0].priority = 3 # 低优先级 tasks[1].priority = 1 # 高优先级 tasks[2].priority = 2 # 中优先级 # 添加任务 for task in tasks: scheduler.add_task(task) # 启动调度器 scheduler.start() # 运行一段时间 time.sleep(5) # 检查状态 status = scheduler.get_status() print(f"\n调度器状态: {status}") # 显示完成的任务 print("\n已完成的任务:") for task in scheduler.completed_tasks: task_info = task.to_dict() print(f"- {task_info['task_id']}: {task_info['status']} - 结果: {task_info['result']}") # 停止调度器 scheduler.stop() if __name__ == "__main__": main()

7.2 与OpenAI Agents SDK集成

展示如何将基础Python技能应用到实际的Agent开发框架中。

""" OpenAI Agents SDK 集成示例 注意：需要先安装 openai-agents 包 pip install openai-agents """ import os from datetime import datetime # 模拟Agents SDK的基本使用模式 class MockAgentFramework: """模拟Agent框架，展示集成概念""" def __init__(self, api_key=None): self.api_key = api_key or os.getenv('OPENAI_API_KEY') self.sessions = {} self.message_history = [] def create_agent(self, name, instructions, tools=None): """创建Agent实例""" agent = { 'name': name, 'instructions': instructions, 'tools': tools or [], 'created