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

引言

在基于大模型开发应用时，如何向 LLM 发送请求、接收回复直接决定程序性能与用户体验。LangChain 基于 init_chat_model 封装了三套核心同步调用接口 invoke、stream、batch，同时配套异步版本适配高并发服务场景。三者底层共用同一套消息结构，但在返回形式、阻塞逻辑、适用业务场景上差异显著。本文聚焦三大核心调用方法，同步对比异步调用，理清选型标准与实战用法。

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

所有示例共用一套 OpenRouter 兼容模型配置，支持字典消息、LangChain 内置 HumanMessage/SystemMessage 传参：

from langchain.chat_models import init_chat_modelfrom langchain_core.messages import HumanMessage, SystemMessagefrom dotenv import load_dotenvimport osimport asynciofrom 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：单次同步完整调用（基础通用方案）

核心特性

1. 执行逻辑：阻塞式等待模型生成全部内容，完整输出后一次性返回；
2. 返回结构：单个 AIMessage 对象，内置完整文本、Token 消耗、接口延迟、模型元数据、工具调用信息；
3. 消息兼容：纯字符串、字典对话数组、内置消息对象均可作为入参；
4. 适配场景：简短问答、后台单次查询、多轮对话机器人。

基础示例

# 基础单轮提问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)

优缺点

优点：返回信息完整，可直接获取 Token 统计、响应耗时等全量元数据，代码简洁；缺点：长文本生成时全程阻塞，前端会出现长时间空白，交互体验差。

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

核心特性

1. 执行逻辑：返回迭代器，模型每生成一段文字就分片推送，无需等待全文；
2. 返回结构：分段 Chunk 对象，通过 .text 实时拼接文本；
3. 适配场景：网页聊天、长文案生成、实时展示模型推理过程。

实战代码

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

优缺点

优点：首字返回速度快，实现打字机实时输出，大幅优化用户体感；缺点：分片无法直接拿到完整 Token 统计，需要手动累加所有 chunk 的消耗数据。

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

批量接口一次性提交多条独立问题，减少重复网络握手、鉴权开销，相比循环多次调用 invoke 成本更低、耗时更短，分为两种执行模式。

1. batch()：等待所有任务全部完成，按输入顺序统一返回结果列表；
2. batch_as_completed()：任务完成一条立即返回一条，输出顺序打乱，适合任务耗时差异大的场景。

batch 有序输出示例

tasks = ["1+99等于多少","什么是大语言模型","中国的首都"]results = llm.batch(tasks)for index, ans in enumerate(results):print(f"问题{index+1}：{ans.content}n")

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, "n")asyncio.run(async_batch_demo())

异步接口核心价值

1. 不阻塞事件循环，Web 服务可同时处理上千用户请求，提升并发吞吐量；
2. 搭配异步数据库、异步 HTTP 客户端，实现全链路异步架构；
3. 适合后端接口、机器人服务、大规模批量离线任务。

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

表格

五、业务选型指南

1. 本地脚本、简单测试：优先 invoke，代码简单，信息完整；
2. 网页聊天、前端交互：同步选 stream，线上服务选 astream；
3. 批量处理上万条独立问答：batch / abatch，降低接口开销；
4. FastAPI、异步 Web 后端、高并发服务：统一使用全套异步 ainvoke/astream/abatch；
5. 多轮对话机器人：任意调用方法都需要手动维护完整对话历史，与调用方式无关。

六、总结

invoke、stream、batch 构成 LangChain 模型请求的三大基础能力，分别覆盖单次问答、实时交互、批量处理三大业务场景。同步接口适合本地开发、低并发场景，异步接口是线上高并发服务的标准方案。所有调用方法共用一套消息传递规则，上下文记忆均依赖手动拼接对话列表，开发时只需根据并发量、交互形式选择对应接口，即可兼顾性能与用户体验。