同一段模型能力，为什么 LangChain 要提供 invoke()、stream()、batch()，甚至还有一组异步版本？

因为它们解决的不是「模型会不会回答」，而是「结果以什么方式交付给用户和业务系统」。选错之后，聊天界面会一直空白，批处理会把结果配错输入，对话会突然失忆，流式半截内容甚至会被直接写进数据库。

本文用一个客服问答服务贯穿三个方法。读完后，你应该能判断一次请求应该等待完整结果、逐块展示，还是与其他独立请求一起处理，并知道消息历史和返回对象为什么不能被忽略。

先用交付方式做选择

方法	返回什么	适用场景	最大误区
invoke()	一条完整 AIMessage	单次问答、提取、普通接口	以为会自动记住历史
stream()	多个 AIMessageChunk	聊天界面、长文本展示	把 Chunk 当完整结果
batch()	多条完整结果	独立任务批处理	以为必然更快

先建立一个可复用的模型实例

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


load_dotenv()

model = init_chat_model(
    "openai:gpt-5.5",
    api_key=os.environ["OPENAI_API_KEY"],
    temperature=0.2,
    timeout=30,
    max_retries=2,
)

示例把模型实例放在请求函数外部，避免每个请求重复创建客户端。生产服务还应根据部署方式管理连接池、密钥和模型生命周期。

invoke, 适合需要一个完整答案的场景

response = model.invoke("用两句话解释订单状态中的已签收。")
print(response.content)

这很简单，但字符串输入只适合单轮任务。对话服务应该显式传递消息角色和历史。

conversation = [
    {"role": "system", "content": "你是订单支持助手，只回答已提供的信息。"},
    {"role": "user", "content": "订单 A-100 的状态是已签收，这代表什么?"},
]

first = model.invoke(conversation)
conversation.append({"role": "assistant", "content": first.content})
conversation.append({"role": "user", "content": "我刚才问的是哪张订单?"})

second = model.invoke(conversation)
print(second.content)

模型不会自动保存上一轮对话。保存多少历史、何时摘要、如何隔离不同用户、如何脱敏，都是应用层的责任。历史无限增长还会带来上下文挤占、成本上升和隐私风险。

不要只拿 content, 先认识 AIMessage

invoke() 返回的通常是 AIMessage。文本在 content，但模型调用的其他信息对于生产服务同样重要。

response = model.invoke("解释什么是检索增强生成。")

print(response.content)
print(response.usage_metadata)
print(response.response_metadata.get("finish_reason"))
print(response.tool_calls)

不同服务商提供的元数据并不完全相同。读取 token、模型名和延迟字段时使用 get()，不要假设每一个响应都有同样的结构。业务逻辑尤其不应依赖某个服务商私有字段。

stream, 既要实时展示，也要聚合完整结果

聊天界面中，用户不愿意盯着空白屏幕等待长答案。stream() 返回一系列 AIMessageChunk，每个 Chunk 包含部分文本或其他内容块。

full = None

for chunk in model.stream("解释 RAG 的完整工作流程，控制在五句话内。"):
    if chunk.text:
        print(chunk.text, end="", flush=True)
    full = chunk if full is None else full + chunk

print("\n最终完整文本", full.text)

这里有两个不同的消费者。界面消费 Chunk，获得即时反馈。业务存储消费完整消息，确保不会把半句话、半个 JSON 或未完成工具调用误当最终答案。

如果下游必须校验 JSON、创建工单或调用另一个系统，应在流结束后再做。流式体验不应破坏数据契约。

另外，流式返回不等于能展示完整推理过程。服务商是否发送推理内容、产品是否允许展示、内容是否应被记录，必须按模型和业务边界单独处理。

batch, 处理独立任务，但要控制并发和顺序

每天导入一千封邮件分类、把一组商品描述翻译成多语言、批量提取文档字段，这些任务彼此独立，适合 batch()。

inputs = [
    "将这句话归类为退款、物流、咨询或其他，包裹一直没到。",
    "将这句话归类为退款、物流、咨询或其他，我想了解会员权益。",
    "将这句话归类为退款、物流、咨询或其他，我想取消昨天的订单。",
]

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

for source, response in zip(inputs, responses, strict=True):
    print(source, "=>", response.content)

batch() 可能通过并行处理提高吞吐，但并不保证一定比串行快。服务商限流、网络、模型队列、当前并发和集成实现都会影响结果。必须在真实配额下压测，并限制 max_concurrency。

当希望优先处理先完成的任务时，可使用 batch_as_completed()。结果会乱序返回，必须使用返回的索引重新关联输入。

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

三个容易踩到的坑

对话为什么失忆

第二轮请求没有携带第一轮消息。修复方式不是换更大模型，而是把会话状态存到应用层，并设计长度控制和摘要策略。

流式 JSON 为什么经常解析失败

因为单个 Chunk 只是结果的一部分。先聚合完整内容，再校验 Schema。若业务需要真正的增量结构化事件，应使用专门协议，而不是对半截 JSON 调 json.loads()。

batch 为什么触发 429

因为批量调用提高的是并发机会，不是服务商配额。降低 max_concurrency，为可重试错误增加退避，并把不可重试错误单独报告。

选择时再问四个问题

1. 用户是否需要立刻看到过程，还是只需要完整结果。
2. 输入之间是否真正独立，是否允许乱序完成。
3. 下游能否处理不完整内容，还是必须等待完整 Schema。
4. 会话状态、失败重试和用户取消由谁负责。

调用方法不是能力等级。它们是把同一模型能力交付给用户、界面和后台任务的不同方式。先把消息状态和输出契约设计好，invoke、stream、batch 就不再只是需要背诵的 API 名称。

延伸阅读

• LangChain model invocation
• LangChain Messages
• LangChain Streaming