# 使用LangChain+LangGraph开发AI应用（2） - 模型抽象


![](https://static.xiongneng.me/langchain-langgraph-cover-20261017122621.png)

## 学习进度

模型是应用的大脑，这一篇讲怎么把它接进来。LangChain 的模型抽象把所有提供商收拢到一个接口后面，你的代码只面对 `BaseChatModel`，面对 `invoke`、`stream`、`batch` 三个方法，换模型只改一个字符串，不用再碰每家 SDK 各自的鉴权、参数名和流式协议。

示例在 `source/hello_models.py`，从选型到调参一次讲完。完整案例是一个模型试用台 `source/ask.py`，接新模型之前，先用它过一遍。

## 前言用途

模型是 Agent 最重要的一次选型，也是第一个绕不开的决策。接哪个模型，直接决定回答质量、每个月的话费、数据出不出你的机器。选错了，后面全白搭。

但我见过太多人把模型接进来这件事做复杂了。每家的 SDK 形状都不一样，鉴权、参数名、流式协议也各有各的规矩。今天用 OpenAI，明天想换 DeepSeek 试试水，代码要重写一半。更麻烦的是一篇文章里想对比三个模型的输出，得维护三套调用代码。

LangChain 的模型抽象就是来解决这个问题的。它把所有提供商收拢到一个接口后面，你的代码只面对 `BaseChatModel`，面对 `invoke`、`stream`、`batch` 三个方法，换模型只改一个字符串。这一篇把这个接口从设计到用法讲透，再配一个能上手的模型试用台案例。接新模型之前，先用它过一遍。

## 原理概念

没有抽象的世界长什么样？OpenAI 的 SDK 是 `client.chat.completions.create`，Anthropic 是 `client.messages.create`，Google 又是另一套。鉴权方式、参数名、流式协议，家家都不一样。你的业务代码如果直接写这些调用，就等于和某一家提供商签了卖身契。

LangChain 的做法是定义统一接口。所有聊天模型都实现 `BaseChatModel` 这个基类，对外只有三个方法，`invoke` 单次调用，`stream` 流式输出，`batch` 批量调用。输入统一是消息列表，输出统一是 `AIMessage`。至于背后怎么拼 HTTP 请求、怎么解析流式响应，是各家接入包自己的事。

![](https://static.xiongneng.me/model-abstraction-20261017122518.png)

在这个接口之上，LangChain 提供了一个工厂函数 `init_chat_model`，这就是本篇的主角。它接收一个字符串，返回一个可用的模型实例。字符串用 `提供商:模型名` 的格式，冒号前面是提供商，后面是模型名。

```python
from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-5.5")
```

这一行代码背后，工厂函数做了三件事，按提供商找到对应的接入包，读取该提供商的凭证（通常从环境变量），把模型参数组装成一个 `BaseChatModel` 实例返回。你拿到的 `model` 对象，用起来和任何其他模型一模一样。

裸模型名也接受，工厂会按前缀推断提供商，比如 `gpt-` 开头的推断为 OpenAI。但官方文档明确说了，推断是尽力而为，不保证正确。我的建议和官方一致，**永远显式写提供商前缀**，一个字符串的事，不值得赌。

消息是这个接口的通用货币。`SystemMessage` 装系统提示，`HumanMessage` 装用户输入，`AIMessage` 装模型输出，`ToolMessage` 装工具返回。多轮对话就是这些消息排成的列表，`invoke` 原样收下。1.x 之后推荐从 `langchain.messages` 导入这些类，它和第一篇里用的 `langchain_core.messages` 是同一批类，换个更短的入口。

`init_chat_model` 还支持**可配置模型**。构造时不指定模型名，调用时通过 `config` 字典动态指定，同一个实例就能在多个模型间切换。

```python
from langchain.chat_models import init_chat_model

model = init_chat_model(temperature=0)

# 同一个实例，运行时换模型，代码不用动
model.invoke("你好", config={"configurable": {"model": "openai:gpt-5.5"}})
model.invoke("你好", config={"configurable": {"model": "openai:gpt-5.5-nano"}})
```

做模型对比、灰度切换时很有用，本篇案例的思路和它一脉相承。

模型能力还能自省。LangChain 1.1 起，模型实例带一个 `profile` 属性，返回能力字典，上下文多长、支不支持图片输入、支不支持工具调用、是不是推理模型。数据来自 models.dev 开源项目。接一个新模型，先打一句 `model.profile` 看看，比翻半天文档快。

## 核心用法

四种调用方式，两个传参时机。这一节用 `hello_models.py` 逐个演示，你仓库里能看到全部源码。

![](https://static.xiongneng.me/call-modes-20261017122521.png)

### 1. invoke（单次调用）

最简单的用法，传一个字符串，框架自动包成 `HumanMessage`。

```python
response = model.invoke("北京天气怎么样？")
print(type(response).__name__)  # AIMessage
print(response.text)            # 模型的回答
```

注意返回的是 `AIMessage` 对象，不是字符串。回答文本用 `.text` 属性取，它会把内容块拼成纯文本。如果你拿到的是裸字符串，说明用的是旧版 text-completion 接口，该升级了。

带历史的调用要传消息列表。`SystemMessage` 放最前面定人设，然后交替放用户和模型的消息。

```python
from langchain.messages import AIMessage, HumanMessage, SystemMessage

conversation = [
    SystemMessage("你是一个简洁的地理助手，回答不超过两句话。"),
    HumanMessage("北京是哪座城市？"),
    AIMessage("北京是中华人民共和国的首都，位于华北平原北部。"),
    HumanMessage("它有多少年的建城史？"),
]
response = model.invoke(conversation)
```

字符串入参只能聊一轮，因为框架不知道历史。多轮对话必须自己维护消息列表，把每一轮的问答追加进去。这个习惯从第一篇的 Agent 示例就开始了，`create_agent` 内部帮你做的也是这件事。

### 2. stream（流式输出）

长回答等完整响应，用户体验很差。`stream` 返回一个迭代器，模型每产出一点就 `yield` 一个 `AIMessageChunk`。

```python
full = None  # None | AIMessageChunk
for chunk in model.stream("演示一下流式输出"):
    full = chunk if full is None else full + chunk
    print(chunk.text, end="", flush=True)
print()
print(full.text)  # 聚合后的完整回答
```

这里有个关键设计，**chunk 之间用加法聚合**。`AIMessageChunk` 重载了 `+` 运算符，内容、工具调用、用量元数据都会正确合并。别自己去拼 `chunk.content`，一是它可能是列表（多模态内容块），直接字符串相加会炸；二是工具调用的增量在 `content` 里根本看不到。

流式还有两个实战细节。第一，推理模型（比如本篇测试用的 `deepseek-v4.1-flash`）会先输出一批**空文本 chunk**，那是思考 token，`chunk.text` 是空串，不是 bug，聚合结果不受影响。第二，`chunk_position` 属性标了 `last` 的 chunk 是最后一个，需要在流结束时 flush UI 的场合会用到。

本地替身上跑一遍，`stream` 的输出长这样。

```text
== stream（流式输出） ==
  收到文本: '流式'
  收到文本: ' '
  收到文本: '输出'
  收到文本: '把'
  收到文本: '回答'
  收到文本: '切成'
  收到文本: '小块'
  收到文本: '逐块'
  收到文本: '送达'
共 15 个 chunk，聚合后: 流式 输出 把 回答 切成 小块 逐块 送达
chunk 类型: AIMessageChunk（AIMessage 的子类）
```

替身模型按空白切片，所以 chunk 数很整齐。真实模型一个 token 一个 chunk，数量完全不规则，累积逻辑却一模一样。

### 3. batch（批量并行）

要问模型一堆问题，别写循环 `invoke`。`batch` 把输入列表一起发出去，内部并行执行，返回顺序和输入一致。

```python
questions = ["第一个批量问题", "第二个批量问题"]
responses = model.batch(questions)
for q, r in zip(questions, responses):
    print(f"问: {q} -> 答: {r.text}")
```

输出长这样。

```text
== batch（批量调用） ==
  问: 第一个批量问题 -> 答: 第一个批量问题的回答。
  问: 第二个批量问题 -> 答: 第二个批量问题的回答。
```

`batch_as_completed` 则按完成顺序产出，哪个先完成先处理，带输入索引。并发数用 `config` 控制，写成 `model.batch(inputs, config={"max_concurrency": 5})`。

要强调一句，**这是客户端并行**，LangChain 起多个线程同时打 API，不是 OpenAI、Anthropic 那种服务端 Batch API。服务端 Batch 通常有折扣但按批次处理、以小时计；客户端并行立刻返回，但打的是你的并发额度。不限并发容易撞 429 限流，生产上一定要设 `max_concurrency`。

### 4. 参数传递（两个时机）

模型参数有两类，传错的时机不生效，这是新手最常踩的迷糊。

**构造时传模型参数**，通过 `init_chat_model` 的关键字参数，对整个实例生效。

```python
model = init_chat_model(
    "openai:gpt-5.5",
    temperature=0.7,    # 随机性，0 最确定，1 最发散
    max_tokens=1000,    # 回答 token 上限
    timeout=30,         # 单次请求超时（秒）
    max_retries=6,      # 失败重试次数，默认 6，指数退避加抖动
)
```

网络错误、429 限流、5xx 服务端错误会自动重试；401、404 这类客户端错误不重试，重试也没用。网络不好的环境把 `max_retries` 提到 10 到 15，`timeout` 加到 120。

**调用时传运行配置**，通过 `invoke` 的 `config` 字典。它不改变模型行为，供追踪和检索用。

```python
response = model.invoke(
    "这一趟调用带了哪些运行配置？",
    config={
        "run_name": "demo_params",              # 这次调用在追踪里的名字
        "tags": ["lesson-02", "models"],        # 标签，便于检索
        "metadata": {"source": "hello_models"}, # 自定义元数据
    },
)
```

配了 LangSmith，这些字段就是你检索执行的入口。部分参数（如 `reasoning_effort`）按模型支持，构造时或调用时传都行，调用时传更灵活。

### 5. 换提供商（只改一个字符串）

抽象的全部价值，最后浓缩成这一行 diff。同一份代码，换个模型名就完成了迁移。

```python
model = init_chat_model("anthropic:claude-sonnet-4-6")   # 换 Anthropic
model = init_chat_model("google_genai:gemini-3.7-flash") # 换 Google
model = init_chat_model("ollama:qwen3:8b")               # 换本地 Ollama
```

调用代码一行不用动，`invoke`、`stream`、`batch` 原样工作。这就是一个接口，百家模型。

本系列的测试环境走 API易 网关，它是 OpenAI 兼容的，用 `openai:` 前缀加自定义 `base_url` 接入。我写的示例自动读取环境变量，你只要设好三个变量就能跑真实模型。

```bash
# Linux / macOS
export AGENT_MODEL="openai:deepseek-v4.1-flash"
export YI_BASE_URL="https://你的网关地址/v1"
export YI_API_KEY="你的key"

# Windows PowerShell
$env:AGENT_MODEL="openai:deepseek-v4.1-flash"
$env:YI_BASE_URL="https://你的网关地址/v1"
$env:YI_API_KEY="你的key"
```

不设 `AGENT_MODEL` 时，示例用本地脚本化模型演示，不需要任何 key，行为和第一篇一样。

## 完整案例

用法都演示过了，收个尾，写一个模型试用台 `ask.py`，把接入新模型时要问的问题一次问清。接进来要写几行代码？temperature 调了有什么变化？流式长什么样？一次调用烧多少 token、等多久？

核心函数不到二十行。

```python
def ask(model, question, on_token=None):
    """流式提问，返回回答文本与统计数据。"""
    started = time.perf_counter()
    full = None  # None | AIMessageChunk
    n_chunks = 0
    for chunk in model.stream(question):
        full = chunk if full is None else full + chunk
        n_chunks += 1
        if on_token is not None:
            on_token(chunk.text)
    elapsed = time.perf_counter() - started
    usage = full.usage_metadata if full is not None else None
    return {
        "text": full.text if full is not None else "",
        "chunks": n_chunks,
        "usage": usage,
        "elapsed_s": round(elapsed, 3),
    }
```

四个字段各管一件事，`text` 是聚合后的完整回答，`chunks` 是收到的块数，`usage` 是 token 用量，`elapsed_s` 是端到端耗时。`on_token` 回调让调用方自己决定怎么显示，命令行就打印，网页就追加节点。

命令行入口用 argparse 收三个参数，`--model` 换模型，`--temperature` 调随机性，`--max-tokens` 限长度。不传模型名时读环境变量，再没有就用本地替身。

先看不传参数的本地演示。

```bash
uv run python ask.py "北京天气怎么样？"
```

```text
模型: ScriptedChatModel（本地替身）
参数: temperature=None max_tokens=None
问题: 北京天气怎么样？
回答: 北京 今天 晴天，气温 26℃，适合出行。

---- 统计 ----
chunk 数: 7
耗时: 0.041s
token 用量: 无（fake 模型不产用量，真实模型才有）
```

替身模型不产用量，`usage` 是 `None`，程序明确告诉你原因，不装。接上真实模型再跑一遍。

```bash
# 先设好 AGENT_MODEL / YI_BASE_URL / YI_API_KEY
uv run python ask.py "用一句话解释什么是模型抽象" --temperature 0
```

```text
模型: openai:deepseek-v4.1-flash
参数: temperature=0.0 max_tokens=None
问题: 用一句话解释什么是模型抽象
回答: 模型抽象是指从具体问题或系统中提取关键特征、忽略次要细节，形成可计算、可分析的简化表示。

---- 统计 ----
chunk 数: 70
耗时: 2.562s
token 用量: 输入 36 / 输出 68 / 合计 104
```

四行代码接入，一次调用拿到全部统计。接一个新模型，先跑它十来个问题，答得好不好、快不快、烧多少 token，一看就知道，再决定用不用。

这个案例还顺带回答了一个隐性成本问题。同一条问题，另一次运行我的统计是这样的，输入 36 token，输出 131 token，合计 167。看着不多，拆开 `usage_metadata` 的明细，`output_token_details` 里 `reasoning` 占了 100。推理模型烧的 token 一大半在思考上，你看到的回答只是成品。按输出计费的模型，这笔钱做预算时必须算进去。

案例的测试在 `tests/test_ask.py`，四条用例，返回值结构、`on_token` 回调、替身分支、真实模型分支的参数透传。全部用替身模型跑，不花一分钱。

## 源码

[https://github.com/yidao620c/xneng-langchain-langgraph](https://github.com/yidao620c/xneng-langchain-langgraph)

## 参考

- [LangChain 官方文档 - 模型](https://docs.langchain.com/oss/python/langchain/models)
- [LangChain 官方文档 - 概览](https://docs.langchain.com/oss/python/langchain/overview)
- [LangChain 官方文档 - 安装指南](https://docs.langchain.com/oss/python/langchain/install)


---

> 作者: XiongNeng  
> URL: https://xiongneng.me/posts/ai/langchain/langchain-langgraph-02-models/