AI项目博客模型建好之后，就是调用。LangChain 提供六种核心调用方法，区别主要在「等不等结果」和「一次发几个」：
| 方法 | 特点 | 适用场景 |
|---|---|---|
| invoke() | 阻塞式，一次性返回完整结果 | 问答、批处理任务、无需实时反馈 |
| ainvoke() | 非阻塞式，提高系统吞吐量 | 高并发 Web 应用、IO 密集型任务 |
| stream() | 流式输出，实时返回每个 token | 聊天机器人、长文本生成、重视体验的交互 |
| astream() | 非阻塞的流式输出 | 高并发场景下的流式输出 |
| batch() | 批量处理多个输入 | 需要同时处理大量请求 |
| abatch() | 非阻塞的批量处理 | 高并发批量任务 |
10.1 invoke()
invoke 是最核心的方法，工作模式是阻塞式的：程序等待模型完全生成整个响应后，再一次性返回结果。它做三件事：接收输入（问题、指令、对话历史）→ 发送给 LLM → 返回响应（文本加元数据）。
response = model.invoke(input, config=None)| 参数 | 类型 | 说明 | 必需 | 默认值 |
|---|---|---|---|---|
| input | str / list[dict] / list[Message] 等 | 要发送给模型的内容 | 必需 | 无 |
| config | dict | 高级配置（回调函数、元数据、标签等） | 可选 | None |
10.1.1 三种输入形式
1、文本输入（最简单）。直接传一个问题或指令，适合快速测试，不需要保留对话历史的简单生成任务。缺点是无法设置系统提示，也无法传递对话历史。
prompt = "翻译成英文：你好世界"
response = model.invoke(prompt)
print(response)2、字典列表（推荐，最灵活）。用 role 和 content 组成消息列表，可以设置系统提示、表达多轮对话历史，JSON 兼容、易于序列化和网络传输，生产环境推荐。
messages = [
{"role": "system", "content": "系统提示"},
{"role": "user", "content": "用户消息"},
{"role": "assistant", "content": "AI 回复"}, # 可选，用于对话历史
{"role": "user", "content": "继续提问"},
]
response = model.invoke(messages)| 角色 | 英文 | 作用 | 示例 |
|---|---|---|---|
| system | System | 设定 AI 的行为、角色、规则 | 「你是一个专业的 Python 导师」 |
| user | Human/User | 用户的输入或问题 | 「什么是装饰器？」 |
| assistant | AI/Assistant | AI 的历史回复，用于对话上下文 | 「装饰器是一种设计模式……」 |
user 和 human 有时可以互换，但遵循主流模型提供商（如 OpenAI）的惯例，使用 user 最稳妥。
from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os
load_dotenv(override=True)
model = init_chat_model(
model="gpt-5.4-mini",
model_provider="openai",
api_key=os.getenv("CLOSEAI_API_KEY"),
base_url=os.getenv("CLOSEAI_BASE_URL"),
)
messages = [
{"role": "system", "content": "你是一个专业的数学老师。"},
{"role": "user", "content": "什么是斐波那契数列？"},
]
response = model.invoke(messages)
print(f"AI的回复：{response.content}")3、消息对象列表。使用内置的 SystemMessage、HumanMessage、AIMessage，适合需要类型检查、IDE 自动补全的大型项目；缺点是代码更长、不如字典简洁、难以序列化。
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
messages = [
SystemMessage("你是一个专业的数学老师。"),
HumanMessage("2 + 3 * 2 = ？"),
AIMessage("8"),
HumanMessage("我刚才问什么问题了？"),
]
response = model.invoke(messages)
print(response.content)| 消息类 | 对应字典格式 | 作用 |
|---|---|---|
| SystemMessage | {“role”: “system”, …} | 系统提示 |
| HumanMessage | {“role”: “user”, …} | 用户输入 |
| AIMessage | {“role”: “assistant”, …} | AI 回复 |
10.1.2 不带历史就会「失忆」
第一次问「你好，我叫小明」，第二次问「我叫什么名字」。如果第二次只发新问题，模型会回答「我不知道你的名字」——两个请求之间没有任何记忆。把上一轮的 assistant 回复和本轮问题追加进同一个列表，模型就能答对。
conversation = [
{"role": "system", "content": "你是一个非常友好的AI助手"},
{"role": "user", "content": "你好，我叫小明"},
]
response1 = model.invoke(conversation)
print(f"AI的回复1：{response1.content}")
# 把 AI 回复和下一轮问题追加进历史
conversation.append({"role": "assistant", "content": response1.content})
conversation.append({"role": "user", "content": "我叫什么名字？"})
response2 = model.invoke(conversation)
print(f"AI的回复2：{response2.content}") # 你叫小明。10.1.3 返回值：AIMessage
invoke 返回一个 AIMessage 对象，源码签名如下：
def invoke(
self,
input: LanguageModelInput,
config: RunnableConfig | None = None,
*,
stop: list[str] | None = None,
**kwargs: Any,
) -> AIMessage:AIMessage 的字段可以分成五类：
- 核心内容：
content是模型生成的文本；id是本次运行在 LangChain 内部的唯一标识；additional_kwargs存厂商附加参数，refusal表示拒答原因。 - 消耗统计：
prompt_tokens/input_tokens是输入 Token，completion_tokens/output_tokens是输出 Token，total_tokens是总消耗；推理模型还有reasoning_tokens，命中缓存时有cached_tokens。 - 响应元数据：
model_name、model_provider、finish_reason（stop 正常结束、length 达到长度限制）、system_fingerprint。 - 性能与延迟：
total_duration_ms总耗时、user_visible_ttft_ms用户看到首字的等待时间、engine_ttft_ms引擎首字时间、service_tbt_ms字与字之间的间隔（决定打字机效果是否丝滑）、pre_inference_ms推理前处理耗时。 - 工具调用：
tool_calls是模型决定调用的工具列表，invalid_tool_calls是格式错误的调用尝试。
AIMessage(
content='2 + 3 * 2 = **8**',
additional_kwargs={'refusal': None},
response_metadata={
'token_usage': {
'completion_tokens': 15,
'prompt_tokens': 16,
'total_tokens': 31,
'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0},
'latency_checkpoint': {
'engine_tbt_ms': 4,
'engine_ttft_ms': 36,
'engine_ttlt_ms': 100,
'pre_inference_ms': 86,
'service_tbt_ms': 4,
'service_ttft_ms': 280,
'service_ttlt_ms': 338,
'total_duration_ms': 259,
'user_visible_ttft_ms': 194,
},
},
'model_provider': 'openai',
'model_name': 'gpt-5.4-mini-2026-03-17',
'system_fingerprint': None,
'id': 'chatcmpl-DgWobsxhDOqzjqVFwbZYKRnovpEiV',
'service_tier': 'default',
'finish_reason': 'stop',
'logprobs': None,
},
id='lc_run--019e3659-5ee2-7b62-bc8a-741e27374b43-0',
tool_calls=[],
invalid_tool_calls=[],
usage_metadata={
'input_tokens': 16,
'output_tokens': 15,
'total_tokens': 31,
'input_token_details': {'audio': 0, 'cache_read': 0},
'output_token_details': {'audio': 0, 'reasoning': 0},
},
)按需访问这些字段：
response = model.invoke("用一句话解释什么是 AI")
print("AI 回复:", response.content)
meta = response.response_metadata
print(f"使用的模型: {meta['model_name']}")
print(f"结束原因: {meta['finish_reason']}")
print(f"模型提供商：{meta['model_provider']}")
usage = meta.get('token_usage', {})
print(f"输入 tokens: {usage.get('prompt_tokens')}")
print(f"输出 tokens: {usage.get('completion_tokens')}")
print(f"总计 tokens: {usage.get('total_tokens')}")
print(f"消息 ID: {response.id}")10.2 流式调用 stream()
invoke 在模型输出完成后才一次性返回，输出文本很长时用户体验不好。stream 返回一个迭代器，边生成边返回 chunk，可以逐 token 渲染。注意：流式输出依赖模型供应商对流式的支持。
for chunk in model.stream("写一首七言律诗，总结大模型的发展"):
print(chunk.text, end="", flush=True)输出不再是一整段，而是逐字出现。stream 的优点：响应更快，用户不必等待完整输出；交互更流畅，尤其在长文本或复杂推理场景；可以实时展示模型的生成过程。
10.3 批量调用 batch()
batch 一次发送一组请求，模型在后台并行处理，然后按原始输入顺序返回结果列表。相比逐个调用 invoke，能大幅减少网络往返和等待时间，适合文档摘要、批量问答、数据预处理、多样本分类。
messages = [
"你好，你是谁？",
"2 + 3 * 5 = ?",
"中国首都在哪里？",
]
responses = model.batch(messages)
for response in responses:
print(response)如果输入列表很大，或单个请求耗时差异明显，用 batch_as_completed()：谁先完成就先返回，结果可能乱序。每个元素是 (原始输入索引, AIMessage) 元组，可以按索引重新排序。
for index, response in model.batch_as_completed(messages):
print(index, response.content)性能对比：4 条翻译请求，batch 耗时约 1.91 秒，循环 invoke 耗时约 3.87 秒，批量调用节省约 50.7%。
import time
inputs = [
"翻译成英文：春天来了",
"翻译成英文：夏天很热",
"翻译成英文：秋天落叶",
"翻译成英文：冬天下雪",
]
start = time.time()
responses = model.batch(inputs)
batch_time = time.time() - start
print(f"批量调用耗时: {batch_time:.2f}秒")10.4 异步调用
同步：发起任务后必须等待任务完成，才能继续执行后续任务，当前执行流会被阻塞。异步：发起任务后不必等待，可以继续执行其他任务；任务完成后仍可通过特定方式获取结果，执行流不会被阻塞。
LangChain 的异步方法 ainvoke、astream、abatch 与同步版本一一对应，价值在于避免阻塞主线程、提高资源利用率。以下示例建议放在 .py 文件中执行，不要在 Jupyter 里跑。
ainvoke()：
import asyncio
import time
from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os
load_dotenv(override=True)
model = init_chat_model(
model="openai:gpt-5.4-mini",
api_key=os.getenv("CLOSEAI_API_KEY"),
base_url=os.getenv("CLOSEAI_BASE_URL"),
)
async def demo_async_invoke():
start = time.perf_counter()
print(">>> 发起异步模型调用 (ainvoke)...")
task = asyncio.create_task(model.ainvoke("用一句话解释人工智能。"))
for i in range(3):
await asyncio.sleep(1)
print(f">>> 正在执行第{i + 1}个任务... (已耗时 {time.perf_counter() - start:.2f}s)")
response = await task
print(f">>> 模型返回: {response.content}")
print(f"=== 总运行耗时: {time.perf_counter() - start:.2f}s ===")
asyncio.run(demo_async_invoke())三个本地任务各 1 秒，模型请求同时在后台执行，总耗时约 3.02 秒；如果串行，模型调用时间还要再叠加。
astream()：
async def demo_async_stream():
start = time.perf_counter()
stream_resp = model.astream("请用一句话解释机器学习的基本概念。")
for i in range(3):
await asyncio.sleep(1)
print(f">>> 正在执行第{i + 1}个任务... (已耗时 {time.perf_counter() - start:.2f}s)")
print(">>> 流式输出: ", end="", flush=True)
async for chunk in stream_resp:
content = chunk.content if hasattr(chunk, "content") else str(chunk)
print(content, end="", flush=True)
print(f"\n=== 总运行耗时: {time.perf_counter() - start:.2f}s ===")
asyncio.run(demo_async_stream())abatch()：
async def demo_async_batch():
start = time.perf_counter()
questions = [
"用一句话说明深度学习与传统机器学习的区别",
"中国首都在哪里？",
]
batch_task = asyncio.create_task(model.abatch(questions))
for i in range(3):
await asyncio.sleep(1)
print(f">>> 正在执行第{i + 1}个任务... (已耗时 {time.perf_counter() - start:.2f}s)")
responses = await batch_task
for response in responses:
print(f">>> 响应内容: {response.content}")
print(f"=== 总运行耗时: {time.perf_counter() - start:.2f}s ===")
asyncio.run(demo_async_batch())10.5 如何处理 API 调用失败
网络、限流、配置错误都可能让调用失败，线上代码至少要用 try-except 包住：
try:
response = model.invoke("Hello")
print(response.content)
except ValueError as e:
print(f"配置错误: {e}")
except ConnectionError as e:
print(f"网络错误: {e}")
except Exception as e:
print(f"未知错误: {e}")10.6 小结
优先用字典列表传消息，能带 system 和历史；长文本用 stream；批量任务用 batch，并可用 max_concurrency 控制并发；高并发服务用异步方法；线上务必加异常处理。
下一篇：LangChain 1.2 入门（十一）：输出美化、profile 与 config；上一篇：Ollama 本地模型部署与调用。
本文根据公开课程课件整理，并结合 LangChain 官方文档核对与补充。
