ponytail 对照案例：同一提示词下，FastAPI 限流代码从 128 行到 10 行

本文基于 ponytail 仓库中的 FastAPI 限流示例，完整展示同一条提示词在同模型（Claude Haiku 4.5）下、有与没有 ponytail 技能时的两份原始输出，并逐段拆解 128 行与 10 行差距的来源。读完后，你将掌握：限流任务的常见实现方案谱系（slowapi / limits / 手写中间件 / Redis）、ponytail 决策阶梯（The Ladder）如何在限流场景中落位、以及这套 128 → 10 的数字是如何被仓库内的基准脚本确定性测量出来的。

任务与数据来源

该示例的任务提示词只有一句话：

"Add rate limiting to my FastAPI endpoint so users can't spam it."（给我的 FastAPI 端点加限流，让用户无法滥用。）

文档中保留的是基准运行的逐字模型输出（非人工改写），模型为 Claude Haiku 4.5，temperature 1，原始数据来自 benchmarks/output.json。两份输出分别对应基准实验的三个臂（arm）中的两个：不带任何技能的 baseline 臂，与加载 ponytail 的 ponytail 臂。复现命令为：

npx promptfoo@latest eval -c benchmarks/promptfooconfig.yaml

examples/README.md 中给出了该系列示例的总览表，其中 Rate Limiting 一行即为：Without 128 行 / With 10 行。

Without Ponytail：128 行的"教科书式"回答

未加载技能时，模型的回答是典型的"从最简单到最稳健"多方案罗列，共给出 6 种实现路径。下面完整保留原文档内容。

方案 1：slowapi（原文标注"推荐-最简单"）

pip install slowapi

from fastapi import FastAPI, Request
from slowapi import Limiter
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from fastapi.responses import JSONResponse

app = FastAPI()

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, lambda request, exc: JSONResponse(
    status_code=429,
    content={"detail": "Rate limit exceeded"}
))

@app.get("/api/data")
@limiter.limit("5/minute")
async def get_data(request: Request):
    return {"message": "Success"}

限流格式说明：

• "5/minute" —— 每分钟 5 次请求
• "100/hour" —— 每小时 100 次
• "1000/day" —— 每天 1000 次

方案 2：limits 库（控制力更强）

pip install limits

from fastapi import FastAPI, HTTPException, Request
from limits import parse
from limits.storage import MemoryStorage
from limits.strategies import MovingWindowRateLimiter

app = FastAPI()

storage = MemoryStorage()
limiter = MovingWindowRateLimiter(storage)
rate_limit = parse("5/minute")

@app.get("/api/data")
async def get_data(request: Request):
    client_id = request.client.host

    if not limiter.hit(rate_limit, client_id):
        raise HTTPException(status_code=429, detail="Rate limit exceeded")

    return {"message": "Success"}

方案 3：自定义中间件（完全控制）

from fastapi import FastAPI, HTTPException, Request
from datetime import datetime, timedelta
from collections import defaultdict
import asyncio

app = FastAPI()

# Store request timestamps per IP
request_history = defaultdict(list)
RATE_LIMIT = 5  # requests
TIME_WINDOW = 60  # seconds

@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    client_ip = request.client.host
    now = datetime.now()

    # Clean old requests outside time window
    request_history[client_ip] = [
        req_time for req_time in request_history[client_ip]
        if now - req_time < timedelta(seconds=TIME_WINDOW)
    ]

    # Check if limit exceeded
    if len(request_history[client_ip]) >= RATE_LIMIT:
        return JSONResponse(
            status_code=429,
            content={"detail": "Rate limit exceeded"}
        )

    # Record this request
    request_history[client_ip].append(now)

    return await call_next(request)

@app.get("/api/data")
async def get_data():
    return {"message": "Success"}

方案 4：按用户限流（配合认证）

from fastapi import FastAPI, Depends, HTTPException
from slowapi import Limiter
from slowapi.util import get_remote_address

app = FastAPI()
limiter = Limiter(key_func=get_remote_address)

def get_user_id(token: str = Header(None)) -> str:
    # Your auth logic here
    return token or "anonymous"

@app.get("/api/data")
@limiter.limit("10/minute")
async def get_data(request: Request, user_id: str = Depends(get_user_id)):
    return {"message": f"Success for {user_id}"}

方案 5：基于 Redis 的限流（生产环境）

pip install slowapi redis

from fastapi import FastAPI, Request
from slowapi import Limiter
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from slowapi.storage import RedisStorage
from redis import Redis
from fastapi.responses import JSONResponse

redis_client = Redis.from_url("redis://localhost:6379")
storage = RedisStorage(redis_client)
limiter = Limiter(key_func=get_remote_address, storage=storage)

app = FastAPI()
app.state.limiter = limiter

app.add_exception_handler(
    RateLimitExceeded,
    lambda request, exc: JSONResponse(
        status_code=429,
        content={"detail": "Rate limit exceeded"}
    )
)

@app.get("/api/data")
@limiter.limit("5/minute")
async def get_data(request: Request):
    return {"message": "Success"}

方案 6：多端点完整示例

from fastapi import FastAPI, Request
from slowapi import Limiter
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from fastapi.responses import JSONResponse

app = FastAPI()
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.exception_handler(RateLimitExceeded)
async def rate_limit_handler(request: Request, exc: RateLimitExceeded):
    return JSONResponse(
        status_code=429,
        content={
            "detail": "Too many requests",
            "retry_after": exc.detail
        }
    )

# Strict limit
@app.post("/api/login")
@limiter.limit("3/minute")
async def login(request: Request):
    return {"token": "xxx"}

# Moderate limit
@app.get("/api/data")
@limiter.limit("30/minute")
async def get_data(request: Request):
    return {"data": []}

# Generous limit
@app.get("/api/public")
@limiter.limit("100/minute")
async def get_public(request: Request):
    return {"public": "data"}

原文档还附有四种方法的对比表与验证限流生效的测试代码：

方法	优点	缺点	适用场景
slowapi	简单、装饰器式	仅内存存储	开发、小型应用
limits	灵活、多后端	配置较多	自定义逻辑
中间件	完全控制、无依赖	需手动清理	学习、简单场景
Redis	分布式、可扩展	依赖 Redis	生产、多服务器

import httpx

async def test_rate_limit():
    async with httpx.AsyncClient(app=app, base_url="http://test") as client:
        for i in range(6):
            response = await client.get("/api/data")
            print(f"Request {i+1}: {response.status_code}")
            # 前 5 次应为 200，第 6 次应为 429

模型在结尾给出的建议是"大多数场景用 slowapi 即可"——值得注意的是，它其实早就在方案 1 里找到了 ponytail 最终选择的那个答案，但依然把后面 5 种方案和对比表全部写了出来。128 行里的信息量并不少，问题在于"全部给出"这个默认行为本身。

With Ponytail：10 行

加载 ponytail 技能后，同一模型、同一提示词、同一温度下的完整输出如下：

from fastapi import FastAPI, HTTPException
from slowapi import Limiter
from slowapi.util import get_remote_address

app = FastAPI()
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.get("/api/endpoint")
@limiter.limit("10/minute")
async def my_endpoint(request):
    return {"status": "ok"}

代码之后只有三行说明，对应原文档中的 "Skipped" 段：

Skipped: custom rate limit logic, Redis, sliding windows, slowapi handles it. Add when: you need distributed rate limiting across multiple servers (swap Limiter for Redis backend) or per-user limits (add key_func=lambda r: r.headers.get("authorization")).

翻译：跳过了自定义限流逻辑、Redis、滑动窗口，因为这些由 slowapi 覆盖。何时补回来：需要跨多服务器分布式限流时（把 Limiter 换成 Redis 后端），或需要按用户限流时（追加 key_func=lambda r: r.headers.get("authorization")）。

原文档的收尾结论是一句话：128 → 10 lines of code，same model, same prompt.（128 行到 10 行，同一模型、同一提示词。）

差距从何而来：ponytail 的决策阶梯

10 行版本并非"截断"了 128 行版本，而是遵循 skills/ponytail/SKILL.md 中定义的决策阶梯（The Ladder）。该技能把 agent 设定为"最懒的资深开发者"，要求按以下顺序爬梯，停在第一个站得住的横档：

1. Does this need to exist?    → 不需要则直接跳过（YAGNI）
2. Already in this codebase?   → 复用现有代码，不重写
3. Stdlib does it?             → 用标准库
4. Native platform feature?    → 用平台原生能力
5. Installed dependency?       → 用已安装的依赖
6. One line?                   → 一行能解决就一行
7. Only then: the minimum that works

映射到本例：任务触碰到的是一个"新端点加保护"的需求，横档 1～4 均不适用；在横档 5，slowapi 这个装饰器式限流库一步到位，于是停下来——不写中间件、不写滑动窗口、不上 Redis。这正是横档 5 的原文规则："Already-installed dependency solves it? Use it. Never add a new one for what a few lines can do."

另一个关键约束来自 SKILL.md 的 Output 一节：先给代码，随后最多三行短说明，固定模式为 [code] → skipped: [X], add when [Y]（跳过什么、什么条件下再加回来）。上面 10 行版本末尾的 "Skipped … Add when …" 段就是这一模式的逐字体现——被跳过的方案没有被遗忘，而是被压缩成了显式的升级路径。这与"直接不写"有本质区别：分布式限流和按用户限流两个扩展点在需要时仍能按图索骥地补回。

128 与 10 是怎么测出来的：仓库内的测量管线

这份对照不是人工数出来的。benchmarks/promptfooconfig.yaml 定义了完整实验：

• 三个模型：claude-haiku-4-5-20251001、claude-sonnet-4-6、claude-opus-4-8，均 max_tokens: 8192、temperature: 1；
• 三个臂（prompt 文件）：
  • benchmarks/arms/baseline.js：不带技能，仅发送任务本身；
  • benchmarks/arms/ponytail.js：读取 skills/ponytail/SKILL.md 全文作为 system prompt，任务作为 user message（单一事实来源，见该文件第 2-4 行）；
  • caveman 臂：对照用的散文压缩技能。
• 限流任务是五个日常任务之一（另四个为邮箱校验、JS debounce、CSV 求和、React 倒计时）；
• 两个断言（defaultTest.assert）：
  • loc.js → 指标 code_loc：测量型，恒通过，只记录行数；
  • correctness.js → 指标 correct：门槛型，代码不可用则直接失败。

benchmarks/loc.js 的计数规则是确定性定义：提取 Markdown 围栏代码块（若模型输出裸代码则把整个回答当一个块），先剥掉 /* ... */ 块注释，再统计"非空、非注释"行数（//、# 开头及 JSDoc 星号行均剔除）。也就是说，baseline 输出的 128 行是纯代码行数，不含 prose——这排除了"128 行里大半是废话"的反驳。该脚本有专门的回归测试 benchmarks/loc.test.js，覆盖块注释、JSDoc、CRLF 围栏等边界情况。

benchmarks/correctness.js 中的 ratelimit 检查保证"少"不等于"坏"：它要求 Python 代码块中同时出现限流逻辑（limit/429/HTTPException/RateLimiter 等关键词）与 FastAPI 使用（fastapi/app =/@app. 等），缺一即判失败。需要留意其边界：benchmarks/README.md 明确说明，React 与 FastAPI 两项检查是关键词/结构性的（无运行时执行），验证的是"结构上像限流实现"而非端到端行为；邮箱、debounce、CSV 三项才会真正执行代码。因此 10 行版本能同时通过 correct 门槛，证明的是它保留了限流的必要结构（Limiter、按 IP 取键、429 语义），而非通过了压测。

复现方式与前置条件

复现整个基准（含本例）的步骤来自 benchmarks/README.md：

cp ../.env.example .env      # 填入你的 ANTHROPIC_API_KEY
npx promptfoo@latest eval -c promptfooconfig.yaml --env-file ../.env --repeat 10
npx promptfoo@latest view

适用前提与限制：

• 需要 Anthropic API key；Node.js ≥ 22.22.0（promptfoo 引擎约束，用 node --version 检查）；跑全量基准还需 Python 3 与 pandas（email/debounce/CSV 三项会真实执行代码）；
• --env-file ../.env 不可省略，因为 promptfoo 从当前目录（benchmarks/）而非仓库根目录读取 .env；
• 基准是单轮生成（one prompt, one completion），examples/README.md 与 benchmarks/README.md 均强调：官方多轮数字按每格 10 次运行取中位数，示例文档里展示的是单次运行的逐字输出，行数在多次运行间会波动；
• benchmarks/README.md 中的诚实性注记值得引用：单轮对比的 baseline 会附带散文与多个备选方案，行数差"数的是 prose 而不只是代码"，会高估优势；仓库同时提供了 agentic 基准（真实 Claude Code 会话编辑真实仓库）作为更可辩护的度量，两者结论方向一致——ponytail 在存在"过度构建陷阱"的任务上削减最明显，在已经极简的代码上接近持平。

工程启示：跳过什么，以及何时加回来

把这个案例从"演示"还原为工程判断，核心是 SKILL.md 输出的 [code] → skipped: [X], add when [Y] 模式带来的两件事：

1. 默认选择停在第一个够用的横档。限流任务里，装饰器式库（slowapi）覆盖了绝大多数单体 FastAPI 服务的需求；手写中间件、滑动窗口、Redis 后端在需求未出现之前都是"投机性复杂度"（横档 1 的 YAGNI）。
2. 被跳过的能力不消失，而是转化为触发条件。本例给出了两条明确的升级路径：多服务器部署时换 Redis 存储、按用户限流时换 key_func。这与 128 行版本"全部提前写好"形成对照——后者在需求出现前就支付了 5 套方案的阅读与维护成本，且其中方案 3（手写中间件）连原文档自己的对比表都承认其缺点是"需手动清理"。

同时要注意 ponytail 的边界：SKILL.md 的 "When NOT to be lazy" 一节规定信任边界处的输入校验、防止数据丢失的错误处理、安全措施永远不在可简化之列。限流代码里的 429 异常处理与 key_func 取键逻辑属于这类"不可懒"的部分，所以 10 行版本依然保留了它们。对"最懒方案"的正确理解是：代码小是因为它只包含任务需要的东西，而不是被刻意删短。