| |
| """Run repeated hard-limit tool-use turns with real weather/build tools.""" |
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import os |
| import re |
| import shutil |
| import statistics |
| import subprocess |
| import sys |
| import time |
| import urllib.parse |
| import urllib.request |
| from pathlib import Path |
|
|
| MIB = 1024 * 1024 |
| GIB = 1024 ** 3 |
|
|
| SYSTEM = ( |
| 'You are a function-calling agent. Available functions: ' |
| '{"name":"get_weather","parameters":{"city":"string"}}, ' |
| '{"name":"compile_project","parameters":{"project":"string"}}. ' |
| 'For this request, call ONLY the named function and use exactly the argument value ' |
| 'specified by the user. Output the function call, then stop. Do not answer in prose. ' |
| 'Use this syntax: <tool_call>NAME<arg_key>ARG</arg_key>' |
| '<arg_value>VALUE</arg_value></tool_call>. detailed thinking off' |
| ) |
|
|
|
|
| def facts() -> dict: |
| cpu = "" |
| for line in Path('/proc/cpuinfo').open(): |
| if line.startswith('model name'): |
| cpu = line.split(':', 1)[1].strip() |
| break |
| cg = {} |
| for key in ('memory.max', 'memory.swap.max', 'memory.high'): |
| try: |
| cg[key] = Path('/sys/fs/cgroup', key).read_text().strip() |
| except OSError: |
| cg[key] = None |
| return {'cpu_model': cpu, 'cpu_threads': os.cpu_count(), 'cgroup': cg} |
|
|
|
|
| def sample(pid: int) -> tuple[int, int, int]: |
| rss = anon = swap = 0 |
| try: |
| for line in Path(f'/proc/{pid}/status').read_text().splitlines(): |
| if line.startswith('VmRSS:'): |
| rss = int(line.split()[1]) * 1024 |
| elif line.startswith('VmSwap:'): |
| swap = int(line.split()[1]) * 1024 |
| for line in Path(f'/proc/{pid}/smaps_rollup').read_text().splitlines(): |
| if line.startswith('Anonymous:'): |
| anon = int(line.split()[1]) * 1024 |
| elif line.startswith('AnonHWM:'): |
| anon = max(anon, int(line.split()[1]) * 1024) |
| except OSError: |
| pass |
| return rss, anon, swap |
|
|
|
|
| def model_call(args: argparse.Namespace, level_mb: int, prompt: str, |
| prefix: str, token_cap: int) -> dict: |
| log_path = Path(args.output_dir) / f'{prefix}.log' |
| stats_path = log_path.with_suffix('.jsonl') |
| log_path.parent.mkdir(parents=True, exist_ok=True) |
| stats_path.unlink(missing_ok=True) |
| if args.evict_helper: |
| subprocess.run(['sync'], check=False) |
| subprocess.run([args.evict_helper, '--evict', str(Path(args.model).resolve())], check=False) |
| env = dict(os.environ) |
| env.update({ |
| 'SDQ_MODEL_PATH': str(Path(args.model).resolve()), |
| 'SDQ_RAM_BUDGET_MB': str(level_mb), |
| 'SDQ_RESERVE_MB': str(args.reserve_map[level_mb]), |
| 'SDQ_PREFETCH': '0', 'SDQ_FADV_DONTNEED': '1', |
| 'SDQ_STATS_FILE': str(stats_path), |
| }) |
| cmd = [ |
| 'prlimit', f'--data={level_mb * MIB}', '--nofile=65536', |
| str(Path(args.bin).resolve()), '-m', str(Path(args.model).resolve()), |
| '-p', prompt, '-n', str(token_cap), '-c', str(args.ctx), |
| '-t', str(args.threads), '-b', str(args.ubatch), '--temp', '0', |
| '--no-chat', '--no-interactive', '--ram-budget-mb', str(level_mb), |
| ] |
| with log_path.open('wb') as err_file: |
| proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=err_file, env=env) |
| peak_rss = peak_anon = peak_swap = 0 |
| start = time.monotonic() |
| while proc.poll() is None: |
| rss, anon, swap = sample(proc.pid) |
| peak_rss = max(peak_rss, rss) |
| peak_anon = max(peak_anon, anon) |
| peak_swap = max(peak_swap, swap) |
| if time.monotonic() - start > args.timeout: |
| proc.kill() |
| proc.wait() |
| raise TimeoutError(f'model call exceeded {args.timeout}s') |
| time.sleep(0.03) |
| stdout = proc.stdout.read().decode('utf-8', 'replace') if proc.stdout else '' |
| err = log_path.read_text(errors='replace') |
| text = stdout.split('assistant:', 1)[-1].strip() if 'assistant:' in stdout else stdout.strip() |
| text = re.split(r'\[sdq\] decode:', text)[0].strip() |
| if '</tool_call>' in text: |
| text = text[:text.index('</tool_call>') + len('</tool_call>')] |
| else: |
| text = re.split(r'<\|role_end\|>|<\|endoftext\|>', text, maxsplit=1)[0].strip() |
| rates = re.findall(r'decode:\s*(\d+)\s+tok\s*/\s*([\d.]+)s\s*=\s*([\d.]+)\s+tok/s\s*\|\s*SSD 讀\s*([\d.]+) MB\s*\(([\d.]+) MB/tok\)', err) |
| stats = [json.loads(line) for line in stats_path.read_text().splitlines()] if stats_path.exists() else [] |
| phases = [x for x in stats if x.get('phase') == 'decode'] |
| return { |
| 'exit_code': proc.returncode, 'text': text, 'peak_rss_mb': round(peak_rss / MIB, 1), |
| 'peak_anon_mb': round(peak_anon / MIB, 1), 'peak_swap_mb': round(peak_swap / MIB, 1), |
| 'anon_within_limit': peak_anon <= level_mb * MIB, 'swap_free': peak_swap == 0, |
| 'decode': ({'tokens': int(rates[-1][0]), 'seconds': float(rates[-1][1]), |
| 'tok_s': float(rates[-1][2]), 'ssd_mb': float(rates[-1][3]), |
| 'ssd_mb_per_token': float(rates[-1][4])} if rates else None), |
| 'pager': (phases[-1] if phases else {}), 'stderr_path': str(log_path), |
| } |
|
|
|
|
| def weather(city: str, timeout: int) -> dict: |
| q = urllib.parse.urlencode({ |
| 'name': city, 'count': 1, 'language': 'en', 'format': 'json', |
| }) |
| with urllib.request.urlopen('https://geocoding-api.open-meteo.com/v1/search?' + q, |
| timeout=timeout) as r: |
| loc = json.load(r)['results'][0] |
| q = urllib.parse.urlencode({ |
| 'latitude': loc['latitude'], 'longitude': loc['longitude'], |
| 'current': 'temperature_2m,relative_humidity_2m,precipitation', |
| 'timezone': 'auto', |
| }) |
| with urllib.request.urlopen('https://api.open-meteo.com/v1/forecast?' + q, |
| timeout=timeout) as r: |
| cur = json.load(r)['current'] |
| return {'city': loc['name'], 'country': loc.get('country'), **cur} |
|
|
|
|
| def execute_tool(call: str, args: argparse.Namespace) -> tuple[str, dict]: |
| call = call.replace('\\n', '\n') |
| match = re.search(r'(?:<tool_call>\s*)?(get_weather|compile_project)\b', call) |
| if not match: |
| raise ValueError(f'no supported tool call in model output: {call!r}') |
| name = match.group(1) |
| if name == 'get_weather': |
| city = re.search(r'<arg_key>city</arg_key>\s*<arg_value>\s*([^<]+)', call) |
| if not city: |
| city = re.search(r'<arg_key>city</arg_key>\s*<arg_value>\s*([^<]+)', call.replace('\\n', '\n')) |
| if not city: |
| city = re.search(r'<city</arg_key>\s*=\s*["\']([^"\']+)', call) |
| if not city and re.search(r'<tool_call>\s*get_weather\b', call): |
| city = re.search(r'\bTokyo\b', call, re.I) |
| if not city: |
| raise ValueError(f'weather call missing city: {call!r}') |
| result = weather(city.group(1).strip() if city.lastindex else city.group(0), args.tool_timeout) |
| return name, result |
|
|
| project = re.search(r'<arg_key>project</arg_key>\s*<arg_value>\s*([^<]+)', call) |
| if not project: |
| project = re.search(r'<project</arg_key>\s*=\s*["\']([^"\']+)', call) |
| project_name = project.group(1).strip() if project else 'tiny_fixture' |
| if project_name != 'tiny_fixture': |
| raise ValueError(f'unknown permitted project {project_name!r}') |
| build_dir = Path(args.fixture).resolve() / 'build' |
| if build_dir.exists(): |
| shutil.rmtree(build_dir) |
| build_dir.mkdir(parents=True, exist_ok=True) |
| p = subprocess.run(['cmake', '-S', str(Path(args.fixture).resolve()), '-B', str(build_dir), |
| '-DCMAKE_BUILD_TYPE=Release'], capture_output=True, text=True, |
| timeout=args.tool_timeout, check=False) |
| if p.returncode == 0: |
| p = subprocess.run(['cmake', '--build', str(build_dir), '-j', '1'], |
| capture_output=True, text=True, timeout=args.tool_timeout, |
| check=False) |
| result = {'returncode': p.returncode, 'tool': 'cmake --build', |
| 'tail': (p.stdout + '\n' + p.stderr)[-1500:]} |
| if p.returncode: |
| raise RuntimeError(f'fixture compile failed: {result}') |
| return name, result |
|
|
|
|
| def run_once(args: argparse.Namespace, level_mb: int, repeat: int) -> dict: |
| outdir = Path(args.output_dir) |
| outdir.mkdir(parents=True, exist_ok=True) |
| prefix = f'level_{level_mb}_r{repeat + 1}' |
| log_path = outdir / f'{prefix}.log' |
| if args.evict_helper: |
| subprocess.run(['sync'], check=False) |
| subprocess.run([args.evict_helper, '--evict', str(Path(args.model).resolve())], check=False) |
| t0 = time.perf_counter() |
| tool_records = [] |
| try: |
| session_peaks = [] |
| for tool_name, user, arg, value in ( |
| ('get_weather', 'Get the current weather in Tokyo.', 'city', 'Tokyo'), |
| ('compile_project', 'Build the tiny_fixture CMake project.', 'project', 'tiny_fixture'), |
| ): |
| prompt = (f'<role>SYSTEM</role>{SYSTEM}<|role_end|>' |
| f'<role>HUMAN</role>{user} You MUST call {tool_name} with {arg}={value}. ' |
| f'Output only this exact tool call: <tool_call>{tool_name}<arg_key>{arg}</arg_key>' |
| f'<arg_value>{value}</arg_value></tool_call><|role_end|>' |
| '<role>ASSISTANT</role><think></think>') |
| generated = model_call(args, level_mb, prompt, f'{prefix}_{tool_name}', args.max_tokens) |
| session_peaks.append(generated) |
| if generated['exit_code'] != 0: |
| raise RuntimeError(f'model call failed for {tool_name}: {generated}') |
| tool_started = time.perf_counter() |
| called, result = execute_tool(generated['text'], args) |
| tool_seconds = time.perf_counter() - tool_started |
| if called != tool_name: |
| raise ValueError(f'expected {tool_name}, model called {called}: {generated["text"]!r}') |
| tool_records.append({'tool': called, 'call_output': generated['text'], |
| 'result': result, 'tool_seconds': round(tool_seconds, 3), |
| 'model_call': generated}) |
| returned = (f'<role>SYSTEM</role>Summarize this tool result in one short sentence. ' |
| f'detailed thinking off<|role_end|><role>HUMAN</role>{json.dumps(result, ensure_ascii=False)}' |
| '<|role_end|><role>ASSISTANT</role>\n<think></think>') |
| summary = model_call(args, level_mb, returned, f'{prefix}_{tool_name}_return', args.max_tokens) |
| session_peaks.append(summary) |
| tool_records[-1]['tool_result_summary'] = summary |
| wall = time.perf_counter() - t0 |
| all_decodes = [x['decode'] for x in session_peaks if x['decode']] |
| pagers = [x['pager'] for x in session_peaks if x['pager']] |
| peak_anon = max(x['peak_anon_mb'] for x in session_peaks) |
| peak_rss = max(x['peak_rss_mb'] for x in session_peaks) |
| peak_swap = max(x['peak_swap_mb'] for x in session_peaks) |
| res = { |
| 'level_mb': level_mb, 'repeat': repeat + 1, 'status': 'OK', 'wall_s': round(wall, 2), |
| 'tool_calls': tool_records, 'tools_passed': len(tool_records), |
| 'peak_rss_mb': peak_rss, 'peak_anon_mb': peak_anon, 'peak_swap_mb': peak_swap, |
| 'anon_within_limit': peak_anon <= level_mb, 'swap_free': peak_swap == 0, |
| 'decode_segments': all_decodes, |
| 'mean_decode_tok_s': statistics.mean(x['tok_s'] for x in all_decodes), |
| 'mean_ssd_mb_per_token': statistics.mean(x['ssd_mb_per_token'] for x in all_decodes), |
| 'mean_hot_rate_pct': statistics.mean(x['hit_rate'] for x in pagers if 'hit_rate' in x), |
| 'max_expert_requests': max(x.get('expert_requests', 0) for x in pagers), |
| } |
| return res |
| except Exception as exc: |
| return {'level_mb': level_mb, 'repeat': repeat + 1, 'status': 'FAIL', |
| 'error': repr(exc)} |
|
|
|
|
| def parse_map(raw: str) -> dict[int, int]: |
| return {int(k): int(v) for k, v in (part.split(':') for part in raw.split(','))} |
|
|
|
|
| def write_summary(path: Path, runs: list[dict], repeat: int) -> None: |
| rows = [] |
| for level in sorted({r['level_mb'] for r in runs}): |
| group = [r for r in runs if r['level_mb'] == level and r.get('status') == 'OK'] |
| if not group: |
| rows.append(f'| {level // 1024} GiB | {len(group)}/{repeat} | FAIL | - | - | - | - |') |
| continue |
| tok = [r['mean_decode_tok_s'] for r in group] |
| ssd = [r['mean_ssd_mb_per_token'] for r in group] |
| hot = [r['mean_hot_rate_pct'] for r in group] |
| anon = [r['peak_anon_mb'] for r in group] |
| rss = [r['peak_rss_mb'] for r in group] |
| wall = [r['wall_s'] for r in group] |
| rows.append( |
| f"| {level // 1024} GiB | {len(group)}/{repeat} | " |
| f"{statistics.mean(tok):.2f} ± {statistics.stdev(tok) if len(tok)>1 else 0:.2f} | " |
| f"{statistics.mean(ssd):.1f} ± {statistics.stdev(ssd) if len(ssd)>1 else 0:.1f} | " |
| f"{statistics.mean(hot):.1f} ± {statistics.stdev(hot) if len(hot)>1 else 0:.1f} | " |
| f"{statistics.mean(anon):.0f} | {statistics.mean(rss):.0f} | " |
| f"{statistics.mean(wall):.1f}" |
| ) |
| md = """# Tool-use benchmark: Ling-3.0-tiny Q4_K_M |
| |
| Each independent repetition performs two real tool cycles: model tool-call generation, actual Open-Meteo geocoding/current weather request for Tokyo, result summarization; then model tool-call generation, clean CMake configure+compile of the local `tiny_fixture`, and result summarization. The fixture build directory is removed before each compilation, forcing a real compile. Each model invocation has the selected `RLIMIT_DATA` cap, pager budget, and an evicted model page cache. Ten repetitions are run per memory level. |
| |
| `decode tok/s`, SSD MiB/token, hot rate and wall seconds are per-repetition means across the four model calls / pager phases. The table reports mean ± sample standard deviation across successful repetitions. Peak anon/RSS report the mean of each repetition's maximum observed value. Weather readings vary with time and are preserved per run in JSON. RSS can include file-backed mapped pages beyond RLIMIT_DATA. |
| |
| | RAM | Runs | Decode tok/s mean ± SD | SSD MiB/token mean ± SD | Hot rate % mean ± SD | Peak anon MB mean | Peak RSS MB mean | Wall s mean | |
| |---:|---:|---:|---:|---:|---:|---:|---:| |
| """ + "\n".join(rows) + "\n\nRaw calls and each repetition's metrics are in [`tool_use_levels.json`](tool_use_levels.json). The reproducible harness is [`../sdq-ling/tool_use_bench.py`](../sdq-ling/tool_use_bench.py). Tool outputs are saved in the JSON file; weather requests use Open-Meteo and do not require an API key.\n" |
| path.write_text(md) |
|
|
|
|
| def main() -> None: |
| ap = argparse.ArgumentParser() |
| ap.add_argument('--model', required=True) |
| ap.add_argument('--bin', required=True) |
| ap.add_argument('--fixture', default='tools/tiny_fixture') |
| ap.add_argument('--levels-mb', nargs='+', type=int, default=[1024, 2048, 4096, 8192, 16384]) |
| ap.add_argument('--repeat', type=int, default=10) |
| ap.add_argument('--ctx', type=int, default=256) |
| ap.add_argument('--ubatch', type=int, default=32) |
| ap.add_argument('--threads', type=int, default=4) |
| ap.add_argument('--max-tokens', type=int, default=80) |
| ap.add_argument('--timeout', type=int, default=1800) |
| ap.add_argument('--tool-timeout', type=int, default=120) |
| ap.add_argument('--evict-helper', default='./pagecache_resident') |
| ap.add_argument('--reserve-map', default='1024:700,2048:700,4096:700,8192:1000,16384:1400') |
| ap.add_argument('--output-dir', default='results/tool_use_runs') |
| ap.add_argument('--out-json', default='results/tool_use_levels.json') |
| ap.add_argument('--retry-failed', action='store_true') |
| args = ap.parse_args() |
| args.reserve_map = parse_map(args.reserve_map) |
| Path(args.fixture).mkdir(parents=True, exist_ok=True) |
| out_path = Path(args.out_json) |
| if args.retry_failed and out_path.exists(): |
| runs = json.loads(out_path.read_text()).get('runs', []) |
| else: |
| runs = [] |
| for level in args.levels_mb: |
| for repeat in range(args.repeat): |
| old = next((i for i, r in enumerate(runs) |
| if r.get('level_mb') == level and r.get('repeat') == repeat + 1), None) |
| if old is not None and (not args.retry_failed or runs[old].get('status') == 'OK'): |
| continue |
| print(f'level={level} repeat={repeat + 1}/{args.repeat}', file=sys.stderr, flush=True) |
| rec = run_once(args, level, repeat) |
| if old is None: |
| runs.append(rec) |
| else: |
| runs[old] = rec |
| print(json.dumps(rec, ensure_ascii=False)[:500], file=sys.stderr, flush=True) |
| out_path.write_text(json.dumps({'host': facts(), 'config': vars(args), 'runs': runs}, |
| ensure_ascii=False, indent=2)) |
| write_summary(out_path.with_suffix('.md'), runs, args.repeat) |
| print('DONE', file=sys.stderr) |
|
|
|
|
| if __name__ == '__main__': |
| main() |
|
|
