玩了几年单片机、开发板,现在想转型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,所有操作都是阻塞的。下一步:
引入 asyncio:把
handle改成async def,让多个工具并发执行接入真实 API:把
mock_data换成异步 HTTP 请求(天气、翻译等)引入大模型:让
Agent能理解自然语言,而不是死板的命令词引入向量数据库:给 Agent 加长期记忆(从 JSON 文件升级到 Chroma/Milvus)
Agent 开发的核心不是调 API,而是架构设计能力:如何把"理解意图 → 选择工具 → 执行动作"这套流程抽象成可扩展的代码结构。
当你亲手写完这个任务路由器,再看 LangChain 的Tool和AgentExecutor,会发现它们做的本质上也是这些事——只是更完善、更工程化。
先造轮子,再用框架。理解原理后,工具只是加分项。