从“有卡”到“跑顺”：上海开源大赛 DaoCloud 赛题背后的智算云工程题

在请求失败率不超过基线的前提下：
1. 将 TTFT P95 降低 20%；
2. 将重复前缀请求的计算量降低 30%；
3. 将模型副本启动时间降低 40%；
4. 不增加单节点峰值显存占用。

#!/usr/bin/env python3
import argparse
import json
import statistics
import time
import urllib.request


def run_once(endpoint, model, token, prompt):
    body = json.dumps({
        'model': model,
        'stream': True,
        'temperature': 0,
        'messages': [
            {'role': 'user', 'content': prompt}
        ]
    }).encode('utf-8')

    headers = {'Content-Type': 'application/json'}
    if token:
        headers['Authorization'] = f'Bearer {token}'

    request = urllib.request.Request(
        endpoint.rstrip('/') + '/v1/chat/completions',
        data=body,
        headers=headers,
        method='POST'
    )

    started = time.perf_counter()
    first_content_at = None
    output = []

    with urllib.request.urlopen(request, timeout=300) as response:
        for raw_line in response:
            line = raw_line.decode('utf-8').strip()
            if not line.startswith('data:'):
                continue

            data = line[5:].strip()
            if data == '[DONE]':
                break

            event = json.loads(data)
            choices = event.get('choices', [])
            if not choices:
                continue

            content = choices[0].get('delta', {}).get('content')
            if content:
                if first_content_at is None:
                    first_content_at = time.perf_counter()
                output.append(content)

    finished = time.perf_counter()
    return {
        'ttft_ms': round((first_content_at - started) * 1000, 2)
        if first_content_at else None,
        'total_ms': round((finished - started) * 1000, 2),
        'output_chars': len(''.join(output))
    }


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--endpoint', required=True)
    parser.add_argument('--model', required=True)
    parser.add_argument('--token', default='')
    parser.add_argument('--rounds', type=int, default=5)
    args = parser.parse_args()

    fixed_prefix = (
        '你是集群故障分析助手。以下规则在每次请求中保持不变：'
        '优先检查资源、调度、网络、存储和缓存，并给出验证命令。'
    ) * 200
    prompt = fixed_prefix + '\n问题：为什么推理服务的首 Token 时延突然升高？'

    results = []
    for index in range(args.rounds):
        result = run_once(
            args.endpoint, args.model, args.token, prompt
        )
        result['round'] = index + 1
        results.append(result)
        print(json.dumps(result, ensure_ascii=False))

    valid_ttft = [r['ttft_ms'] for r in results if r['ttft_ms'] is not None]
    if valid_ttft:
        print(json.dumps({
            'ttft_median_ms': round(statistics.median(valid_ttft), 2),
            'first_round_ms': valid_ttft[0],
            'repeated_round_median_ms': round(
                statistics.median(valid_ttft[1:] or valid_ttft), 2
            )
        }, ensure_ascii=False))


if __name__ == '__main__':
    main()

python3 benchmark_ttft.py \
  --endpoint http://127.0.0.1:8000 \
  --model your-model-name \
  --rounds 8

--token "$INFERENCE_API_TOKEN"

model:
  repository: object-storage/models/example-llm
  revision: sha256-7b8d3f2
  expectedDigest: sha256:REPLACE_WITH_REAL_DIGEST
cache:
  nodeLocalPath: /var/lib/model-cache
  maxSizeGiB: 500
  evictionPolicy: lru