Model I/O - 模型调用（中）

Model I/O - 模型调用

3、模型调用方式详解

调用大模型就像打电话：你得先知道"说什么"（消息类型），再知道"怎么说"（传入方式），最后知道"怎么拨号"（调用方式）。

3.1 消息类型：传什么

LangChain 用标准化的消息格式来传递不同角色的内容。理解这四种消息类型，是构建对话应用的基础。

消息类型	类名	用途	示例
系统消息	SystemMessage	设定AI的行为、角色和规则	"你是一个有帮助的助手"
用户消息	HumanMessage	用户的输入	"帮我解释一下量子计算"
AI消息	AIMessage	AI的回复,可用于对话历史	"量子计算是..."
工具消息	ToolMessage	工具执行返回的结果	工具调用的输出

记忆口诀：系统定规则，用户提问题，AI给回复，工具报结果。

3.1.1 基础示例：

fromlangchain_core.messagesimport(HumanMessage)fromlangchain_openaiimportChatOpenAI llm=ChatOpenAI(model="gpt-4o-mini")# 单个消息response=llm.invoke([HumanMessage(content="你好"])])print(response.content)

3.1.2 实战示例：构建完整对话历史

fromlangchain_core.messagesimport(HumanMessage,SystemMessage,AIMessage)# 对话历史--模拟多轮对话conversation=[SystemMessage(content="你是一个有帮助的AI助手"),HumanMessage(content="你好，我叫hzk"),AIMessage(content="你好！hzk，有什么我可以帮助你的吗？"),HumanMessage(content="我叫什么名字？"),]response=llm.invoke(conversation)print(response.content)

3.1.3 为什么需要区分消息类型？

系统消息：设定AI的"人设"和行为规则

用户消息：真正的问题或指令

AI消息：保留历史上下文，让AI"记得"之前说过什么

工具消息：当AI调用外部工具时，工具返回的结果

3.2 传入方式：怎么传

知道"传什么"之后，下一个问题是："怎么传给模型？"常用三种传入方式，各有适用场景。

快速决策表

你的需求	推荐方式	代码示例
简单问答,不需要上下文	直接传字符串	1lm.invoke("你好")
需要角色设定或对话历史	传消息列表	1lm.invoke([SystemMessage(...), HumanMessage(...)])
动态构建消息,或从其他格式转换	用元组/字典	1lm.invoke([("system", "..."), ("user", "...")])

一句话记忆：

能用字符串就用字符串（最简单）

需要对话/角色时用消息列表（最常见）

动态构建时用元组/字典（最灵活）

3.2.1 方式一：直接传入字符串（最简单）

fromlangchain_openaiimportChatOpenAI llm=ChatOpenAI(model="gpt-4o-mini")response=llm.invoke("你好，介绍一下LangChain")print(response.content)

1. 适用场景：

单轮问答，不需要上下文

不需要设定 AI 的角色或行为规则

快速测试或调试

优势：代码最简洁，一行搞定

局限：无法保留对话历史，无法设定系统提示词

3.2.2 方式二：传入消息列表（最常用）

fromlangchain_core.messagesimportHumanMessage,SystemMessage,AIMessage llm=ChatOpenAI(model="gpt-4o-mini")response=llm.invoke([SystemMessage(content="你是一个专业的Python编程助手"),HumanMessage(content="什么是装饰器？")])print(response.content)

1. 适用场景：

需要设定 AI 角色（比如"你是一个翻译助手"）

需要保留对话历史

需要区分系统指令/用户输入/AI回复

2. 实战示例：多轮对话

fromlangchain_core.messagesimportHumanMessage,AIMessage# 对话历史conversation=[HumanMessage(content="什么是LangChain?"),AIMessage(content="LangChain是一个用于开发大模型应用的框架。"),HumanMessage(content="它有哪些核心组件？")# 这依赖于上一轮的上下文]response=llm.invoke(conversation)print(response.content)

3.2.3 方式三：使用元组或字典（最灵活）

# 元组方式：（角色，内容）tuple_messages=[("system","你是一个专业的Python编程助手"),("user","什么是装饰器？")]# 字典方式：{"role": 角色, "content": 内容}dict_messages=[{"role":"system","content":"你是一个专业的Python编程助手"},{"role":"user","content":"什么是装饰器？"}]print(llm.invoke(tuple_messages))print(llm.invoke(dict_messages))

1.适用场景：

从 API 返回的 JSON 数据直接转成消息列表

从配置文件或数据库读取对话模板

动态构建消息列表

2.实战示例：从配置读取对话模板

# 假设这是从配置文件读取的prompt_template=[{"role":"system","content":"你是一个{role}"},{"role":"user","content":"请解释{topic}"}]# 动态填充messages=[{"role":t["role"],"content":t["content"].format(role="翻译助手",topic="机器翻译",),}fortinprompt_template]print(llm.invoke(messages).content)

3.3 调用方式：怎么调

知道"传什么"和"怎么传"之后，最后一个问题是："怎么调用模型？"LangChain 提供了多种调用方式，适应不同场景。

3.3.1 同步调用 - invoke() （最常用）

最基础的调用方式，适合大多数场景：

response=llm.invoke("什么是LangChain? ")print(response.content)

适用场景：

单次调用，不需要高并发

简单问答、文本生成

快速原型开发

3.3.2 异步调用 - ainvoke() （高并发）

适用于需要同时处理多个请求的高并发场景。要理解 ainvoke() 的价值，需要先搞清楚一个前置知识：Python 的异步编程（async/await）。

前置知识：同步 vs 异步

同步（invoke）：排队买奶茶 你 → 点单 → 等10分钟 → 拿到 → 再点下一杯 → 等10分钟 → 拿到5杯奶茶总耗时：50分钟（串行等待） 异步（ainvoke）：扫码下单 你 → 5杯同时下单 → 哪杯好了取哪杯 5杯奶茶总耗时：≈10分钟（并行等待）

为什么模型调用特别适合异步？ 因为 llm.invoke() 的耗时几乎全花在"等网络响应"上，CPU 其实是闲着的。异步让你在等第1个响应的同时，把第2、3、4、5个请求也发出去，所有等待时间重叠，总耗时 ≈单次最慢的那个请求。

1. 基础用法

importasynciofromlangchain_openaiimportChatOpenAI llm=ChatOpenAI(model="gpt-4o-mini")asyncdefcall_llm_async():response=awaitllm.ainvoke("什么是LangChain? ")print(response.content)# Jupyter Notebook 中直接 await（见下方说明）awaitcall_llm_async()# 方式一asyncio.run(call_llm_async())# 方式二

2.invoke() vs ainvoke() 性能对比

importtimeimportasynciofromlangchain_openaiimportChatOpenAI llm=ChatOpenAI(model="gpt-4o-mini")

# 准备 5 个测试问题prompts=["用一句话介绍一下北京","用一句话介绍一下上海","用一句话介绍一下广州","用一句话介绍一下深圳","用一句话介绍一下杭州"]# ======== 测试一：同步 invoke（串行） ========deftest_sync_invoke():print("=== 同步 invoke ===")start_time=time.time()fori,promptinenumerate(prompts):print(f" [同步] 正在发送第{i+1}个请求...")llm.invoke(prompt)# 死等，拿到结果才进入下一次循环print(f"总耗时：{time.time()-start_time:.2f}秒\n")# ======== 测试二：异步 ainvoke（并行） ========asyncdeftest_async_ainvoke():print("=== 异步 ainvoke ===")start_time=time.time()# 关键：用 asyncio.gather 同时派发所有请求print(" [异步] 瞬间派发 5 个请求...")tasks=[llm.ainvoke(prompt)forpromptinprompts]results=awaitasyncio.gather(*tasks)forrinresults:print(f" 回答：{r.content[:20]}...")print(f"总耗时：{time.time()-start_time:.2f}秒\n")# ======== 运行对比 ========asyncdefmain():test_sync_invoke()# 先跑同步awaittest_async_ainvoke()# 再跑异步awaitmain()

3.典型输出：

=== 同步 invoke === [同步] 正在发送第 1 个请求... [同步] 正在发送第 2 个请求... ... 总耗时：8.73 秒 === 异步 ainvoke === [异步] 瞬间派发 5 个请求... 总耗时：1.92 秒

5个请求，同步耗时 ~9秒，异步耗时 ~2秒——快了 4-5 倍。请求越多，差距越大。

关键知识点：ainvoke() 和 asyncio.gather() 各自干了什么？

看完对比你可能会问：既然 ainvoke() 是异步函数，为什么逐个 await ainvoke() 跟同步一样慢？

因为 ainvoke() 和 gather() 解决的是两个不同的问题：

ainvoke() 解决的是 → "等待时不阻塞"（让出 CPU，别的任务有机会插进来） gather() 解决的是 → "同时派发多个任务"（把多个协程塞进事件循环并行跑）

回到奶茶店的比喻：

invoke() = 你站在柜台前死等，奶茶没做好之前你哪儿也去不了，后面的人也点不了单

ainvoke() = 你扫码下单后去旁边坐着，不占柜台了，后面的人可以继续点单

asyncio.gather() = 同时帮5个人下单，让奶茶店并行制作

所以：如果只有你一个人买奶茶，ainvoke() （坐着等）和 invoke()（站着等）时间一样长——因为没有"后面的人"需要你让位。ainvoke() 的价值在于"让出控制权"，而 gather() 的价值在于"利用让出的控制权塞入更多任务"。两者缺一不可。

# 错误理解：用了 ainvoke 就会快# 实际效果：还是串行，因为每次 await 都在等当前这个完成asyncdefwrong_way():r1=awaitllm.ainvoke("问题1")# 等第1个完成（2秒）r2=awaitllm.ainvoke("问题2")# 再等第2个完成（2秒）r3=awaitllm.ainvoke("问题3")# 再等第3个完成（2秒）# 总耗时：~6秒（串行）# 正确写法：ainvoke 负责"能让出", gather 负责"同时跑"asyncdefright_way():tasks=[llm.ainvoke("问题1"),# 创建协程，但不等待llm.ainvoke("问题2"),# 创建协程，但不等待llm.ainvoke("问题3"),# 创建协程，但不等待]r1,r2,r3=awaitasyncio.gather(*tasks)# 三个请求同时发出，同时等待# 总耗时：~2秒（并行）

写法 效果 类比 invoke() 逐个调用 串行，阻塞 站在柜台前死等，一杯一杯买

写法	效果	类比
await ainvoke()逐个调用	串行,不阻塞但没利用起来	扫码后坐着等,但只点了1杯,没人需要你让位
asyncio.gather(*tasks)	并行,耗时≈最慢的那个	同时下单5杯,谁好了取谁

*tasks 是 Python 的解包语法：gather(*[a, b, c]) 等价于 gather(a, b, c)。

4. 运行环境差异：Jupyter vs 普通 .py 文件

你可能注意到了，上面的代码直接写了 await main() ，而不是 asyncio.run(main()) 。这是因为运行环境不同：

环境	启动异步的方式	原因
Jupyter Notebook / IPython	await main()	Jupyter 内部已经有一个事件循环在运行,不能再创建新的
普通 .py 文件	asyncio.run(main())	需要自己创建并启动事件循环

# ===== Jupyter Notebook 中 ====asyncdefmain():response=awaitllm.ainvoke("你好")print(response.content)awaitmain()# ✓ 直接 await# asyncio.run(main()) # ✗ 会报错: Cannot run nested event loops# === 普通 .py 文件中 ====importasyncioasyncdefmain():response=awaitllm.ainvoke("你好")print(response.content)# await main() # ✗ 会报错: await 只能在 async 函数内使用asyncio.run(main())# ✓ 创建事件循环并运行

在 Jupyter 环境中运行，所以统一使用 await 写法。如果你在 PyCharm 的 .py 文件中运行，

把 await main() 改成 asyncio.run(main()) 即可。

5.适用场景：

需要同时处理多个请求（批量调用、并行对比）

Web 服务、API 接口（FastAPI 等异步框架）

对响应时间有要求的应用

3.3.3 流式调用 - stream() （打字机效果）

实现打字机效果，提升用户体验：

defstreaming_example():fromlangchain_openaiimportChatOpenAI llm=ChatOpenAI(model="gpt-4o-mini")print("AI回答：")full_message=Noneforchunkinllm.stream("请写一首关于春天的诗"):# 累积消息块full_message=chunkiffull_messageisNoneelsefull_message+chunkprint(chunk.content,end="",flush=True)# 完整消息print(f"\n\n完整消息:\n{full_message.content}")streaming_example()

flush=True 的作用是强制将内存缓冲区中的内容立刻推送到屏幕上显示，而不是等攒够了一定数量或者遇到换行符才显示。

流式事件监听（高级用法）：

asyncdefstream_events():asyncforeventinllm.astream_events("你好"):ifevent["event"]=="on_chat_model_start":print(f"输入：{event['data']['input']}")elifevent["event"]=="on_chat_model_stream":print(f"Token:{event['data']['chunk'].content}",end="",flush=True)elifevent["event"]=="on_chat_model_end":print(f"\n完成！")awaitstream_events()

适用场景：

• 聊天机器人、对话系统
• =长文本生成（让用户看到进度）
• 实时交互应用

3.3.4 批次调用 - batch() （并行处理）

并行处理多个独立请求：

defbatch_example():fromlangchain_openaiimportChatOpenAI llm=ChatOpenAI(model="gpt-4o-mini")questions=["什么是Python?","什么是JavaScript?","什么是Go语言?"]responses=llm.batch(questions)forq,rinzip(questions,responses):print(f"Q:{q}")

print(f"A: {r.content}\n") batch_example()

适用场景：

• 批量处理多个独立请求
• 数据分析、批量内容生成
• 不需要按顺序返回结果

批量异步调用：

asyncdefbatch_async():questions=["什么是LangChain?","LangChain的核心组件有哪些?","如何使用LangChain构建Agent?"]responses=awaitllm.abatch(questions)forq,rinzip(questions,responses):print(f"Q:{q}\nA:{r.content}\n")awaitbatch_async()

3.4 调用配置与高级特性

3.4.1 运行时配置

1. 通过 config 参数传递运行时配置：

fromlangchain_openaiimportChatOpenAI llm=ChatOpenAI(model="gpt-4o-mini")response=llm.invoke("讲一个笑话",config={"tags":["humor","demo"],# 标签"metadata":{"user_id":"123"},# 元数据})print(response)

config 字典是在 LangChain 框架层流转的“元数据”，专门用于系统的观测、调试和日志记录，它与业务层面的问答结果 response 是严格分离的。

2. 用途：

serialized：这辆快递车的车辆行驶证。上面写着车牌号（模型名称）、排量

• 调试和追踪
• 日志记录（通过 metadata 传递额外信息）
• 监控和分析

3. 这些 config 数据到底去哪了？

这些数据被 LangChain 的 回调系统（Callback System） 拦截并收集起来了。它们主要用于以下两个场景：

场景 A：云端监控与可视化（LangSmith）通过在环境变量中配置了 LangSmith（LangChain 官方的监控平台）

场景 B：本地回调拦截,通过回调处理器（Callback Handler）来拦截它们

fromlangchain_openaiimportChatOpenAIfromlangchain_core.callbacksimportBaseCallbackHandler# 1. 自定义一个回调处理器，拦截并打印运行信息classMyDebugCallback(BaseCallbackHandler):defon_chat_model_start(self,serialized,prompts,**kwargs):print("\n"+"="*40)print(" [拦截到 LLM 启动请求] ")print(f" Tags:{kwargs.get('tags')}")print(f" Metadata:{kwargs.get('metadata')}")print"="*40+"\n")llm=ChatOpenAI(model="gpt-4o-mini")# 2. 在调用时，将你的回调处理器传进去response=llm.invoke("讲一个短笑话",config={"tags":["humor","demo"],"metadata":{"user_id":"123"},"callbacks":[MyDebugCallback()]# 关键：挂载你的回调})print("最终返回的 response 依然只有大模型的内容：")print(response.content)

messages prompts： 包裹里的货品。

config / kwargs ：运单上的加急标签、客户编号（比如 tags , metadata ）。

（Temperature）、车辆品牌（OpenAI 类路径），它不影响包裹的内容，但对车队管理员（框架和开发者）来说是必不可少的档案。

3.4.2 运行时动态切换模型

2.4 节介绍了用 init_chat_model 在初始化时选择不同模型。这里展示一个更高级的用法——不重新初始化，在调用时通过 config 参数动态切换：

fromlangchain.chat_modelsimportinit_chat_model configurable_model=init_chat_model(temperature=0)# 使用GPT-4result1=configurable_model.invoke("你好",config={"configurable":{"model":"gpt-4o-mini"})# 使用Clauderesult2=configurable_model.invoke("你好",config={"configurable":{"model":"claude-sonnet-4-6"})

1.适用场景：

• 需要在运行时根据用户选择切换模型
• A/B 测试不同模型效果
• 多租户系统（不同客户使用不同模型）