LangChain 模型调用方案：invoke、stream、batch 的同步与异步

引言

基于大模型开发应用，打交道最多的，其实就是怎么给 LLM 发请求、收结果。这一步做得顺不顺，直接决定了程序的性能和最终的用户体验。LangChain 在这方面做得挺周到，它用 init_chat_model 统一封装了三套核心的同步调用接口——invoke、stream 和 batch，还配套了对应的异步版本，专门应对高并发场景。这三兄弟底层共用同一套消息结构，但返回形式、阻塞逻辑和适用的业务场景差别很大。这篇就来把这三种核心方法掰开揉碎讲清楚，顺带对比下异步调用，理一理在实际项目中到底该怎么选、怎么用。

一、前置统一环境与模型初始化

下面所有示例都基于同一套 OpenRouter 兼容模型配置，支持字典消息，也支持 LangChain 内置的 HumanMessage/SystemMessage 传参：

from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage
from dotenv import load_dotenv
import os
import asyncio
from rich import print as rprint

# 加载环境变量
load_dotenv(override=True)

llm = init_chat_model(
    model="deepseek-v4-flash",
    model_provider="openai",
    api_key=os.getenv("OPENROUTER_API_KEY"),
    base_url=os.getenv("OPENROUTER_BASE_URL")
)

二、同步三大核心调用方法

2.1 invoke：单次同步完整调用（基础通用方案）

核心特性

同步等待，模型得把整段话全都想好了，才一次性吐出来给你。返回的是一个完整的 AIMessage 对象，里面装着完整的文本、Token 消耗、接口延迟、模型元数据，甚至还能包含工具调用信息。入参方式也很灵活，纯字符串、字典形式的对话数组、或者内置的消息对象，随便哪种都行。最适合的场景就是那些简短问答、后台单次查询，还有传统的多轮对话机器人。

基础示例

# 基础单轮提问
res = llm.invoke([
    SystemMessage(content="专业数学老师"),
    HumanMessage(content="什么是斐波那契数列")
])
print("完整回答：", res.content)

# 查看计费Token消耗
print("输入Token：", res.usage_metadata["input_tokens"])
print("输出Token：", res.usage_metadata["output_tokens"])

多轮对话记忆实现

LLM 本身不带记忆功能，这一点不管是哪种调用方式都一样。要实现多轮对话，就得手动把完整的历史拼接好传进去：

conversation = [
    {"role": "system", "content": "友好AI助手"},
    {"role": "user", "content": "我叫小张"}
]

# 第一轮调用
res1 = llm.invoke(conversation)

# 追加AI回复与新提问，留存上下文
conversation.append({"role": "assistant", "content": res1.content})
conversation.append({"role": "user", "content": "我叫什么名字？"})

res2 = llm.invoke(conversation)
print(res2.content)

优缺点

优点很直接：代码简单，信息完整，拿到一个 AIMessage，里面既有文本，又有 Token 统计、响应耗时这些元数据。缺点也明显：生成长文本时，整个程序都得干等着，前端会长时间一片空白，用户的体验感会大打折扣。

2.2 stream：流式分片调用（前端交互专用）

核心特性

返回的是一个迭代器，模型每生成一小段文字，就立刻推给你，完全不用等全文。返回的是分段的 Chunk 对象，通过 .text 字段可以提取到当前的字块，然后实时拼起来。这个方案简直就是为网页聊天和长文案生成量身定做的。

实战代码

print("流式实时输出：")
for chunk in llm.stream("详细讲解高等数学的作用"):
    print(chunk.text, end="", flush=True)

优缺点

最大的好处是首字返回的速度极快，可以轻松实现打字机一样的实时输出效果，用户体验会好很多。缺点是分片太零碎，没法直接拿到完整的 Token 统计，得自己手动把所有 chunk 的消耗数据累加起来。

2.3 batch / batch_as_completed：批量任务调用（离线数据处理）

批量接口可以一次性提交多条独立的问答，减少了重复的网络握手和鉴权开销。跟循环调用 invoke 相比，成本更低，耗时也更短。它有两种模式：

• batch()：老老实实等所有任务全部完成，然后按输入顺序，统一返回一个结果列表。
• batch_as_completed()：哪个任务先完成，就先返回哪个，输出顺序是乱的。这种模式最适合那些任务耗时差异比较大的场景。

batch 有序输出示例

tasks = [
    "1+99等于多少",
    "什么是大语言模型",
    "中国的首都"
]
results = llm.batch(tasks)

for index, ans in enumerate(results):
    print(f"问题{index+1}：{ans.content}
")

batch_as_completed 异步流式批量输出

tasks = ["介绍Python", "解释微积分", "一年多少天"]
for res in llm.batch_as_completed(tasks):
    print("单条批量结果：", res.content)

优缺点

批量处理海量问答、做数据清洗的时候，节约接口开销的效果很明显。但缺点也摆在那：只适合那些相互之间没有关联的独立问题，不能直接用来批量传递多轮对话的上下文。

三、异步调用：高并发服务必备方案

线上后端、FastAPI Web 服务、多任务并发这些场景里，用同步接口会把事件循环死死地卡住，服务的并发承载能力直线下降。所以 LangChain 为上面那三种方法都配上了异步版：

• ainvoke：异步单次完整调用，对应同步的 invoke
• astream：异步流式分片调用，对应同步的 stream
• abatch：异步批量调用，对应同步的 batch

3.1 ainvoke 异步单次调用示例

async def async_invoke_demo():
    res = await llm.ainvoke(HumanMessage(content="2+5*8等于几"))
    print("异步回答：", res.content)

asyncio.run(async_invoke_demo())

3.2 astream 异步流式输出示例

async def async_stream_demo():
    print("异步流式输出：")
    async for chunk in llm.astream("解释人工智能定义"):
        print(chunk.text, end="", flush=True)

asyncio.run(async_stream_demo())

3.3 abatch 异步批量调用示例

async def async_batch_demo():
    tasks = ["春天的特点", "什么是向量", "地球公转周期"]
    results = await llm.abatch(tasks)
    for r in results:
        print(r.content, "
")

asyncio.run(async_batch_demo())

异步接口核心价值

• 不阻塞事件循环，Web 服务处理上千用户请求也能游刃有余，并发吞吐量显著提升。
• 可以跟异步数据库、异步 HTTP 客户端搭在一起，搞成全链路异步架构。
• 特别适合后端接口、机器人服务、还有大规模的批量离线任务。

四、同步 / 异步调用方法对比汇总

五、业务选型指南

1. 本地跑个脚本、做个简单测试，直接用 invoke，代码最简单，信息也最全。
2. 要做网页聊天、前端这种交互场景，同步开发用 stream，线上服务就得上 astream。
3. 要批量处理上万条彼此独立的问答，用 batch 或 abatch，接口开销能省不少。
4. 如果是在 FastAPI 这类异步 Web 后端，或者需要支撑高并发的服务，直接全套异步走起：ainvoke、astream、abatch。
5. 多轮对话机器人这块，无论选哪种调用方法，都需要手动维护完整的对话历史，这一点跟调用方式本身没任何关系。

六、总结

invoke、stream、batch 这三招，构成了 LangChain 模型请求的三大基础能力，分别对应单次问答、实时交互和批量处理这三类核心业务场景。同步接口最适合本地开发和低并发场景，异步接口则是线上高并发服务的标配。不管选哪种，它们都共用一套消息传递规则，上下文记忆全靠手动拼接对话列表来实现。开发的时候，只要根据并发量和交互形式，挑对了接口，性能和用户体验就都能顾得上。