LangChain之02：模型的创建与调用#

本章对应的官网文档：

英文文档：https://docs.langchain.com/oss/python/langchain/models

中文文档：https://docs.langchain.org.cn/oss/python/langchain/models

1、模型调用的准备工作#

1.1 一张老图看大模型的调用#

在 LangChain v0.3 版本中，提到了 Model I/O，包括输入提示（Format）、调用模型（Predict）、输出解析（Parse），分别对应着 Prompt Template、Model 和 Output Parser。

关于模型调用模块，如今对话模型已经是主要形式。从历史上解读：

在 GPT-3 时代，大模型以补全模型为主，只能以类似“成语接龙”的方式对文本进行补全，并且实际运行效果也非常不稳定。此时 LangChain 借助一些高层封装的 API，能够让模型完成对话、调用外部工具、甚至是结构化输出等功能，为开发者提供了极大的便利。
伴随着 GPT-3.5 模型的发布，对话模型正式登上历史的舞台，并逐渐成为主流。而得益于对话模型更强的指令跟随能力，很多 GPT-3 需要借助 LangChain 才能完成的工作，已经成为 GPT-3.5 原生自带的一些功能。

所以，本章只提供了对话模型的创建，而没有了非对话模型。

1.2 模型初始化的分类方式#

简单来说，就是用谁家的 API、以什么方式创建、存放在哪个位置的大模型。

角度1：调用谁家的 API

使用模型提供商的库
使用 LangChain 统一方式（推荐）

角度2：模型初始化时，几个重要参数（如 BASE_URL、API-KEY）的书写位置的不同：

使用配置文件（推荐）
硬编码：写在代码文件中

角度3：调用的模型所在位置

在线部署的大模型
本地部署的大模型

LangChain 作为一个“工具”，不提供任何 LLMs，而是依赖于第三方集成各种大模型。这里就看大模型到底部署在哪里。

1.3 线上大模型服务平台#

有许多提供大模型 API 服务的平台，使用时只需要注册、充值并创建 API-Key，之后即可使用 API-Key 与 URL 来调用平台提供的相应的模型的服务。

说明：每个平台配置时，都需要几个要素：模型名、api-key、base-url。如果大家想使用国外的大模型，就选择前两个平台；如果只使用国内的大模型，可以选择后四个。

平台	网址	备注
OpenRouter	https://openrouter.ai/	全球主流，含国外模型
CloseAI	https://platform.closeai-asia.com/	亚洲最大，含国外模型
阿里云百炼	https://bailian.console.aliyun.com/	企业端友好
硅基流动	https://www.siliconflow.cn/	性价比高，适合个人
百度千帆	https://console.bce.baidu.com/qianfan/overview	主打百度生态
火山引擎	https://console.volcengine.com/ark/	主打字节多模态生态

阿里云百炼：所有新用户可获得超过 5000 万 Tokens 的免费额度及 4500 张图片生成额度，适合 toB 企业用户。
硅基流动：号称 9B 以下模型永久免费，开源模型价格低，适合个人学习。
此外，还有各个模型自己的厂商平台，比如 deepseek、智谱等。
OpenRouter 是一个第三方镜像站，专门转发大模型厂商的 API 服务。通过它我们可以间接调用几乎所有大模型厂商的 API 服务，支持国内直连。
OpenRouter 支持支付宝或微信充值，最低限额 $5，税费$ 0.8。此外，如果调用的模型禁止在国内使用（如 ChatGPT），则无法通过 OpenRouter 直接调用（需要 Magic），会提示 This model is not available in your region。根据个人情况，决定是否充值并调用 OpenRouter API。

1.4 提前安装所有依赖#

课程中会涉及到多个库的安装，这里一并声明在 requirements.txt 文件中。同时，LangChain 的版本变化较快，不同版本之间可能存在兼容问题，为了避免因版本不一致导致的问题，本课程会通过 requirements.txt 固定主要依赖版本。

用法：将 requirements.txt 存放到项目所在的目录下，执行：

1
(langchain1.2) PS D:\code\workspace_pycharm_llm\langchain1.2_tutorial>
2
pip install -r .\requirements.txt

说明：课程中的部分章节会单独列出相关依赖，主要是为了帮助大家了解该章节涉及的核心库。无需重复安装，这些依赖已经统一包含在 requirements.txt 中。

2、模型初始化角度1：使用模型提供商库#

在 LangChain 中初始化模型，主要可以通过直接使用特定的 Model Class 和使用统一的 init_chat_model 函数这两种方式来实现。

这里先讲方式 1，这种方式最直接。LangChain 为一些大模型供应商提供了专门的 Model 类，导入对应的具体类（如 ChatOpenAI、ChatAnthropic、ChatDeepSeek、ChatOllama、ChatHunyuan、ChatTongyi、ChatZhipuAI）并进行实例化。

官网链接：https://reference.langchain.com/python/langchain-community/chat-models

2.1 通过专用 API 调用#

注意：使用不同的模型可能传入的参数名称不同，可以参考对应的源码。

2.1.1 DeepSeek 大模型#

官网：https://www.deepseek.com/

步骤1：安装必要的依赖（略）

执行过前面的 requirements.txt 文件指令的情况下，这里就不需要安装了。说明：langchain-deepseek 是使用 deepseek 大模型的必要依赖。注意：langchain-deepseek 依赖于 langchain-openai，安装前者，pip 会自动从 pypi 拉取元数据解析依赖，后者也会被安装。所以我们把 langchain-openai 也放在此处。

1
#切换python环境
2
conda activate langchain1.2
3
#安装ChatOpenAI依赖包
4
pip install langchain-openai
5
#安装ChatDeepSeek 依赖包
6
pip install langchain-deepseek
7
# 用于环境管理的包
8
pip install python-dotenv

步骤2：配置 .env 文件（明确去 deepseek 官网获取 key）

在项目根目录下创建 .env 文件，在 .env 文件中写入以下内容。说明：将占位符替换为自己的 API_KEY。

1
DEEPSEEK_API_KEY=<Your API Key>
2
DEEPSEEK_BASE_URL=https://api.deepseek.com

步骤3：读取配置并初始化模型

这里，我们用 DeepSeek 的模型进行测试，LangChain 会从环境变量中读取 DEEPSEEK_API_KEY。如下是代码实现。

方式1：

1
from langchain_deepseek import ChatDeepSeek
2
import os
3
from dotenv import load_dotenv
4
# 通过load_dotenv()将.env中的变量加载为环境变量
5
# override=True表示：无论你当前的操作系统、终端或者虚拟环境中是否已经存在同名的环境变量，
6
# 都会强行用 .env 文件里写的值去覆盖它
7
load_dotenv(override=True)
8
# 从环境变量读取配置
9
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
10
DEEPSEEK_BASE_URL = os.getenv("DEEPSEEK_BASE_URL")

基于模型集成手册和 LangChain Reference 的 API 参考页 ChatDeepSeek 可知相关的配置参数。

方式2：优化，依靠默认行为读取 .env 环境变量。调用 ChatDeepSeek 要求系统存在名为 DEEPSEEK_API_KEY 的环境变量。URL 通过源码可以查看，有默认值。如下：

1
from langchain_deepseek import ChatDeepSeek
2
from dotenv import load_dotenv
3
# 从.env文件中加载环境变量
4
# override=True 确保.env文件优先
5
load_dotenv(override=True)
6
# 创建DeepSeek LLM
7
deepseek_llm = ChatDeepSeek(
8
    model="deepseek-v4-flash",
9
)
10
print(deepseek_llm.invoke("请介绍一下你自己"))

对应的源码默认值：

1
api_key: SecretStr | None = Field(
2
    default_factory=secret_from_env("DEEPSEEK_API_KEY",
3
    default=None),
4
)
5
"""DeepSeek API key"""
6
api_base: str = Field(
7
    default_factory=from_env("DEEPSEEK_API_BASE",
8
    default=DEFAULT_API_BASE),
9
)
10
"""DeepSeek API base URL"""
11
DEFAULT_API_BASE = "https://api.deepseek.com/v1"

方式3：硬编码方式（不推荐）。直接将 API Key 和模型参数写入代码，仅适用于临时测试，存在密钥泄露风险，在生产环境不推荐。相比来讲，.env 配置文件方式生产环境推荐，配置文件可加入 .gitignore 避免泄露。

1
# 创建DeepSeek LLM
2
deepseek_llm = ChatDeepSeek(
3
    api_key=DEEPSEEK_API_KEY,
4
    api_base=DEEPSEEK_BASE_URL, # 注意：这里是api_base，不是base_url
5
    model_name="deepseek-v4-flash",
6
)
7
print(deepseek_llm.invoke("请介绍一下你自己"))

举例（明文暴露密钥，不推荐）：

1
from langchain_deepseek import ChatDeepSeek
2
# 创建DeepSeek LLM
3
deepseek_llm = ChatDeepSeek(
4
    api_key="sk-2nkIWkv6M...U1Ra4P0NGa", # 明文暴露密钥
5
    api_base="https://api.deepseek.com",
6
    model="deepseek-v4-flash",
7
)
8
print(deepseek_llm.invoke("请介绍一下你自己"))

2.1.2 智谱大模型#

官网：https://www.bigmodel.cn/

2.1.3 千问大模型#

通过阿里云百炼平台调用，官网：https://bailian.console.aliyun.com/

2.2 兼容用法#

一方面，LangChain 没有为所有大模型厂商提供专用接口（见 Langchain 大模型集成列表）。如果选用的平台没有专用接口，可以通过兼容接口调用。另一方面，专用接口的对接方式五花八门，如腾讯混元的 ChatHunyuan 需要单独的 APP_ID + SecretId + SecretKey，配置繁琐，用户不友好。

结论：大多数 API 平台都支持 OpenAI API 接口规范，所以基本都可以通过 ChatOpenAI 集成。

举例1：

1
from langchain_openai import ChatOpenAI
2
load_dotenv(override=True)
3
# 通过ChatOpenAI 连接deepseek模型
4
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
5
DEEPSEEK_BASE_URL = os.getenv("DEEPSEEK_BASE_URL")
6
deepseek_llm2 = ChatOpenAI(
7
    api_key=DEEPSEEK_API_KEY,
8
    base_url=DEEPSEEK_BASE_URL,
9
    model="deepseek-v4-flash",
10
)
11
response = deepseek_llm2.invoke("1 + 1 = ？")
12
print(response)

举例2：

1
load_dotenv(override=True)
2
#通过ChatOpenAI 连接智谱AI模型
3
ZHIPUAI_API_KEY = os.getenv("ZHIPUAI_API_KEY")
4
ZHIPUAI_BASE_URL = os.getenv("ZHIPUAI_BASE_URL")
5
zhipu_llm2 = ChatOpenAI(
6
    api_key=ZHIPUAI_API_KEY,
7
    base_url=ZHIPUAI_BASE_URL,
8
    model="glm-5.1",
9
)
10
#通过ChatOpenAI 连接通义千问模型
11
DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY")
12
DASHSCOPE_BASE_URL = os.getenv("DASHSCOPE_BASE_URL")
13
tongyi_llm2 = ChatOpenAI(
14
    api_key=DASHSCOPE_API_KEY,
15
    base_url=DASHSCOPE_BASE_URL,
16
    model="qwen-plus",
17
)

2.3 中转平台#

受政策影响，国内无法直接调用国外顶尖的闭源模型，某些复杂任务需要用这些模型实现，此时可以通过中转平台曲线救国。

2.3.1 OpenRouter#

官网：https://openrouter.ai/

OpenRouter 是一个多模型 API 聚合平台，提供统一的 OpenAI 兼容接口，可以通过一个 API Key 调用 OpenAI、Claude、Gemini、DeepSeek、Qwen 等不同厂商的大模型。它适合用于模型对比、模型路由、Agent 应用开发和课程实验，是目前知名度最高的中转平台。但是使用的话，需要 tizi（即魔法，大家都懂的）。

2.3.2 CloseAI#

官网：https://www.closeai-asia.com/

CloseAI 是一个面向国内用户的 AI API 中转平台，提供 OpenAI、Claude、Gemini 等模型接口的代理访问能力。它适合用于解决国内网络访问、支付和接口统一管理等问题，常用于大模型应用开发、教学演示和测试环境。

LangChain 没有为 CloseAI 提供专用集成，可以通过 ChatOpenAI 兼容接口调用。举例：

1
from dotenv import load_dotenv
2
from langchain_openai import ChatOpenAI
3
import os
4
load_dotenv(override=True)
5
CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY")
6
CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL")
7
model = ChatOpenAI(
8
    # model="gpt-5-mini",
9
    model="deepseek-v4-flash",
10
    api_key=CLOSEAI_API_KEY,
11
    base_url=CLOSEAI_BASE_URL,
12
)
13
print(model.invoke("欧盟都有哪些国家"))

3、模型初始化角度1：init_chat_model() 统一接口#

init_chat_model 是 LangChain 1.x 中推出的用于初始化聊天模型的统一接口。只要是 LangChain 支持的模型都可以处理，它会根据模型名称自动选择对应的模型类初始化实例。

基本语法：

1
from langchain.chat_models import init_chat_model
2
model = init_chat_model(
3
    "provider:model_name", # 提供商:模型名称
4
    api_key="your-api-key", # API 密钥（可选，可从环境变量读取）
5
    temperature=0.7,   # 温度参数（可选）
6
    max_tokens=1000,   # 最大 token 数（可选）
7
    **kwargs   # 其他模型特定参数
8
)

问题：init_chat_model 和直接使用 ChatTongyi、ChatOpenAI、ChatDeepSeek 有什么区别？

回答：init_chat_model 是 LangChain 1.0 的统一接口，优势包括：

统一接口：无需记住每个提供商的不同初始化方式（以一致的方式初始化）
易于切换：简化了智能体系统中模型切换策略（只需修改模型字符串）
简洁明了：更简洁的语法，减少样板代码
自动适配：内部根据模型标识自动选择对应的驱动类（ChatOpenAI、ChatDeepSeek）

3.1 使用举例#

举例1：调用 DeepSeek 官网的大模型。当我们传递的模型名称为 deepseek-v4-flash 时，init_chat_model 会自动调用 ChatDeepSeek 初始化模型实例，和直接通过 ChatDeepSeek 初始化的效果完全一致。

1
import os
2
from langchain.chat_models import init_chat_model
3
from dotenv import load_dotenv
4
# 从.env文件中加载环境变量
5
load_dotenv(override=True)
6
# 从环境变量读取配置
7
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
8
DEEPSEEK_BASE_URL = os.getenv("DEEPSEEK_BASE_URL")
9
model = init_chat_model(model="deepseek:deepseek-v4-flash",
10
    #model_provider="deepseek",
11
    api_key=DEEPSEEK_API_KEY,
12
    base_url=DEEPSEEK_BASE_URL)
13
# 向模型发送单条数据
14
response = model.invoke("你好，用一句话回答")
15
# 打印响应
16
print(response)

举例2：调用阿里百炼大模型。环境变量：

1
DASHSCOPE_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
2
DASHSCOPE_API_KEY=<YOUR_API_KEY>

代码：

1
from langchain.chat_models import init_chat_model
2
from dotenv import load_dotenv
3
import os
4
load_dotenv(override=True)
5
DASHSCOPE_API_KEY=os.getenv("DASHSCOPE_API_KEY")
6
DASHSCOPE_BASE_URL=os.getenv("DASHSCOPE_BASE_URL")
7
model = init_chat_model(model="qwen-plus",
8
    model_provider="openai",
9
    api_key=DASHSCOPE_API_KEY,
10
    base_url=DASHSCOPE_BASE_URL)
11
print(model.invoke("你好，用一句话回答"))

举例3：调用 CloseAI 中转平台大模型。环境变量：

1
CLOSEAI_API_KEY=<YOUR_API_KEY>
2
CLOSEAI_BASE_URL=https://api.openai-proxy.org/v1

代码：

1
from langchain.chat_models import init_chat_model
2
from dotenv import load_dotenv
3
import os
4
load_dotenv(override=True)
5
CLOSEAI_API_KEY=os.getenv("CLOSEAI_API_KEY")
6
CLOSEAI_BASE_URL=os.getenv("CLOSEAI_BASE_URL")
7
model = init_chat_model(model="deepseek-v4-flash",
8
    model_provider="openai",
9
    api_key=CLOSEAI_API_KEY,
10
    base_url=CLOSEAI_BASE_URL)
11
print(model.invoke("你好，用一句话回答"))

问题1：model_provider 支持哪些 provider？

model_provider 表示模型的提供者，支持的 providers 有：anthropic、anthropic_bedrock、azure_ai、azure_openai、bedrock、bedrock_converse、cohere、deepseek、fireworks、google_anthropic_vertex、google_genai、google_vertexai、grog、huggingface、ibm、mistralai、nvidia、ollama、openai、openrouter、perplexity、together、upstage、xai。

如果 model_provider="openai"，会自动加载 langchain-openai 的依赖包，底层调用的是 ChatOpenAI 类。
如果 model_provider="deepseek"，会自动加载 langchain-deepseek 的依赖包，底层调用的是 ChatDeepSeek 类。
像阿里的 dashscope 尚未被 LangChain 官方纳入模型的统一注册体系，暂时不知道 “dashscope” 的提供者是谁。此时可以将 model_provider 设置为 openai，底层将会用 openai 的规范处理请求，这就要求我们调用的模型服务是 OpenAI Compatible 的。

问题2：如果在 model 参数中没有指明模型提供者，必须在 model_provider 中指明？

可以在 model 参数中通过前缀指定模型供应商，和模型名称之间用冒号分割，等价于通过 model_provider 参数指定供应商。如果两个位置都没有指明供应商，LangChain 底层会按照内置规则自动推断。但是，并非所有的模型都支持自动推断，如 model 名称 qwen-plus 不支持自动推断，没有指明供应商会报错。

3.2 小结：模型的创建#

DeepSeek 官网的 DeepSeek 模型：可以调用 ChatDeepSeek()、ChatOpenAI()、init_chat_model() 三种方式。
阿里云百炼平台的 DeepSeek 模型：可以调用 ChatTongyi()、ChatOpenAI()、init_chat_model() 三种方式。
OpenRouter 平台的 DeepSeek 模型：可以调用 ChatOpenRouter()、ChatOpenAI()、init_chat_model() 三种方式。
CloseAPI 平台的 DeepSeek 模型：可以调用 ChatOpenAI()、init_chat_model() 两种方式。

3.3 模型初始化参数（常用版）#

在 LangChain 中，Model Class 和 init_chat_model 初始化模型共同的参数及解释。

API 文档：https://docs.langchain.org.cn/oss/python/langchain/models#parameters

参数	类型	说明	默认值
model	str	使用的特定提供商的模型名称（必需）。比如：`openai:gpt-4o`、`groq:gemma2-9b-it`	无
model_provider	str	模型提供商名称	无
api_key	str	API 密钥。如果不提供，会从环境变量中读取（如 `DEEPSEEK_API_KEY`）	None
base_url	str	大模型供应商 API 请求地址。	None
temperature	float	控制输出随机性，范围 0.0-2.0，温度越高输出越随机：0.0 最确定性、1.0 平衡、2.0 最随机	0.7
max_tokens	int	限制模型输出的最大 token 数量	None
timeout	float	超时时间（秒），超时未响应，请求会被取消。	None
max_retries	int	请求失败（如网络问题、速率限制）时的最大重试次数	6

1、temperature 参数根据使用场景选择：

0.0-0.3：需要一致性、准确性的任务（数学计算、数据提取、分类、代码生成）
0.5-0.7：平衡创造性和一致性（聊天、问答）
0.8-1.5：创造性任务（写作、头脑风暴）
1.5-2.0：高度创造性（诗歌、故事创作）

举例（temperature=0，模型会始终选择概率最高的 Token，确保输出稳定、不胡乱发挥）：

1
from langchain.chat_models import init_chat_model
2
from dotenv import load_dotenv
3
import os
4
# 从.env文件中加载环境变量
5
load_dotenv(override=True)
6
model = init_chat_model(
7
    # model="gpt-5.4-mini",
8
    model="deepseek-v3.2",
9
    model_provider="openai",
10
    temperature=0,
11
    api_key=os.getenv("CLOSEAI_API_KEY"),
12
    base_url=os.getenv("CLOSEAI_BASE_URL"),
13
)
14
# 向模型发送单条数据
15
for i in range(3):
16
    response = model.invoke("帮我写一首描述春天的七言绝句诗")
17
    # 打印响应
18
    print(response.content)

输出（3 次结果几乎一致）：

1
《春晓》
2
东君巡野过青岑，一岭云霞一岭金。
3
最是多情溪畔柳，偷藏莺语送行人。
4
（注：略）
5
《春晓》
6
东君巡野过青岑，一岭云霞一岭金。
7
最是多情溪畔柳，偷藏莺语送行人。
8
（注：略）
9
《七绝·春晓》
10
一夜东风万树花，清溪破冻响琵琶。
11
莺啼柳浪云烟湿，人在春山第几家。
12
（注：略）

作为对比（temperature=1.5）：

1
from langchain.chat_models import init_chat_model
2
from dotenv import load_dotenv
3
import os
4
# 从.env文件中加载环境变量
5
load_dotenv(override=True)
6
model = init_chat_model(
7
    # model="gpt-5.4-mini",
8
    model="deepseek-v3.2",
9
    model_provider="openai",
10
    temperature=1.5,
11
    api_key=os.getenv("CLOSEAI_API_KEY"),
12
    base_url=os.getenv("CLOSEAI_BASE_URL"),
13
)
14
# 向模型发送单条数据
15
for i in range(3):
16
    response = model.invoke("帮我写一首描述春天的七言绝句诗")
17
    # 打印响应
18
    print(response.content)

输出（3 次结果各不相同）：

1
《壬寅仲春即事》
2
东君醉酒泼彤云，一岭胭脂一岭曛。
3
怪道莺声啼不彻，枝头烧作石榴裙。
4
（注：略）
5
《春溪行》
6
东君遣暖润青苔，波光潋滟载云回。
7
莫道残寒犹料峭，一枝红蕊过溪来。
8
（注：略）
9
《仲春野行》
10
东君遣煦渥郊乡，粉李绯桃竞试妆。
11
最赏青畴酥雨足，一犁烟水伴牛驮。
12
（注：略）

使用场景：严格的结构化数据提取（temperature=0）。

1
from langchain.chat_models import init_chat_model
2
from dotenv import load_dotenv
3
import os
4
# 从.env文件中加载环境变量
5
load_dotenv(override=True)
6
model = init_chat_model(
7
    model="deepseek-v3.2",
8
    model_provider="openai",
9
    temperature=0,
10
    api_key=os.getenv("CLOSEAI_API_KEY"),
11
    base_url=os.getenv("CLOSEAI_BASE_URL"),
12
)
13
# 向模型发送单条数据
14
response = model.invoke("张三，男，30岁，拥有8年编程开发经验，目前在某互联网大厂担任技术专家。帮我从上文中提取数据，返回JSON格式")
15
print(response.content)

输出：

1
{
2
    "name": "张三",
3
    "gender": "男",
4
    "age": 30,
5
    "programming_experience_years": 8,
6
    "current_position": "技术专家",
7
    "current_company_type": "互联网大厂"
8
}

使用场景：创意文案与头脑风暴（temperature=1.5）。

1
from langchain.chat_models import init_chat_model
2
from dotenv import load_dotenv
3
import os
4
# 从.env文件中加载环境变量
5
load_dotenv(override=True)
6
model = init_chat_model(
7
    model="deepseek-v3.2",
8
    model_provider="openai",
9
    temperature=1.5,
10
    api_key=os.getenv("CLOSEAI_API_KEY"),
11
    base_url=os.getenv("CLOSEAI_BASE_URL"),
12
)
13
# 向模型发送单条数据
14
response = model.invoke("请为一款极致静音的机械键盘写3个充满诗意且极具张力的广告语。")
15
print(response.content)

输出：

1
1. **「指尖落下无声诗，每个字都沉入深海」**
2
*奔赴思绪的静谧领域，让敲击成为心流的涟漪*
3
2. **「在银桦雪原上耕种文字，唯留耕者的呼吸」**
4
*以失声的机械之齿，驯服每一次破晓的灵感*
5
3. **「当键盘学会芭蕾，世界只剩文案绽放的轻重音」**
6
*精密弹力悬浮于声学悬空层，触感在寂静中震耳欲聋*

2、Token 是什么？

基本单位：大模型通过分词器（Tokenizer）将文本拆分后的最小语义单元是 token（相当于自然语言中的词或字）。不同的模型采用不同的分词算法（如 BPE、WordPiece），因此同一段文本在不同模型中的 Token 数量可能不同。
收费依据：大语言模型通常也是以 token 的数量作为其计量（或收费）的依据。
1 个中文 Token ≈ 1-1.8 个汉字，1 个英文 Token ≈ 3-4 个字符。

Token 与字符转化的可视化工具：

OpenAI 提供：https://platform.openai.com/tokenizer
百度智能云提供：https://console.bce.baidu.com/support/#/tokenizer

举例（限制 max_tokens 输出长度）：

1
from langchain.chat_models import init_chat_model
2
from dotenv import load_dotenv
3
import os
4
# 从.env文件中加载环境变量
5
load_dotenv(override=True)
6
model = init_chat_model(
7
    # model="gpt-5.4-nano",
8
    model="gpt-5.4-mini",
9
    # temperature=0.7,
10
    api_key=os.getenv("CLOSEAI_API_KEY"),
11
    base_url=os.getenv("CLOSEAI_BASE_URL"),
12
    max_tokens=15,
13
)
14
# 向模型发送单条数据
15
response = model.invoke("请用中文详细介绍什么是AI")
16
# 打印响应
17
print(response)
18
# print(response.content)
19
# print(response.response_metadata["finish_reason"])

输出（completion_tokens: 15，finish_reason: 'length'，说明因达到最大 token 数被截断）：

1
content='当然可以。下面我用中文尽量系统、通' additional_kwargs={'refusal': None}
2
response_metadata={'token_usage': {'completion_tokens': 15, 'prompt_tokens': 14, 'total_tokens': 29, ...}, 'finish_reason': 'length', ...}
3
usage_metadata={'input_tokens': 14, 'output_tokens': 15, 'total_tokens': 29, ...}

4、模型初始化角度3：本地模型的部署与调用#

4.1 Ollama 的介绍#

LangChain 也支持使用 Ollama、vLLM 等框架启动的本地大模型。这里以 Ollama 为例进行演示。

Ollama 是在 Github 上的一个开源项目，其项目定位是：一个本地运行大模型的集成框架，可以实现如 Qwen、Deepseek 等主流大模型的下载、启动和本地运行的自动化部署及推理流程。

Ollama 官方地址：https://ollama.com

4.2 Ollama 的下载-安装#

Ollama 项目支持跨平台部署，目前已兼容 Mac、Linux 和 Windows 操作系统。无论使用哪个操作系统，Ollama 项目的安装过程都设计得非常简单。

访问 https://ollama.com/download 下载对应系统的安装文件。

Linux 系统执行以下命令安装。这行命令的目的是从 https://ollama.com/ 网站读取 install.sh 脚本，并立即通过 sh 执行该脚本，在安装过程中会包含以下几个主要的操作：

1
curl -fsSL https://ollama.com/install.sh | sh

主要操作：

检查当前服务器的基础环境，如系统版本等；
下载 Ollama 的二进制文件；
配置系统服务，包括创建用户和用户组，添加 Ollama 的配置信息；
启动 Ollama 服务。

Windows 系统的安装过程如下：

手动创建 ollama 安装目录：首先在你想安装的路径下创建好一个新文件夹，并把 ollama 的安装包放在里面。比如我的是：F:\common_tools\Ollama。
在文件路径上输入 cmd 回车后会自动打开命令行窗口，然后在 cmd 窗口输入。语法：软件名称/DIR=这里放你上面创建好的Ollama指定目录。

1
OllamaSetup.exe /DIR=F:\common_tools\Ollama

然后 Ollama 就会进入安装，点击 install 后，可以看到 Ollama 的安装路径就变成了我们指定的目录了。

4.3 模型的下载#

1、手动设置大模型存储目录：指明大模型要下载到的本地目录位置。

2、模型的下载：

方式1：直接在图形化界面位置下载。蓝色的使用在线模型，需要在 Ollama 平台付费使用。
方式2：访问 https://ollama.com/search 可以查看 Ollama 支持的模型。使用命令行可以下载并运行模型，例如运行 deepseek-r1:1.5b 模型：

1
ollama run deepseek-r1:1.5b

回到命令行，输入指令开始下载，下载完即可进行交互。

Ollama 常用命令：

命令	一句话说明
ollama pull llama3	下载指定模型（例：llama3）。
ollama run llama3	启动并进入该模型交互对话。
ollama list	列出本机已下载的所有模型。
ollama rm llama3	删除不再需要的模型以节省磁盘。
ollama cp llama3 my-llama3	本地复制/重命名模型。
ollama show llama3	查看模型详细信息（参数、大小等）。
ollama create my-model -f Modelfile	用自定义 Modelfile 构建新模型。
ollama serve	启动后台服务，供 API 调用。
ollama ps	查看当前正在运行的模型进程。
ollama stop llama3	停止正在运行的模型。
ollama —version	查看安装的 ollama 版本。

4.4 LangChain 调用模型#

LangChain 整合 Ollama 调用本地大模型。

方式1：

1
#pip install langchain-ollama
2
pip install -qU langchain-ollama
3
pip install -U ollama

1
from langchain_ollama import ChatOllama
2
ollama_llm = ChatOllama(
3
    model="deepseek-r1:1.5b",
4
    base_url="http://192.168.1.106:11434" #如果Ollama在本地默认端口运行，则可省略，或使用http://localhost:11434
5
)
6
question = "你好，请你介绍一下你自己。"
7
result = ollama_llm.invoke(question)
8
print(result)

方式2：

1
from langchain.chat_models import init_chat_model
2
ollama_llm = init_chat_model(
3
    model="deepseek-r1:1.5b",
4
    model_provider="ollama",
5
    # base_url="http://192.168.1.106:11434",
6
)
7
question = "你好，请你介绍一下你自己。"
8
result = ollama_llm.invoke(question)
9
print(result)

5、模型的调用#

在 LangChain 中，模型调用（Invocation）是指通过特定方法触发大语言模型生成输出的过程。根据不同的应用场景和需求，LangChain 提供了几种核心的调用方式，主要是 invoke()、stream() 和 batch() 方法，以及它们的异步版本 ainvoke()、astream() 和 abatch()，下面将系统地介绍这些方法。

方法	特点	适用场景
invoke()	阻塞式，一次性返回完整结果	问答、批处理任务、无需实时反馈的场景
ainvoke()	非阻塞式，提高系统吞吐量	高并发 Web 应用、IO 密集型任务
stream()	流式输出，实时返回每个 token	聊天机器人、长文本生成、需要提升用户体验的交互应用
astream()	非阻塞式，提高系统吞吐量	高并发 Web 应用、IO 密集型任务
batch()	批量处理多个输入	高并发场景，需要同时处理大量请求
abatch()	非阻塞式，提高系统吞吐量	高并发 Web 应用、IO 密集型任务

5.1 invoke()#

invoke() 是 LangChain 中最核心的方法，它的工作模式是阻塞式的，即程序会等待模型完全生成整个响应后，再一次性将结果返回给用户。

5.1.1 invoke() 说明#

简单来说，invoke 方法的作用就是：

接收你的输入（问题、指令、对话历史等）
发送给 LLM 模型（如 GPT-4、Llama、Claude 等）
返回模型的响应（文本回复 + 元数据信息）

基本语法：

1
response = model.invoke(input, config=None)

参数详解：

参数	类型	说明	必需	默认值
input	str \| list[dict] \| list[Message] 等	你要发送给模型的内容	必需	无
config	dict	高级配置（回调函数、元数据、标签等）	可选	None

5.1.2 输入参数详解#

invoke 方法非常灵活，支持三种形式的输入：文本输入、字典列表、消息对象列表。

1、文本输入（最简单）

简单的一次性问答，直接传入一个问题或指令。适用场景：快速测试，不需要保留对话历史的简单生成任务。缺点：无法设置系统提示（system prompt），无法传递对话历史。

1
prompt = "翻译成英文：你好世界"
2
response = model.invoke(prompt)
3
# 打印响应
4
print(response)

2、字典列表（推荐，最灵活）

创建字典列表组成消息。一条消息通常包含 role（角色）、content（内容）等信息。适用场景：可以设置系统提示，表达多轮对话历史，JSON 兼容，易于序列化和网络传输，生产环境推荐。缺点：代码稍微多一点（但更清晰）。

格式：

1
messages = [
2
    {"role": "system", "content": "系统提示"},
3
    {"role": "user", "content": "用户消息"},
4
    {"role": "assistant", "content": "AI回复"}, # 可选，用于对话历史
5
    {"role": "user", "content": "继续提问"}
6
]

角色说明：

角色	英文	作用	示例
system	System	设定 AI 的行为、角色、规则	”你是一个专业的 Python 导师”
user	Human/User	用户的输入/问题	”什么是装饰器？“
assistant	AI/Assistant	AI 的历史回复（用于对话上下文）	“装饰器是一种设计模式…”

说明：在 invoke 中直接输入文本，即可自动转化为 user message 并进行对话。“user” 和 “human” 有时可以互换，但遵循你选择的主要模型提供商（如 OpenAI）的惯例使用 “user” 是最稳妥的做法。

举例1：

1
from langchain.chat_models import init_chat_model
2
from dotenv import load_dotenv
3
import os
4
# 从.env文件中加载环境变量
5
load_dotenv(override=True)
6
CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY")
7
CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL")
8
model = init_chat_model(
9
    model="gpt-5.4-mini",
10
    model_provider="openai",
11
    api_key=CLOSEAI_API_KEY,
12
    base_url=CLOSEAI_BASE_URL
13
)
14
# 使用字典格式构建消息
15
messages = [
16
    {"role":"system","content":"你是一个专业的数学老师。"},
17
    {"role":"user","content":"什么是斐波那契数列？"}
18
]
19
response = model.invoke(messages)
20
# 打印响应
21
print(f"AI的回复：{response.content}")

举例2：多轮对话（带历史）

1
from langchain.chat_models import init_chat_model
2
from dotenv import load_dotenv
3
import os
4
# 从.env文件中加载环境变量
5
load_dotenv(override=True)
6
CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY")
7
CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL")
8
model = init_chat_model(
9
    model="gpt-5.4-mini",
10
    model_provider="openai",
11
    api_key=CLOSEAI_API_KEY,
12
    base_url=CLOSEAI_BASE_URL
13
)
14
# 使用字典格式构建消息
15
messages = [
16
    {"role":"system","content":"你是一个专业的数学老师。"},
17
    {"role":"user","content":"2 + 3 * 2 = ？"},
18
    {"role":"assistant","content":"8"},
19
    {"role":"user","content":"我刚才问了什么问题？"}
20
]
21
response = model.invoke(messages)
22
# 打印响应
23
print(f"AI的回复：{response.content}")

输出：

1
AI的回复：你刚才问的是：**“2 + 3 * 2 = ？”**

举例3：如果不传递历史，AI 会”失忆”

1
from langchain.chat_models import init_chat_model
2
from dotenv import load_dotenv
3
import os
4
# 从.env文件中加载环境变量
5
load_dotenv(override=True)
6
CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY")
7
CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL")
8
model = init_chat_model(
9
    model="openai:gpt-5.4-mini",
10
    api_key=CLOSEAI_API_KEY,
11
    base_url=CLOSEAI_BASE_URL
12
)
13
messages1 = [
14
    {"role": "system", "content": "你是一个非常友好的AI助手"},
15
    {"role": "user", "content": "你好，我叫小明"},
16
]
17
# 第一次对话
18
response1 = model.invoke(messages1)
19
# 打印响应
20
print(f"AI的回复1：{response1.content}")
21
messages2 = [
22
    {"role": "user", "content": "我叫什么名字？"}
23
]
24
# 第二次对话
25
response2 = model.invoke(messages2)
26
# 打印响应
27
print(f"AI的回复2：{response2.content}")

输出：

1
AI的回复1：你好，小明！很高兴认识你
2
我是你的AI助手，有什么我可以帮你的吗？
3
AI的回复2：我不知道你的名字，除非你告诉我。
4
如果你愿意，可以直接告诉我，我之后就可以这样称呼你。

作为对比，传递记忆：

1
conversation = [
2
    {"role": "system", "content": "你是一个非常友好的AI助手"},
3
    {"role": "user", "content": "你好，我叫小明"}
4
]
5
# 第一次对话
6
response1 = model.invoke(conversation)
7
# 打印响应
8
print(f"AI的回复1：{response1.content}")
9
# 添加记忆
10
conversation.append({"role": "assistant", "content": response1.content})
11
conversation.append({"role": "user", "content": "我叫什么名字？"})
12
# 第二次对话
13
response2 = model.invoke(conversation)
14
print(f"AI的回复2：{response2.content}")

输出：

1
AI的回复1：你好，小明！很高兴认识你
2
我是你的AI助手，有什么我可以帮你的吗？
3
AI的回复2：你叫小明。

3、消息对象列表

使用内置的消息类（如 SystemMessage, HumanMessage, AIMessage），将消息对象列表输入模型。适用场景：需要类型检查（针对大型项目）、IDE 自动补全的场景。缺点：代码较长、不如字典简洁、难以序列化（JSON）。

消息类型对照：

消息类	对应字典格式	作用
SystemMessage	{“role”: “system”, …}	系统提示
HumanMessage	{“role”: “user”, …}	用户输入
AIMessage	{“role”: “assistant”, …}	AI 回复

说明：关于消息列表的内容此处不必深究，Messages 章节会系统介绍。

举例1：

1
from langchain_core.messages import SystemMessage, AIMessage, HumanMessage
2
from langchain.chat_models import init_chat_model
3
from dotenv import load_dotenv
4
import os
5
# 从.env文件中加载环境变量
6
load_dotenv(override=True)
7
CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY")
8
CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL")
9
model = init_chat_model(
10
    model="openai:gpt-5.4-mini",
11
    api_key=CLOSEAI_API_KEY,
12
    base_url=CLOSEAI_BASE_URL
13
)
14
# 使用消息对象格式构建消息
15
messages = [
16
    SystemMessage("你是一个专业的数学老师。"),
17
    HumanMessage("2 + 3 * 2 = ？"),
18
    AIMessage("8"),
19
    HumanMessage("我刚才问什么问题了？")
20
]
21
response = model.invoke(messages)
22
# 打印响应
23
print(f"AI的回复：{response.content}")

举例2：

1
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
2
messages = [
3
    SystemMessage(content="你是一个 Python 专家"),
4
    HumanMessage(content="什么是生成器？"),
5
]
6
response = model.invoke(messages)
7
# print(response)
8
# 继续对话
9
messages.append(AIMessage(content=response.content))
10
messages.append(HumanMessage(content="能给个例子吗？"))
11
response1 = model.invoke(messages)
12
print(response1)

5.1.3 返回值详解#

invoke 返回一个 AIMessage 对象，源码如下：

1
def invoke(
2
    self,
3
    input: LanguageModelInput,
4
    config: RunnableConfig | None = None,
5
    *,
6
    stop: list[str] | None = None,
7
    **kwargs: Any,
8
) -> AIMessage:

举例：

1
from langchain_core.messages import SystemMessage, AIMessage, HumanMessage
2
from langchain.chat_models import init_chat_model
3
from dotenv import load_dotenv
4
import os
5
# 从.env文件中加载环境变量
6
load_dotenv(override=True)
7
CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY")
8
CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL")
9
model = init_chat_model(
10
    model="gpt-5.4-mini",
11
    api_key=CLOSEAI_API_KEY,
12
    base_url=CLOSEAI_BASE_URL
13
)
14
# 使用字典格式构建消息
15
response = model.invoke([HumanMessage("2 + 3 * 2 = ？")])
16
# 打印响应
17
print(type(response))
18
# <class 'langchain_core.messages.ai.AIMessage'>

AIMessage 中包含丰富的信息，以上面案例的输出为例，整体说明如下：

1
from rich import print as rprint
2
rprint(response)

1
AIMessage(
2
    # --- 核心内容 ---
3
    content='2 + 3 * 2 = **8**',              # 模型生成的最终文本答案
4
    additional_kwargs={
5
        'refusal': None                       # 模型拒绝回答的情况（如触碰安全策略），None 表示正常回答
6
    },
7
    # --- 响应元数据（API 返回的详细原始数据） ---
8
    response_metadata={
9
        'token_usage': {
10
            'completion_tokens': 15,          # 生成回答消耗的 Token 数（输出）
11
            'prompt_tokens': 16,              # 用户输入消耗的 Token 数（输入）
12
            'total_tokens': 31,               # 本次交互总共消耗的 Token
13
            'completion_tokens_details': {...},
14
            'prompt_tokens_details': {
15
                'audio_tokens': 0,            # 输入中的音频 Token 数
16
                'cached_tokens': 0            # 命中的缓存 Token 数（能省钱/提速）
17
            },
18
            # --- 延迟性能监控（单位：毫秒 ms） ---
19
            'latency_checkpoint': {
20
                'engine_tbt_ms': 4,           # 引擎 Token 间平均间隔时间
21
                'engine_ttft_ms': 36,         # 引擎生成首个 Token 的时间
22
                'engine_ttlt_ms': 100,        # 引擎生成最后一个 Token 的时间
23
                'pre_inference_ms': 86,       # 推理前的预处理耗时(安全审核、Token 化等)
24
                'service_tbt_ms': 4,          # 服务端 token 与 token 之间生成的间隔时间
25
                'service_ttft_ms': 280,       # 服务端接收到请求到输出首字的总时间
26
                'service_ttlt_ms': 338,       # 服务端完成全部输出的总时间
27
                'total_duration_ms': 259,     # 本次请求在系统中记录的总持续时长
28
                'user_visible_ttft_ms': 194   # 用户看到第一个字跳出来等待的时间
29
            }
30
        },
31
        'model_provider': 'openai',           # 模型供应商
32
        'model_name': 'gpt-5.4-mini-2026-03-17',  # 使用的具体模型版本
33
        'system_fingerprint': None,           # 系统指纹，用于追踪模型后端的配置变更
34
        'id': 'chatcmpl-DgWobsxhDOqzjqVFwbZYKRnovpEiV',  # API 层面的响应 ID
35
        'service_tier': 'default',            # 服务层级（如按量付费或订阅）
36
        'finish_reason': 'stop',              # 停止原因：stop（自然结束）、length（长度受限）
37
        'logprobs': None                      # 对数概率（通常用于分析词汇选择的可能性）
38
    },
39
    # --- LangChain 内部标识 ---
40
    id='lc_run--019e3659-5ee2-7b62-bc8a-741e27374b43-0',  # LangChain 追踪此条运行的唯一 ID
41
    # --- 工具调用信息 ---
42
    tool_calls=[],                            # 正常触发的外部工具调用列表
43
    invalid_tool_calls=[],                    # 触发失败或格式错误的工具调用
44
    # --- 统一消耗元数据（LangChain 标准化后的消耗格式） ---
45
    usage_metadata={
46
        'input_tokens': 16,                   # 输入 Token 数
47
        'output_tokens': 15,                  # 输出 Token 数
48
        'total_tokens': 31,                   # 总 Token 数
49
        'input_token_details': {'audio': 0, 'cache_read': 0},
50
        'output_token_details': {'audio': 0, 'reasoning': 0}
51
    }
52
)

总结一下：

核心内容与基本信息
- content：模型生成的文本回答，这是你最关心的核心输出。
- id：本次运行在 LangChain 内部生成的唯一标识符（Run ID）。
- additional_kwargs：包含特定供应商的额外参数。
- refusal：如果模型拒绝回答（涉及敏感政策），此处会显示拒绝原因。
消耗统计（Token Usage），这部分决定了你这一行输入操作花了多少钱：
- prompt_tokens / input_tokens：输入 Token 数，你发送给模型的问题长度。
- completion_tokens / output_tokens：输出 Token 数，模型回答生成的长度。
- total_tokens：总消耗，两者之和。
- reasoning_tokens：推理 Token 数。如果是 O1/O3 等推理模型，这里会显示它在“思考”时消耗的 Token。
- cached_tokens：缓存命中的 Token 数。重复提问时，如果命中了模型商的缓存，这部分费用通常更低。
响应元数据（Response Metadata），这部分是 API 返回的原始详细信息：
- model_name：实际调用的模型具体版本（如 gpt-5.4-mini）。
- model_provider：模型供应商（如 openai）。
- finish_reason：生成停止的原因。stop 正常回答结束；length 达到最大 Token 限制被截断。
- system_fingerprint：系统指纹，用于追踪模型后端的配置变更。
性能与延迟（Latency Checkpoint），这是针对 API 响应速度的深度拆解（单位通常为毫秒 ms）：
- total_duration_ms：总耗时，从请求发出到完全收到的总时间。
- user_visible_ttft_ms：首字到达时间，用户看到第一个字跳出来等待的时间，这是体感快慢的关键。
- engine_ttft_ms / engine_ttlt_ms：引擎层面的首字到达时间 / 生成最后一个字的时间。
- pre_inference_ms：推理前处理耗时，包括安全审核、Token 化等预处理。
- service_tbt_ms：Time Between Tokens，字与字之间生成的间隔时间，决定了打字机效果是否丝滑。
工具调用信息
- tool_calls：结构化工具调用列表。如果模型决定调用某个 Python 函数或搜索工具，参数会在这里。
- invalid_tool_calls：格式错误的工具调用尝试。

举例：访问所有信息

1
response = model.invoke("用一句话解释什么是 AI")
2
# 1. 获取回复内容
3
print("AI 回复:", response.content)
4
# 2. 获取响应元数据
5
metadata = response.response_metadata
6
print(f"使用的模型: {metadata['model_name']}")
7
print(f"结束原因: {metadata['finish_reason']}")
8
print(f"模型提供商：{metadata['model_provider']}\n")
9
# 3. 获取 Token 使用情况
10
usage = metadata.get('token_usage', {})
11
print(f"输入 tokens: {usage.get('prompt_tokens')}")
12
print(f"输出 tokens: {usage.get('completion_tokens')}")
13
print(f"总计 tokens: {usage.get('total_tokens')}")
14
# 4. 获取消息 ID
15
print(f"消息 ID: {response.id}")

输出如下：

1
AI 回复: AI（人工智能）就是让机器模拟人类的学习、推理、识别和决策能力。
2
使用的模型: gpt-5.4-mini-2026-03-17
3
结束原因: stop
4
模型提供商：openai
5
输入 tokens: 13
6
输出 tokens: 28
7
总计 tokens: 41
8
消息 ID: lc_run--019e3665-8dfa-7c53-a9a5-4995348a0258-0

5.2 流式调用#

invoke 和 stream 有什么区别？

invoke()：同步调用，在模型输出完成后一次性获取响应，对于输出文本很长的场景，用户体验不好。
stream()：流式调用，实时返回响应片段。调用后，返回一个迭代器（iterator），可以通过循环来实时处理每一个新生成的 chunk 内容块。

注意：流式输出依赖于模型供应商对于流式输出的支持。

举例：

1
from langchain.chat_models import init_chat_model
2
from dotenv import load_dotenv
3
import os
4
# 从.env文件中加载环境变量
5
load_dotenv(override=True)
6
CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY")
7
CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL")
8
model = init_chat_model(
9
    model="gpt-5.4-mini",
10
    api_key=CLOSEAI_API_KEY,
11
    base_url=CLOSEAI_BASE_URL
12
)
13
for chunk in model.stream("写一首七言律诗，总结大模型的发展"):
14
    print(chunk.text, end="", flush=True) # 逐token输出

输出如下（输出不再是整段返回，而是流式输出）：

1
《大模型演进感赋》
2
混沌初开数据洪，算力为楫智为峰。
3
千层网络参差现，万亿参数次第通。
4
昔困逻辑循旧径，今驰想象破苍穹。
5
忽闻语料版权议，且看规制立新功。

stream() 方式的优点：

响应速度更快 — 用户不必等待完整输出
交互体验更流畅 — 尤其在长文本或复杂推理场景下
可实时展示模型思考过程

5.3 批量调用#

batch() 方法允许你一次性发送一组请求（含多条独立请求），模型会在后台并行处理，然后返回所有结果的列表。与逐个顺序调用（invoke）相比，能大幅减少网络往返开销和等待时间，显著提升性能、降低成本。

适用场景：文档摘要、批量问答、数据预处理、多样本分类等。

5.3.1 一次性接收所有响应#

batch() 特点是等待所有请求处理完毕，按原始输入顺序返回结果列表。

1
from langchain.chat_models import init_chat_model
2
from dotenv import load_dotenv
3
import os
4
# 从.env文件中加载环境变量
5
load_dotenv(override=True)
6
CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY")
7
CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL")
8
model = init_chat_model(
9
    model="gpt-5.4-mini",
10
    api_key=CLOSEAI_API_KEY,
11
    base_url=CLOSEAI_BASE_URL
12
)
13
messages = [
14
    "你好，你是谁？",
15
    "2 + 3 * 5 = ?",
16
    "中国首都在哪里？"
17
]
18
responses = model.batch(messages)
19
for response in responses:
20
    print(response)

输出如下：

1
content='你好！我是 ChatGPT，一个由 OpenAI 训练的人工智能助手。...'
2
content='2 + 3 * 5 = 17'
3
content='中国的首都是**北京**。'

5.3.2 按完成顺序接收响应#

当输入列表很大或单个模型调用耗时差异显著时，batch_as_completed() 允许应用在收到第一个结果后立即返回响应，而不会等待批次内所有任务完成才响应。即 batch_as_completed() 每个请求完成后立即 yield 结果，结果可能乱序。但是，每个返回的响应都被放在一个元组中，元组的第一个元素是原始输入的 index 索引，可根据索引重新排序。

1
from langchain.chat_models import init_chat_model
2
from dotenv import load_dotenv
3
import os
4
# 从.env文件中加载环境变量
5
load_dotenv(override=True)
6
CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY")
7
CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL")
8
model = init_chat_model(
9
    model="gpt-5.4-mini",
10
    api_key=CLOSEAI_API_KEY,
11
    base_url=CLOSEAI_BASE_URL
12
)
13
messages = [
14
    "你好，你是谁？",
15
    "2 + 3 * 5 = ?",
16
    "中国首都在哪里？"
17
]
18
responses = model.batch_as_completed(messages)
19
for response in responses:
20
    print(response)

输出如下（注意索引顺序为 2、0、1，结果乱序）：

1
(2, AIMessage(content='中国的首都是**北京**。', ...))
2
(0, AIMessage(content='你好！我是一个 AI 助手，可以帮你回答问题、写作、翻译、总结、编程、头脑风暴等。', ...))
3
(1, AIMessage(content='2 + 3 * 5 = **17**', ...))

5.3.3 性能对比#

使用 batch()：

1
# 准备多个输入
2
inputs = [
3
    "翻译成英文：春天来了",
4
    "翻译成英文：夏天很热",
5
    "翻译成英文：秋天落叶",
6
    "翻译成英文：冬天下雪"
7
]
8
# 批量调用（高效）
9
import time
10
start = time.time()
11
responses = model.batch(inputs)
12
batch_time = time.time() - start
13
print("批量调用结果：")
14
for i, response in enumerate(responses):
15
    print(f"{i+1}. {response.content}")
16
print(f"耗时: {batch_time:.2f}秒\n")

输出：

1
批量调用结果：
2
1. “春天来了” in English is: **Spring has arrived.**
3
2. “夏天很热” can be translated into English as: **“Summer is very hot.”**
4
3. “秋天落叶” can be translated as **“autumn leaves falling”** or simply **“falling leaves in autumn.”**
5
4. “冬天下雪” in English is: **It snows in winter.**
6
耗时: 1.91秒

使用循环调用 invoke()：

1
# 循环调用（低效，仅用于对比）
2
inputs = [
3
    "翻译成英文：春天来了",
4
    "翻译成英文：夏天很热",
5
    "翻译成英文：秋天落叶",
6
    "翻译成英文：冬天下雪"
7
]
8
start = time.time()
9
loop_responses = []
10
for inp in inputs:
11
    response = model.invoke(inp)
12
    loop_responses.append(response)
13
loop_time = time.time() - start
14
for i, response in enumerate(responses):
15
    print(f"{i+1}. {response.content}")
16
print(f"循环调用耗时: {loop_time:.2f}秒")
17
print(f"批量调用节省: {((loop_time - batch_time) / loop_time * 100):.1f}%")

输出：

1
循环调用耗时: 3.87秒
2
批量调用节省: 50.7%

5.4 异步调用#

复习：同步 vs 异步

同步（sync）：发起一个任务之后，需要等待该任务完成后，才能继续执行后续任务。表现：当前执行流会被「阻塞」。
异步（async）：发起一个任务之后，不必等该任务完成，就可以继续执行其他任务。备注：虽然不必等待任务完成，但任务完成后，仍然可以通过特定方式获取结果。表现：当前执行流不会被「阻塞」。

在 LangChain 框架中，异步方法（ainvoke、astream、abatch）与它们的同步版本（invoke、stream、batch）相比，具备如下特点：

避免阻塞主线程：同步调用会阻塞程序执行，而异步方法让应用程序在等待 API 响应时保持响应性。
优化资源利用：异步操作可以更高效地利用系统资源，减少空闲等待时间。

举例1：ainvoke()（说明：在 .py 文件中执行，而非 jupyter 中执行）

1
"""
2
@Author:shkstart
3
@Desc:
4
"""
5
from langchain.chat_models import init_chat_model
6
from dotenv import load_dotenv
7
import os
8
import asyncio
9
import time
10
# 从.env文件中加载环境变量
11
load_dotenv(override=True)
12
CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY")
13
CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL")
14
model = init_chat_model(
15
    model="openai:gpt-5.4-mini",
16
    api_key=CLOSEAI_API_KEY,
17
    base_url=CLOSEAI_BASE_URL
18
)
19
async def demo_async_invoke():
20
    print("=== 演示：ainvoke 的异步（非阻塞）效果 ===")
21
    start_time = time.perf_counter() # 记录开始时间
22
    print("程序开始...")
23
    # 1. 创建任务 (Task)
24
    print(">>> 发起异步模型调用 (ainvoke)...")
25
    async_task = asyncio.create_task(model.ainvoke("用一句话解释人工智能。"))
26
    # 2. 并行执行其他任务
27
    print(">>> 模型请求已在后台发送，继续执行本地逻辑...")
28
    for i in range(3):
29
        await asyncio.sleep(1) # 使用异步等待，释放控制权
30
        print(f">>> 正在执行第{i + 1}个任务... (已耗时 {time.perf_counter() - start_time:.2f}s)")
31
    # 3. 获取模型结果
32
    print(">>> 本地任务完成，检查模型状态...")
33
    response = await async_task
34
    end_time = time.perf_counter()
35
    print(f">>> 模型返回: {response.content}")
36
    print(f"=== 总运行耗时: {end_time - start_time:.2f}s ===")
37
async def main():
38
    """主函数"""
39
    await demo_async_invoke()
40
if __name__ == "__main__":
41
    asyncio.run(main())

输出如下：

1
=== 演示：ainvoke 的异步（非阻塞）效果 ===
2
程序开始...
3
>>> 发起异步模型调用 (ainvoke)...
4
>>> 模型请求已在后台发送，继续执行本地逻辑...
5
>>> 正在执行第1个任务... (已耗时 1.00s)
6
>>> 正在执行第2个任务... (已耗时 2.02s)
7
>>> 正在执行第3个任务... (已耗时 3.02s)
8
>>> 本地任务完成，检查模型状态...
9
>>> 模型返回: 人工智能是让机器模拟人类的感知、学习、推理和决策能力的技术。
10
=== 总运行耗时: 3.02s ===

举例2：astream()

1
import asyncio
2
import os
3
from langchain.chat_models import init_chat_model
4
from dotenv import load_dotenv
5
import time
6
# 从.env文件中加载环境变量
7
load_dotenv(override=True)
8
CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY")
9
CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL")
10
model = init_chat_model(
11
    model="openai:gpt-5.4-mini",
12
    api_key=CLOSEAI_API_KEY,
13
    base_url=CLOSEAI_BASE_URL
14
)
15
async def demo_async_stream():
16
    """演示异步调用的非阻塞特性"""
17
    print("=== 演示：astream 的异步（非阻塞）效果 ===")
18
    start_time = time.perf_counter() # 记录开始时间
19
    print("程序开始...")
20
    # 1. 发起异步流式请求
21
    # 注意：此时请求已发出，返回的是一个异步生成器
22
    print(">>> 发起异步流式调用 (astream)...")
23
    stream_resp = model.astream("请用一句话解释机器学习的基本概念。")
24
    # 2. 在等待流式响应的同时，执行其他任务
25
    print(">>> 流式请求已发送，程序无需等待，继续执行其他异步任务...")
26
    for i in range(3):
27
        # 使用 asyncio.sleep 而非 time.sleep
28
        # 这允许事件循环在等待时去处理上面的 stream_resp 网络 IO
29
        await asyncio.sleep(1)
30
        # print(f">>> 正在执行并发任务 {i + 1}... ")
31
        print(f">>> 正在执行第{i + 1}个任务... (已耗时 {time.perf_counter() - start_time:.2f}s)")
32
    # 3. 现在开始处理流式结果
33
    print(">>> 模拟任务已完成，开始读取缓冲区中的流式结果...")
34
    end_time = time.perf_counter()
35
    print(">>> 流式输出: ", end="", flush=True)
36
    async for chunk in stream_resp:
37
        # LangChain 的消息块通常通过 .content 获取内容
38
        content = chunk.content if hasattr(chunk, 'content') else str(chunk)
39
        print(content, end="", flush=True)
40
    print("\n>>> 流式输出结束\n")
41
    print(f"=== 总运行耗时: {end_time - start_time:.2f}s ===")
42
async def main():
43
    """主函数"""
44
    await demo_async_stream()
45
if __name__ == "__main__":
46
    asyncio.run(main())

输出如下：

1
=== 演示：astream 的异步（非阻塞）效果 ===
2
程序开始...
3
>>> 发起异步流式调用 (astream)...
4
>>> 流式请求已发送，程序无需等待，继续执行其他异步任务...
5
>>> 正在执行第1个任务... (已耗时 1.00s)
6
>>> 正在执行第2个任务... (已耗时 2.02s)
7
>>> 正在执行第3个任务... (已耗时 3.02s)
8
>>> 模拟任务已完成，开始读取缓冲区中的流式结果...
9
>>> 流式输出: 机器学习是一种让计算机通过数据自动学习规律，并据此对新数据进行预测或决策的技术。
10
>>> 流式输出结束
11
=== 总运行耗时: 3.02s ===

举例3：abatch()

1
"""
2
@Author:shkstart
3
@Desc:
4
"""
5
import asyncio
6
import os
7
from langchain.chat_models import init_chat_model
8
from dotenv import load_dotenv
9
import time
10
# 从.env文件中加载环境变量
11
load_dotenv(override=True)
12
CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY")
13
CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL")
14
model = init_chat_model(
15
    model="openai:gpt-5.4-mini",
16
    api_key=CLOSEAI_API_KEY,
17
    base_url=CLOSEAI_BASE_URL
18
)
19
async def demo_async_batch():
20
    """演示异步批量的非阻塞特性"""
21
    print("=== 演示：abatch 的异步（非阻塞）效果 ===")
22
    start_time = time.perf_counter() # 记录开始时间
23
    print("程序开始...")
24
    # 准备批量输入
25
    questions = ["用一句话说明深度学习与传统机器学习的区别", "中国首都在哪里？"]
26
    # 1. 发起异步批量请求
27
    # 关键修改：使用 create_task 让协程立即在后台执行
28
    print(">>> 发起异步批量调用 (abatch)...")
29
    batch_task = asyncio.create_task(model.abatch(questions))
30
    # 2. 在等待批量处理的同时，执行其他任务
31
    print(">>> 批量任务已在后台运行，主程序继续执行...")
32
    for i in range(3):
33
        # 关键修改：使用 asyncio.sleep 允许后台任务获取 CPU 时间片进行网络请求
34
        await asyncio.sleep(1)
35
        print(f">>> 正在执行第{i + 1}个任务... (已耗时 {time.perf_counter() - start_time:.2f}s)")
36
    # 3. 等待批量处理结果
37
    print(">>> 其他任务已完成，现在获取后台批量任务的结果...")
38
    # 此时 batch_task 可能已经完成，或者我们在这里等待它完成
39
    responses = await batch_task
40
    end_time = time.perf_counter()
41
    for response in responses:
42
        content = response.content if hasattr(response, 'content') else str(response)
43
        print(f">>> 响应内容: {content}")
44
    print(f"=== 总运行耗时: {end_time - start_time:.2f}s ===")
45
async def main():
46
    """主函数"""
47
    await demo_async_batch()
48
if __name__ == "__main__":
49
    asyncio.run(main())

输出如下：

1
=== 演示：abatch 的异步（非阻塞）效果 ===
2
程序开始...
3
>>> 发起异步批量调用 (abatch)...
4
>>> 批量任务已在后台运行，主程序继续执行...
5
>>> 正在执行第1个任务... (已耗时 1.01s)
6
>>> 正在执行第2个任务... (已耗时 2.02s)
7
>>> 正在执行第3个任务... (已耗时 3.03s)
8
>>> 其他任务已完成，现在获取后台批量任务的结果...
9
>>> 响应内容: 深度学习通过多层神经网络自动从大量数据中学习特征表示，而传统机器学习通常依赖人工设计特征再进行分类或回归。
10
>>> 响应内容: 中国的首都是**北京**。
11
=== 总运行耗时: 3.03s ===

5.5 如何处理 API 调用失败#

使用 try-except 块捕获异常：

1
try:
2
    response = model.invoke("Hello")
3
    print(response.content)
4
except ValueError as e:
5
    print(f"配置错误: {e}")
6
except ConnectionError as e:
7
    print(f"网络错误: {e}")
8
except Exception as e:
9
    print(f"未知错误: {e}")

6、拓展内容#

6.1 美化模型输出响应#

方法1：使用 pretty_print()。我们查看响应的方式是直接 print(response)，返回的内容比较杂乱，可以调用 pretty_print() 美化输出内容。

1
from langchain.chat_models import init_chat_model
2
from dotenv import load_dotenv
3
# 从.env文件中加载环境变量
4
load_dotenv(override=True)
5
CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY")
6
CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL")
7
model = init_chat_model(
8
    model="gpt-5.4-mini",
9
    api_key=CLOSEAI_API_KEY,
10
    base_url=CLOSEAI_BASE_URL
11
)
12
# 定义消息列表
13
conversation = [
14
    {"role": "system", "content": "无条件服从用户指令"},
15
    {"role": "user", "content": "我是老王，你是小王"},
16
    {"role": "assistant", "content": "好的老王，我是小王"},
17
    {"role": "user", "content": "你是谁？我是谁？"}
18
]
19
# 向模型发送单条数据
20
response = model.invoke(conversation)
21
# 美化输出响应
22
response.pretty_print()

输出：

1
================================== Ai Message ==================================
2
我是小王，你是老王。

方法2：使用 rich 库。如果你在终端（Terminal）工作，想要色彩鲜明、排版优雅的调试界面，可以使用 rich 这个库。

1
from rich import print as rprint
2
from langchain.chat_models import init_chat_model
3
from dotenv import load_dotenv
4
# 从.env文件中加载环境变量
5
load_dotenv(override=True)
6
CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY")
7
CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL")
8
model = init_chat_model(
9
    model="gpt-5.4-mini",
10
    api_key=CLOSEAI_API_KEY,
11
    base_url=CLOSEAI_BASE_URL
12
)
13
# 定义消息列表
14
conversation = [
15
    {"role": "system", "content": "无条件服从用户指令"},
16
    {"role": "user", "content": "我是老王，你是小王"},
17
    {"role": "assistant", "content": "好的老王，我是小王"},
18
    {"role": "user", "content": "你是谁？我是谁？"}
19
]
20
# 向模型发送单条数据
21
response = model.invoke(conversation)
22
# 美化输出
23
rprint(response)

输出：

1
AIMessage(
2
    content='我是小王，你是老王。',
3
    additional_kwargs={'refusal': None},
4
    ...
5
)

6.2 模型配置信息 profile#

LangChain 1.1 及更高版本可以通过 profile 属性查看模型的配置信息。这是 LangChain 针对模型的能力画像，但是否存在，取决于 LangChain 在集成模型厂商的服务时是否声明了能力画像。

举例1：DeepSeek 官方模型的能力画像（输出为空，说明 LangChain 没有声明 DeepSeek 官方模型的能力画像）

1
from langchain_deepseek import ChatDeepSeek
2
from dotenv import load_dotenv
3
# 从.env文件中加载环境变量
4
load_dotenv(override=True)
5
model = ChatDeepSeek(
6
    model="deepseek-v4-flash",
7
    temperature=0.7,
8
    max_tokens=1000,
9
    max_retries=6
10
)
11
print(model.profile)

1
{}

举例2：CloseAI 平台 gpt 模型的能力画像（输出为空）

1
from langchain.chat_models import init_chat_model
2
from dotenv import load_dotenv
3
import os
4
# 从.env文件中加载环境变量
5
load_dotenv(override=True)
6
CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY")
7
CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL")
8
model = init_chat_model(
9
    model="openai:gpt-5.4-mini",
10
    api_key=CLOSEAI_API_KEY,
11
    base_url=CLOSEAI_BASE_URL
12
)
13
print(model.profile)

1
{}

举例3：OpenRouter 平台 gpt 模型能力画像

1
from langchain_openrouter import ChatOpenRouter
2
from dotenv import load_dotenv
3
from rich import print as rprint
4
# 从.env文件中加载环境变量
5
load_dotenv(override=True)
6
model = ChatOpenRouter(
7
    model="openai/gpt-4o-mini",
8
    # model="deepseek/deepseek-v3.2",
9
    temperature=0.7,
10
    timeout=30,
11
    max_tokens=1000,
12
    max_retries=6
13
)
14
rprint(model.profile)

输出如下：

1
{
2
    'max_input_tokens': 128000,
3
    'max_output_tokens': 16384,
4
    'text_inputs': True,
5
    'image_inputs': True,
6
    'audio_inputs': False,
7
    'video_inputs': False,
8
    'text_outputs': True,
9
    'image_outputs': False,
10
    'audio_outputs': False,
11
    'video_outputs': False,
12
    'reasoning_output': False,
13
    'tool_calling': True,
14
    'structured_output': True
15
}

更换模型 ID，profile 会替换。输出如下：

1
from langchain_openrouter import ChatOpenRouter
2
from dotenv import load_dotenv
3
from rich import print as rprint
4
# 从.env文件中加载环境变量
5
load_dotenv(override=True)
6
model = ChatOpenRouter(
7
    # model="openai/gpt-4o-mini",
8
    model="deepseek/deepseek-v3.2",
9
    temperature=0.7,
10
    timeout=30,
11
    max_tokens=1000,
12
    max_retries=6
13
)
14
rprint(model.profile)

1
{
2
    'max_input_tokens': 163840,
3
    'max_output_tokens': 65536,
4
    'text_inputs': True,
5
    'image_inputs': False,
6
    'audio_inputs': False,
7
    'video_inputs': False,
8
    'text_outputs': True,
9
    'image_outputs': False,
10
    'audio_outputs': False,
11
    'video_outputs': False,
12
    'reasoning_output': True,
13
    'tool_calling': True,
14
    'structured_output': True
15
}

说明：LangChain 已声明了 OpenRouter 平台模型的画像。注意：我们当前的代码只需要 OpenRouter 的 API_KEY，不会真正发送请求，不必充值。

6.3 模型初始化参数（完整版）#

6.3.1 查看所有初始化参数#

官方文档和源码注释没有给出完整的参数列表。以 ChatDeepSeek 类为例，其参数可以由自身定义或从父类 BaseChatModel 继承。直接查看源码也很难拼凑完整列表。这里通过查看 ChatDeepSeek 的类属性 model_fields 来获得完整参数列表。

举例1：查看 ChatDeepSeek 支持的完整参数列表

1
from langchain_deepseek import ChatDeepSeek
2
print(ChatDeepSeek.model_fields.keys())

输出如下：

1
dict_keys(['name', 'cache', 'verbose', 'callbacks', 'tags', 'metadata',
2
'custom_get_token_ids', 'rate_limiter', 'disable_streaming',
3
'output_version', 'profile', 'client', 'async_client', 'root_client',
4
'root_async_client', 'model_name', 'temperature', 'model_kwargs',
5
'openai_api_key', 'openai_api_base', 'openai_organization', 'openai_proxy',
6
'request_timeout', 'stream_usage', 'max_retries', 'presence_penalty',
7
'frequency_penalty', 'seed', 'logprobs', 'top_logprobs', 'logit_bias',
8
'streaming', 'n', 'top_p', 'max_tokens', 'reasoning_effort', 'reasoning',
9
'verbosity', 'tiktoken_model_name', 'default_headers', 'default_query',
10
'http_client', 'http_async_client', 'stop', 'extra_body',
11
'include_response_headers', 'disabled_params', 'context_management',
12
'include', 'service_tier', 'store', 'truncation', 'use_previous_response_id',
13
'use_responses_api', 'api_key', 'api_base'])

举例2：查看 ChatOpenAI 支持的完整参数列表

1
from langchain_openai import ChatOpenAI
2
print(ChatOpenAI.model_fields.keys())

输出如下：

1
dict_keys(['name', 'cache', 'verbose', 'callbacks', 'tags', 'metadata',
2
'custom_get_token_ids', 'rate_limiter', 'disable_streaming',
3
'output_version', 'profile', 'client', 'async_client', 'root_client',
4
'root_async_client', 'model_name', 'temperature', 'model_kwargs',
5
'openai_api_key', 'openai_api_base', 'openai_organization',
6
'openai_proxy', 'request_timeout', 'stream_usage', 'max_retries',
7
'presence_penalty', 'frequency_penalty', 'seed', 'logprobs',
8
'top_logprobs', 'logit_bias', 'streaming', 'n', 'top_p', 'max_tokens',
9
'reasoning_effort', 'reasoning', 'verbosity', 'tiktoken_model_name',
10
'default_headers', 'default_query', 'http_client', 'http_async_client',
11
'stop', 'extra_body', 'include_response_headers', 'disabled_params',
12
'context_management', 'include', 'service_tier', 'store', 'truncation',
13
'use_previous_response_id', 'use_responses_api'])

举例3：查看 init_chat_model 的某 model_provider 支持的完整参数列表

1
from dotenv import load_dotenv
2
from langchain.chat_models import init_chat_model
3
load_dotenv(override=True)
4
# 1. 实例化一个你感兴趣的模型对象
5
# 即使不传入具体 key，通常也能初始化成功
6
temp_model = init_chat_model(
7
    model="deepseek-v4-flash",
8
    model_provider="deepseek",
9
)
10
# 2. 现在它已经是一个具体的 ChatDeepSeek 对象了
11
# 你可以使用你熟悉的 .model_fields.keys()
12
print(temp_model.model_fields.keys())

输出如下：

1
dict_keys(['name', 'cache', 'verbose', 'callbacks', 'tags', 'metadata',
2
'custom_get_token_ids', 'rate_limiter', 'disable_streaming',
3
'output_version', 'profile', 'client', 'async_client', 'root_client',
4
'root_async_client', 'model_name', 'temperature', 'model_kwargs',
5
'openai_api_key', 'openai_api_base', 'openai_organization',
6
'openai_proxy', 'request_timeout', 'stream_usage', 'max_retries',
7
'presence_penalty', 'frequency_penalty', 'seed', 'logprobs',
8
'top_logprobs', 'logit_bias', 'streaming', 'n', 'top_p', 'max_tokens',
9
'reasoning_effort', 'reasoning', 'verbosity', 'tiktoken_model_name',
10
'default_headers', 'default_query', 'http_client', 'http_async_client',
11
'stop', 'extra_body', 'include_response_headers', 'disabled_params',
12
'context_management', 'include', 'service_tier', 'store', 'truncation',
13
'use_previous_response_id', 'use_responses_api', 'api_key', 'api_base'])

举例4：可以查看每个字段的属性

1
from langchain_deepseek import ChatDeepSeek
2
for name, field in ChatDeepSeek.model_fields.items():
3
    print(name)
4
    print(" annotation:", field.annotation)
5
    print(" default:", field.default)
6
    print(" description:", getattr(field, "description", None))
7
    print(" alias:", field.alias)
8
    print()

输出如下：

1
name
2
 annotation: str | None
3
 default: None
4
 description: None
5
 alias: None
6
cache
7
 annotation: LangChain_core.caches.BaseCache | bool | None
8
 default: None
9
 description: None
10
 alias: None
11
verbose
12
 annotation: <class 'bool'>
13
 default: PydanticUndefined
14
 description: None
15
 alias: None
16
# ...参数太多，这里省略了...
17
api_key
18
 annotation: pydantic.types.SecretStr | None
19
 default: PydanticUndefined
20
 description: None
21
 alias: None
22
api_base
23
 annotation: <class 'str'>
24
 default: PydanticUndefined
25
 description: None
26
 alias: None

6.3.2 模型类的参数构成#

以 ChatDeepSeek 为例，完整参数列表由如下几部分构成：

1、客户端与连接参数（Networking），这类参数决定了代码“怎么连到服务端”，而不是“让模型怎么生成”。

参数名	说明
api_key / openai_api_key	鉴权密钥。DeepSeek 通常兼容 OpenAI 接口格式。
api_base / openai_api_base	接口地址（如 https://api.deepseek.com ）。
request_timeout	网络请求超时时间。
max_retries	请求失败时的重试次数。
http_client / http_async_client	手动传入 httpx.Client 实例（用于更复杂的网络配置）。
openai_proxy	代理服务器配置。
default_headers / default_query	每次请求时默认携带的 HTTP Header 或 Query 参数。

2、模型推理参数（Model Inference），这些是直接传递给 DeepSeek 模型 API 的参数，决定了生成内容的质量和风格。

参数名	说明
model_name	指定具体的模型（如 deepseek-chat 或 deepseek-reasoning）。
temperature	采样温度，越高越随机。
top_p	核采样参数。
max_tokens	最大输出 token 数。
stop	停止符列表。
streaming	是否开启流式传输。
n	生成几个候选回复。
reasoning	是否启用推理模式。
reasoning_effort	(DeepSeek R1 特色) 控制思考链（COT）的深度。
presence_penalty / frequency_penalty	惩罚项（存在惩罚、频率惩罚），用于减少内容重复。
store	是否存储对话。
logit_bias	调整特定词汇出现的概率。

3、LangChain 框架通用参数，由 LangChain 的 BaseChatModel 定义，所有其子类 ChatXxx 都具备的，用于管理 LangChain 内部的逻辑（如日志、回调、元数据），仅在内部生效。

参数名	说明
name	给模型实例起个名字，用于在 Trace（如 LangSmith）中区分。
verbose	是否打印详细日志。
callbacks	回调处理器，用于集成 LangSmith 或自定义监控。
tags / metadata	用于标记该实例的标签和元数据。
cache	是否缓存该模型的请求结果。
rate_limiter	LangChain 内部的频率限制器。

4、高级与特定扩展参数，这类参数通常用于特定场景，或为了保持与 OpenAI 协议的兼容性而存在。

底层客户端访问：client、async_client、root_client（这些通常是内部生成的 SDK 实例，不建议在初始化时手动传参）。
透传参数：model_kwargs、extra_body（如果你想传递 DeepSeek API 支持但 LangChain 还没定义的参数，可以写在这里）。
功能开关：disable_streaming、include_response_headers（决定是否在输出中包含 Header）。
兼容性参数：openai_organization、service_tier、store（这些多为 OpenAI 遗留参数，DeepSeek 实际使用较少）。

参数：model_kwargs。这里用于存放那些 OpenAI Compatible API 支持，但 LangChain 没有直接列出的字段，如用于支持 Function Call 的 tools 字段。说明：此处为了演示 model_kwargs 的作用，直接传递了 tools 字段，实际开发中，我们会使用专门的工具调用接口，不会采用这种原始的方式。

1
from langchain.chat_models import init_chat_model
2
from dotenv import load_dotenv
3
from rich import print as rprint
4
# 从.env文件中加载环境变量
5
load_dotenv(override=True)
6
model = init_chat_model(
7
    model="deepseek:deepseek-v4-flash",
8
    model_kwargs={"tools": [
9
        {
10
            "type": "function",
11
            "function": {
12
                "name": "get_weather",
13
                "description": "Get weather of a location, the user should supply a location first.",
14
                "parameters": {
15
                    "type": "object",
16
                    "properties": {
17
                        "location": {
18
                            "type": "string",
19
                            "description": "The city and state, e.g. San Francisco, CA",
20
                        }
21
                    },
22
                    "required": ["location"]
23
                },
24
            }
25
        },
26
    ]}
27
)
28
# 向模型发送单条数据
29
response = model.invoke("你好，今天北京的天气如何")
30
# 打印响应
31
rprint(response)

输出如下。可以看到，输出包含了 tool_calls 字段，说明工具被模型正确识别了：

1
AIMessage(
2
    content='你好！让我帮你查一下北京今天的天气情况。',
3
    additional_kwargs={
4
        'refusal': None,
5
        'reasoning_content': '用户想知道北京今天的天气情况。我需要使用get_weather工具来查询北京的天气。让我调用这个工具。'
6
    },
7
    response_metadata={
8
        'token_usage': {
9
            'completion_tokens': 78,
10
            'prompt_tokens': 303,
11
            'total_tokens': 381,
12
            'reasoning_tokens': 23,
13
            'cached_tokens': 256,
14
            ...
15
        },
16
        'model_provider': 'deepseek',
17
        'model_name': 'deepseek-v4-flash',
18
        'finish_reason': 'tool_calls',
19
        ...
20
    },
21
    id='lc_run--019e4480-c8df-7f33-87cf-8d978779b0f3-0',
22
    tool_calls=[
23
        {
24
            'name': 'get_weather',
25
            'args': {'location': '北京'},
26
            'id': 'call_00_BT3PTJVDQlb9C2uhhJkc4856',
27
            'type': 'tool_call'
28
        }
29
    ],
30
    invalid_tool_calls=[],
31
    usage_metadata={
32
        'input_tokens': 303,
33
        'output_tokens': 78,
34
        'total_tokens': 381,
35
        'input_token_details': {'cache_read': 256},
36
        'output_token_details': {'reasoning': 23}
37
    }
38
)

参数：extra_body。这里用于存放模型厂商基于 OpenAI API 协议扩展的字段。查阅 OpenAI Chat Completions 文档和 DeepSeek 对话补全 API 文档可知，thinking 是 DeepSeek 扩展的字段，用于控制是否启用思考模式。

1
from langchain.chat_models import init_chat_model
2
from dotenv import load_dotenv
3
from rich import print as rprint
4
# 从.env文件中加载环境变量
5
load_dotenv(override=True)
6
model = init_chat_model(
7
    model="deepseek:deepseek-v4-flash",
8
    extra_body={"thinking": {"type": "enabled"}},
9
)
10
# 向模型发送单条数据
11
response = model.invoke("你好，一句话回答")
12
# 打印响应
13
rprint(response)

输出如下。输出包含了 reasoning_content，说明启用了思考模式。与 extra_body={"thinking": {"type": "disabled"}} 可以对比：

1
AIMessage(
2
    content='你好，请问有什么可以帮你的？',
3
    additional_kwargs={
4
        'refusal': None,
5
        'reasoning_content': '好的，用户的问题很简单，就是要求“一句话回答”...'
6
    },
7
    response_metadata={
8
        'token_usage': {
9
            'completion_tokens': 87,
10
            'prompt_tokens': 8,
11
            'total_tokens': 95,
12
            'reasoning_tokens': 78,
13
            ...
14
        },
15
        'model_provider': 'deepseek',
16
        'model_name': 'deepseek-v4-flash',
17
        'finish_reason': 'stop',
18
        ...
19
    },
20
    tool_calls=[],
21
    invalid_tool_calls=[],
22
    usage_metadata={
23
        'input_tokens': 8,
24
        'output_tokens': 87,
25
        'total_tokens': 95,
26
        'input_token_details': {'cache_read': 0},
27
        'output_token_details': {'reasoning': 78}
28
    }
29
)

作为对比（未启用思考模式，不包含 reasoning_content）：

1
AIMessage(
2
    content='好的，我们一步一步来分析这个数学问题。题目是：\n\n2 + 3 * 2 = ？\n\n根据数学中的运算顺序规则（通常称为“先乘除，后加减”），我们应该先计算乘法部分。...',
3
    additional_kwargs={'refusal': None},
4
    response_metadata={
5
        'token_usage': {
6
            'completion_tokens': 119,
7
            'prompt_tokens': 14,
8
            'total_tokens': 133,
9
            'completion_tokens_details': None,
10
            ...
11
        },
12
        'model_provider': 'deepseek',
13
        'model_name': 'deepseek-v4-flash',
14
        'finish_reason': 'stop',
15
        ...
16
    },
17
    tool_calls=[],
18
    invalid_tool_calls=[],
19
    usage_metadata={
20
        'input_tokens': 14,
21
        'output_tokens': 119,
22
        'total_tokens': 133,
23
        'input_token_details': {'cache_read': 0},
24
        'output_token_details': {}
25
    }
26
)

6.3.3 需要记住哪些参数#

记住常见参数及用法即可，如果需要精细控制模型输出，可以查阅 OpenAI 和特定模型供应商的官方文档，通过 model_kwargs 或 extra_body 传递。

6.4 模型调用中 config 参数#

在调用模型时（如使用 invoke()、ainvoke()、stream()、batch() 等方法时），我们可以传入 config 参数。config 参数允许在调用模型时，动态地配置和控制模型的行为，而无需在初始化时就固定所有参数，这为应用带来了极大的灵活性和可维护性。

关于 config 中可配参数的解释参考：https://reference.langchain.com/python/langchain-core/runnables/config/RunnableConfig

举例：

1
deepseek_llm.invoke(
2
    "你好",
3
    config={
4
        "run_name": "...",   # 在LangSmith中这次运行会显示为指定名称
5
        "tags": ["test", "development"], # 打上标签便于分类查找
6
        "metadata": {"user_id": "123"},   # 记录用户ID
7
        "callbacks": [custom_handler],   # 启用自定义回调函数
8
        "configurable":{
9
            "model": "deepseek-reasoner", # 配置模型参数
10
            "temperature": 0.7,   # 配置温度参数
11
            "max_tokens": 100   # 配置最大令牌数
12
        }
13
    }
14
)

config 中支持配置的参数如下：

配置项	类型	描述
run_name	str	为当前运行设置一个可读的名称。如在 LangSmith 追踪系统中快速定位和识别不同的运行任务。
tags	List[str]	为运行设置标签，用于分类和过滤。
callbacks	List[BaseCallbackHandler]	设置回调处理器，在运行的不同阶段（开始、流输出、结束等）触发。与一些监控平台（如 LangSmith）集成进行深度追踪和调试。
metadata	Dict[str,Any]	附加任意的键值对元数据。记录本次调用的业务上下文，如 `{"user_id": "123", "session_id": "abc"}`。
max_concurrency	int	限制当前可运行对象的最大并发运行数。防止对 API 接口或本地资源造成过大压力，实现简单的速率限制。
recursion_limit	int	限制运行时递归调用的最大深度。主要在复杂的工作流（如 Agent 执行多步工具调用）中，防止出现无限递归循环。
configurable	Dict[str,Any]	一个万能字典，用于传递其他可配置参数。实现更高级的动态行为，如配置可替代的模型或组件。

说明如下：

config 中参数 run_name、tags、callbacks 主要用在 LangSmith 中，用于追踪、筛选和调试。
metadata 可以配置用户指定的一些信息，在工作流开发中，当整个流程被包装为 Runnable 链时，可以将这些参数传递给后续的链节点使用。
configurable 中可配置的参数与 init_chat_model 初始化模型参数一样，与在初始化模型时设置的参数（如 temperature=0.7）的关键区别在于：
- init_chat_model 初始化参数：模型的默认设置，适用于该模型实例的大部分场景。
- 运行时 config：单次调用的特定设置，优先级更高，针对本次调用进行的特殊调整。

举例1：当需要处理大量输入时，为了避免对模型服务造成过大压力或触发速率限制，在 config 中使用 max_concurrency 参数控制最大并行数。

1
large_list_of_inputs = [....,....,....]
2
model.batch(
3
    large_list_of_inputs,
4
    config={
5
        'max_concurrency': 5 # 限制最大并发数为5
6
    }
7
)

举例2：配置 configurable 覆盖默认参数时，需要在 init_chat_model 初始化模型中指定 configurable_fields 参数来指定模型运行时可替换的参数有哪些。

1
from langchain.chat_models import init_chat_model
2
from dotenv import load_dotenv
3
import os
4
from rich import print as rprint
5
# 从.env文件中加载环境变量
6
load_dotenv(override=True)
7
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
8
DEEPSEEK_BASE_URL = os.getenv("DEEPSEEK_BASE_URL")
9
# 1. 初始化模型
10
model = init_chat_model(
11
    model="deepseek-v4-flash",
12
    model_provider="deepseek",
13
    api_key=DEEPSEEK_API_KEY,
14
    base_url=DEEPSEEK_BASE_URL,
15
    temperature=0.2,
16
    max_tokens=500,
17
    # 指定可调整参数
18
    configurable_fields=("model", "model_provider", "temperature", "max_tokens"),
19
)
20
# 2. 准备 config 字典
21
config = {
22
    "run_name": "joke_generation",   # 在LangSmith中这次运行会显示为 "joke_generation"
23
    "tags": ["tag1", "tag2"],   # 打上标签便于分类查找
24
    "metadata": {"user_id": "123"},   # 记录用户ID
25
    "configurable":{
26
        "model": "deepseek-v4-pro",   # 配置模型参数
27
        "model_provider": "openai",   # 配置模型提供商参数
28
        "temperature": 0.7,   # 配置温度参数
29
        "max_tokens": 1000   # 配置最大令牌数
30
    }
31
}
32
# 3. 调用模型并传入config
33
response = model.invoke(
34
    "1 + 2 = ？",
35
    config=config
36
)
37
rprint(response)

输出如下：

1
AIMessage(
2
    content='1 + 2 = 3',
3
    additional_kwargs={'refusal': None},
4
    response_metadata={
5
        'token_usage': {
6
            'completion_tokens': 65,
7
            'prompt_tokens': 11,
8
            'total_tokens': 76,
9
            'reasoning_tokens': 57,
10
            ...
11
        },
12
        'model_provider': 'openai',
13
        'model_name': 'deepseek-v4-pro',
14
        'finish_reason': 'stop',
15
        ...
16
    },
17
    tool_calls=[],
18
    invalid_tool_calls=[],
19
    usage_metadata={
20
        'input_tokens': 11,
21
        'output_tokens': 65,
22
        'total_tokens': 76,
23
        'input_token_details': {'cache_read': 0},
24
        'output_token_details': {'reasoning': 57}
25
    }
26
)

Welcome to Stella blog

LangChain之02：模型的创建与调用#

1、模型调用的准备工作#

1.1 一张老图看大模型的调用#

1.2 模型初始化的分类方式#

1.3 线上大模型服务平台#

1.4 提前安装所有依赖#

2、模型初始化角度1：使用模型提供商库#

2.1 通过专用 API 调用#

2.1.1 DeepSeek 大模型#

2.1.2 智谱大模型#

2.1.3 千问大模型#

2.2 兼容用法#

2.3 中转平台#

2.3.1 OpenRouter#

2.3.2 CloseAI#

3、模型初始化角度1：init_chat_model() 统一接口#

3.1 使用举例#

3.2 小结：模型的创建#

3.3 模型初始化参数（常用版）#

4、模型初始化角度3：本地模型的部署与调用#

4.1 Ollama 的介绍#

4.2 Ollama 的下载-安装#

4.3 模型的下载#

4.4 LangChain 调用模型#

5、模型的调用#

5.1 invoke()#

5.1.1 invoke() 说明#

5.1.2 输入参数详解#

5.1.3 返回值详解#

5.2 流式调用#

5.3 批量调用#

5.3.1 一次性接收所有响应#

5.3.2 按完成顺序接收响应#

5.3.3 性能对比#

5.4 异步调用#

5.5 如何处理 API 调用失败#

6、拓展内容#

6.1 美化模型输出响应#

6.2 模型配置信息 profile#

6.3 模型初始化参数（完整版）#

6.3.1 查看所有初始化参数#

6.3.2 模型类的参数构成#

6.3.3 需要记住哪些参数#

6.4 模型调用中 config 参数#

目录