12. LangChain 6大核心调用方法：invoke/stream/batch同步异步全解析，新手也能轻松学会

12. LangChain 6大核心调用方法invoke/stream/batch同步异步全解析新手也能轻松学会在LangChain开发中如何高效调用LLM是每个开发者必须掌握的核心技能。无论你是构建聊天机器人、文档问答系统还是自动化工作流理解invoke、stream、batch以及它们的异步版本ainvoke、astream、abatch将极大提升你的代码效率和用户体验。本文将从实战角度出发用大量代码示例带你逐一攻破这6大方法让你从新手秒变高手。—## 1. 准备工作环境与基础模型在开始之前请确保已安装LangChain和OpenAI库或使用其他LLM如Ollama。这里我们使用OpenAI作为示例但方法同样适用于其他模型。bashpip install langchain langchain-openai接下来设置API Key并创建一个简单的LLM实例pythonimport osfrom langchain_openai import ChatOpenAI# 设置API Key建议使用环境变量或配置文件不要硬编码os.environ[OPENAI_API_KEY] your-api-key-here# 创建一个Chat模型实例llm ChatOpenAI(modelgpt-3.5-turbo, temperature0)注意如果你没有OpenAI API可以使用ChatOllama等开源模型替代方法完全一样。—## 2. invoke最基础的同步调用invoke是LangChain中最简单、最常用的调用方法。它接收一个输入同步返回一个结果。python# 示例1使用invoke进行单次调用from langchain_core.messages import HumanMessage# 创建一个消息列表作为输入messages [HumanMessage(content请用一句话解释什么是量子计算)]# 调用invoke同步等待结果result llm.invoke(messages)# 输出结果print( invoke 示例 )print(result.content)运行结果示例 invoke 示例 量子计算是一种利用量子力学原理如叠加和纠缠来进行信息处理的计算方式能解决某些传统计算机难以高效解决的问题。invoke适用于**简单问答**、**单次推理**等场景。它的优点是简单直接缺点是如果LLM响应慢程序会阻塞等待。---## 3. stream流式输出实时体验当需要给用户实时显示生成内容时如ChatGPT打字效果stream就是你的最佳选择。它将LLM的输出按token或chunk逐步返回。python# 示例2使用stream实现流式输出from langchain_core.messages import HumanMessageprint( stream 示例 “)print(“AI正在思考…”, end”“, flushTrue)# 创建一个消息messages [HumanMessage(content“写一首关于人工智能的短诗不超过4行。”)]# 使用stream方法逐步获取生成内容for chunk in llm.stream(messages): # chunk.content包含部分生成文本 print(chunk.content, end”“, flushTrue)print(”\n 流式输出完成 )**运行效果**实时逐个字符显示 stream 示例 AI正在思考…硅基智慧闪星光数据海洋任翱翔。深度学习破迷障未来世界共辉煌。 流式输出完成 stream特别适合**聊天机器人**、**实时翻译**、**代码生成预览**等需要即时反馈的场景。---## 4. batch批量处理效率翻倍当你有多个独立的问题需要同时处理时batch方法可以并行发送请求大幅提升效率。它接收一个输入列表返回一个结果列表。python# 示例3使用batch批量处理多个问题from langchain_core.messages import HumanMessage# 准备多个独立的问题questions [ 什么是Python, 什么是Java, 什么是JavaScript, 什么是Go语言]# 创建消息列表的列表messages_list [[HumanMessage(contentq)] for q in questions]print( batch 示例 )print(f正在处理 {len(questions)} 个问题...)# 使用batch并行调用results llm.batch(messages_list)# 输出每个结果for i, result in enumerate(results): print(f问题{i1}: {questions[i]}) print(f回答: {result.content[:30]}...) # 截取前30个字符 print(- * 40)**运行结果示例** batch 示例 正在处理 4 个问题...问题1: 什么是Python回答: Python是一种高级、解释型、通用...----------------------------------------问题2: 什么是Java回答: Java是一种面向对象、跨平台的...----------------------------------------问题3: 什么是JavaScript回答: JavaScript是一种轻量级、解释型...----------------------------------------问题4: 什么是Go语言回答: Go又称Golang是一种静态...----------------------------------------batch适用于**批量文档摘要**、**多轮问答预处理**、**数据标注**等需要大量独立调用的场景。注意如果LLM有速率限制请谨慎使用高并发。---## 5. 异步三剑客ainvoke, astream, abatch在异步编程中如FastAPI、WebSocket服务同步方法会阻塞事件循环。LangChain提供了对应的异步版本ainvoke、astream、abatch。### 5.1 ainvoke异步单次调用python# 示例4使用ainvoke进行异步调用import asynciofrom langchain_core.messages import HumanMessageasync def async_invoke_demo(): messages [HumanMessage(content异步编程有什么优点)] # 使用ainvoke不会阻塞其他协程 result await llm.ainvoke(messages) print( ainvoke 示例 ) print(result.content[:50] ...)# 运行异步函数asyncio.run(async_invoke_demo())### 5.2 astream异步流式输出python# 示例5使用astream实现异步流式输出import asynciofrom langchain_core.messages import HumanMessageasync def async_stream_demo(): messages [HumanMessage(content用3句话解释云计算。)] print( astream 示例 ) # 异步遍历流式输出 async for chunk in llm.astream(messages): print(chunk.content, end, flushTrue) print(\n 异步流式完成 )asyncio.run(async_stream_demo())### 5.3 abatch异步批量处理python# 示例6使用abatch进行异步批量调用import asynciofrom langchain_core.messages import HumanMessageasync def async_batch_demo(): questions [ 什么是机器学习, 什么是深度学习, 什么是强化学习 ] messages_list [[HumanMessage(contentq)] for q in questions] print( abatch 示例 ) print(f异步处理 {len(questions)} 个问题...) # 异步批量调用 results await llm.abatch(messages_list) for i, result in enumerate(results): print(f问题{i1}: {questions[i]}) print(f回答: {result.content[:40]}...) print(- * 30)asyncio.run(async_batch_demo())异步方法在**构建高性能API服务**、**处理大量并发请求**、**集成到异步框架**如FastAPI、Tornado时必不可少。---## 6. 实战对比同步 vs 异步性能测试为了让你直观感受性能差异我们做一个简单测试pythonimport timeimport asynciofrom langchain_core.messages import HumanMessage# 准备20个简单问题questions [什么是苹果] * 20messages_list [[HumanMessage(contentq)] for q in questions]def sync_test(): start time.time() results llm.batch(messages_list) end time.time() print(f同步batch耗时: {end - start:.2f}秒)async def async_test(): start time.time() results await llm.abatch(messages_list) end time.time() print(f异步batch耗时: {end - start:.2f}秒)# 运行测试print( 性能对比测试 )sync_test()asyncio.run(async_test())**输出示例** 性能对比测试 同步batch耗时: 8.23秒异步batch耗时: 7.95秒 虽然在这个简单测试中差异不大但在高并发、网络延迟大的场景下异步的优势会非常明显。---## 7. 最佳实践与选择指南| 场景 | 推荐方法 | 原因 ||------|----------|------|| 简单问答一次请求 |invoke| 简单直接代码最简洁 || 实时聊天/打字效果 |stream| 提升用户体验减少等待感 || 批量文档处理 |batch| 并行请求提高吞吐量 || FastAPI/WebSocket服务 |ainvoke/astream| 不阻塞事件循环 || 高并发异步任务 |abatch| 高效处理大量独立请求 |**注意事项**- 使用batch/abatch时注意API速率限制必要时添加延迟- 异步方法必须在协程中调用使用async def和await- 对于复杂链Chain这些方法同样适用只需将llm替换为链对象—## 8. 总结本文通过6个实战代码示例详细解析了LangChain的6大核心调用方法-invoke同步单次调用新手入门首选-stream流式输出打造实时体验-batch同步批量处理提升效率-ainvoke/astream/abatch异步版本高性能编程必备掌握这些方法后你可以根据业务场景灵活选择无论是构建简单的问答系统还是复杂的实时AI应用都能游刃有余。建议你动手运行每个示例并尝试修改参数加深理解。LangChain的世界很大但掌握这些基础调用方法你已经迈出了坚实的第一步