用 Mercury 2.5 跑扩散 LLM 高吞吐推理实战

这篇能做出什么

跟着做完，你手上会有四样东西：

1. 一段能直接跑通的 Mercury 2.5 API 调用代码（curl 与 Python 两版）。

2. 一个参数扫描脚本，用来摸清「去噪步数 / 并行解码宽度 / 输出长度」三者对速度和质量的拉扯。

3. 一个压测脚本，输出首字延迟（TTFT）、单请求输出速率、并发下的聚合吞吐（tokens/s）和 P95 延迟。

4. 一张把扩散 LLM 和自回归模型放在一起对比的选型表，知道什么场景该用哪个。

关于吞吐：扩散 LLM 的设计目标就是并行解码，在合适的并发与参数下，服务端聚合输出速率跑到 700+ tokens/s 这一量级是常见预期。但请记住，这个数字跟你选的模型、并发数、输出长度、网络环境强相关，本文的示例数值只用来说明口径，真实结果以你自己的压测为准。

前置条件清单

• 一个能访问外网的开发机或容器，Python 3.9 以上（以官方文档当前要求为准）。
• Mercury 2.5 的 API Key，以及官方文档给出的 Base URL 与模型名。这两项会变，务必从控制台和官方文档现取，不要照抄教程里的字符串。
• 一个用于对比的自回归模型接口。可以是同一家的对话模型，也可以是任何 OpenAI 兼容接口的服务。
• 基础依赖：

```bash

python -m venv .venv

source .venv/bin/activate # Windows 用 .venv\Scripts\activate

pip install openai httpx asyncio_throttle

```

如果你的环境里已经有 openai 和 httpx，第二条命令可以跳过。

步骤一：先搞清楚扩散 LLM 为什么快

自回归模型生成 100 个 token，就要串行跑 100 次前向。AI 词典：扩散语言模型">扩散语言模型换了个思路：它从一团噪声出发，通过若干轮「去噪」迭代，每一轮同时确定多个位置的 token，几轮到几十轮就能把整段文本定下来。

这带来两个直接后果：

• 单请求的延迟曲线不一样。 自回归是稳定的「每 token 一个固定间隔」，扩散是「前几轮什么都没出，然后一次涌出一批」。所以拿传统 TTFT 去衡量它，含义会有偏差。
• 批量效率高。 因为一轮计算覆盖多个 token，同样的算力能吐出更多内容，服务端在并发场景下的聚合吞吐更可观。

理解了这一点，后面调参就有方向了：你要调的是「跑几轮（去噪步数）」和「每轮放多少位置一起定（并行宽度）」。

步骤二：最小可用调用

先用 curl 确认接口通。字段名以官方文档为准，下面的结构是 OpenAI 兼容接口的常见形态：

```bash

export MERCURY_API_KEY="你的 Key"

export MERCURY_BASE_URL="官方文档给出的 Base URL"

curl -s "$MERCURY_BASE_URL/chat/completions" \

-H "Authorization: Bearer $MERCURY_API_KEY" \

-H "Content-Type: application/json" \

-d '{

"model": "官方文档给出的模型名",

"messages": [

{"role": "user", "content": "用 100 字解释什么是扩散语言模型"}

],

"max_tokens": 256,

"stream": true

}'

```

返回 200 且有内容，说明凭据和网络没问题。返回 401 换 Key，404 多是 Base URL 少了或多了路径段。

Python 版本：

```python

import os

from openai import OpenAI

client = OpenAI(

base_url=os.environ["MERCURY_BASE_URL"],

api_key=os.environ["MERCURY_API_KEY"],

)

resp = client.chat.completions.create(

model=os.environ["MERCURY_MODEL"],

messages=[{"role": "user", "content": "用 100 字解释什么是扩散语言模型"}],

max_tokens=256,

)

print(resp.choices[0].message.content)

print("usage:", resp.usage)

```

步骤三：并行解码相关参数怎么调

扩散 LLM 的可调项通常落在三类上。具体字段名各家不同，先翻官方文档的参数表，再按下面的思路取值。

1）去噪步数（denoising steps / diffusion steps）

这是核心旋钮。步数少，延迟低、吞吐高，但文本可能不连贯；步数多，质量稳，但每一轮都要前向，总耗时上去。

经验做法：从官方默认值出发，做一次向下扫描（例如默认值、默认值的一半、默认值的四分之一），在业务测试集上看质量是否掉到不可接受。很多场景会发现，步数减半之后质量几乎看不出来差别，而延迟省了一大截。

2）并行宽度 / 每步生成 token 数

控制单轮能同时确定多少位置。宽度越大，单轮算力开销越大，但总轮数减少。它和步数是一对需要一起调的参数，单独调一个容易得出错误结论。

3）max_tokens 与温度

max_tokens 直接影响服务端占用时间，压测时一定要固定它，否则吞吐数字没有可比性。温度用于采样多样性，做吞吐基准时建议设成接近 0 或官方推荐的确定性档位。

传参方式（OpenAI 兼容接口多用 extra_body 透传非标字段）：

```python

resp = client.chat.completions.create(

model=os.environ["MERCURY_MODEL"],

messages=[{"role": "user", "content": "写一个二分查找函数"}],

max_tokens=512,

temperature=0,

extra_body={

字段名以官方文档为准，这里演示透传方式

"num_denoising_steps": 8, # 去噪步数

"parallel_tokens": 64, # 并行宽度

},

)

```

如果接口不认这些字段，会直接报参数错误，那就说明命名不同，去文档确认后替换键名即可。

写一个扫描脚本，把结果落成表格：

```python

import itertools, json, time, os

from openai import OpenAI

client = OpenAI(base_url=os.environ["MERCURY_BASE_URL"],

api_key=os.environ["MERCURY_API_KEY"])

PROMPT = "用 300 字说明什么是向量数据库，给出三个使用场景。"

def run(steps, width):

t0 = time.perf_counter()

r = client.chat.completions.create(

model=os.environ["MERCURY_MODEL"],

messages=[{"role": "user", "content": PROMPT}],

max_tokens=400, temperature=0,

extra_body={"num_denoising_steps": steps, "parallel_tokens": width},

)

dt = time.perf_counter() - t0

out = r.choices[0].message.content or ""

return {"steps": steps, "width": width, "sec": round(dt, 3),

"chars": len(out), "text": out}

results = []

for steps, width in itertools.product([4, 8, 16, 32], [32, 64, 128]):

try:

results.append(run(steps, width))

except Exception as e:

print("skip", steps, width, e)

for r in sorted(results, key=lambda x: x["sec"]):

print(r["steps"], r["width"], r["sec"], r["chars"])

json.dump(results, open("sweep.json", "w"), ensure_ascii=False, indent=2)

```

跑完打开 sweep.json，人工读几段文本，把「速度提升明显、质量没塌」的组合圈出来，作为后续压测的配置。

步骤四：单请求打点，量出 TTFT 与输出速率

流式请求才能测首字延迟：

```python

import os, time, statistics

from openai import OpenAI

client = OpenAI(base_url=os.environ["MERCURY_BASE_URL"],

api_key=os.environ["MERCURY_API_KEY"])

def one_shot(prompt, max_tokens=400):

t0 = time.perf_counter()

ttft = None

chunks = 0

stream = client.chat.completions.create(

model=os.environ["MERCURY_MODEL"],

messages=[{"role": "user", "content": prompt}],

max_tokens=max_tokens, temperature=0, stream=True,

)

for ev in stream:

if not ev.choices:

continue

delta = ev.choices[0].delta

if delta and delta.content:

if ttft is None:

ttft = time.perf_counter() - t0

chunks += 1

total = time.perf_counter() - t0

return ttft, total, chunks

ttfts, totals, rates = [], [], []

for i in range(5):

ttft, total, chunks = one_shot("解释一下 Transformer 的注意力机制")

ttfts.append(ttft); totals.append(total)

rates.append(chunks / total)

print("TTFT 中位:", round(statistics.median(ttfts), 3), "s")

print("总耗时中位:", round(statistics.median(totals), 3), "s")

print("输出块速率中位:", round(statistics.median(rates), 1), "chunk/s")

```

注意：扩散模型的流式输出常常是「一批一批」到的，chunks 并不等于 token 数。想要精确的 token 速率，用响应里的 usage.completion_tokens 除以总耗时。

```python

非流式版本，拿 usage 算准确速率

t0 = time.perf_counter()

r = client.chat.completions.create(

model=os.environ["MERCURY_MODEL"],

messages=[{"role": "user", "content": "解释一下 Transformer 的注意力机制"}],

max_tokens=400, temperature=0,

)

dt = time.perf_counter() - t0

n = r.usage.completion_tokens

print("tokens/s:", round(n / dt, 1))

```

单请求能跑出几百 tokens/s 是正常的，因为并行解码把多轮迭代压进了很短的时间里。

步骤五：并发压测，算清聚合吞吐与并发成本

聚合吞吐才是「700+ tokens/s」这类数字的出处：所有并发请求的输出 token 总数 ÷ 压测墙钟时间。

```python

import asyncio, os, time, statistics

import httpx

BASE = os.environ["MERCURY_BASE_URL"]

KEY = os.environ["MERCURY_API_KEY"]

MODEL = os.environ["MERCURY_MODEL"]

PROMPT = "用 200 字说明什么是消息队列，并给出两个使用场景。"

async def worker(client, sem, idx, out):

async with sem:

t0 = time.perf_counter()

ttft = None

tokens = 0

payload = {

"model": MODEL,

"messages": [{"role": "user", "content": PROMPT}],

"max_tokens": 300, "temperature": 0, "stream": True,

需要时在这里加 extra_body 等价字段

}

async with client.stream("POST", f"{BASE}/chat/completions",

json=payload,

headers={"Authorization": f"Bearer {KEY}"}) as r:

r.raise_for_status()

async for line in r.aiter_lines():

if not line.startswith("data:"):

continue

data = line[5:].strip()

if data == "[DONE]":

break

if ttft is None:

ttft = time.perf_counter() - t0

tokens += 1 # 粗略计数，精确值用 usage

total = time.perf_counter() - t0

out.append({"idx": idx, "ttft": ttft, "total": total, "tokens": tokens})

async def bench(concurrency, n=60):

sem = asyncio.Semaphore(concurrency)

out = []

t0 = time.perf_counter()

async with httpx.AsyncClient(timeout=120) as client:

await asyncio.gather(*[worker(client, sem, i, out) for i in range(n)])

wall = time.perf_counter() - t0

ttfts = [o["ttft"] for o in out if o["ttft"]]

toks = sum(o["tokens"] for o in out)

print(f"并发={concurrency} 墙钟={wall:.2f}s "

f"聚合吞吐={toks/wall:.1f} chunk/s "

f"TTFT中位={statistics.median(ttfts):.3f}s "

f"P95={sorted(ttfts)[int(len(ttfts)*0.95)-1]:.3f}s")

asyncio.run(bench(8))

```

把 bench(8) 依次换成 1、2、4、8、16、32，观察三个量：

• 聚合吞吐随并发上升，到某个点开始走平甚至下降 —— 那个点就是你这套配置的容量上限。
• TTFT 中位数会随并发缓慢上升，P95 上升更快，这是排队造成的。
• 单请求总耗时也被拉长，说明并发不是白来的。

并发成本怎么算。 用一个简单口径：单位输出的成本 ≈（实例小时单价 ÷ 该小时内的总输出 token 数）。把不同并发下的聚合吞吐代进去，就能算出「每百万 token 的实例成本」随并发怎么变。单价从你的账单或官方定价页取，写死在脚本外。

步骤六：和自回归模型对比

把同一段压测脚本换个 BASE 和 MODEL 指向自回归模型，跑同样的并发序列，就能得到一张可比表。下面是要重点看的三件事。

首字延迟。 自回归模型的 TTFT 通常很稳定，跟输出长度关系不大；扩散模型因为要跑若干轮去噪，早期可能有一段「沉默期」，但一旦开始吐就是一批。所以在交互式补全这种「用户盯着屏幕等第一个字」的场景，要专门测扩散模型的 TTFT，别只看吞吐。

并发成本。 扩散模型在批量场景下的聚合吞吐更可观，意味着达到同样 QPS 需要的实例更少。但它的算力是「一轮一轮」吃的，峰值占用可能更集中，容量规划要留余量。

适用场景。 对照着看：

一句话选型：吞吐敏感、批量大、对逐字流畅度要求不那么苛刻的任务，优先试扩散；长链路推理和强交互任务，先用自回归。

常见坑与排错

报参数错误但字段名看着没错。 扩散参数通常是各家自定义字段，OpenAI SDK 需要放进 extra_body，直接当顶层关键字传会被过滤掉。

流式返回的 chunk 数不等于 token 数。 扩散模型一批输出多个 token，用 chunk 计数会严重低估速率，务必以 usage.completion_tokens 为准做对比。

压测数字忽高忽低。 常见的三个原因：没固定 max_tokens；没固定温度导致输出长度漂移；客户端并发没做限流，瞬时打爆连接池。把这三项钉死再测。

并发上去吞吐反而掉了。 多半是撞到了服务端限流或排队，也可能是客户端所在的出口带宽打满。先降到并发 4 复现，再逐级升。

本地网络成了瓶颈。 千兆以下的出口 + 大量并发流式连接，会让客户端本身成为限制项。压测尽量放在离服务近的机器上跑。

质量下降但速度上去了。 这是去噪步数减半的必然代价，不要只看平均分，挑几个业务里真正难的 case 人工读一遍。

Key 泄漏。 压测脚本里的 Key 一律走环境变量，不要把字符串硬编码进仓库。

下一步建议

1. 建自己的评测集。 从线上捞 50～100 条真实请求，覆盖短问答、长生成、代码、结构化输出四类，给每条打分。参数扫描只在这套集子上做，结论才有意义。

2. 画曲线而不是记数字。 把「并发 — 聚合吞吐 — P95 延迟」画成折线，容量规划看拐点，别背单次跑出来的那个数。

3. 做 A/B 而不是二选一。 把扩散模型放在批量生成、异步任务、代码补全上，自回归留在长对话和强推理上，两条路并行。

4. 盯住长尾。 高吞吐容易让人只关注平均值，但用户骂的往往是 P99。每轮压测都把 P95、P99 一起记下来。

5. 留意版本变化。 Mercury 2.5 的参数表、默认值、定价都可能迭代，脚本里的配置项建议集中放一个文件，改的时候只改一处。