在使用 LangChain 进行大语言模型开发时，选择合适的调用方法对于用户体验和系统性能至关重要。本博文将详细介绍大模型的两种核心调用方式：invoke 阻塞式调用和 stream 流式调用，帮助你根据实际场景选择最优方案。

大模型调用方法概览

LangChain 为开发者提供了灵活的模型调用接口，主要分为同步和异步两大类：

1. 同步方法调用：

   • invoke - 阻塞式单次调用
   • stream - 流式输出调用
   • batch - 批量处理调用

2. 异步方法调用：

   • ainvoke - 异步阻塞式调用
   • astream - 异步流式调用
   • abatch - 异步批量调用

接下来，我们将重点探讨最常用的 invoke 和 stream 两种方法。

---

方法一：invoke 阻塞式调用

什么是 invoke？

invoke 是 LangChain 与大语言模型交互时的默认调用方式，采用阻塞式/非流式的工作模式。这种方法的核心特点是：系统在后台等待模型完成整个生成过程，然后一次性将完整结果返回给用户。

适用场景

invoke 方法特别适合以下场景：

• 需要获取完整答案后再进行后续处理
• 对响应时间不敏感的批处理任务
• 需要对完整输出进行验证或格式化
• 简单的问答系统或脚本自动化

示例代码

import os
import dotenv
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI

# 加载环境变量
dotenv.load_dotenv()
os.environ['OPENAI_API_KEY'] = os.getenv('OPENAI_API_KEY')
os.environ['OPENAI_BASE_URL'] = os.getenv('OPENAI_BASE_URL')

# 初始化大模型
chat_model = ChatOpenAI(
    model="deepseek-chat",
    temperature=0.7
)

# 创建消息
messages = [HumanMessage(content="你好,请介绍一下自己")]

# 阻塞式调用 LLM
print("正在等待模型响应...")
response = chat_model.invoke(messages)

# 一次性输出完整结果
print(response.content)

优缺点分析

阻塞式 invoke() 有如上优点，但是也存在如下固有缺陷：

• 用户需要等待较长时间才能看到结果
• 对于长文本生成，用户体验较差
• 无法提前中断响应生成

---

方法二：stream 流式调用

什么是 stream？

stream 采用流式输出方式，让模型逐个 token 地实时返回生成内容。用户不再需要等待完整答案，而是能够立即看到模型开始"思考"并输出结果，极大提升了交互体验。

适用场景

stream 方法特别适合以下场景：

• 对话式聊天机器人
• 需要实时反馈的交互应用
• 长文本生成(文章、报告、代码等)
• 提升用户体验的前端应用

示例代码

import os
import dotenv
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()
os.environ['OPENAI_API_KEY'] = os.getenv('OPENAI_API_KEY')
os.environ['OPENAI_BASE_URL'] = os.getenv('OPENAI_BASE_URL')

# 初始化大模型(需要设置 streaming=True)
chat_model = ChatOpenAI(
    model="deepseek-chat",
    streaming=True
)

# 创建消息
messages = [HumanMessage(content="你好,请介绍一下自己")]

# 流式调用 LLM 获取响应
print("开始流式输出:")
for chunk in chat_model.stream(messages):
    # 逐个打印内容块
    print(chunk.content, end="", flush=True)
    # end="" 表示不添加换行符,保持连续输出
    # flush=True 表示立即刷新缓冲区,确保实时显示

print("\n流式输出结束")

代码解析

在 stream 方法中，有几个关键点需要注意:

1. streaming=True：初始化模型时必须设置此参数，启用流式输出功能
2. for chunk in chat_model.stream()：通过迭代器逐个获取生成的内容块
3. end=“”：print 函数的参数，去除默认换行符，让输出连续呈现
4. flush=True：强制刷新输出缓冲区，确保每个 token 立即显示，而非等待缓冲区满。

---

如何选择调用方法？

根据实际应用场景，可以参考以下建议：

选择 invoke 的情况：

• 构建数据处理管道或批量任务
• 需要对完整输出进行格式验证
• 后台服务或定时任务
• API 调用后需要完整结果再返回

选择 stream 的情况：

• 构建聊天机器人或对话系统
• 网页或移动应用的前端交互
• 内容创作工具（写作助手、代码生成等）
• 任何需要提升用户体验的场景

---

进阶技巧：异步调用

对于高并发场景，可以使用异步版本的调用方法：

# 异步阻塞式调用
response = await chat_model.ainvoke(messages)

# 异步流式调用
async for chunk in chat_model.astream(messages):
    print(chunk.content, end="", flush=True)

异步调用可以有效提升系统吞吐量，特别适合需要同时处理多个请求的服务端应用。

---

总结

invoke 和 stream 是大模型调用的两种基础方法，各有优势。invoke 简单稳定，适合批处理和后台任务；stream 交互性强，适合需要实时反馈的应用场景。理解这两种方法的特点和适用场景，能够帮助你构建更优秀的 AI 应用。在实际开发中，根据具体需求灵活选择，甚至可以在同一个应用中结合使用，以达到最佳效果。

---

2025.10.02 吉林·松原