用FunctionCall实现文件解析(二):创建单例客户端

前言

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

创建客户端

是不是以为我要开始创建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了,所以我也就不区分什么AsyncClientFactorySyncClientFactory了,毕竟有奥卡姆剃刀原则在,我们直接给一个全功能的超级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

# ===== 私有方法:懒加载 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_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对象,所以最好还是由每个子类实现。

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