![]()
邊界限制最先失敗。不是模型。不是 schema。是上限。我不再詢問代理是否「真的在思考」,而是開始問一個更笨的問題,這個問題依然能見血。如果工具掛起,這個 while 會永遠停不下來嗎?
這個問題不浪漫。它也是讓共享的箱子不會變成熱燈的那個問題。我想要一個可以在星期一重新執行的數字,而不是來自已經奉承過我的聊天記錄的感覺。
我讀過的大多數代理重試程式碼都是一個穿著風衣的迴圈。你知道那件風衣。try、except、sleep、continue,也許還有一句註解寫著「要有韌性」。沒有預算的韌性只是無上限的帳單。作者有設定嘗試次數上限嗎?他們有設定執行時間上限嗎?還是他們只限制了自己的樂觀程度?
我做了一個評分器。不是排行榜。是一個有意見的 lint。它讀取 Python,遍歷 AST,並以我希望程式碼審查能做到的方式為重試輔助函式評分:邊界限制優先,詩意永遠不要。產出物才是重點。如果你把這篇文章裡的所有產品名稱都刪掉,你應該還是能儲存檔案並得到相同的整數。
揭露:本文是作為 MonkeyCode 的產品推廣所準備。
當我需要一個我在晚上 11 點不想手寫的候選迴圈時,我會透過 MonkeyCode 詢問一個免費模型。然後我把輸出丟進 fixtures/,並拒絕信任它,直到評分器印出分數為止。免費伺服器選項不是那個步驟的吉祥物。它是實驗的後半段,是 time.sleep 遇到排程雜訊的那一半。我的筆電是個糟糕的證人。它太快了,而且它喜歡我。
我不評「推理品質」。我評的是你即將部署的產出物中是否存在停止條件。它們是不同的論文。其中一篇是主題演講。另一篇是單元測試。
重試輔助函式從零開始。我為嘗試次數上限、執行時間逾時、退避、抖動,以及隨請求一起傳遞的冪等性金鑰加分。我為沒有中斷預算的 while True,以及無法看到截止時間的 sleep 扣分。這很嚴苛嗎?很好。一個能活得比你的耐心還久的迴圈並不穩健。它是一條塗層被剝掉的保險絲。
將此儲存為 loop_hygiene.py。它是整個靜態的一半。
#!/usr/bin/env python3
"""Score retry-loop hygiene from Python source. Fixture-calibrated, not a bake-off."""
from __future__ import annotations
import ast
import json
import sys
from dataclasses import asdict, dataclass, field
from pathlib import Path
ATTEMPT_NAMES = {"max_attempts", "max_retries", "retries", "attempts", "n_tries"}
TIMEOUT_NAMES = {"timeout", "deadline", "max_seconds", "wall_timeout", "budget_s"}
BACKOFF_NAMES = {"backoff", "backoff_s", "delay", "base_delay"}
JITTER_NAMES = {"jitter", "jitter_s", "jitter_ratio"}
IDEM_NAMES = {"idempotency_key", "idempotency", "request_id", "dedupe_key"}
@dataclass
class LoopScore:
path: str
has_max_attempts: bool = False
has_timeout: bool = False
has_backoff: bool = False
has_jitter: bool = False
has_idempotency: bool = False
unbounded_while: bool = False
sleep_without_budget: bool = False
score: int = 0
notes: list[str] = field(default_factory=list)
class HygieneVisitor(ast.NodeVisitor):
def __init__(self) -> None:
self.names: set[str] = set()
self.unbounded_while = False
self.sleep_calls = 0
self.breaks_in_loop = 0
def visit_Name(self, node: ast.Name) -> None:
self.names.add(node.id)
self.generic_visit(node)
def visit_While(self, node: ast.While) -> None:
constant_true = (
isinstance(node.test, ast.Constant) and node.test.value is True
) or (
isinstance(node.test, ast.Constant) and node.test.value == 1
)
if constant_true:
self.unbounded_while = True
for child in ast.walk(node):
if isinstance(child, ast.Break):
self.breaks_in_loop += 1
self.generic_visit(node)
def visit_Call(self, node: ast.Call) -> None:
func = node.func
name = ""
if isinstance(func, ast.Attribute):
name = func.attr
elif isinstance(func, ast.Name):
name = func.id
if name in {"sleep", "usleep"}:
self.sleep_calls += 1
self.generic_visit(node)
def score_source(path: Path, src: str) -> LoopScore:
tree = ast.parse(src)
v = HygieneVisitor()
v.visit(tree)
row = LoopScore(path=str(path))
row.has_max_attempts = bool(ATTEMPT_NAMES & v.names)
row.has_timeout = bool(TIMEOUT_NAMES & v.names)
row.has_backoff = bool(BACKOFF_NAMES & v.names)
row.has_jitter = bool(JITTER_NAMES & v.names)
row.has_idempotency = bool(IDEM_NAMES & v.names)
row.unbounded_while = v.unbounded_while and v.breaks_in_loop == 0
budget = row.has_max_attempts or row.has_timeout
row.sleep_without_budget = v.sleep_calls > 0 and not budget
n = 0
if row.has_max_attempts:
n += 2
row.notes.append("+2 attempt cap")
if row.has_timeout:
n += 2
row.notes.append("+2 wall clock")
if row.has_backoff:
n += 1
row.notes.append("+1 backoff")
if row.has_jitter:
n += 1
row.notes.append("+1 jitter")
if row.has_idempotency:
n += 2
row.notes.append("+2 idempotency key")
if row.unbounded_while:
n -= 3
row.notes.append("-3 unbounded while")
if row.sleep_without_budget:
n -= 2
row.notes.append("-2 sleep with no ceiling")
row.score = n
return row
def main(argv: list[str]) -> int:
root = Path(argv[1] if len(argv) > 1 else "fixtures")
rows = []
for path in sorted(root.glob("*.py")):
rows.append(score_source(path, path.read_text(encoding="utf-8")))
print(json.dumps([asdict(r) for r in rows], indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main(sys.argv))
進入全螢幕模式
退出全螢幕模式
像測試一樣執行它,而不是像示範一樣。
mkdir -p fixtures
python3 loop_hygiene.py fixtures/
進入全螢幕模式
退出全螢幕模式
如果該指令沒有印出任何東西,你得到的是一個空資料夾,而不是及格的分數。空也不是有韌性。
我用四個我故意寫的檔案來校準評分器。這不是關於任何具名模型的宣稱。這是宣稱當我彎曲它時,尺不會彎曲。兩個測試檔是當你說「讓它穩健」時,聊天視窗會吐出的「在 PR 看起來沒問題」的迴圈。另外兩個是我實際願意放在會呼叫我的 cron 下的迴圈。
fixtures/a_trench_coat.py 就是那件風衣。
import time
def call_tool(url):
while True:
try:
return fetch(url)
except Exception:
time.sleep(1)
進入全螢幕模式
退出全螢幕模式
fixtures/b_range_but_naked.py 看起來有邊界,直到你注意到請求可能被收取兩次。
import time
def call_tool(url):
max_attempts = 5
for _ in range(max_attempts):
try:
return fetch(url)
except Exception:
time.sleep(0.5)
raise RuntimeError("gave up")
進入全螢幕模式
退出全螢幕模式
fixtures/c_budget.py 終於承認時間的存在。
import random, time
def call_tool(url, timeout=8.0):
max_attempts = 5
backoff = 0.2
deadline = time.monotonic() + timeout
for attempt in range(max_attempts):
if time.monotonic() >= deadline:
raise TimeoutError("wall clock")
try:
return fetch(url)
except Exception:
jitter = random.random() * 0.05
sleep_for = min(backoff, max(0.0, deadline - time.monotonic()))
time.sleep(sleep_for + jitter)
backoff *= 2
raise RuntimeError("attempts exhausted")
進入全螢幕模式
退出全螢幕模式
fixtures/d_idempotent.py 是我實際會在審查中主張的那一個。
import random, time, uuid
def call_tool(url, timeout=8.0, idempotency_key=None):
max_attempts = 5
backoff = 0.2
jitter = 0.05
deadline = time.monotonic() + timeout
key = idempotency_key or str(uuid.uuid4())
for attempt in range(max_attempts):
if time.monotonic() >= deadline:
raise TimeoutError("wall clock")
try:
return fetch(url, headers={"Idempotency-Key": key})
except Exception:
sleep_for = min(backoff, max(0.0, deadline - time.monotonic()))
time.sleep(sleep_for + random.random() * jitter)
backoff *= 2
raise RuntimeError("attempts exhausted")
進入全螢幕模式
退出全螢幕模式
在我的機器上,針對這四個檔案,評分器印出的分數是 -5, 2, 6, 8。再讀一次。風衣不是「有點草率」。它是負分。審查者喜歡的 range 迴圈得到 2 分,因為它記住了 max_attempts,卻立刻忘記了時間、抖動和重複的副作用。有預算的迴圈是 6 分。冪等的那个是 8 分。這個差距就是實驗在運作。如果你的尺無法分辨保險絲和暖氣機,你只是在收集軼事。
我會部署得到 2 分的程式嗎?不會部署在我沒有監視的箱子上。我會說得到 8 分的程式已可投入生產嗎?還是會說不。一個 AST 訪問器無法看到隱藏在好看標頭名稱後面的非冪等 POST。這個數字是一道閘門,不是一枚勳章。
當我去煩一個免費模型時,我不會叫它「寫一個穩健的代理」。那個提示詞就是你得到風衣的方式。我會要求一個函式,其合約是評分器能看見的。
Write a Python function call_tool(url, timeout=8.0, idempotency_key=None)
that retries a failing fetch(). Requirements:
- max_attempts is an int cap
- timeout is a wall-clock budget using time.monotonic()
- exponential backoff plus jitter
- send Idempotency-Key on every attempt
- no while True
Return only the function.
進入全螢幕模式
退出全螢幕模式
然後我把回傳的任何東西存成 fixtures/model_candidate.py 並執行相同的指令。我不會先讀程式碼周圍的文字。先看整數。如果這聽起來很無禮,請問問自己,為什麼一個重試輔助函式值得擁有單元測試所沒有的禮貌。
模型有加入抖動,還是只是把 sleep(1) 穿上一件更漂亮的外套?它有命名 timeout 卻從未查詢時鐘嗎?名稱很便宜。訪問器更便宜。這就是整個把戲。
靜態分數能抓住缺少上限的情況。它們抓不到紙面上存在、卻在真實延遲面前落敗的上限。所以我加入了一個動態探測。它故意很醜。一個模擬工具會在 0.25 秒、1 秒、3 秒後回應,然後就永遠不回應。被測試的迴圈必須尊重 max_attempts 和執行時間上限,而且行程必須結束。
#!/usr/bin/env python3
"""Dynamic probe. Label: run this; do not treat my laptop timings as yours."""
import threading
import time
from http.server import BaseHTTPRequestHandler, HTTPServer
DELAYS = [0.25, 1.0, 3.0, 999.0]
HITS = {"n": 0}
class SlowTool(BaseHTTPRequestHandler):
def do_GET(self):
i = min(HITS["n"], len(DELAYS) - 1)
HITS["n"] += 1
time.sleep(DELAYS[i])
self.send_response(200)
self.end_headers()
self.wfile.write(b"ok")
def log_message(self, fmt, *args):
return
def main() -> None:
server = HTTPServer(("127.0.0.1", 8765), SlowTool)
t = threading.Thread(target=server.serve_forever, daemon=True)
t.start()
deadline = time.monotonic() + 10.0
# Import YOUR candidate here and call it against http://127.0.0.1:8765/
# If this process is still alive past deadline, the loop has no ceiling.
while time.monotonic() < deadline:
time.sleep(0.05)
server.shutdown()
print({"hits": HITS["n"], "exited_before_watchdog": True})
if __name__ == "__main__":
main()
進入全螢幕模式
退出全螢幕模式
在我的筆電上,3 秒的延遲是一個禮貌的暫停。在共享的免費伺服器上,那個暫停會與其他人的工作以及箱子實際擁有的任何監控程式發生碰撞。如果你的停止條件是「我看轉圈看膩了」,伺服器不會替你看膩。這就是為什麼免費伺服器應該屬於這個方法,而不是屬於一句口號。延遲是證人。安靜的 SSD 是被告的朋友。
我不會為那一半發布對抗賽表格,因為我不會捏造模型名稱、硬體規格或永久性宣稱。協定就是結果。你讓候選程式對抗模擬工具。要嘛行程在 10 秒監控程式之前結束,不然你就是剛剛看著一個無邊界迴圈花掉別人的電。你希望被哪個結果驚訝,是整數還是生產環境?
這沒有測量一個模型是否「比大多數開發者更會寫程式」。我完全不知道你怎麼抽樣那句話而不對自己說謊。它沒有測量 MCP 伺服器、社群知識庫,或代理是否其實是 if 陳述式。很多 if 陳述式都很誠實。不誠實的是那些拒絕成為帶有計數器的 if 陳述式的迴圈。
AST 評分器是一支手電筒。它會錯過藏在 exec、我沒有命名的裝飾器,或是因為有人認為那是架構而重新啟動自己的子行程中的重試。免費模型輸出會變化。免費伺服器不是 SLO。如果你需要固定的模型身分、保留的 CPU,或是會標明延遲的廠商合約,不要使用這個工作流程。請使用付費的、具名的端點,以及一個你可以 SSH 進去而不用猜測還有誰在同一台機器上的箱子。
誰應該跳過它:任何在出貨計費、醫療保健,或是圍繞非冪等收費的重試的人。8 分的衛生分數不是 PCI 稽核。如果你不打算閱讀產生的迴圈,也請跳過它。一個你忽略的評分器只是另一個儀表板,而儀表板不會在發票來之前呼叫你。
我仍然會在爭論「代理品質」之前先執行評分器。邊界限制在聊天中是可選的。在行程表中卻不是。如果你希望動態的那一半在不是你筆電的機器上變得醜陋,我已經把那個探測放在 MonkeyCode 的免費伺服器上,並讓延遲保持不誠實。先偷走這個測試架構。再來和我爭論評分標準。
https://dev.to/hackhub_6179/i-scored-retry-loops-boundedness-was-optional-pe7
https://www.worldprogramming.org/posts/i-scored-retry-loops-boundedness-was-optional-uihs26