玩了几年单片机、开发板，现在想转型AI Agent开发，本文记录作者从Python基础语法到搭建第一个Agent原型的全过程。适合刚学完Python面向对象、想理解Agent核心机制但不知道从何下手的同学

        一、为啥要学Agent开发？？？

        目前大模型API已经普及，但是单纯的“问答机器人”的价值还是太有限了。为啥现在各个大厂都在砸钱进军Agent市场，因为Agent的核心能力相比于大模型来说还是太全面了。Agent的核心能力在于：理解用户意图——>自主决策——>调用工具执行

        而我不想做调包侠，而是想从Agent底层架构开始，一步步往上搭梯子，所以决定先造轮子再造车。

        二、学习路线

        对于已经学会一门编程语言或者刚学会Python的同学来说，我们不可能从头学到尾，一点点啃下来，那样速度太慢了，而且工作可不会等你，所以这条路线只抓刚需，不瞎卷。

        首先Agent开发的Python知识是很聚焦滴！不需要去刷算法题，因为对于我这种能力不足的人来说，卷算法无疑是死路一条，所以我们剑走偏锋，曲线救国。

阶段	内容	目的
1	Python基础语法+面向对象	能独立写类、继承、方法重写
2	Pydantic+类型注解+装饰器	理解Agent框架的源码风格
3	asyncio	Agent并发调用多个工具的命脉
4	第一个Agent项目（同步版）	理解注册-路由-执行机制
5	接入大模型API	让Agent有“脑子”
6	LangChain/LlamaIndex	站在巨人的肩膀上

        三、Python新概念的通用理解

Python 概念	一句话解释	本项目中的体现
类继承	子类继承父类的规范，必须实现指定方法	WeatherHandler 继承 TaskHandler，必须写 can_handle 和 handle
多态	父类引用指向子类对象，调用时自动执行子类版本	Agent 只认 TaskHandler 类型，自动调用具体子类的实现
抽象方法      NotImplementedError	强制约束：子类必须重写，否则运行报错	忘写 handle 就抛异常，防止静默失败
字典 .get()	安全查表：查到返回值，查不到给默认值	mock_data.get(city, "未知天气")，防止 KeyError 崩溃
with open	上下文管理：自动开关文件，异常时也保证关闭	记事本读写文件，不用手动 f.close()
try-except	异常捕获：出错时执行兜底逻辑，程序不崩	文件不存在或损坏时，初始化空列表继续运行
json.load/dump	Python 对象与 JSON 文本的双向转换	记事本数据持久化到本地文件
split(maxsplit=1)	字符串切分，限制切分次数	把用户输入拆成"命令 + 参数"两部分
f-string	字符串格式化，变量直接嵌入	f"{city}天气：{weather}"
set + all()	集合用于白名单，all() 做批量校验	计算器里逐字检查算式是否只含合法字符

        四、Agent工作流程图

        五、代码

        1、先看基类

import json
from datetime import datetime
from abc import ABC, abstractmethod


#==============基类============
"""
注意一下，这里TaskHandler是继承ABC的，一般来说是继承于object的，
但是由于object 是普通基类，子类可以偷懒不实现；
ABC 是抽象基类，子类必须实现所有抽象方法，否则实例化时就报错。
在这个项目里，用 ABC 就是为了
防止有人写 Handler 时
忘记重写 can_handle 和 handle
把错误扼杀在编译/启动阶段。
"""

class TaskHandler(ABC):
    '''所有任务处理器的基类'''

    @abstractmethod  
    def can_handle(self, command: str)->bool:
        """判断是否能执行这个命令"""
        pass
        
    """
    原本的代码应该是这么写的：
        def can_handle(self, command)
            raise NotImplementedError("子类必须实现") //提醒子类必须重写，否则报错

    command: str  这个是新写法，告诉函数这个参数是字符串
    ->bool      告诉函数返回的应该是True或者Faults
    """
  
    @abstractmethod
    def handle(self, args:str)->str:
        """执行具体逻辑，返回结果字符串"""
        pass

    def get_name(self)->str:
        """返回处理器名称（默认实现，子类可重写）"""
        return self.__class__.__name__

        2、基于TaskHandler基类，我们写以下子类

#========子类：具体处理器==========
"""天气查询(模拟)"""
class WeatherHandler(TaskHandler):  #新建一个子类继承父类TaskHandler
    def can_handle(self, command: str) ->bool:
        #父类函数重写,检测用户输入的命令关键词是否是"weather","w","天气"中的其中一个（w是weather的简写，可替换成其他的）
        #"weather","w","天气"这仨是自定义的，可以随便写，我这里是因为这个代码是用来查询天气的，所以用这三个
        return command in ("weather","w","天气")

    def handle(self, city:str) ->str:
        #判定用户是否输入了正确的城市名，从而判定输出的数据
        #如果用户没有输入 城市名 ，则return "请告诉我城市，比如：weather 北京"
        #如果用户没有输入 正确的城市名（字典中的城市） 则输出：  xx天气：未知天气（模拟数据有限）
        #如果输入了正确的城市名，则进行天气的输出，由于这里没有接入天气API，目前只用固定的字典进行应答
        if not city:
            return "请告诉我城市，比如：weather 北京"
        #模拟天气数据（后面学asyncio时，改成调用真实API）
        mock_data = {
            "北京": "晴 25°C",
            "上海": "小雨 22°C",
            "广州": "多云 28°C"
        }
        weather = mock_data.get(city, "未知天气（模拟数据有限）")
        #注意这里的.get()不是get函数，只是恰巧名字一样。这里的  字典.get() 是字典的方法，查字典用的
        #字典.get（key, default）  key:必须填  default:非必填
        # .get()  就是拿着现成的 key（city）去字典里查，查到给真的，查不到给默认值，然后存进  weather ，最后  return  拼起来输出
        return f"{city}天气：{weather}"

class ClacHandler(TaskHandler):
    """计算器"""
    def can_handle(self, command: str) ->bool:
        return command in ("calc","c","计算")

    def handle(self, expression:str) ->str:
        if not expression:
            return "请输入算式，比如： clac 1+2*3"

        try:
            # 安全计算：只允许数字和运算符
            allowed = set("0123456789+-*/.()")#创建一个集合
            if all(c in allowed for c in expression):

                #for c in expression：把算式里的每个字符都拎出来，叫c
                #c in allowed：检查这个字符c在不在白名单里
                #all(...)：所有字符都必须通过检查，有一个不通过就是False

                result = eval(expression)
                #eval()  是 Python 的字符串执行器，把字符串当数学公式算，"1+2*3"
                # 传进去 → 返回  7
                # 然后 f-string 拼成  "结果：7"  返回

                return f"结果：{result}"
            return"算式包含非法字符"

        # try  里的代码如果报错（比如除以0、括号不匹配），不会崩溃，而是跳到这里
        # e是错误信息（比如"division by zero"）
        # 返回"计算错误：division by zero" ，程序继续运行
        #except  是关键字（捕获）， Exception 是所有错误的基类， as e 是把错误信息抓出来存到变量 e 里
        except Exception as e:
            return f"计算错误：{e}"

class NoteHandler(TaskHandler):
    """记事本（持久化到本地）"""
    #这边定义了一个文件叫：notes.json
    FILE = "notes.json"

    def can_handle(self, command: str) ->bool:
        return command in ("note", "n", "记事")

    def handle(self, content:str) ->str:
        if not content:
            return  "请输入内容，比如：note 记得交作业"

        #读取已有记录
        try:
            #打开文件，用完自动关，不用手动close
            #"r"  读模式
            #encoding="utf-8"  指定编码，防止中文乱码

            #这里是用open打开file，然后将打开的file命名为f，用with来保证自动关闭
            with open(self.FILE,"r",encoding="utf-8")as f:
                #json.load(f)  把JSON文件内容变成Python列表/字典
                notes = json.load(f)
        #捕获错误
        except (FileNotFoundError,json.JSONDecodeError):
        #出错时就当没有旧纪录，从空列表开始
            notes = []
        #添加新纪录
        note = {
            "time":datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "content":content
        }
        #追加到列表
        notes.append(note)
        #保存
        with open(self.FILE,"w",encoding="utf-8")as f:
            # json.dump(notes, f) 把Python列表变成JSON字符串写入文件
            # ensure_ascii=False  中文不转义，直接写中文
            # index = 2 格式化缩进，让所有JSON文件有换行和空格，方便人看
            json.dump(notes,f,ensure_ascii=False,indent=2)

        return f"已记录：{content}"
    """
    上面那个版本是用了with方法，下面这个版本不用with
    
    def handler(self, content: str) -> str:
        if not content:
            return "请输入内容，比如：note 记得交作业"
            
        #====读文件====
        notes = []
        try:
            f = open(self.FILE, "r", encoding = "utf-8")  //手动打开
            notes = json.load(f)                          //读取
            f.close()                                     //必须手动关闭
        except (FileNotFoundError,json.JSONDecodeError):
            notes = []
        ////注意：如果上面的json.load(f)报错了，f.close()根本执行不到，文件就会被一直占着
        
        #====造数据====
        note = {
            "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "content": content
        }
        notes.append(note)
        
        # ========== 写文件 ==========
        f = open(self.FILE, "w", encoding="utf-8")  # 1. 手动打开
        json.dump(notes, f, ensure_ascii=False, indent=2)
        f.close()                                     # 2. 必须手动关闭
    
        return f"已记录：{content}" 
        
    不用with的坏处就是：
        1. 
        每次都要记着手动  f.close() ，忘了就内存泄漏
        2. 
        如果  json.load(f)  那行报错了，程序直接跳到  except ， f.close()  永远执行不到
        3. 
        就像你开了串口、读了数据、中途出异常，串口没关，后面再开就报错
    """


class HelpHandler(TaskHandler):
    """帮助"""
    def can_handle(self, command: str) ->bool:
        return command in ("help","h","帮助")

    def handle(self, args:str) ->str:
        return """可用命令：
  weather <城市>  查天气（示例：weather 北京）
  calc <算式>     计算（示例：calc 1+2*3）
  note <内容>     记事（示例：note 记得喝水）
  help            显示帮助
  exit            退出"""

        3、接下来是Agent开发的核心出装

#============核心出装：Agent路由器============
class Agent:
    """任务路由器——————Agent核心机制"""

    def __init__(self):
        #注册所有处理器（后面学框架时，这就是Tool Registry）
        self.handlers = [
            WeatherHandler(),
            ClacHandler(),
            NoteHandler(),
            HelpHandler(),
        ]
    #路由分发
    def route(self, user_input:str)->str:
        """解析用户输入，路由到对应处理器"""
        parts = user_input.strip().split(maxsplit=1)
        #strip()   去掉用户输入字符串前后的空格和换行
        #split(maxsplit = 1)   按照空格切割，最多切一次，分成两段
        if not parts:
            return "请输入命令"

        command = parts[0].lower()  #把切割出来的第一部分赋值给command，并转成小写
        args = parts[1] if len(parts) > 1 else ""   #把第二部分赋值给args  计算parts长度，大于1则args为该parts，否则为空

        #多态：遍历所有处理器，找到能处理的
        for i in self.handlers:
            if i.can_handle(command):
                return i.handle(args)

        return f"未知命令：{command}，输入 help 查看可用命令"

        4、主函数

#========主程序=========
def main():
    agent = Agent()
    print("=====简易任务路由器=====")
    print("输入 help 查看命令， exit 退出")

    while True:
        try:
            user_input = input("\n> ").strip()
            if user_input.lower() in ("exit","quit","退出"):
                print("再见")
                break

            result = agent.route(user_input)
            print(result)

        except KeyboardInterrupt:
            print("\n再见")
            break
        except Exception as e:
            print(f"出错了：{e}")

if __name__ == "__main__":
    main()

        六、项目亮点和感悟

        1. 即插即用的扩展性

        新增功能只需要写一个类，继承 TaskHandler，实现 can_handle + handle，然后在         ​​​​​​ Agent.__init__ 里注册。主程序完全不需要改动。这就是开闭原则：对扩展开放，对修改关闭。

        2. 异常安全设计

   try-except 捕获文件读写异常，防止程序崩溃

   with open 自动关闭文件，防止资源泄漏

   .get() 安全查字典，防止 KeyError

        3. 安全第一

   CalcHandler 里的 eval() 是危险函数，必须先做白名单字符过滤。任何直接执行用户输入的代码，都必须先校验。

        4. 面向对象不是炫技

        很多人学 OOP 只会写"学生管理系统"，但在这个项目里，继承 + 多态 + 抽象方法真正解决了"如何统一调度不同功能模块"的问题。

        七、结语

        目前这是同步版 Agent，所有操作都是阻塞的。下一步：

1. 引入 asyncio：把 handle 改成 async def，让多个工具并发执行

2. 接入真实 API：把 mock_data 换成异步 HTTP 请求（天气、翻译等）

3. 引入大模型：让 Agent 能理解自然语言，而不是死板的命令词

4. 引入向量数据库：给 Agent 加长期记忆（从 JSON 文件升级到 Chroma/Milvus）

Agent 开发的核心不是调 API，而是架构设计能力：如何把"理解意图 → 选择工具 → 执行动作"这套流程抽象成可扩展的代码结构。

当你亲手写完这个任务路由器，再看 LangChain 的 Tool 和 AgentExecutor，会发现它们做的本质上也是这些事——只是更完善、更工程化。

先造轮子，再用框架。理解原理后，工具只是加分项。