本文聚焦 LangChain 对语言模型接口的核心抽象 ——
BaseLanguageModel，结合源码深入剖析其在多模型适配中的角色，并探讨工厂模式等设计思想如何提升模型集成能力。
一、什么是 BaseLanguageModel？
在 LangChain 中，BaseLanguageModel 是所有语言模型类的统一基类，位于模块 langchain_core.language_models.base 中。
它作为 LangChain 的 Runnable 抽象体系的一部分，定义了语言模型统一的配置接口、token 使用信息获取、流式生成支持等结构，但 并不直接实现文本生成或对话功能。
class BaseLanguageModel( RunnableSerializable[LanguageModelInput, LanguageModelOutputVar], ABC): """Abstract base class for interfacing with language models. All language model wrappers inherited from BaseLanguageModel. """该基类是为以下两类模型服务的抽象基类：
BaseLLM：传统的纯文本生成语言模型（如 text-davinci），位于 langchain_core.language_models.llms.py 模块中。class BaseLLM(BaseLanguageModel[str], ABC): """Base LLM abstract interface. It should take in a prompt and return a string. """ callback_manager: Optional[BaseCallbackManager] = Field(default=None, exclude=True) """[DEPRECATED]""" model_config = ConfigDict( arbitrary_types_allowed=True, )BaseChatModel：基于对话格式的语言模型（如 GPT-4、Claude、Mistral），位于 langchain_core.language_models.chat_models.py 模块中。class BaseChatModel(BaseLanguageModel[BaseMessage], ABC): """Base class for chat models. Key imperative methods: Methods that actually call the underlying model. +---------------------------+----------------------------------------------------------------+---------------------------------------------------------------------+--------------------------------------------------------------------------------------------------+ | Method | Input | Output | Description | +===========================+================================================================+=====================================================================+==================================================================================================+ | `invoke` | str | list[dict | tuple | BaseMessage] | PromptValue | BaseMessage | A single chat model call. | +---------------------------+----------------------------------------------------------------+---------------------------------------------------------------------+--------------------------------------------------------------------------------------------------+ | `ainvoke` | ''' | BaseMessage | Defaults to running invoke in an async executor. | +---------------------------+----------------------------------------------------------------+---------------------------------------------------------------------+--------------------------------------------------------------------------------------------------+ | `stream` | ''' | Iterator[BaseMessageChunk] | Defaults to yielding output of invoke. | +---------------------------+----------------------------------------------------------------+---------------------------------------------------------------------+--------------------------------------------------------------------------------------------------+ | `astream` | ''' | AsyncIterator[BaseMessageChunk] | Defaults to yielding output of ainvoke. | +---------------------------+----------------------------------------------------------------+---------------------------------------------------------------------+--------------------------------------------------------------------------------------------------+ | `astream_events` | ''' | AsyncIterator[StreamEvent] | Event types: 'on_chat_model_start', 'on_chat_model_stream', 'on_chat_model_end'. | +---------------------------+----------------------------------------------------------------+---------------------------------------------------------------------+--------------------------------------------------------------------------------------------------+ | `batch` | list['''] | list[BaseMessage] | Defaults to running invoke in concurrent threads. | +---------------------------+----------------------------------------------------------------+---------------------------------------------------------------------+--------------------------------------------------------------------------------------------------+ | `abatch` | list['''] | list[BaseMessage] | Defaults to running ainvoke in concurrent threads. | +---------------------------+----------------------------------------------------------------+---------------------------------------------------------------------+--------------------------------------------------------------------------------------------------+ | `batch_as_completed` | list['''] | Iterator[tuple[int, Union[BaseMessage, Exception]]] | Defaults to running invoke in concurrent threads. | +---------------------------+----------------------------------------------------------------+---------------------------------------------------------------------+--------------------------------------------------------------------------------------------------+ | `abatch_as_completed` | list['''] | AsyncIterator[tuple[int, Union[BaseMessage, Exception]]] | Defaults to running ainvoke in concurrent threads. | +---------------------------+----------------------------------------------------------------+---------------------------------------------------------------------+--------------------------------------------------------------------------------------------------+ ... """二、invoke 方法
虽然 BaseLanguageModel 没有显式实现 invoke()，但它继承自 RunnableSerializable → Runnable，后者为 LangChain 中所有模块提供了统一的执行入口：
# langchain_core.runnables.baseclass Runnable(Protocol): def invoke(self, input: InputType, config: RunnableConfig = None) -> OutputType: ...因此，任何继承自 BaseLanguageModel 的模型类 自动拥有了 invoke() 方法，可用于链式组合执行。BaseLLM 与 BaseChatModel 都在各自子类中重写了 invoke() 方法，它们的输入接口保持一致（通常为 LanguageModelInput），但输出类型不同：BaseLLM 输出纯文本结果，而 BaseChatModel 输出结构化的消息对象（BaseMessage）。
BaseLLM.invoke()@overridedef invoke( self, input: LanguageModelInput, config: Optional[RunnableConfig] = None, *, stop: Optional[list[str]] = None, **kwargs: Any,) -> str: config = ensure_config(config) return ( self.generate_prompt( [self._convert_input(input)], stop=stop, callbacks=config.get("callbacks"), tags=config.get("tags"), metadata=config.get("metadata"), run_name=config.get("run_name"), run_id=config.pop("run_id", None), **kwargs, ) .generations[0][0] .text )BaseChatModel.invoke()@overridedef invoke( self, input: LanguageModelInput, config: Optional[RunnableConfig] = None, *, stop: Optional[list[str]] = None, **kwargs: Any,) -> BaseMessage: config = ensure_config(config) return cast( "ChatGeneration", self.generate_prompt( [self._convert_input(input)], stop=stop, callbacks=config.get("callbacks"), tags=config.get("tags"), metadata=config.get("metadata"), run_name=config.get("run_name"), run_id=config.pop("run_id", None), **kwargs, ).generations[0][0], ).message👀 cast() 是 Python 类型系统中的一个函数，来自标准库模块 typing。它的作用主要是给类型系统看的注释，不会改变实际数据类型，所以请确保你的 cast 是合理的，不然只是自欺欺人。
三、工厂模式与多模型适配结构
LangChain 在早期版本中将主流模型厂商（如 OpenAI、Anthropic、Alibaba 等）的封装类直接集中维护在主包中，并统一继承自 BaseChatModel。然而在最新版本中，LangChain 已将这类具体实现迁移到子项目 langchain-community，形成“核心逻辑”和“社区接入”的分离架构。
在 langchain-community 包中，你可以找到如下模型的适配器类：
langchain_community.chat_models.openai.ChatOpenAIclass ChatOpenAI(BaseChatModel): """`OpenAI` Chat large language models API. To use, you should have the ``openai`` python package installed, and the environment variable ``OPENAI_API_KEY`` set with your API key. Any parameters that are valid to be passed to the openai.create call can be passed in, even if not explicitly saved on this class. Example: .. code-block:: python from langchain_community.chat_models import ChatOpenAI openai = ChatOpenAI(model="gpt-3.5-turbo") """langchain_community.chat_models.anthropic.ChatAnthropicclass ChatAnthropic(BaseChatModel, _AnthropicCommon): """`Anthropic` chat large language models. To use, you should have the ``anthropic`` python package installed, and the environment variable ``ANTHROPIC_API_KEY`` set with your API key, or pass it as a named parameter to the constructor. Example: .. code-block:: python import anthropic from langchain_community.chat_models import ChatAnthropic model = ChatAnthropic(model="<model_name>", anthropic_api_key="my-api-key") """langchain_community.chat_models.tongyi.ChatTongyiclass ChatTongyi(BaseChatModel): """Alibaba Tongyi Qwen chat model integration. ... """这些类通过重写 invoke/_generate 方法，实现：
- 模型输入转换（参数、消息格式）模型 API 调用（REST / SDK）模型输出统一封装（
BaseMessage）错误重试 / 输出解析 / 日志跟踪ChatOpenAI._generate()
def _generate( self, messages: List[BaseMessage], stop: Optional[List[str]] = None, run_manager: Optional[CallbackManagerForLLMRun] = None, stream: Optional[bool] = None, **kwargs: Any,) -> ChatResult: should_stream = stream if stream is not None else self.streaming if should_stream: stream_iter = self._stream( messages, stop=stop, run_manager=run_manager, **kwargs ) return generate_from_stream(stream_iter) message_dicts, params = self._create_message_dicts(messages, stop) params = { **params, **({"stream": stream} if stream is not None else {}), **kwargs, } response = self.completion_with_retry( messages=message_dicts, run_manager=run_manager, **params ) return self._create_chat_result(response)ChatTongyi._generate()
def _generate( self, messages: List[BaseMessage], stop: Optional[List[str]] = None, run_manager: Optional[CallbackManagerForLLMRun] = None, **kwargs: Any,) -> ChatResult: generations = [] if self.streaming: generation_chunk: Optional[ChatGenerationChunk] = None for chunk in self._stream( messages, stop=stop, run_manager=run_manager, **kwargs ): if generation_chunk is None: generation_chunk = chunk else: generation_chunk += chunk assert generation_chunk is not None generations.append(self._chunk_to_generation(generation_chunk)) else: params: Dict[str, Any] = self._invocation_params( messages=messages, stop=stop, **kwargs ) resp = self.completion_with_retry(**params) generations.append( ChatGeneration(**self._chat_generation_from_qwen_resp(resp)) ) return ChatResult( generations=generations, llm_output={ "model_name": self.model_name, }, )示例：OpenAI 聊天模型
import osfrom langchain_community.chat_models import ChatOpenAIfrom langchain_core.messages import HumanMessagemodel = ChatOpenAI( model='deepseek-chat', openai_api_key=os.getenv('DEEPSEEK_API_KEY'), openai_api_base='https://api.deepseek.com', max_tokens=1024)response = model.invoke( [HumanMessage(content="一句话回答这个问题：AI会成为第一生产力吗？")])print(response.content)我没有 OpenAI 的 API Key，DeepSeek API 使用与 OpenAI 兼容的 API 格式，所以替代一下。
输出：
四、BaseLLM 与 BaseChatModel 的区别
在 LangChain 的语言模型抽象体系中，BaseLLM 与 BaseChatModel 是两个并列的核心基类，它们都继承自 BaseLanguageModel，并作为 Runnable 可组合模块用于构建链、代理和应用框架。
它们的区别主要体现在输入输出格式、生成逻辑、使用场景等方面：
| 特性 | BaseLLM | BaseChatModel |
|---|---|---|
| 输入类型 | str（纯文本） | List[BaseMessage]（结构化消息列表） |
| 输出类型 | str 或 LLMResult | BaseMessage 或 ChatResult |
| 使用场景 | 文本生成、摘要、续写、补全 | 多轮对话、工具调用、代理决策、意图识别等 |
是否重写 invoke() | ✅ 重写 | ✅ 重写 |
| 内部核心方法 | _generate(prompt: PromptValue) | _generate(messages: list[BaseMessage]) |
| 是否支持流式输出 | ✅ 支持（token 级流式输出） | ✅ 支持（message 内容流式分块） |
📌 典型子类包括：
BaseLLM → OpenAI(text-davinci)、HuggingFaceHubBaseChatModel → ChatOpenAI、ChatAnthropic、ChatBaichuan🎯 使用接口一致性
尽管输入输出有所不同，二者都遵循 Runnable 接口标准，统一支持：
.invoke()：执行一次推理.stream()：流式输出（用于实时响应）.batch()：批量请求.with_config() / .with_retry() 等链式组合方式五、自定义一个 Chat 模型类（继承 BaseChatModel）
自定义一个对话模型 EchoChatModel，用于模拟回显响应。
from langchain_core.language_models.chat_models import BaseChatModelfrom langchain_core.messages import BaseMessage, AIMessagefrom langchain_core.outputs import ChatResult, ChatGenerationclass EchoChatModel(BaseChatModel): @property def _llm_type(self) -> str: return "echo-chat" def _generate( self, messages: list[BaseMessage], stop: list[str] = None, **kwargs ) -> ChatResult: content = f"Echo: {messages[-1].content}" return ChatResult( generations=[ ChatGeneration( message=AIMessage(content=content) ) ] )这在测试代理行为或调试链路时非常有用。
from langchain_core.messages import HumanMessagemodel = EchoChatModel()response = model.invoke([HumanMessage(content="我向你提出了一个问题！")])print(response.content)输出：
六、设计模式与解耦价值
LangChain 采用“抽象基类 + 工厂方法 +统一协议”组合的方式封装 LLM：
- 所有语言模型遵循统一的
Runnable 调用接口（invoke / stream / batch）不同模型类型拆分为 BaseLLM 与 BaseChatModel 两个方向各厂商通过继承封装模型差异，实现极高的复用性这让我们在构建 Chain、Agent、工具执行器等复杂流程时，可以自由切换底层模型而不影响业务逻辑。
🔚 小结
BaseLanguageModel 是 LangChain 中抽象所有语言模型的基础invoke() 来自 Runnable 接口体系，支持链式组合BaseLLM / BaseChatModel 提供了具体模型的适配规范工厂模式实现解耦、多模型统一适配、可测性与可扩展性强接下来我们将深入解析 LangChain 的 PromptTemplate 模板系统，掌握变量绑定、FewShot 示例注入与格式化机制，构建高可控的提示生成链路。
