在快速发展的人工智能领域，语言模型（LLMs）和聊天模型（Chat Models）的应用变得越来越普遍。然而，不同模型的接口和使用方式各不相同，给开发者带来了诸多挑战。为了解决这一问题，LangChain 提供了一套标准化的模型接口，使得开发者可以方便地切换模型而无需重构代码。深入学习 LangChain 的核心模块——Model I/O包括模型输入（Prompts）、模型输出（Output Parsers）和模型本身（Models）三个方面，本文将结合具体的源码和业务场景，探讨模型本身（Models），后面将分享模型输入（Prompts）、模型输出（Output Parsers），敬请期待。
一、模型抽象 Model
1、语言模型（LLMs）
LangChain 的核心组件之一是语言模型（LLMs）。值得注意的是，LangChain 并不提供自己的 LLMs，而是为与许多不同的 LLMs（如 OpenAI、Cohere、Hugging Face 等）进行交互提供了一个标准接口。这一设计极大地提高了开发者在不同模型之间切换的灵活性。2、聊天模型（Chat Models）
聊天模型是语言模型的一种变体。尽管它们在内部使用了语言模型，但它们提供的接口略有不同。与“输入文本，输出文本”的传统 API 不同，聊天模型提供了一个以“聊天消息”作为输入和输出的接口。这使得聊天模型在处理对话场景时更加高效和直观。二、语言模型（LLMs）
1、类继承关系
在 LangChain 中，语言模型的类继承关系如下：- BaseLanguageModel --> BaseLLM --> LLM --> <name>
- 示例：HuggingFaceHub, OpenAI
2、主要抽象
- CallbackManagerForLLMRun, AsyncCallbackManagerForLLMRun
- CallbackManager, AsyncCallbackManager
3、源码解析
该基类为语言模型定义了一个接口，允许用户以不同的方式与模型交互（例如通过提示或消息）。其中一个主要方法是 generate_prompt，它接受一系列提示并返回模型的生成结果。# 定义 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 是一个抽象基类，主要目的是提供一个基本的接口来处理大型语言模型（LLM）。这个基类使用了 Pydantic 的功能，特别是 Field 方法，用于定义默认值和序列化行为。BaseLLM 的子类需要提供实现具体功能的方法。# 定义 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
LLM 类继承自 BaseLLM，其目的是为用户提供一个简化的接口来处理 LLM，而不期望用户实现完整的 _generate 方法。# 继承自 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)
4、生成模式（Completion）
LangChain 已支持多种 LLM 模型，详细列表可以参考https://python.langchain.com/docs/integrations/llms/。使用 LangChain 调用 OpenAI GPT Completion API
开发者可以通过 LangChain 调用 OpenAI GPT Completion API，主要源码BaseOpenAI Class解析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时出错，您可以在此处指定要使用的模型名称。"""
5、案例
以下是一个简单的案例，展示了如何使用 LangChain 调用 OpenAI GPT Completion API：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个给程序员听得笑话"))
三、聊天模型（Chat Models）
1、类继承关系
在 LangChain 中，聊天模型的类继承关系如下：- BaseLanguageModel --> BaseChatModel --> <name>
- 示例：ChatOpenAI, ChatGooglePalm
2、主要抽象
3、源码解析
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:
该基类定义了聊天模型的基本接口，允许用户通过聊天消息与模型进行交互。ChatOpenAI 类用于调用 OpenAI 的 Chat Completion API。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时出错，您可以在这里指定要使用的模型名称。"""
4、案例
以下是一个简单的案例，展示了如何使用 LangChain 调用 OpenAI 的 Chat Completion API：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]))
四、业务场景中的应用
1. 客服自动化系统
现代企业越来越多地使用自动化客服系统来提高客户服务效率。这些系统需要理解客户问题，并生成准确、自然的回复。传统的客服系统往往需要预定义大量规则和模板，而通过引入 LLMs，可以大大简化这一过程，并提升回复的智能化水平。通过 LangChain，开发者可以轻松集成 OpenAI 的 GPT 模型来构建智能客服系统。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)
这种方式使得客服系统可以更快速地响应客户问题，并根据上下文提供更为精准的回复。同时，LangChain 的统一接口使得系统在需要切换到其他模型（如 Hugging Face 的模型）时，无需进行大量代码重构。2. 电商推荐系统
在电商平台中，个性化推荐系统是提高用户体验和销售额的重要手段。推荐系统需要分析用户行为数据，并生成个性化的商品推荐。通过 LangChain，开发者可以使用 Cohere 的语言模型来实现这一需求。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)
这种方式使得推荐系统能够根据用户的历史行为，生成更为个性化和动态的推荐结果，提升用户的购物体验和平台的销售业绩。3. 内容生成与校对
内容创作和校对是许多公司面临的常见任务，特别是在媒体、出版和营销领域。通过使用 LLMs，可以大幅提高内容生成和校对的效率和质量。通过 LangChain，开发者可以使用 Hugging Face 的模型来实现自动内容生成和校对。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)
这种方式使得内容创作和校对变得更加高效，减少了人工的重复劳动，并提升了内容的质量和一致性。4. 在线教育问答系统
在线教育平台需要一个智能问答系统来解答学生的疑问，并提供即时的学习帮助。传统的问答系统需要大量预先定义的问题和答案，维护成本高，且难以覆盖所有学生的需求。通过 LangChain，开发者可以使用 OpenAI 的聊天模型来构建智能问答系统。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)
这种方式使得问答系统能够实时生成高质量的答案，覆盖更广泛的知识领域，提高学生的学习效率和平台的用户满意度。通过以上案例，我们可以看到 LangChain 如何通过统一接口封装不同的模型，极大地简化了模型的调用和切换。希望本文能够帮助大家更好地理解和应用 LangChain，在人工智能开发的道路上更加得心应手。
立即咨询 JOTO
JOTO 提供覆盖企业智能体规划与搭建、AI 平台私有化部署、RAG 知识工程、AI 安全治理、FDE 驻场共创及持续运营优化的全周期 AI 落地服务，帮助企业把验证中的 AI 能力转化为安全、可控、可持续迭代的生产力。 联系 JOTO 获取 AI 落地咨询
