Langchain的models学习与实践（国内大模型）

整体步骤

1.导入兼容 OpenAI 接口的国内模型和设置环境变量
​​

from langchain_openai import ChatOpenAI

许多国内大模型平台（如阿里云通义千问、智谱 AI、DeepSeek 等）在设计 API 时，采用了与 OpenAI 官方 API 高度相似的接口规范，如使用相同的 HTTP 方法（POST）和路径结构（如 /v1/chat/completions）；请求体和响应体的 JSON 格式与 OpenAI 保持一致（例如 messages 数组、model 字段、choices 等）；认证方式通常也是通过 HTTP 头部 Authorization: Bearer <API_KEY> 实现。
所以我们可以直接使用 ChatOpenAI 来调用这些国内模型，只需更换 base_url（基础地址）和 api_key 即可，这大大简化了多模型集成的代码复杂度。

import os os.environ['OPENAI_API_KEY'] = "你的API-KEY" # 注意"你的API-KEY"改成你申请的api key os.environ['OPENAI_API_BASE'] = "https://open.bigmodel.cn/api/paas/v4/" # 兼容端点,即base url。

通过环境变量设置 API 密钥和基础 URL，本篇文章用的智谱。

2.使用 ChatOpenAI 类实例化模型

类名是 ChatOpenAI，因为它兼容 OpenAI 的接口规范。
ChatOpenAI 是 LangChain 中用于调用兼容 OpenAI 接口的聊天模型的类，继承自 BaseChatModel，常用参数如下（按功能分类）:

llm = ChatOpenAI( model="glm-4-flash", #指定国内模型的名字，如 qwen-plus, qwen-max, deepseek-chat 等 temperature=0, # 也可以在这里直接传入 openai_api_key 和 openai_api_base # openai_api_key="你的API-KEY", # openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1", # max_tokens=512, # 可选，根据需要设置 )

调用模型

直接输出结果

response = llm.invoke("你叫什么名字？") print(response.content)

invoke 是 LangChain 中所有可运行组件（Runnable）的核心方法，用于同步调用语言模型。invoke 会将输入的消息发送给模型，等待模型生成回复后返回结果。
invoke 返回一个 AIMessage 对象（在较新版本中，返回类型为 BaseMessage 的子类，其 type 为 "ai"）。该对象包含模型生成的回复内容及可能的其他元数据。

常用属性和方法：

.content：获取生成的文本内容（字符串）

.response_metadata：获取 API 返回的元数据，如 token 用量、模型名称等（可能因平台而异）

.type：消息类型，通常为 "ai"。

流式打印每个块

for chunk in llm.stream("你叫什么名字？"): print(chunk.content, end="", flush=True)

.stream(...)返回一个迭代器，每次迭代生成一个AIMessageChunk对象，chunk 是 AIMessageChunk 对象，其 content 属性包含文本片段。在打印时设置flush=True可以强制立即输出，避免缓冲，从而实现实时的流式效果。

批

questions = [ "什么是langchain？", "什么是langgraph？", ] print("使用batch()") responses = llm.batch(questions) # 返回一个列表，顺序与输入一致 for i, response in enumerate(responses): print(f"Q{i+1}: {questions[i]}") print(f"A: {response.content}\n") print("使用batch_as_completed()") # batch_as_completed 返回一个迭代器，每个元素是 (index, response) 元组 for idx, response in llm.batch_as_completed(questions): print(f"Q{idx+1}: {questions[idx]}") print(f"A: {response.content}\n")

输入格式：两个方法都接受一个列表，列表元素可以是字符串（代表用户消息），也可以是更复杂的消息格式（如 HumanMessage 对象），这里直接使用字符串即可。
返回类型：batch() 返回 List[AIMessage]；batch_as_completed() 返回迭代器，元素为 (int, AIMessage)，其中 int 是原始输入列表中的索引。

方法一：使用 batch() 一次性获取所有结果（按输入顺序返回）

batch() 方法可以并行发送多个请求，等待所有请求完成后，按输入顺序返回一个包含所有 AIMessage 对象的列表。
结果顺序与输入严格一致，便于处理；必须等待最慢的那个请求完成后才能得到所有结果（整体延迟取决于最慢的请求）。

方法二：使用 batch_as_completed() 按完成顺序实时获取结果（可能乱序）

batch_as_completed() 方法则同样并行发送多个请求，但一旦某个请求完成，就立即通过迭代器返回 (index, response) 对，不保证返回顺序（先完成的先返回）。
batch_as_completed() 方法可以实时处理先返回的结果，提升用户体验（尤其当请求处理时间差异较大时）；需要额外处理 index 来匹配原始输入，因为返回顺序可能乱序。

整体代码

from langchain_openai import ChatOpenAI import os os.environ['OPENAI_API_KEY'] = "" os.environ['OPENAI_API_BASE'] = "https://open.bigmodel.cn/api/paas/v4/" llm = ChatOpenAI( model="glm-4-flash", # 指定国内模型的名字，如 qwen-plus, qwen-max, deepseek-chat 等 temperature=0, # 也可以在这里直接传入 openai_api_key 和 openai_api_base # openai_api_key="你的API-KEY", # openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1", # max_tokens=512, # 可选，根据需要设置 ) response = llm.invoke("你叫什么名字？") print(response.content) ####################################################### # 流式打印每个块（用 "" 分隔） for chunk in llm.stream("你叫什么名字？"): # chunk 是 AIMessageChunk 对象，其 content 属性包含文本片段 print(chunk.content, end="", flush=True) ####################################################### questions = [ "什么是langchain？", "什么是langgraph？", ] print("使用batch()") responses = llm.batch(questions) # 返回一个列表，顺序与输入一致 for i, response in enumerate(responses): print(f"Q{i+1}: {questions[i]}") print(f"A: {response.content}\n") print("使用batch_as_completed()") for idx, response in llm.batch_as_completed(questions): print(f"Q{idx+1}: {questions[idx]}") print(f"A: {response.content}\n")