LangChain 6大核心调用方法：invoke/stream/batch同步异步全解析，新手也能轻松学会

选方法就看任务数和返回方式。如果喜欢看视频学习的可以看这个《12. LangChain 6大核心调用方法》喜欢看文章的接着往下看。选同步还是异步最直白的判断标准就是怕不怕「卡住」。如果不怕卡或者说卡了完全不影响 —— 比如本地跑单条脚本、离线处理小批量数据就安安稳稳等模型返回结果再往下走那直接选同步方法invoke/stream/batch就行逻辑简单写着省事不用进行额外处理。但凡你怕卡住卡了会出问题 —— 比如做 Web 后端接口、聊天机器人前端交互不能因为等模型返回就把整个服务、整个页面卡死没法接新请求、响应用户中断操作那就选异步方法ainvoke/astream/abatch。异步的核心就是不阻塞主线程等模型结果的空档照样能处理其他任务完全不会出现 “点一下整个界面都动不了” 的情况。选同步还是异步就看有没有卡住。按如下流程图走一遍流程便可直接选对方法。首先看有多少个任务。如果只有1个任务接着看是否需要逐块显示如果是就在异步环境或UI界面里选astream否则选 stream。如果不需要逐块显示就在异步环境或UI界面里选ainvoke否则选 invoke。如果是多个任务就在异步环境或UI界面里选abatch否则选batch。以下为.env 环境配置文件的内容记得必将其中的 API_KEY 占位值替换为您自行在对应平台申请的有效API密钥。QWEN_API_KEY你的QWEN API KEY QWEN_BASE_URLhttps://dashscope.aliyuncs.com/compatible-mode/v1前面我们已经实操掌握了 invoke 单次调用和 stream 流式输出接下来学习 batch 批量调用。我们先定义多条任务请求虽说传入了多个任务但批量调用的语法和 invoke 完全一致。批量任务执行完毕后通过循环遍历依次输出每一个任务的返回结果。import os from dotenv import load_dotenv from langchain.chat_models import init_chat_model load_dotenv() prefix QWEN llm init_chat_model( model_provideropenai, configurable_fields[model, api_key, base_url], config_prefixprefix, temperature0.5, max_tokens200 ) config { configurable: { f{prefix}_model: os.getenv(f{prefix}_MODEL), f{prefix}_api_key: os.getenv(f{prefix}_API_KEY), f{prefix}_base_url: os.getenv(f{prefix}_BASE_URL) } } prompts [ 生成Java打印Hello World的代码只提供一种最标准的写法, 生成C#打印Hello World的代码只提供一种最标准的写法, 生成Python打印Hello World的代码只提供一种最标准的写法 ] responses llm.batch(prompts, configconfig) for res in responses: print(res.content)熟悉同步 batch 用法后我们再进阶切换到异步abatch。首先定义异步函数 async_batch在函数内部调用大模型的 abatch 方法。最后用 asyncio.run 启动并运行这个异步函数。import os from dotenv import load_dotenv from langchain.chat_models import init_chat_model import asyncio load_dotenv() prefix QWEN llm init_chat_model( model_provideropenai, configurable_fields[model, api_key, base_url], config_prefixprefix, temperature0.5, max_tokens200 ) config { configurable: { f{prefix}_model: os.getenv(f{prefix}_MODEL), f{prefix}_api_key: os.getenv(f{prefix}_API_KEY), f{prefix}_base_url: os.getenv(f{prefix}_BASE_URL) } } prompts [ 生成Java打印Hello World的代码只提供一种最标准的写法, 生成C#打印Hello World的代码只提供一种最标准的写法, 生成Python打印Hello World的代码只提供一种最标准的写法 ] async def async_batch(): responses await llm.abatch(prompts, configconfig) for resp in responses: print(resp.content) asyncio.run(async_batch())接下来我们来学习异步单次调用方法 ainvoke。我们把函数名修改一下内部通过ainvoke传入单个提示词和配置参数等待异步请求完成后直接打印返回内容。同样使用 asyncio.run 运行函数就完成了大模型的异步单次调用。import os from dotenv import load_dotenv from langchain.chat_models import init_chat_model import asyncio load_dotenv() prefix QWEN llm init_chat_model( model_provideropenai, configurable_fields[model, api_key, base_url], config_prefixprefix, temperature0.5, max_tokens200 ) config { configurable: { f{prefix}_model: os.getenv(f{prefix}_MODEL), f{prefix}_api_key: os.getenv(f{prefix}_API_KEY), f{prefix}_base_url: os.getenv(f{prefix}_BASE_URL) } } prompt 生成Python打印Hello World的代码只提供一种最标准的写法 async def async_invoke(): response await llm.ainvoke(prompt, configconfig) print(response.content) asyncio.run(async_invoke())继续进阶学习异步流式输出 astream 方法。提示词保持不变重新定义异步函数 async_stream。它不会等待完整结果返回而是通过 async for循环监听llm.astream的实时数据流。大模型每生成一段内容片段就立即打印输出设置end不换行、flush强制刷新输出实现逐字打字机效果。最后依旧用 asyncio.run 运行函数就能看到实时流式输出的效果。import os from dotenv import load_dotenv from langchain.chat_models import init_chat_model import asyncio load_dotenv() prefix QWEN llm init_chat_model( model_provideropenai, configurable_fields[model, api_key, base_url], config_prefixprefix, temperature0.5, max_tokens200 ) config { configurable: { f{prefix}_model: os.getenv(f{prefix}_MODEL), f{prefix}_api_key: os.getenv(f{prefix}_API_KEY), f{prefix}_base_url: os.getenv(f{prefix}_BASE_URL) } } prompt 生成Python打印Hello World的代码只提供一种最标准的写法 async def async_stream(): async for chunk in llm.astream(prompt, configconfig): print(chunk.content, end, flushTrue) asyncio.run(async_stream())接下来我们将通过一个完整的实战项目带你将astream流式响应功能集成到 PySide6 搭建的UI界面中。本项目结构清晰只有3 个文件分别为环境变量配置文件 .env、环境变量读写工具模块 env_util.py以及 PySide6 界面交互与 astream 流式功能集成的核心业务文件 chat_page.py。在正式编写代码、运行项目前你需要先通过 pip 命令安装项目所需的全部依赖开发包具体安装指令如下pip install pyside6 dotenv langchain langchain-openai完成依赖安装后我们先从项目的基础环境配置入手.env 环境变量配置文件的完整内容如下# 阿里巴巴-通义千问 QWEN_API_KEY你的QWEN API KEY QWEN_BASE_URLhttps://dashscope.aliyuncs.com/compatible-mode/v1 QWEN_CONSOLE_URLhttps://dashscope.console.aliyun.com/ QWEN_MODELSqwen3.6-plus,qwen-3.5-plus,qwen-3.5-72b-instruct # 月之暗面 - Kimi KIMI_API_KEY KIMI_BASE_URLhttps://api.moonshot.cn/v1 KIMI_CONSOLE_URLhttps://platform.moonshot.cn/ KIMI_MODELSmoonshot-v1-8k,moonshot-v1-128k,moonshot-v1-256k # MiniMax MINIMAX_API_KEY MINIMAX_BASE_URLhttps://api.minimaxi.com/v1 MINIMAX_CONSOLE_URLhttps://platform.minimax.chat/ MINIMAX_MODELSminimax-m2.7,abab-6.5-pro,minimax-m2-her # 智谱AI-智谱清言 ZHIPU_API_KEY ZHIPU_BASE_URLhttps://open.bigmodel.cn/api/paas/v4 ZHIPU_CONSOLE_URLhttps://open.bigmodel.cn/ ZHIPU_MODELSglm-4.5-flash,glm-4.5-pro,glm-4-air # 字节跳动-豆包 DOUBAO_API_KEY DOUBAO_BASE_URLhttps://ark.cn-beijing.volces.com/api/v3 DOUBAO_CONSOLE_URLhttps://console.volcengine.com/ark/ # 重要豆包不直接使用模型名称必须替换为火山方舟创建的推理接入点ID DOUBAO_MODELSdoubao-pro-32k,doubao-lite-32k,doubao-pro-128k # 百度-文心一言 ERNIE_API_KEY ERNIE_BASE_URLhttps://qianfan.baidubce.com/v2 ERNIE_CONSOLE_URLhttps://console.bce.baidu.com/qianfan/ ERNIE_MODELSernie-4.0-turbo-8k,ernie-3.5-turbo-128k,ernie-speed-128k # 腾讯-混元 HUNYUAN_API_KEY HUNYUAN_BASE_URLhttps://api.hunyuan.cloud.tencent.com/v1 HUNYUAN_CONSOLE_URLhttps://console.cloud.tencent.com/hunyuan/ HUNYUAN_MODELShunyuan-turbo,hunyuan-pro,hunyuan-lite # DeepSeek DEEPSEEK_API_KEY DEEPSEEK_BASE_URLhttps://api.deepseek.com/v1 DEEPSEEK_CONSOLE_URLhttps://platform.deepseek.com/ DEEPSEEK_MODELSdeepseek-chat,deepseek-reasoner为了便捷、安全地管理.env文件中的配置项我们专门封装了环境变量处理工具模块env_util.py的完整源代码如下import os from dotenv import load_dotenv, set_key # 老陈说编程 哔哩哔哩、今日头条 class EnvUtil: def __init__(self, env_pathNone): if env_path is None: self.env_path os.path.join(os.path.dirname(os.path.abspath(__file__)), .env) else: self.env_path env_path load_dotenv(self.env_path) def load_config(self, providers): config {} for provider in providers: api_key os.getenv(f{provider}_API_KEY, ) base_url os.getenv(f{provider}_BASE_URL, ) console_url os.getenv(f{provider}_CONSOLE_URL, ) models_str os.getenv(f{provider}_MODELS, ) models [m.strip() for m in models_str.split(,)] if models_str else [] config[provider] { api_key: api_key, base_url: base_url, console_url: console_url, models: models } return config def save_config(self, config): if not os.path.exists(self.env_path): return for provider, config_item in config.items(): set_key(self.env_path, f{provider}_API_KEY, config_item[api_key]) set_key(self.env_path, f{provider}_BASE_URL, config_item[base_url]) set_key(self.env_path, f{provider}_MODELS, ,.join(config_item[models])) load_dotenv(self.env_path, overrideTrue)完成环境配置与工具模块的封装后就进入了项目的核心实现环节。以下是基于 PySide6 实现 UI 界面渲染、并完整集成 astream 流式响应能力的chat_page.py 完整源代码import sys import os from PySide6.QtCore import Qt, QSize, Signal, QThread, QTimer from PySide6.QtWidgets import QWidget, QVBoxLayout, QHBoxLayout, QTextEdit, QTextBrowser, QFrame, QSizePolicy, QComboBox, QApplication, QPushButton from PySide6.QtGui import QIcon import asyncio from env_util import EnvUtil # 老陈说编程 哔哩哔哩、今日头条 class IconButton(QPushButton): def __init__(self, icon_name, parentNone): super().__init__(parent) self.icon_name icon_name self.setFixedSize(40, 40) self.setStyleSheet( QPushButton { border: none; background-color: transparent; } QPushButton:hover { background-color: rgba(0, 0, 0, 0.1); border-radius: 8px; } ) self.set_icon(icon_name) def set_icon(self, icon_name): self.icon_name icon_name icon_path os.path.join(os.path.dirname(os.path.abspath(__file__)), Icons, icon_name) if os.path.exists(icon_path): self.setIcon(QIcon(icon_path)) self.setIconSize(QSize(24, 24)) class ChatPage(QWidget): def __init__(self, parentNone): super().__init__(parent) self.current_output self.worker None self.thinking_timer QTimer() self.thinking_timer.timeout.connect(self.update_thinking) self.dot_count 0 self.thinking_prefix self.is_thinking False self.env_util EnvUtil() self.model_configs {} self.all_models [] self.load_models_from_env() layout QVBoxLayout(self) layout.setContentsMargins(0, 0, 0, 0) layout.setSpacing(0) self.model_combo QComboBox() self.model_combo.addItems(self.all_models) self.model_combo.setFixedWidth(180) self.apply_combo_style(self.model_combo) self.model_combo.currentTextChanged.connect(self.on_model_changed) self.current_model self.model_combo.currentText() chat_area QWidget() chat_layout QVBoxLayout(chat_area) chat_layout.setContentsMargins(40, 20, 40, 0) chat_layout.setSpacing(0) output_wrapper QWidget() output_wrapper_layout QHBoxLayout(output_wrapper) output_wrapper_layout.setContentsMargins(0, 0, 0, 0) output_wrapper_layout.addStretch() self.output_text QTextBrowser() self.output_text.setMinimumWidth(820) self.output_text.setMaximumWidth(1040) self.output_text.setHorizontalScrollBarPolicy(Qt.ScrollBarAlwaysOff) self.output_text.setVerticalScrollBarPolicy(Qt.ScrollBarAlwaysOff) self.output_text.setStyleSheet( QTextBrowser { background-color: #ffffff; border: none; padding: 12px; font-size: 14px; color: #333333; } ) output_wrapper_layout.addWidget(self.output_text) output_wrapper_layout.addStretch() chat_layout.addWidget(output_wrapper, 1) layout.addWidget(chat_area, 1) bottom_area QWidget() bottom_area.setStyleSheet(background-color: #ffffff;) bottom_layout QVBoxLayout(bottom_area) bottom_layout.setContentsMargins(40, 0, 40, 20) bottom_layout.setSpacing(0) input_frame QFrame() input_frame.setSizePolicy(QSizePolicy.Preferred, QSizePolicy.Fixed) input_frame.setMinimumWidth(800) input_frame.setMaximumWidth(1000) self.input_frame input_frame self.apply_input_frame_style() frame_layout QVBoxLayout(input_frame) frame_layout.setContentsMargins(16, 12, 16, 12) frame_layout.setSpacing(4) self.input_text QTextEdit() self.input_text.setPlaceholderText(请输入消息...) self.input_text.setMaximumHeight(60) self.input_text.setMinimumHeight(20) self.input_text.setHorizontalScrollBarPolicy(Qt.ScrollBarAlwaysOff) self.input_text.setVerticalScrollBarPolicy(Qt.ScrollBarAlwaysOff) self.apply_input_text_style() bottom_bar QWidget() bottom_bar_layout QHBoxLayout(bottom_bar) bottom_bar_layout.setContentsMargins(0, 0, 0, 0) bottom_bar_layout.addWidget(self.model_combo) bottom_bar_layout.addSpacing(8) bottom_bar_layout.addStretch() self.clear_btn IconButton(clear.png) self.clear_btn.setFixedSize(40, 40) self.clear_btn.hide() self.clear_btn.clicked.connect(self.clear_input) self.send_btn IconButton(send_default.png) self.send_btn.setFixedSize(40, 40) self.send_btn.clicked.connect(self.on_send_clicked) bottom_bar_layout.addWidget(self.clear_btn) bottom_bar_layout.addWidget(self.send_btn) frame_layout.addWidget(self.input_text) frame_layout.addWidget(bottom_bar) center_layout QHBoxLayout() center_layout.addStretch() center_layout.addWidget(input_frame) center_layout.addStretch() bottom_layout.addLayout(center_layout) layout.addWidget(bottom_area) self.setStyleSheet(background-color: #ffffff;) self.input_text.installEventFilter(self) self.input_text.textChanged.connect(self.on_text_changed) self.init_chat_model() def closeEvent(self, event): # 窗口关闭时清理线程 if self.worker and self.worker.isRunning(): self.worker.stop() self.worker.wait() self.thinking_timer.stop() event.accept() def apply_input_frame_style(self): self.input_frame.setStyleSheet( QFrame { border: 1px solid #cccccc; border-radius: 16px; } ) def apply_input_text_style(self): self.input_text.setStyleSheet( QTextEdit { background-color: transparent; border: none; font-size: 14px; color: #333333; padding: 0; } QTextEdit:focus { outline: none; } ) def apply_combo_style(self, combo): combo.setStyleSheet( QComboBox { background-color: #f5f5f5; color: #333333; border: 1px solid #cccccc; border-radius: 4px; padding: 4px 8px; } QComboBox::drop-down { border: none; width: 20px; } QComboBox::down-arrow { width: 12px; height: 12px; } QComboBox QAbstractItemView { background-color: #ffffff; color: #333333; selection-background-color: #0078d4; selection-color: white; } ) def clear_input(self): self.input_text.clear() def on_text_changed(self): has_text len(self.input_text.toPlainText().strip()) 0 if has_text: self.clear_btn.show() self.send_btn.set_icon(send.png) else: self.clear_btn.hide() self.send_btn.set_icon(send_default.png) def eventFilter(self, obj, event): if obj self.input_text and event.type() event.Type.KeyPress: if event.key() Qt.Key.Key_Return or event.key() Qt.Key.Key_Enter: if not event.modifiers() Qt.KeyboardModifier.ShiftModifier: self.on_send_clicked() return True return super().eventFilter(obj, event) def scroll_to_bottom(self): cursor self.output_text.textCursor() cursor.movePosition(cursor.MoveOperation.End) self.output_text.setTextCursor(cursor) self.output_text.ensureCursorVisible() def clear_chat(self): self.current_output self.is_thinking False self.dot_count 0 self.thinking_prefix self.output_text.clear() self.input_text.clear() self.send_btn.setEnabled(True) self.input_text.setEnabled(True) self.worker None self.thinking_timer.stop() def load_models_from_env(self): providers [QWEN, KIMI, MINIMAX, ZHIPU, DOUBAO, ERNIE, HUNYUAN, DEEPSEEK] config self.env_util.load_config(providers) for provider, provider_config in config.items(): for model in provider_config[models]: self.model_configs[model] { api_key: provider_config[api_key], base_url: provider_config[base_url], provider: provider.lower() } self.all_models.append(model) def on_model_changed(self, model_name): if model_name in self.model_configs: self.init_chat_model(model_name) def init_chat_model(self, model_nameNone): from langchain.chat_models import init_chat_model if model_name is None: model_name self.model_combo.currentText() if model_name in self.model_configs: config self.model_configs[model_name] self.model init_chat_model( modelmodel_name, model_provideropenai, api_keyconfig[api_key], base_urlconfig[base_url], temperature0.5 ) def update_thinking(self): self.dot_count (self.dot_count 1) % 4 dots . * self.dot_count self.thinking_prefix 思考中 dots if \n\n in self.current_output: temp_output self.current_output.rsplit(\n\n, 1)[0] else: temp_output self.current_output self.output_text.setPlainText(temp_output \n\n self.thinking_prefix) self.scroll_to_bottom() def on_send_clicked(self): user_input self.input_text.toPlainText().strip() if not user_input or self.worker: return self.current_model self.model_combo.currentText() if self.current_output: self.current_output \n\n self.current_output user_input self.output_text.setPlainText(self.current_output) self.scroll_to_bottom() self.current_output \n\n思考中 self.thinking_prefix 思考中 self.output_text.setPlainText(self.current_output) self.scroll_to_bottom() self.input_text.clear() self.send_btn.setEnabled(False) self.input_text.setEnabled(False) self.dot_count 0 self.is_thinking True self.thinking_timer.start(500) self.worker AsyncWorker(self.model, user_input) self.worker.text_received.connect(self.on_text_received) self.worker.finished.connect(self.on_finished) self.worker.start() def on_text_received(self, ai_text): if self.thinking_timer.isActive(): self.thinking_timer.stop() if self.is_thinking: self.is_thinking False self.current_output self.current_output.rsplit(\n\n, 1)[0] self.current_output \n\n new_output self.current_output ai_text current_text self.output_text.toPlainText() if current_text ! new_output: self.output_text.setPlainText(new_output) self.scroll_to_bottom() def on_finished(self): self.thinking_timer.stop() self.is_thinking False self.send_btn.setEnabled(True) self.input_text.setEnabled(True) self.worker None class AsyncWorker(QThread): text_received Signal(str) finished Signal() def __init__(self, model, user_input): super().__init__() self.model model self.user_input user_input self.full_text self.last_emit_time 0 self._is_running True def stop(self): self._is_running False def run(self): import time import asyncio try: asyncio.run(self.async_run()) except Exception as e: print(fAsyncWorker error: {e}) finally: self.finished.emit() async def async_run(self): import time try: async for chunk in self.model.astream(self.user_input): if not self._is_running: break if chunk.content: self.full_text chunk.content current_time time.time() if current_time - self.last_emit_time 0.033: self.text_received.emit(self.full_text) self.last_emit_time current_time if self._is_running: self.text_received.emit(self.full_text) except Exception as e: print(fasync_run error: {e}) if self._is_running: self.text_received.emit(self.full_text) if __name__ __main__: app QApplication(sys.argv) chat_page ChatPage() chat_page.setWindowTitle(AI 聊天) chat_page.resize(1200, 800) chat_page.show() sys.exit(app.exec())