趋近智
与大型语言模型(LLM)的交互常常是LangChain应用中延迟和运营成本的主要因素。因此,在准备生产部署时,优化这些交互是首要任务。这里将介绍几种提高LLM调用效率的实用方法。
许多LLM应用会遇到重复的请求或子请求。对于相同输入多次执行相同的LLM调用会浪费时间和金钱(API成本)。缓存提供了一种简单方法来存储和重用之前的LLM响应。
LangChain提供了内置的缓存机制。最简单的形式是内存缓存,适用于开发或单进程应用:
from langchain.globals import set_llm_cache
from langchain_community.cache import InMemoryCache
set_llm_cache(InMemoryCache())
# 现在,此进程中任何后续相同的LLM调用(相同模型、参数、提示)
# 都将命中缓存。
# llm.invoke("天空为什么是蓝色的?") # 第一次调用 - 访问LLM
# llm.invoke("天空为什么是蓝色的?") # 第二次调用 - 命中缓存
对于生产环境,特别是涉及多个服务器实例或需要持久化的场景,外部缓存是必需的。选项包括数据库支持的缓存(如用于SQL数据库的SQLAlchemyCache或RedisCache)或专门的向量存储缓存(GPTCache,但需要单独配置)。
# 使用Redis的例子(需要Redis服务器和redis-py包)
# from langchain.globals import set_llm_cache
# from langchain_community.cache import RedisCache
# import redis
# client = redis.Redis(decode_responses=True)
# set_llm_cache(RedisCache(client))
缓存失效: 缓存的一个主要问题是确定缓存数据何时失效。如果LLM可能访问的底层知识发生变化,或者特定提示的预期行为发生改变,则需要更新缓存。方法有:
实施适当的缓存策略很大程度上依据应用对潜在过期数据的容忍度,以及其对低延迟和成本降低的需求。
LLM API成本通常根据输入和输出令牌的数量计算。延迟也通常与处理的令牌数量相关。因此,最小化令牌数量是一种直接的优化手段。
1. 提示工程:
2. 上下文管理:
3. 输出限制:
并非所有任务都需要使用最强大(且通常最昂贵、最慢)的LLM。请权衡模型能力、速度和成本。
下面的图表展示了比较结果:
不同类型语言模型的相对成本和延迟。实际值因供应商和具体模型版本而异。
如果您的LangChain应用涉及多个独立的LLM调用(例如,处理列表中的项目,查询一个主题的多个方面),并发执行这些调用可以显著缩短总耗时。
LangChain的表达式语言(LCEL)内置支持异步操作和并行执行,如第一章介绍的。使用 ainvoke、abatch 和 astream 等方法,以及Python的 asyncio 库,可以使LLM API调用等I/O密集型操作并发运行。
import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableParallel
# 假设llm是一个已初始化的ChatOpenAI实例
prompt = ChatPromptTemplate.from_template("讲一个关于{topic}的笑话")
chain = prompt | llm
# 并发运行多个独立调用的例子
topics = ["bears", "programmers", "cheese"]
coroutines = [chain.ainvoke({"topic": t}) for t in topics]
# 并发运行它们
results = await asyncio.gather(*coroutines)
# results将包含每个主题的响应
对于更复杂的工作流,LCEL的 RunnableParallel 允许在链中定义并行步骤:
# 在链中使用RunnableParallel进行并行LLM调用的例子
joke_chain = ChatPromptTemplate.from_template("讲一个关于{topic}的笑话") | llm
poem_chain = ChatPromptTemplate.from_template("写一首关于{topic}的短诗") | llm
map_chain = RunnableParallel(joke=joke_chain, poem=poem_chain)
# 这将并发执行joke_chain和poem_chain
result = await map_chain.ainvoke({"topic": "robots"})
# result将是一个字典:{'joke': ..., 'poem': ...}
利用异步执行对于构建在每个用户请求中执行多次LLM交互的响应式应用程序非常重要。
对于涉及长文本生成(如聊天机器人或内容创建工具)的应用,在显示任何内容之前等待整个LLM响应可能会导致糟糕的用户体验。流式传输允许应用在LLM输出生成时,逐个令牌接收和显示。
大多数LangChain LLM集成通过 stream 或 astream 方法支持流式传输。
# 流式输出的例子
async for chunk in llm.astream("写一个关于勇敢骑士的短故事。"):
# 处理每个到达的块(例如,打印到控制台,发送到UI)
print(chunk.content, end="", flush=True)
尽管流式传输不会减少总处理时间或成本,但它显著改善了最终用户的感知延迟,使应用感觉更具响应性。实现流式传输通常需要改变应用前端处理传入数据的方式,但它是生产级对话式AI的标准做法。
通过系统地应用这些方法:缓存、减少令牌、选择合适的模型、并行化和流式传输,您可以显著提高LangChain应用的性能和成本效益,使其适用于要求高的生产工作负载。后面的章节将讨论扩展其他组件,如数据检索系统。
简洁的语法。内置调试功能。从第一天起就可投入生产。
为 ApX 背后的 AI 系统而构建
这部分内容有帮助吗?
asyncio库的官方文档,该库是实现本节讨论的异步和并行LLM调用的基础。© 2026 ApX Machine Learning用心打造