用FunctionCall实现文件解析(三)：ChatOpenAI单例工厂

前言

在前面的文章中，我们尝试了结构和客户端的构建，接下来我们就开始新的尝试：创建ChatOpenAI实例。

在BaseFactory基础上再抽一部分逻辑

在上一篇文章中，我们完成了基类BaseFactory，并实现了ClientFactory的单例。接下来，我们进一步在BaseFactory的基础上实现ChatOpenAI的实例化。

既然ClientFactory是全局唯一的，那我们也将ChatOpenAI的工厂也定义为全局唯一的。虽然这样会使得项目始终保存每一个工厂的实例，但是起码来说，比起反复创建又销毁，这样还是稍微简单一点。

既然所有的内容都是相同的，我们不妨再将一些逻辑抽离出来。比如，工厂构造的单例逻辑和构建大模型的逻辑。

单例逻辑

我们首先将单例逻辑抽离出来。

他们都使用了一个_instance和一个_instance_lock方法，所以把这部分抽离出来：

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")# ---------- 单例相关 ----------_instance: ClassVar[Optional["ClientFactory"]] = None_instance_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

看上去没啥问题。

P.S.：

根据这篇文章的描述，类方法get_instance虽然定义在了父类，但是子类继承之后，所传入的cls实际上就成了子类。所以，如果父类有这个方法，子类方法同样会按照父类的逻辑实现单例。非常的方便。

大模型逻辑

大模型相对来说更简单一些。既然已经传入了base_url、api_key和timeout，那么基本上也就能够确定一系列的ChatOpenAI对象了。剩下的参数我们就放到build方法中，让build去创建对应的对象就好了。

但是呢，build方法如果放在BaseFactory中，那么BaseFactory的功能也就太多了，这看起来不太好。我们直接继承一个新的：

from httpx import Client
from langchain_openai.chat_models.base import ChatOpenAI
class BaseLLMFactory(BaseFactory):def build(self,model: str,temperature: float = 0.7,max_tokens: int = 256,client: Client = None) -> ChatOpenAI:return ChatOpenAI(model=model,temperature=temperature,max_tokens=max_tokens,client=client)

看着不错。当然，你也可以将client设置为必填或者在方法中检测并报错，这都是比较细节的小问题了。

以通义千问为例构建大模型工厂

在BaseLLMFactory的基础上，我们就可以进一步确定一些具体厂商的大模型啦。比如说，我们创建一个千问大模型的类：

class TongyiFactory(BaseLLMFactory):...

是的，没错，他什么逻辑都不需要，定义出来就够用了。

我们尝试着使用一下：

import streamlit as st
from langchain_openai.chat_models.base import ChatOpenAIfrom factory.client import Clientllm: ChatOpenAI = TongyiFactory.get_instance(base_url=st.secrets["DASH_URL"],api_key=st.secrets["DASH_KEY"],_client=ClientFactory.get_instance(base_url=st.secrets["DASH_URL"],api_key=st.secrets["DASH_KEY"],timeout=st.secrets["DASH_TIMEOUT"],).client(),
).build(model="qwen-max",
)

看着好像有那么一点不太像python，甚至有点像Java。

那我们再换个写法：

client = ClientFactory.get_instance(base_url=st.secrets["DASH_URL"],api_key=st.secrets["DASH_KEY"],timeout=st.secrets["DASH_TIMEOUT"],
).client()
llm: ChatOpenAI = TongyiFactory.get_instance(base_url=client.base_url,api_key=client.api_key,_client = client,
).build()

嗯……总之，挑你喜欢的方案就行。

自定义大模型工厂

既然通义千问可以，我自定义的行不行？

比如说，现在我在华为昇腾的卡上部署了一个DeepSeek-R1-Dstill-Llama-70B模型，于是我就用这样的模型再配一个工厂：

class DeepSeekFactory(BaseLLMFactory):...

同样的，定义出来就够用了。

试一下：

import streamlit as st
from langchain_openai.chat_models.base import ChatOpenAIfrom factory.client import ClientFactoryllm: ChatOpenAI = DeepSeekFactory.get_instance(base_url=st.secrets["deepseek_url"],api_key=st.secrets["deepseek_api_key"],_client = ClientFactory.get_instance(base_url=st.secrets["deepseek_url"],api_key=st.secrets["deepseek_api_key"],timeout=st.secrets["deepseek_timeout"],).client()
).build(model="DeepSeek-70B",temperature=0.7,max_tokens=4096,
)

然后就可以开心的使用了。

当然，你完全可以自己定义的时候将一些参数配死，这样的话build过程也就更简单了。