模型的调用

模型的调用

2026年9月21日·#编程学习/langchain学习笔记LangChain/AI·6027 字 30 分钟
浏览量加载中...
AI 摘要

invoke/stream/batch 三种调用方式与它们的异步版本,invoke 的三种输入形式,以及 AIMessage 返回值结构

六种调用方式总览#

模型调用(Invocation)是指通过特定方法触发大语言模型生成输出的过程。

方法特点适用场景
invoke()阻塞式,一次性返回完整结果问答、批处理任务、无需实时反馈
stream()流式输出,实时返回每个 token聊天机器人、长文本生成、需要提升体验
batch()批量处理多个输入高并发、需要同时处理大量请求
ainvoke()非阻塞式(异步)高并发 Web 应用、IO 密集型任务
astream()非阻塞式流式同上
abatch()非阻塞式批量同上
Tip

记法:前缀 a = async(异步),同步版与异步版一一对应。

invoke():最核心的方法#

阻塞式:程序会等模型完全生成整个响应后,再一次性返回。

response = model.invoke(input, config=None)

它做的事:① 接收你的输入(问题、指令、对话历史)② 发送给 LLM ③ 返回响应(文本 + 元数据)。

参数详解#

参数类型说明必需默认值
inputstr | list[dict] | list[Message]你要发送给模型的内容必需
configdict高级配置(回调函数、元数据、标签等)可选None

invoke 的源码签名(除了 config,还有一个直接传的 stop 停止符参数):

def invoke(
self,
input: LanguageModelInput,
config: RunnableConfig | None = None,
*,
stop: list[str] | None = None,
**kwargs: Any,
) -> AIMessage:
Note

config 属于进阶用法(run_nametagsmetadatacallbacksconfigurable 等),本笔记最后一节专门讲。

三种输入形式#

① 文本输入(最简单)

response = model.invoke("帮我写一首描述春天的七言绝句诗")
print(response.content)

文本会自动转成一条 user 消息。✅ 适合快速测试;❌ 无法设置系统提示、无法传递对话历史。

② 字典列表(推荐,最灵活)

messages = [
{"role": "system", "content": "你是一个专业的 Python 导师"},
{"role": "user", "content": "什么是装饰器?"},
{"role": "assistant", "content": "装饰器是一种设计模式..."}, # 可选,用于对话历史
{"role": "user", "content": "继续提问"},
]
response = model.invoke(messages)

✅ 可以设置系统提示、表达多轮对话历史、JSON 兼容、易于序列化与网络传输(生产环境推荐);❌ 代码稍多一点。

角色说明:

role英文作用
systemSystem设定 AI 的行为、角色、规则(如”你是一个专业的 Python 导师”)
userHuman / User用户的输入/问题
assistantAI / AssistantAI 的历史回复,用于对话上下文

③ 消息对象列表

SystemMessageHumanMessageAIMessage 这些专门的类来构造:

from langchain_core.messages import HumanMessage
response = model.invoke([HumanMessage("2 + 3 * 2 = ?")])
Note

"user""human" 在某些实现里可互换,但跟着你用的大模型提供商的惯例走最稳(如 OpenAI 用 user)。

对话历史:带历史 vs 不带历史#

字典列表的价值在”多轮”上体现得最明显。举例 2:多轮对话(带历史)——把上一轮的问答按 assistant / user 追加进列表,模型就能”记得”:

# 使用字典格式构建消息
messages = [
{"role": "system", "content": "你是一个专业的数学老师。"},
{"role": "user", "content": "2 + 3 * 2 = ?"},
{"role": "assistant", "content": "8"},
{"role": "user", "content": "我刚才问了什么问题?"}
]
response = model.invoke(messages)
print(f"AI的回复:{response.content}")
AI的回复:你刚才问的是:**"2 + 3 * 2 = ?"**

举例 3:如果不传递历史,AI 会”失忆”——两次调用互相独立,第二次问”我叫什么名字”它答不上来:

messages1 = [
{"role": "system", "content": "你是一个非常友好的AI助手"},
{"role": "user", "content": "你好,我叫小明"},
]
# 第一次对话
response1 = model.invoke(messages1)
print(f"AI的回复1:{response1.content}")
# 第二次对话:只发了新问题,没有带上上一轮的消息
messages2 = [
{"role": "user", "content": "我叫什么名字?"}
]
response2 = model.invoke(messages2)
print(f"AI的回复2:{response2.content}")
AI的回复1:你好,小明!很高兴认识你 😊
我是你的AI助手,有什么我可以帮你的吗?
AI的回复2:我不知道你的名字,除非你告诉我。
如果你愿意,可以直接告诉我,我之后就可以这样称呼你。

作为对比:传递记忆——把上一轮的回复 append 进同一个列表再调用,它就记得住了:

conversation = [
{"role": "system", "content": "你是一个非常友好的AI助手"},
{"role": "user", "content": "你好,我叫小明"}
]
# 第一次对话
response1 = model.invoke(conversation)
print(f"AI的回复1:{response1.content}")
# 添加记忆
conversation.append({"role": "assistant", "content": response1.content})
conversation.append({"role": "user", "content": "我叫什么名字?"})
# 第二次对话
response2 = model.invoke(conversation)
print(f"AI的回复2:{response2.content}")
AI的回复1:你好,小明!很高兴认识你 😊
我是你的AI助手,有什么我可以帮你的吗?
AI的回复2:你叫小明。
Important

结论:模型的”记忆”完全来自你传进 invoke 的消息列表,它自己不保存任何上下文。要实现多轮对话,就必须在每次调用时把完整历史一起传过去(消息类型的系统讲解见「Message与对话历史」笔记)。

返回值:AIMessage#

invoke 返回的是 AIMessage 对象(不是纯字符串):

print(type(response)) # <class 'langchain_core.messages.ai.AIMessage'>
print(response.content) # 模型回复的文本

rich 打印能看清它的结构(from rich import print as rprint):

AIMessage(
content='2 + 3 * 2 = **8**',
additional_kwargs={'refusal': None},
response_metadata={
'token_usage': {
'completion_tokens': 15,
'prompt_tokens': 16,
'total_tokens': 31,
...
}, ...
}, ...
)
字段说明
content回复的文本(最常用)
response_metadata元数据:token_usage(输入/输出/总 token)、模型名、结束原因等
additional_kwargs附加参数(不同厂商不同,如 refusal
tool_calls模型要求调用的工具(第 5 章 Tools 会用到)

图:美化打印 AIMessage 的实际效果——content 就是模型回复的文本

返回值字段全解析#

AIMessage 里的字段可以分成五组来看,回复内容、账单、体检报告、追踪 ID 全在里面

① 核心内容与基本信息

字段说明
content模型生成的文本回答,你最关心的核心输出
id本次运行在 LangChain 内部生成的唯一标识符(Run ID),形如 lc_run--019e3659-5ee2-7b62-bc8a-741e27374b43-0
additional_kwargs包含特定供应商的额外参数
refusal模型拒绝回答(涉及敏感政策)时给出拒绝原因,正常回答时为 None

② 消耗统计(Token Usage)——决定你这一行输入操作花了多少钱

字段说明
prompt_tokens / input_tokens输入 Token 数:你发送给模型的问题长度
completion_tokens / output_tokens输出 Token 数:模型回答生成的长度
total_tokens总消耗,两者之和
reasoning_tokens推理 Token 数:o1/o3 这类推理模型”思考”时消耗的 Token
cached_tokens缓存命中的 Token 数:重复提问命中模型商缓存时,这部分费用通常更低

③ 响应元数据(Response Metadata)——API 返回的原始详细信息

字段说明
model_name实际调用的模型具体版本,如 gpt-5.4-mini-2026-03-17
model_provider模型供应商,如 openaideepseek
finish_reason生成停止的原因:stop = 正常回答结束;length = 达到最大 Token 限制被截断(调用工具时还会出现 tool_calls
system_fingerprint系统指纹,用于追踪模型后端的配置变更
idAPI 层面的响应 ID(如 chatcmpl-DgWobsxhDOqzjqVFwbZYKRnovpEiV),和上面那个 LangChain Run ID 不是一回事
service_tier服务层级(如按量付费或订阅)
logprobs对数概率,通常用于分析词汇选择的可能性

④ 性能与延迟(Latency Checkpoint)——针对 API 响应速度的深度拆解,单位毫秒(ms)

字段说明
total_duration_ms总耗时:从请求发出到完全收到的总时间
user_visible_ttft_ms首字到达时间:用户看到第一个字跳出来等待的时间,这是体感快慢的关键
engine_ttft_ms引擎层面的首字到达时间(TTFT = Time To First Token)
engine_ttlt_ms引擎生成最后一个字的时间(TTLT = Time To Last Token)
pre_inference_ms推理前处理耗时:包括安全审核、Token 化等预处理
service_tbt_ms服务端 token 与 token 之间生成的间隔时间(TBT = Time Between Tokens),决定了打字机效果是否丝滑
Note

latency_checkpoint 不是所有供应商都会返回(它来自 OpenAI 兼容的高阶接口);实测 DeepSeek 官方接口的返回里就没有这一项。

⑤ 工具调用信息

字段说明
tool_calls结构化工具调用列表:模型决定调用某个函数或搜索工具时,参数都在这里
invalid_tool_calls触发失败或格式错误的工具调用尝试

一次访问所有信息

response = model.invoke("用一句话解释什么是 AI")
# 1. 获取回复内容
print("AI 回复:", response.content)
# 2. 获取响应元数据
metadata = response.response_metadata
print(f"使用的模型: {metadata['model_name']}")
print(f"结束原因: {metadata['finish_reason']}")
print(f"模型提供商:{metadata['model_provider']}\n")
# 3. 获取 Token 使用情况
usage = metadata.get('token_usage', {})
print(f"输入 tokens: {usage.get('prompt_tokens')}")
print(f"输出 tokens: {usage.get('completion_tokens')}")
print(f"总计 tokens: {usage.get('total_tokens')}")
# 4. 获取消息 ID
print(f"消息 ID: {response.id}")
AI 回复: AI(人工智能)就是让机器模拟人类的学习、推理、识别和决策能力。
使用的模型: gpt-5.4-mini-2026-03-17
结束原因: stop
模型提供商:openai
输入 tokens: 13
输出 tokens: 28
总计 tokens: 41
消息 ID: lc_run--019e3665-8dfa-7c53-a9a5-4995348a0258-0

stream():流式调用#

invoke()stream()
返回时机全部生成完才返回实时返回响应片段
返回值AIMessage迭代器(iterator)
适合短回答、后台任务长文本、聊天机器人(用户体验好)
for chunk in model.stream("请用中文详细介绍什么是AI"):
print(chunk.content, end="", flush=True)
Warning

流式输出依赖模型供应商对流式的支持——不是所有模型/平台都支持。

batch():批量调用#

batch() 允许一次性发送一组独立请求,模型在后台并行处理,然后返回所有结果的列表(按原始输入顺序)。

responses = model.batch([
"请为一款机械键盘写一句广告语",
"请为无线耳机写一句广告语",
"请为显示器写一句广告语",
])
for r in responses:
print(r.content)

适用于高并发场景——比写 for 循环逐个 invoke 快得多(并行)。

batch_as_completed():谁先完成谁先返回#

输入列表很大、或者单个请求耗时差异明显时,batch()”等全部完成才算完”就显得慢了。batch_as_completed()在每个请求完成后立即 yield 结果,所以结果可能乱序;每个响应被放在一个元组里,元组的第一个元素是原始输入的索引,可以据此重新排序:

messages = [
"你好,你是谁?",
"2 + 3 * 5 = ?",
"中国首都在哪里?"
]
responses = model.batch_as_completed(messages)
for response in responses:
print(response)
(2, AIMessage(content='中国的首都是**北京**。', ...))
(0, AIMessage(content='你好!我是一个 AI 助手,可以帮你回答问题、写作、翻译…', ...))
(1, AIMessage(content='2 + 3 * 5 = **17**', ...))

索引 2 最先返回、索引 1 最后返回——谁先算完谁先到,适合”先到先处理”的场景。

batch() 与循环 invoke() 的性能对比#

同一个任务(4 条翻译),分别用 batch() 和 for 循环调用 invoke()

import time
inputs = [
"翻译成英文:春天来了",
"翻译成英文:夏天很热",
"翻译成英文:秋天落叶",
"翻译成英文:冬天下雪"
]
# ✅ 批量调用(高效)
start = time.time()
responses = model.batch(inputs)
batch_time = time.time() - start
print(f"批量调用耗时: {batch_time:.2f}\n")
# ❌ 循环调用(低效,仅用于对比)
start = time.time()
loop_responses = []
for inp in inputs:
response = model.invoke(inp)
loop_responses.append(response)
loop_time = time.time() - start
print(f"循环调用耗时: {loop_time:.2f}秒")
print(f"批量调用节省: {((loop_time - batch_time) / loop_time * 100):.1f}%")

课程环境实测:批量调用 1.91 秒,循环调用 3.87 秒,批量调用节省约 50.7%

Tip

batch() 快在并行发送更少的网络往返开销上;输入条数越多、单条应答越慢,差距越明显。

异步调用:ainvoke / astream / abatch#

异步方法与同步版相比:调用时不会阻塞程序,可以继续执行本地逻辑,等模型返回后再取结果。用 asyncio 组织:

图:同步 vs 异步的直观类比——同步要等对方准备好才能一起走,异步则各自准备、最后会合

import asyncio
async def main():
task = asyncio.create_task(model.ainvoke("用一句话解释人工智能。"))
# 模型在后台请求的同时,本地逻辑继续跑
for i in range(1, 4):
await asyncio.sleep(1)
print(f">>> 正在执行第{i}个任务...")
result = await task
print(">>> 模型返回:", result.content)
asyncio.run(main())

运行结果能直观看到:模型请求已经在后台发送,本地循环照样每秒打印一次,最后才取回模型结果。

同步异步说明
invoke()ainvoke()一次性返回
stream()astream()流式
batch()abatch()批量

异步调用的两个注意事项#

Warning
  1. 异步示例请.py 文件中执行,而不是 Jupyter Notebook 里——Jupyter 自带事件循环,和 asyncio.run() 会打架。
  2. 等待时要用 await asyncio.sleep(),不要用 time.sleep()time.sleep() 会把整个线程阻塞住,后台的网络请求也一起被卡住,异步就白写了。

astream() 与 abatch() 长什么样#

astream() 返回的是异步生成器,用 async for 消费;请求发出后可以不急着读,先干别的:

async def demo_async_stream():
print(">>> 发起异步流式调用 (astream)...")
stream_resp = model.astream("请用一句话解释机器学习的基本概念。")
# 流式请求已发送,程序无需等待,继续执行其他异步任务
for i in range(3):
await asyncio.sleep(1) # 用 asyncio.sleep 而非 time.sleep,事件循环才能去处理上面的网络 IO
print(f">>> 正在执行第{i + 1}个任务...")
# 现在开始读取缓冲区中的流式结果
print(">>> 流式输出: ", end="", flush=True)
async for chunk in stream_resp:
print(chunk.content, end="", flush=True)
print()
asyncio.run(demo_async_stream())

abatch()ainvoke() 一样,用 create_task 丢到后台再 await 取结果:

async def demo_async_batch():
questions = ["用一句话说明深度学习与传统机器学习的区别", "中国首都在哪里?"]
print(">>> 发起异步批量调用 (abatch)...")
batch_task = asyncio.create_task(model.abatch(questions)) # 关键:立刻在后台执行
print(">>> 批量任务已在后台运行,主程序继续执行...")
for i in range(3):
await asyncio.sleep(1) # 用 asyncio.sleep 让后台任务拿到 CPU 时间片做网络请求
print(f">>> 正在执行第{i + 1}个任务...")
responses = await batch_task # 取回批量结果
for response in responses:
print(f">>> 响应内容: {response.content}")
asyncio.run(demo_async_batch())

两次运行的总耗时都稳定在 3.0 秒左右(= 本地循环的 3 秒),说明模型请求确实是和本地任务并行跑的,几乎没有额外等待。

处理 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}")
异常类型典型触发场景处理建议
ValueError配置错误:参数名写错、model_provider 推断失败、密钥缺失或格式不对检查 .env 与初始化参数,改完再重试
ConnectionError网络错误:连不上服务端、代理/魔法失效、base_url 写错检查网络与 base_url,配合 max_retries 自动重试
Exception未知错误:兜底捕获(限流、余额不足、模型侧 5xx 等)打日志、告警,必要时降级到备用模型
Tip

except 的顺序很重要:它是从上往下匹配的,所以具体异常(ValueErrorConnectionError)必须写在 Exception 前面,否则全被兜底那个吃掉,排查问题时就看不到真正原因了。

美化输出:pretty_print()#

直接 print(response) 打出来的内容比较杂乱,AIMessage 自带 pretty_print() 方法可以美化输出:

# 定义消息列表
conversation = [
{"role": "system", "content": "无条件服从用户指令"},
{"role": "user", "content": "我是老王,你是小王"},
{"role": "assistant", "content": "好的老王,我是小王"},
{"role": "user", "content": "你是谁?我是谁?"}
]
response = model.invoke(conversation)
# 美化输出响应
response.pretty_print()
================================== Ai Message ==================================
我是小王,你是老王。
Note

另一种美化方式是 rich 库的 rprint(response)(前面的返回值一节已经用过):它会把 contentresponse_metadatausage_metadata 全部彩色展开,适合在终端里做深度调试;pretty_print() 只呈现”谁说了什么”,更干净。

模型能力画像:profile#

LangChain 1.1 及更高版本可以通过 profile 属性查看模型的配置信息。这是 LangChain 针对模型做的能力画像,但它是否存在,取决于 LangChain 在集成模型厂商的服务时是否声明了能力画像

例 1:DeepSeek 官方模型——输出为空 {}(本机实测结果一致):

from langchain_deepseek import ChatDeepSeek
from dotenv import load_dotenv
# 从.env文件中加载环境变量
load_dotenv(override=True)
model = ChatDeepSeek(
model="deepseek-v4-flash",
temperature=0.7,
max_tokens=1000,
max_retries=6,
)
print(model.profile) # {} —— LangChain没有声明DeepSeek官方模型的能力画像

例 2:CloseAI 平台的 gpt 模型——同样输出为空 {}

例 3:OpenRouter 平台的模型——LangChain 已声明了画像,能看到完整的能力清单:

from langchain_openrouter import ChatOpenRouter
from dotenv import load_dotenv
from rich import print as rprint
# 从.env文件中加载环境变量
load_dotenv(override=True)
model = ChatOpenRouter(
model="openai/gpt-4o-mini",
# model="deepseek/deepseek-v3.2",
temperature=0.7,
timeout=30,
max_tokens=1000,
max_retries=6,
)
rprint(model.profile)
{
'max_input_tokens': 128000,
'max_output_tokens': 16384,
'text_inputs': True,
'image_inputs': True,
'audio_inputs': False,
'video_inputs': False,
'text_outputs': True,
'image_outputs': False,
'audio_outputs': False,
'video_outputs': False,
'reasoning_output': False,
'tool_calling': True,
'structured_output': True
}

更换模型 ID,profile 会跟着替换——把模型换成 deepseek/deepseek-v3.2 后(本机实测一致):

{'max_input_tokens': 163840, 'max_output_tokens': 65536, 'text_inputs': True,
'image_inputs': False, 'audio_inputs': False, 'video_inputs': False,
'text_outputs': True, 'image_outputs': False, 'audio_outputs': False,
'video_outputs': False, 'reasoning_output': True, 'tool_calling': True,
'structured_output': True}
Tip

这段代码只需要 OpenRouter 的 API Key(用于实例化),不会真正发送请求,所以不必充值就能看到画像。

画像的实用价值:写代码前先确认该模型支不支持工具调用(tool_calling)、结构化输出(structured_output)、图片输入(image_inputs,能省掉一大堆”跑起来才发现不支持”的坑。

调用时传入 config#

调用模型时(invoke()ainvoke()stream()batch() 等方法)都可以额外传入 config 参数。它允许在调用时动态配置和控制模型行为,而不必在初始化时就固定所有参数,为应用带来更大的灵活性和可维护性。

deepseek_llm.invoke(
"你好",
config={
"run_name": "...", # 在LangSmith中这次运行会显示为指定名称
"tags": ["test", "development"], # 打上标签便于分类查找
"metadata": {"user_id": "123"}, # 记录用户ID
"callbacks": [custom_handler], # 启用自定义回调函数
"configurable": {
"model": "deepseek-reasoner", # 配置模型参数
"temperature": 0.7, # 配置温度参数
"max_tokens": 100 # 配置最大令牌数
}
}
)

config 中支持的配置项:

配置项类型描述
run_namestr为当前运行设置一个可读的名称,便于在 LangSmith 追踪系统中快速定位和识别不同的运行任务
tagsList[str]为运行设置标签,用于分类和过滤,同样服务于 LangSmith 的检索
callbacksList[BaseCallbackHandler]设置回调处理器,在运行的不同阶段(开始、流式输出、结束等)触发,用于与监控平台(如 LangSmith)集成做深度追踪和调试
metadataDict[str, Any]附加任意键值对元数据,记录本次调用的业务上下文,如 {"user_id": "123", "session_id": "abc"}
max_concurrencyint限制当前可运行对象的最大并发运行数,防止对 API 接口或本地资源造成过大压力,实现简单的速率限制
recursion_limitint限制运行时递归调用的最大深度,主要在复杂工作流(如 Agent 执行多步工具调用)中防止无限递归循环
configurableDict[str, Any]一个万能字典,用于传递其他可配置参数,实现更高级的动态行为(如配置可替代的模型或组件)

说明:

  • run_nametagscallbacks 主要用在 LangSmith 中,用于追踪、筛选和调试;
  • metadata 可以放业务信息;在工作流开发中,当整个流程被包装为 Runnable 链时,这些参数还能传递给后续的链节点使用;
  • configurable 里可配置的参数与 init_chat_model 的初始化参数一样,区别在于生效范围与优先级
生效范围优先级
init_chat_model 初始化参数该模型实例的默认设置,适用于大部分场景
运行时 config单次调用的特定设置更高(针对本次调用做特殊调整)

举例 1:用 configurable 覆盖初始化参数

# 1. 初始化模型
model = init_chat_model(
model="deepseek-v4-flash",
model_provider="deepseek",
api_key=DEEPSEEK_API_KEY,
base_url=DEEPSEEK_BASE_URL,
temperature=0.2,
max_tokens=500,
# 指定可调整参数
configurable_fields=("model", "model_provider", "temperature", "max_tokens"),
)
# 2. 准备 config 字典
config = {
"run_name": "joke_generation", # 在LangSmith中这次运行会显示为 joke_generation
"tags": ["tag1", "tag2"], # 打上标签便于分类查找
"metadata": {"user_id": "123"}, # 记录用户ID
"configurable": {
"model": "deepseek-v4-pro", # 配置模型参数
"model_provider": "openai", # 配置模型提供商参数
"temperature": 0.7, # 配置温度参数
"max_tokens": 1000 # 配置最大令牌数
}
}
# 3. 调用模型并传入config
response = model.invoke("1 + 2 = ?", config=config)
rprint(response)

返回的元数据里能看到实际生效的是 config 里指定的模型

AIMessage(
content='1 + 2 = 3',
...
response_metadata={..., 'model_provider': 'openai', 'model_name': 'deepseek-v4-pro', ...},
usage_metadata={'input_tokens': 11, 'output_tokens': 65, 'total_tokens': 76, ..., 'output_token_details': {'reasoning': 57}}
)
Important

想用 config["configurable"] 覆盖默认参数,必须在 init_chat_model 时用 configurable_fields 声明”哪些参数允许在运行时替换”,否则配置不会生效。

举例 2:用 max_concurrency 限流

处理大量输入时,为了避免把模型服务打爆或触发速率限制:

large_list_of_inputs = [...., ...., ....]
model.batch(
large_list_of_inputs,
config={
'max_concurrency': 5 # 限制最大并发数为5
}
)

相关#

练习题#

一、回忆填空(写完再展开对答案)#

  1. 六种调用方式:____ / ____ / ____ 以及它们的异步版本 ____ / ____ / ____
  2. invoke 是____式,一次性返回完整结果;stream 返回一个 ____,可循环处理每个 chunk
  3. batch 会并行处理请求,返回结果按 ____ 顺序排列
  4. 异步方法的前缀 a 表示 ____;异步调用后可以继续执行本地逻辑,最后用 await 取结果
  5. invoke 的三种输入形式:(推荐)、____
  6. 字典形式的消息三个常用 role:____ 设定行为规则、____ 用户输入、____ AI 历史回复
  7. invoke 的返回值类型是 ____ 对象,取文本用 response.____
  8. 想看本次消耗了多少 token,看 response.response_metadata 里的 ____
  9. 流式输出依赖 ____ 对流式的支持
填空答案(做完再点开)
  1. invoke / stream / batch;ainvoke / astream / abatch 2. 阻塞 / 迭代器(iterator) 3. 原始输入 4. 异步(async) 5. 文本 / 字典列表 / 消息对象列表 6. system / user / assistant 7. AIMessage / content 8. token_usage 9. 模型供应商

二、裸写题#

  • 2-1 三种输入形式各写一遍 分别用文本、字典列表、消息对象列表三种方式问模型同一句话,打印 type(response)response.content

    提示(先自己想,实在想不出再点开)

    一级 · 思路:三种输入形式的结果类型应该一致(都是 AIMessage) 二级 · 方法:文本直接传;字典用 {"role": ..., "content": ...};消息对象用 HumanMessage(...) 三级 · 骨架from langchain_core.messages import HumanMessage

  • 2-2 给”导师”设系统提示 用字典列表形式构造两条消息:system 设定”你是一个专业的 Python 导师”,user 问”什么是装饰器”,打印回复。

    提示

    一级 · 思路:system 放在列表第一条 二级 · 方法[{"role": "system", ...}, {"role": "user", ...}] 三级 · 骨架:对比不加 system 时的回答风格差异

  • 2-3 流式 vs 非流式 同一个问题分别用 invokestream 调用,感受两者的返回差异(一次性 vs 逐字)。

    提示

    一级 · 思路:stream 返回的是可迭代对象,要循环 二级 · 方法for chunk in model.stream(...) 三级 · 骨架print(chunk.content, end="", flush=True)

  • 2-4 批量调用batch 一次问三个不同的问题,打印全部回复。

    提示

    一级 · 思路:一次调用传一组输入 二级 · 方法model.batch([...]) 返回列表 三级 · 骨架:结果顺序与输入顺序一致

参考答案(做完再点开)
import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
load_dotenv(override=True)
model = init_chat_model(
model="deepseek-v4-flash",
model_provider="openai",
base_url=os.getenv("DEEPSEEK_BASE_URL"),
api_key=os.getenv("DEEPSEEK_API_KEY"),
)
# 2-1 三种输入形式
r1 = model.invoke("用一句话介绍 Python")
r2 = model.invoke([{"role": "user", "content": "用一句话介绍 Python"}])
r3 = model.invoke([HumanMessage("用一句话介绍 Python")])
for r in (r1, r2, r3):
print(type(r).__name__, "→", r.content)
# 2-2 系统提示
r = model.invoke([
{"role": "system", "content": "你是一个专业的 Python 导师"},
{"role": "user", "content": "什么是装饰器"},
])
print(r.content)
# 2-3 流式 vs 非流式
r = model.invoke("用一句话介绍装饰器")
print("[非流式]", r.content)
print("[流式] ", end="")
for chunk in model.stream("用一句话介绍装饰器"):
print(chunk.content, end="", flush=True)
print()
# 2-4 批量调用(并行,结果顺序与输入一致)
responses = model.batch([
"用一句话介绍列表",
"用一句话介绍字典",
"用一句话介绍集合",
])
for i, r in enumerate(responses, 1):
print(i, r.content)

评论区

[ 标签 ]
# AI37# AI 编程2# AI工具1# Ajax2# Apifox1# AstrBot3# Astro2# CC Switch1# CDN2# Claude Code1# claudecode2# ClaudeCode1# Cloudflare2# CloudFlare2# CloudFlare-ImgBed3# coc3# CSS6# DeepSeek6# deepseek2# DELETE1# Docker1# EdgeOne3# Gist1# git1# GitHub1# hexo-circle-of-friends1# HTML6# HTTP5# ImageManager1# Java23# java13# JavaScript5# JDBC3# JSON2# JUnit1# LangChain25# Logback1# Maven6# Muse Spark1# Mybatis1# MyBatis4# MySQL28# MySql1# NapCat1# Node.js1# obsidian2# Obsidian5# OpenCode4# ORM1# PathVariable1# PicGo1# PyCharm1# Python65# RequestBody1# RequestMapping1# RESTful风格1# skills1# Slf4j1# SpringBoot11# SQL2# Streamlit5# Svelte2# TailwindCSS1# Telegram3# Tlias2# Vercel1# vscode2# Vue7# Waline3# WebDAV1# Web基础6# Web开发6# WinSCP1# YAML1# 三层架构1# 中二宣言1# 书籍1# 使用文档10# 写作1# 函数2# 刷步数1# 前端32# 动态1# 动漫1# 包1# 单词2# 博客7# 博客工作流1# 博客开发2# 参数接收1# 友链1# 反思2# 图床6# 地图1# 备份2# 大模型1# 奇思妙想1# 存储1# 学习方法6# 学校1# 宝塔面板3# 宝宝10# 对象1# 导航栏1# 工具2# 开发1# 开发工具1# 开发规范1# 开心1# 异常处理1# 影视2# 微信1# 性能优化2# 总结1# 想法15# 感受1# 感悟11# 指南1# 提示词工程1# 插件5# 故障排除1# 效率工具2# 教程10# 数据分析9# 数据库27# 数据结构1# 文件操作2# 斩神1# 日常92# 日志框架1# 朋友圈1# 朱元璋1# 模块1# 模板1# 正则表达式2# 测试1# 游戏2# 爬虫7# 生活迁移1# 电影2# 电脑1# 碎碎念1# 视觉识别1# 类1# 类型注解1# 网络基础2# 网络教室1# 羊毛2# 脚本2# 脚本工具1# 自动化2# 蓝奏云1# 订阅推荐2# 记录2# 评论系统1# 词根1# 词缀1# 说说1# 足迹1# 跑步2# 路径参数1# 转载2# 运动1# 部落冲突1# 配置1# 随机图1# 面向对象5# 音乐3# 音标1# 饮食1# 驼峰命名1# 高德地图1
[ 公告 ]

如果你喜欢,那么欢迎来到我的世界!

了解更多
[ 音乐 ]
封面

音乐

暂未播放

0:000:00
暂无歌词
找不到相关结果。
[ contents ]
[ 全部文章 ]