在AI Agent（智能体）的开发中，Python是无可替代的核心工具——无论是Agent的任务调度、数据处理、模型交互，还是工具调用，都需要Python作为底层支撑。对于刚入门AI Agent的开发者来说，无需一开始钻研复杂的Agent框架（如LangChain、AutoGPT），先夯实Python核心能力，才能更高效地理解和开发AI Agent。

本文将围绕AI Agent入门所需的Python技能，从基础语法、数据结构、核心算法，到Numpy、Pandas数据处理与可视化，再到代码规范，循序渐进讲解，并结合AI Agent场景举例，让你学完就能用。

一、Python基础：AI Agent的“语法基石”

AI Agent的核心逻辑（如任务执行、状态判断、工具调用）都依赖Python基础语法实现。重点掌握以下内容：

1.1 核心数据类型与变量

AI Agent中常用的数据类型包括：int（任务ID、优先级）、str（任务描述、用户指令）、bool（任务完成状态）、list（任务列表）、dict（Agent配置、任务属性）。

实战示例：定义AI Agent的基础属性
# 1. 基本数据类型：Agent的基础信息
agent_name = "TaskSchedulerAgent"  # str：Agent名称
agent_version = 1.0  # float：版本号
is_active = True  # bool：是否激活
max_tasks = 10  # int：最大并发任务数

# 2. 复合数据类型：Agent的任务列表（每个任务用dict描述）
tasks = [
    {"task_id": 1, "description": "处理用户邮件", "priority": 3, "is_completed": False},
    {"task_id": 2, "description": "生成日报", "priority": 1, "is_completed": True},
    {"task_id": 3, "description": "调用天气API", "priority": 2, "is_completed": False}
]

# 打印Agent信息
print(f"Agent名称：{agent_name}")
print(f"当前激活状态：{is_active}")
print(f"待处理任务数：{sum(1 for task in tasks if not task['is_completed'])}")
输出结果：
Agent名称：TaskSchedulerAgent
当前激活状态：True
待处理任务数：2

1.2 流程控制：Agent的任务逻辑判断

AI Agent需要根据条件执行不同操作（如“任务优先级>2则优先执行”“任务完成则标记状态”），核心依赖if-else（条件判断）、for（遍历任务）、while（循环执行任务）。

实战示例：Agent的任务筛选与执行
# 需求：遍历任务列表，优先执行未完成且优先级≥2的任务
for task in tasks:
    # 条件判断：未完成且优先级≥2
    if not task["is_completed"] and task["priority"] >= 2:
        print(f"正在执行任务：{task['description']}（ID：{task['task_id']}）")
        # 模拟任务执行（修改状态）
        task["is_completed"] = True
        print(f"任务{task['task_id']}执行完成！")
    elif task["is_completed"]:
        print(f"任务{task['task_id']}已完成，跳过执行")
    else:
        print(f"任务{task['task_id']}优先级低（{task['priority']}），暂不执行")
输出结果：
正在执行任务：处理用户邮件（ID：1）
任务1执行完成！
任务2已完成，跳过执行
正在执行任务：调用天气API（ID：3）
任务3执行完成！

1.3 函数与类：Agent的模块化封装

AI Agent通常以“类”的形式实现（如TaskAgent类包含add_task、run_task等方法），函数则用于封装重复逻辑（如“计算任务优先级”“检查任务合法性”）。

实战示例：实现一个基础TaskAgent类
class TaskAgent:
    # 初始化方法：创建Agent时定义名称和最大任务数
    def __init__(self, agent_name, max_tasks=5):
        self.agent_name = agent_name  # Agent名称（实例属性）
        self.max_tasks = max_tasks    # 最大任务数（实例属性）
        self.task_list = []           # 任务列表（实例属性）

    # 方法1：添加任务（带合法性检查）
    def add_task(self, task_id, description, priority):
        # 检查任务数是否超过上限
        if len(self.task_list) >= self.max_tasks:
            return f"错误：{self.agent_name}的任务数已达上限（{self.max_tasks}个）"
        # 检查任务ID是否重复
        for task in self.task_list:
            if task["task_id"] == task_id:
                return f"错误：任务ID {task_id} 已存在"
        # 添加合法任务
        new_task = {
            "task_id": task_id,
            "description": description,
            "priority": priority,
            "is_completed": False
        }
        self.task_list.append(new_task)
        return f"任务 {task_id} 添加成功"

    # 方法2：执行指定ID的任务
    def run_task(self, task_id):
        for task in self.task_list:
            if task["task_id"] == task_id:
                if task["is_completed"]:
                    return f"任务 {task_id} 已完成，无需重复执行"
                # 模拟任务执行
                task["is_completed"] = True
                return f"任务 {task_id}（{task['description']}）执行完成"
        return f"错误：未找到任务ID {task_id}"

    # 方法3：查看所有任务状态
    def show_tasks(self):
        print(f"\n{self.agent_name} 任务列表：")
        for task in self.task_list:
            status = "已完成" if task["is_completed"] else "待执行"
            print(f"ID：{task['task_id']} | 描述：{task['description']} | 优先级：{task['priority']} | 状态：{status}")


# 实例化Agent并测试
if __name__ == "__main__":
    # 创建一个最大任务数为3的Agent
    my_agent = TaskAgent(agent_name="DailyTaskAgent", max_tasks=3)
    
    # 添加任务
    print(my_agent.add_task(1, "写Python笔记", 2))
    print(my_agent.add_task(2, "调试RAG代码", 1))
    print(my_agent.add_task(3, "整理Agent文档", 3))
    print(my_agent.add_task(4, "测试新工具", 2))  # 超过任务上限
    
    # 执行任务
    print(my_agent.run_task(2))
    
    # 查看任务状态
    my_agent.show_tasks()
输出结果：
任务 1 添加成功
任务 2 添加成功
任务 3 添加成功
错误：DailyTaskAgent的任务数已达上限（3个）
任务 2（调试RAG代码）执行完成

DailyTaskAgent 任务列表：
ID：1 | 描述：写Python笔记 | 优先级：2 | 状态：待执行
ID：2 | 描述：调试RAG代码 | 优先级：1 | 状态：已完成
ID：3 | 描述：整理Agent文档 | 优先级：3 | 状态：待执行

二、Python数据结构：AI Agent的“任务管理工具”

AI Agent的核心是“任务调度与数据处理”，而Python内置的数据结构（及collections库）是实现这一功能的关键。重点掌握以下结构：

2.1 常用内置数据结构

数据结构 核心用途（AI Agent场景） 示例代码
list 有序任务列表（可修改） task_list = [task1, task2, task3]
dict 存储任务属性/Agent配置 task = {"id":1, "desc":"查天气", "priority":2}
tuple 存储不可修改的固定信息（如Agent的IP+端口） agent_address = ("127.0.0.1", 8080)
set 去重（如Agent的已处理用户ID） processed_user_ids = {"user001", "user002"}

2.2 collections库：Agent的高级数据结构

collections提供了更贴合Agent场景的结构，如任务队列、多值映射等。

1. deque：高效任务队列（FIFO）

AI Agent的任务调度常需“先进先出”（FIFO），deque的popleft()比list.pop(0)效率高100倍以上。

from collections import deque

# 初始化任务队列（FIFO）
task_queue = deque()

# 添加任务（入队）
task_queue.append({"id":1, "desc":"处理用户查询"})
task_queue.append({"id":2, "desc":"生成报告"})
task_queue.append({"id":3, "desc":"调用数据库"})

# 执行任务（出队）
while task_queue:
    current_task = task_queue.popleft()  # 取出队首任务
    print(f"执行任务：{current_task['desc']}（ID：{current_task['id']}）")

# 输出结果：
# 执行任务：处理用户查询（ID：1）
# 执行任务：生成报告（ID：2）
# 执行任务：调用数据库（ID：3）
2. defaultdict：多任务结果分组

当Agent处理多个用户的任务时，可用defaultdict按“用户ID”分组存储结果。

from collections import defaultdict

# 初始化：key=用户ID，value=该用户的任务结果列表
user_task_results = defaultdict(list)

# 模拟Agent处理任务
tasks = [
    ("user001", "查天气", "北京今日晴"),
    ("user002", "算快递费", "10元"),
    ("user001", "查股票", "XX股涨2%")
]

for user_id, task_desc, result in tasks:
    user_task_results[user_id].append({"task": task_desc, "result": result})

# 查看用户任务结果
for user_id, results in user_task_results.items():
    print(f"\n用户 {user_id} 的任务结果：")
    for res in results:
        print(f"- 任务：{res['task']} | 结果：{res['result']}")
输出结果：
用户 user001 的任务结果：
- 任务：查天气 | 结果：北京今日晴
- 任务：查股票 | 结果：XX股涨2%

用户 user002 的任务结果：
- 任务：算快递费 | 结果：10元

三、核心算法：AI Agent的“决策大脑”

AI Agent的“任务调度”“优先级排序”“结果筛选”依赖基础算法，无需高深算法，掌握以下4类即可：

3.1 遍历算法：处理所有任务/数据

Agent需遍历任务列表、用户数据等，常用for循环（列表/字典遍历）。

示例：遍历Agent的任务列表，统计待执行任务
tasks = [
    {"id":1, "desc":"写代码", "completed":True},
    {"id":2, "desc":"测接口", "completed":False},
    {"id":3, "desc":"改BUG", "completed":False}
]

# 遍历统计待执行任务
pending_tasks = []
for task in tasks:
    if not task["completed"]:
        pending_tasks.append(task["id"])

print(f"待执行任务ID：{pending_tasks} | 数量：{len(pending_tasks)}")
# 输出：待执行任务ID：[2, 3] | 数量：2

3.2 排序算法：任务优先级调度

Agent常需按“优先级”“截止时间”排序任务，用Python内置sorted()（支持自定义排序键）。

示例：按优先级降序执行Agent任务
tasks = [
    {"id":1, "desc":"写文档", "priority":2},
    {"id":2, "desc":"修线上BUG", "priority":5},
    {"id":3, "desc":"测新功能", "priority":3}
]

# 按优先级降序排序（key=lambda函数指定排序字段）
sorted_tasks = sorted(tasks, key=lambda x: x["priority"], reverse=True)

# 按排序后的顺序执行任务
print("按优先级执行任务：")
for task in sorted_tasks:
    print(f"- 优先级{task['priority']}：任务{task['id']}（{task['desc']}）")
输出结果：
按优先级执行任务：
- 优先级5：任务2（修线上BUG）
- 优先级3：任务3（测新功能）
- 优先级2：任务1（写文档）

3.3 搜索算法：查找特定任务/结果

Agent需快速查找“指定ID的任务”“用户的历史结果”，用for循环（线性搜索）或dict（O(1)查找）。

示例：用dict实现任务ID的快速查找
# 用dict存储任务（key=任务ID，value=任务详情），实现O(1)查找
task_dict = {
    1: {"desc":"写Python笔记", "priority":2},
    2: {"desc":"调试RAG", "priority":1},
    3: {"desc":"整理Agent文档", "priority":3}
}

# 查找任务ID=2的详情（无需遍历，直接通过key获取）
target_task_id = 2
if target_task_id in task_dict:
    print(f"找到任务{target_task_id}：{task_dict[target_task_id]}")
else:
    print(f"未找到任务{target_task_id}")

# 输出：找到任务2：{'desc': '调试RAG', 'priority': 1}

3.4 贪心算法：简单任务调度

Agent在资源有限时（如同时只能执行1个任务），用贪心算法“选当前最优”（如优先级最高的任务）。

示例：Agent用贪心算法选择下一个任务
def select_next_task(tasks):
    """贪心算法：选择当前优先级最高的未完成任务"""
    max_priority = -1
    next_task = None
    for task in tasks:
        if not task["completed"] and task["priority"] > max_priority:
            max_priority = task["priority"]
            next_task = task
    return next_task


# 模拟任务列表
tasks = [
    {"id":1, "desc":"写代码", "completed":False, "priority":2},
    {"id":2, "desc":"修BUG", "completed":False, "priority":5},
    {"id":3, "desc":"测接口", "completed":True, "priority":3}
]

# 选择下一个任务
next_task = select_next_task(tasks)
print(f"下一个执行的任务：ID={next_task['id']}，描述={next_task['desc']}，优先级={next_task['priority']}")
# 输出：下一个执行的任务：ID=2，描述=修BUG，优先级=5

四、数据处理库：AI Agent的“数据引擎”

AI Agent需处理数值数据（如传感器数据、模型输入）和结构化数据（如用户日志、任务记录），Numpy（数值计算）和Pandas（结构化数据）是核心工具。

4.1 Numpy：Agent的数值计算工具

Numpy用于高效处理数组（如Agent收集的多维度传感器数据、模型输出的向量），比Python列表快10~100倍。

实战示例：Agent处理传感器数据

假设Agent收集了某设备5分钟内的温度数据（每1分钟1个值），用Numpy计算均值、最大值、最小值，判断设备是否异常（温度>35℃为异常）。

import numpy as np

# 1. 创建Numpy数组：5分钟的温度数据（单位：℃）
temp_data = np.array([32.5, 33.1, 35.2, 34.8, 36.0])

# 2. 数值计算：均值、最大值、最小值
temp_mean = np.mean(temp_data)  # 均值
temp_max = np.max(temp_data)    # 最大值
temp_min = np.min(temp_data)    # 最小值

# 3. 筛选异常数据（温度>35℃）
abnormal_temps = temp_data[temp_data > 35]
abnormal_times = np.where(temp_data > 35)[0] + 1  # 异常时间（第1~5分钟）

# 输出结果
print(f"5分钟温度数据：{temp_data}℃")
print(f"均值：{temp_mean:.2f}℃ | 最大值：{temp_max}℃ | 最小值：{temp_min}℃")
print(f"异常温度：{abnormal_temps}℃ | 对应分钟：{abnormal_times}分钟")
输出结果：
5分钟温度数据：[32.5 33.1 35.2 34.8 36. ]℃
均值：34.32℃ | 最大值：36.0℃ | 最小值：32.5℃
异常温度：[35.2 36. ]℃ | 对应分钟：[3 5]分钟

4.2 Pandas：Agent的结构化数据处理工具

Pandas用于处理表格类数据（如Agent的任务日志、用户交互记录），核心是Series（一维）和DataFrame（二维表格）。

实战示例：Agent分析用户交互日志

假设Agent有一份用户交互日志（user_logs.csv），包含“用户ID、交互时间、交互类型、是否完成”，用Pandas分析：

  1. 读取日志数据；
  2. 统计每个用户的交互次数；
  3. 筛选“查询类”且未完成的交互。
步骤1：准备user_logs.csv数据
user_id,interact_time,interact_type,is_completed
user001,2024-05-01 09:30,查询天气,True
user002,2024-05-01 10:15,生成报告,False
user001,2024-05-01 11:00,查询股票,True
user003,2024-05-01 14:20,计算税费,True
user002,2024-05-01 15:30,查询天气,False
步骤2：Pandas代码实现分析
import pandas as pd

# 1. 读取CSV日志数据（生成DataFrame）
df = pd.read_csv("user_logs.csv")

# 2. 查看数据基本信息
print("数据基本信息：")
print(df.info())  # 查看数据类型、非空值
print("\n前3行数据：")
print(df.head(3))

# 3. 统计每个用户的交互次数（分组聚合）
user_interact_count = df.groupby("user_id").size().reset_index(name="interact_count")
print("\n各用户交互次数：")
print(user_interact_count)

# 4. 筛选“查询类”且未完成的交互（条件筛选）
query_uncompleted = df[(df["interact_type"].str.contains("查询")) & (df["is_completed"] == False)]
print("\n查询类未完成的交互：")
print(query_uncompleted[["user_id", "interact_time", "interact_type"]])
输出结果：
数据基本信息：
<class 'pandas.core.frame.DataFrame'>
RangeIndex: 5 entries, 0 to 4
Data columns (total 4 columns):
 #   Column         Non-Null Count  Dtype 
---  ------         --------------  ----- 
 0   user_id        5 non-null      object
 1   interact_time  5 non-null      object
 2   interact_type  5 non-null      object
 3   is_completed   5 non-null      bool  
dtypes: bool(1), object(3)
memory usage: 288.0+ bytes

前3行数据：
   user_id    interact_time interact_type  is_completed
0  user001  2024-05-01 09:30        查询天气          True
1  user002  2024-05-01 10:15        生成报告          False
2  user001  2024-05-01 11:00        查询股票          True

各用户交互次数：
   user_id  interact_count
0  user001               2
1  user002               2
2  user003               1

查询类未完成的交互：
   user_id    interact_time interact_type
4  user002  2024-05-01 15:30        查询天气

五、基础可视化：AI Agent的“结果展示”

AI Agent的运行状态（如任务完成率、用户交互趋势）需要可视化呈现，Matplotlib是入门首选（简单易上手）。

实战示例：可视化Agent的任务执行情况

用Matplotlib绘制：

  1. 柱状图：各任务类型的执行次数；
  2. 饼图：任务完成率；
  3. 折线图：一周内每天的任务完成数量。
import matplotlib.pyplot as plt
import numpy as np

# 设置中文字体（避免乱码）
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

# ---------------------- 1. 柱状图：各任务类型执行次数 ----------------------
task_types = ["查询天气", "生成报告", "查询股票", "计算税费"]
execute_counts = [12, 8, 5, 7]

plt.subplot(2, 2, 1)  # 2行2列，第1个子图
plt.bar(task_types, execute_counts, color=['#1f77b4', '#ff7f0e', '#2ca02c', '#d62728'])
plt.title("各任务类型执行次数")
plt.xlabel("任务类型")
plt.ylabel("执行次数")
plt.xticks(rotation=45)  # 旋转x轴标签，避免重叠

# ---------------------- 2. 饼图：任务完成率 ----------------------
task_status = ["已完成", "未完成"]
status_counts = [25, 10]
colors = ['#2ca02c', '#d62728']

plt.subplot(2, 2, 2)  # 2行2列，第2个子图
plt.pie(status_counts, labels=task_status, colors=colors, autopct='%1.1f%%')
plt.title("任务完成率")

# ---------------------- 3. 折线图：一周任务完成趋势 ----------------------
days = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"]
completed_tasks = [5, 3, 6, 4, 7, 2, 1]

plt.subplot(2, 1, 2)  # 2行1列，第2个子图（占下方整行）
plt.plot(days, completed_tasks, marker='o', linewidth=2, color='#1f77b4')
plt.title("一周任务完成数量趋势")
plt.xlabel("日期")
plt.ylabel("完成任务数")
plt.grid(True, alpha=0.3)  # 显示网格

# 调整子图间距
plt.tight_layout()

# 保存图片（或直接显示）
plt.savefig("agent_task_analysis.png", dpi=300, bbox_inches='tight')
plt.show()
生成的可视化图表效果：
  • 上方左：柱状图清晰展示“查询天气”任务执行次数最多（12次）；
  • 上方右：饼图显示任务完成率71.4%（25/35）；
  • 下方：折线图显示周五完成任务数最多（7次），周日最少（1次）。

六、Python代码规范：AI Agent的“协作基础”

AI Agent开发常需多人协作（如一人写任务调度、一人写数据处理），遵循PEP8规范能让代码更易读、易维护。核心规则如下：

6.1 核心PEP8规范（必遵守）

规范类别 具体要求 示例（正确/错误）
缩进 用4个空格（禁止用Tab） 正确：def func(): + 4空格缩进代码
错误：def func(): + Tab缩进
命名 函数/变量：小写+下划线（snake_case）
类：首字母大写（CamelCase）		 正确：task_queue（变量）、run_task()（函数）、TaskAgent（类）
错误：TaskQueue（变量）、RunTask()（函数）
空行 函数/类之间空2行
函数内逻辑块之间空1行		 正确：
class A:
  …

def func():
  …
  
  if …:
注释 类/函数：用文档字符串（"""说明"""）
复杂逻辑：用# 注释		 正确：
def add_task():
  """添加任务，返回添加结果"""
  # 检查任务数上限
  if …:
行长度 每行代码不超过79个字符（注释不超过72个） 长表达式用\换行：
result = calculate_task_priority(task_id) \
  + get_user_level(user_id)

6.2 工具自动格式化：Black

手动遵守PEP8效率低，用Black工具可自动格式化代码（无需手动调整缩进、空行）。

使用步骤：
  1. 安装Black：pip install black；
  2. 格式化代码：black your_agent_code.py（会直接修改文件）；
  3. 检查格式：black --check your_agent_code.py（仅检查不修改）。

6.3 规范代码示例（AI Agent工具类）

"""
AI Agent 工具类：提供任务格式校验、优先级计算功能
作者：XXX
日期：2024-05-20
"""

from typing import Dict, Optional  # 类型提示（增强代码可读性）


class TaskUtils:
    """任务工具类：包含任务校验、优先级计算等静态方法"""

    @staticmethod
    def is_valid_task(task: Dict) -> bool:
        """
        校验任务是否合法（必须包含id、desc、priority字段）
        
        Args:
            task: 任务字典，格式如{"id":1, "desc":"查天气", "priority":2}
        
        Returns:
            bool: 合法返回True，否则返回False
        """
        # 检查是否包含必需字段
        required_fields = ["id", "desc", "priority"]
        for field in required_fields:
            if field not in task:
                print(f"错误：任务缺少必需字段 '{field}'")
                return False
        
        # 检查字段类型（id为int，priority为int）
        if not isinstance(task["id"], int):
            print(f"错误：任务ID必须是整数，当前是 {type(task['id'])}")
            return False
        if not isinstance(task["priority"], int) or task["priority"] < 1:
            print(f"错误：优先级必须是≥1的整数，当前是 {task['priority']}")
            return False
        
        return True

    @staticmethod
    def calculate_task_score(task: Dict, user_level: int) -> int:
        """
        计算任务得分（优先级×用户等级，用于任务排序）
        
        Args:
            task: 合法的任务字典
            user_level: 用户等级（1~5）
        
        Returns:
            int: 任务得分
        """
        # 先校验任务合法性
        if not TaskUtils.is_valid_task(task):
            return 0
        
        # 计算得分（优先级×用户等级）
        score = task["priority"] * user_level
        return score


# 测试工具类
if __name__ == "__main__":
    # 测试合法任务
    valid_task = {"id": 1, "desc": "查天气", "priority": 3}
    print(f"合法任务校验结果：{TaskUtils.is_valid_task(valid_task)}")
    print(f"任务得分（用户等级2）：{TaskUtils.calculate_task_score(valid_task, 2)}")

    # 测试非法任务（缺少desc字段）
    invalid_task = {"id": 2, "priority": 2}
    print(f"非法任务校验结果：{TaskUtils.is_valid_task(invalid_task)}")

七、综合实战：实现一个“基础任务调度AI Agent”

将前面的知识点串联，实现一个能“添加任务、按优先级排序、执行任务、统计结果、可视化分析”的AI Agent。

完整代码

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from collections import deque
from typing import Dict, List


class TaskSchedulerAgent:
    """基础任务调度AI Agent：支持任务添加、排序、执行、统计、可视化"""

    def __init__(self, agent_name: str, max_tasks: int = 10):
        self.agent_name = agent_name
        self.max_tasks = max_tasks
        self.task_queue = deque()  # 任务队列（FIFO）
        self.task_history = []     # 任务执行历史

    def add_task(self, task_id: int, desc: str, priority: int, user_id: str) -> str:
        """添加任务，返回添加结果"""
        # 校验任务数上限
        if len(self.task_queue) >= self.max_tasks:
            return f"❌ 任务添加失败：{self.agent_name}已达最大任务数（{self.max_tasks}）"
        
        # 校验任务ID是否重复
        for task in self.task_queue:
            if task["task_id"] == task_id:
                return f"❌ 任务添加失败：任务ID {task_id} 已存在"
        
        # 校验优先级合法性
        if not isinstance(priority, int) or priority < 1:
            return f"❌ 任务添加失败：优先级必须是≥1的整数"
        
        # 添加任务到队列
        new_task = {
            "task_id": task_id,
            "desc": desc,
            "priority": priority,
            "user_id": user_id,
            "is_completed": False,
            "execute_time": None
        }
        self.task_queue.append(new_task)
        return f"✅ 任务添加成功：ID={task_id}（{desc}）"

    def run_tasks_by_priority(self) -> None:
        """按优先级降序执行任务，更新历史记录"""
        if not self.task_queue:
            print(f"⚠️ {self.agent_name} 暂无任务可执行")
            return
        
        # 将队列转为列表，按优先级排序
        task_list = list(self.task_queue)
        task_list_sorted = sorted(task_list, key=lambda x: x["priority"], reverse=True)
        
        # 执行排序后的任务
        print(f"\n🚀 {self.agent_name} 开始按优先级执行任务：")
        for task in task_list_sorted:
            print(f"- 执行任务 ID={task['task_id']}（优先级：{task['priority']}）：{task['desc']}")
            # 模拟执行（标记为完成，记录时间）
            task["is_completed"] = True
            task["execute_time"] = pd.Timestamp.now().strftime("%Y-%m-%d %H:%M:%S")
            # 添加到历史记录
            self.task_history.append(task)
        
        # 清空任务队列（已执行完毕）
        self.task_queue.clear()

    def get_task_statistics(self) -> pd.DataFrame:
        """生成任务统计DataFrame（用户、类型、完成情况）"""
        if not self.task_history:
            print(f"⚠️ {self.agent_name} 暂无任务历史记录")
            return pd.DataFrame()
        
        # 提取历史记录的关键字段
        stats_data = [
            {
                "task_id": task["task_id"],
                "user_id": task["user_id"],
                "desc": task["desc"],
                "priority": task["priority"],
                "execute_time": task["execute_time"],
                "is_completed": task["is_completed"]
            }
            for task in self.task_history
        ]
        
        # 转为DataFrame
        stats_df = pd.DataFrame(stats_data)
        return stats_df

    def visualize_task_stats(self) -> None:
        """可视化任务统计结果：用户任务数、优先级分布、执行时间趋势"""
        stats_df = self.get_task_statistics()
        if stats_df.empty:
            return
        
        # 设置中文字体
        plt.rcParams['font.sans-serif'] = ['SimHei']
        plt.rcParams['axes.unicode_minus'] = False

        # 1. 柱状图：各用户任务数
        plt.subplot(2, 2, 1)
        user_task_count = stats_df.groupby("user_id").size().reset_index(name="count")
        plt.bar(user_task_count["user_id"], user_task_count["count"], color='#1f77b4')
        plt.title("各用户任务执行数量")
        plt.xlabel("用户ID")
        plt.ylabel("任务数")

        # 2. 饼图：任务优先级分布
        plt.subplot(2, 2, 2)
        priority_count = stats_df.groupby("priority").size().reset_index(name="count")
        plt.pie(priority_count["count"], labels=priority_count["priority"], autopct='%1.1f%%')
        plt.title("任务优先级分布")

        # 3. 折线图：任务执行时间趋势（按分钟排序）
        plt.subplot(2, 1, 2)
        stats_df["execute_time"] = pd.to_datetime(stats_df["execute_time"])
        stats_df_sorted = stats_df.sort_values("execute_time")
        plt.plot(
            stats_df_sorted["execute_time"],
            stats_df_sorted["task_id"],
            marker='o', linewidth=2, color='#ff7f0e'
        )
        plt.title("任务执行时间趋势（按时间排序）")
        plt.xlabel("执行时间")
        plt.ylabel("任务ID")
        plt.xticks(rotation=45)
        plt.grid(True, alpha=0.3)

        # 调整布局并保存
        plt.tight_layout()
        plt.savefig(f"{self.agent_name}_stats.png", dpi=300, bbox_inches='tight')
        print(f"\n📊 任务统计图表已保存为：{self.agent_name}_stats.png")
        plt.show()


# ---------------------- 测试AI Agent ----------------------
if __name__ == "__main__":
    # 1. 实例化Agent
    my_scheduler = TaskSchedulerAgent(agent_name="WorkTaskAgent", max_tasks=5)

    # 2. 添加任务
    print("=== 添加任务 ===")
    print(my_scheduler.add_task(1, "写Python文档", 2, "user001"))
    print(my_scheduler.add_task(2, "调试RAG代码", 3, "user002"))
    print(my_scheduler.add_task(3, "整理Agent日志", 1, "user001"))
    print(my_scheduler.add_task(4, "测试任务队列", 2, "user003"))
    print(my_scheduler.add_task(5, "优化优先级算法", 4, "user002"))

    # 3. 按优先级执行任务
    print("\n=== 执行任务 ===")
    my_scheduler.run_tasks_by_priority()

    # 4. 查看任务统计
    print("\n=== 任务统计 ===")
    stats_df = my_scheduler.get_task_statistics()
    print(stats_df)

    # 5. 可视化任务结果
    print("\n=== 可视化任务 ===")
    my_scheduler.visualize_task_stats()

代码说明与输出

  1. 功能覆盖：包含任务添加（校验）、优先级执行、统计（Pandas）、可视化（Matplotlib），串联所有核心知识点；
  2. 输出结果：
    • 添加任务：成功添加5个任务，无重复ID和优先级错误；
    • 执行任务：按优先级4→3→2→2→1的顺序执行；
    • 统计表格：显示每个任务的ID、用户、优先级、执行时间；
    • 可视化图表：生成3个子图，展示用户任务分布、优先级占比、执行时间趋势。

八、总结：AI Agent入门的Python学习路径

通过本文的学习，你已掌握AI Agent开发所需的Python核心技能。后续学习路径可参考：

  1. 巩固基础：熟练使用本文讲解的语法、数据结构、Numpy/Pandas；
  2. 学习Agent框架：基于Python基础，学习LangChain（任务链）、AutoGPT（自主决策）等框架；
  3. 结合大模型：用Python调用OpenAI/文心一言API，实现“大模型+Agent”的智能决策（如工具调用、多轮对话）。

AI Agent的核心是“用Python解决实际问题”，建议多动手实践（如修改本文的TaskSchedulerAgent，添加“任务重试”“超时处理”功能），逐步积累实战经验。

如果在学习过程中有任何问题，欢迎在评论区留言讨论！

# 人工智能 # python # microsoft
Logo
2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

cover

五款AI论文软件深度横评：学术助手选对了，科研效率大不同！

avatar 2048 AI社区
cover

10 个继续教育任务书工具，AI 写作降重推荐

avatar 2048 AI社区
cover

便宜且好用的论文哪里找？

avatar 2048 AI社区