Dakota Huang

Posted on Sep 4

Cache the Repetition: A Quota-Saving Proxy for Free Model Endpoints

#ai #python #tutorial

重复请求浪费免费配额。缓存响应节省时间和令牌。本教程构建一个零依赖缓存代理。它运行在免费服务器上。它拦截对免费模型端点的调用。

免费模型端点有速率限制。许多请求是重复的。例如，测试时相同的提示词被多次发送。缓存可以避免重复计算。这减少了令牌消耗。也降低了延迟。

The Problem: Duplicate Calls Burn Quota

每个免费模型端点都有速率限制。你的应用可能多次发送相同请求。比如，一个脚本在循环中调用同一个提示词。或者多个服务共享同一个端点。缓存是自然的解决方案。

一个简单的缓存代理可以拦截请求。它检查是否已经缓存了相同的请求体。如果是，它返回缓存响应。如果不是，它转发到上游并存储结果。

Step 1: Design the Cache Key

缓存键必须唯一标识一个请求。使用请求体的SHA256哈希。忽略HTTP头，因为提示词在请求体中。

import hashlib
key = hashlib.sha256(payload).hexdigest()

如果请求包含时间戳或随机数，缓存将永远不命中。确保你的客户端不发送这些。或者从缓存键中排除它们。

Step 2: Build the Proxy with Python's http.server

Python标准库提供了HTTP服务器。我们用它构建一个轻量代理。它监听POST请求。它检查SQLite缓存。它转发到上游模型端点。

#!/usr/bin/env python3
import hashlib
import json
import os
import sqlite3
import time
import urllib.request
from http.server import BaseHTTPRequestHandler, HTTPServer

UPSTREAM = os.environ["MODEL_ENDPOINT"]
TOKEN = os.environ.get("MODEL_TOKEN", "")
DB_PATH = os.environ.get("CACHE_DB", "cache.db")
TTL = int(os.environ.get("CACHE_TTL", "3600"))

conn = sqlite3.connect(DB_PATH, check_same_thread=False)
conn.execute("CREATE TABLE IF NOT EXISTS cache (key TEXT PRIMARY KEY, body TEXT, created INTEGER)")

class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        length = int(self.headers.get("Content-Length", 0))
        payload = self.rfile.read(length)
        key = hashlib.sha256(payload).hexdigest()
        row = conn.execute("SELECT body, created FROM cache WHERE key=?", (key,)).fetchone()
        if row and time.time() - row[1] < TTL:
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("X-Cache", "HIT")
            self.end_headers()
            self.wfile.write(row[0].encode())
            return
        # forward to upstream
        req = urllib.request.Request(UPSTREAM, data=payload, method="POST")
        req.add_header("Content-Type", "application/json")
        if TOKEN:
            req.add_header("Authorization", f"Bearer {TOKEN}")
        try:
            with urllib.request.urlopen(req, timeout=30) as resp:
                body = resp.read()
                status = resp.status
        except urllib.error.HTTPError as e:
            body = e.read()
            status = e.code
        if status == 200:
            conn.execute("INSERT OR REPLACE INTO cache VALUES (?,?,?)", (key, body.decode(), int(time.time())))
            conn.commit()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("X-Cache", "MISS")
        self.end_headers()
        self.wfile.write(body)
    def log_message(self, format, *args):
        pass

if __name__ == "__main__":
    server = HTTPServer(("0.0.0.0", int(os.environ.get("PORT", 8080))), Handler)
    print("Cache proxy running on :8080")
    server.serve_forever()

这个代理只有约50行。它没有外部依赖。它使用SQLite进行持久化。它支持TTL。

Step 3: Add TTL and Persistence

SQLite数据库存储缓存条目。每个条目包含键、响应体和创建时间。TTL检查在查询时进行。过期条目被忽略。

你可以添加一个清理函数。定期删除旧条目。这防止数据库无限增长。

def cleanup():
    conn.execute("DELETE FROM cache WHERE created < ?", (time.time() - TTL,))
    conn.commit()

在服务器启动时调用一次。或者使用cron定期调用。

Step 4: Deploy to a Free Server

我使用MonkeyCode的免费服务器选项来运行这个代理。它提供了一个持久的shell环境。这适合运行长时间运行的进程。

Disclosure: This article was prepared as part of MonkeyCode's product outreach.

任何免费服务器都可以。你需要Python 3和SQLite。大多数免费服务器都内置了。

将脚本上传到服务器。设置环境变量。启动进程。

export MODEL_ENDPOINT="https://your-model-endpoint"
export MODEL_TOKEN="your-token"
export PORT=8080
python3 cache_proxy.py

使用nohup或systemd保持运行。

Step 5: Verify with a Test Script

验证缓存是否工作。发送相同请求两次。检查X-Cache头。

curl -s -X POST http://localhost:8080 -d '{"prompt":"What is TCP?"}' -H "Content-Type: application/json" -D - | grep X-Cache
curl -s -X POST http://localhost:8080 -d '{"prompt":"What is TCP?"}' -H "Content-Type: application/json" -D - | grep X-Cache

第一次应该返回MISS。第二次应该返回HIT。

你还可以检查延迟。缓存命中通常快得多。

Limitations and Who Should Skip This

缓存只适用于确定性输出。如果模型有随机性，响应可能不同。设置较短的TTL可以减轻问题。

缓存键基于请求体。如果请求包含非确定参数，缓存会失效。确保你的客户端发送稳定的请求。

免费服务器没有持久化保证。重启后SQLite文件可能丢失。使用外部存储或接受丢失。

这个代理不处理认证。如果你的端点需要认证，确保代理正确传递。或者让代理自己认证。

谁不应该使用这个？需要实时数据的应用。需要每次不同响应的应用。有严格合规要求的团队。

对于大多数开发工作流，缓存是一个简单的胜利。它减少配额消耗。它降低延迟。它让你在免费配额下做更多事情。

你的缓存命中率是多少？在下方留言。

MonkeyCode provides free models that can run this workflow.

DEV Community