最近在AI Agent开发领域很多开发者反馈Python基础不扎实导致项目推进困难。吴恩达教授的AI课程结合Python实践是绝佳的学习路径但网上资料分散不成体系。本文整合一套完整的Python入门教程特别针对Agent开发需求包含环境搭建、核心语法、实战案例到Agent项目衔接新手能快速上手有经验的开发者也能查漏补缺。1. Python环境搭建与配置1.1 Python安装详细步骤Python环境是Agent开发的基础正确的安装能避免后续很多兼容性问题。推荐使用Python 3.10及以上版本这是大多数AI框架的兼容要求。Windows系统安装访问Python官网下载页面选择Windows installer (64-bit)运行安装程序时务必勾选Add Python to PATH选项选择自定义安装确保pip和IDLE等组件被选中完成安装后打开命令提示符验证python --versionmacOS系统安装# 使用Homebrew安装推荐 brew install python # 或从官网下载macOS安装包 # 验证安装 python3 --versionLinux系统安装# Ubuntu/Debian sudo apt update sudo apt install python3 python3-pip # CentOS/RHEL sudo yum install python3 python3-pip # 验证安装 python3 --version pip3 --version1.2 环境变量配置详解环境变量配置是初学者常遇到的问题正确配置后可以在任何目录下运行Python。Windows环境变量配置右键此电脑 → 属性 → 高级系统设置点击环境变量在系统变量中找到Path点击编辑添加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 ~/.bashrc1.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.txt2. 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任务{i1}: {task}) # while循环 count 0 while count 5: print(f执行第{count1}次迭代) 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, greetingHello): 打招呼函数 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(fBMI指数: {bmi:.2f}) message greet(Agent开发者) print(message) process_data(1, 2, 3, nameAlice, age25)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(levellogging.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_okTrue) def save_json(self, data, filename): 保存JSON数据 filepath f{self.base_path}/{filename} try: with open(filepath, w, encodingutf-8) as f: json.dump(data, f, ensure_asciiFalse, indent2) print(fJSON数据已保存到: {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, encodingutf-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(fAgent状态已保存: {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_okTrue) for dir_name in directories: dir_path os.path.join(base_path, dir_name) os.makedirs(dir_path, exist_okTrue) 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, indent2) 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(secondsdelay_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, timestampNone): 格式化时间戳 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_urlNone, timeout30): 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, dataNone, paramsNone): 发起HTTP请求 url f{self.base_url}{endpoint} if self.base_url else endpoint try: response self.session.request( methodmethod, urlurl, jsondata, paramsparams, timeoutself.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, datapayload) 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, periods100, freqH) # 创建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, threshold2): 异常值检测 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], labelTemperature) plt.title(Temperature Over Time) plt.ylabel(Temperature (°C)) plt.legend() plt.subplot(2, 1, 2) plt.plot(df[timestamp], df[humidity], labelHumidity, colororange) 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(timeout1) print(fWorker {worker_id} 开始执行任务: {task.task_id}) task.execute() self.completed_tasks.append(task) print(fWorker {worker_id} 完成任务: {task.task_id} - 结果: {task.result}) self.task_queue.task_done() except queue.Empty: continue except Exception as e: print(fWorker {worker_id} 执行任务时出错: {e}) def start(self): 启动调度器 self.is_running True self.workers [] # 创建工作线程 for i in range(self.max_workers): worker threading.Thread(targetself.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_workers2) # 创建各种任务 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_keyNone): self.api_key api_key or os.getenv(OPENAI_API_KEY) self.sessions {} self.message_history [] def create_agent(self, name, instructions, toolsNone): 创建Agent实例 agent { name: name, instructions: instructions, tools: tools or [], created
最近在AI Agent开发领域很多开发者反馈Python基础不扎实导致项目推进困难。吴恩达教授的AI课程结合Python实践是绝佳的学习路径但网上资料分散不成体系。本文整合一套完整的Python入门教程特别针对Agent开发需求包含环境搭建、核心语法、实战案例到Agent项目衔接新手能快速上手有经验的开发者也能查漏补缺。1. Python环境搭建与配置1.1 Python安装详细步骤Python环境是Agent开发的基础正确的安装能避免后续很多兼容性问题。推荐使用Python 3.10及以上版本这是大多数AI框架的兼容要求。Windows系统安装访问Python官网下载页面选择Windows installer (64-bit)运行安装程序时务必勾选Add Python to PATH选项选择自定义安装确保pip和IDLE等组件被选中完成安装后打开命令提示符验证python --versionmacOS系统安装# 使用Homebrew安装推荐 brew install python # 或从官网下载macOS安装包 # 验证安装 python3 --versionLinux系统安装# Ubuntu/Debian sudo apt update sudo apt install python3 python3-pip # CentOS/RHEL sudo yum install python3 python3-pip # 验证安装 python3 --version pip3 --version1.2 环境变量配置详解环境变量配置是初学者常遇到的问题正确配置后可以在任何目录下运行Python。Windows环境变量配置右键此电脑 → 属性 → 高级系统设置点击环境变量在系统变量中找到Path点击编辑添加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 ~/.bashrc1.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.txt2. 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任务{i1}: {task}) # while循环 count 0 while count 5: print(f执行第{count1}次迭代) 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, greetingHello): 打招呼函数 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(fBMI指数: {bmi:.2f}) message greet(Agent开发者) print(message) process_data(1, 2, 3, nameAlice, age25)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(levellogging.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_okTrue) def save_json(self, data, filename): 保存JSON数据 filepath f{self.base_path}/{filename} try: with open(filepath, w, encodingutf-8) as f: json.dump(data, f, ensure_asciiFalse, indent2) print(fJSON数据已保存到: {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, encodingutf-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(fAgent状态已保存: {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_okTrue) for dir_name in directories: dir_path os.path.join(base_path, dir_name) os.makedirs(dir_path, exist_okTrue) 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, indent2) 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(secondsdelay_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, timestampNone): 格式化时间戳 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_urlNone, timeout30): 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, dataNone, paramsNone): 发起HTTP请求 url f{self.base_url}{endpoint} if self.base_url else endpoint try: response self.session.request( methodmethod, urlurl, jsondata, paramsparams, timeoutself.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, datapayload) 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, periods100, freqH) # 创建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, threshold2): 异常值检测 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], labelTemperature) plt.title(Temperature Over Time) plt.ylabel(Temperature (°C)) plt.legend() plt.subplot(2, 1, 2) plt.plot(df[timestamp], df[humidity], labelHumidity, colororange) 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(timeout1) print(fWorker {worker_id} 开始执行任务: {task.task_id}) task.execute() self.completed_tasks.append(task) print(fWorker {worker_id} 完成任务: {task.task_id} - 结果: {task.result}) self.task_queue.task_done() except queue.Empty: continue except Exception as e: print(fWorker {worker_id} 执行任务时出错: {e}) def start(self): 启动调度器 self.is_running True self.workers [] # 创建工作线程 for i in range(self.max_workers): worker threading.Thread(targetself.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_workers2) # 创建各种任务 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_keyNone): self.api_key api_key or os.getenv(OPENAI_API_KEY) self.sessions {} self.message_history [] def create_agent(self, name, instructions, toolsNone): 创建Agent实例 agent { name: name, instructions: instructions, tools: tools or [], created
![[C++]内存管理：串顺序存储的内存回收](http://pic.xiahunao.cn/yaotu/[C++]内存管理：串顺序存储的内存回收)
