概述

为什么一行 init_chat_model() 能切换不同厂商

前面文章里我们多次写过：

from langchain.chat_models import init_chat_model


model = init_chat_model("openai:gpt-5.4-mini", temperature=0)
response = model.invoke("你好，介绍一下 LangChain")

如果要换成其他模型，通常只改一行：

model = init_chat_model("anthropic:claude-sonnet-4-5", temperature=0)

或者：

model = init_chat_model("ollama:llama3.1", temperature=0)

从使用者视角看，这叫“统一模型入口”。

从源码视角看，它背后至少做了五件事：

init_chat_model("openai:gpt-5.4-mini")
  |
  v
解析 provider 和 model name
  |
  v
根据 provider 找到集成包和 ChatModel 类
  |
  v
动态 import 供应商包
  |
  v
实例化具体 ChatModel，例如 ChatOpenAI
  |
  v
返回 BaseChatModel 统一接口

也就是说：

"openai:gpt-5.4-mini"
  -> provider = "openai"
  -> model = "gpt-5.4-mini"
  -> langchain_openai.ChatOpenAI(model="gpt-5.4-mini")
  -> BaseChatModel.invoke(...)

这就是 LangChain 模型抽象层的核心价值：

• 上层代码只依赖 BaseChatModel。
• 不同供应商通过独立集成包提供具体实现。
• init_chat_model() 负责做字符串解析、provider 路由、动态导入和实例化。
• invoke、stream、batch、bind_tools、with_structured_output 这些方法形成统一能力面。

init_chat_model() 是模型层的工厂函数，BaseChatModel 是统一抽象接口，供应商包是可插拔实现。

源码入口：langchain.chat_models 只暴露两个东西

先定位入口文件：

langchain/chat_models/__init__.py

内容非常短：

from langchain_core.language_models import BaseChatModel

from langchain.chat_models.base import init_chat_model

__all__ = ["BaseChatModel", "init_chat_model"]

这说明 langchain.chat_models 的公共入口主要就是：

名称	来源	作用
init_chat_model	langchain/chat_models/base.py	创建具体聊天模型
BaseChatModel	langchain_core/language_models/chat_models.py	所有聊天模型的统一基类

也就是说：

from langchain.chat_models import init_chat_model

实际进入的是：

langchain/chat_models/base.py

而你拿到的模型对象最终要满足：

isinstance(model, BaseChatModel)

源码结构可以画成：

langchain.chat_models
  |
  +-- init_chat_model()
  |     |
  |     +-- 解析模型字符串
  |     +-- 动态导入 provider 包
  |     +-- 实例化供应商 ChatModel
  |
  +-- BaseChatModel
        |
        +-- invoke / ainvoke
        +-- stream / astream
        +-- batch / abatch
        +-- bind_tools
        +-- with_structured_output

langchain 包本身不把所有厂商代码塞进来，而是提供一个统一入口和一套核心抽象。

第一层：_BUILTIN_PROVIDERS 是 provider 路由表

在 langchain/chat_models/base.py 顶部，有一个非常关键的字典：

_BUILTIN_PROVIDERS: dict[str, tuple[str, str, Callable[..., BaseChatModel]]] = {
    "anthropic": ("langchain_anthropic", "ChatAnthropic", _call),
    "azure_openai": ("langchain_openai", "AzureChatOpenAI", _call),
    "deepseek": ("langchain_deepseek", "ChatDeepSeek", _call),
    "google_genai": ("langchain_google_genai", "ChatGoogleGenerativeAI", _call),
    "google_vertexai": ("langchain_google_vertexai", "ChatVertexAI", _call),
    "ollama": ("langchain_ollama", "ChatOllama", _call),
    "openai": ("langchain_openai", "ChatOpenAI", _call),
    "openrouter": ("langchain_openrouter", "ChatOpenRouter", _call),
    "xai": ("langchain_xai", "ChatXAI", _call),
    ...
}

真实源码支持的 provider 更多，这里只截取常见部分。

这个表的每一项都是：

provider -> (module_path, class_name, creator_func)

比如：

provider	module path	class name	实例化结果
openai	langchain_openai	ChatOpenAI	ChatOpenAI(model=...)
azure_openai	langchain_openai	AzureChatOpenAI	AzureChatOpenAI(model=...)
anthropic	langchain_anthropic	ChatAnthropic	ChatAnthropic(model=...)
deepseek	langchain_deepseek	ChatDeepSeek	ChatDeepSeek(model=...)
ollama	langchain_ollama	ChatOllama	ChatOllama(model=...)

这其实就是一个轻量级 SPI 机制。

SPI 在这里不是 Java 那套 META-INF/services，而是一种架构思想：

核心包定义接口和注册约定
供应商包提供具体实现
工厂函数根据 provider 动态加载实现

LangChain 的模型层没有强迫所有供应商代码都进入 langchain 主包，而是拆成独立包：

langchain-openai
langchain-anthropic
langchain-deepseek
langchain-ollama
langchain-google-genai
langchain-google-vertexai
...

这带来三个好处：

• 依赖隔离：不用 OpenAI 的人不必安装 OpenAI SDK。
• 版本隔离：不同供应商 SDK 更新节奏不同，独立包更容易维护。
• 能力扩展：新增 provider 不需要改上层 Agent、Chain、Prompt 代码。

_BUILTIN_PROVIDERS 是 LangChain ChatModel 工厂的路由表，把 provider 字符串映射到独立集成包里的具体类。

第二层：_parse_model() 负责拆出 provider 和 model name

init_chat_model() 支持两种写法。

第一种：推荐的前缀形式。

model = init_chat_model("openai:gpt-5.4-mini")

第二种：分开传。

model = init_chat_model(
    "gpt-5.4-mini",
    model_provider="openai",
)

源码里负责解析的是 _parse_model()：

def _parse_model(model: str, model_provider: str | None) -> tuple[str, str]:
    if (
        not model_provider
        and ":" in model
        and model.split(":", maxsplit=1)[0] in _BUILTIN_PROVIDERS
    ):
        model_provider = model.split(":", maxsplit=1)[0]
        model = ":".join(model.split(":")[1:])

    model_provider = model_provider or _attempt_infer_model_provider(model)

    if not model_provider:
        raise ValueError(...)

    model_provider = model_provider.replace("-", "_").lower()
    return model, model_provider

把 "openai:gpt-5.4-mini" 代进去：

model = "openai:gpt-5.4-mini"
model_provider = None

发现冒号前缀 openai 在 _BUILTIN_PROVIDERS 中
  -> model_provider = "openai"
  -> model = "gpt-5.4-mini"

返回：
  ("gpt-5.4-mini", "openai")

如果传：

init_chat_model("gpt-5.4-mini")

没有 provider 前缀，源码会尝试自动推断：

model_provider = model_provider or _attempt_infer_model_provider(model)

如果推断失败，就抛错，提示你显式传 model_provider。

_parse_model() 先识别 provider:model 前缀，再尝试按模型名前缀推断 provider，最后把 provider 标准化为小写下划线格式。

第三层：provider 推断是 best-effort，不是强保证

_attempt_infer_model_provider() 根据模型名前缀猜 provider。

源码规则可以简化成：

def _attempt_infer_model_provider(model_name: str) -> str | None:
    model_lower = model_name.lower()

    if model_lower.startswith(("gpt-", "o1", "o3", "chatgpt", "text-davinci")):
        return "openai"

    if model_lower.startswith("claude"):
        return "anthropic"

    if model_lower.startswith("command"):
        return "cohere"

    if model_lower.startswith("gemini"):
        return "google_vertexai"

    if model_lower.startswith(("amazon.", "anthropic.", "meta.")):
        return "bedrock"

    if model_lower.startswith(("mistral", "mixtral")):
        return "mistralai"

    if model_lower.startswith("deepseek"):
        return "deepseek"

    if model_lower.startswith("grok"):
        return "xai"

    if model_lower.startswith("sonar"):
        return "perplexity"

    if model_lower.startswith("solar"):
        return "upstage"

    return None

所以这些写法大概率能推断：

init_chat_model("gpt-5.4-mini")        # openai
init_chat_model("claude-sonnet-4-5")  # anthropic
init_chat_model("deepseek-chat")      # deepseek
init_chat_model("grok-4")             # xai

但生产代码里更推荐写明确：

init_chat_model("openai:gpt-5.4-mini")
init_chat_model("anthropic:claude-sonnet-4-5")
init_chat_model("deepseek:deepseek-chat")

原因很简单：

• 不同供应商可能出现相似模型名前缀。
• 某些模型别名会随时间变化。
• gemini 这种前缀本身存在 Vertex AI 和 Google GenAI 两种接入方式。
• 自动推断属于便利功能，不适合做强依赖。

源码里对 gemini 也有提示：当前推断为 google_vertexai，但未来默认可能变化；如果要锁定行为，应该显式写：

init_chat_model("google_vertexai:gemini-...", ...)

或者：

init_chat_model("google_genai:gemini-...", ...)

provider 推断只是为了方便交互式开发，工程代码建议使用 provider:model 前缀，避免模型名和供应商路由漂移。

第四层：_get_chat_model_creator() 动态导入供应商包

provider 和 model name 解析出来以后，下一步是找到具体类。

核心函数：

@functools.lru_cache(maxsize=len(_BUILTIN_PROVIDERS))
def _get_chat_model_creator(provider: str) -> Callable[..., BaseChatModel]:
    if provider not in _BUILTIN_PROVIDERS:
        raise ValueError(...)

    pkg, class_name, creator_func = _BUILTIN_PROVIDERS[provider]
    module = _import_module(pkg, class_name)
    cls = getattr(module, class_name)
    return functools.partial(creator_func, cls=cls)

拆开看：

provider = "openai"
  |
  v
_BUILTIN_PROVIDERS["openai"]
  |
  v
("langchain_openai", "ChatOpenAI", _call)
  |
  v
importlib.import_module("langchain_openai")
  |
  v
getattr(module, "ChatOpenAI")
  |
  v
partial(_call, cls=ChatOpenAI)

_import_module() 的错误处理也很友好：

def _import_module(module: str, class_name: str):
    try:
        return importlib.import_module(module)
    except ImportError as e:
        pkg = module.split(".", maxsplit=1)[0].replace("_", "-")
        raise ImportError(
            f"Initializing {class_name} requires the {pkg} package. "
            f"Please install it with `pip install {pkg}`"
        ) from e

所以如果你没有安装 langchain-openai，却写了：

init_chat_model("openai:gpt-5.4-mini")

报错会提示：

Please install it with `pip install langchain-openai`

这也是为什么 LangChain v1 之后常见安装方式是：

pip install -U langchain langchain-openai

而不是只装：

pip install langchain

init_chat_model() 不直接 import 所有供应商包，而是在知道 provider 后按需动态 import，因此依赖更轻、错误提示也更明确。

第五层：_init_chat_model_helper() 是真正的实例化入口

init_chat_model() 本身要处理固定模型和可配置模型两种情况。

如果是固定模型，最终会进入：

def _init_chat_model_helper(
    model: str,
    *,
    model_provider: str | None = None,
    **kwargs: Any,
) -> BaseChatModel:
    model, model_provider = _parse_model(model, model_provider)
    creator_func = _get_chat_model_creator(model_provider)
    return creator_func(model=model, **kwargs)

继续用 OpenAI 举例：

init_chat_model("openai:gpt-5.4-mini", temperature=0)

执行过程是：

_parse_model("openai:gpt-5.4-mini", None)
  -> ("gpt-5.4-mini", "openai")

_get_chat_model_creator("openai")
  -> partial(_call, cls=ChatOpenAI)

creator_func(model="gpt-5.4-mini", temperature=0)
  -> ChatOpenAI(model="gpt-5.4-mini", temperature=0)

其中 _call() 更简单：

def _call(cls: type[BaseChatModel], **kwargs: Any) -> BaseChatModel:
    return cls(**kwargs)

某些 provider 会有特殊实例化方式。

比如 HuggingFace：

"huggingface": (
    "langchain_huggingface",
    "ChatHuggingFace",
    lambda cls, model, **kwargs: cls.from_model_id(model_id=model, **kwargs),
)

IBM：

"ibm": (
    "langchain_ibm",
    "ChatWatsonx",
    lambda cls, model, **kwargs: cls(model_id=model, **kwargs),
)

这就是 creator_func 存在的意义：不同供应商构造参数不完全一致，工厂层可以做最后一层适配。

_init_chat_model_helper() 是固定模型的核心路径，它把“解析结果 + provider 工厂 + 用户参数”合成一个具体 BaseChatModel 实例。

完整调用链：从字符串到 ChatOpenAI

把前面几层连起来：

用户代码：
  init_chat_model("openai:gpt-5.4-mini", temperature=0)

源码路径：
  langchain/chat_models/base.py
    init_chat_model()
      |
      +-- configurable_fields 为空？
      |     |
      |     +-- 是：走固定模型
      |
      +-- _init_chat_model_helper(
            model="openai:gpt-5.4-mini",
            temperature=0
          )
            |
            +-- _parse_model(...)
            |     |
            |     +-- model = "gpt-5.4-mini"
            |     +-- model_provider = "openai"
            |
            +-- _get_chat_model_creator("openai")
            |     |
            |     +-- _BUILTIN_PROVIDERS["openai"]
            |     +-- import langchain_openai
            |     +-- getattr(ChatOpenAI)
            |
            +-- creator_func(model="gpt-5.4-mini", temperature=0)
                  |
                  +-- ChatOpenAI(model="gpt-5.4-mini", temperature=0)

返回：
  BaseChatModel 实例

用 Mermaid 表示：

init_chat_model() 的固定模型路径，本质就是一个“解析 + 查表 + 动态 import + 工厂实例化”的流程。

可配置模型：_ConfigurableModel 让模型在运行时切换

init_chat_model() 还有一个非常实用的能力：运行时切模型。

示例：

from langchain.chat_models import init_chat_model


model = init_chat_model(
    configurable_fields=("model", "model_provider", "temperature"),
)

response = model.invoke(
    "写一句中文介绍 LangChain",
    config={
        "configurable": {
            "model": "openai:gpt-5.4-mini",
            "temperature": 0,
        }
    },
)

如果你不传 model，并且不传 configurable_fields：

model = init_chat_model(temperature=0)

源码会默认让这些字段可配置：

if not model and not configurable_fields:
    configurable_fields = ("model", "model_provider")

也就是说，下面这种写法是成立的：

configurable_model = init_chat_model(temperature=0)

configurable_model.invoke(
    "你好",
    config={"configurable": {"model": "openai:gpt-5.4-mini"}},
)

它不会在 init_chat_model() 调用时立刻创建真实模型，而是返回一个 _ConfigurableModel。

源码：

return _ConfigurableModel(
    default_config=kwargs,
    config_prefix=config_prefix,
    configurable_fields=configurable_fields,
)

真正的模型实例化延迟到调用时：

def _model(self, config: RunnableConfig | None = None) -> Runnable[Any, Any]:
    params = {**self._default_config, **self._model_params(config)}
    model = _init_chat_model_helper(**params)
    for name, args, kwargs in self._queued_declarative_operations:
        model = getattr(model, name)(*args, **kwargs)
    return model

调用 invoke() 时：

def invoke(self, input, config=None, **kwargs):
    return self._model(config).invoke(input, config=config, **kwargs)

也就是说：

_ConfigurableModel.invoke(...)
  -> 从 config["configurable"] 取 model/model_provider/temperature
  -> _init_chat_model_helper(**params)
  -> 得到真实 ChatModel
  -> 调用真实 ChatModel.invoke(...)

这就是运行时切模型的实现。

_ConfigurableModel 是一个延迟实例化代理，它在每次调用时根据 RunnableConfig 创建真实模型。

config_prefix：多个可配置模型如何避免参数冲突

如果一个链里有两个模型：

• 一个负责改写问题。
• 一个负责生成答案。

它们都可能需要配置 model 和 temperature。

这时就需要 config_prefix。

rewriter = init_chat_model(
    "openai:gpt-5.4-mini",
    configurable_fields=("model", "temperature"),
    config_prefix="rewrite",
    temperature=0,
)

answerer = init_chat_model(
    "openai:gpt-5.4-mini",
    configurable_fields=("model", "temperature"),
    config_prefix="answer",
    temperature=0.2,
)

调用时：

result = chain.invoke(
    {"question": "LangChain 的 Runnable 是什么？"},
    config={
        "configurable": {
            "rewrite_model": "openai:gpt-5.4-mini",
            "rewrite_temperature": 0,
            "answer_model": "anthropic:claude-sonnet-4-5",
            "answer_temperature": 0.3,
        }
    },
)

源码里 config_prefix 会自动补下划线：

self._config_prefix = (
    config_prefix + "_"
    if config_prefix and not config_prefix.endswith("_")
    else config_prefix
)

然后 _model_params() 只读取对应前缀的配置：

model_params = {
    _remove_prefix(k, self._config_prefix): v
    for k, v in config.get("configurable", {}).items()
    if k.startswith(self._config_prefix)
}

所以：

rewrite_model -> model
rewrite_temperature -> temperature

answer_model -> model
answer_temperature -> temperature

config_prefix 是多个可配置模型共存时的命名空间机制，防止所有模型都争抢同一组 config key。

安全提醒：不要轻易开放 configurable_fields="any"

init_chat_model() 支持：

model = init_chat_model(
    "openai:gpt-5.4-mini",
    configurable_fields="any",
)

这意味着所有模型参数都可以通过运行时 config 改写。

听起来很灵活，但源码文档里明确有安全提示：

Setting configurable_fields="any" means fields like api_key, base_url, etc.,
can be altered at runtime.

风险在于：

• 用户可能改 api_key。
• 用户可能改 base_url，把请求导向未知服务。
• 用户可能改 model_provider，绕过你预期的供应商。
• 用户可能改 max_tokens、temperature，造成成本和输出质量不可控。

所以生产环境更推荐白名单：

model = init_chat_model(
    "openai:gpt-5.4-mini",
    configurable_fields=("model", "temperature", "max_tokens"),
)

如果 provider 不应该被用户切换，就不要把 model_provider 放进去。

如果 endpoint 不应该被用户改，就不要开放 base_url。

可配置模型很强，但 configurable_fields="any" 不适合接收不可信用户输入，生产环境应该显式枚举允许变更的字段。

声明式方法排队：为什么可配置模型也能 .bind_tools()

有一个很有意思的细节。

普通模型可以这样绑定工具：

model = init_chat_model("openai:gpt-5.4-mini")
model_with_tools = model.bind_tools([get_weather])

可配置模型也可以：

model = init_chat_model(
    configurable_fields=("model", "model_provider"),
)

model_with_tools = model.bind_tools([get_weather])

但问题是：可配置模型此时还没有真实 ChatOpenAI 或 ChatAnthropic 对象，怎么调用 .bind_tools()？

答案在 _ConfigurableModel.__getattr__()。

源码里有：

_DECLARATIVE_METHODS = ("bind_tools", "with_structured_output")

当你访问这些方法时，_ConfigurableModel 不会立刻执行，而是把操作记录到队列：

def __getattr__(self, name: str) -> Any:
    if name in _DECLARATIVE_METHODS:
        def queue(*args: Any, **kwargs: Any) -> _ConfigurableModel:
            queued_declarative_operations = list(self._queued_declarative_operations)
            queued_declarative_operations.append((name, args, kwargs))
            return _ConfigurableModel(
                default_config=dict(self._default_config),
                configurable_fields=list(self._configurable_fields),
                config_prefix=self._config_prefix,
                queued_declarative_operations=queued_declarative_operations,
            )

        return queue

真正创建模型时再回放：

model = _init_chat_model_helper(**params)
for name, args, kwargs in self._queued_declarative_operations:
    model = getattr(model, name)(*args, **kwargs)
return model

也就是说：

configurable_model.bind_tools([get_weather])
  -> 不执行真实 bind_tools
  -> 记录 ("bind_tools", ([get_weather],), {})

invoke(config={"model": "openai:gpt-5.4-mini"})
  -> 创建 ChatOpenAI
  -> 在 ChatOpenAI 上执行 bind_tools([get_weather])
  -> 再调用 invoke

这就是为什么可配置模型也能参与 Agent 工具调用。

_ConfigurableModel 对 bind_tools 和 with_structured_output 做了延迟排队，等真实模型创建后再回放这些声明式操作。

BaseChatModel：所有聊天模型共同遵守的统一接口

init_chat_model() 解决的是“创建哪个模型”。

BaseChatModel 解决的是“上层如何调用模型”。

源码位于：

langchain_core/language_models/chat_models.py

BaseChatModel 文档里把方法分成两类。

命令式方法：真正触发调用

方法	输入	输出	作用
invoke	字符串、消息列表、PromptValue	BaseMessage	单次调用
ainvoke	同上	BaseMessage	异步调用
stream	同上	BaseMessageChunk 迭代器	流式输出
astream	同上	异步 chunk 迭代器	异步流式输出
batch	输入列表	消息列表	批量调用
abatch	输入列表	消息列表	异步批量调用

声明式方法：返回新的 Runnable

方法	作用
bind_tools	绑定工具调用 schema
with_structured_output	包装结构化输出
with_retry	增加重试
with_fallbacks	增加 fallback
configurable_fields	声明运行时可配置字段
configurable_alternatives	声明运行时可切换候选模型

这也是为什么模型对象可以直接参与 LCEL：

chain = prompt | model | parser

因为 BaseChatModel 本身是 Runnable。

BaseChatModel 是 Chat 模型的统一 Runnable 接口，上层 Chain、Agent、Middleware 都只需要面向这套接口编程。

invoke() 调用链：字符串如何变成消息，消息如何进供应商 API

BaseChatModel.invoke() 是最核心的方法之一。

源码可以简化成：

def invoke(self, input, config=None, *, stop=None, **kwargs):
    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

第一步是 _convert_input()：

def _convert_input(self, input):
    if isinstance(input, PromptValue):
        return input
    elif isinstance(input, str):
        return StringPromptValue(text=input)
    elif isinstance(input, Sequence):
        return ChatPromptValue(messages=convert_to_messages(input))
    else:
        raise ValueError(...)

所以这些输入都能被接受：

model.invoke("你好")

from langchain_core.messages import HumanMessage, SystemMessage

model.invoke([
    SystemMessage(content="你是一个严谨的技术讲师"),
    HumanMessage(content="解释 BaseChatModel"),
])

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个 {role}"),
    ("human", "{question}"),
])

model.invoke(prompt.invoke({
    "role": "源码讲师",
    "question": "解释 init_chat_model",
}))

它们最终都会变成：

PromptValue
  -> messages
  -> generate_prompt
  -> generate
  -> _generate_with_cache
  -> _generate

其中 _generate 是供应商类必须实现的方法。

invoke() 的统一性来自 _convert_input() 和 generate_prompt()，不同输入格式最终都会被归一化成消息列表。

_generate_with_cache()：缓存、流式回退、回调和元数据都在这里

BaseChatModel 的核心执行层是 _generate_with_cache()。

它做了几类事情：

_generate_with_cache(messages, stop, run_manager, **kwargs)
  |
  +-- 检查 cache
  |
  +-- 如果 cache 命中，直接返回 ChatResult
  |
  +-- 如果请求流式，并且模型实现了 _stream，走 _stream
  |
  +-- 否则走 _generate
  |
  +-- 给 message 补 id / response_metadata
  |
  +-- 写入 cache
  |
  +-- 返回 ChatResult

源码里有一个关键判断：

if type(self)._stream != BaseChatModel._stream and kwargs.pop("stream", ...):
    chunks = []
    for chunk in self._stream(messages, stop=stop, **kwargs):
        ...
    result = generate_from_stream(iter(chunks))
else:
    result = self._generate(messages, stop=stop, run_manager=run_manager, **kwargs)

这说明：

• 如果供应商模型实现了 _stream，并且当前调用要求流式，可以走流式。
• 否则走 _generate。
• 最终统一返回 ChatResult。

这也解释了为什么不同供应商虽然 SDK 差异很大，但 LangChain 上层看到的都是：

AIMessage(content="...")

因为供应商实现会在 _generate() 或 _stream() 中把原始 SDK response 转成 LangChain 的 ChatResult、ChatGeneration、AIMessage。

_generate_with_cache() 是 BaseChatModel 的执行中枢，负责缓存、流式、回调、元数据和最终 ChatResult 统一化。

供应商类：ChatOpenAI 如何接入 BaseChatModel

以本地安装的 langchain_openai 为例，ChatOpenAI 位于：

langchain_openai/chat_models/base.py

类定义大致是：

class ChatOpenAI(BaseChatOpenAI):
    """Interface to OpenAI chat model APIs."""

而 BaseChatOpenAI 会实现 BaseChatModel 需要的核心方法，比如 _generate()：

def _generate(
    self,
    messages: list[BaseMessage],
    stop: list[str] | None = None,
    run_manager: CallbackManagerForLLMRun | None = None,
    **kwargs: Any,
) -> ChatResult:
    payload = self._get_request_payload(messages, stop=stop, **kwargs)

    if "response_format" in payload:
        raw_response = self.root_client.chat.completions.with_raw_response.parse(**payload)
        response = raw_response.parse()
    elif self._use_responses_api(payload):
        raw_response = self.root_client.responses.with_raw_response.create(**payload)
        response = raw_response.parse()
        return _construct_lc_result_from_responses_api(...)
    else:
        raw_response = self.client.with_raw_response.create(**payload)
        response = raw_response.parse()

    return self._create_chat_result(response, generation_info)

你不需要记住 OpenAI 具体 SDK 调用细节，重点是这条适配链：

LangChain messages
  |
  v
_get_request_payload(...)
  |
  v
OpenAI SDK request payload
  |
  v
OpenAI API response
  |
  v
_create_chat_result(...)
  |
  v
ChatResult / AIMessage

这就是供应商类的核心职责：

把 LangChain 标准输入转换成供应商 API 请求，
再把供应商 API 响应转换回 LangChain 标准输出。

ChatOpenAI 不是重新定义调用范式，而是在 BaseChatModel 的 _generate() 扩展点里完成 OpenAI SDK 到 LangChain 消息协议的双向适配。

bind_tools()：统一接口背后仍然有供应商差异

BaseChatModel 定义了 bind_tools()，但默认实现是：

def bind_tools(self, tools, **kwargs):
    raise NotImplementedError()

也就是说，工具调用不是所有模型都天然支持。

具体供应商类要自己实现。

以 ChatOpenAI.bind_tools() 为例，源码里会把 LangChain 工具、Pydantic 类、普通函数等转换成 OpenAI tool schema：

formatted_tools = [
    convert_to_openai_tool(tool, strict=strict)
    for tool in tools
]

return super().bind(
    tools=formatted_tools,
    **kwargs,
)

这说明 bind_tools() 的统一只是方法入口统一，真正 tool schema 如何表示，仍然取决于供应商。

比如：

• OpenAI 使用 OpenAI tool/function calling 格式。
• Anthropic 使用自己的 tool use 格式。
• 部分本地模型可能通过 OpenAI-compatible server 模拟工具调用。
• 有些模型可能根本不支持原生 tool calling。

LangChain 的做法是：

BaseChatModel 定义能力入口
供应商类负责转换成本供应商能理解的 schema
上层 Agent 只调用 bind_tools()

这也是 create_agent() 可以对不同模型复用工具调用逻辑的前提。

一句话总结：bind_tools() 是统一能力入口，不是统一底层协议；供应商类负责把工具 schema 翻译成各自 API 能接受的格式。

with_structured_output()：结构化输出也是模型层能力

结构化输出同样挂在 BaseChatModel 上：

def with_structured_output(
    self,
    schema: dict | type[BaseModel],
    *,
    include_raw: bool = False,
    **kwargs,
) -> Runnable:
    ...

它的目标是：

from pydantic import BaseModel, Field


class Person(BaseModel):
    name: str = Field(description="姓名")
    age: int = Field(description="年龄")


structured_model = model.with_structured_output(Person)
result = structured_model.invoke("张三今年 18 岁")

上层希望拿到：

Person(name="张三", age=18)

但底层可能有多种实现方式：

• provider-native structured output。
• tool calling。
• JSON mode。
• 输出解析器。

第 9 篇我们讲过结构化输出策略，第 23 篇又看到 create_agent() 会根据 response_format 决定是否绑定 provider strategy 或 tool strategy。

从模型抽象层看，关键仍然是：

BaseChatModel 提供统一声明式方法
具体模型决定如何绑定 schema
上层得到 Runnable

结构化输出不是简单 prompt 技巧，而是模型抽象层暴露出来的声明式能力，具体实现由 provider 能力决定。

为什么说这是“工厂模式 + SPI 机制”

把源码抽象成设计模式，会比较清楚。

工厂模式：

model = init_chat_model("openai:gpt-5.4-mini")

调用者不直接写：

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-5.4-mini")

而是把创建逻辑交给 init_chat_model()。

工厂内部根据字符串决定创建哪个类。

SPI 机制：

BaseChatModel 定义接口
langchain-openai 提供 ChatOpenAI
langchain-anthropic 提供 ChatAnthropic
langchain-ollama 提供 ChatOllama
_BUILTIN_PROVIDERS 负责把 provider key 映射到实现类

这让 LangChain 核心包只需要依赖抽象，不需要硬编码每个 SDK 的调用细节。

图示：

业务代码 / Chain / Agent
        |
        v
   BaseChatModel
        |
        v
 init_chat_model()
        |
        v
 _BUILTIN_PROVIDERS
        |
        +-- openai    -> langchain_openai.ChatOpenAI
        +-- anthropic -> langchain_anthropic.ChatAnthropic
        +-- deepseek  -> langchain_deepseek.ChatDeepSeek
        +-- ollama    -> langchain_ollama.ChatOllama

这套设计的收益是明显的：

• prompt | model | parser 不关心模型来自哪个供应商。
• create_agent(model=...) 不关心模型类名。
• Middleware 的 ModelRequest.model 只要求是 BaseChatModel。
• 工具调用和结构化输出通过统一方法挂接。
• 供应商适配逻辑被隔离在独立集成包。

init_chat_model() 是工厂，BaseChatModel 是接口，供应商集成包是插件实现，三者组合起来构成了 LangChain 的模型层扩展机制。

常见问题一：只安装 langchain，没有安装供应商包

很多初学者会这样：

pip install langchain

然后运行：

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-5.4-mini")

结果报错：

Initializing ChatOpenAI requires the langchain-openai package.
Please install it with `pip install langchain-openai`

这是符合设计的。

langchain 核心包不内置 OpenAI SDK，OpenAI 集成在：

pip install -U langchain-openai

Anthropic：

pip install -U langchain-anthropic

Ollama：

pip install -U langchain-ollama

DeepSeek：

pip install -U langchain-deepseek

init_chat_model() 能路由到供应商类，但不会替你安装供应商集成包。

常见问题二：模型名能推断，不代表应该依赖推断

下面写法可以工作：

model = init_chat_model("gpt-5.4-mini")

因为 gpt- 会被推断成 openai。

但更推荐：

model = init_chat_model("openai:gpt-5.4-mini")

尤其是团队项目、生产系统、教程代码里。

原因：

• 显式 provider 更容易阅读。
• 避免未来模型别名变化。
• 避免同名模型在不同供应商托管。
• 避免 gemini 这类 provider 接入方式差异。

交互式试验可以偷懒，工程代码应该写 provider:model。

常见问题三：把 OpenAI-compatible 接口都当成 ChatOpenAI

很多模型服务提供 OpenAI-compatible API。

于是有人会写：

model = init_chat_model(
    "openai:deepseek-chat",
    base_url="https://api.deepseek.com",
)

这可能能跑，但不一定是最佳实践。

ChatOpenAI 的源码注释里也强调：它面向官方 OpenAI API 规格。第三方供应商额外返回的字段，例如某些推理字段，可能不会被完整提取或保留。

如果 LangChain 已经提供 provider-specific 包，优先使用：

model = init_chat_model("deepseek:deepseek-chat")

或者：

from langchain_deepseek import ChatDeepSeek

model = ChatDeepSeek(model="deepseek-chat")

OpenAI-compatible endpoint 适合：

• 内部代理完全兼容 OpenAI 格式。
• 只需要基础文本生成。
• 不依赖供应商特有响应字段。

如果你需要供应商特有能力，应该用对应集成包。

base_url 能改请求地址，但不能自动让 ChatOpenAI 理解其他供应商的所有扩展语义。

常见问题四：以为 BaseChatModel.bind_tools() 一定可用

BaseChatModel 有 bind_tools() 方法，但基类默认抛：

raise NotImplementedError()

所以：

model.bind_tools([tool])

能不能成功，取决于具体模型类是否实现了 tool calling。

如果报错，要检查：

• 当前供应商模型是否支持工具调用。
• 集成包版本是否支持 bind_tools()。
• 传入工具 schema 是否能转换成该供应商支持的格式。
• 模型 ID 本身是否支持 tool calling。

这也是为什么 Agent 选型时不能只看“模型能聊天”，还要看：

能力	对 Agent 的影响
tool calling	能否让模型可靠调用工具
structured output	能否稳定返回 schema
streaming	能否流式展示中间结果
token usage metadata	能否做成本统计
system message 支持	能否可靠注入系统提示词

BaseChatModel 定义统一方法名，但具体能力是否可用，仍要看 provider 和模型本身。

实战建议：生产项目里如何初始化模型

下面是一种更稳的写法。

先封装模型工厂：

from langchain.chat_models import init_chat_model
from langchain_core.language_models import BaseChatModel


def build_chat_model(
    model: str,
    *,
    temperature: float = 0,
    timeout: float = 30,
    max_retries: int = 2,
) -> BaseChatModel:
    return init_chat_model(
        model,
        temperature=temperature,
        timeout=timeout,
        max_retries=max_retries,
    )

业务里显式写 provider：

model = build_chat_model("openai:gpt-5.4-mini")

如果允许运行时切换，只开放白名单字段：

model = init_chat_model(
    "openai:gpt-5.4-mini",
    temperature=0,
    configurable_fields=("model", "temperature", "max_tokens"),
    config_prefix="chat",
)

调用：

response = model.invoke(
    "解释 LCEL",
    config={
        "configurable": {
            "chat_model": "gpt-5.4-mini",
            "chat_temperature": 0.2,
            "chat_max_tokens": 800,
        }
    },
)

如果接入 Agent：

from langchain.agents import create_agent


agent = create_agent(
    model=build_chat_model("openai:gpt-5.4-mini"),
    tools=[search_tool, calculator_tool],
    system_prompt="你是一个严谨的技术助手。",
)

或者直接传字符串，让 create_agent() 内部处理：

agent = create_agent(
    model="openai:gpt-5.4-mini",
    tools=[search_tool, calculator_tool],
)

如果你需要统一控制 timeout、max_retries、base_url 等参数，封装 build_chat_model() 更清晰。

生产项目里建议显式 provider、封装模型工厂、白名单开放运行时配置，不要把所有模型参数都暴露给外部输入。

总结：模型抽象层的关键设计

最后用一张图收束：

业务代码
  |
  v
init_chat_model("openai:gpt-5.4-mini")
  |
  v
_parse_model()
  |
  +-- provider = openai
  +-- model = gpt-5.4-mini
  |
  v
_get_chat_model_creator("openai")
  |
  +-- _BUILTIN_PROVIDERS
  +-- import langchain_openai
  +-- getattr ChatOpenAI
  |
  v
ChatOpenAI(model="gpt-5.4-mini")
  |
  v
BaseChatModel.invoke()
  |
  v
_convert_input()
  |
  v
generate_prompt()
  |
  v
_generate_with_cache()
  |
  v
ChatOpenAI._generate()
  |
  v
OpenAI API
  |
  v
AIMessage

记住四个结论：

• init_chat_model() 是工厂函数：负责解析字符串、查 provider、动态 import、实例化具体模型。
• _BUILTIN_PROVIDERS 是轻量 SPI 注册表：把 provider key 绑定到独立集成包和模型类。
• _ConfigurableModel 是延迟代理：支持通过 RunnableConfig 在运行时切模型、切参数。
• BaseChatModel 是统一调用协议：上层 Chain、Agent、Middleware 只依赖 invoke/stream/batch/bind_tools/with_structured_output 这些统一方法。