一、调用方式总览
"模型调用（Invocation）"是指通过特定方法触发大语言模型生成输出的过程。LangChain 提供以下核心调用方法：
二、invoke() —— 阻塞式调用
invoke() 是最核心的方法，工作模式是阻塞式：程序等待模型完整生成响应后，再一次返回结果。
基本语法：
response = model.invoke(input, config=None)参数说明：
三种输入形式
1. 文本输入（最简单）
model = init_chat_model(
model="openai:gpt-5.4-mini",
api_key=CLOSEAI_API_KEY,
base_url=CLOSEAI_BASE_URL,
)
prompt = "翻译成英文：你好世界"
response = model.invoke(prompt)
print(response)✅ 适用快速测试；❌ 缺点是无法设置系统提示（system prompt）和传递对话历史。
2. 字典列表（推荐，最灵活）
messages = [
{"role": "system", "content": "系统提示"},
{"role": "user", "content": "用户消息"},
{"role": "assistant", "content": "AI回复"}, # 可选，用于对话历史
{"role": "user", "content": "继续提问"},
]
response = model.invoke(messages)角色说明：
✅ 可设置系统提示、表达多轮对话历史、JSON 兼容、易于序列化，生产环境推荐。
3. 消息对象列表
将字符串封装为 LangChain 的消息对象（如 SystemMessage、HumanMessage），功能与字典列表等价。
三、stream() —— 流式输出
stream() 以流式方式返回结果，实时返回每个 token，能显著提升长文本生成的用户体验。
for chunk in model.stream("请用一句话介绍你自己"):
print(chunk.content, end="", flush=True)适用场景：聊天机器人、长文本生成等需要实时反馈的交互应用。
异步流式 astream()
import asyncio
async def demo():
stream_resp = model.astream("请用一句话解释机器学习的基本概念。")
async for chunk in stream_resp:
content = chunk.content if hasattr(chunk, "content") else str(chunk)
print(content, end="", flush=True)
asyncio.run(demo())四、batch() —— 批量处理
batch() 用于一次性处理多个输入，适合高并发、大量请求的场景。
messages = [
"你好，你是谁？",
"2 + 3 * 5 = ?",
"中国首都在哪里？",
]
responses = model.batch(messages)
for response in responses:
print(response.content)batch_as_completed() —— 按完成顺序接收
当输入很多或单次调用耗时差异大时，batch_as_completed() 允许收到第一个结果就立即返回（结果可能乱序），每个返回值是 (index, response) 元组：
responses = model.batch_as_completed(messages)
for response in responses:
print(response) # (2, AIMessage(...))性能对比
实测对 4 条输入：
batch() 批量调用：约 1.91 秒；
循环 invoke()：约 3.87 秒；
批量调用可节省约 50.7% 的时间。
五、异步调用
同步 vs 异步
同步（sync）：发起任务后需等待完成才能继续，当前执行流被"阻塞"；
异步（async）：发起任务后不必等待即可继续执行其他任务，执行流"不阻塞"，结果可通过特定方式获取。
异步方法的特点
避免阻塞主线程：同步调用会阻塞程序，异步方法让应用在等待 API 响应时保持响应性；
优化资源利用：更高效地利用系统资源，减少空闲等待时间。
ainvoke() 示例
import asyncio
import time
async def demo_async_invoke():
# 1. 发起异步调用，创建 Task
async_task = asyncio.create_task(model.ainvoke("用一句话解释人工智能。"))
# 2. 并行执行其他任务（不阻塞）
for i in range(3):
await asyncio.sleep(1)
print(f">>> 正在执行第{i+1}个任务...")
# 3. 获取模型结果
response = await async_task
print(f">>> 模型返回: {response.content}")
asyncio.run(demo_async_invoke())abatch() 示例
import asyncio
async def demo_async_batch():
questions = ["用一句话说明深度学习的区别", "中国首都在哪里？"]
# 使用 create_task 让协程立即在后台执行
batch_task = asyncio.create_task(model.abatch(questions))
# 主程序继续执行其他任务
for i in range(3):
await asyncio.sleep(1)
# 获取批量结果
responses = await batch_task
for response in responses:
print(response.content)
asyncio.run(demo_async_batch()
评论区