ChatModel

你要把同一套业务逻辑接到 OpenAI、Anthropic 或本地 Ollama，若直接写各厂商 SDK，messages 字段名、流式回调、错误类型都不一致。换模型等于重写调用层。ChatModel 把「发 messages、收 AIMessage」收成同一套 invoke / stream 协议，上层 Prompt 与 LCEL 链不必跟着改。

段末注释：ChatModel = LangChain 对「对话式大语言模型（large language model，LLM）」的统一抽象，输入输出均为消息列表或单条消息。

1. 一句话定位

维度	内容
角色	能力层模型入口：封装厂商 API，产出 `AIMessage`
输入 → 输出	`LanguageModelInput`（多为 `list[BaseMessage]` 或 str）→ `AIMessage`
典型调用入口	`model.invoke(messages)`、`model.bind_tools(...)`、`model.with_structured_output(...)`
与 LangGraph	图节点内调用同一 ChatModel；环与 checkpoint 不在本篇

2. 实现逻辑

一次同步调用的运行时步骤：

1. 构造 messages：SystemMessage / HumanMessage / 历史 AIMessage、ToolMessage
2. （可选）bind_tools / with_structured_output 返回新 Runnable 实例
3. model.invoke(input, config=...)  → 内部序列化为厂商 payload
4. HTTP/SDK 请求 → 解析为 AIMessage（content、tool_calls、usage_metadata）
5. 下游：StrOutputParser 取 content，或读 tool_calls 进入工具循环

字段级变形示例：

1
2
3

[HumanMessage(content="北京天气？")]
  → invoke
AIMessage(content="", tool_calls=[{name:"get_weather", args:{city:"北京"}, id:"call_1"}])

3. 原理说明

3.1 BaseChatModel 协议

所有 Chat 集成继承 langchain_core.language_models.chat_models.BaseChatModel，内部实现 _generate / _agenerate；对外由 Runnable 包装成 invoke / batch / stream / ainvoke。

BaseChatModel（抽象类）
功能：各厂商 Chat Completions → 统一 messages 进、AIMessage 出。
默认值：temperature 视集成（OpenAI 兼容端常为 None=服务端默认）；max_retries=2；timeout=None。
最小维度：messages: list[BaseMessage]，len ≥ 1；空列表多数实现报错或空回复。

invoke(input, config=None)（方法）
功能：同步一次补全。config=None 时内部填空 tags/metadata。
最小维度：input 为 str（自动包成 HumanMessage）或 list[BaseMessage]（len≥1）。
返回：单条 AIMessage。

1	ai = model.invoke([HumanMessage(content="你好")])

_generate / _agenerate（子类方法）
功能：真正发厂商请求；业务代码不要直接调。最小维度同 invoke。

batch(inputs, config=None, *, return_exceptions=False)（方法）
功能：多条输入一次调度。默认 return_exceptions=False（一条失败整批抛错）。
最小维度：inputs 为 list，len≥1；空列表返回 []。

1	ais = model.batch([[HumanMessage("1+1")]], return_exceptions=False)

stream(input, config=None)（方法）
功能：同步迭代 AIMessageChunk。参数与 invoke 相同。生产异步用 astream / ainvoke。

1	text = "".join(c.content for c in model.stream([HumanMessage("首都？")]))

3.2 与 OpenAI Chat Completions 的映射

LangChain	OpenAI API
`SystemMessage`	`role: system`
`HumanMessage`	`role: user`
`AIMessage.tool_calls`	`tool_calls` 数组
`temperature`	同名参数

Partner 包负责双向转换；换厂商时 Message 类型不变。

SystemMessage / HumanMessage（数据类）
功能：系统指令 / 用户轮次。
数据结构：content: str | list（必填）；name: str | None = None；id: str | None = None。
最小维度：content 至少 1 个字符才有语义（空字符串语法合法）。

1 2	SystemMessage(content="只答事实。") HumanMessage(content="北京天气？")

AIMessage（数据类）
功能：模型回复。工具轮次 content 可为 ""。

字段	类型	默认值	最小维度
`content`	str / list	必填	0 字符合法
`tool_calls`	list	`[]`	0=纯文本；≥1 进入工具循环
`usage_metadata`	dict / None	`None`	—
`response_metadata`	dict	`{}`	—

tool_calls 单项（dict）
name: str（非空）、args: dict（无参为 {}）、id: str（非空，与 ToolMessage 对齐）。最小一条调用 = 这 3 个键。

3.3 bind 与 immutable 配置

bind(**kwargs)（方法）
功能：返回新 Runnable，不改原 model。未传入的字段保持原值。
最小维度：至少绑 1 个参数才有意义。

1	bound = model.bind(temperature=0, max_tokens=64)

bind_tools(tools, *, tool_choice=None)（方法）
功能：注册工具 schema。tool_choice 默认视厂商（多为 "auto"）。
最小维度：tools len≥1；空列表等价未绑定。

1	model_with_tools = model.bind_tools([get_weather])

3.4 流式 stream

AIMessageChunk（数据类）
功能：流式增量，可用 + 与其它 chunk 合并；拼起来与 invoke 的 AIMessage.content 同义。
数据结构：与 AIMessage 同形；content 常为短 str，默认 "" 的空 chunk 可能出现。
最小维度：一次 stream 至少 yield 1 个 chunk（视实现）。

4. 最小可运行示例

验证于 langchain-openai + 本地 Ollama。需 ollama serve 且已 ollama pull qwen3.5:9b。云端改为 ChatOpenAI(model="gpt-4o-mini") 并设置 OPENAI_API_KEY。

1	pip install -U langchain-openai

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

model = ChatOpenAI(
    model="qwen3.5:9b",
    api_key="ollama",
    base_url="http://localhost:11434/v1",
    temperature=0,
)

messages = [
    SystemMessage(content="只答一个词，不要标点。"),
    HumanMessage(content="中国的首都是哪座城市？"),
]
result = model.invoke(messages)
print(type(result).__name__, result.content)
# 预期形态：AIMessage 北京（措辞随模型可能略有不同）

print("stream:", end=" ")
for chunk in model.stream(messages):
    print(chunk.content, end="")
print()
# 预期：多个 AIMessageChunk，拼起来与 invoke 同义

bound = model.bind(max_tokens=32)
print(bound.invoke([HumanMessage("用四个字解释什么是列表。")]).content)

重要配置参数

参数（API 名）	类型 / 默认值	功能说明	作用与影响	参考起点	配置指导
`model`	str，必填，构造器	指定调用哪一个 Chat 模型 ID	能力、单价、延迟随 ID 变；换模型不必改 messages	`gpt-4o-mini`	原型用小模型；上线前按任务评测
`temperature`	float，视厂商，构造器/`bind`	控制采样随机性	越高越发散、越低越稳定；对延迟几乎无影响	0～0.3（事实/工具）	工具调用、JSON 输出宜偏低
`max_tokens`	int，可选	限制本次回复最多生成多少 token	过小截断 tool JSON；过大费钱、变慢	1024～4096	工具调用至少留足 schema 长度
`timeout`	float，可选	单次 HTTP 请求最长等待秒数	过短易误杀慢请求；过长拖垮线程	30～120	RAG 长上下文可加大
`max_retries`	int，默认 2	SDK 层对网络/5xx 自动重试次数	次数多则更稳、尾延迟更高	2～5	网络抖动场景保留
`api_key`	str，可选	厂商鉴权凭证	缺失则请求失败	环境变量	勿硬编码进仓库
`streaming`	bool，默认 False	是否走流式 Completions API	True 时须用 `stream` 消费；invoke 仍一次返回	False	打字机效果在调用侧用 `stream`

5. 易踩坑

只装 langchain 不装 partner 包：ChatOpenAI 在 langchain-openai，import 失败与 LangChain 版本无关。
把 dict 当 messages 长期混用：短期可 {role, content}，但 tool 回填必须用 ToolMessage 且带 tool_call_id，否则下一轮报错。
在 async 路由里用同步 invoke：FastAPI 会阻塞事件循环；改用 ainvoke。

小结

ChatModel 统一 messages 进、AIMessage 出，是 LCEL 与 Agent 的模型底座。
换厂商只换构造器，链结构可保留。
工具与结构化输出通过 bind_tools / with_structured_output 挂在同一 model 上。
最易踩坑：partner 包缺失 与 ToolMessage 缺 id。