益阳建设网站建设银行的官方网站

前言

在上一篇文章中，我们建立了这样一个基本的结构，所以我们下一步就是开始做最基础的部分。

创建客户端

是不是以为我要开始创建ChatOpenAI了？还没有呢。正如我们上一篇文章所说的，一切的一切都起源于httpx.client。能够直接复用精心设计好的http线程池那当然是最好的。

于是呢，我们就想着，先创建一个客户端：

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__。这俩结合锁一起用，其实就能够线程安全地创建对象了。

我们来试试：

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._instancedef __init__(self, *args, **kwargs):if getattr(self, '_initialized', False): returnself._initialized = True

看上去没问题。

优化可读性

只不过呢，这样子可读性还差点。为了装逼，呃，我是说，为了可读性，我们也是直接上pydantic，代码可以更清晰的同时，也能够用pydantic机制做很多检查。

from threading import Lock
from pydantic import BaseModel, Field
class Singleton(BaseModel):_instance: ClassVar[Any] = None_instance_lock: ClassVar[Lock] = Lock()@classmethoddef 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：

from typing import ClassVar, Optional
from threading import Lockfrom httpx import Client, AsyncClient, Timeout
from pydantic import BaseModel, Fieldclass 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()# ===== 单例入口 =====@classmethoddef 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# ===== 私有方法：懒加载 httpx 客户端 =====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_clientdef async_client(self) -> AsyncClient:self._ensure_clients()return self._async_client

看上去顺眼多了不是吗？

来试试看：

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。

当然，最终也是没有抛出。

或者说，你也可以改成：

assert f1.client() is not f1.client(), "一样的！"

这样的话他就会抛出一个AssertError，说明这俩是一个玩意儿了。

这样，咱就有了全局唯一的一个Client或者AsyncClient了。

而且，这样创建也相当的方便易懂。

抽取抽象类

或者说，我们有一个更大胆的想法。

既然Client需要一个base_url、一个api-key和一个timeout，而ChatOpenAI也同样需要，那，抽出来？

当然没问题：

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就可以直接继承这个类了：

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对象，所以最好还是由每个子类实现。

虽然带来了一点点冗余，但是能接受。