Cerebras 万亿参数模型 API 接入、压测与成本对照

Cerebras AI 词典：Inference">Inference 把超大模型跑在自研的晶圆级芯片上，对外提供 OpenAI 兼容的 HTTP 接口。也就是说，不需要采购显卡、不用装驱动、不用管显存，只要能发 HTTPS 请求，就能调用总参数达到万亿量级的 MoE 模型。下面是一套可以照着敲一遍的流程：拿 Key、跑通第一个请求、选模型、做并发压测，最后把「按 token 计费」和「按 GPU 小时计费」折算到同一把尺子上比较。

适用场景

适合三类人：一是想在现有 OpenAI 兼容代码上换一个推理后端做 A/B 对比的团队，改两行配置就能切；二是对首 token 延迟和输出吞吐敏感的业务，比如实时对话、代码补全、批量摘要；三是需要在「Serverless 按量计费」和「GPU 云自建」之间做成本决策的技术负责人。核心解决的问题是：不建集群的前提下，把万亿参数模型接进业务，并用数据而不是感觉来判断该走哪条路。

环境与前置条件

• 操作系统：Linux 或 macOS 均可；Windows 建议用 WSL2，避免终端与代理配置的坑。
• 运行时：Python 3.9 及以上（具体下限以官方 SDK 的说明为准），能正常使用 pip 与虚拟环境。
• 本机资源：因为推理在云端完成，压测机不需要 GPU、不需要显存。建议压测机至少 2 vCPU / 2 GB 内存；如果要压到上百并发，注意调高文件描述符上限。
• 磁盘：几百 MB 即可，主要是虚拟环境和日志。
• 网络：能出网访问 api.cerebras.ai 的 443 端口。如果公司网络有出口代理，注意代理本身可能限制并发连接数，会污染压测结果。
• 账号：一个可以接收验证邮件的账号，用于在控制台创建 API Key。免费额度、速率限制档位以控制台和官方文档当前展示为准。
• 成本预算：压测会产生真实调用量，先在控制台确认当前额度，别一上来就开高并发。

分步骤部署

第 1 步：创建 API Key 并写入环境变量

登录 Cerebras Cloud 控制台，在 API Keys 页面创建一个新 Key。Key 通常只在创建时完整显示一次，复制后立刻存进密码管理器。

不要把它写进代码，也不要提交到 Git。用环境变量传递：

```bash

export CEREBRAS_API_KEY="你的密钥"

export MODEL="你从模型列表里选定的模型 id"

```

把这两行写进 ~/.bashrc 或项目的 .env（.env 要加进 .gitignore）。检查一下是否生效：

```bash

echo ${#CEREBRAS_API_KEY}

```

输出应该是一个大于 20 的数字（长度），而不是 0。如果是 0，说明变量没加载成功。

这一步的成功标志：环境变量存在，且下面第 3 步的 curl 能返回 200。

第 2 步：准备虚拟环境与依赖

这一步在做什么：把官方 SDK 和 OpenAI SDK 装进一个隔离环境，避免和系统里其他 Python 包打架。

```bash

mkdir -p ~/cerebras-bench && cd ~/cerebras-bench

python3 -m venv .venv

source .venv/bin/activate

pip install --upgrade pip

pip install cerebras-cloud-sdk openai python-dotenv

```

包名是 cerebras-cloud-sdk，导入名是 cerebras.cloud.sdk，两者不一致是新手最容易卡住的地方，别按包名去 import。

同时装 openai 是因为 Cerebras 的接口兼容 OpenAI 协议，很多存量代码只改 base_url 和 model 就能迁移，省一大笔改造量。

成功标志：pip show cerebras-cloud-sdk 能打印出包信息，且 which python 指向 .venv/bin/python。

第 3 步：用 curl 打通最小可行请求

先用最原始的方式确认网络、鉴权和模型名三件事都对，再上 Python。这样出问题时能快速定位是哪一层。

先列模型：

```bash

curl -s https://api.cerebras.ai/v1/models \

-H "Authorization: Bearer $CEREBRAS_API_KEY" | head -c 2000

```

返回的 JSON 里会有一个 data 数组，里面每项的 id 就是可用的模型名。把其中一个赋值给 MODEL：

```bash

export MODEL="上面返回的某个 id"

```

再发一次对话请求：

```bash

curl -s https://api.cerebras.ai/v1/chat/completions \

-H "Authorization: Bearer $CEREBRAS_API_KEY" \

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

-d '{

"model": "'"$MODEL"'",

"messages": [{"role": "user", "content": "用一句话解释什么是 MoE 模型"}],

"max_tokens": 128

}'

```

成功标志：返回 JSON 里有 choices[0].message.content 且内容非空，同时 usage 里 completion_tokens 大于 0。

注意模型 id 会随官方更新变化，不要硬编码进代码，一律走环境变量或配置中心。

第 4 步：接入 Python SDK

新建 hello.py：

```python

import os

from cerebras.cloud.sdk import Cerebras

client = Cerebras(api_key=os.environ["CEREBRAS_API_KEY"])

resp = client.chat.completions.create(

model=os.environ["MODEL"],

messages=[

{"role": "system", "content": "你是简洁的技术助手，回答不超过三句话。"},

{"role": "user", "content": "解释一下什么是首 token 延迟（TTFT）"},

],

max_tokens=256,

temperature=0.2,

)

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

print(resp.usage)

```

运行 python hello.py，能看到一段中文回答和 usage 对象，就说明 SDK 这一层通了。

如果团队已有基于 OpenAI SDK 的代码，改成这样即可：

```python

import os

from openai import OpenAI

client = OpenAI(

base_url="https://api.cerebras.ai/v1",

api_key=os.environ["CEREBRAS_API_KEY"],

)

```

其余调用方式与原来完全一致。这一步的价值在于：迁移成本几乎为零，可以先小流量灰度再全量。

第 5 步：选模型，先小后大

选型看四个维度，不要只看参数量：

1. 架构：万亿参数级的 MoE 模型每次推理只激活一小部分参数，单 token 的计算量远小于总参数，这正是它能跑出高吞吐的原因。

2. 上下文长度：决定你能塞多长的文档。超长输入会同时推高延迟和费用。

3. 能力特性：是否支持工具调用（function calling）、结构化输出（JSON mode）。这两项直接影响能不能接进生产链路。

4. 限速档位：不同模型的速率限制可能不一样，以控制台和官方文档当前说明为准。

实操建议：调试阶段用列表里较小的模型，把脚本、重试逻辑、指标采集全部调通，最后一步再切到万亿参数模型。否则每次调试都在烧大模型的额度。

第 6 步：封装带超时与退避的调用函数

压测时一定会遇到 429，生产环境也一定会。把重试逻辑收进一个函数，别在业务代码里到处写 try。

```python

import os, time, random

from cerebras.cloud.sdk import Cerebras

client = Cerebras(api_key=os.environ["CEREBRAS_API_KEY"])

def chat_with_retry(messages, max_tokens=512, retries=5):

delay = 1.0

last_err = None

for attempt in range(retries):

try:

return client.chat.completions.create(

model=os.environ["MODEL"],

messages=messages,

max_tokens=max_tokens,

temperature=0.2,

)

except Exception as e:

last_err = e

status = getattr(e, "status_code", None)

参数错误、鉴权错误重试没有意义，直接抛

if status in (400, 401, 403, 404):

raise

if attempt == retries - 1:

raise

sleep = min(delay * (2 ** attempt), 30) + random.random()

print(f"[retry {attempt+1}] status={status} sleep={sleep:.1f}s")

time.sleep(sleep)

raise last_err

```

判断标准：status 是 429 或 5xx 才退避重试，4xx 直接抛出来，因为重试只会浪费时间。

第 7 步：写压测脚本，测 TTFT 和输出吞吐

关键点：必须开流式（stream），否则拿不到首 token 延迟。同时固定 prompt 和 max_tokens，否则不同请求之间的数据没法比。

新建 bench.py：

```python

import os, time, statistics, csv

from concurrent.futures import ThreadPoolExecutor

from cerebras.cloud.sdk import Cerebras

client = Cerebras(api_key=os.environ["CEREBRAS_API_KEY"])

MODEL = os.environ["MODEL"]

固定输入：同样的问题、同样的输出上限，保证可比

PROMPT = "请写一段约 300 字的产品说明，主题是一款面向中小团队的日志分析工具。要求分三段。"

MAX_TOKENS = 256

def one_call(_):

t0 = time.perf_counter()

ttft = None

chars = 0

stream = client.chat.completions.create(

model=MODEL,

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

max_tokens=MAX_TOKENS,

temperature=0.2,

stream=True,

)

for chunk in stream:

if not chunk.choices:

continue

delta = chunk.choices[0].delta.content

if delta:

if ttft is None:

ttft = time.perf_counter() - t0

chars += len(delta)

total = time.perf_counter() - t0

return {"ttft": ttft, "total": total, "chars": chars}

def percentile(values, p):

values = sorted(v for v in values if v is not None)

if not values:

return float("nan")

k = int(round((p / 100) * (len(values) - 1)))

return values[k]

def run(concurrency, n):

t_start = time.perf_counter()

with ThreadPoolExecutor(max_workers=concurrency) as ex:

results = list(ex.map(one_call, range(n)))

wall = time.perf_counter() - t_start

ttfts = [r["ttft"] for r in results]

total_chars = sum(r["chars"] for r in results)

return {

"concurrency": concurrency,

"requests": n,

"wall_s": round(wall, 2),

"ttft_p50": round(percentile(ttfts, 50), 3),

"ttft_p95": round(percentile(ttfts, 95), 3),

"chars_per_s": round(total_chars / wall, 1),

"req_per_s": round(n / wall, 2),

}

if __name__ == "__main__":

rows = []

for c in [1, 2, 4, 8, 16, 32]:

r = run(concurrency=c, n=max(c * 4, 8))

print(r)

rows.append(r)

with open("bench_result.csv", "w", newline="") as f:

w = csv.DictWriter(f, fieldnames=rows[0].keys())

w.writeheader()

w.writerows(rows)

```

运行前先放开文件描述符上限，否则并发一高就报 Too many open files：

```bash

ulimit -n 65535

python bench.py

```

说明几点：

• 用字符数近似输出量，够用来做相对比较。要精确的 token 数，单独跑一次非流式请求读 usage.completion_tokens，再算一个「字符 → token」的换算比例，套回压测结果。
• 并发梯度从小往大扫，别一上来就 32。目的是找到吞吐曲线变平、p95 延迟开始陡增的那个拐点。
• 如果压测中途大量 429，说明已经触到账号的速率限制，这个并发值就是当前档位的实际上限，而不是模型的极限。要更高并发得看控制台的提额说明。

第 8 步：做成本对照，把两种计费方式拉到同一把尺子上

Serverless 推理按 token 计费，GPU 云按 GPU 小时计费，直接比数字没意义。统一折算成每百万输出 token 的成本。

云端 API 侧，从官方定价页读到输入/输出的单价（价格会调整，以官方页面为准）：

```python

def api_cost_per_million(price_per_million_output_tokens,

input_tokens_ratio=0.3,

price_per_million_input_tokens=0.0):

折算成「按输出 token 摊」的综合单价

return (price_per_million_output_tokens

+ input_tokens_ratio * price_per_million_input_tokens)

```

GPU 云侧，先在自己租的机器上跑同一套 bench.py，得到稳定的输出速率，再折算：

```python

def gpu_cost_per_million(gpu_hour_price, tokens_per_sec, utilization=0.7):

"""

gpu_hour_price: 单卡每小时价格

tokens_per_sec: 你自己实测的单卡输出吞吐（token/s）

utilization: 有效利用率，扣除冷启动、排队、空转

"""

tokens_per_hour = tokens_per_sec * 3600 * utilization

return gpu_hour_price / tokens_per_hour * 1_000_000

def breakeven_tokens_per_month(api_price_per_million, gpu_hour_price,

hours_per_month=730):

"""24 小时包月的 GPU 用满多少 token 才比按量调用便宜"""

gpu_month = gpu_hour_price * hours_per_month

return gpu_month / api_price_per_million * 1_000_000

```

必须强调：tokens_per_sec 只能自己测。同尺寸模型在不同框架、不同批处理配置、不同输入输出长度下差好几倍，拿别人的跑分来算成本会得出完全错误的结论。测的时候输入长度、输出长度都要和业务真实分布对齐。

对照表可以按这个维度铺开：

结论的写法建议是：先算出盈亏平衡的月 token 量，再叠加「运维人力」和「峰值弹性」两个非货币因素。很多团队的实际情况是——日常量用按量调用，大促或批处理任务再临时开 GPU，混合使用。

验证部署是否成功

按顺序跑这几条，全过就算部署完成：

1. 鉴权与连通性

```bash

curl -s -o /dev/null -w "%{http_code}\n" \

https://api.cerebras.ai/v1/models \

-H "Authorization: Bearer $CEREBRAS_API_KEY"

```

预期输出：200。出现 401 说明 Key 有问题。

2. 模型列表非空

```bash

curl -s https://api.cerebras.ai/v1/models \

-H "Authorization: Bearer $CEREBRAS_API_KEY" \

| python -c "import sys,json; d=json.load(sys.stdin); print(len(d['data'])); print([m['id'] for m in d['data']][:5])"

```

预期输出：先是一个大于 0 的数字，然后是若干个模型 id。

3. 一次完整对话

```bash

python hello.py

```

预期输出：一段中文回答，以及 usage 对象，其中 completion_tokens 明显大于 0。

4. 压测脚本能跑出指标

```bash

ulimit -n 65535 && python bench.py

```

预期输出：6 行字典，每行包含 concurrency、ttft_p50、ttft_p95、chars_per_s。正常情况下，随着并发上升 chars_per_s 先涨后平，ttft_p95 在拐点后明显变大。同时目录下生成 bench_result.csv。

5. 密钥没泄漏到代码里

```bash

grep -rn "CEREBRAS_API_KEY" . --include="*.py"

```

预期输出：只出现 os.environ["CEREBRAS_API_KEY"] 这类读取方式，不出现明文密钥字符串。

常见报错与解决

报错：401 Unauthorized / invalid api key

→ 原因：环境变量没加载、复制时带了多余的空格或引号、或者在 .env 里写了 export 前缀被重复解析。

→ 解决：

```bash

unset CEREBRAS_API_KEY

export CEREBRAS_API_KEY="重新从控制台复制的密钥"

echo ${#CEREBRAS_API_KEY} # 确认长度正常

python hello.py

```

报错：404 model_not_found / model does not exist

→ 原因：模型 id 拼错，或该模型已从列表下线（模型 id 会随官方更新变化）。

→ 解决：

```bash

curl -s https://api.cerebras.ai/v1/models \

-H "Authorization: Bearer $CEREBRAS_API_KEY" \

| python -c "import sys,json; [print(m['id']) for m in json.load(sys.stdin)['data']]"

export MODEL="从上面输出里挑一个"

```

报错：429 Too Many Requests / rate limit exceeded

→ 原因：并发或每分钟 token 数超过了账号当前档位的限制。

→ 解决：先降并发确认是限速而不是代码问题，再启用第 6 步的指数退避；需要更高额度按控制台说明申请。

```bash

压测时把并发梯度改小，例如从 [1,2,4,8] 开始

python bench.py

```

报错：400 context_length_exceeded 或提示输入过长

→ 原因：输入 token 加上 max_tokens 超过了模型上下文窗口。

→ 解决：截断输入或调小 max_tokens：

```python

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

resp = client.chat.completions.create(model=MODEL, messages=messages, max_tokens=512)

```

报错：ModuleNotFoundError: No module named 'cerebras'

→ 原因：装的是 cerebras-cloud-sdk 但导入路径写成了 cerebras，或者根本没激活虚拟环境。

→ 解决：

```bash

source .venv/bin/activate

pip show cerebras-cloud-sdk

python -c "from cerebras.cloud.sdk import Cerebras; print('ok')"

```

报错：压测时 OSError: [Errno 24] Too many open files

→ 原因：并发连接数超过进程文件描述符上限。

→ 解决：

```bash

ulimit -n 65535

ulimit -n # 确认已生效

```

报错：连接超时 / SSLError / Connection reset

→ 原因：公司出口代理限制了长连接数，或 DNS 解析异常。

→ 解决：先用 curl 单独验证网络层，再决定是否配代理：

```bash

curl -v https://api.cerebras.ai/v1/models \

-H "Authorization: Bearer $CEREBRAS_API_KEY" 2>&1 | head -30

如需代理

export HTTPS_PROXY="http://你的代理地址:端口"

```

后续维护

密钥管理：至少每季度轮换一次 API Key；旧 Key 在新 Key 生效后再删除。密钥只放环境变量或密钥管理服务，.env 必须进 .gitignore。CI 里用平台提供的 secret 机制，不要回显。

额度与成本监控：每次调用都把 usage.prompt_tokens、usage.completion_tokens 写进日志，按天聚合。设定日消耗阈值告警，避免异常重试把额度烧光。每月用真实账单除以真实 token 数，回头验证第 8 步的估算是否仍然成立——单价和模型都会变。

日志字段建议：至少记录 request_id、model、prompt_tokens、completion_tokens、ttft_ms、total_ms、status_code、retry_count。prompt 内容按合规要求脱敏或不落盘。

降级与熔断：主模型连续 429 或 5xx 时，自动切到列表里的小尺寸模型或排队重试，保证业务不中断。熔断阈值用压测得到的 p95 延迟做基准。

升级策略：升级 SDK 前看 changelog；模型 id 变更时先灰度 5% 流量，对比延迟和输出质量再放量。SDK 版本号以官方文档当前版本为准，别盲目 pip install -U 到生产。

压测常态化：把 bench.py 放进仓库，每次换模型、改 prompt 模板、调整 max_tokens 之后重跑一遍，结果追加到 CSV。攒够几个月的曲线，容量规划和成本预测就不用拍脑袋了。

数据合规：确认业务数据能否出境、能否经过第三方推理服务，这部分以服务条款和公司法务意见为准，不要在技术层面自行判断。