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

学习进度

模型是应用的大脑，这一篇讲怎么把它接进来。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 请求、怎么解析流式响应，是各家接入包自己的事。

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

1
2
3

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 字典动态指定，同一个实例就能在多个模型间切换。

1
2
3
4
5
6
7

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 逐个演示，你仓库里能看到全部源码。

1. invoke（单次调用）

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

1
2
3

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

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

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

1
2
3
4
5
6
7
8
9

from langchain.messages import AIMessage, HumanMessage, SystemMessage

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

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

2. stream（流式输出）

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

1
2
3
4
5
6

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 的输出长这样。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

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

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

3. batch（批量并行）

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

1
2
3
4

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

输出长这样。

1
2
3

== 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 的关键字参数，对整个实例生效。

1
2
3
4
5
6
7

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 字典。它不改变模型行为，供追踪和检索用。

1
2
3
4
5
6
7
8

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

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

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

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

1
2
3

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 接入。我写的示例自动读取环境变量，你只要设好三个变量就能跑真实模型。

1
2
3
4
5
6
7
8
9

# 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、等多久？

核心函数不到二十行。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

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 限长度。不传模型名时读环境变量，再没有就用本地替身。

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

1

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

1
2
3
4
5
6
7
8
9

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

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

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

1
2

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

1
2
3
4
5
6
7
8
9

模型: 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

参考

• LangChain 官方文档 - 模型
• LangChain 官方文档 - 概览
• LangChain 官方文档 - 安装指南