大语言模型（LLM）和嵌入模型的统一调用接口

ChatModelFactory、EmbeddingModelFactory

1. 讲解代码：import os
2. from dotenv import load_dotenv, find_dotenv
4. _ = load_dotenv(find_dotenv())
6. from langchain_openai import ChatOpenAI, OpenAIEmbeddings, AzureChatOpenAI, AzureOpenAIEmbeddings
9. class ChatModelFactory:
10.     model_params = {
11.         "temperature": 0,
12.         "seed": 42,
13.     }
15.     @classmethod
16.     def get_model(cls, model_name: str, use_azure: bool = False):
17.         if "gpt" in model_name:
18.             if not use_azure:
19.                 return ChatOpenAI(model=model_name, **cls.model_params)
20.             else:
21.                 return AzureChatOpenAI(
22.                     azure_deployment=model_name,
23.                     api_version="2024-05-01-preview",
24.                     **cls.model_params
25.                 )
26.         elif model_name == "deepseek":
27.             # 换成开源模型试试
28.             # https://siliconflow.cn/
29.             # 一个 Model-as-a-Service 平台
30.             # 可以通过与 OpenAI API 兼容的方式调用各种开源语言模型。
31.             return ChatOpenAI(
32.                 model="deepseek-ai/DeepSeek-V2-Chat",  # 模型名称
33.                 openai_api_key=os.getenv("SILICONFLOW_API_KEY"),  # 在平台注册账号后获取
34.                 openai_api_base="https://api.siliconflow.cn/v1",  # 平台 API 地址
35.                 **cls.model_params,
36.             )
38.     @classmethod
39.     def get_default_model(cls):
40.         return cls.get_model("gpt-3.5-turbo")
43. class EmbeddingModelFactory:
45.     @classmethod
46.     def get_model(cls, model_name: str, use_azure: bool = False):
47.         if model_name.startswith("text-embedding"):
48.             if not use_azure:
49.                 return OpenAIEmbeddings(model=model_name)
50.             else:
51.                 return AzureOpenAIEmbeddings(
52.                     azure_deployment=model_name,
53.                     openai_api_version="2024-05-01-preview",
54.                 )
55.         else:
56.             raise NotImplementedError(f"Model {model_name} not implemented.")
58.     @classmethod
59.     def get_default_model(cls):
60.         return cls.get_model("text-embedding-ada-002")

复制代码

以下是这段代码的逐层解析，该代码实现了大语言模型（LLM）和嵌入模型的统一调用接口，支持多种云服务和开源模型：

---

一、环境初始化

1. import os
2. from dotenv import load_dotenv, find_dotenv
4. # 加载.env文件中的环境变量
5. _ = load_dotenv(find_dotenv())

复制代码

• 关键技术：
  
  
  
  • dotenv库主动探求项目根目次下的.env文件（包括上级目次）
    
  • 将环境变量（如API Key）注入体系环境变量
    
  
  
• 典范.env内容：
  1. OPENAI_API_KEY = "sk-xxx"
  2. AZURE_OPENAI_KEY = "azure-xxx"
  3. SILICONFLOW_API_KEY = "sf-xxx"

  复制代码

---

二、聊天模型工厂（ChatModelFactory）

1. 基础配置

1. class ChatModelFactory:
2.     model_params = {
3.         "temperature": 0,  # 控制生成随机性（0=确定性最高）
4.         "seed": 42,        # 固定随机种子保证可复现性
5.     }

复制代码

• 核心参数：
  
  
  
  • temperature: 影响天生多样性（0为最保守，1为最随机）
    
  • seed: 确保相同输入得到相同输出（对测试和调试至关紧张）
    
  
  

2. 模型选择逻辑

1. @classmethod
2. def get_model(cls, model_name: str, use_azure: bool = False):
3.     # OpenAI官方服务
4.     if "gpt" in model_name:
5.         if not use_azure:
6.             return ChatOpenAI(model=model_name, **cls.model_params)
7.         else:
8.             return AzureChatOpenAI(
9.                 azure_deployment=model_name,  # Azure部署名称
10.                 api_version="2024-05-01-preview",  # 最新API版本
11.                 **cls.model_params
12.             )
13.    
14.     # 第三方开源模型平台
15.     elif model_name == "deepseek":
16.         return ChatOpenAI(
17.             model="deepseek-ai/DeepSeek-V2-Chat",
18.             openai_api_key=os.getenv("SILICONFLOW_API_KEY"),
19.             openai_api_base="https://api.siliconflow.cn/v1",
20.             **cls.model_params,
21.         )

复制代码

• 多服务支持：
     服务类型适用场景关键参数OpenAI官方直接利用OpenAI的APImodel="gpt-3.5-turbo"Azure OpenAI企业级Azure云服务azure_deploymentSiliconFlow平台调用DeepSeek等国产开源模型自定义API Base URL
  
• 设计亮点：
  
  
  
  • 统一接口：不同服务利用相同调用方式（均继承自BaseChatModel）
    
  • 无缝切换：通过use_azure布尔参数切换云服务商
    
  • 开箱即用：预置最新API版本（2024-05-01-preview）
    
  
  

3. 默认模型配置

1. @classmethod
2. def get_default_model(cls):
3.     return cls.get_model("gpt-3.5-turbo")

复制代码

• 最佳实践：将gpt-3.5-turbo设为默认模型（性价比与性能均衡）
  

---

三、嵌入模型工厂（EmbeddingModelFactory）

1. 模型分发逻辑

1. class EmbeddingModelFactory:
2.     @classmethod
3.     def get_model(cls, model_name: str, use_azure: bool = False):
4.         if model_name.startswith("text-embedding"):
5.             if not use_azure:
6.                 return OpenAIEmbeddings(model=model_name)
7.             else:
8.                 return AzureOpenAIEmbeddings(
9.                     azure_deployment=model_name,
10.                     openai_api_version="2024-05-01-preview",
11.                 )
12.         else:
13.             raise NotImplementedError(f"Model {model_name} not implemented.")

复制代码

• 模型识别：通过text-embedding前缀判定为嵌入模型
  
• 服务兼容性：
  
  
  
  • OpenAI：直接指定模型名称（如text-embedding-ada-002）
    
  • Azure：利用部署名称（需与Azure门户中的部署名同等）
    
  
  

2. 默认配置

1. @classmethod
2. def get_default_model(cls):
3.     return cls.get_model("text-embedding-ada-002")

复制代码

• 行业尺度：text-embedding-ada-002是OpenAI最常用的嵌入模型
  

---

四、关键技术点解析

1. 环境变量管理

• 安全实践：通过.env文件隔离敏感信息
  
• 多平台支持：
  1. # SiliconFlow示例
  2. openai_api_key=os.getenv("SILICONFLOW_API_KEY")
  3. openai_api_base="https://api.siliconflow.cn/v1"

  复制代码
  纵然利用非OpenAI服务，也复用openai_前缀的参数名，保持接口统一
  

2. 工厂模式优势

优势详细表现解耦合业务代码无需关心模型初始化细节扩展性新增模型只需修改工厂类配置集中化所有模型参数在单个类中管理 3. 多云服务兼容

1. # Azure的特殊参数
2. AzureChatOpenAI(
3.     azure_deployment="gpt-35-turbo",  # 部署名称（可能与模型名不同）
4.     api_version="2024-05-01-preview"   # 必须指定
5. )

复制代码

• 留意差异：Azure的部署名称可能与OpenAI模型名称不同（如gpt-35-turbo对应OpenAI的gpt-3.5-turbo）
  

---

五、利用示例

1. 基础调用

1. # 获取默认聊天模型
2. chat_model = ChatModelFactory.get_default_model()
4. # 获取Azure版嵌入模型
5. embed_model = EmbeddingModelFactory.get_model(
6.     "text-embedding-ada-002",
7.     use_azure=True
8. )

复制代码

2. 完整工作流

1. from langchain.schema import HumanMessage
3. # 初始化模型
4. model = ChatModelFactory.get_model("gpt-4", use_azure=False)
6. # 执行对话
7. messages = [HumanMessage(content="你好！")]
8. response = model.invoke(messages)
9. print(response.content)

复制代码

---

六、扩展建议

1. 添加新模型支持

1. # 在ChatModelFactory中添加
2. elif model_name == "moonshot":
3.     return ChatOpenAI(
4.         model="moonshot-v1",
5.         openai_api_key=os.getenv("MOONSHOT_API_KEY"),
6.         openai_api_base="https://api.moonshot.cn/v1",
7.         **cls.model_params
8.     )

复制代码

2. 动态参数配置

1. # 允许运行时修改参数
2. @classmethod
3. def get_model(cls, model_name: str, **kwargs):
4.     params = {**cls.model_params, **kwargs}
5.     return ChatOpenAI(model=model_name, **params)

复制代码

3. 错误处理增强

1. try:
2.     return AzureChatOpenAI(...)
3. except AuthenticationError as e:
4.     print("Azure认证失败，请检查API Key和部署名称")

复制代码

---

七、架构图示

     该代码为构建多模型AI应用提供了尺度化基础办法，开辟者可以通过简单的参数切换实现：

• 不同云服务商之间的迁徙
  
• 开源模型与商业模型的混淆利用
  
• 嵌入模型与对话模型的协同工作
  

@Classmethod

@classmethod 是 Python 中一个紧张的装饰器，用于定义类方法。它的核心作用是为类提供一种不依靠实例却能操作类本身或类层级数据的途径。以下是逐层解析：

---

一、核心特性

1. 方法签名

1. class MyClass:
2.     @classmethod
3.     def my_method(cls, arg1, arg2):  # 第一个参数必须是类本身（惯例命名为`cls`）
4.         pass

复制代码

• cls参数：主动接收当前类（而非实例）作为第一个参数
  
• 调用方式：可通过类名.方法名()或实例.方法名()调用
  

2. 与普通实例方法的对比

方法类型第一个参数可访问内容典范用途实例方法self实例属性/方法操作实例级数据类方法cls类属性/其他类方法工厂模式、类配置操作

---

二、核心应用场景

1. *工厂模式（核心代码中的用法）

1. class ChatModelFactory:
2.     @classmethod
3.     def get_model(cls, model_name: str):  # 不创建实例即可获取模型
4.         if model_name == "gpt":
5.             return cls._create_openai_model()
6.         else:
7.             return cls._create_custom_model()
9. # 直接通过类调用
10. model = ChatModelFactory.get_model("gpt")

复制代码

• 优势：无需实例化工厂类即可创建目的对象
  
• 扩展性：子类可重写方法改变创建逻辑
  

2. 替代构造函数

1. class Date:
2.     def __init__(self, year, month, day):
3.         self.year = year
5.     @classmethod
6.     def from_string(cls, date_str):  # 接受不同格式的输入
7.         year, month, day = map(int, date_str.split("-"))
8.         return cls(year, month, day)  # 调用主构造函数
10. # 使用示例
11. date = Date.from_string("2023-08-01")

复制代码

3. 类状态管理

1. class Counter:
2.     _count = 0
4.     @classmethod
5.     def increment(cls):
6.         cls._count += 1  # 修改类属性
8.     @classmethod
9.     def get_count(cls):
10.         return cls._count  # 读取类属性
12. Counter.increment()
13. print(Counter.get_count())  # 输出: 1

复制代码

---

三、继承场景下的行为

[code]class Animal:
    @classmethod
    def identify(cls):
        print(f"I am {cls.__name__}")

class Dog(Animal):
    pass

Animal.identify()  # 输出: I am Animal
Dog.identify()     # 输出: I am Dog