language_models/chat_models.py:272，类 docstring 自带两张表，这里裁出最关键的第三张——"写一个自定义聊天模型要实现什么"：

# libs/core/langchain_core/language_models/chat_models.py:272
class BaseChatModel(BaseLanguageModel[AIMessage], ABC):
    r"""Base class for chat models.

    Key imperative methods:  （真正调模型的方法）
      invoke: str | list[dict|tuple|BaseMessage] | PromptValue → BaseMessage
      stream: 同上输入 → Iterator[BaseMessageChunk]
      batch:  list[...] → list[BaseMessage]
      ...
    Creating custom chat model:
      | Method/Property        | Description                    | Required |
      | _generate              | 从一组消息生成一个聊天结果        | Required |  ★必修
      | _llm_type (property)   | 模型类型唯一标识（用于日志）      | Required |  ★必修
      | _identifying_params    | 模型参数化描述（用于追踪）        | Optional |
      | _stream                | 实现流式                        | Optional |  ★选修
      | _agenerate / _astream  | 原生异步版                      | Optional |
    """

D04 留下的钩子今天补上——BaseChatModel.invoke 全文（chat_models.py:463-487）：

# libs/core/langchain_core/language_models/chat_models.py:463（完整正文）
def invoke(self, input, config=None, *, stop=None, **kwargs) -> AIMessage:
    config = ensure_config(config)                       # D04 的老朋友：补齐工单
    return cast(
        "AIMessage",
        cast(
            "ChatGeneration",
            self.generate_prompt(
                [self._convert_input(input)],            # ① 输入归一化，且包成"单元素列表"
                stop=stop,
                callbacks=config.get("callbacks"),       # ② 把工单拆开递给 generate 体系
                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],                         # ③ 批量结果里取第 [0][0] 份
        ).message,                                       # ④ 从 Generation 里剥出 AIMessage
    )

generate（chat_models.py:1564）签名收的是 list[list[BaseMessage]]（多组对话）。它的核心段落——给每组输入开监理、逐组调用缓存版生成（chat_models.py:1642-1673，裁剪）：

# libs/core/langchain_core/language_models/chat_models.py:1642 附近（裁剪）
run_managers = callback_manager.on_chat_model_start(   # ① 每组输入开一个 run（监理）
    self._serialized, messages_to_trace,
    invocation_params=params, options=options,
    name=run_name, run_id=run_id, batch_size=len(messages),
)
results = []
input_messages = [_normalize_messages(m) for m in messages]
for i, m in enumerate(input_messages):                 # ② 逐组处理
    try:
        results.append(
            self._generate_with_cache(                 # ③ ★交给"总调度"（L05）
                m, stop=stop,
                run_manager=run_managers[i] if run_managers else None,
                **kwargs,
            )
        )
    except BaseException as e:
        if run_managers:
            run_managers[i].on_llm_error(e, response=...)  # ④ 哪组失败哪组报
        raise
# 之后：把逐组的 ChatResult 汇总成 LLMResult，并逐组 on_llm_end

今天最重要的函数（chat_models.py:1864）。三步走——查缓存（:1871-1901）、限流（:1907）、选路执行（:1941-1998），最后回填缓存（:2020）：

# libs/core/langchain_core/language_models/chat_models.py:1864（裁剪，行号真实）
def _generate_with_cache(self, messages, stop=None, run_manager=None, **kwargs):
    llm_cache = self.cache if isinstance(self.cache, BaseCache) else get_llm_cache()
    check_cache = self.cache or self.cache is None      # False=明确关闭才不查
    if check_cache and llm_cache:
        llm_string = self._get_llm_string(stop=stop, **kwargs)  # 模型+参数 → 指纹
        prompt = dumps(normalized_messages)                     # 消息序列化 → 钥匙
        cache_val = llm_cache.lookup(prompt, llm_string)        # :1888 ★查缓存
        if isinstance(cache_val, list):
            return ChatResult(generations=...)                  # 命中 → 直接返回，零 API
    # :1907  Apply the rate limiter after checking the cache（源码注释：
    #        缓存查询不限流，真要发 API 请求才限流）
    if self.rate_limiter:
        self.rate_limiter.acquire(blocking=True)

    # 选路：三条路挑一条
    if self._should_use_protocol_streaming(...):        # 路A：v2 协议事件流（新）
        ...
    elif self._should_stream(async_api=False, run_manager=run_manager, **kwargs):
        for chunk in self._stream(messages, stop=stop, **kwargs):   # :1953 路B
            run_manager.on_llm_new_token(chunk.message.content, chunk=chunk)
            chunks.append(chunk)
        result = generate_from_stream(iter(chunks))     # 攒块拼成完整结果
    elif inspect.signature(self._generate).parameters.get("run_manager"):
        result = self._generate(messages, stop=stop,    # :1994 路C：普通一次性生成
                                run_manager=run_manager, **kwargs)
    else:
        result = self._generate(messages, stop=stop, **kwargs)     # :1998
    ...
    if check_cache and llm_cache:
        llm_cache.update(prompt, llm_string, result.generations)   # :2020 ★回填缓存
    return result

用户直接调 model.stream(...) 时走这里（chat_models.py:715-780，裁剪）：

# libs/core/langchain_core/language_models/chat_models.py:715（裁剪）
def stream(self, input, config=None, *, stop=None, **kwargs):
    if not self._should_stream(async_api=False, **{**kwargs, "stream": True}):
        # Model doesn't implement streaming, so use default implementation
        yield cast("AIMessageChunk",
                   self.invoke(input, config=config, stop=stop, **kwargs))  # ① 兜底
    else:
        config = ensure_config(config)
        messages = self._convert_input(input).to_messages()
        ...
        (run_manager,) = callback_manager.on_chat_model_start(...)   # ② 开监理
        chunks: list[ChatGenerationChunk] = []
        if self.rate_limiter:
            self.rate_limiter.acquire(blocking=True)                 # ③ 限流
        try:
            for chunk in self._stream(input_messages, stop=stop, **kwargs):  # ④ ★真流式
                ...
                run_manager.on_llm_new_token(chunk.message.content, chunk=chunk)
                chunks.append(chunk)                                 # ⑤ 边吐边攒
                yield cast("AIMessageChunk", chunk.message)          # ⑥ 吐给调用方
        ...
        # 结束后：attach 一个 chunk_position="last" 的收尾块，并 on_llm_end