本地服务、兼容接口与维护：练习答案

参考实现

以下仅测试预算规则，不启动模型。

def check_budget(input_tokens, output_tokens, window):
    if min(input_tokens,output_tokens,window) < 0:
        raise ValueError('negative budget')
    return input_tokens+output_tokens <= window
assert check_budget(100,28,128)
assert not check_budget(100,29,128)
try:
    check_budget(-1,20,128)
except ValueError:
    pass
else:
    raise AssertionError('negative accepted')

token数必须来自对应模板和tokenizer；字符串长度只适合作粗警报。

返回题目。

链路变体

服务运行时执行下面客户端，只有超预算分支才接受400：

import json
from urllib.request import Request,urlopen
from urllib.error import HTTPError
url='http://127.0.0.1:8087/generate'
def send(prompt,n):
    return json.load(urlopen(Request(url,data=json.dumps({'prompt':prompt,'max_new_tokens':n}).encode(),headers={'Content-Type':'application/json'})))
assert send('Refund policy:',1)['output_tokens']==1
try:
    send('hello '*200,8)
except HTTPError as exc:
    assert exc.code==400
else:
    raise AssertionError('over-budget request accepted')

错误目录在启动载入阶段抛异常，尚未绑定端口。不要同时保留旧服务再测错误目录，否则旧/health响应会让你误判新进程启动成功。

HTTP长度边界

先启动正文服务，另存http_limits.py运行。原始socket避免高级客户端自动修正Content-Length；无效请求不发送声明的大体积内容。

import socket
cases=[('',400),('Content-Length: -1\r\n',400),
       ('Content-Length: xyz\r\n',400),('Content-Length: 8193\r\n',413),
       ('Content-Length: 2\r\nContent-Length: 2\r\n',400),
       ('Transfer-Encoding: chunked\r\n',400)]
for headers,expected in cases:
    with socket.create_connection(('127.0.0.1',8087),timeout=5) as s:
        s.sendall(('POST /generate HTTP/1.1\r\nHost: localhost\r\n'+headers+'\r\n').encode())
        response=s.recv(4096)
        assert int(response.split(b' ',2)[1])==expected,response
print('length headers rejected before body read')

错误响应关闭连接；负数不会传给read，非数字不会逃出处理器。可以另发Content-Length:2但不发送体，超过读取等待后应收到408。该实验没有证明高并发服务防护；生产需要连接数限制、总deadline与受控队列。