📢 转载信息
原文链接:https://www.kdnuggets.com/5-powerful-python-decorators-to-optimize-llm-applications
原文作者:Iván Palomares Carrascosa
#
简介
Python 装饰器是专门设计的解决方案,旨在帮助简化各种应用(包括基于 LLM 的应用)中的复杂软件逻辑。处理 LLM 通常需要应对不可预测、缓慢且常常昂贵的第三方 API,而装饰器可以通过封装(例如,用优化逻辑封装 API 调用)使这项任务更加清晰。
让我们来看看五个有用的 Python 装饰器,它们可以在不带来明显额外负担的情况下帮助您优化基于 LLM 的应用。
附带的示例说明了使用每个装饰器的语法和方法。它们有时会显示为没有实际 LLM 使用,但它们是最终旨在成为更大应用程序一部分的代码片段。
#
1. 内存缓存
此解决方案来自 Python 的 functools 标准库,对于像使用 LLM 这样的昂贵函数很有用。如果我们要在下面定义的函数中使用 LLM API 调用,用 LRU(最近最少使用)装饰器包装它会添加一个缓存机制,该机制可以防止在同一执行或会话中包含相同输入的(提示)的重复请求。这是优化延迟问题的优雅方法。
此示例说明了其用法:
from functools import lru_cache import time @lru_cache(maxsize=100) def summarize_text(text: str) -> str: print("Sending text to LLM...") time.sleep(1) # A simulation of network delay return f"Summary of {len(text)} characters." print(summarize_text("The quick brown fox.")) # Takes one second print(summarize_text("The quick brown fox.")) # Instant
#
2. 磁盘持久化缓存
谈到缓存,外部库 diskcache 通过在磁盘上实现持久化缓存(即通过 SQLite 数据库)将其更进一步:这对于存储耗时函数的(例如 LLM API 调用)结果非常有用。这样,以后需要时就可以快速检索结果。当内存缓存不足时(因为脚本或应用程序的执行可能会停止),请考虑使用此装饰器模式。
import time from diskcache import Cache # Creating a lightweight local SQLite database directory cache = Cache(".local_llm_cache") @cache.memoize(expire=86400) # Cached for 24 hours def fetch_llm_response(prompt: str) -> str: print("Calling expensive LLM API...") # Replace this by an actual LLM API call time.sleep(2) # API latency simulation return f"Response to: {prompt}" print(fetch_llm_response("What is quantum computing?")) # 1st function call print(fetch_llm_response("What is quantum computing?")) # Instant load from disk happens here!
#
3. 网络弹性应用
由于 LLM 经常会因瞬时错误、超时和互联网上的“502 Bad Gateway”响应而失败,因此使用像 tenacity 这样的网络弹性库并结合 @retry 装饰器可以帮助拦截这些常见的网络故障。
下面的示例通过随机模拟 70% 的网络错误几率来说明这种弹性行为的实现。尝试多次,早晚会看到这个错误出现:完全在意料之中,也是故意的!
import random from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception_type class RateLimitError(Exception): pass # Retrying up to 4 times, waiting 2, 4, and 8 seconds between each attempt @retry( wait=wait_exponential(multiplier=2, min=2, max=10), stop=stop_after_attempt(4), retry=retry_if_exception_type(RateLimitError) ) def call_flaky_llm_api(prompt: str): print("Attempting to call API...") if random.random() < 0.7: # Simulating a 70% chance of API failure raise RateLimitError("Rate limit exceeded! Backing off.") return "Text has been successfully generated!" print(call_flaky_llm_api("Write a haiku"))
#
4. 客户端限流
此组合装饰器使用 ratelimit 库来控制对(通常是需求量很大的)函数的调用频率:这对于避免在使用外部 API 时遇到客户端限制非常有用。下面的示例通过定义每分钟请求数 (RPM) 限制来实现这一点。当启动过多的并发请求时,提供者将拒绝来自客户端应用程序的提示。
from ratelimit import limits, sleep_and_retry import time # Strictly enforcing a 3-call limit per 10-second window @sleep_and_retry @limits(calls=3, period=10) def generate_text(prompt: str) -> str: print(f"[{time.strftime('%X')}] Processing: {prompt}") return f"Processed: {prompt}" # First 3 print immediately, the 4th pauses, thereby respecting the limit for i in range(5): generate_text(f"Prompt {i}")
#
5. 结构化输出绑定
列表中的第五个装饰器使用 magentic 库和 Pydantic 结合,通过 API 提供了一种高效的与 LLM 交互的机制,并获得结构化的响应。它简化了调用 LLM API 的过程。这个过程对于可靠地引导 LLM 返回如 JSON 对象之类的格式化数据非常重要。该装饰器将处理底层的系统提示和 Pydantic 驱动的解析,从而优化令牌的使用并有助于保持更清晰的代码库。
要尝试此示例,您需要一个 OpenAI API 密钥。
# IMPORTANT: An OPENAI_API_KEY set is required to run this simulated example from magentic import prompt from pydantic import BaseModel class CapitalInfo(BaseModel): capital: str population: int # A decorator that easily maps the prompt to the Pydantic return type @prompt("What is the capital and population of {country}?") def get_capital_info(country: str) -> CapitalInfo: ... # No function body needed here! info = get_capital_info("France") print(f"Capital: {info.capital}, Population: {info.population}")
#
总结
在本文中,我们列出了并说明了五个基于各种库的 Python 装饰器,它们在 LLM 应用的上下文中具有特殊意义,用于简化逻辑、提高效率或提高网络弹性等方面。
Iván Palomares Carrascosa 是人工智能、机器学习、深度学习和 LLM 领域的领导者、作家、演讲者和顾问。他培训和指导他人掌握现实世界中的人工智能。
🚀 想要体验更好更全面的AI调用?
欢迎使用青云聚合API,约为官网价格的十分之一,支持300+全球最新模型,以及全球各种生图生视频模型,无需翻墙高速稳定,文档丰富,小白也可以简单操作。
评论区