前言
在上一篇文章中,我们建立了这样一个基本的结构,所以我们下一步就是开始做最基础的部分。
创建客户端
是不是以为我要开始创建ChatOpenAI
了?还没有呢。正如我们上一篇文章所说的,一切的一切都起源于httpx.client
。能够直接复用精心设计好的http
线程池那当然是最好的。
于是呢,我们就想着,先创建一个客户端:
1 2 3 4 5 6 7
| client = httpx.Client( base_url = "https://api.example.com/v1", headers = { "Authorization": f"Bearer {API_KEY}" }, timeout = httpx.Timeout(timeout=10), )
|
当然,上面这段是随便乱写的,但是基本上是这么回事。
单例控制
但是这样也就只是创建了一个客户端的逻辑,还没有让所有的任务都用上这个客户端。之前我们说的是复用,也就是说整个项目其实都是只有唯一一个客户端的。
该怎么做呢?
有些有Java
基础的小伙伴应该知道,因为静态变量是全局只有一个的,所以每次调用创建的时候,检查一下静态变量是不是存在,存在就直接引用,没有才创建。
但是Python
似乎不是这么一个逻辑。该怎么办呢?
Python
提供了一个魔法函数,也就是__new__
和__init__
。这俩结合锁一起用,其实就能够线程安全地创建对象了。
我们来试试:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
| class Singleton(object): _instance_lock = threading.Lock() def __new__(cls, *args, **kwargs): if not hasattr(cls, "_instance"): with cls._instance_lock: if not hasattr(cls, "_instance"): cls._instance = object.__new__(cls) return cls._instance def __init__(self, *args, **kwargs): if getattr(self, '_initialized', False): return self._initialized = True
|
看上去没问题。
优化可读性
只不过呢,这样子可读性还差点。为了装逼呢,我们也是直接上pydantic
,代码可以更清晰的同时,也能够用pydantic
机制做很多检查。
1 2 3 4 5 6 7 8 9 10 11 12
| from threading import Lock from pydantic import BaseModel, Field class Singleton(BaseModel): _instance: ClassVar[Any] = None _instance_lock: ClassVar[Lock] = Lock() @classmethod def get_instance(cls, *args, **kwargs): if cls._instance is None: with cls._instance_lock: if cls._instance is None: cls._instance = cls(*args, **kwargs) return cls._instance
|
看上去更可靠了,也更清晰了。
在这里,主要的优化点在于,采用pydantic
之后,我们没有必要再实现__init__
逻辑与__new__
逻辑,我们其实只需要更关心单例模式的实现逻辑。至于没能将单例模式的实现植入__init__
逻辑与__new__
逻辑,这确实不是啥问题。再后续业务代码中可以完成植入。
工厂模式
或者说,你觉得你就是要解决这样一个问题,把get_instance
方法植入到构建实例的过程中,那其实解决办法也很简单:再包装一层,然后用这个包装类去调用构建实例的方法。
当然,这个方法就叫工厂模式,是Java
老哥熟悉得不能再熟悉的家伙事儿了。
由于这基本上就是整个系统中唯一一个Client
专用的Factory
了,所以我也就不区分什么AsyncClientFactory
和SyncClientFactory
了,毕竟有奥卡姆剃刀原则在,我们直接给一个全功能的超级ClientFactory
:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55
| from typing import ClassVar, Optional from threading import Lock
from httpx import Client, AsyncClient, Timeout from pydantic import BaseModel, Field
class ClientFactory(BaseModel): base_url: str = Field(..., description="API Base URL") api_key: str = Field(..., description="API Key") timeout: float = Field(60.0, description="API Timeout (seconds)")
_instance: ClassVar[Optional["ClientFactory"]] = None _instance_lock: ClassVar[Lock] = Lock()
_sync_client: ClassVar[Optional[Client]] = None _async_client: ClassVar[Optional[AsyncClient]] = None _client_lock: ClassVar[Lock] = Lock()
@classmethod def get_instance(cls, **kwargs) -> "ClientFactory": """双重检查锁的线程安全单例""" if cls._instance is None: with cls._instance_lock: if cls._instance is None: cls._instance = cls(**kwargs) return cls._instance
def _ensure_clients(self) -> None: if self._sync_client is None: with self._client_lock: if self._sync_client is None: ClientFactory._sync_client = Client( base_url=self.base_url, headers={"Authorization": f"Bearer {self.api_key}"}, timeout=Timeout(self.timeout) ) ClientFactory._async_client = AsyncClient( base_url=self.base_url, headers={"Authorization": f"Bearer {self.api_key}"}, timeout=Timeout(self.timeout) )
def client(self) -> Client: self._ensure_clients() return self._sync_client
def async_client(self) -> AsyncClient: self._ensure_clients() return self._async_client
|
看上去顺眼多了不是吗?
来试试看:
1 2 3 4 5 6 7 8 9 10 11 12
| if __name__ == "__main__": f1 = ClientFactory.get_instance( base_url="https://api.example.com", api_key="sk-xxxx", timeout=30, ) f2 = ClientFactory.get_instance( base_url="https://api.sakebow.com", api_key="sk-xxxx", timeout=30, ) assert f1.client() is f2.client(), "不一样!"
|
其实虽然看上去是构建了两个对象,实际上,如果没有单例模式的话,其实是构建了$4$个对象,各有$1$个工厂对象和$1$个客户端对象。
但是呢,咱加了单例模式,所以的话,这俩绝对是一样的。不然的话,这个assert
就会抛出一个AssertError
。
当然,最终也是没有抛出。
或者说,你也可以改成:
1
| assert f1.client() is not f1.client(), "一样的!"
|
这样的话他就会抛出一个AssertError
,说明这俩是一个玩意儿了。
这样,咱就有了全局唯一的一个Client
或者AsyncClient
了。
而且,这样创建也相当的方便易懂。
抽取抽象类
或者说,我们有一个更大胆的想法。
既然Client
需要一个base_url
、一个api-key
和一个timeout
,而ChatOpenAI
也同样需要,那,抽出来?
当然没问题:
1 2 3 4 5
| from pydantic import BaseModel, Field class BaseFactory(BaseModel): base_url: str = Field(..., description="API Base URL") api_key: str = Field(..., description="API Key") timeout: float = Field(60.0, description="API Timeout")
|
于是呢,咱的ClientFactory
就可以直接继承这个类了:
1 2 3 4 5 6 7 8 9 10 11 12
| class ClientFactory(BaseFactory): _instance: ClassVar[Optional["ClientFactory"]] = None _instance_lock: ClassVar[Lock] = Lock()
_sync_client: ClassVar[Optional[Client]] = None _async_client: ClassVar[Optional[AsyncClient]] = None _client_lock: ClassVar[Lock] = Lock() """ 下面的方法都没变 """
|
当然,有些聪明人已经注意到了,因为继承了BaseFactory
的关系,重复的那些变量就不需要再声明啦。
但是呢,单例还是由每个子类实现。如果需要整个族都只生成一个类,说实话也没有那么大的必要控制得那么严格。再说了,每个子类功能差别其实很大。比如,ClientFactory
需要给出Client
对象,而其他的大模型模块需要给出ChatOpenAI
对象,所以最好还是由每个子类实现。
虽然带来了一点点冗余,但是能接受。