这篇教程带你走完一条完整的迁移链路:先用最少的代码把 Mistral Large 4 的对话接口跑通,再把现有工作流里的提示词和评测集原样搬过来做一次对照,最后得到一张能拿给团队看的成本与时延对比表。
这篇能做出什么
照着往下做,你会得到四样东西:
1. 一个能直接复制运行的调用脚本,支持普通返回与流式返回两种模式;
2. 一层薄薄的供应商抽象,让同一个提示词可以在旧模型和新模型之间一键切换;
3. 一份基于官方定价页填数的成本估算脚本,输出「每条请求花多少钱」而不用手算;
4. 一份评测跑批结果 CSV,包含延迟、token 用量、原始输出,方便人工打分或接自动打分。
不追求一步到位,重点是先把链路打通,再谈调优。
前置条件清单
开工前确认这几项:
- 一个 Mistral 平台的账号,并在控制台里创建 API Key;
- 本机有 Python 3.10 及以上,或者 Node.js 18 及以上(两者选一即可,示例两种都给);
- 有
curl,方便在写代码前先验证网络与鉴权是否通; - 网络能访问
https://api.mistral.ai; - 一份你现有工作流的提示词,以及一份包含 20~100 条真实问题的评测集(JSONL 格式即可)。
需要提前说明的是:模型 ID 的具体字符串、AI 词典:上下文窗口">上下文窗口大小、是否支持某个参数、单价与速率限制,都以官方文档当前版本为准。本文里所有涉及这些的地方都用环境变量占位,你照着官方页面替换即可。
第一步:跑通最小可运行调用
1.1 准备环境变量
```bash
export MISTRAL_API_KEY="在控制台创建的 Key"
export MISTRAL_BASE_URL="https://api.mistral.ai/v1"
export MODEL_ID="以官方文档当前可用的模型 ID 为准"
```
把这三行写进 ~/.bashrc 或 .env 文件,避免每次开新终端都要重设。
1.2 用 curl 先验证一次
```bash
curl -s "$MISTRAL_BASE_URL/chat/completions" \
-H "Authorization: Bearer $MISTRAL_API_KEY" \
-H "Content-Type: application/json" \
-d "{
\"model\": \"$MODEL_ID\",
\"messages\": [
{\"role\": \"system\", \"content\": \"你是简洁的中文技术助理,回答控制在三句话内。\"},
{\"role\": \"user\", \"content\": \"解释一下什么是向量数据库。\"}
],
\"temperature\": 0.3,
\"max_tokens\": 512
}" | python -m json.tool
```
返回体里重点看两个字段:choices[0].message.content 是正文,usage 里有 prompt_tokens、completion_tokens、total_tokens。后面算成本全靠 usage。
1.3 Python 版:不装依赖也能跑
用标准库就够了,省掉环境折腾:
```python
chat_client.py
import os, json, time, urllib.request, urllib.error
BASE = os.environ["MISTRAL_BASE_URL"].rstrip("/")
KEY = os.environ["MISTRAL_API_KEY"]
MODEL = os.environ["MODEL_ID"]
def chat(messages, temperature=0.3, max_tokens=1024,
json_mode=False, timeout=60):
payload = {
"model": MODEL,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
}
if json_mode:
payload["response_format"] = {"type": "json_object"}
req = urllib.request.Request(
f"{BASE}/chat/completions",
data=json.dumps(payload).encode("utf-8"),
headers={
"Authorization": f"Bearer {KEY}",
"Content-Type": "application/json",
},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
return json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as e:
把服务端返回的错误正文原样抛出,排错时非常有用
detail = e.read().decode("utf-8", "replace")
raise RuntimeError(f"HTTP {e.code}: {detail}") from None
if __name__ == "__main__":
r = chat([
{"role": "system", "content": "你是简洁的中文技术助理。"},
{"role": "user", "content": "用一句话说明什么是幂等。"},
])
print(r["choices"][0]["message"]["content"])
print(r["usage"])
```
except 里把错误正文打出来这一步很关键。很多 4xx 报错的原因就写在响应体里,吞掉它会让排错时间翻几倍。
1.4 流式输出
长回答用流式能明显改善体感。注意 SSE 分片不一定按行对齐,要自己缓冲:
```python
import json, urllib.request, os
def chat_stream(messages):
payload = {
"model": os.environ["MODEL_ID"],
"messages": messages,
"stream": True,
}
req = urllib.request.Request(
f"{os.environ['MISTRAL_BASE_URL'].rstrip('/')}/chat/completions",
data=json.dumps(payload).encode("utf-8"),
headers={
"Authorization": f"Bearer {os.environ['MISTRAL_API_KEY']}",
"Content-Type": "application/json",
"Accept": "text/event-stream",
},
)
with urllib.request.urlopen(req, timeout=120) as resp:
buf = ""
for raw in resp:
buf += raw.decode("utf-8")
while "\n" in buf:
line, buf = buf.split("\n", 1)
line = line.strip()
if not line.startswith("data:"):
continue
data = line[5:].strip()
if data == "[DONE]":
return
chunk = json.loads(data)
delta = chunk["choices"][0].get("delta", {})
if delta.get("content"):
yield delta["content"]
```
第二步:参数怎么设
大多数参数的行为和其他厂商的对话接口一致,迁移时认知负担不大。几个要点:
| 参数 | 作用 | 建议 |
|---|---|---|
temperature | 随机性 | 抽取类任务 0~0.3,创意类 0.7 以上 |
top_p | 采样范围 | 通常和 temperature 只调一个 |
max_tokens | 输出上限 | 按最坏情况设,超了会被截断且不易察觉 |
response_format | 强制 JSON | 需要结构化输出时打开,提示词里仍要写明 schema |
stream | 流式 | 前端交互场景打开 |
random_seed | 固定随机种子 | 做回归测试时打开,可复现 |
max_tokens 被截断的表现是内容在句子中间断掉,finish_reason 会给出提示。跑批时一定要检查这个字段,否则会把截断样本当成「模型能力不足」。
第三步:封装一层可切换的客户端
要对照成本,就得让同一份提示词能在两个模型上跑。加一层很薄的封装:
```python
providers.py
import os
from chat_client import chat
PROVIDERS = {
"new": lambda msgs, kw: chat(msgs, kw),
"legacy": lambda msgs, kw: legacy_chat(msgs, kw),
}
def legacy_chat(messages, **kw):
"""旧供应商的调用,保持返回结构一致。"""
...
```
统一约定返回一个三元组:(文本, usage, 耗时秒)。这样上层的评测脚本完全不用关心底层是谁。新旧模型的差异(比如角色名、JSON 模式的写法)全部收在这一层里处理。
第四步:成本对照
4.1 计费口径
对话接口一般按输入 token 和输出 token 分别计价,缓存命中、批处理等场景可能有折扣。单价随时会变,务必以官方定价页为准,不要凭记忆填数。把单价写成配置:
```python
pricing.py
单位:美元 / 百万 token。数值请从官方定价页复制,不要臆测。
PRICES = {
"new_model_id": {"in": None, "out": None},
"legacy_model_id": {"in": None, "out": None},
}
def cost_usd(model_key, usage):
p = PRICES[model_key]
if p["in"] is None or p["out"] is None:
raise SystemExit(f"请先到官方定价页填入 {model_key} 的单价")
return (usage["prompt_tokens"] / 1e6 * p["in"]
+ usage["completion_tokens"] / 1e6 * p["out"])
```
故意留成 None 并在运行时抛错,是为了防止有人拿着占位数字去汇报。
4.2 不要只看单价
实际对比时至少记录四个量:
- 单次请求平均成本;
- 单次请求平均延迟(首 token 延迟和总延迟分开看);
- 达到同等效果所需的重试次数或轮次;
- 人工修正输出所花的时间。
单价低的模型如果需要两轮才答对,总成本可能反而更高。这些小数字单看没感觉,乘以每天几万次请求就很直观了。
第五步:旧提示迁移
这一步最容易出问题,因为提示词里往往藏着大量针对旧模型的「补丁」。按下面的清单逐条过:
角色与格式
- 旧提示用了非标准的角色名(例如
developer之类),统一改成system/user/assistant三件套; - 检查是否依赖了旧模型特有的特殊标记或占位符,这些通常要删掉。
指令风格
- 负向指令(「不要输出 Markdown」)改成正向约束(「输出纯文本,不使用任何标记符号」),迁移后效果一般更稳;
- 把笼统的人设(「你是一个有用的助手」)换成一个具体岗位加一条输出格式约束。
结构化输出
- 打开 JSON 模式后,提示词里仍要写清字段名和类型,不能只靠参数;
- 给一个字段齐全的最小示例,比列十条规则管用。
语言
- 如果业务要中文,在 system 里显式写「始终使用简体中文回答」,不要指望默认行为。
改造前后的对比例子:
```text
改造前
你是乐于助人的助手。不要输出 Markdown,不要啰嗦,不要编造。
请从下面的客服对话里抽取用户诉求。
改造后
你是客服工单分析员,只输出简体中文。
从输入对话中抽取用户诉求,返回 JSON:
{"issue": "一句话描述", "urgency": "high|medium|low", "order_id": "无则填 null"}
只输出 JSON,不要任何解释文字。
```
第六步:把评测集跑起来
6.1 评测集格式
每行一条,字段自己定,保持稳定即可:
```json
{"id": "c001", "system": "你是客服工单分析员……", "input": "用户:我上周买的东西还没发货……"}
```
6.2 跑批脚本
```python
run_eval.py
import csv, json, time
import providers
def run(path, provider_key, out_csv):
fn = providers.PROVIDERS[provider_key]
with open(path, encoding="utf-8") as f, \
open(out_csv, "w", newline="", encoding="utf-8") as g:
w = csv.writer(g)
w.writerow(["id", "latency_s", "prompt_tokens",
"completion_tokens", "answer"])
for line in f:
case = json.loads(line)
messages = [
{"role": "system", "content": case["system"]},
{"role": "user", "content": case["input"]},
]
t0 = time.time()
text, usage, _ = fn(messages, temperature=0, max_tokens=1024)
w.writerow([
case["id"],
round(time.time() - t0, 3),
usage.get("prompt_tokens"),
usage.get("completion_tokens"),
text.replace("\n", " "),
])
if __name__ == "__main__":
run("eval_set.jsonl", "new", "out_new.csv")
run("eval_set.jsonl", "legacy", "out_legacy.csv")
```
temperature=0 是为了减少波动,让两次结果可比。
6.3 怎么判好坏
先抽样人工看 20 条,把明显不合格的挑出来,再决定要不要自动化打分。常见的做法是:有标准答案的字段(如 urgency、order_id)用规则比对;开放文本用另一个模型当裁判,但裁判模型要固定,否则没法比。
常见坑与排错
401 Unauthorized:Key 没导出、复制时带了空格、或者引号没包住。先在终端 echo $MISTRAL_API_KEY 看一眼长度对不对。
404 model not found:模型 ID 拼错或大小写不符。直接以官方文档当前版本的写法为准,不要凭记忆。
422 参数校验失败:messages 里某条的 content 是 null,或者 temperature 给了字符串。打开错误正文看,通常写得很具体。
429 限流:加指数退避重试。
```python
import random, time
def with_retry(fn, tries=5):
for i in range(tries):
try:
return fn()
except RuntimeError as e:
if "429" not in str(e) or i == tries - 1:
raise
time.sleep(min(2 ** i + random.random(), 30))
```
流式输出乱码或缺字:没有做分片缓冲,直接按行切。回到 1.4 节的写法。
token 用量对不上:不同模型的 tokenizer 不同,同一段中文的 token 数差别可能不小。做成本估算时要用真实调用返回的 usage,不要用别的模型的口径套。
迁移后效果变差:九成情况是提示词里的某个约束在新模型上失效了。做法是二分排查:把旧提示逐段删掉重跑,看哪一段删掉后效果崩了,那就是需要重写的部分。
输出被截断:检查 finish_reason,并把 max_tokens 调到业务上界。
下一步建议
跑通之后,可以往三个方向走:
第一,把评测跑批接进 CI。每次改提示词或换模型都自动跑一遍,输出对比表,避免靠感觉判断。
第二,做分层路由。简单分类、抽取类请求走小模型,复杂推理走大模型。用同一层抽象切换,改一行配置就能调。
第三,把成本监控做成看板。按业务线、按天统计 token 消耗,设置预算告警。成本问题的特点是平时无声无息,月底账单才暴露。
最后提醒一句:模型规格、价格、速率限制这些数字变化得很快,落地前花五分钟核对官方页面,比事后返工划算得多。
