09. LLM 应用工程:Token、Prompt、Eval、资产化
读完这篇你能:
- 做成本治理:预算、限流、模型路由、缓存
- 管理 Prompt 版本,像管理代码一样
- 搭 Eval 体系,知道改 prompt 是变好还是变差
- 沉淀组织资产,让团队的 LLM 能力可复用
对应架构层:应用层(LLM 专属)
前置知识:01-08 篇
预计用时:4 小时(含动手)
0. 为什么算法工程师要学这个
你做完了前 8 篇,应用上线了,能跑、能扩、能监控、能容错。
但你发现 LLM 应用有独有的工程问题:
问题 1:成本失控
月底 OpenAI 账单 $50,000——你公司 CFO 跳起来。但你怎么解释?哪个用户烧的?哪个接口?哪些 prompt 效率最低?
问题 2:Prompt 是"魔法数字"
团队 5 个人,每个人在自己代码里写 prompt:
# Alice 的代码
prompt = "你是一个助手。请回答:..."
# Bob 的代码(改了一版)
prompt = "You are an assistant. Answer:..."
# Charlie 的代码(又改了一版,但忘了告诉 Alice)
prompt = "你是 AI 助手。请简洁回答:..."
没人知道哪个版本最好,改了出问题没法回退。
问题 3:不知道改 prompt 是好是坏
PM 说"让回答更简洁一点"。你改了 prompt,感觉好像简洁了——但真的更好吗?有多少场景变好了?有多少变差了?
没有 Eval 体系,所有改动都是"凭感觉"。
问题 4:能力不沉淀
某员工搭了个超棒的 RAG 流程,半年后他离职了,所有经验跟着他走。新来的人从零开始。
这一篇讲 LLM 工程化(LLMOps)——把这些"零散的 LLM 经验"变成"系统化的工程能力"。
1. 核心概念(最少必要理论)
1.1 LLM 应用的特殊性
| 维度 | 传统软件 | LLM 应用 |
|---|---|---|
| 输出确定性 | 输入 X,输出 Y 固定 | 同样输入,输出可能不同 |
| 失败模式 | 崩溃 / 报错 | 看起来对,其实错(幻觉) |
| 成本 | 固定(服务器) | 变动(按 token 计费) |
| 测试 | 单元测试足够 | 需要 LLM-as-Judge |
| 部署 | 二进制 artifact | 二进制 + Prompt + Eval |
| 监控 | QPS、延迟 | + Token、成本、反馈率 |
LLM 工程化(LLMOps)就是处理这些特殊性的方法论。
1.2 LLMOps 七大能力
LLMOps
├── 1. Token 经济与成本治理
├── 2. Prompt 即代码 + 版本管理
├── 3. 模型路由(多模型协同)
├── 4. Eval 评估体系
├── 5. 灰度发布
├── 6. 一键回退
└── 7. 组织资产化
2. Token 经济与成本治理
2.1 Token 成本怎么算
成本 = 输入 token 数 × 输入单价 + 输出 token 数 × 输出单价
2026 年主流模型价格(每 1M token,美元):
| 模型 | 输入 | 输出 |
|---|---|---|
| GPT-4o | $2.5 | $10 |
| GPT-4o-mini | $0.15 | $0.6 |
| Claude 3.5 Sonnet | $3 | $15 |
| Claude 3 Haiku | $0.25 | $1.25 |
| Llama 3.1 70B(自托管) | ~$0.5 | ~$0.5 |
LLM 应用的典型成本结构:
日活 1000,人均 10 次调用,每次 500 输入 + 200 输出 token
日 token 消耗:
1000 × 10 × (500 + 200) = 700 万 token
用 GPT-4o:
输入:500 万 × $2.5/1M = $12.5
输出:200 万 × $10/1M = $20
日成本:$32.5
月成本:$975
用 GPT-4o-mini:
日成本:$1.6
月成本:$48
结论:模型选择对成本影响 20 倍。不是所有任务都要用最强模型。
2.2 成本治理的五个手段
手段 1:模型路由
简单任务用便宜模型,复杂任务才用贵模型:
async def smart_llm_call(prompt: str, complexity: str = "auto") -> str:
if complexity == "auto":
complexity = await classify_complexity(prompt)
model_map = {
"simple": "gpt-4o-mini", # $0.15/$0.6
"medium": "claude-3-haiku", # $0.25/$1.25
"complex": "gpt-4o", # $2.5/$10
}
return await call_model(model_map[complexity], prompt)
async def classify_complexity(prompt: str) -> str:
"""用最便宜的模型分类"""
resp = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "判断问题复杂度:simple/medium/complex"},
{"role": "user", "content": prompt[:200]},
],
max_tokens=10
)
return resp.choices[0].message.content.strip()
效果:70% 简单任务 + 20% 中等 + 10% 复杂 → 总成本下降 5-10 倍。
手段 2:Prompt 压缩
LLM 输入按 token 计费,prompt 越长越贵。
# ❌ 啰嗦
prompt = """
请你作为一个专业的助手,用中文回答用户的问题。
回答时要保持礼貌和专业,不要使用任何不当言论。
如果不确定,请明确说明,不要编造信息。
回答长度适中,不要过长也不要过短。
"""
# ✅ 精炼(同样意思,token 减半)
prompt = "中文专业助手。不确定时直说,勿编造。"
工具:LLMLingua(微软,自动压缩 prompt 2-10 倍)。
手段 3:缓存(03 篇讲过)
精确缓存 + 语义缓存 → 节省 20-50% 成本。
手段 4:批处理 API
OpenAI / Anthropic 的 Batch API:
- 异步处理(24 小时内返回)
- 价格便宜 50%
适合非实时场景(夜间汇总、批量打标签)。
batch = await client.batches.create(
completion_window="24h",
endpoint="/v1/chat/completions",
input_file_id=file.id
)
手段 5:预算和限流
每个用户 / 部门设预算,超了拒绝:
class UserBudget(SQLModel, table=True):
user_id: int
daily_token_limit: int = 100_000
monthly_cost_limit_usd: float = 50.0
async def check_budget(user: User, estimated_tokens: int, estimated_cost: float):
today_tokens = await get_today_tokens(user.id)
if today_tokens + estimated_tokens > user.budget.daily_token_limit:
raise HTTPException(429, "Daily token limit exceeded")
month_cost = await get_month_cost(user.id)
if month_cost + estimated_cost > user.budget.monthly_cost_limit_usd:
raise HTTPException(402, "Monthly budget exceeded") # 402 Payment Required
2.3 Token 监控
每次调用都记 token:
from prometheus_client import Counter
token_counter = Counter(
"llm_tokens_total",
"Tokens consumed",
["model", "direction", "user_tier"] # direction: input/output
)
cost_counter = Counter(
"llm_cost_usd_total",
"USD spent on LLM",
["model", "user_tier"]
)
# 在调用后
token_counter.labels(model="gpt-4o", direction="input", user_tier=user.tier).inc(usage.input_tokens)
token_counter.labels(model="gpt-4o", direction="output", user_tier=user.tier).inc(usage.output_tokens)
cost = calculate_cost(model, usage)
cost_counter.labels(model="gpt-4o", user_tier=user.tier).inc(cost)
Grafana 看板:
- 日 / 月成本曲线
- 按用户分组的 Top 10 消费者
- 按接口的 token 消耗
- 异常告警:单用户 1 小时 > $10
3. Prompt 即代码
3.1 Prompt 管理的三个阶段
阶段 1:写在代码里(失败)
def summarize(text):
prompt = f"总结以下文本:{text}" # 改 prompt 要改代码 + 发版
return llm(prompt)
阶段 2:写在配置文件(入门)
# prompts/summarize.yaml
system: "你是总结助手"
user: "总结:{text}"
prompt = load_yaml("prompts/summarize.yaml")
阶段 3:版本化 + Eval + CI(生产级)
prompts/
└── summarize/
├── metadata.json
├── v1.0.0/
│ ├── prompt.md
│ ├── variables.json
│ ├── tests.jsonl
│ └── eval_results.json
├── v1.1.0/
│ └── ...
└── latest -> v1.1.0
3.2 Prompt 文件结构
prompts/summarize/v1.0.0/prompt.md:
---
version: 1.0.0
model: gpt-4o-mini
temperature: 0.3
max_tokens: 500
---
# 角色
你是中文总结助手。
# 任务
总结用户提供的文本,要求:
- 字数不超过 {{ max_length }}
- 保留关键信息
- 客观中立
# 输入
{{ text }}
prompts/summarize/v1.0.0/tests.jsonl:
{"input": {"text": "FastAPI 是..."}, "expected_keywords": ["FastAPI", "Web 框架"], "expected_max_length": 100}
{"input": {"text": "Python 是..."}, "expected_keywords": ["Python", "编程"]}
3.3 加载和渲染
from jinja2 import Template
import frontmatter
from pathlib import Path
class PromptManager:
def __init__(self, prompts_dir: str = "prompts"):
self.dir = Path(prompts_dir)
def load(self, name: str, version: str = "latest") -> dict:
if version == "latest":
version = self._read_latest(name)
path = self.dir / name / version / "prompt.md"
post = frontmatter.load(path)
return {
"template": post.content,
"metadata": post.metadata, # model, temperature...
"version": version,
}
def render(self, name: str, variables: dict, version: str = "latest") -> str:
prompt = self.load(name, version)
return Template(prompt["template"]).render(**variables)
def _read_latest(self, name: str) -> str:
link = self.dir / name / "latest"
return link.resolve().name
# 使用
manager = PromptManager()
prompt_text = manager.render("summarize", {"text": "...", "max_length": 100}, version="latest")
metadata = manager.load("summarize")["metadata"]
resp = await client.chat.completions.create(
model=metadata["model"],
messages=[{"role": "user", "content": prompt_text}],
temperature=metadata["temperature"],
max_tokens=metadata["max_tokens"],
)
3.4 Langfuse Prompt Management
不想自己造轮子,用 Langfuse:
pip install langfuse
from langfuse import Langfuse
langfuse = Langfuse()
# 从 Langfuse 拉 prompt(在 UI 里编辑)
prompt_client = langfuse.get_prompt("summarize", label="production")
# 渲染
compiled = prompt_client.compile(text="...", max_length=100)
# 调用 + 自动记录
resp = await client.chat.completions.create(
model=prompt_client.config["model"],
messages=[{"role": "user", "content": compiled}],
)
好处:
- PM 在 UI 里改 prompt,不用改代码
- 自动版本化
- 集成 Eval
- A/B 测试
4. Eval 评估体系
4.1 为什么需要 Eval
没有 Eval,所有改动都是凭感觉:
- prompt 改一个字,效果变好还是变差?
- 换一个模型,精度差多少?
- 缓存命中返回,用户体验降多少?
Eval 给你"客观尺子":每次改动,数字告诉你好坏。
4.2 Eval 类型
| 类型 | 怎么做 | 例子 |
|---|---|---|
| Exact Match | 字符串完全相同 | 数学题答案 |
| Keyword Match | 包含关键词 | 答案有 "FastAPI" |
| Regex | 正则匹配 | 输出符合 JSON 格式 |
| LLM-as-Judge | 用另一个 LLM 打分 | GPT-4 给 GPT-4o-mini 的答案打分 |
| Embedding 相似度 | 与参考答案的余弦相似度 | 开放式问答 |
| Human Eval | 人工标注 | 高风险场景 |
4.3 搭建 Eval 数据集
// datasets/summarize_eval.jsonl
{"input": {"text": "..."}, "expected_keywords": ["FastAPI"], "expected_max_length": 100, "quality_score": 5}
{"input": {"text": "..."}, "expected_keywords": ["Python"], "expected_max_length": 100, "quality_score": 4}
积累来源:
- 人工标注种子样本(100 条起步)
- 生产真实 query 采样(脱敏后)
- 用户反馈 👍/👎 的样本
- 边界 / 对抗样本(防止 prompt injection)
4.4 跑 Eval
from openai import AsyncOpenAI
import json
from pathlib import Path
client = AsyncOpenAI()
async def run_eval(prompt_name: str, version: str, dataset_path: str):
dataset = [json.loads(line) for line in open(dataset_path)]
results = []
for sample in dataset:
prompt_text = manager.render(prompt_name, sample["input"], version=version)
resp = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt_text}],
)
output = resp.choices[0].message.content
# 各种指标
metrics = {
"keyword_match": all(kw in output for kw in sample.get("expected_keywords", [])),
"length_ok": len(output) <= sample["expected_max_length"],
"quality_score": await llm_judge(output, sample["input"]), # 1-5 分
}
results.append({"output": output, "metrics": metrics})
# 汇总
avg_quality = sum(r["metrics"]["quality_score"] for r in results) / len(results)
keyword_pass_rate = sum(r["metrics"]["keyword_match"] for r in results) / len(results)
return {
"prompt_version": version,
"sample_count": len(results),
"avg_quality": avg_quality,
"keyword_pass_rate": keyword_pass_rate,
"results": results,
}
async def llm_judge(output: str, input_data: dict) -> int:
"""用 GPT-4 给答案打分"""
resp = await client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "给总结质量打 1-5 分。只返回数字。"},
{"role": "user", "content": f"输入:{input_data}\n\n总结:{output}"},
],
max_tokens=5
)
return int(resp.choices[0].message.content.strip())
# 命令行
# python eval.py summarize v1.0.0 --dataset datasets/summarize_eval.jsonl
4.5 Eval 集成到 CI
# .github/workflows/eval.yml
name: Eval Prompt
on:
pull_request:
paths:
- 'prompts/**'
jobs:
eval:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- run: pip install -r requirements.txt
- name: Run eval on new prompt version
run: python eval.py summarize ${{ github.event.pull_request.head.ref }}
- name: Compare with baseline
run: python compare_eval.py summarize latest ${{ github.event.pull_request.head.ref }}
- name: Fail if regression
run: |
# 如果新版本质量比 baseline 差 5% 以上,PR 不让合
python check_regression.py --threshold 0.05
效果:PR 里改 prompt,自动跑 Eval,GPT-4 打分,质量退化就阻止合并。
4.6 进阶工具
- promptfoo:命令行跑 Eval,本地开发用
- DeepEval:Python 单元测试风格的 Eval
- Langfuse:在线 Eval + 真实流量采样
- OpenAI Evals:官方开源框架
# promptfoo 例子
promptfoo eval -p prompts/summarize.txt -d datasets/summarize.csv --metrics accuracy,length
5. 灰度发布
5.1 为什么不能全量上新 Prompt
新 Prompt 改了 100 条 Eval 样本里 95 条变好——但线上有 100 万种 query,可能某 5% 场景变差。
解决:灰度发布,先放 5% 流量,观察,逐步加到 100%。
5.2 实现灰度
import random
async def call_with_rollout(prompt_name: str, variables: dict, user_id: int):
# 稳定的 hash,保证同一用户始终走同一版本
bucket = hash(str(user_id)) % 100
if bucket < 5: # 5% 灰度
version = "v1.1.0" # 新版本
else:
version = "v1.0.0" # 老版本
return await call_with_version(prompt_name, variables, version), version
# 监控两个版本的效果
为什么用 hash 不用随机:同一用户始终走同一版本,体验一致,便于 A/B 对比。
5.3 灰度指标
灰度期间监控:
- 用户反馈率(👎 比例)
- 平均会话长度(突然变短 → 用户放弃)
- 重试率(用户重发同样问题 → 不满意)
- 成本(token / 调用)
任何指标显著恶化,立即回退。
6. 一键回退
6.1 回退三件事
代码回退(05 篇讲过)+ Prompt 回退 + 数据回退。
Prompt 回退:
# Latest symlink 切回旧版本
def rollback_prompt(name: str, to_version: str):
link = Path(f"prompts/{name}/latest")
link.unlink()
link.symlink_to(to_version)
# 或者用配置
rollback_service.update("summarize", "v1.0.0")
Langfuse:UI 里点几下切换 production 标签到旧版本。
6.2 回退触发条件
- 错误率 > 5%(持续 5 分钟)
- p99 延迟 > 30 秒
- 用户反馈率下降 > 20%
- 成本飙升 > 200%
# 自动回退(谨慎)
async def auto_rollback_monitor():
while True:
metrics = await get_recent_metrics(window="5m")
if metrics.error_rate > 0.05:
logger.critical("auto_rollback_triggered", reason="high_error_rate")
rollback_prompt("summarize", "previous")
notify_oncall("Auto rollback triggered")
await asyncio.sleep(60)
警告:自动回退很危险(可能回退到一个本来就有问题的版本)。生产强烈建议人工确认。
7. 组织资产化
7.1 什么是组织资产
个人经验 vs 组织资产:
| 类型 | 例子 | 持有者 |
|---|---|---|
| 个人经验 | Alice 知道怎么调 prompt 让 LLM 不幻觉 | Alice 的脑子 |
| 组织资产 | 团队 wiki 写了 "防幻觉 prompt 模板 + Eval 数据集" | 公司的仓库 |
算法工程师价值的关键:不只是做出能跑的东西,而是把经验沉淀成可复用的资产。
7.2 八类 LLM 组织资产
组织资产
├── 1. Prompt 库(版本化、Eval 覆盖)
├── 2. Eval 数据集(每个场景 100+ 样本)
├── 3. 模型路由策略(什么任务用什么模型)
├── 4. Guardrail 规则库(注入检测、PII 脱敏)
├── 5. RAG 知识库(文档切片、embedding 策略)
├── 6. Eval 自动化工具(打分器、CI 集成)
├── 7. 监控仪表盘模板(成本、质量、反馈)
└── 8. 反模式文档(踩过的坑 + 解决方案)
7.3 资产字段规范
每个 prompt 资产的字段:
name: summarize_v2
domain: content-generation
use_case: "长文本摘要,适用于博客、论文、报告"
prompt: |
...
variables:
- name: text
type: string
description: 要总结的原文
max_length: 10000
model: gpt-4o-mini
temperature: 0.3
eval_dataset: datasets/summarize_v2.jsonl
eval_score:
accuracy: 4.5
concise: 4.2
total_samples: 150
author: [email protected]
created_at: 2026-06-23
last_used: 2026-06-30
usage_count: 12500
related_prompts: [outline_v1, rewrite_v3]
known_issues:
- "对超长文档(>5000 字)摘要可能丢细节"
tags: [summarization, chinese, gpt-4o-mini]
7.4 资产复用证据
团队 wiki 里维护"谁用了什么资产":
## summarize_v2
### 用在哪里
- 产品 A 的文章摘要功能
- 产品 B 的搜索结果摘要
- 内部知识库的 RAG 摘要
### 效果数据
- 产品 A:用户满意度 4.6/5,日均调用 5000 次
- 产品 B:节省 30% 阅读时间
好处:
- 新项目启动时,先查资产库,90% 场景能复用
- 资产升级时,知道会影响哪些下游
- 员工离职时,资产仍在
7.5 反模式文档
踩过的坑,写下来:
## 反模式:缓存 LLM 输出但没考虑时效性
**场景**:新闻摘要应用,缓存了 LLM 输出。一个月后用户看到"今天的新闻"是上个月的旧摘要。
**根因**:TTL 设过长(永久缓存),内容实际有时效性。
**解决**:
- 时效性内容(news、stock)TTL < 1 小时
- 静态内容(知识、文档)TTL 长
- 用户主动刷新时不返回缓存
**发现时间**:2026-03-15
**影响**:5000 用户看到错误信息,客诉 30+
8. 完整的 LLMOps 流水线
开发期:
Prompt 设计 → 写本地 Eval → 跑 100 条 → 优化
↓
预发:
提 PR → CI 自动跑 Eval(对比 baseline)
↓
PR Review(人审 Prompt)
↓
合并 → 部署到 Staging
↓
灰度:
5% 流量 → 监控 24h
↓
50% 流量 → 监控 48h
↓
全量:
100% 流量 → 持续监控
↓
异常:
自动告警 → 人审 → 一键回退
↓
复盘:
写反模式文档 → 更新 Eval 数据集
9. 常见踩坑
坑 1:Prompt 写死在代码里
改一次 prompt 要发版,周期长,迭代慢。
解决:Prompt 独立成文件 / Langfuse,与代码解耦。
坑 2:Eval 数据集太小
10 条样本打分,完全不能代表线上分布。
解决:
- 至少 100 条覆盖各种边界
- 持续从生产采样补充
- 关注"长尾"场景(低频但重要的)
坑 3:LLM-as-Judge 自己评自己
用 GPT-4 评 GPT-4 的输出 → 偏向自己。
解决:用不同模型族(用 Claude 评 GPT,用 GPT 评 Claude),或人工抽检校准。
坑 4:成本监控滞后
月底才发现成本爆炸 → 已经烧了 $50000。
解决:
- 实时监控(每 5 分钟刷新)
- 日预算告警
- 单用户预算硬限制
坑 5:资产库没人维护
搭了个资产库,半年后没人用,内容过期。
解决:
- 资产使用记录(谁在用,什么时候用)
- 定期 review(每季度)
- 资产负责人(每个资产有 owner)
坑 6:灰度不监控
灰度发了,但忘了看监控,出问题一周后才发现。
解决:
- 灰度必须有监控仪表盘
- 指标恶化自动告警
- 灰度期固定(7 天/14 天)
坑 7:Eval 只看"好"的指标
只看准确率,不看延迟、成本、安全。
解决:Eval 要多维度(质量 + 性能 + 成本 + 安全)。
10. 检查清单
- 我的应用每次 LLM 调用都记 token 数和成本
- 我有 Grafana 看板看日 / 月成本
- 我做了模型路由(简单任务用便宜模型)
- 我的 Prompt 在文件里,不在代码里
- 我的 Prompt 有版本号,能回退
- 我有 Eval 数据集(至少 100 条)
- 我用 LLM-as-Judge 自动打分
- Eval 集成到 CI,质量退化阻止合并
- 我用灰度发布(5% → 50% → 100%)
- 我有回退机制,5 分钟内能切回老版本
- 我的团队有资产库(prompt / eval / guardrails)
- 我有反模式文档,记录踩过的坑
11. 系列总结
恭喜,你读完了 10 篇!
回头看 00 篇的分层架构图,你应该能在脑子里点亮每一格:
| 层 | 技术 | 你学到的 |
|---|---|---|
| 接入层 | DNS、CDN、LB、HTTPS、VPC | 07 篇 |
| 应用层 | FastAPI、JWT、RBAC、SSE、Guardrails | 01、04、08 篇 |
| 数据层 | Postgres、Redis、S3、pgvector | 02、03 篇 |
| 协调层 | K8s etcd、Kafka、Nacos、Celery | 07 篇 |
| 部署层 | Docker、Cloud Run、CI/CD | 05 篇 |
| 可观测层 | Prometheus、Grafana、ELK、OTel、Langfuse | 06 篇 |
| LLM 工程 | Token 治理、Prompt 版本、Eval、资产化 | 09 篇 |
下一步该做什么?
- 做个综合项目:从零做一个完整的 LLM 应用(前端 + 后端 + 数据库 + 部署 + 监控 + 护栏 + Eval),目标 1000 真实日活
- 深入某一层:挑你最感兴趣的(如可观测性、K8s),读 1-2 本专著
- 持续学习:订阅 ByteByteGo、Chip Huyen、Simon Willison 的博客
- 教别人:写博客 / 内部分享,把知识"用出来"才算真懂
最后:这个系列的终点不是"读完 10 篇文章",而是你脑子里有完整的架构地图,遇到任何技术问题都能定位、判断、解决。
这就是算法工程师到架构师的路。你已经走完了。
12. 进阶学习资源
必读
- Chip Huyen《Designing Machine Learning Systems》:ML 生产系统的圣经
- 《LLM Engineer's Handbook》:Paul Iusztin,最新最全的 LLM 工程指南
- LangFuse 官方文档:https://langfuse.com/docs
- OpenAI Cookbook:https://github.com/openai/openai-cookbook
- Anthropic Prompt Engineering:https://docs.anthropic.com/claude/docs/prompt-engineering
必关注
- Chip Huyen 博客:https://huyenchip.com/blog/
- Eugene Yan 博客:
https://eugeneyan.com/(Eval / ML 生产) - Simon Willison 博客:
https://simonwillison.net/(LLM 实战) - Andrej Karpathy YouTube:从零实现 GPT、讲 LLM 原理
- Hamel Husain 博客:
https://hamel.dev/(LLM Eval)
必用工具
- LangFuse:LLM 可观测 + Prompt 管理
- LangSmith:LangChain 官方 trace
- Braintrust:Eval 平台
- Promptfoo:本地 Eval CLI
- Weights & Biases:ML 全流程实验追踪