模型的调用
模型的调用
invoke/stream/batch 三种调用方式与它们的异步版本,invoke 的三种输入形式,以及 AIMessage 返回值结构
六种调用方式总览
模型调用(Invocation)是指通过特定方法触发大语言模型生成输出的过程。
| 方法 | 特点 | 适用场景 |
|---|---|---|
invoke() | 阻塞式,一次性返回完整结果 | 问答、批处理任务、无需实时反馈 |
stream() | 流式输出,实时返回每个 token | 聊天机器人、长文本生成、需要提升体验 |
batch() | 批量处理多个输入 | 高并发、需要同时处理大量请求 |
ainvoke() | 非阻塞式(异步) | 高并发 Web 应用、IO 密集型任务 |
astream() | 非阻塞式流式 | 同上 |
abatch() | 非阻塞式批量 | 同上 |
记法:前缀 a = async(异步),同步版与异步版一一对应。
invoke():最核心的方法
阻塞式:程序会等模型完全生成整个响应后,再一次性返回。
response = model.invoke(input, config=None)它做的事:① 接收你的输入(问题、指令、对话历史)② 发送给 LLM ③ 返回响应(文本 + 元数据)。
参数详解
| 参数 | 类型 | 说明 | 必需 | 默认值 |
|---|---|---|---|---|
input | str | list[dict] | list[Message] 等 | 你要发送给模型的内容 | 必需 | 无 |
config | dict | 高级配置(回调函数、元数据、标签等) | 可选 | None |
invoke 的源码签名(除了 config,还有一个直接传的 stop 停止符参数):
def invoke( self, input: LanguageModelInput, config: RunnableConfig | None = None, *, stop: list[str] | None = None, **kwargs: Any,) -> AIMessage:config 属于进阶用法(run_name、tags、metadata、callbacks、configurable 等),本笔记最后一节专门讲。
三种输入形式
① 文本输入(最简单)
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 | 英文 | 作用 |
|---|---|---|
system | System | 设定 AI 的行为、角色、规则(如”你是一个专业的 Python 导师”) |
user | Human / User | 用户的输入/问题 |
assistant | AI / Assistant | AI 的历史回复,用于对话上下文 |
③ 消息对象列表
用 SystemMessage、HumanMessage、AIMessage 这些专门的类来构造:
from langchain_core.messages import HumanMessage
response = model.invoke([HumanMessage("2 + 3 * 2 = ?")])"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:你叫小明。结论:模型的”记忆”完全来自你传进 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 | 模型供应商,如 openai、deepseek |
finish_reason | 生成停止的原因:stop = 正常回答结束;length = 达到最大 Token 限制被截断(调用工具时还会出现 tool_calls) |
system_fingerprint | 系统指纹,用于追踪模型后端的配置变更 |
id | API 层面的响应 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),决定了打字机效果是否丝滑 |
latency_checkpoint 不是所有供应商都会返回(它来自 OpenAI 兼容的高阶接口);实测 DeepSeek 官方接口的返回里就没有这一项。
⑤ 工具调用信息
| 字段 | 说明 |
|---|---|
tool_calls | 结构化工具调用列表:模型决定调用某个函数或搜索工具时,参数都在这里 |
invalid_tool_calls | 触发失败或格式错误的工具调用尝试 |
一次访问所有信息
response = model.invoke("用一句话解释什么是 AI")
# 1. 获取回复内容print("AI 回复:", response.content)
# 2. 获取响应元数据metadata = response.response_metadataprint(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. 获取消息 IDprint(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-0stream():流式调用
| invoke() | stream() | |
|---|---|---|
| 返回时机 | 全部生成完才返回 | 实时返回响应片段 |
| 返回值 | AIMessage | 迭代器(iterator) |
| 适合 | 短回答、后台任务 | 长文本、聊天机器人(用户体验好) |
for chunk in model.stream("请用中文详细介绍什么是AI"): print(chunk.content, end="", flush=True)流式输出依赖模型供应商对流式的支持——不是所有模型/平台都支持。
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() - startprint(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() - startprint(f"循环调用耗时: {loop_time:.2f}秒")print(f"批量调用节省: {((loop_time - batch_time) / loop_time * 100):.1f}%")课程环境实测:批量调用 1.91 秒,循环调用 3.87 秒,批量调用节省约 50.7%。
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() | 批量 |
异步调用的两个注意事项
- 异步示例请在
.py文件中执行,而不是 Jupyter Notebook 里——Jupyter 自带事件循环,和asyncio.run()会打架。 - 等待时要用
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 等) | 打日志、告警,必要时降级到备用模型 |
except 的顺序很重要:它是从上往下匹配的,所以具体异常(ValueError、ConnectionError)必须写在 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 ==================================
我是小王,你是老王。另一种美化方式是 rich 库的 rprint(response)(前面的返回值一节已经用过):它会把 content、response_metadata、usage_metadata 全部彩色展开,适合在终端里做深度调试;pretty_print() 只呈现”谁说了什么”,更干净。
模型能力画像:profile
LangChain 1.1 及更高版本可以通过 profile 属性查看模型的配置信息。这是 LangChain 针对模型做的能力画像,但它是否存在,取决于 LangChain 在集成模型厂商的服务时是否声明了能力画像。
例 1:DeepSeek 官方模型——输出为空 {}(本机实测结果一致):
from langchain_deepseek import ChatDeepSeekfrom 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 ChatOpenRouterfrom dotenv import load_dotenvfrom 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}这段代码只需要 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_name | str | 为当前运行设置一个可读的名称,便于在 LangSmith 追踪系统中快速定位和识别不同的运行任务 |
tags | List[str] | 为运行设置标签,用于分类和过滤,同样服务于 LangSmith 的检索 |
callbacks | List[BaseCallbackHandler] | 设置回调处理器,在运行的不同阶段(开始、流式输出、结束等)触发,用于与监控平台(如 LangSmith)集成做深度追踪和调试 |
metadata | Dict[str, Any] | 附加任意键值对元数据,记录本次调用的业务上下文,如 {"user_id": "123", "session_id": "abc"} |
max_concurrency | int | 限制当前可运行对象的最大并发运行数,防止对 API 接口或本地资源造成过大压力,实现简单的速率限制 |
recursion_limit | int | 限制运行时递归调用的最大深度,主要在复杂工作流(如 Agent 执行多步工具调用)中防止无限递归循环 |
configurable | Dict[str, Any] | 一个万能字典,用于传递其他可配置参数,实现更高级的动态行为(如配置可替代的模型或组件) |
说明:
run_name、tags、callbacks主要用在 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. 调用模型并传入configresponse = 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}})想用 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 })相关
练习题
一、回忆填空(写完再展开对答案)
- 六种调用方式:____ / ____ / ____ 以及它们的异步版本 ____ / ____ / ____
- invoke 是____式,一次性返回完整结果;stream 返回一个 ____,可循环处理每个 chunk
- batch 会并行处理请求,返回结果按 ____ 顺序排列
- 异步方法的前缀
a表示 ____;异步调用后可以继续执行本地逻辑,最后用await取结果 - invoke 的三种输入形式:、(推荐)、____
- 字典形式的消息三个常用 role:____ 设定行为规则、____ 用户输入、____ AI 历史回复
- invoke 的返回值类型是 ____ 对象,取文本用
response.____ - 想看本次消耗了多少 token,看
response.response_metadata里的____ - 流式输出依赖 ____ 对流式的支持
填空答案(做完再点开)
- 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 非流式 同一个问题分别用
invoke和stream调用,感受两者的返回差异(一次性 vs 逐字)。提示一级 · 思路:stream 返回的是可迭代对象,要循环 二级 · 方法:
for chunk in model.stream(...)三级 · 骨架:print(chunk.content, end="", flush=True) -
2-4 批量调用 用
batch一次问三个不同的问题,打印全部回复。提示一级 · 思路:一次调用传一组输入 二级 · 方法:
model.batch([...])返回列表 三级 · 骨架:结果顺序与输入顺序一致
参考答案(做完再点开)
import osfrom dotenv import load_dotenvfrom langchain.chat_models import init_chat_modelfrom 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)评论区
如果你喜欢,那么欢迎来到我的世界!
了解更多













