LangChain 模型调用：invoke、stream 与 batch

LangChain 为模型提供了一套统一的调用接口。本文以聊天模型（Chat Model）为例，介绍三种常用的同步调用方法，以及它们对应的异步版本：

这些方法都属于 LangChain Runnable 接口，因此模型、提示词模板和链等对象通常都支持类似的调用方式。

invoke：单次调用

invoke 是最常用的调用方法。它采用同步阻塞方式：程序将输入发送给模型后，会等待模型生成完毕，再一次性返回完整结果。

准备输入 → 调用模型 → 等待模型生成 → 返回完整响应

基本语法：

response = model.invoke(input, config=None)

常用参数：

不同模型集成还可能支持 stop、temperature 等额外参数，具体以对应供应商的集成文档为准。

初始化模型

下面使用兼容 OpenAI 接口的 DeepSeek 模型进行演示。后文将省略初始化代码，直接使用这里创建的 model。

import os

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model

load_dotenv()

model = init_chat_model(
    model="deepseek-v4-flash",
    model_provider="openai",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_BASE_URL"),
)

response = model.invoke("翻译成英文：今天天气真不错！")
print(response.text)

调用结果：

The weather is really nice today!

说明

model_provider="openai" 表示使用 OpenAI 兼容接口，并不表示实际调用的是 OpenAI 模型；最终使用哪个服务由 base_url 和模型名称共同决定。

输入格式

聊天模型常用的输入格式包括纯文本、字典消息列表、元组消息列表和消息对象列表。

1. 纯文本

直接传入字符串，适合一次性的独立问题或指令：

response = model.invoke("用一句话解释什么是大语言模型。")

纯文本输入不会自动携带之前的对话历史。如果需要多轮上下文，应在每次调用时显式传入完整的消息列表，或另外使用聊天历史管理机制。

2. 字典消息列表

每条消息通常包含 role（角色）和 content（内容）两个字段。把多条消息组成列表后，可以向模型提供系统指令和对话历史。

messages = [
    {"role": "system", "content": "你是一名专业的中英文翻译。"},
    {"role": "user", "content": "翻译成英文：今天天气真不错！"},
    {"role": "assistant", "content": "The weather is really nice today!"},
    {"role": "user", "content": "我刚才让你做了什么？"},
]

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

调用结果：

你刚才让我把“今天天气真不错！”翻译成英文。

常见角色如下：

• system：设置模型的行为、角色和规则。
• user：表示用户输入。
• assistant：表示模型之前的回复。
• tool：表示工具执行结果，通常在工具调用流程中使用。

字典格式建议使用供应商通用的 system、user、assistant 和 tool 角色名称。LangChain 中的 HumanMessage、AIMessage 是对应的消息对象类型，不应简单理解为所有供应商都接受 human、ai 作为字典中的角色值。

重要

模型之所以能回答“我刚才让你做了什么”，是因为本次调用显式传入了完整的对话历史，而不是因为 invoke 自动保存了记忆。

3. 元组消息列表

消息也可以写成 (role, content) 元组，从而省略字段名：

messages = [
    ("system", "你是一名专业的中英文翻译。"),
    ("user", "翻译成英文：今天天气真不错！"),
    ("assistant", "The weather is really nice today!"),
    ("user", "我刚才让你做了什么？"),
]

response = model.invoke(messages)

4. 消息对象列表

LangChain 提供了与不同角色对应的消息类，例如 SystemMessage、HumanMessage 和 AIMessage。消息类型由类本身表达，并且可以携带更丰富的元数据。

from langchain.messages import AIMessage, HumanMessage, SystemMessage

messages = [
    SystemMessage("你是一名专业的中英文翻译。"),
    HumanMessage("翻译成英文：今天天气真不错！"),
    AIMessage("The weather is really nice today!"),
    HumanMessage("我刚才让你做了什么？"),
]

response = model.invoke(messages)

消息对象适合需要类型检查、工具调用、多模态内容或消息元数据的场景。

返回值：AIMessage

聊天模型的 invoke 方法通常返回一个 AIMessage 对象，而不是纯字符串。它既包含模型生成的内容，也可以包含 Token 用量、工具调用和供应商元数据等信息。

下面是一份经过格式化的示例输出。实际字段及结构会因模型、供应商和 LangChain 集成版本而异。

AIMessage(
    content="你刚才让我把“今天天气真不错！”翻译成英文。",
    additional_kwargs={"refusal": None},
    response_metadata={
        "token_usage": {
            "completion_tokens": 125,
            "prompt_tokens": 39,
            "total_tokens": 164,
            "completion_tokens_details": {
                "accepted_prediction_tokens": None,
                "audio_tokens": None,
                "reasoning_tokens": 103,
                "rejected_prediction_tokens": None,
            },
            "prompt_tokens_details": {
                "audio_tokens": None,
                "cached_tokens": 0,
            },
            "prompt_cache_hit_tokens": 0,
            "prompt_cache_miss_tokens": 39,
        },
        "model_provider": "openai",
        "model_name": "deepseek-v4-flash",
        "system_fingerprint": "fp_8b330d02d0_prod0820_fp8_kvcache_20260402",
        "id": "7846ae37-1772-46b9-9c0e-80540d510261",
        "finish_reason": "stop",
        "logprobs": None,
    },
    id="lc_run--019f8d0f-a0e1-71e2-85c0-5c3efff192e5-0",
    tool_calls=[],
    invalid_tool_calls=[],
    usage_metadata={
        "input_tokens": 39,
        "output_tokens": 125,
        "total_tokens": 164,
        "input_token_details": {"cache_read": 0},
        "output_token_details": {"reasoning": 103},
    },
)

content、text 与 content_blocks

• content：模型返回的原始内容，可能是字符串，也可能是内容块列表。
• text：从消息中提取出的文本内容，适合只关心文字输出的场景。
• content_blocks：LangChain 标准化后的内容块，可用于统一处理文本、推理、多模态内容等数据。

print(response.content)         # 原始内容
print(response.text)            # 文本内容
print(response.content_blocks)  # 标准化内容块

对于普通文本响应，content 和 text 通常一致；在多模态或包含推理内容的响应中，两者可能不同。

其他常用字段

response_metadata 中经常出现以下信息：

• token_usage：供应商原始的 Token 用量统计。
• model_name：本次实际使用的模型名称。
• finish_reason：生成停止原因，例如正常结束、达到输出长度上限或请求调用工具。
• id：供应商接口返回的响应 ID。
• system_fingerprint：用于识别模型后端配置的指纹；并非所有供应商都提供。
• logprobs：Token 对数概率；只有部分模型和接口支持。

这些字段没有完全统一的结构，不应让跨模型业务逻辑强依赖某个供应商的 response_metadata 格式。

Token 用量

如果需要编写跨模型的 Token 统计逻辑，优先读取 usage_metadata：

usage = response.usage_metadata

if usage:
    print(f"输入 Token：{usage['input_tokens']}")
    print(f"输出 Token：{usage['output_tokens']}")
    print(f"总 Token：{usage['total_tokens']}")

其中：

• input_tokens：输入消耗的 Token 数。
• output_tokens：模型输出消耗的 Token 数。
• total_tokens：本次调用的 Token 总数。
• input_token_details：缓存读取、音频输入等更细粒度的输入统计。
• output_token_details：推理 Token、音频输出等更细粒度的输出统计。

具体的明细字段取决于模型供应商；某些模型可能不返回 Token 用量。

实际开发中的常用读取方式

# 获取文本
print(response.text)

# 获取原始内容
print(response.content)

# 获取工具调用
print(response.tool_calls)

# 获取标准化 Token 用量
print(response.usage_metadata)

# 查看供应商原始响应信息
print(response.response_metadata)

stream：流式调用

stream 会在模型生成内容的过程中持续返回响应片段，而不是等待完整响应生成后再一次性返回。它适合回答较长、需要实时展示内容的场景。

stream 返回一个同步迭代器，其中的每一项通常是 AIMessageChunk：

full_response = None

for chunk in model.stream("写一首七言绝句。"):
    print(chunk.text, end="", flush=True)
    full_response = chunk if full_response is None else full_response + chunk

flush=True 可以让未换行的内容尽快输出到终端，避免暂存在输出缓冲区中。

多个 AIMessageChunk 可以通过 + 合并，得到完整消息：

if full_response is not None:
    print("\n\n完整响应：")
    print(full_response.text)

需要注意：

• stream 的输入格式与 invoke 基本一致，但返回的是消息块迭代器，而不是单个完整的 AIMessage。
• 真正逐块返回内容需要底层模型和集成支持流式输出。如果没有原生流式实现，LangChain 的默认实现可能只返回一个包含完整结果的消息块。
• 流中除了文本块，还可能出现工具调用、推理信息和用量统计等内容，因此生产代码不应假设每个消息块都一定包含非空文本。

batch：批量调用

batch 用于处理一组相互独立的输入。默认情况下，LangChain 会在客户端并发执行多次 invoke，并按照输入顺序返回结果列表。

inputs = [
    "翻译成英文：你好，世界！",
    "5 + 3 等于多少？",
    "中国的首都是哪里？",
]

responses = model.batch(inputs)

for response in responses:
    print(response.text)

调用结果示例：

Hello, world!
5 + 3 等于 8。
中国的首都是北京。

batch 的输入必须是一个列表。列表中每一项的格式与 invoke 的单个输入格式一致，返回值通常是 list[AIMessage]。

限制并发量

请求数量较多时，可以通过 max_concurrency 控制最大并发数，避免触发供应商的速率限制：

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

按完成顺序处理结果

batch 会等待所有请求结束后按输入顺序返回结果。如果希望某个请求一完成就立即处理，可以使用 batch_as_completed：

for index, response in model.batch_as_completed(inputs):
    print(f"第 {index} 个输入：{response.text}")

结果的完成顺序可能与输入顺序不同，因此需要使用返回的索引进行对应。

说明

LangChain 的 batch 默认是客户端并发调用，与模型供应商提供的离线 Batch API 不是同一个概念。它通常可以缩短一组独立请求的总等待时间，但不会自动减少单次请求的 Token 用量或费用；是否有额外的批处理优化，取决于具体模型集成和供应商。

异步调用

三种同步方法都提供了对应的异步版本：

其中，ainvoke 和 abatch 需要使用 await，astream 则需要通过 async for 消费异步迭代器。

下面继续使用前文初始化的 model：

import asyncio


async def main():
    # 单次异步调用
    response = await model.ainvoke("用一句话介绍你自己。")
    print(response.text)

    # 异步流式调用
    async for chunk in model.astream("写一首五言绝句。"):
        print(chunk.text, end="", flush=True)

    print()

    # 异步批量调用
    responses = await model.abatch(
        [
            "1 + 1 等于多少？",
            "法国的首都是哪里？",
        ],
        config={"max_concurrency": 2},
    )

    for item in responses:
        print(item.text)


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

在已经运行事件循环的环境中（例如部分 Jupyter Notebook），可以直接执行：

await main()

异步调用可以避免模型请求阻塞事件循环，适合 Web 服务、高并发任务和异步工作流。底层是否使用原生异步或原生流式能力，仍取决于具体模型集成。

参考资料

• LangChain 官方文档：Models
• LangChain 官方文档：Messages
• LangChain API Reference：BaseChatModel
• LangChain API Reference：Runnable