1序列乘法 ：

[None]*capacity ，表示长度为capacity的列表
只要对象属于序列类型（Sequence）且实现了 重复语义，就能用 *：
字符串：“ab” * 3 → “ababab”
元组：(“x”, “y”) * 2 → (“x”, “y”, “x”, “y”)
bytes / bytearray：b"A" * 4 → b"AAAA"
range（3.8+）：range(3) * 2 → range(0, 3, 1) 重复两次的扁平化 range（需 itertools.chain.from_iterable 展开）
非序列（如 set、dict）或可变序列若未实现重复语义，则不能使用 *。

3、可变变量赋值特性

变量之间赋值的是变量的物理地址，即"引用/绑定"

4、del

删除 例： del self.data[0]

5、python的join操作

join为split的逆操作，“/”.join(stack) → “usr/local/bin”（用 / 把 list 里的元素拼接起来

6、os.path.join

os.path.join(‘a’, ‘b’, ‘c’) → 拼接文件系统路径，自动根据操作系统用 \ 或 / 分隔。

7、python切片语法

seq[start:stop:step]，左闭右开，三个参数都可省，省略时 start=0、stop=len、step=1，支持负索引与负步长。
例子： ax.set_xticks(x[::5]) # 每隔5分钟显示一个刻度

8、def fun(*a): 中 * 的作用

*用于收集位置参数（positional arguments）。它允许函数接收任意数量的位置参数(按照位置输入的参数)，并将它们存储在一个元组（tuple）中。

9、def fun( ** a): 中** 的作用

** 用于收集关键字参数（keyword arguments）。它允许函数接收任意数量的关键字参数(对应的参数名=xxx输入的参数)，并将它们存储在一个字典（dict）中。

10、字典的索引：

top =1
bottom =100
chat_pages = {
‘top’: top,
‘bottom’: bottom,

}
chat_pages[“top”]

11、批量创建变量的方法：

1. 创建一个注册于self的字典。2. 的次数建立多个字典，假设次数为idx，可以创建一个dict[idx] = {“val”:myValue}的变量。 3. 此时就可以在别的地方使用所有变量，只需要先访问对应次数的字典，比如访问第一次循环的变量：mydict = dict[idx] ，mydict[“val”]就可以获取val变量

12、astype用于修改数据类型

13、一次性批量更新字典

batch = dict(zip(keys, values)) # 先得到临时字典
data.update(batch) # 再合并

14、批量读取文件名

for file_name in os.listdir(‘…/data/fruit/img’): 之后可结合join组成完整的文件名，也可以不用，视情况而定

14、enumerate的作用

enumerate是 Python 内置的一个函数，主要作用是在遍历一个可迭代对象（比如列表、字符串、元组等）时，同时获得元素的索引和值。例如：

fruits = ['苹果', '香蕉', '橘子']

for index, fruit in enumerate(fruits):
    print(index, fruit)

输出：
0 苹果
1 香蕉
2 橘子

15 Pandas .Series.map(注意与python内置函数map不是同一个)

Series.map(arg)
arg：可以是字典（键对应原元素，值对应映射后元素），或者函数。

作用：

1、用字典把分类编号映射成类别名称。

2、用函数对每个元素做计算或格式转换。

第一个作用比较常用，也容易理解，比如我有一列sex，值为0和1，我想要将0和1置为男、女，就通过可以把enumerate创建一个键值对然后用dict封装，最后用map完成替换即可

16 select_dtypes

分离出分类特征。include改为[‘int64’, ‘float64’]则为数值特征

categorical_features = data.select_dtypes(include=[‘object’]).columns

17 set转list

List = sorted(all_genres) 推荐该方法，保证结果可复现

List = list(all_genres)

18 set 增删元素

增加元素：

• add(elem)：添加单个元素
• update(iterable)：添加多个元素（批量添加）

删除元素：

• remove(elem)：删除指定元素（元素不存在会报错）
• discard(elem)：删除指定元素（元素不存在不报错）
• pop()：随机删除一个元素 # 因为set是基于哈希表的，无序，自然也不存在最后一个元素的说法也不能pop删除最后一个元素
• clear()：清空集合

19 list 增删元素

增加元素：

• append(elem)：添加单个元素到末尾
• extend(iterable)：批量添加元素到末尾
• insert(index, elem)：在指定位置插入单个元素

删除元素：

• remove(elem)：删除首次出现的指定元素
• pop([index])：删除指定位置元素（默认最后一个）
• clear()：清空列表

20 tuple

元组是不可变类型，没有用于增加或删除元素的函数。

21 dict 增删元素

增加元素：

• update([other])：增加或更新一个或多个键值对（批量增加）

删除元素：

• pop(key)：删除指定键并返回对应的值
• popitem()：删除并返回最后一个插入的键值对（Python 3.7+）
• clear()：清空字典

22 zip

作用：把两个（或多个）序列按位配对，得到元组迭代器。

names = ['Ann', 'Bob', 'Cid']
ages  = [20, 25, 30]
for n, a in zip(names, ages):
    print(n, a)
# Ann 20
# Bob 25
# Cid 30

• 长度不一致时，以最短的为准（除非用 itertools.zip_longest）。
• 常与拆包、并行遍历一起用。

23 strip()

去除空格、换行、制表符

stopwords = [x.strip() for x in stopwords] # 遍历stopwords的每一个元素并去除空格、换行符等

24 双端队列deque

来自于py标准库 ：from collections import deque

25 os批量读取文件名

list = []
for i in os.listdir(“xxxx”):
list.append(i)
print(list)

以下为目录操作---------------------

25.1 os.getcwd()

获取当前工作目录

current_dir = os.getcwd() # 返回当前工作目录的绝对路径

25.2 os.chdir()

切换目录

os.chdir(“/path/to/new/directory”)

25.3 os.mkdir()、os.makedirs

创建目录

os.mkdir(“new_folder”) # 创建单层目录（若已存在会报错）

os.makedirs(“a/b/c”) # 递归创建多层目录

25.4 os.rmdir()

删除目录

os.rmdir(“empty_folder”) # 只能删除空目录 # 非空目录需用 shutil 库
import shutil
shutil.rmtree(“non_empty_folder”) # 强制删除目录及内容

25.5 os.listdir()、os.walk()

遍历目录内容

# 列出当前目录下的所有文件和子目录名
items = os.listdir() # 返回列表形式
print(items)

递归遍历目录（含子目录）
for root, dirs, files in os.walk(“target_folder”):
print(f"当前目录：{root}“)
print(f"子目录：{dirs}”)
print(f"文件：{files}")

以下为路径操作----------------------

25.6 os.path.join()

full_path = os.path.join(“folder”, “subfolder”, “file.txt”)

25.7 路径分割与解析

path = “/home/user/docs/file.txt”

目录路径：dir_path = os.path.dirname(path) # /home/user/docs

文件名：file_name = os.path.basename(path) # file.txt

路径与文件分割：directory, filename = os.path.split(path) #(‘/home/user/docs’, ‘file.txt’)

文件名分割：base_name, ext = os.path.splitext(file_name) # (‘file’, ‘.txt’)

25.8 路径存在性检查

is_file = os.path.isfile(“file.txt”) # 是否为文件

is_dir = os.path.isdir(“folder”) # 是否为目录

exists = os.path.exists(“path”) # 路径是否存在

以下为文件与系统操作----------------------

25.9 os.remove()

删除文件‌

os.remove(“file.txt”) # 若文件不存在会报错

25.10 os.rename()

重命名/移动文件‌

os.rename(“old.txt”, “new.txt”) # 也可用于移动文件到其他目录

25.11 os.system()、os.popen()

执行系统命令‌

os.system(“ls -l”) # 执行命令并返回状态码（慎用！） output = os.popen(“date”).read() # 获取命令输出

26 对于空的列表、元组、字典等的识别

直接判断 if book_id is None:是无法识别的，空的列表元组字典不是None

需要使用if not reader_id: 这样可以成功判断

27 上下文管理器：with

上下文管理器：用于控制文件、数据库等的自动关闭

with可以自动关闭文件，例如：

with open 打开文件，当操作完成时无论操作是否成功都会自动关闭文件

with sqlite3.connect()连接数据库，当数据库操作结束时无论什么情况都会自动断连数据库。

28 回调函数

回调函数：就是被当作参数传递给另一个函数，在合适的时机由这个函数调用的函数。
换句话说：
函数 X：提供了“注册回调”的机制（比如 cv2.setMouseCallback）。
函数 Y：作为参数传递给 X，被 X 在事件触发时调用，这个 Y 就是回调函数。

常常和hook 的编程方法一起使用，回调函数很简单来描述就是：回调函数会作为其他函数的参数被使用。

29 除法

1 / —— 真除法（浮点除法）

height / 2

• 无论 height 是整数还是浮点数，结果都会是 浮点数

2 //—— 地板除法（整数除法）

height // 2

• 如果 height 是整数，结果一定是 整数
• 如果 height 是浮点数，结果是 向下取整后的浮点数

3 % 取余

30 sqlite和numpy

numpy的数据类型不能直接传入sqlite，必须转为pyhton的原生int，否则会导致sqlite读错数据类型

31 pandas直接读取sql的查询结果(read_sql_query)

通过read_sql_query

# 构建查询语句，排除face_id列
query = """
SELECT * FROM (
    SELECT * FROM detections WHERE face_id = ?
) AS subquery
"""

# 执行查询
df = pd.read_sql_query(query, conn, params=(face_id,))

32 pycharm如何将自己写的代码识别为包

在文件夹里创建一个"__ init __.py"即可,

创建后就可以通过from xxxx import xxxx的方式使用

33 数组拼接

1. numpy.concatenate

• 功能：最基础的数组拼接函数，可在指定轴上拼接多个数组

• 特点

  ：

  • 要求所有输入数组的形状在非拼接轴上必须一致
  • 需要显式指定拼接轴（axis参数）
  • 可以拼接任意维度的数组

2. numpy.vstack

• 功能：垂直（按行）拼接数组，相当于concatenate(..., axis=0)

• 特点

  ：

  • 不需要指定轴参数，固定沿第一个轴（axis=0）拼接
  • 要求输入数组在除第一个轴外的其他维度上形状必须一致
  • 对于 1D 数组，会自动将其视为行向量处理

numpy.hstack

• 功能：水平（按列）拼接数组，相当于concatenate(..., axis=1)

• 特点

  ：

  • 不需要指定轴参数，固定沿第二个轴（axis=1）拼接
  • 要求输入数组在除第二个轴外的其他维度上形状必须一致
  • 对于 1D 数组，直接按水平方向拼接

34 线性代数运算

“np.linalg” 是 NumPy 中的线性代数模块。它提供了一系列用于线性代数运算的函数，例如矩阵求逆、求解线性方程组、计算特征值和特征向量等。例如，可以使用 “np.linalg.inv” 来计算矩阵的逆矩阵，使用 “np.linalg.solve” 来求解线性方程组。

矩阵运算

• np.linalg.inv(a)：计算方阵a的逆矩阵。例如：

import numpy as np a = np.array([[1, 2], [3, 4]]) print(np.linalg.inv(a))

• np.linalg.pinv(a)：计算矩阵a的伪逆矩阵，对于非方阵或奇异矩阵也能计算。
• np.linalg.matrix_power(a, n)：计算方阵a的n次幂。

求解线性方程组

• np.linalg.solve(a, b)：求解线性方程组ax = b，其中a为系数矩阵，b为常数项向量。例如：

import numpy as np a = np.array([[1, 2], [3, 4]]) b = np.array([5, 6]) print(np.linalg.solve(a, b))

• np.linalg.lstsq(a, b, rcond=None)：计算线性方程组ax = b的最小二乘解。

特征值与特征向量

• np.linalg.eig(a)：计算方阵a的特征值和特征向量。例如：

import numpy as np a = np.array([[1, 2], [3, 4]]) eigenvalues, eigenvectors = np.linalg.eig(a) print(“特征值：”, eigenvalues) print(“特征向量：”, eigenvectors)

• np.linalg.eigh(a)：专门用于计算对称矩阵的特征值和特征向量，效率更高。

矩阵分解

• np.linalg.qr(a)：对矩阵a进行 QR 分解，得到正交矩阵Q和上三角矩阵R。
• np.linalg.svd(a)：对矩阵a进行奇异值分解，得到三个矩阵U、S和V。

其他

• np.linalg.norm(a, ord=None)：计算矩阵或向量的范数。
• np.linalg.det(a)：计算方阵a的行列式。例如：

import numpy as np a = np.array([[1, 2], [3, 4]]) print(np.linalg.det(a))

35try except追溯具体报错行数

import traceback

try:
    # 这里放可能会出错的代码
    result = 10 / 0  # 这行会引发 ZeroDivisionError
    # 或者
    # num = int("abc")  # 这行会引发 ValueError
except (ValueError, ZeroDivisionError) as e:
    # 获取异常的详细信息
    print(f"捕获异常: {e}")
    # 打印异常发生的位置（文件名、行号等）
    print("异常发生在:")
    traceback.print_exc()  # 打印完整的异常追踪信息

36 yield

yield可以循环返回一个生成器，这个迭代是实时的，函数每次执行到yield就返回一个值，主程序就可以直接读取，然后依次等待返回+读取
yield返回的值必须用循环来迭代接收,最好是while True+next的组合用法
如果想要yield、return混用(例如训练中实时yield中间结果，最后return训练完成后的评估结果)，需要通过以下方式：

try:
    while True:
        # 每次迭代获取一个 yield 的 fig
        fig = next(generator)
        fig_list.append(fig)
        # （可选）在 Streamlit 中实时展示
        st.pyplot(fig)
except StopIteration as e:
    # 3. 生成器结束后，获取 return 的值（包装在 e.value 中）
    confusion_matrix, acc, f1 = e.value

生成器函数（用 yield 定义）
生成器表达式（推导式语法 (expr for x in iterable)）
返回迭代器/生成器的标准库函数（map/filter/zip/enumerate/itertools 等）

生成器是迭代器的一种实现。
所有生成器都是迭代器（实现了 __iter__ 和 __next__）。
但并不是所有迭代器都是生成器，比如 iter(range(10)) 返回的是迭代器，不是生成器。

生成器才可以用next()获取下一个元素,迭代器是不行的,迭代器要用for 循环

37 断言机制

断言机制，仅用于调试阶段，验证程序运行到此处必须满足的条件，多用于函数参数的判断
语法：asser 条件 “提示语句”

38 **kwargs (传入关键字参数)

kwargs ={
    "active_fn":active_fn,
    "n_samples":n_samples,
    "dataset_type" :dataset_type,
    'layer_sizes': layer_sizes,
    'learning_rate': learning_rate,
    'batch_size': batch_size,
    "num_epochs": num_epochs
}

if st.button("开始训练"):
    net = visualization_nn.TrainNet(**kwargs)
    

39*args(传入位置参数)

def sum_all(*args):
    """求和：支持1个、多个甚至0个参数"""
    total = 0
    for num in args:  # args是元组，可迭代遍历
        total += num
    return total

# 调用示例
print(sum_all(1))          # 1（单个参数）
print(sum_all(1, 2, 3))    # 6（多个参数）
print(sum_all())           # 0（无参数，args为空元组()）

40 钩子函数(hook function)

Hook（钩子函数） 通常指的是：
在某个框架、库、系统或函数的执行流程里，预留了一些“挂钩点”，用户可以在这些点上注册自己的函数。当特定条件发生或者某个事件被触发时，这个钩子函数就会被自动调用。

新理解：也就是说，hook函数早就已经写好了，只是开发者写好的时候它们没有实际意义，就像一个占位符。而当使用者向hook传入了一个有实际行为的函数（大多数是作为回调函数传入），这个hook就会和之前一样运行，只是这时候hook被传入了使用者自定义的逻辑。

41 (接36 yield) 补充迭代器

所有生成器都可以用next()来获取下一个元素

42 .pkl文件

.pkl文件是python中用于序列化反序列化对象的文件格式，通过import pickle实现
1、 pkl文件本质是二级制文件，可存储python对象(列表、字典、模型、类等)状态信息
2、 序列化就是存储：pickle.dump(data,f)     类似mode ='wb' 
3、反序列化就是加载 ：data = pickle.load(f)    类似mode ='rb'

43 exit()表示退出程序

常用于程序调试，py文件中运行到exit即不再继续往下运行

@装饰器

在 Python 装饰器的概念中：

• 装饰函数（也称为装饰器）：是用于修饰其他函数的函数，它接受一个函数作为参数，并返回一个新的函数（通常是对原函数进行增强或修改后的版本）。
• 被装饰函数：是被装饰器修饰的函数，即作为参数传递给装饰器的函数。

举个简单的例子：

# 这是装饰函数（装饰器）
def my_decorator(func):
    def wrapper():
        print("装饰器添加的功能")
        func()  # 调用被装饰的函数
    return wrapper

# 这是被装饰函数
@my_decorator
def target_function():
    print("这是被装饰的函数")

在这个例子中：

• my_decorator 是装饰函数（装饰器）
• target_function 是被装饰函数

44 在代码层面访问外网

在py文件的开头使用os.environ[“HTTPS_PROXY”] = “http://127.0.0.1:7890” 即可，http://127.0.0.1:7890是个人的代理端口

45 读取json并解析

import json

# 打开 JSON 文件并解析
with open("QA.jsonl", "r", encoding="utf-8") as f:
    # load() 函数将文件内容解析为 Python 字典
    data = json.load(f)[0]

结果取决于具体的json文件，类型可能是列表也可能是dict(或者说一定有dict但不一定是第一层，可能嵌套在列表里)，需要自己用type检查

46在普通函数中异步执行逻辑

在普通 def 函数里，无法直接用 await 调用异步函数（async def），
但你仍然想让异步函数执行而不阻塞主程序，这时就可以用事件循环来“桥接”：

loop = asyncio.get_event_loop_policy().get_event_loop() # 获取当前线程的 事件循环对象（Event Loop）。每个线程可以有自己的事件循环；事件循环是 异步任务的调度中心，负责管理协程的执行和暂停。
audio_data = loop.run_until_complete(text_to_sound.run(sentence)) # 将协程 text_to_sound.run(sentence) 放到事件循环里执行；等待协程执行完成后返回结果;主线程会在这一行等待，但协程内部的 I/O 操作（比如调用 TTS 服务）不会阻塞整个程序的其他异步任务。

效果类似于在async函数中进行 await text_to_sound.run(sentence)

• 步本身不加速操作，它的意义在于在等待的时候还能做其他事情。

47 python库里面的__init__文件里的__all__

__all__ = [
    "AgentState",
    "create_agent",
]

表示有几个类可以import ，例如例子中的 from xxx import AgentState create_agent

49 静态方法

通过装饰器@staticmethod创建。
通过装饰器@staticmethod创建的静态方法，可以通过类直接调用，例如：

class MathUtils:
    @staticmethod
    def add(a, b):
        return a + b
    
    @staticmethod
    def multiply(a, b):
        return a * b

# 调用工具方法
print(MathUtils.add(2, 3))  # 输出: 5
print(MathUtils.multiply(2, 3))  # 输出: 6

此时可以发现，add和multiply函数是没有self这个东西的，也就是说它不能用类的实例，它是不能被复用的。
常常可以用这种方法来创建线程独立的东西，避免复用，每个线程创调用类时使用函数创建的东西都是不同的

50上下文管理器装饰器@contextmanager

作用：让函数支持 with 语法，自动处理 “资源获取” 和 “资源释放” 逻辑，无需手动编写 enter 和 exit 方法。

51defaultdict

from collections import defaultdict的defaultdict
作用：当访问字典中不存在的键时，会自动为该键创建并初始化一个默认值，而不是抛出 KeyError 异常

from collections import defaultdict

# 示例1：默认值为列表（list）
d = defaultdict(list)
d["a"].append(1)  # 键"a"不存在，自动创建空列表[]，再追加1
d["a"].append(2)
print(d)  # defaultdict(<class 'list'>, {'a': [1, 2]})

# 示例2：默认值为整数（int）
d = defaultdict(int)
d["count"] += 1  # 键"count"不存在，自动初始化为0，再加1
print(d["count"])  # 1

# 示例3：默认值为字典（dict）
d = defaultdict(dict)
d["user"]["name"] = "Alice"  # 键"user"不存在，自动创建空字典{}
print(d["user"])  # {'name': 'Alice'}

52 会话与实例管理

1. 不同用户的会话可以与实例绑定，来进行多个实例的创建，以便于异步和多线程**（一般可以通过框架(如FastAPI)做一些基本的异步和多线程）**。通常需要实现：会话和实例的绑定、根据会话获取实例、释放会话、释放实例等功能。
2. 可以通过批量初始化多个实例，存入列表来实现实例池，当用户访问时直接取出空闲实例来绑定用户会话进行与用户的操作，用户退出后归还实例，此方法无法在用户下一次使用时提供操作过程中产生的一些临时变量（如果需要保存则额外处理），但agent的记忆是可以正常保存的，只要thread_id不变即可。
3. 会话直接通过用户的uuid、thread_id、user_id等信息创建即可，可以用json、list、queue等方式储存。

53 线程（具体看pycharm项目）

1. 线程锁：

作用：（做对象/实例管理，因为框架会自动创建多个线程，某些全局对象/实例需要上锁防止多个线程同时操作）

和sql一样，线程也是有锁的，避免多个线程对同一个文件、对象同时进行操作，一般通过threading.Lock()来加锁

2. 线程创建：

monitor_thread = threading.Thread(target=detector.run, args=(thread_id,))
monitor_thread.start()

54 异步（最常用）（具体看pycharm项目）

作用：单个线程内进行某些IO操作的等待过程中不阻塞程序

1. 父级异步 → 子级必须异步异步函数（如async def定义的函数）内部若调用阻塞操作或同步函数，会阻塞整个事件循环，导致异步失效。因此子逻辑必须用异步 API（如asyncio.sleep()而非time.sleep()）或通过loop.run_in_executor()包装同步代码。
2. 异步调用必须在事件循环中执行异步函数无法直接调用（调用仅返回协程对象），必须通过事件循环调度（如asyncio.run()、await、loop.create_task()）。
3. 一般来说await xxxx 表示需要等待xxxx完成，此时可以进行其他操作（具体要进行什么操作由事件循环决定），若这个xxxx是某个函数，则进入子级异步，该子级异步中也可以使用await，那就是再嵌套一个，当子级异步整体逻辑完成后，父级异步的await xxxx 结束，可以继续下面的代码运行。

53 FastAPI的生命周期管理函数

可以通过一个函数来管理后端各个对象的生命周期

关键：yield，表示挂起程序，回到with调用lifespan的程序中，with结束再自动执行yield后面的代码

@asynccontextmanager  # 异步上下文管理器：管理应用的启动和关闭生命周期
async def lifespan(app: FastAPI):
    """
    FastAPI应用生命周期管理函数：
    - 启动阶段：初始化全局资源（ChatBot池、会话管理器、语音管理器）
    - 关闭阶段：清理资源（停止所有语音监听、归还ChatBot实例）
    """
    global chatbot_pool, session_instance_manager, voice_monitor_manager

    # ===== 应用启动时执行 =====
    logging.info("🚀 正在初始化 ChatBot 池...")
    # 创建异步ChatBot池（初始化NLP/CV/Voice三类实例池）
    chatbot_pool = await AsyncChatBotPool.create(
        nlp_size=3,  # NLP池大小：最多3个实例
        cv_size=2,   # CV池大小：最多2个实例
        voice_size=2,  # Voice池大小：最多2个实例
        parallel_init=True  # 并行初始化实例，加速启动
    )
    # 初始化会话实例管理器
    session_instance_manager = SessionInstanceManager()
    # 初始化语音监听管理器（关联ChatBot池）
    voice_monitor_manager = VoiceMonitorManager(chatbot_pool)
    logging.info("✅ ChatBot 池初始化完成")

    yield  # 应用运行阶段（yield前为启动逻辑，yield后为关闭逻辑）

    # ===== 应用关闭时执行 =====
    logging.info("🔒 正在关闭所有语音监听...")
    # 停止所有活跃的语音监听会话
    if voice_monitor_manager and voice_monitor_manager.sessions:
        for thread_id in list(voice_monitor_manager.sessions.keys()):
            await voice_monitor_manager.stop_monitor(thread_id)

    logging.info("🔒 正在清理 ChatBot 池...")
    # 清理ChatBot池（归还所有实例、释放资源）
    if chatbot_pool:
        await chatbot_pool.cleanup()
    logging.info("✅ 应用关闭完成")


# 创建FastAPI应用实例，并绑定生命周期管理函数
app = FastAPI(lifespan=lifespan)

54 异步(gather) （具体看pycharm项目）

一、gather 的核心用途

1. 并发执行多个异步任务，提升效率

gather 最常用的场景是同时启动多个独立的异步任务，等待所有任务完成后统一获取结果，避免逐个等待的串行耗时。例如在多模态 ChatBot 中，同时调用「语音转文字」「意图识别」「知识库检索」三个独立接口：

import asyncio
import aiohttp

async def speech_to_text(audio_data):
    """异步调用语音转文字API"""
    async with aiohttp.ClientSession() as session:
        resp = await session.post("https://api.example.com/stt", data=audio_data)
        return await resp.json()

async def intent_recognition(text):
    """异步调用意图识别API"""
    async with aiohttp.ClientSession() as session:
        resp = await session.post("https://api.example.com/intent", json={"text": text})
        return await resp.json()

async def knowledge_retrieval(query):
    """异步检索知识库"""
    async with aiohttp.ClientSession() as session:
        resp = await session.post("https://api.example.com/retrieval", json={"query": query})
        return await resp.json()

async def process_multi_modal_request(audio_data):
    """并发处理多模态请求"""
    # 用gather并发执行三个异步任务，而非逐个await（串行）
    stt_result, intent_result, retrieval_result = await asyncio.gather(
        speech_to_text(audio_data),
        intent_recognition("placeholder"),  # 先占位，实际可依赖stt_result
        knowledge_retrieval("placeholder")
    )
    # 后续处理：结合结果生成响应
    return {"stt": stt_result, "intent": intent_result, "retrieval": retrieval_result}

55 GIL

python自带的全局锁，避免线程出现竞态条件，虽然由于GIL的存在各个线程不会出现问题，但是对于某些情况仍然要加锁，比如创建任务，不能让两个线程去创建同一个会话，虽然它们不会竞态条件冲突，但它们创造的同一个会话会冲突。

56 uv

新一代的包管理器，更适合管理项目各种依赖的版本，强于poetry和pip

57 StringIO

StringIO可直接将字符串模拟为文件读取，例如pd.readcsv(StringIO(“xxxxx”))

58 GRAPH RAG

知识图谱，实体、关系。

59 exec

exec 是 Python 内置函数，作用是：

执行一段 Python 代码（字符串形式）

60 .parquet

.parquet可以由pandas的read_csv读取

61 nginx

具体的端口通常就是后端所在的地方，niginx可以把一个域名分发给后端所在的端口

比如把www.xxxx.com分发到127.0.0.1：3200 和127.0.0.1：2513

62 报错解析

除了靠AI解读之外，

可尝试直接点到对应的源码，也并不是特别难理解，很多时候由于源码的try except机制导致无法给出完全正确的错误解析，需要自己看一看。

63 Docker Compose

核心概念

通过一个 YAML 文件（通常叫 docker-compose.yml）来配置应用的所有服务，然后用一条命令就能启动所有容器。

不用 Compose 的痛点

需要手动启动多个容器，例如redis、db、web、backend等

使用 `docker-compose

一条命令启动所有服务,例如：

# docker-compose.yml
version: '3'
services:
  db:
    image: postgres
    environment:
      POSTGRES_PASSWORD: secret
  
  redis:
    image: redis
  
  web:
    build: .
    ports:
      - "8000:8000"
    depends_on:
      - db
      - redis

# 一条命令启动所有服务
docker-compose up

64 argparse.ArgumentParser(管理命令行参数)

argparse.ArgumentParser用于管理用户运行时允许的命令行参数。

例如，以下表示创建ArgumentParser

    parser = argparse.ArgumentParser(
        description="检查模型部署是否成功的工具",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
使用示例:
  python scripts/check_deployment_cn.py --base-url http://localhost:8000/v1 --apikey your-key --model autoglm-phone-9b
  python scripts/check_deployment_cn.py --base-url http://localhost:8000/v1 --apikey your-key --model autoglm-phone-9b --messages-file custom.json
        """,
    )

第二步：通过以下操作进行添加参数：

parser.add_argument(
    "--base-url",
    type=str,
    required=True,
    help="API 服务的 base URL，例如: http://localhost:8000/v1",
)

第三步，使用parse_args进行参数检查，检查是否输入正确。

args = parser.parse_args()

使用方法：直接将上述内容写入代码的开头，或者if name == “main”:的开头，当使用命令允许该函数时自动触发。

65 ABC(抽象基类)

新的补充：更规范的说法是它规范了一种"接口"，一般结合组合的编程思路，形成一种接口+组合的现代编程思路来代替传统的继承。

ABC 是 abc 模块（Python 内置）中 ABCMeta（抽象元类）和 ABC（抽象基类）的统称，class X(ABC): 本质是让类 X 成为「抽象基类」，具备强制子类实现特定方法的能力，用抽象基类定义一套必须遵守的接口，强制子类实现关键方法，避免因方法缺失导致的运行时错误，同时让代码结构更清晰。

比如pytorch的dataset类就必须要实现两个方法，就是用的这个ABC

再例如：

from abc import ABC, abstractmethod

# 抽象基类：继承ABC
class BaseDataProcessor(ABC):
    # 抽象方法：子类必须实现
    @abstractmethod
    def load_data(self, path):
        # 无具体实现，仅定义接口规范
        pass

    @abstractmethod
    def process(self):
        pass

    # 普通方法：子类可直接复用（可选）
    def save_result(self, result, path):
        print(f"保存结果到 {path}")
        # 通用保存逻辑...

# 子类1：实现抽象方法（合法）
class Neo4jDataProcessor(BaseDataProcessor):
    def load_data(self, path):
        # 实现Neo4j数据加载逻辑
        print(f"从{path}加载Neo4j数据")
    
    def process(self):
        # 实现Neo4j数据处理逻辑
        print("处理Neo4j数据")

# 子类2：未实现抽象方法（非法）
class CsvDataProcessor(BaseDataProcessor):
    def load_data(self, path):
        print(f"从{path}加载CSV数据")
    # 漏写process()方法

# 测试
if __name__ == "__main__":
    # 合法：实例化实现了所有抽象方法的子类
    neo4j_processor = Neo4jDataProcessor()
    neo4j_processor.load_data("neo4j://localhost:7687")
    neo4j_processor.process()
    neo4j_processor.save_result("result.json", "./output")

    # 非法：实例化未实现全部抽象方法的子类，直接报错
    csv_processor = CsvDataProcessor()  # 抛出TypeError

66 @property

@property 是 Python 内置的装饰器，核心目的是把类的方法「伪装成属性」 —— 让你可以像访问普通属性一样调用方法，同时还能在「获取 / 修改 / 删除」属性时，隐藏内部逻辑（比如数据校验、计算、权限控制），兼顾属性的易用性和逻辑的封装性。

67 sys.stdout（标准输出流对象）

接收程序中 print()、sys.stdout.write() 等操作的输出，默认指向「控制台 / 终端」。

简单说：

• 你在代码里写 print("hello")，本质就是把字符串 "hello" 写入 sys.stdout；
• 默认情况下，sys.stdout 会把内容输出到「终端窗口」（比如 PyCharm 的运行面板、CMD 命令行）；
• sys.stdout 是一个可替换的对象（这是关键），你可以把它指向文件、内存缓冲区等，实现「输出重定向」。

当需要获取本应输出在终端的信息时，需要临时劫持sys.stdout，实际上就是对sys.stdout进行赋值，在赋值前一般需要将sys.stdout先被赋值为一个对象进行保存，用于后续恢复

例如：

def _run_code(self, code: str, result_dict: dict, safe_globals: dict) -> None:
    """
    由于Agent运行代码时需要获取对应的结果，所以需要通过sys.stdout相关操作将原本应输出到控制台的内容获取过来。
    """
    original_stdout = sys.stdout  # 将sys.stdout被赋值为original_stdout
    try:
        output_buffer = StringIO() # 创建一个内存缓冲区
        sys.stdout = output_buffer # 劫持sys.stdout，意义为让代码输出不要输出到控制台而是写入刚才创建的内存缓冲区(output_buffer)
        exec(code, safe_globals, safe_globals) # 执行传入的code代码
        result_dict["observation"] = output_buffer.getvalue() # 读取内存缓冲区的结果并保存
        result_dict["success"] = True # 表示成功
    except Exception as e:
        result_dict["observation"] = str(e) #若执行失败，保存错误
        result_dict["success"] = False # 表示失败
    finally:
        sys.stdout = original_stdout # 恢复sys.stdout，让后续的输出回到控制台

68 StringIO()

用于创建内存缓冲区，可以在内存中直接进行IO，对于不需要长期保存的数据可以优化运行速度。

69 Pydantic

举例：

llm: LLM = Field(default_factory=LLM, description="Language model instance")

解释：
在当前 Pydantic 数据模型中，定义一个名为 llm 的字段，它的类型必须是 LLM 类的实例；如果创建模型时没有手动指定 llm 的值，会自动新建一个 LLM 实例作为默认值；该字段的描述是 “Language model instance”。

70 类型注解中的 |

例如：
def func(x: str | int) -> str | None: …
表示：x类型可以是str或int，输出可以是str或None

71 TYPE_CHECKING

# 避免循环导入， if BrowserAgent needs BrowserContextHelper
# 只给静态工具(IDE)看,避免IDE找不到BaseAgent而报错，运行时不会运行到这一段
if TYPE_CHECKING:
    from app.agent.base import BaseAgent  # Or wherever memory is defined


class BrowserContextHelper:
    def __init__(self, agent: "BaseAgent"):   #  agent: "BaseAgent"在代码运行时不严格保证类型正确，这只是给IDE和人看的，当类型不对时IDE会提示错误
        self.agent = agent
        self._current_base64_image: Optional[str] = None

72 python的Optional[]类型注解

例如：
Optional[dict]
表示：
类型可以是dict也可以是None，本质上是Union[dict, None]的语法糖

73 hasattr

hasattr作用：检查一个对象是否拥有指定名称的属性 / 方法
例如：
hasattr(browser_tool, “get_current_state”) 用于检查browser_tool是否存在get_current_state方法

74 __new__

__new__ 是一个特殊方法（魔法方法），核心作用是 创建并返回类的实例对象—— 它是实例化过程的 “第一步”，比我们更熟悉的 __init__ 更早执行。

75 fastAPI挂载路由

挂载管理 API（无前缀）

try:
app.include_router(manage_router)
except Exception as e:
print(f"Failed to include manage_router: {e}")
raise # Fail fast，立即终止启动

挂载工具管理 API（带 /manage 前缀）

try:
    app.include_router(tools_router, prefix="/manage")
except Exception as e:
    print(f"Failed to include tools_router: {e}")
    raise

```
---

## 路由挂载的效果

假设 `manage_router` 定义了：
- `GET /agents`
- `POST /agents`
- `GET /agents/{id}`

挂载后，这些端点会变成主应用的一部分：
```

http://localhost:8000/agents          → manage_router 的 GET /agents
http://localhost:8000/agents/{id}     → manage_router 的 GET /agents/{id}

```
如果使用 `prefix="/manage"` 挂载 `tools_router`：
```

http://localhost:8000/manage/tools    → tools_router 的 /tools
http://localhost:8000/manage/config   → tools_router 的 /configa

76 Literal

typing.Literal直接注解参数仅能取指定值，编辑器会提示、类型检查工具（如 mypy）能校验：

from typing import Literal

# 明确标注status只能是"a"或"b"
async def call_tool(status: Literal["a", "b"]):
    # 可选：仍保留运行时校验，防止绕过类型检查的情况
    if status not in ("a", "b"):
        raise ValueError("status参数仅支持'a'/'b'")
    print(f"当前状态: {status}")

77 断点

断点可以打到源码，不一定只能在自己代码打断点，并且“恢复程序”按钮会执行代码到下一个断电，不必总是点击单步执行

78 本地源码下载

本地源碼下載需要pyproject.toml或setup.py，所以下载的时候找有pyproject.toml或setup.py的目录然后执行"pip install . " 就行了。

79 setattr()

setattr() 是 Python 内置函数，其核心作用是动态地为一个对象（object）设置指定属性（attribute）及其对应的值。

80 getattr（）

getattr（） 是 Python 内置函数，其核心作用是获取对象的内置属性，通过指定名称获取指定的属性。这个方法可以避免对某个属性的直接引用如x.toolname，这种用法，万一输入的参数x没有这个属性，就会报错，而getattr（）不会，如果获取失败可以指定返回None。

例子：
close_fn = getattr(t, “close”, None)

81 type(x).__name

type(x).__name可用于查看x的类型名字，如果用于类型判断，可以判断名字是否为xxx，可平替用isinstance实现的相关逻辑。

82 callable

callable()是 Python 内置函数，其核心作用是检查（判断）一个对象是否是 “可调用” 的，简单来说就是判断该对象是否能够被使用 () 括号进行调用。

例：
if callable(close_fn):

83 inspect.iscoroutine(coro)

inspect.iscoroutine(coro)用于判断coro是否是协程对象，即是否是async def 定义 函数的返回结果。

84 any

any()是 Python 内置标准函数，其核心作用是判断一个可迭代对象（iterable）中是否存在至少一个 “为真”（真值，truthy）的元素。

85 id()

id() 是 Python 内置标准函数，它的核心作用是：返回对象的 “身份标识”（唯一标识符），这个标识可以理解为对象在内存中的唯一 “编号”，用于区分不同的对象实例。

使用示例：

# 验证引用一致性：确保 agents 列表就是 os.agents 的引用
if id(agents) != id(AGENT_OS_REF.agents):
    # 引用不一致，强制同步
    AGENT_OS_REF.agents = agents
    agents = AGENT_OS_REF.agents
    print(f"[DEBUG replace_or_add_agent] Fixed reference inconsistency")

86json序列化

接收到json有时候需要做的事情可能就是要先序列化（即存储为字符串或字节流）：便于传输和存储

87 AsyncIterator

AsyncIterator：异步迭代器，类似于同步里面的retrun结尾的普通函数和yield结尾的迭代器函数的区别。

可以用async for 去循环获取结果，并且有状态，会记录记录迭代位置，如当前索引。

88 流式传输/Event

流式传输：流式传输并不是真正的连续传输数据（接近数学上的连续定义），而是密集的离散数据块（Chunks）的连续传输。

Event：流式传输往往伴随着Event，通常作为Response对象的一种属性，用来标记流式传输中当前正在传输的内容是在干什么。比如可能有tool_calling_steat、mcp_start等等事件，这里的Event不参与实际逻辑，只做标记，用于前端分辨内容用途。

在agno中，发送event的方式就是，在流式输出过程中，给run_response的event属性(一个列表)添加event，让前端可以取出run_response的event决定接下来的内容如何渲染。

一般来说，Event通常都用于记录步骤，来分别当前的行为是哪个行为

89 拦截或模拟该域名的请求

如果要检查某个域名是否被访问了，可以在hosts文件那边改：C:\Windows\System32\drivers\etc\hosts。

需要管理员模式打开记事本再打开hosts才有权限更改。

比如我现在发现https://api.cloud.copilotkit.ai域名老是传出来一个报错，然后这个域名又没什么用，就可以尝试把它定向到127.0.0.1来尝试屏蔽它。

90内存和存储(共享资源)

对于共享资源，在编程中不止要思考存储的增删改查，也要考虑内存的增删，不能说在存储更新了某样东西，但内存没删除，这可能会导致某种问题。

但也不是每次都要手动去写删除逻辑，python自己有垃圾回收机制，只是需要保留这个思想，毕竟如果某个资源还对这个旧存在引用，它默认就不会被垃圾回收。

91 raise

在try except中，正常不会报Traceback错误堆栈。但如果在末尾加上raise就会返回错误堆栈，例如：

    ........（省略）
    try:
        response.raise_for_status()
    except requests.exceptions.HTTPError as e:
        print("HTTPError body:", e.response.text)
        raise

    return response.json()
except requests.exceptions.RequestException as e:
    print(f"Request failed: {e}")
    raise

92 asyncio.Queue

asyncio.Queue 是基于 asyncio 的事件循环（Event Loop） 和 Future/Task 机制 实现的，核心流程如下：

1. 消费者执行 await queue.get()：
   • 检查队列是否有数据：如果有，直接取走返回；
   • 如果没有，创建一个 “等待对象（Future）”，并把这个消费者任务加入队列的 “等待列表”，然后将该任务从事件循环的 “就绪队列” 中移除（挂起）。
2. 生产者执行 await queue.put(data)：
   • 把数据放入队列；
   • 检查队列的 “等待列表”：如果有挂起的消费者任务，就把第一个任务重新加入事件循环的 “就绪队列”（唤醒）；
   • 被唤醒的消费者任务会继续执行，取走刚放入的数据。

整个过程是纯事件驱动，没有任何定时检查（轮询），CPU 在消费者挂起期间可以去处理其他任务（比如其他异步函数），直到被明确唤醒。

93 （重要）异步调用的生命周期

1. “在同一个 Task 里 async for / await 下游，就把自己的生命周期绑定在下游上；上游被取消，下游也会一起被取消。”（例如上游的StreamResponse被取消）
 2. “想让长任务不跟请求共死，就把‘算结果’和‘推结果’拆成两个 Task，用队列/通道解耦。”（让输出结果这个行为去和上游绑定，以做到把执行任务的行为摘出来不和上游绑定，避免第一点的情况。）

94 局部变量

如果期望一个变量百分之百不会滞留在内存中，就可以在函数内使用局部变量，不要注册到self，这在函数结束后就会自动被回收。

95 @dataclass

@dataclass 核心作用:个人认为还有一种作用就是协作开发时方便共享数据类型。

@dataclass 是 Python 的语法糖（来自 dataclasses 模块），核心是简化纯数据载体类的编写，自动为类生成 __init__（初始化）、__repr__（友好打印）、__eq__（字段相等比较）等样板方法，无需手动编写，提高开发效率，专注于数据本身。

96 nginx

Nginx 的反向代理 / 路由分发能力：它决定哪些 API 对外暴露、不同 API 该转发给哪个后端服务，相当于前端和后端之间的 “API 网关”。

97 AST

AST 是源代码经过 “语法分析” 后，生成的一种结构化、抽象化的树形数据结构

重点可以用于解析LLM的字符串，让LLM生成的字符串变成可以运行的对象

例子：

def parse_action(response: str) -> dict[str, Any]:
    """
    从模型响应中解析动作。
    :arg：response：来自模型的原始响应字符串。
    :return：解析后的动作字典。
    :raise：ValueError：如果无法解析响应。
    """
    try:
        response = response.strip()
        if response.startswith("do"):
            # 使用 AST 用于语法解析，避免代码注入风险(例如避免do(exec("删除所有文件")))
            try:
                # tree.body返回的是ast.call(即对「函数调用」的结构化描述)
                tree = ast.parse(response, mode="eval")
                if not isinstance(tree.body, ast.Call):
                    raise ValueError("Expected a function call")

                call = tree.body  # tree.body返回的是ast.call(即对「函数调用」的结构化描述)
                # 安全地提取关键字参数
                action = {"_metadata": "do"}  # _metadata设置为do
                for keyword in call.keywords:
                    # 这里期望的是'action'
                    key = keyword.arg  # 提取解析出的键(参数名，这里期望的是'action')
                    # 这里期望的是'action的类型'
                    value = ast.literal_eval(keyword.value)  # 提取解析出的值(参数的值，这里期望的是'action的类型')
                    action[key] = value  # 键值对映射
                logging.info(action,"*"*300)
                return action  # 期望的结果是{"_metadata": "xxx","action","xxxxx"}
            except (SyntaxError, ValueError) as e:
                raise ValueError(f"Failed to parse do() action: {e}")

        elif response.startswith("finish"):
            action = {
                "_metadata": "finish",
                "message": response.replace("finish(message=", "")[1:-2],
            }
        else:
            raise ValueError(f"Failed to parse action: {response}")
        return action
    except Exception as e:
        raise ValueError(f"Failed to parse action: {e}")

98 Agent的多租户

单个agent多租户的实现（agno的做法：根据DB里面已建好的agent，deep_copy一份来使用，对于DB、memory等重资源，使用共享的资源，暂不知道用户不使用之后还要不要销毁？）

99 全局单例模式开发

def __new__(cls):
    if cls._instance is None:
        cls._instance = super().__new__(cls)
        cls._instance._initialized = False
    return cls._instance

用于保证类实例的全局单例，避免重复创建多个实例

100 python装饰器开发

装饰器本质是闭包函数，利用 Python “函数是一等对象” 的特性：

• 外层函数接收被装饰的函数 / 类作为参数
• 内层函数实现增强逻辑（比如计时、日志、异常捕获），并调用原函数
• 最后返回内层函数，替换原函数

例如：

import time
# 1. 定义装饰器函数（外层接收被装饰函数）
def timer_decorator(func):
    # 2. 内层函数实现增强逻辑（*args/**kwargs 适配任意参数的函数）
    def wrapper(*args, **kwargs):
        start_time = time.time()  # 增强逻辑：记录开始时间
        result = func(*args, **kwargs)  # 调用原函数，保留返回值
        end_time = time.time()  # 增强逻辑：记录结束时间
        print(f"函数 {func.__name__} 耗时：{end_time - start_time:.4f} 秒")
        return result  # 返回原函数的结果
    # 3. 返回内层函数（替换原函数）
    return wrapper

# 2. 使用装饰器（@+装饰器名，放在函数定义上方）
@timer_decorator
def my_func(n):
    """一个简单的循环函数"""
    total = 0
    for i in range(n):
        total += i
    return total

# 测试
my_func(1000000)
# 输出：函数 my_func 耗时：0.0458 秒

101 闭包函数

闭包是嵌套函数中内层函数引用外层变量，且外层函数返回内层函数，使得内层函数脱离外层作用域后仍能访问并保留这些变量的状态。

简单来讲，闭包函数就像它的名字，把这个函数包装起来了，调用的时候不带括号，当闭包函数作为数据传输时，这个函数的内部数据会被一起打包传输，就像Agent框架里面的took函数。

例如：

def outer(n):          # 外层函数定义变量n
    def inner(x):      # 内层函数引用外层的n
        return x * n
    return inner       # 外层返回内层函数（不是调用）

# 闭包保留n=2，即使outer已执行完毕
double = outer(2)
print(double(5))       # 输出：10（inner仍能拿到n=2）

102 .gitignore && .git/info/exclude

特性	.gitignore（项目级忽略）	.git/info/exclude（本地仓库级忽略）
作用范围	对「所有克隆该仓库的人」生效（全局共享）	仅对「当前本地仓库」生效（私人独享，不共享）
是否被 Git 追踪	可以被 git add 加入版本控制，随仓库同步	不会被 Git 追踪，仅存于当前本地仓库，不同步
配置位置	仓库根目录（或子目录），文件名固定 .gitignore	仓库内 .git/info/exclude（隐藏目录，需手动打开）
适用场景	项目通用忽略（如编译产物、日志、环境配置）	个人本地临时忽略（如个人编辑器配置、本地测试文件）
你的配置示例	若把 /aaa_todo/ 写在这里，所有人克隆仓库都会忽略该目录	你当前把 /aaa_todo/ 写在这里，仅你自己的本地仓库忽略，其他人克隆仓库不会忽略

103 获取某个函数的所有位置参数

agent_param=Agent.init.code.co_varnames

104 Node js体系的常见文件

src表示源码，dist是编译后的可执行代码（非源码），index是模块的入口文件

105 GraphQL

REST API的平替？

106 请求体

请求体，即request库的get/post/put函数中的json参数。用于向API传递参数

请求不一定必须要有请求体，有可能参数都在params传递
在fastapi中，params和请求体都可以用来向API传递参数，区别相对不大

107 suppress

suppress 是 Python 标准库 contextlib 中的上下文管理器，用于优雅忽略指定类型的异常；
核心优势是替代 try-except Pass，让代码更简洁、可读性更高；
使用时需明确指定要忽略的异常类型，避免滥用导致隐藏 bug。

108 Post/get/put的区别

POST/GET/PUT 核心区别对比表

特性	GET	POST	PUT
核心语义	读取 / 查询资源（查）	创建 / 提交资源（创）	更新 / 替换资源（全量改）
幂等性	幂等（多次请求结果一致）	非幂等（多次请求可能创建多份）	幂等（多次请求结果一致）
数据传输位置	仅通过 URL 参数（Params）	主要通过请求体（Body）	主要通过请求体（Body）
数据可见性	数据暴露在 URL 中	数据在请求体中，URL 不可见	数据在请求体中，URL 不可见
数据容量	受 URL 长度限制（KB 级）	无限制（可传大 / 复杂数据）	无限制（可传大 / 复杂数据）
缓存支持	可被浏览器 / 服务器缓存	默认不缓存	默认不缓存
安全性	低（明文暴露）	高（HTTPS 下加密）	高（HTTPS 下加密）
RESTful 示例	/users?page=1（查用户列表）	/users（创建新用户）	/users/123（全量更新 ID=123 的用户）

109 Agent的工具并发执行

工具的并发执行需要一个支持并行函数调用的模型。例如，OpenAI模型有一个parallel_tool_calls参数（默认启用），该参数允许同时请求和执行多个工具调用。

110 调用自己的API（ASGI内部调用）

transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://agno.local", timeout=30.0) as client:
    for db_id in _AGNO_DB_IDS:
        with suppress(Exception):
            await client.post("/metrics/refresh", params={"db_id": db_id}, headers=headers)