LangChain 模型调用全指南：从 API 配置到异步并发与能力画像系列文章

---

前言

---

`# LangChain 模型调用全指南：从 API 配置到异步并发与能力画像

本文面向刚开始使用 LangChain 的开发者。从最基础的 API 配置开始，逐步讲清模型初始化、消息对象、流式输出、批量调用、异步任务、Token 元数据和模型能力检查。示例以 Python 为主，并覆盖 DeepSeek、OpenAI 兼容平台、OpenRouter 与 Ollama 本地模型。

一、LangChain 到底解决什么问题？

LangChain 不是一个大模型，而是连接大模型、数据、工具和应用流程的开发框架。它常见的应用场景包括：

场景	解决的问题	核心能力
自动化写作	输出格式不稳定	提示词模板、输出解析器
数据连接	PDF、Excel 等数据难以直接利用	文档加载、切分、结构化处理
对话系统	多轮对话容易丢失历史	消息历史、会话记忆
RAG	知识滞后与幻觉	外部知识检索、向量数据库
多模态	只能处理文字	图片、语音、视频输入
Agent	模型不能直接执行复杂任务	任务规划、工具调用、结果检查

推荐的学习顺序是：

API 配置 → 模型初始化 → 单次调用 → 消息与记忆
→ 流式与批量调用 → 异步并发 → 元数据与能力检查
→ RAG、多模态与 Agent

---

二、用 .env 安全配置 API

2.1 为什么不把 API Key 直接写进代码？

API Key 属于敏感信息。如果直接写入 Python 文件并上传到代码仓库，其他人可能盗用额度。更安全的做法是把密钥保存在项目根目录的 .env 文件中。

安装依赖：

pip install python-dotenv

创建 .env：

DEEPSEEK_API_KEY=你的API密钥
DEEPSEEK_BASE_URL=你的API地址

Python 中读取配置：

import os
from dotenv import load_dotenv

load_dotenv(override=True)

api_key = os.getenv("DEEPSEEK_API_KEY")
base_url = os.getenv("DEEPSEEK_BASE_URL")

不要忘记在 .gitignore 中加入：

.env

2.2 load_dotenv(override=True) 是什么意思？

load_dotenv() 默认不会覆盖系统中已经存在的同名环境变量：

load_dotenv()                 # 等价于 override=False
load_dotenv(override=True)    # 让 .env 覆盖同名系统变量

假设系统变量和 .env 中都存在 DEEPSEEK_API_KEY：

写法	最终优先使用
load_dotenv()	系统环境变量
load_dotenv(override=False)	系统环境变量
load_dotenv(override=True)	.env 文件

本地开发时，override=True 能让项目配置更直观；生产环境要谨慎使用，避免覆盖服务器注入的安全配置。

---

三、四种常见模型接入方式

3.1 厂商专用类：ChatDeepSeek

pip install -U langchain-deepseek

import os
from dotenv import load_dotenv
from langchain_deepseek import ChatDeepSeek

load_dotenv(override=True)

DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
DEEPSEEK_BASE_URL = os.getenv("DEEPSEEK_BASE_URL")

deepseek_llm = ChatDeepSeek(
    api_key=DEEPSEEK_API_KEY,
    api_base=,DEEPSEEK_BASE_URL,
    model="deepseek-v4-pro",
    
)

response = deepseek_llm.invoke("请用一句话介绍 LangChain")
print(response.content)

专用类的优势是厂商能力适配更完整；缺点是切换服务商时可能需要更换依赖和初始化代码。

api_base、base_url、model、model_name 等参数名称可能随集成包版本变化，请以当前版本文档为准。

3.2 OpenAI 兼容接口：ChatOpenAI

很多模型厂商和中转平台提供 OpenAI 兼容协议，因此可以用统一客户端接入：

pip install -U langchain-openai

from langchain_openai import ChatOpenAI
load_dotenv(override=True)
import os

DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
DEEPSEEK_BASE_URL = os.getenv("DEEPSEEK_BASE_URL")

deepseek_llm = ChatOpenAI(
    api_key=DEEPSEEK_API_KEY,
    base_url=DEEPSEEK_BASE_URL,
    model="deepseek-v4-pro",
)
response = deepseel_llm.invoke("1 + 9 = ?")
print(response.content)

这里使用 ChatOpenAI，不代表实际调用的一定是 OpenAI 模型。它也可以只是一个 OpenAI 兼容协议客户端。

3.3 统一入口：init_chat_model()

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

load_dotenv(override=True)

DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
DEEPSEEK_BASE_URL = os.getenv("DEEPSEEK_BASE_URL")

model = init_chat_model(
    model="deepseek-v4-pro",
    model_provider="deepseek",
    api_key=DEEPSEEK_API_KEY
    base_url=DEEPSEEK_BASE_URL
)
response = model.invoke("1 + 1 = ?")
print(response)

也可以把提供商写进模型字符串：

model = init_chat_model(
    model="deepseek:deepseek-chat",
    api_key=DEEPSEEK_API_KEY,
    base_url=DEEPSEEK_BASE_URL,
)

判断规则可以简单记成：

model="deepseek-chat"
→ 可能需要 model_provider="deepseek"

model="deepseek:deepseek-chat"
→ 通常不需要重复指定 model_provider

如果调用的是 OpenAI 兼容中转平台，model_provider 可能需要写成 openai：

model = init_chat_model(
    model="平台规定的模型名称",
    model_provider="openai",
    api_key=PLATFORM_API_KEY,
    base_url=PLATFORM_BASE_URL,
)

这里的 model_provider 更准确地说是 LangChain 选择的接口适配器，不一定等于模型实际所属厂商。

3.4 OpenRouter 等聚合平台

import os
from langchain_openrouter import ChatOpenRouter
from dotenv import load_dotenv

load_dotenv(override=True)

OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")
# OPENROUTER_API_BASE = os.getenv("OPENROUTER_API_BASE")

model = ChatOpenRouter(
    api_key=OPENROUTER_API_KEY,
    model="deepseek/deepseek-v4-pro",
    #base_url=OPENROUTER_API_BASE,
)

print(model.invoke("一句话介绍你自己"))

聚合平台中的模型标识通常采用：

厂商名称/模型名称

模型名会随平台更新，调用前应查看平台当前模型列表，不要把教程中的示例名称当成永久可用值。

3.5 Ollama 本地模型

Ollama 可以在本地下载、运行模型并提供 HTTP API。

先安装 Ollama 桌面程序，再安装 LangChain 集成：

pip install -U langchain-ollama
ollama pull deepseek-r1:1.5b

调用本机模型：

from langchain_ollama import ChatOllama

ollama_llm = ChatOllama(
    model="deepseek-r1:1.5b",
    base_url="http://localhost:11434",
    temperature=0.3,
)
question = "你好，请你介绍一下你自己。"

response = ollama_llm.invoke(question)
print(response.content)

如果 Ollama 位于局域网另一台机器，可将 base_url 改为它的局域网地址。不要把无鉴权的 11434 端口直接暴露到公网。

---

四、初始化参数：temperature 如何影响输出？

temperature 控制采样随机性。它不会让模型“变聪明”，只是改变输出的集中程度和多样性。

范围	表现	常见场景
0	最稳定，输出更集中	数学、代码、字段提取
0.1～0.3	准确优先，略有表达变化	RAG、技术问答
0.4～0.7	准确性与多样性平衡	聊天、总结、博客初稿
0.8～1.2	更灵活，也更容易出错	标题、创意方案
1.3～2.0	高度发散，可能不连贯	实验性创作

需要注意：

• 不同模型允许的范围可能不同。
• temperature=0 也不保证每次结果完全一致。
• 技术内容生成可用 0.3～0.7，事实核验和结构化提取建议更低。

---

五、模型调用的六种核心方法

LangChain 的模型调用可以按“同步/异步”和“单次/流式/批量”两个维度理解：

类型	同步	异步
单次完整调用	invoke()	ainvoke()
流式调用	stream()	astream()
批量调用	batch()	abatch()

5.1 invoke()：同步单次调用

response = model.invoke("什么是 LangChain？")
print(response.content)

程序会等待模型生成完整答案，适合脚本、后台任务和简单问答。

5.2 stream()：同步流式输出

for chunk in model.stream("详细介绍 LangChain"):
    print(chunk.content, end="", flush=True)

stream() 返回迭代器，模型生成一部分，程序就处理一部分。chunk 是消息分块，不保证每块恰好等于一个 Token。

如果需要保留完整文本：

parts = []

for chunk in model.stream("详细介绍 LangChain"):
    print(chunk.content, end="", flush=True)
    parts.append(chunk.content)

full_content = "".join(parts)

5.3 batch()：批量调用

inputs = [
    "翻译成英文：春天来了",
    "翻译成英文：夏天很热",
    "翻译成英文：秋天落叶",
    "翻译成英文：冬天下雪",
]

responses = model.batch(
    inputs,
    config={"max_concurrency": 3},
)

for i, response in enumerate(responses):
    print(f"{i + 1}. {response.content}")

enumerate(responses) 同时给出下标和响应对象：

i           → 0、1、2、3
response    → 当前 AIMessage
i + 1       → 显示为 1、2、3、4
response.content → 模型回答正文

batch() 的主要优势是提高独立任务吞吐量、减少总等待时间。它不一定只发送一次网络请求，也不必然减少 Token 或费用。

5.4 测量批量调用耗时

import time

start = time.perf_counter()
responses = model.batch(inputs)
batch_time = time.perf_counter() - start

print(f"耗时：{batch_time:.2f}秒")

这一行：

batch_time = time.perf_counter() - start

表示“当前高精度计时值减去开始计时值”，结果就是本次批量调用耗时。:.2f 表示保留两位小数。

5.5 batch_as_completed()：谁先完成就先处理谁

for index, response in model.batch_as_completed(inputs):
    print(f"输入 {index}：{inputs[index]}")
    print(f"回答：{response.content}")

它返回：

(index, AIMessage)

结果按完成顺序产生，可能乱序。index 用来关联原始输入。如果需要恢复原顺序：

results = [None] * len(inputs)

for index, response in model.batch_as_completed(inputs):
    results[index] = response.content

5.6 异步版本

response = await model.ainvoke("什么是 LangChain？")
responses = await model.abatch(inputs)

异步按完成顺序返回：

async for index, response in model.abatch_as_completed(inputs):
    print(index, response.content)

异步流式输出：

async for chunk in model.astream("详细介绍 LangChain"):
    print(chunk.content, end="", flush=True)

---

六、消息对象与手动传递记忆

6.1 三种核心消息对象

from langchain_core.messages import (
    SystemMessage,
    HumanMessage,
    AIMessage,
)

对象	角色	作用
SystemMessage	system	设置身份、规则和回答风格
HumanMessage	user	表示用户输入
AIMessage	assistant	表示模型历史回复

示例：

messages = [
    SystemMessage(content="你是一个专业的数学老师。"),
    HumanMessage(content="2 + 3 * 2 = ?"),
    AIMessage(content="8"),
    HumanMessage(content="我刚才问了什么问题？"),
]

response = model.invoke(messages)
print(response.content)

6.2 “记忆”的本质

大多数模型 API 是无状态的。模型不会自动记住上一次请求，程序需要在下一次调用中重新传入历史消息：

messages = [
    SystemMessage(content="你是一个友好的助手"),
    HumanMessage(content="你好，我叫小明"),
]

response1 = model.invoke(messages)
messages.append(response1)  # response1 通常就是 AIMessage

messages.append(
    HumanMessage(content="我叫什么名字？")
)

response2 = model.invoke(messages)
print(response2.content)

这不是模型内部永久记忆，而是：

程序保存历史 → 下一次请求重新发送历史 → 模型根据上下文回答

这种方式的局限包括：

• 程序关闭后列表会丢失。
• 历史越长，输入 Token 越多。
• 可能超过模型上下文窗口。
• 多用户应用必须为每个会话分别保存历史。

进阶方案通常是会话存储、历史裁剪、摘要记忆和数据库持久化。

---

七、异步到底在解决什么问题？

异步不会让单个模型推理得更快。它的价值是：等待网络响应时，程序可以处理其他任务。

7.1 async、await 与 asyncio.run()

import asyncio

async def main():
    response = await model.ainvoke("什么是 LangChain？")
    print(response.content)

asyncio.run(main())

• async def：定义异步函数。
• await：等待异步操作，同时允许事件循环执行其他任务。
• asyncio.run()：在普通脚本中启动事件循环并运行异步主函数。

7.2 asyncio.create_task()

import asyncio
import time

async def demo_async_invoke():
    start_time = time.perf_counter()

    async_task = asyncio.create_task(
        model.ainvoke("用一句话解释人工智能。")
    )

    for i in range(3):
        await asyncio.sleep(1)
        elapsed = time.perf_counter() - start_time
        print(f"正在执行第 {i + 1} 个本地任务，已耗时 {elapsed:.2f}s")

    response = await async_task
    print(response.content)

asyncio.run(demo_async_invoke())

工作过程是：

创建模型任务
→ 当前协程在 await 处让出控制权
→ 事件循环推进模型请求和其他任务
→ await async_task 获取最终结果

如果执行 await async_task 时模型已经完成，会立即取得结果；否则继续等待。

7.3 同时运行多个异步任务

async def ask(question):
    response = await model.ainvoke(question)
    return response.content

async def main():
    tasks = [
        asyncio.create_task(ask("什么是 RAG？")),
        asyncio.create_task(ask("什么是 Agent？")),
        asyncio.create_task(ask("什么是 LangChain？")),
    ]

    results = await asyncio.gather(*tasks)

    for i, result in enumerate(results):
        print(f"{i + 1}. {result}")

asyncio.run(main())

真实项目还要考虑平台限流、超时、异常处理和最大并发数。

---

八、理解 AIMessage 与响应元数据

model.invoke() 返回的通常不是普通字符串，而是 AIMessage：

response = model.invoke("2 + 3 * 2 = ?")

8.1 content

print(response.content)

这是模型生成的正文，也是业务中最常用的字段。

8.2 additional_kwargs

print(response.additional_kwargs)

保存服务商返回、但没有被 LangChain单独标准化的附加字段。例如：

{"refusal": None}

通常表示回答没有被安全策略拒绝，但具体含义取决于平台。

8.3 usage_metadata

LangChain 标准化后的 Token 用量：

print(response.usage_metadata)

常见结构：

{
    "input_tokens": 16,
    "output_tokens": 15,
    "total_tokens": 31,
}

跨模型平台统计用量时，优先使用它。

8.4 response_metadata

print(response.response_metadata)

这里更接近服务商原始响应，可能包含：

• model_name：实际模型名称
• finish_reason：停止原因
• token_usage：详细 Token 数据
• id：响应标识
• 延迟与服务层级信息

常见 Token 字段：

字段	含义
prompt_tokens	输入 Token
completion_tokens	输出 Token
total_tokens	总 Token
reasoning_tokens	推理 Token
cached_tokens	缓存命中的输入 Token
audio_tokens	音频相关 Token

元数据结构取决于模型和平台，业务代码应允许字段缺失。

8.5 工具调用字段

response.tool_calls
response.invalid_tool_calls

• tool_calls：成功解析的工具调用。
• invalid_tool_calls：格式错误或无法解析的工具调用。

普通问答中它们通常为空。

---

九、检查模型能力：model.profile

部分 LangChain 集成支持通过 profile 查看标准化能力画像：

from rich import print as rprint

rprint(model.profile)

可能得到：

{
    "max_input_tokens": 128000,
    "max_output_tokens": 16384,
    "text_inputs": True,
    "image_inputs": True,
    "audio_inputs": False,
    "video_inputs": False,
    "text_outputs": True,
    "image_outputs": False,
    "tool_calling": True,
    "structured_output": True,
}

其中：

• image_inputs=True 表示能理解图片，不代表能生成图片。
• image_outputs 才表示是否能直接输出图片。
• tool_calling 表示是否支持工具调用。
• structured_output 表示是否支持结构化输出。

profile 可能缺失或不完全准确，关键能力仍应以服务商文档和实际小规模调用验证为准。

---

十、查看初始化字段：model_fields

如果想知道模型类定义了哪些字段，可以利用 Pydantic 的类字段信息：

from langchain_deepseek import ChatDeepSeek

print(ChatDeepSeek.model_fields.keys())

排序后逐行输出：

for field_name in sorted(ChatDeepSeek.model_fields.keys()):
    print(field_name)

查看某个字段详情：

field = ChatDeepSeek.model_fields["temperature"]

print("类型：", field.annotation)
print("默认值：", field.default)
print("说明：", field.description)

为什么字段很多？因为其中可能同时包含：

• 当前模型集成自定义的字段
• BaseChatModel 等父类继承的公共字段
• OpenAI 兼容层字段
• 同步和异步 HTTP 客户端字段
• 旧参数的兼容别名

需要特别注意：model_fields 表示 Python 类能够接收或管理的字段，不代表底层 API 一定支持所有字段。确认顺序建议为：

服务商 API 文档
→ LangChain 当前集成文档
→ model_fields 与源码
→ 实际调用验证

---

十一、常见误区汇总

误区 1：ChatOpenAI 只能调用 OpenAI 模型

不一定。只要平台提供 OpenAI 兼容协议，就可能通过 ChatOpenAI 调用其他厂商模型。

误区 2：model_provider 永远等于实际模型厂商

不一定。通过兼容接口时，它可能表示 LangChain 应选择的协议适配器。

误区 3：stream() 每次严格返回一个 Token

不一定。它返回消息分块，分块边界由模型和接口实现决定。

误区 4：batch() 一定减少费用

不一定。它主要改善吞吐量和总等待时间，Token 总量和请求数量可能没有下降。

误区 5：异步会让模型本身生成得更快

不会。异步提高的是等待期间的资源利用率和系统整体并发能力。

误区 6：安装 Python 的 ollama 包就安装了本地模型服务

不是。Ollama 服务需要单独安装和运行，Python 包只是客户端。

误区 7：把历史消息放进列表就是永久记忆

不是。这只是当前程序内的上下文传递，程序退出后会丢失，还会持续增加 Token 消耗。

---

十二、总结：一张路线图串起全部知识

完成本文后，可以把 LangChain 模型调用理解为四层：

第一层：连接模型
.env、API Key、base_url、官方 SDK、兼容接口、Ollama

第二层：组织输入输出
SystemMessage、HumanMessage、AIMessage、content、元数据

第三层：选择执行方式
invoke、stream、batch、batch_as_completed 及其异步版本

第四层：构建可靠应用
会话记忆、并发控制、能力画像、参数检查、RAG、多模态、Agent

真正开始做项目时，不需要一上来就搭建复杂 Agent。先让一个模型调用稳定运行，再加入消息历史、流式输出和并发控制，最后根据业务需要扩展 RAG、工具调用与持久化记忆，开发路径会清晰很多。

---

附：最小可运行模板

import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage

load_dotenv()

model = init_chat_model(
    model="deepseek-chat",
    model_provider="deepseek",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_BASE_URL"),
    temperature=0.2,
)

messages = [
    SystemMessage(content="你是一名严谨的 Python 教师。"),
    HumanMessage(content="请用简单语言解释 LangChain。"),
]

response = model.invoke(messages)

print(response.content)
print(response.usage_metadata)

这份模板涵盖了环境变量、统一模型初始化、消息对象、单次调用、正文读取和 Token 用量查看，可以作为后续学习 RAG、Agent 与多模态应用的起点。
提示：以下是本篇文章正文内容，下面案例可供参考`

一、pandas是什么？

示例：pandas 是基于NumPy 的一种工具，该工具是为了解决数据分析任务而创建的。

二、使用步骤

1.引入库

代码如下（示例）：

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
import warnings
warnings.filterwarnings('ignore')
import  ssl
ssl._create_default_https_context = ssl._create_unverified_context

2.读入数据

代码如下（示例）：

data = pd.read_csv(
    'https://labfile.oss.aliyuncs.com/courses/1283/adult.data.csv')
print(data.head())

该处使用的url网络请求的数据。

---

总结

提示：这里对文章进行总结：

例如：以上就是今天要讲的内容，本文仅仅简单介绍了pandas的使用，而pandas提供了大量能使我们快速便捷地处理数据的函数和方法。