用FunctionCall实现文件解析(三)：ChatOpenAI实例化

发表于 分类于 阅读次数： 本文字数： 979 阅读时长 ≈ 4 分钟

前言

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

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

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

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

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

单例逻辑

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

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
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()

    # ===== 单例入口 =====
    @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

看上去没啥问题。

P.S.：

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

大模型逻辑

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

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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的基础上，我们就可以进一步确定一些具体厂商的大模型啦。比如说，我们创建一个千问大模型的类：

1
2
class TongyiFactory(BaseLLMFactory):
    ...

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

我们尝试着使用一下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import streamlit as st
from langchain_openai.chat_models.base import ChatOpenAI

from factory.client import Client

llm: 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。

那我们再换个写法：

1
2
3
4
5
6
7
8
9
10
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模型，于是我就用这样的模型再配一个工厂：

1
2
class DeepSeekFactory(BaseLLMFactory):
    ...

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

试一下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import streamlit as st
from langchain_openai.chat_models.base import ChatOpenAI

from factory.client import ClientFactory

llm: 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过程也就更简单了。