深度解析LangChain模型 I/O 之Models封装：解锁智能化业务应用的全新可能

# 定义 BaseLanguageModel 抽象基类，它从 Serializable, Runnable 和 ABC 继承class BaseLanguageModel(Serializable, Runnable[LanguageModelInput, LanguageModelOutput], ABC):"""与语言模型交互的抽象基类。
所有语言模型的封装器都应从 BaseLanguageModel 继承。
主要提供三种方法：- generate_prompt: 为一系列的提示值生成语言模型输出。提示值是可以转换为任何语言模型输入格式的模型输入（如字符串或消息）。- predict: 将单个字符串传递给语言模型并返回字符串预测。- predict_messages: 将一系列 BaseMessages（对应于单个模型调用）传递给语言模型，并返回 BaseMessage 预测。
每种方法都有对应的异步方法。"""
# 定义一个抽象方法 generate_prompt，需要子类进行实现@abstractmethoddef generate_prompt(self,prompts: List[PromptValue],# 输入提示的列表stop: Optional[List[str]] = None,# 生成时的停止词列表callbacks: Callbacks = None,# 回调，用于执行例如日志记录或流式处理的额外功能**kwargs: Any,# 任意的额外关键字参数，通常会传递给模型提供者的 API 调用) -> LLMResult:"""将一系列的提示传递给模型并返回模型的生成。
对于提供批处理 API 的模型，此方法应使用批处理调用。
使用此方法时：1. 希望利用批处理调用，2. 需要从模型中获取的输出不仅仅是最顶部生成的值，3. 构建与底层语言模型类型无关的链（例如，纯文本完成模型与聊天模型）。
参数:prompts: 提示值的列表。提示值是一个可以转换为与任何语言模型匹配的格式的对象（对于纯文本生成模型为字符串，对于聊天模型为 BaseMessages）。stop: 生成时使用的停止词。模型输出在这些子字符串的首次出现处截断。callbacks: 要传递的回调。用于执行额外功能，例如在生成过程中进行日志记录或流式处理。**kwargs: 任意的额外关键字参数。通常这些会传递给模型提供者的 API 调用。
返回值:LLMResult，它包含每个输入提示的候选生成列表以及特定于模型提供者的额外输出。"""

# 定义 BaseLLM 抽象基类，它从 BaseLanguageModel[str] 和 ABC（Abstract Base Class）继承class BaseLLM(BaseLanguageModel[str], ABC):"""Base LLM abstract interface.It should take in a prompt and return a string."""
# 定义可选的缓存属性，其初始值为 Nonecache: Optional[bool] = None
# 定义 verbose 属性，该属性决定是否打印响应文本# 默认值使用 _get_verbosity 函数的结果verbose: bool = Field(default_factory=_get_verbosity)"""Whether to print out response text."""
# 定义 callbacks 属性，其初始值为 None，并从序列化中排除callbacks: Callbacks = Field(default=None, exclude=True)
# 定义 callback_manager 属性，其初始值为 None，并从序列化中排除callback_manager: Optional[BaseCallbackManager] = Field(default=None, exclude=True)
# 定义 tags 属性，这些标签会被添加到运行追踪中，其初始值为 None，并从序列化中排除tags: Optional[List[str]] = Field(default=None, exclude=True)"""Tags to add to the run trace."""
# 定义 metadata 属性，这些元数据会被添加到运行追踪中，其初始值为 None，并从序列化中排除metadata: Optional[Dict[str, Any]] = Field(default=None, exclude=True)"""Metadata to add to the run trace."""
# 内部类定义了这个 pydantic 对象的配置class Config:"""Configuration for this pydantic object."""
# 允许使用任意类型arbitrary_types_allowed = True

# 继承自 BaseLLM 的 LLM 类class LLM(BaseLLM):"""Base LLM abstract class.
The purpose of this class is to expose a simpler interface for workingwith LLMs, rather than expect the user to implement the full _generate method."""
# 使用 @abstractmethod 装饰器定义一个抽象方法，子类需要实现这个方法@abstractmethoddef _call(self,prompt: str,# 输入提示stop: Optional[List[str]] = None,# 停止词列表run_manager: Optional[CallbackManagerForLLMRun] = None,# 运行管理器**kwargs: Any,# 其他关键字参数) -> str:"""Run the LLM on the given prompt and input."""# 此方法的实现应在子类中提供
# _generate 方法使用了 _call 方法，用于处理多个提示def _generate(self,prompts: List[str],# 多个输入提示的列表stop: Optional[List[str]] = None,run_manager: Optional[CallbackManagerForLLMRun] = None,**kwargs: Any,) -> LLMResult:"""Run the LLM on the given prompt and input."""# TODO: 在此处添加缓存逻辑generations = []# 用于存储生成的文本# 检查 _call 方法的签名是否支持 run_manager 参数new_arg_supported = inspect.signature(self._call).parameters.get("run_manager")for prompt in prompts:# 遍历每个提示# 根据是否支持 run_manager 参数来选择调用方法text = (self._call(prompt, stop=stop, run_manager=run_manager, **kwargs)if new_arg_supportedelse self._call(prompt, stop=stop, **kwargs))# 将生成的文本添加到 generations 列表中generations.append([Generation(text=text)])# 返回 LLMResult 对象，其中包含 generations 列表return LLMResult(generations=generations)

class BaseOpenAI(BaseLLM):"""OpenAI 大语言模型的基类。"""
@propertydef lc_secrets(self) -> Dict[str, str]:return {"openai_api_key": "OPENAI_API_KEY"}
@propertydef lc_serializable(self) -> bool:return True
client: Any#: :meta private:model_name: str = Field("text-davinci-003", alias="model")"""使用的模型名。"""temperature: float = 0.7"""要使用的采样温度。"""max_tokens: int = 256"""完成中生成的最大令牌数。-1表示根据提示和模型的最大上下文大小返回尽可能多的令牌。"""top_p: float = 1"""在每一步考虑的令牌的总概率质量。"""frequency_penalty: float = 0"""根据频率惩罚重复的令牌。"""presence_penalty: float = 0"""惩罚重复的令牌。"""n: int = 1"""为每个提示生成多少完成。"""best_of: int = 1"""在服务器端生成best_of完成并返回“最佳”。"""model_kwargs: Dict[str, Any] = Field(default_factory=dict)"""保存任何未明确指定的`create`调用的有效模型参数。"""openai_api_key: Optional[str] = Noneopenai_api_base: Optional[str] = Noneopenai_organization: Optional[str] = None# 支持OpenAI的显式代理openai_proxy: Optional[str] = Nonebatch_size: int = 20"""传递多个文档以生成时使用的批处理大小。"""request_timeout: Optional[Union[float, Tuple[float, float]]] = None"""向OpenAI完成API的请求超时。默认为600秒。"""logit_bias: Optional[Dict[str, float]] = Field(default_factory=dict)"""调整生成特定令牌的概率。"""max_retries: int = 6"""生成时尝试的最大次数。"""streaming: bool = False"""是否流式传输结果。"""allowed_special: Union[Literal["all"], AbstractSet[str]] = set()"""允许的特殊令牌集。"""disallowed_special: Union[Literal["all"], Collection[str]] = "all""""不允许的特殊令牌集。"""tiktoken_model_name: Optional[str] = None"""使用此类时传递给tiktoken的模型名。Tiktoken用于计算文档中的令牌数量以限制它们在某个限制以下。默认情况下，设置为None时，这将与嵌入模型名称相同。但是，在某些情况下，您可能希望使用此嵌入类与tiktoken不支持的模型名称。这可以包括使用Azure嵌入或使用多个模型提供商的情况，这些提供商公开了类似OpenAI的API但模型不同。在这些情况下，为了避免在调用tiktoken时出错，您可以在此处指定要使用的模型名称。"""

import openaiimport osimport tiktoken# 加载 .env 文件from dotenv import load_dotenv, find_dotenvfrom langchain.prompts import PromptTemplatefrom langchain.llms import OpenAIfrom langchain.chains import LLMChain#from langchain.chat_models import AzureChatOpenAIfrom langchain.chat_models import ChatOpenAI #直接访问OpenAI的GPT服务
_ = load_dotenv(find_dotenv())# 从环境变量中获得你的 OpenAI Key和配置URLopenai.api_key = os.getenv('OPENAI_API_KEY')openai.api_base = os.getenv('OPENAI_API_URL')model = os.getenv('OPENAI_API_MODEL')
llm = OpenAI(model_name=model, temperature=0) #直接访问OpenAI的GPT服务print(llm("今天天气怎么样"))print(llm("讲10个给程序员听得笑话"))

class BaseChatModel(BaseLanguageModel[BaseMessageChunk], ABC):cache: Optional[bool] = None"""是否缓存响应。"""verbose: bool = Field(default_factory=_get_verbosity)"""是否打印响应文本。"""callbacks: Callbacks = Field(default=None, exclude=True)"""添加到运行追踪的回调函数。"""callback_manager: Optional[BaseCallbackManager] = Field(default=None, exclude=True)"""添加到运行追踪的回调函数管理器。"""tags: Optional[List[str]] = Field(default=None, exclude=True)"""添加到运行追踪的标签。"""metadata: Optional[Dict[str, Any]] = Field(default=None, exclude=True)"""添加到运行追踪的元数据。"""
# 需要子类实现的 _generate 抽象方法@abstractmethoddef _generate(self,messages: List[BaseMessage],stop: Optional[List[str]] = None,run_manager: Optional[CallbackManagerForLLMRun] = None,**kwargs: Any,) -> ChatResult:

class ChatOpenAI(BaseChatModel):"""OpenAI Chat大语言模型的包装器。
要使用，您应该已经安装了``openai`` python包，并且环境变量``OPENAI_API_KEY``已使用您的API密钥进行设置。
即使未在此类上明确保存，也可以传入任何有效的参数至openai.create调用。"""
@propertydef lc_secrets(self) -> Dict[str, str]:return {"openai_api_key": "OPENAI_API_KEY"}
@propertydef lc_serializable(self) -> bool:return True
client: Any = None#: :meta private:model_name: str = Field(default="gpt-3.5-turbo", alias="model")"""要使用的模型名。"""temperature: float = 0.7"""使用的采样温度。"""model_kwargs: Dict[str, Any] = Field(default_factory=dict)"""保存任何未明确指定的`create`调用的有效模型参数。"""openai_api_key: Optional[str] = None"""API请求的基础URL路径，如果不使用代理或服务仿真器，请留空。"""openai_api_base: Optional[str] = Noneopenai_organization: Optional[str] = None# 支持OpenAI的显式代理openai_proxy: Optional[str] = Nonerequest_timeout: Optional[Union[float, Tuple[float, float]]] = None"""请求OpenAI完成API的超时。默认为600秒。"""max_retries: int = 6"""生成时尝试的最大次数。"""streaming: bool = False"""是否流式传输结果。"""n: int = 1"""为每个提示生成的聊天完成数。"""max_tokens: Optional[int] = None"""生成的最大令牌数。"""tiktoken_model_name: Optional[str] = None"""使用此类时传递给tiktoken的模型名称。Tiktoken用于计算文档中的令牌数以限制它们在某个限制之下。默认情况下，当设置为None时，这将与嵌入模型名称相同。但是，在某些情况下，您可能希望使用此嵌入类，模型名称不由tiktoken支持。这可能包括使用Azure嵌入或使用其中之一的多个模型提供商公开类似OpenAI的API但模型不同。在这些情况下，为了避免在调用tiktoken时出错，您可以在这里指定要使用的模型名称。"""

import openaiimport osimport tiktoken# 加载 .env 文件from dotenv import load_dotenv, find_dotenvfrom langchain.prompts import PromptTemplatefrom langchain.llms import OpenAIfrom langchain.chains import LLMChain#from langchain.chat_models import AzureChatOpenAIfrom langchain.chat_models import ChatOpenAI #直接访问OpenAI的GPT服务
_ = load_dotenv(find_dotenv())# 从环境变量中获得你的 OpenAI Key和配置URLopenai.api_key = os.getenv('OPENAI_API_KEY')openai.api_base = os.getenv('OPENAI_API_URL')model = os.getenv('OPENAI_API_MODEL')
llm = OpenAI(model_name=model, temperature=0) #直接访问OpenAI的GPT服务#print(llm("今天天气怎么样"))#print(llm("讲10个给程序员听得笑话"))
from langchain.schema import (AIMessage,HumanMessage,SystemMessage)
def message_to_string():messages = [SystemMessage(content="You are a helpful assistant."),HumanMessage(content="Who won the world series in 2020?"),AIMessage(content="The Los Angeles Dodgers won the World Series in 2020."), HumanMessage(content="Where was it played?")]print(messages)
#python 入口函数

if __name__ == '__main__':llm(message_to_string())result = llm("生成可执行的快速排序 Python 代码")print(result)# 使用 `exec` 定义 `quick_sort` 函数#exec(result)# 调用 GPT 生成的快排代码，测试是否可用#print(quick_sort([3,6,8,10,1,2,1,1024]))

from langchain.llms import OpenAI
llm = OpenAI(api_key="your-api-key")prompt = "Customer: 'I cannot log into my account.'\nAI: 'Please try resetting your password using the link on the login page.'"response = llm.generate_prompt([prompt])print(response)

from langchain.llms import Cohere
llm = Cohere(api_key="your-api-key")prompt = "User has recently viewed: 'smartphones, laptops'.\nAI: 'We recommend these new gadgets for you: ...'"response = llm.generate_prompt([prompt])print(response)

from langchain.llms import HuggingFaceHub
llm = HuggingFaceHub(api_key="your-api-key")prompt = "Generate a marketing slogan for a new eco-friendly water bottle."response = llm.generate_prompt([prompt])print(response)

from langchain.chat_models import ChatOpenAIfrom langchain.schema import HumanMessage
chat_model = ChatOpenAI(api_key="your-api-key")message = HumanMessage(content="What is the Pythagorean theorem?")response = chat_model.chat([message])print(response.content)