企业官网建设流程全解析

为什么Agent开发要掌握这3个基本库？

requests负责同步发网络请求；

asyncio负责异步并发；

pydantic负责把大模型乱七八糟输出变成规范可使用的数据对象

一、requests（同步http库）

1、用处

发送同步网络请求，调用大模型 API（专门用来发网络请求，和网上的接口对话。）

• 向 LLM 服务发送 POST 请求，带上 token、模型参数、对话消息
• 获取接口返回状态码、json 结果，处理超时、网络报错

2、安装request

（1）用管理员身份打开cmd，输入

pip install requests

（2）测试是否下载成功，打开vscode，创建python文件

import requests r = requests.get("https://httpbin.org/get",timeout=10) print(r.status_code)

• 200：请求成功
• 401：token 密钥错误,没有权限
• 403：IP 被拦截、拒绝访问
• 404：接口地址写错
• 503：上游服务挂掉（大模型经常遇到）

如果输出200，代表requests库安装正常、GET请求和timeout超时参数全部生效、网络请求链路没问题

3、requests的主要方法

（1）post 关键入参

（2）响应对象 r 的常用成员

json模块四函数：

json.load(json文件)：字符串 → 字典容器

json.loads(json字符串)：字符串 → 字典容器

json.dumps(字典容器)：字典容器 → JSON 字符

（3）方法介绍

·r = requests.get (url,params=None,**kwargs)

url：网页的链接；params：url的额外参数，字典或者字节流格式（可选）；**kwargs：12个控制访问的参数（可选）

requests.get ()：内部会自动构造一个Request 对象，这个对象保存：url、headers、请求体、timeout，代表：我要发给服务器的请求。

r：Response 响应对象，里面装着服务器返回全部东西：状态码、返回文本r.text、r.json()、响应头。

（4）requests方法是同步阻塞（说一句答一句），并发批量调用大模型效率差

二、asyncio异步编程

1、协程（coroutine）

用户态内的上下文切换技术。（一个线程在不同代码之间不断切换游走的去运行），它的意义就是在一个线程中遇到IO等待时间，不会傻傻等待，而是利用等待时间去做一些别的任务；

ps：requests的方法是同步的，一个线程从响应到结束后才能开始下一个线程而async & await就是异步方法，解决了requests的同步效率低下问题

2、实现协程的方法 --async & await关键字（Agent开发只用这个）

（1）async & await 的语法基础

• coroutine function（协程函数） :定义协程函数，async def 函数名
• coroutine object（协程对象）：在调用协程函数的时候，得到的返回值就是协程对象
• await:await + 可等待的对象（协程对象、Future、Task对象 -> IO等待）

ps: await只在协程函数中用，一个函数中可以有多个await，await得到返回值后再继续往下

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，开始启动事件循环

（3）Task对象

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的执行返回None，await 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个返回值，一个是done（done就是已经跑完的协程函数，存储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())

（4）Future对象（不怎么常用，了解就好）

Future：最底层的对象，不怎么用，但是是Task对象的一些基类，就是Task有些功能是靠Future实现的

· Task对象内部对await的结果处理是基于Future的。这就是 task1 = asyncio.create_task(fun())，r = await task1 ，print(r)的原理

1. asyncio.create_task(fun())：创建 Task 对象，Task 内部自带一个 Future。同时把协程fun()交给事件循环去调度执行。
2. 协程fun()运行，执行到return "666"。
3. Task 捕获这个返回值，自动调用 Future 的set_result("666")，把返回值存入内部 Future 盒子。
4. 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。

（5）uvloop（只需要简单了解，怎么替换，不需要深挖代码）

uvloop：是asyncio事件循环的替代方案，asyncio是python自带的，而uvloop属于第三方，在一定程度上可以提高asyncio的循环效率

下载uvloop，只能在Linux或Mac系统下下载

pip install uvloop

import asyncio import uvloop asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()) #asyncio的代码，和之前无差别 #内部循环自动化会自动变为uvloop asyncio.run(...)

（6）aiohttp

aiohttp：异步 HTTP 工具，负责发送请求，必须搭配asyncio事件循环调度使用

aiohttp是基于session对象（基于ClientSession类的实例化对象）进行网络请求发送

# 创建ClientSession类的Session对象,固定用asyncio wth

async withaiohttp.ClientSession() as session: #创建ClientSession对象session

async withsession.get(url) as response : #调用session的get方法,把请求的url信息返回给response

re_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_ssl=False) 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类；的用法

ps：Agent 开发专门用它校验大模型返回的 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_factory=list)：每次新建实例的时候，调用一次`list()`生成一个全新列表。 phone:list[str] = field(default_factory=list) P1 = Person("123",10) p2 = Person("234",12) P1.phone.append("1232323") # 容器没有被共享 print(P1) # Person(name='123', age=10, phone=['1232323']) print(p2) # Person(name='234', age=12, phone=[])

1、BaseModel类

（1）BaseModel：内置了类似dataclass的能力，同时自带数据校验（如果数据标注的）。

（2）但是使用BaseModel的类似dataclass功能时，只支持关键字传参,且所有属性都需要传参！！！输出格式为属性1 = 参数1 属性2 = 参数2.....；

除了关键字传参，json文件或字典都可以进行pydantic实例的传参

（3）如果想解决所有属性都要传参的问题，只需要给参数设置上默认值，或者用 数据类型 | None 表示如果没有参数传入，默认值就是None

（4）在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='张三' age=10 address=None phone=[] print(p2) # name='小王' age=12 address=None phone=['123456677']

2、Field函数

（1）Pydanyic中也有Field（）函数（注意Pydantic的Field中F是大写）

（2）Pydantic的Field和dataclasses中的功能差不多，但是dataclasses中的field主要解决可变默认值；pydantic的Field主要做数据校验、限制参数、加描述

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(name='123', age=10, phone=['1232323']) print(p2) # Person(name='234', age=12, 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(name='123', age=10, phone=['1232323']) print(p2) # Person(name='234', age=12, phone=[])

4、ValidationError类

ValidationError是Pydantic 框架专属异常。 当模型实例化、数据校验失败时，Pydantic 自动抛出这个异常，用来一次性汇总所有字段的错误信息。

from pydantic import BaseModel, Field, field_validator, ValidationError class Person(BaseModel): name: str age: int = Field(lt=200) 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(name="wang", age=1000, 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'}]

一次性把问题都写出来