Model I/O - 模型调用3、模型调用方式详解调用大模型就像打电话你得先知道说什么消息类型再知道怎么说传入方式最后知道怎么拨号调用方式。3.1 消息类型传什么LangChain 用标准化的消息格式来传递不同角色的内容。理解这四种消息类型是构建对话应用的基础。消息类型类名用途示例系统消息SystemMessage设定AI的行为、角色和规则你是一个有帮助的助手用户消息HumanMessage用户的输入帮我解释一下量子计算AI消息AIMessageAI的回复,可用于对话历史量子计算是...工具消息ToolMessage工具执行返回的结果工具调用的输出记忆口诀系统定规则用户提问题AI给回复工具报结果。3.1.1 基础示例fromlangchain_core.messagesimport(HumanMessage)fromlangchain_openaiimportChatOpenAI llmChatOpenAI(modelgpt-4o-mini)# 单个消息responsellm.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我叫什么名字),]responsellm.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 llmChatOpenAI(modelgpt-4o-mini)responsellm.invoke(你好介绍一下LangChain)print(response.content)1. 适用场景单轮问答不需要上下文不需要设定 AI 的角色或行为规则快速测试或调试优势代码最简洁一行搞定局限无法保留对话历史无法设定系统提示词3.2.2 方式二传入消息列表最常用fromlangchain_core.messagesimportHumanMessage,SystemMessage,AIMessage llmChatOpenAI(modelgpt-4o-mini)responsellm.invoke([SystemMessage(content你是一个专业的Python编程助手),HumanMessage(content什么是装饰器)])print(response.content)1. 适用场景需要设定 AI 角色比如你是一个翻译助手需要保留对话历史需要区分系统指令/用户输入/AI回复2. 实战示例多轮对话fromlangchain_core.messagesimportHumanMessage,AIMessage# 对话历史conversation[HumanMessage(content什么是LangChain?),AIMessage(contentLangChain是一个用于开发大模型应用的框架。),HumanMessage(content它有哪些核心组件)# 这依赖于上一轮的上下文]responsellm.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() 最常用最基础的调用方式适合大多数场景responsellm.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 llmChatOpenAI(modelgpt-4o-mini)asyncdefcall_llm_async():responseawaitllm.ainvoke(什么是LangChain? )print(response.content)# Jupyter Notebook 中直接 await见下方说明awaitcall_llm_async()# 方式一asyncio.run(call_llm_async())# 方式二2.invoke() vs ainvoke() 性能对比importtimeimportasynciofromlangchain_openaiimportChatOpenAI llmChatOpenAI(modelgpt-4o-mini)# 准备 5 个测试问题prompts[用一句话介绍一下北京,用一句话介绍一下上海,用一句话介绍一下广州,用一句话介绍一下深圳,用一句话介绍一下杭州]# 测试一同步 invoke串行 deftest_sync_invoke():print( 同步 invoke )start_timetime.time()fori,promptinenumerate(prompts):print(f [同步] 正在发送第{i1}个请求...)llm.invoke(prompt)# 死等拿到结果才进入下一次循环print(f总耗时{time.time()-start_time:.2f}秒\n)# 测试二异步 ainvoke并行 asyncdeftest_async_ainvoke():print( 异步 ainvoke )start_timetime.time()# 关键用 asyncio.gather 同时派发所有请求print( [异步] 瞬间派发 5 个请求...)tasks[llm.ainvoke(prompt)forpromptinprompts]resultsawaitasyncio.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():r1awaitllm.ainvoke(问题1)# 等第1个完成2秒r2awaitllm.ainvoke(问题2)# 再等第2个完成2秒r3awaitllm.ainvoke(问题3)# 再等第3个完成2秒# 总耗时~6秒串行# 正确写法ainvoke 负责能让出, gather 负责同时跑asyncdefright_way():tasks[llm.ainvoke(问题1),# 创建协程但不等待llm.ainvoke(问题2),# 创建协程但不等待llm.ainvoke(问题3),# 创建协程但不等待]r1,r2,r3awaitasyncio.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 / IPythonawait main()Jupyter 内部已经有一个事件循环在运行,不能再创建新的普通 .py 文件asyncio.run(main())需要自己创建并启动事件循环# Jupyter Notebook 中 asyncdefmain():responseawaitllm.ainvoke(你好)print(response.content)awaitmain()# ✓ 直接 await# asyncio.run(main()) # ✗ 会报错: Cannot run nested event loops# 普通 .py 文件中 importasyncioasyncdefmain():responseawaitllm.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 llmChatOpenAI(modelgpt-4o-mini)print(AI回答)full_messageNoneforchunkinllm.stream(请写一首关于春天的诗):# 累积消息块full_messagechunkiffull_messageisNoneelsefull_messagechunkprint(chunk.content,end,flushTrue)# 完整消息print(f\n\n完整消息:\n{full_message.content})streaming_example()flushTrue 的作用是强制将内存缓冲区中的内容立刻推送到屏幕上显示而不是等攒够了一定数量或者遇到换行符才显示。流式事件监听高级用法asyncdefstream_events():asyncforeventinllm.astream_events(你好):ifevent[event]on_chat_model_start:print(f输入{event[data][input]})elifevent[event]on_chat_model_stream:print(fToken:{event[data][chunk].content},end,flushTrue)elifevent[event]on_chat_model_end:print(f\n完成)awaitstream_events()适用场景聊天机器人、对话系统长文本生成让用户看到进度实时交互应用3.3.4 批次调用 - batch() 并行处理并行处理多个独立请求defbatch_example():fromlangchain_openaiimportChatOpenAI llmChatOpenAI(modelgpt-4o-mini)questions[什么是Python?,什么是JavaScript?,什么是Go语言?]responsesllm.batch(questions)forq,rinzip(questions,responses):print(fQ:{q})print(fA: {r.content}\n) batch_example()适用场景批量处理多个独立请求数据分析、批量内容生成不需要按顺序返回结果批量异步调用asyncdefbatch_async():questions[什么是LangChain?,LangChain的核心组件有哪些?,如何使用LangChain构建Agent?]responsesawaitllm.abatch(questions)forq,rinzip(questions,responses):print(fQ:{q}\nA:{r.content}\n)awaitbatch_async()3.4 调用配置与高级特性3.4.1 运行时配置1. 通过 config 参数传递运行时配置fromlangchain_openaiimportChatOpenAI llmChatOpenAI(modelgpt-4o-mini)responsellm.invoke(讲一个笑话,config{tags:[humor,demo],# 标签metadata:{user_id:123},# 元数据})print(response)config 字典是在 LangChain 框架层流转的“元数据”专门用于系统的观测、调试和日志记录它与业务层面的问答结果 response 是严格分离的。2. 用途serialized这辆快递车的车辆行驶证。上面写着车牌号模型名称、排量调试和追踪日志记录通过 metadata 传递额外信息监控和分析3. 这些 config 数据到底去哪了这些数据被 LangChain 的 回调系统Callback System 拦截并收集起来了。它们主要用于以下两个场景场景 A云端监控与可视化LangSmith通过在环境变量中配置了 LangSmithLangChain 官方的监控平台场景 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)llmChatOpenAI(modelgpt-4o-mini)# 2. 在调用时将你的回调处理器传进去responsellm.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_modelinit_chat_model(temperature0)# 使用GPT-4result1configurable_model.invoke(你好,config{configurable:{model:gpt-4o-mini})# 使用Clauderesult2configurable_model.invoke(你好,config{configurable:{model:claude-sonnet-4-6})1.适用场景需要在运行时根据用户选择切换模型A/B 测试不同模型效果多租户系统不同客户使用不同模型