为什么Agent开发要掌握这3个基本库requests负责同步发网络请求asyncio负责异步并发pydantic负责把大模型乱七八糟输出变成规范可使用的数据对象一、requests同步http库1、用处发送同步网络请求调用大模型 API专门用来发网络请求和网上的接口对话。向 LLM 服务发送 POST 请求带上 token、模型参数、对话消息获取接口返回状态码、json 结果处理超时、网络报错2、安装request1用管理员身份打开cmd输入pip install requests2测试是否下载成功打开vscode创建python文件import requests r requests.get(https://httpbin.org/get,timeout10) print(r.status_code)200请求成功401token 密钥错误,没有权限403IP 被拦截、拒绝访问404接口地址写错503上游服务挂掉大模型经常遇到如果输出200代表requests库安装正常、GET请求和timeout超时参数全部生效、网络请求链路没问题3、requests的主要方法方法用法requests.get (url)接口连通性测试对应HTTP中的GETrequests.post ( )提交请求调用大模型 API提交 JSON 对话载荷requests.request ( )构造一个请求底层通用封装1post 关键入参参数作用url大模型接口地址headers存放Authorization: Bearer sk‑xxx鉴权 tokenAPI KEYjson传入 Python 字典自动转为 JSON 请求体model、messages timeout超时时间防止程序卡死每个请求必须配置2响应对象 r 的常用成员成员作用r.status_codeHTTP 状态码r.raise_for_status ( )4xx / 5xx 状态码直接抛出异常r.text原始 UTF‑8 字符串可以用来以字符串类型输出网页内容r.json()把返回 JSON 字符串解析为 Python 字典 dict程序可以里的数据容器用于读取模型返回结果json模块四函数json.load(json文件)字符串 → 字典容器json.loads(json字符串)字符串 → 字典容器json.dumps(字典容器)字典容器 → JSON 字符3方法介绍·r requests.get (url,paramsNone,**kwargs)url网页的链接paramsurl的额外参数字典或者字节流格式可选**kwargs12个控制访问的参数可选requests.get ()内部会自动构造一个Request 对象这个对象保存url、headers、请求体、timeout代表我要发给服务器的请求。rResponse 响应对象里面装着服务器返回全部东西状态码、返回文本r.text、r.json()、响应头。4requests方法是同步阻塞说一句答一句并发批量调用大模型效率差二、asyncio异步编程1、协程coroutine用户态内的上下文切换技术。一个线程在不同代码之间不断切换游走的去运行它的意义就是在一个线程中遇到IO等待时间不会傻傻等待而是利用等待时间去做一些别的任务psrequests的方法是同步的一个线程从响应到结束后才能开始下一个线程而async await就是异步方法解决了requests的同步效率低下问题2、实现协程的方法 --async await关键字Agent开发只用这个1async await 的语法基础coroutine function协程函数 :定义协程函数async def 函数名coroutine object协程对象在调用协程函数的时候得到的返回值就是协程对象await:await 可等待的对象协程对象、Future、Task对象 - IO等待ps: await只在协程函数中用一个函数中可以有多个awaitawait得到返回值后再继续往下async def fun(): pass result fun()注意上面3行代码只是单纯的fun()是不会执行fun()里的代码的只会返回一个协程对象。如果想要运行协程函数中的代码逻辑只能把协程对象交给事件循环来处理【eg1】import asyncio # async def定义一个协程函数fun() async def fun(): print(Hello, World!) # asyncio.sleep(1)是一个协程1秒后自动返回none #执行这个IO的时候队列中只有一个mian()没有其他任务所以等待1秒 await asyncio.sleep(1) print(over) return 返回 #如果没有return则返回None async def main(): print(开始) # 调用fun()就是创建了一个协程对象 # await在main()中进行也就是暂停任务main # 启动执行fun里面代码拿到return返回值再进行下面的 respons1 await fun() print(响应, respons1) # 等上一个await收到fun()的返回值才会进行这一句 print(第一个等待完成) respons2 await fun() print(响应, respons2) if __name__ __main__: asyncio.run(main())2事件循环死循环事件循环也就是 asyncio 的核心调度器一个不断循环的管理程序负责看管所有协程任务根据任务状态标记在 IO 等待间隙切换执行不同协程。通俗的讲事件循环会查看每个任务自带的状态标记任务就绪就放入可执行列表任务执行完毕就放入已完成列表任务处于 IO 等待状态则暂时不处理继续保留在总任务列表中。#伪代码任务列表 [任务1,任务2,任务3,....]while True:可执行的任务列表,已完成的任务列表 去检查任务列表中所有的任务将‘可执行’和‘已完成’的任务返回# 等待操作系统通知哪些IO/定时器就绪了遍历for 就绪任务 in 可执行的任务列表执行已就绪的任务for 已完成任务 in 已经完成的任务列表在任务列表中移除已经完成的任务如果任务列表中的任务都已完成则终止循环import asyncio # 去生成获取一个事件循环如果手动写的事件循环就直接获取循环赋给loop # 如果没有事件循环就创建一个事件循环赋给loop # loop也可以理解为调度器 loop asyncio.get_event_loop() # 执行事件循环筛选就绪任务、已完成任务直到顶层任务全部执行完毕 # 调度器去执行事件循环 loop.run_until_complete(任务) # 额外还要手动关闭循环 loop.close()上面的 3 步是旧写法现在我们只需要一步即可把上面的步骤都包含asyncio.run(任务)asyncio.run(任务)括号里只能接收1个协程对象把这个协程对象自动包装成顶层 Task开始启动事件循环3Task对象ps上面的【eg1】就是一个小的异步编程的代码但是它是串行的只有一个任务 main() ,要想把其他协程也包装成任务让异步编程同时对多个任务进行并发处理就需要Task对象Task对象用于并发调度协程把协程丢进就绪队列asyncio.create_task(协程对象name 任务名 ) # python3.7 之后引用进来的loop.create_task(协程对象)和asyncio.ensure_future() # python3.7之前的import asyncio # async def定义一个协程函数fun() async def fun(): print(Hello, World!) await asyncio.sleep(1) print(over) return 返回 async def main(): print(开始) #创建了2个fun()协程任务fun()函数会被调用两次,task1和task2被加入事件循环的就绪队列中等待被调度执行 task1 asyncio.create_task(fun(),name n1) task2 asyncio.create_task(fun(),name n2) print(协程任务创建完成) #逻辑如下 # 在main()中执行到await task1,暂停main()的执行等待任务task1执行 # 在执行task1的Hello, World!后遇到了sleep(1)暂停了task1的执行返回Noneawait task1暂停事件循环在列表中找可执行任务 # task2也在就绪列表等待执行调度执行task2 # task2在Hello, World!后也遇到了sleep(1)所以又暂停了task2的执行 # 事件循环开始空转等待定时器 # 这时候task1 的sleep(1)时间到了task1继续执行打印over然后返回值返回task1执行完成 # main()中await task1这句恢复执行res1接收task1的返回值返回 # 接下来main()继续执行await task2暂停main()的执行等待任务task2执行 # 此时task2已经执行到sleep(1)时间到了事件循环继续调度执行task2打印over然后返回值返回 res1 await task1 res2 await task2 # task2执行完成main()继续下面的代码 print(响应, res1,res2) if __name__ __main__: asyncio.run(main())优化可以把一个一个创建协程任务的程序代码优化成把创建的协程任务装到一个列表中但是我们上面说了await后只能加可等待的对象协程对象Task任务Futurn列表是不能跟在后面的所以我们加了一个函数 asyncio.wait (任务列表)它返回的是一个Future这样就可以写成 await asyncio.wait(任务列表)然后await会接收2个返回值一个是donedone就是已经跑完的协程函数存储return的值形成一个集合一个是pedding还没跑完的任务集合import asyncio # async def定义一个协程函数fun() async def fun(): print(Hello, World!) await asyncio.sleep(1) print(over) return 返回 async def main(): print(开始) task_list [ asyncio.create_task(fun(),name n1), asyncio.create_task(fun(),name n2) ] print(协程任务创建完成) done, pending await asyncio.wait(task_list) print(响应,done) if __name__ __main__: asyncio.run(main())4Future对象不怎么常用了解就好Future最底层的对象不怎么用但是是Task对象的一些基类就是Task有些功能是靠Future实现的· Task对象内部对await的结果处理是基于Future的。这就是 task1 asyncio.create_task(fun())r await task1 print(r)的原理asyncio.create_task(fun())创建 Task 对象Task 内部自带一个 Future。同时把协程fun()交给事件循环去调度执行。协程fun()运行执行到return 666。Task 捕获这个返回值自动调用 Future 的set_result(666)把返回值存入内部 Future 盒子。await task等待内部 Future 变成完成状态然后取出盒子里的值作为 await 表达式的结果。import asyncio async def fun(f): await asyncio.sleep(1) f.set_result(666) async def main(): #获取当前运行的事件循环 loop asyncio.get_running_loop() #创建一个future对象空盒子如果直接await这个future对象会一直阻塞直到future对象有值 f loop.create_future() #手动配置future的值 await loop.create_task(fun(f)) #await接收返回f的值 data await f print(data) asyncio.run(main())· 使用线程池进程池实现异步操作时用到的对象concurrent.futures.Future用于线程池 / 进程池和asyncio.Future思想相似未来结果占位容器但二者不是同一个类concurrent.futures.Future不支持await协程异步开发使用asyncio.Future。5uvloop只需要简单了解怎么替换不需要深挖代码uvloop是asyncio事件循环的替代方案asyncio是python自带的而uvloop属于第三方在一定程度上可以提高asyncio的循环效率下载uvloop只能在Linux或Mac系统下下载pip install uvloopimport asyncio import uvloop asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()) #asyncio的代码和之前无差别 #内部循环自动化会自动变为uvloop asyncio.run(...)6aiohttpaiohttp异步 HTTP 工具负责发送请求必须搭配asyncio事件循环调度使用aiohttp是基于session对象基于ClientSession类的实例化对象进行网络请求发送# 创建ClientSession类的Session对象,固定用asyncio wthasync withaiohttp.ClientSession() as session: #创建ClientSession对象sessionasync withsession.get(url) as response : #调用session的get方法,把请求的url信息返回给responsere_data awaitresponse.text() # 用await阻塞因为在获取内容的这段事件cpu空闲可以做其他的事等自动解码成文本格式后再继续【eg】我们要下载2张图片用requests.get方法发送请求同步下载就是import requests def downlode_img(url): print(f下载图片{url}中...) #获取url的内容并将内容的相关信息存储到r中 r requests.get(url) #将url以/划分并把url最后一个/后面的一部分作为文件名此时file_name为空文件名为600或400 file_name url.split(/)[-1] #以二进制写入的方式打开刚刚创建的空文件并将r.content写入文件中 with open(file_name, wb) as f: f.write(r.content) print(f图片下载完成保存为{file_name}) if __name__ __main__: url_list [https://picsum.photos/id/237/800/600, https://picsum.photos/id/10/600/400] for url in url_list: downlode_img(url)输出的时候先输出下载图片https://picsum.photos/id/237/800/600中...图片下载完成保存为600间隔几秒后再输出下载图片https://picsum.photos/id/10/600/400中...图片下载完成保存为400用asyncio 异步就是import asyncio import aiohttp async def fetch(session, url): print(发送请求,url) #通过session对象发送get请求服务器收到请求后把相关数据打包返回给response async with session.get(url,verify_sslFalse) as response: #以二进制形式读取response的内容 content await response.content.read() # 用/把二进制url分开取最后一个/后的作为文件名 file_name url.split(/)[-1].jpg # w写入b二进制 # 以二进制写入形式打开当前终端运行目录的文件夹新建以file_name为文件名的空文件 # 新建的文件后把文件对象给f #f就是可以用来操作当前空文件写入等 with open(file_name, wb) as f: #把刚刚读到的链接的二进制内容填给当前file_name文件 f.write(content) print(下载完成,url) async def main(): #创建session对象 async with aiohttp.ClientSession() as session: #url列表把要处理的图片链接放进去 url_list [ https://picsum.photos/id/237/800/600, https://picsum.photos/id/10/600/400, ] #创建任务列表每一轮循环都把fetch()协程对象包装成Task对象加入任务列表 # 将要处理的url作为参数传入 tasks [asyncio.create_task(fetch(session, url)) for url in url_list] # await阻塞main()任务等待task任务完成 await asyncio.wait(tasks) print(所有图片全部下载完成...) if __name__ __main__: asyncio.run(main())发送请求几乎是同时发送的也就是在图1等待的IO过程中发送图2的请求发送请求 https://picsum.photos/id/237/800/600发送请求 https://picsum.photos/id/10/600/400下载也是几乎同时因为在图1下载的过程中中途收到图2的服务器数据图1图2 就会来回快速切换cpu“同时”下载这2张图片下载完成 https://picsum.photos/id/10/600/400下载完成 https://picsum.photos/id/237/800/600三、pydantic模块Pydantic Python 的数据校验 类型转换库。依靠类型标注定义数据长什么样代码运行的时候自动检查数据是否符合规则不对就直接报错类型标注只是提示但是运行并不会报错学习 Pydantic 做 Agent 开发优先掌握pydantic的BaseModel类、Field函数、ValidationError类的用法psAgent 开发专门用它校验大模型返回的 JSON。首先我们拓展一个知识点在我 “面向对象高级” 的文章最后的案例中我提到了抽象方法这个装饰器 abstractmethod 现在我再讲一个装饰器dataclass它自带__init____repr__方法用它定义一个类我们不用初始化属性并且直接print实例对象时输出不再是一串地址而是 “类名属性名1参数1属性名2参数2”属性必须用类型注解但是可变对象列表字典等数据容器的类型注解在这个方法中不能用 容器名容器类型[数据类型] []因为在类中这些可变对象是共享的每个实例对象都可以访问所以如果要给可变对象设置默认值变成不可共享的可变对象就得用容器名:容器类型[数据类型名] field(default_factory容器类型)from dataclasses import dataclass,field dataclass class Person: name:str age:int #field(default_factorylist)每次新建实例的时候调用一次list()生成一个全新列表。 phone:list[str] field(default_factorylist) P1 Person(123,10) p2 Person(234,12) P1.phone.append(1232323) # 容器没有被共享 print(P1) # Person(name123, age10, phone[1232323]) print(p2) # Person(name234, age12, phone[])1、BaseModel类1BaseModel内置了类似dataclass的能力同时自带数据校验如果数据标注的。2但是使用BaseModel的类似dataclass功能时只支持关键字传参,且所有属性都需要传参输出格式为属性1 参数1 属性2 参数2.....除了关键字传参json文件或字典都可以进行pydantic实例的传参3如果想解决所有属性都要传参的问题只需要给参数设置上默认值或者用 数据类型 | None 表示如果没有参数传入默认值就是None4在BaseModel类中内置了一个功能就是定义的可变对象本身就是不被不同实例共享的from pydantic import BaseModel class Person(BaseModel): name:str age:int address:str|None None phone:list[str] [] p1 Person(name 张三,age 10) p2 Person(name 小王,age 12) p2.phone.append(123456677) info {name:王,age:20} #用解包的方式把字典的key和value传给对象 p3 Person(**info) print(p3) import json #先将info字典转化为json格式 info_json json.dumps(info) #再用model_validate_json解析json字符串进行传参 p4 Person.model_validate_json(info_json) print(p4) print(p1) # name张三 age10 addressNone phone[] print(p2) # name小王 age12 addressNone phone[123456677]2、Field函数1Pydanyic中也有Field函数注意Pydantic的Field中F是大写2Pydantic的Field和dataclasses中的功能差不多但是dataclasses中的field主要解决可变默认值pydantic的Field主要做数据校验、限制参数、加描述参数参数说明max_length/min_length字符串长度限制gt / lt大于 / 小于description字段说明调用大模型 API 生成文档时很有用default设置默认值frozenTrue/False是否能被修改from pydantic import BaseModel,Field class Person(BaseModel): name:str Field(max_length 4) age:int Field(lt 100,gt 17,) phone:list[str] Field(default[]) P1 Person(name 123,age 19) p2 Person(name 234,age 20) P1.phone.append(12433242) print(P1) # Person(name123, age10, phone[1232323]) print(p2) # Person(name234, age12, phone[])3、field_validator装饰器在要校验复杂规则时我们会用field_validator装饰器给单个字段自定义校验逻辑field_validator要校验的参数) 必须和 classmethod连在一起用from pydantic import BaseModel,Field,field_validator class Person(BaseModel): name:str Field(max_length 4) age:int Field(lt 100,gt 17,) phone:list[str] Field(default[]) # type: ignore[pylance] # field_validator(phone)中的phone就是我们要校验的东西 field_validator(phone) classmethod # -list[str]为这个方法要返回的类型 def phone(cls,v:list[str]) - list[str]: cls:传入的当前类 v:field_validator(phone)括号中的东西 descrition:这个方法主要用来校验手机号是不是以1开头且11位 for n in v: if (not n.startswith(1))and len(n) ! 11: # raise主动抛出异常中断当前代码执行交给上层 try-except 捕获处理。 raise ValueError(手机号不符合规则) return v P1 Person(name 123,age 19,phone [12345678901]) p2 Person(name 234,age 20) print(P1) # Person(name123, age10, phone[1232323]) print(p2) # Person(name234, age12, phone[])4、ValidationError类ValidationError是Pydantic 框架专属异常。 当模型实例化、数据校验失败时Pydantic 自动抛出这个异常用来一次性汇总所有字段的错误信息。from pydantic import BaseModel, Field, field_validator, ValidationError class Person(BaseModel): name: str age: int Field(lt200) phones: list[str] field_validator(phones) classmethod def check_phone(cls, v:list[str]): for n in v: if len(n)!11 or n[0]!1: raise ValueError(手机号格式错误) return v try: p Person(namewang, age1000, phones[00000]) except ValidationError as e: print(捕获校验异常) print(e.errors())捕获校验异常[{type: less_than, loc: (age,), msg: Input should be less than 200, input: 1000, ctx: {lt: 200}, url: https://errors.pydantic.dev/2.13/v/less_than}, {type: value_error, loc: (phones,), msg: Value error, 手机号格式错误, input: [00000], ctx: {error: ValueError(手机号格式错误)}, url: https://errors.pydantic.dev/2.13/v/value_error}]一次性把问题都写出来