66 lines
3.5 KiB
Python
66 lines
3.5 KiB
Python
"""Small live API acceptance suite; writes evidence without API credentials."""
|
|
import argparse
|
|
import json
|
|
import time
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
p = argparse.ArgumentParser()
|
|
p.add_argument('--url', default='http://127.0.0.1:8000')
|
|
p.add_argument('--key-file', type=Path, default=Path(__file__).resolve().parents[1] / 'secrets/api-key')
|
|
p.add_argument('--output', type=Path, required=True)
|
|
args = p.parse_args()
|
|
key = args.key_file.read_text().strip()
|
|
|
|
def request(path, body=None):
|
|
payload = None if body is None else json.dumps(body, ensure_ascii=False).encode()
|
|
req = urllib.request.Request(args.url + path, data=payload,
|
|
headers={'Authorization': 'Bearer ' + key, 'Content-Type': 'application/json'})
|
|
return urllib.request.urlopen(req, timeout=600)
|
|
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
with args.output.open('x') as out:
|
|
def record(data):
|
|
line = json.dumps(data, ensure_ascii=False)
|
|
out.write(line + '\n'); out.flush(); print(line, flush=True)
|
|
|
|
record({'models': json.load(request('/v1/models'))})
|
|
for name, text, expected in [
|
|
('math', 'Compute 17 * 19. Reply with only the integer.', '323'),
|
|
('chinese', '请只回复：模型已就绪', '模型已就绪'),
|
|
]:
|
|
body = dict(model='qwen3.8-flash-next', messages=[dict(role='user', content=text)],
|
|
temperature=0, reasoning_effort='low', max_tokens=1024,
|
|
stream=True, stream_options={'include_usage': True})
|
|
start = time.monotonic(); first = None; content = ''; reasoning = ''; usage = {}; finish = None
|
|
with request('/v1/chat/completions', body) as response:
|
|
for raw in response:
|
|
if not raw.startswith(b'data: '): continue
|
|
data = raw[6:].strip()
|
|
if data == b'[DONE]': break
|
|
item = json.loads(data)
|
|
if item.get('usage'): usage = item['usage']
|
|
for choice in item.get('choices', []):
|
|
delta = choice.get('delta', {})
|
|
c = delta.get('content') or ''
|
|
r = delta.get('reasoning') or delta.get('reasoning_content') or ''
|
|
if (c or r) and first is None: first = time.monotonic()
|
|
content += c; reasoning += r
|
|
finish = choice.get('finish_reason') or finish
|
|
elapsed = time.monotonic() - start
|
|
passed = content.strip() == expected and finish == 'stop'
|
|
record(dict(test=name, passed=passed, content=content, reasoning_chars=len(reasoning),
|
|
ttft_s=None if first is None else first-start, elapsed_s=elapsed,
|
|
finish_reason=finish, usage=usage))
|
|
if not passed: raise SystemExit('Acceptance failure: ' + name)
|
|
body = dict(model='qwen3.8-flash-next', temperature=0, reasoning_effort='low', max_tokens=1024,
|
|
messages=[dict(role='user',content='Use get_weather to check the weather in Shanghai.')],
|
|
tools=[dict(type='function',function=dict(name='get_weather',description='Get current weather',
|
|
parameters=dict(type='object',properties={'city':{'type':'string'}},required=['city'])))],
|
|
tool_choice='auto')
|
|
result = json.load(request('/v1/chat/completions',body))
|
|
calls = result['choices'][0]['message'].get('tool_calls') or []
|
|
passed = bool(calls) and calls[0]['function']['name']=='get_weather'
|
|
record(dict(test='tool',passed=passed,response=result))
|
|
if not passed: raise SystemExit('Tool acceptance failure')
|