做 AI 应用开发的朋友都有体会，不同业务场景对大模型的调用方式，要求天差地别。为了适配各类开发需求，帮大家省去重复造轮子的麻烦，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_URL="https://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_provider="openai", configurable_fields=["model", "api_key", "base_url"], config_prefix=prefix, temperature=0.5, max_tokens=200 ) 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, config=config) 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_provider="openai", configurable_fields=["model", "api_key", "base_url"], config_prefix=prefix, temperature=0.5, max_tokens=200 ) 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, config=config) 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_provider="openai", configurable_fields=["model", "api_key", "base_url"], config_prefix=prefix, temperature=0.5, max_tokens=200 ) 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, config=config) 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_provider="openai", configurable_fields=["model", "api_key", "base_url"], config_prefix=prefix, temperature=0.5, max_tokens=200 ) 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, config=config): print(chunk.content, end="", flush=True) 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_URL='https://dashscope.aliyuncs.com/compatible-mode/v1' QWEN_CONSOLE_URL='https://dashscope.console.aliyun.com/' QWEN_MODELS='qwen3.6-plus,qwen-3.5-plus,qwen-3.5-72b-instruct' # ========== 月之暗面 - Kimi ========== KIMI_API_KEY='' KIMI_BASE_URL='https://api.moonshot.cn/v1' KIMI_CONSOLE_URL='https://platform.moonshot.cn/' KIMI_MODELS='moonshot-v1-8k,moonshot-v1-128k,moonshot-v1-256k' # ========== MiniMax ========== MINIMAX_API_KEY='' MINIMAX_BASE_URL='https://api.minimaxi.com/v1' MINIMAX_CONSOLE_URL='https://platform.minimax.chat/' MINIMAX_MODELS='minimax-m2.7,abab-6.5-pro,minimax-m2-her' # ========== 智谱AI-智谱清言 ========== ZHIPU_API_KEY='' ZHIPU_BASE_URL='https://open.bigmodel.cn/api/paas/v4' ZHIPU_CONSOLE_URL='https://open.bigmodel.cn/' ZHIPU_MODELS='glm-4.5-flash,glm-4.5-pro,glm-4-air' # ========== 字节跳动-豆包 ========== DOUBAO_API_KEY='' DOUBAO_BASE_URL='https://ark.cn-beijing.volces.com/api/v3' DOUBAO_CONSOLE_URL='https://console.volcengine.com/ark/' # 重要：豆包不直接使用模型名称，必须替换为火山方舟创建的推理接入点ID DOUBAO_MODELS='doubao-pro-32k,doubao-lite-32k,doubao-pro-128k' # ========== 百度-文心一言 ========== ERNIE_API_KEY='' ERNIE_BASE_URL='https://qianfan.baidubce.com/v2' ERNIE_CONSOLE_URL='https://console.bce.baidu.com/qianfan/' ERNIE_MODELS='ernie-4.0-turbo-8k,ernie-3.5-turbo-128k,ernie-speed-128k' # ========== 腾讯-混元 ========== HUNYUAN_API_KEY='' HUNYUAN_BASE_URL='https://api.hunyuan.cloud.tencent.com/v1' HUNYUAN_CONSOLE_URL='https://console.cloud.tencent.com/hunyuan/' HUNYUAN_MODELS='hunyuan-turbo,hunyuan-pro,hunyuan-lite' # ========== DeepSeek ========== DEEPSEEK_API_KEY='' DEEPSEEK_BASE_URL='https://api.deepseek.com/v1' DEEPSEEK_CONSOLE_URL='https://platform.deepseek.com/' DEEPSEEK_MODELS='deepseek-chat,deepseek-reasoner'
为了便捷、安全地管理.env文件中的配置项，我们专门封装了环境变量处理工具模块，env_util.py的完整源代码如下：
import os from dotenv import load_dotenv, set_key # @老陈说编程 哔哩哔哩、今日头条 class EnvUtil: def __init__(self, env_path=None): 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, override=True)
完成环境配置与工具模块的封装后，就进入了项目的核心实现环节。以下是基于 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, parent=None): 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, parent=None): 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_name=None): 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( model=model_name, model_provider="openai", api_key=config["api_key"], base_url=config["base_url"], temperature=0.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(f"AsyncWorker 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(f"async_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())
代码中用到的图标，放在Icons目录下，下载地址：图标下载。
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。