LangChain模型调用四种方式对比：invoke/stream/batch/async怎么选

夏天拐跑了西瓜

114人浏览 · 2026-10-30 10:05:51

夏天拐跑了西瓜 · 2026-10-30 10:05:51 发布

个人主页：夏天拐跑了西瓜
专栏传送门：《Redis入门到精通》、《Spring 生态全家桶体系化实战》、《Python AI Agent开发实战》
学习方向：Java 后端｜AI‑Agent 大模型应用开发爱好者
⭐⭐⭐⭐⭐人生格言：路虽远，行则将至

🔒 本文首发于CSDN，作者：夏天拐跑了西瓜，转载请注明出处及原文链接。

【Python AI Agent实战02】LangChain模型调用四种方式对比：invoke/stream/batch/async怎么选

🔥 本文是《Python AI Agent开发实战》系列教程的第三篇。上一篇我们搞定了模型初始化，这一篇正式进入模型调用（Invocation）。LangChain 提供了 invoke、stream、batch 三组同步方法和对应的 ainvoke、astream、abatch 异步方法，覆盖从简单问答到高并发流式输出的全部场景。本文逐一拆解每种调用方式的用法、适用场景和注意事项。

🖥️ 本文验证环境： Python 3.13 | LangChain 1.2 | langchain-deepseek 1.0.1 | macOS / Windows

🧭 本文速览

invoke 同步调用：三种输入形式（单条消息/字典列表/消息对象），最基础的调用方式
stream 流式输出：逐 token 实时返回，聊天机器人必备
batch 批量调用：并行处理多个请求，效率翻倍
异步调用全解：ainvoke / astream / abatch，高并发场景的核心技能
四种方式对比选型：什么时候用哪种，一张表说清楚

一、模型调用方式总览

在 LangChain 中，模型调用（Invocation）是指通过特定方法触发大语言模型生成输出的过程。不管你是做简单问答、流式聊天还是批量处理，调用入口只有这几个：

核心方法	主要特点	适用场景
invoke()	阻塞式，一次性返回完整结果	简单问答、批处理任务、无需实时反馈
stream()	流式输出，逐 token 实时返回	聊天机器人、长文本生成、提升用户体验
batch()	并行处理多个输入	批量问答、数据标注、情感分析等独立任务
ainvoke()	invoke 的异步版本	高并发 Web 应用、IO 密集型任务
astream()	stream 的异步版本	异步流式输出场景
abatch()	batch 的异步版本	异步批量处理场景

💡 提示：同步方法和异步方法的核心区别——同步方法会阻塞主线程，程序必须等结果返回后才能继续执行；异步方法（以 a 开头）不会阻塞，程序在等待模型响应的同时可以继续执行其他任务。在做 Web 服务（如 FastAPI）时，异步方法是标配。

二、invoke 同步调用

invoke() 是最直接、最常用的模型调用方法。它的工作模式是阻塞式的——程序会等待模型完全生成整个响应后，再一次性将结果返回。

2.1、单条消息调用

最简单的用法，直接传入一个字符串问题：

from init_llm import deepseek_llm

# 单条消息调用
resp = deepseek_llm.invoke("请介绍一下你自己")
print(resp.content)

运行结果：

你好！我是DeepSeek，由深度求索公司创造的AI助手。我擅长编程、数学、逻辑推理和知识问答，能帮你写代码、分析问题、提供建议。有什么需要，尽管说！

🎯 关键观察：invoke() 返回的是一个 AIMessage 对象，你需要通过 .content 属性来获取模型生成的文本内容。直接 print(resp) 会输出完整的 AIMessage 结构（包含 response_metadata 等元数据）。

2.2、消息列表调用（字典格式）

当你需要表达多轮对话历史时，可以传入一个消息列表（字典格式），每条消息通过 role 字段指定角色：

from init_llm import deepseek_llm

# 消息列表（字典格式）调用
conversation = [
    {"role": "system", "content": "你是一个有帮助的助手，可以将汉语翻译成英语。"},
    {"role": "user", "content": "翻译: 我喜欢编程"},
    {"role": "assistant", "content": "I love programming."},
    {"role": "user", "content": "翻译: 我喜欢大模型"}
]
resp = deepseek_llm.invoke(conversation)
print(resp.content)  # 输出: I like large models.

💡 提示：role 字段支持四种角色：

system：设定 AI 的角色和行为准则，相当于"工作说明书"
user：代表用户的输入
assistant：代表 AI 之前的回复，用于构建对话历史
tool：工具执行结果的反馈（在工具调用场景中使用）

2.3、消息列表调用（消息对象格式）

这是 LangChain 推荐的方式——使用内置的消息类（SystemMessage、HumanMessage、AIMessage），类型更安全，功能也更丰富：

from init_llm import deepseek_llm
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

conversation = [
    SystemMessage("你是一个有帮助的助手，可以将汉语翻译成英语。"),
    HumanMessage("翻译: 我喜欢编程"),
    AIMessage("I love programming."),
    HumanMessage("翻译: 我喜欢大模型")
]
resp = deepseek_llm.invoke(conversation)
print(resp.content)  # 输出: I like large models.

⚠️ 避坑提醒：虽然字典格式中 "user" 和 "human" 有时可以互换，但建议遵循 OpenAI 的惯例统一使用 "user"/"human"。使用消息对象格式时，LangChain 提供了 SystemMessage、HumanMessage、AIMessage、ToolMessage 四种类型，比字典格式更不容易出错。

2.4、Message Types 详解

[架构图：LangChain 消息类型体系]
BaseMessage（基类）
├── SystemMessage    → 设定 AI 角色/行为准则
├── HumanMessage     → 用户输入（文本/多模态）
├── AIMessage        → 模型输出（文本/工具调用请求）
└── ToolMessage      → 工具执行结果反馈给模型

消息类型	作用	典型场景
SystemMessage	在对话开始时设定角色、行为准则和上下文背景	“你是一个翻译助手”、“你是一个代码审查专家”
HumanMessage	代表用户的输入，支持简单文本和复杂多模态内容	用户提问、用户上传图片/文件
AIMessage	代表模型的输出，包含生成文本、工具调用、元数据等	模型回复、模型请求调用工具
ToolMessage	将工具执行的结果返回给模型，让模型基于结果继续生成	天气查询结果、数据库查询结果

🎯 关键观察：ChatModel 的 invoke 方法返回的始终是 AIMessage 对象。通过 .content 获取文本，通过 .tool_calls 获取工具调用请求，通过 .response_metadata 获取元数据（如 token 用量）。

三、stream 流式调用

3.1、为什么需要流式输出？

想象一下你在做一个聊天机器人——如果用 invoke()，用户发完消息后要干等 5-10 秒，然后突然看到一大段文字。体验很差。

而用 stream()，模型边生成边输出，用户可以看到文字一个字一个字地"打出来"，就像 ChatGPT 那样。这就是流式输出的核心价值：提升用户体验。

3.2、stream 使用示例

from typing import Iterator
from langchain_core.messages import AIMessageChunk
from init_llm import deepseek_llm

resp: Iterator[AIMessageChunk] = deepseek_llm.stream("使用20个字给我介绍什么是大模型？")

for chunk in resp:
    # end="" 避免打印时自动换行，flush=True 及时刷新输出
    print(chunk.content, end="", flush=True)

运行结果（逐 token 输出）：

大模型是基于海量数据训练的大规模神经网络，能理解和生成自然语言。

💡 提示：

stream() 返回一个迭代器（Iterator），逐个产生 AIMessageChunk 对象
每个 chunk 都包含输出内容的一部分，所有 chunk 拼接后与 invoke() 的结果一致
end="" 避免打印时自动换行，flush=True 确保内容及时显示

🎯 关键观察：流式输出本质上是把 invoke() 一次性返回的大块内容，拆成了多个小块（chunk）逐步返回。对于 Web 应用，这些 chunk 可以通过 SSE（Server-Sent Events）或 WebSocket 实时推送给前端。

四、batch 批量调用

4.1、为什么需要批量调用？

假设你要对 100 条用户评论做情感分析——如果用 invoke() 循环 100 次，每次都要等上一次完成才能开始下一次，效率极低。

batch() 的核心思想是将多个独立请求并行发送给模型处理，大幅减少网络往返开销和等待时间。

4.2、batch 使用示例

from langchain_core.runnables.utils import Output
from init_llm import deepseek_llm

responses: list[Output] = deepseek_llm.batch([
    "为什么鹦鹉的羽毛是彩色的？",
    "飞机是如何飞行的？",
    "什么是量子计算？"
])

for response in responses:
    print(response.content)

运行结果：

========================================
鹦鹉羽毛的绚丽色彩主要源于**色素沉积**、**结构色**以及**两者结合**的复杂机制...
========================================
这是一个非常棒的问题！飞机的飞行主要依赖于四个关键物理原理的协同作用：**升力、推力、重力和阻力**...
========================================
量子计算是一种**利用量子力学原理进行信息处理的新型计算范式**...

💡 提示：batch() 会等待所有请求处理完毕，按原始输入顺序返回结果列表。也就是说，即使第三个问题先完成，它也会排在结果列表的第三个位置。

4.3、batch_as_completed：谁先完成谁先返回

当你不想等最慢的那个请求时，batch_as_completed() 允许你在收到第一个结果后就立即处理：

from typing import Iterator
from langchain_core.runnables.utils import Output
from init_llm import deepseek_llm

responses: Iterator[tuple[int, Output | Exception]] = deepseek_llm.batch_as_completed([
    "为什么鹦鹉的羽毛是彩色的？",
    "飞机是如何飞行的？",
    "什么是量子计算？"
])

for response in responses:
    print(f"第 {response[0]} 个问题回答完毕: {response[1].content}")

运行结果（注意返回顺序可能不同）：

第 0 个问题回答完毕: 鹦鹉羽毛的绚丽色彩主要源于**色素沉积**和**结构色**的复杂结合...
第 2 个问题回答完毕: 量子计算是一种利用量子力学原理进行信息处理的新型计算范式...
第 1 个问题回答完毕: 飞机飞行主要依赖于升力、推力、重力和阻力四个关键物理原理...

🎯 关键观察：batch_as_completed() 返回的结果是乱序的——谁先完成谁先返回。每个结果包含一个元组 (索引, 输出)，索引对应原始列表中的位置。

4.4、控制并发数

当需要处理大量输入时，为了避免对模型服务造成过大压力或触发速率限制，可以通过 RunnableConfig 中的 max_concurrency 参数控制最大并行数：

# 限制最多同时进行 5 个并发调用
model.batch(
    large_list_of_inputs,
    config={
        'max_concurrency': 5  # 限制最大并发数为 5
    }
)

⚠️ 避坑提醒：不设 max_concurrency 时，batch() 会尽可能并行发送所有请求，可能触发 API 的速率限制（429 Too Many Requests）。生产环境中建议始终设置并发数上限。

五、异步调用全解（ainvoke / astream / abatch）

5.1、为什么需要异步？

在做 Web 服务（如 FastAPI/Flask）时，如果用同步方法调用模型，整个请求线程会被阻塞——模型生成需要 3-5 秒，线程就干等 3-5 秒。高并发下，线程池很快就被耗尽了。

异步方法（以 a 开头）让程序在等待模型响应的同时可以继续执行其他任务，大幅提升系统吞吐量。

[对比图：同步 vs 异步调用]
同步：请求1 → 等待3s → 返回 → 请求2 → 等待3s → 返回 → 请求3 → 等待3s → 返回
      总耗时：9s

异步：请求1 ──────────────────────────────────────────→ 返回
      请求2 ──────────────────────→ 返回
      请求3 ──────────→ 返回
      总耗时：~3s（取决于最慢的那个）

5.2、完整异步示例：ainvoke + astream + abatch

import asyncio
import time
from langchain.chat_models import init_chat_model

from env_utils import DEEPSEEK_API_KEY, DEEPSEEK_BASE_URL

# 初始化模型
llm = init_chat_model(
    model="deepseek-chat",
    model_provider="deepseek",
    api_key=DEEPSEEK_API_KEY,
    base_url=DEEPSEEK_BASE_URL,
)

async def demo_async_invoke():
    """演示单个异步调用的非阻塞特性"""
    print("=== 演示：ainvoke 的异步（非阻塞）效果 ===")
    print("程序开始...")

    # 1. 发起一个异步请求，但不立即等待结果
    print(">>> 发起异步模型调用 (ainvoke)...")
    async_task = llm.ainvoke("用一句话解释人工智能。")

    # 2. 在等待模型响应的同时，主程序可以继续执行其他任务
    print(">>> 模型请求已发送，程序无需等待，继续执行...")
    for i in range(3):
        await asyncio.sleep(1)
        print(f">>> 正在执行第{i + 1}个任务... ")

    # 3. 现在需要模型的结果了，用 await 等待它完成
    print(">>> 其他任务已完成，现在等待模型返回结果...")
    response = await async_task
    print(f">>> 模型返回: {response.content}")


async def demo_async_stream():
    """演示异步流式调用"""
    print("=== 演示：astream 的异步（非阻塞）效果 ===")
    print("程序开始...")

    # 1. 发起异步流式请求
    print(">>> 发起异步流式调用 (astream)...")
    stream_resp = llm.astream("请一句话解释机器学习的基本概念。")

    # 2. 在等待流式响应的同时，执行其他任务
    print(">>> 流式请求已发送，程序无需等待，继续执行...")
    for i in range(3):
        await asyncio.sleep(1)
        print(f">>> 正在执行第{i + 1}个任务... ")

    # 3. 开始处理流式结果
    print(">>> 其他任务已完成，开始处理流式结果...")
    print(">>> 流式输出: ", end="", flush=True)
    async for chunk in stream_resp:
        if hasattr(chunk, 'content'):
            print(chunk.content, end="", flush=True)
    print("\n>>> 流式输出结束\n")


async def demo_async_batch():
    """演示异步批量调用"""
    print("=== 演示：abatch 的异步（非阻塞）效果 ===")
    print("程序开始...")

    questions = ["用一句话说明深度学习与传统机器学习的区别"]

    # 1. 发起异步批量请求
    print(">>> 发起异步批量调用 (abatch)...")
    batch_resp = llm.abatch(questions)

    # 2. 在等待批量处理的同时，执行其他任务
    print(">>> 批量请求已发送，程序无需等待，继续执行...")
    for i in range(3):
        await asyncio.sleep(1)
        print(f">>> 正在执行第{i + 1}个任务... ")

    # 3. 等待批量处理结果
    print(">>> 其他任务已完成，现在等待批量处理结果...")
    responses = await batch_resp
    for response in responses:
        print(f">>> 批量响应: {response.content}")


async def main():
    """主函数"""
    await demo_async_invoke()
    await demo_async_stream()
    await demo_async_batch()


if __name__ == "__main__":
    asyncio.run(main())

运行结果：

=== 演示：ainvoke 的异步（非阻塞）效果 ===
程序开始...
>>> 发起异步模型调用 (ainvoke)...
>>> 模型请求已发送，程序无需等待，继续执行...
>>> 正在执行第1个任务... 
>>> 正在执行第2个任务... 
>>> 正在执行第3个任务... 
>>> 其他任务已完成，现在等待模型返回结果...
>>> 模型返回: 人工智能是让机器模拟人类智能行为的技术。

=== 演示：astream 的异步（非阻塞）效果 ===
程序开始...
>>> 发起异步流式调用 (astream)...
>>> 流式请求已发送，程序无需等待，继续执行...
>>> 正在执行第1个任务... 
>>> 正在执行第2个任务... 
>>> 正在执行第3个任务... 
>>> 其他任务已完成，开始处理流式结果...
>>> 流式输出: 机器学习是让计算机从数据中自动学习规律并做出预测或决策的技术。
>>> 流式输出结束

=== 演示：abatch 的异步（非阻塞）效果 ===
程序开始...
>>> 发起异步批量调用 (abatch)...
>>> 批量请求已发送，程序无需等待，继续执行...
>>> 正在执行第1个任务... 
>>> 正在执行第2个任务... 
>>> 正在执行第3个任务... 
>>> 其他任务已完成，现在等待批量处理结果...
>>> 批量响应: 深度学习通过多层神经网络自动学习数据的层次化特征，而传统机器学习更依赖人工设计的特征和浅层模型。

🎯 关键观察：注意每个异步方法中，await asyncio.sleep(1) 模拟的"其他任务"和模型调用是并行执行的。3 个 sleep 共 3 秒，模型响应也大约 3 秒——但总耗时只有 3 秒左右，而不是 6 秒。这就是异步的价值。

六、四种调用方式对比选型

对比维度	invoke	stream	batch	异步系列
返回方式	一次性完整结果	逐 token 流式返回	结果列表（有序/乱序）	与同步对应，但不阻塞
阻塞行为	阻塞等待	阻塞但逐步输出	阻塞等待全部完成	非阻塞，可并行执行其他任务
适用场景	简单问答、脚本测试	聊天机器人、长文本生成	批量数据标注、情感分析	Web 服务、高并发场景
用户体验	等待后一次性看到结果	实时看到文字"打出来"	批量处理，统一返回	不卡顿，响应快
复杂度	最简单	简单	简单	需要 async/await 基础

💡 我的建议：

脚本/学习阶段：用 invoke()，最简单直接。
做聊天机器人/交互界面：必须用 stream()，用户体验完全不同。
批量处理独立任务：用 batch()，效率远高于循环 invoke()。
做 Web 服务（FastAPI 等）：必须用异步系列（ainvoke/astream/abatch），否则线程会被阻塞殆尽。

七、总结

模块	要点
invoke 三种输入	单条消息 / 字典列表 / 消息对象（推荐），返回 AIMessage
stream 流式输出	逐 token 返回，适合聊天机器人，chunk 拼接后等于 invoke 结果
batch 批量调用	并行处理多个请求，batch_as_completed 支持乱序先返回
异步调用	ainvoke/astream/abatch，非阻塞，高并发 Web 服务必备
选型建议	脚本用 invoke，交互用 stream，批处理用 batch，Web 服务用异步

下一篇预告： 我们将讲解 LangChain 模型的结构化输出——用 Pydantic、TypedDict、JSON Schema 三种方式约束模型输出，让 AI 返回的数据可以被程序精准解析和下游处理。