基于 LLM 的系统需要同时考虑用户体验和单位成本。交互设计中常用约 100 毫秒、1 秒和 10 秒作为响应感知的参考尺度：较短响应接近即时，一秒左右通常能够维持注意，等待时间更长则容易使用户分心。因此，流式输出等改善感知速度的方法，与实际吞吐量同样重要。交互应用可以争取一秒以内的首个 token 时间；更长操作应通过进度提示说明当前正在执行的工作。
与此同时，系统成本会随输入输出 token 数量、上下文长度及请求并发量动态变化——提示词模板体积、文档分块大小、重排序候选集规模、缓存命中率、多模型路由策略、离线批处理以及自动扩缩容配置，都会直接影响单次请求的成本。本课介绍如何在保障输出质量的前提下，稳定达成 p95 延迟目标并精细化管控成本。
应当测量的核心指标
工程优化的第一步是准确量化用户实际感知的响应速度与系统实际支付的调用费用。
1. 延迟细分指标
- Time To First Token（TTFT，首字延迟）：从客户端发出请求到收到第一个生成 token 的耗时，主要受请求排队、网络往返以及输入提示词预填充（Prefill）计算耗时的影响。
- Time Per Output Token（TPOT，单输出 token 耗时）与 Inter-Token Latency（ITL，相邻 token 间隔）：反映进入解码生成（Decode）阶段后的持续输出节奏与流畅度。
- Total Latency（TTLT，端到端总延迟）：从请求发起直至收到最后一个 token 的完整耗时。
TTFT 与 TPOT 共同决定了流式界面的主观顺滑度，需在监控仪表板与服务等级目标（SLO）中分别独立跟踪。
每次请求均需记录输入 token 数、输出 token 数、服务商支持的缓存命中 token 数，并将单次请求成本关联至具体的模型名称、提示词版本、功能开关（Feature Flag）与用户分组（团队内部应统一指标属性 schema，避免不同工程师各自定义不一致的标签名称）。在课程工程中主要通过以下两条路径采集：
- 解析模型服务商返回的
usage字段：统计输入、输出及思考链（Reasoning）token 用量。注意，调用百炼聊天流式接口时，需显式设置stream_options.include_usage=true并按文档格式解析最后一个包含用量的数据块；若发生客户端中途断连、长度截断或响应缺失usage，需单独标记记录，并以服务商最终账单为准进行核对。此外，向量化（Embedding）、重排序（Rerank）、模型评判（Judge）与失败重试的消耗必须分别独立计数。 - 上报至自托管 Langfuse：通过 Langfuse 的追踪接口记录脱敏后的请求参数、模型名称、
usage明细与耗时，按目标模型所属地域的实际计费规则核算费用，并将模型评判与重试开销分开统计。
应统一属性 schema，减少各工程人员自行定义不同标签。
结合交互体验要求与成本上限设定量化 SLO：例如将实时聊天的 TTFT 目标设定为 300～700 毫秒，将持续生成速度设定为不低于每秒 6～10 个 token（对应 TPOT 约为 100～167 毫秒/token）；同时按功能模块设定端到端总耗时上限（如常规问答不超过 5 秒，多工具智能体流程不超过 8 秒）。上述目标值需结合具体业务场景实测校准，并与“最大允许输入/输出 token 数”、“重排序最大候选数”等配额预算直接挂钩。监控时需分别观察 p95 与 p99 分位值，重点治理拖慢长尾体验的慢请求。
延迟与成本的协同优化方法
1. 启用流式输出（Streaming）
流式输出在模型生成过程中持续传递 token，收到后立即展示。体验主要由首段文字出现时间和后续显示速度决定。即使总生成时间不变，一秒内开始且持续稳定输出，通常也能改善响应感并减少用户离开。SSE 是通过 HTTP 传递模型增量内容的一种常见方式。
- 开启提供商支持的流式选项。
- 后端收到内容后及时转发到浏览器。
- 前端按到达顺序显示数据块，通常可以直接追加文本。
还可以采用以下做法。
- 在首包到达前展示加载占位提示，首包到达后无缝切换为流式正文。
- 优化系统提示词以删减冗长的客套开场白，让模型第一句话就直奔主题输出高价值信息。
- 当后台工具调用耗时较长时，先向前端流式推送已确认的阶段性信息或当前正在执行的步骤提示，待工具返回后再继续补全回答。
2. 缩减传给 LLM 的上下文体积
减少提示 token：采用保留重要事实的压缩方式。较长上下文通常增加 prefill 时间和输入费用，具体关系取决于模型与服务实现。基于模型或规则的压缩，可以删除重复内容和模板性文字，同时保留引用和关键事实。将压缩实现为明确转换步骤，并通过评估测量质量与成本的取舍。
提升检索精度以减少送入片段数：在当前用户获准访问的资料范围内组合关键词搜索、向量检索与重排序（如调用具有独立原生 endpoint 的百炼 qwen3-rerank 或运行本地 Cross-encoder），通过实验评测审慎设定粗召回 top_k 与精排保留 top_n，避免把大量低相关度文档塞入生成上下文（需分别记录重排耗时、输入规模与生成费用）。
约束输出长度与结构：当业务只需特定字段时，通过 JSON Schema 或 Function Calling 强制要求结构化输出，既能省去长篇大论的废话 token，也能减少因格式解析失败触发的重试开销；同时合理配置 max_tokens、停止序列（Stop Sequences）与较低的 temperature。
3. 模型分级路由与推测解码
可以使用两级或三级模型级联：小型、低成本模型处理多数请求，置信度不足、安全规则触发或验证失败时，再交给能力更强的模型。通常可以在智能体入口设置路由节点。生产系统应明确升级依据，例如可回答程度、检索覆盖或有害内容标记，并记录原因，以便通过后续改进让更多任务由低成本模型可靠处理。
自托管推测解码（Speculative Decoding）：在自托管开源模型时，可利用小草稿模型（Draft Model）快速生成多个候选 token，再由主模型在单次前向计算中并行验证，从而降低平均单 token 解码耗时（TPOT）。由于实际加速比高度依赖硬件配置、提示词类型与解码参数，上线前必须基于真实负载实测端到端收益，切勿直接套用论文标称的理论加速倍数。
模型服务与推理层
高效自托管推理服务：第三方商用 API 的批处理与调度由服务商托管；若自建 GPU 推理集群，可选用 vLLM（通过 PagedAttention 高效管理 KV Cache，并利用 Continuous Batching 连续批处理动态接纳新请求、释放已完成请求，大幅提升 GPU 利用率与并发吞吐）或 Triton Inference Server（提供 Dynamic Batching 与 Ragged Batching，适合无状态模型与变长输入）。调优时需平衡排队等待时间与最大 Batch Size，防止过度攒批拉高 TTFT，并使用贴近生产分布的提示词长度与输出长度开展压测。
p99 长尾治理：针对少数阻塞请求或慢依赖引发的长尾延迟，需为检索、重排、工具调用及模型推理配置分层超时控制（Tiered Timeouts）与指数退避重试；对于幂等的只读请求，还可评估采用对冲请求（Hedged Requests）——在主请求超过设定短延迟仍未返回时立即发出备用请求，并取最先返回的有效结果。
缓存与复用
应用侧语义缓存（Semantic Caching）：在精确字符串匹配缓存之外，利用向量数据库与相似度阈值缓存高频相似提问的回答（在 FAQ、内部规章查询与固定模板工作流中命中率尤高）。请求在进入模型前先查缓存，命中则直接返回；工程上必须精细校准相似度阈值，并为每条缓存绑定语料数据版本号，确保底层知识库更新或权限变更后立即使旧缓存失效，防止返回过期事实。
服务商提示词前缀缓存（Prompt Caching）：许多模型服务商支持复用相同提示词前缀的 Prefill KV Cache，从而降低输入 token 计费并缩短 TTFT。对于大量请求共享长系统提示词、固定 Few-shot 示例、输出 Schema 或工具定义列表的应用，固定前缀顺序往往能获得可观收益（若每次请求的动态前缀差异极大，则应用侧语义缓存更为实用）。
缓存增强生成（Cache-Augmented Generation，CAG）与 RAG 结合：将低频变动的核心背景资料与高频变动的实时资料分流处理：
数据预处理：
- 为 CAG 选择变化较少、又经常与查询相关的数据，将其预先处理为模型可复用的 KV cache，并保存在适当存储中。在模型、前缀和缓存条件保持兼容且缓存仍有效时，后续查询可以复用；相关内容变化或缓存失效时需要更新。
- 为 RAG 按需预先计算 embedding，并写入兼容数据库。某些任务只需普通数据库即可，无需向量检索。
- 缓存上下文
- 低频变化数据经过一次预处理，形成可重复使用的缓存上下文。
- 系统提示词与用户查询
- 用户查询与系统提示词一同进入上下文窗口。
- 检索上下文
- 按用户查询检索高频变化数据，并将结果加入上下文窗口。
查询流程：
- 组织用户查询和系统提示，说明如何使用缓存上下文及检索获得的外部上下文。
- 需要语义搜索时，对查询计算 embedding 并访问向量数据库；其他情况可直接查询实时数据库或网页。
- 将检索到的外部内容加入最终提示。
- 生成最终回答并返回用户。
6. 无需实时响应时使用离线批处理（Batch Processing）
大规模语料向量化与离线模型评判应走独立异步队列，完整记录任务版本、完成期限、失败原因与重试次数。若使用云厂商提供的原生批处理服务，需提前核验当前地域与目标模型是否支持 Batch API（不能因为同步接口可用就默认其支持批处理；百炼向量化的批处理接入方式可参阅百炼向量化说明）。
LLM 推理的总拥有成本（TCO）与自托管权衡
在 AI Cloud 或本地 GPU 上部署及维护模型，需要额外工程和运维成本。部分费用随已配置 GPU 数量增加，在固定资源期间相对稳定。规模扩大时，某些人工管理成本按 GPU 分摊后可能下降，具体仍取决于系统复杂度。
Total Cost of Ownership（TCO，总拥有成本）与安全要求应共同影响厂商 API 和自托管的选择。按 token 计费的 API 成本主要随输入、输出及应用流量变化。
自托管模型需要配置计算资源。最简单的固定配置中，单位时间成本可以近似为水平线；实际生产环境通常还需要自动扩缩容，以控制每个 token 的成本并维持响应速度。
AI Cloud 的吞吐量、延迟与弹性扩缩容
评估时需要明确应用所需吞吐量和延迟。
- 吞吐量：单位时间能够处理的请求数量，通常以每秒请求数衡量，也可根据推理场景记录 token 吞吐量。
- 延迟：用户等待首项有效输出的时间。聊天应用常关注 TTFT，之后的生成速度只要足够支持阅读即可。降低 prefill 时间是改善 TTFT 的一种方法，同时也需检查排队和网络时间。
增加提供服务的 GPU 容量可以提高吞吐量。有时在多台机器部署相同模型即可水平扩展，有时则需要让同一模型使用更多 GPU。具体方案取决于模型规模和服务要求。
应预估 GPU 数量和各时段流量。流量较少时所需吞吐量降低，可以通过扩容与缩容管理 TCO。
自托管还提供更详细的吞吐量和延迟控制。充分调整后，有机会在特定场景中达到低于外部 API 的延迟，但需要实际测量。将检索和其他服务部署在模型附近，例如同一区域，也可以减少网络开销。
原型开发阶段通常没有持续用户流量。按调用计费 API 可以在无调用时停止相应用量费用；支持缩容到零的云环境也可以在空闲期间停止计算资源，但仍需考虑存储等剩余费用。
厂商 API 与自托管的成本对比方法
将商用 API 的波动成本曲线与自托管云资源的成本曲线叠加在同一时间轴上，即可直观对比不同业务发展阶段的经济性：
图中绿色区域表示 API 成本高于图示云资源成本的时段，红色区域表示 API 成本更低的时段。比较实际方案时，需要累计观察期内的费用，并将运维、容量闲置及其他支出计入 TCO。采用云端自托管时，可以通过提高有效利用率和适当扩缩容，使其在满足性能要求的同时具有成本优势。
记录实际模型用量与限流
在课程主线实践中，需对照百炼接口文档核验聊天响应的 usage 字段及流式模式下的 include_usage 输出，并将 embedding、judge 评判、工具调用轮次与失败重试分别独立统计；模型费用按实际所属地域、模型单价与计费日期核算（本地 JSONL 检索指标评估直接比对来源 ID，不调用外部 judge 模型）。
练习：记录本课所用配置、输入、预期结果与失败分支，并用脱敏数据复现。
通过专用 Batch 接口处理离线作业
百炼的 Batch File API 使用独立的 BATCH_BASE_URL、API Key 及受支持的地域模型列表（北京地域示例 endpoint 为 [https://dashscope.aliyuncs.com/compatible-mode/v1](https://dashscope.aliyuncs.com/compatible-mode/v1)；注意输入 JSONL 文件内每行记录的 url 路径必须与创建批处理任务时指定的 endpoint 参数严格一致）。
import json
import os
from pathlib import Path
from openai import OpenAI
def create_batch_client():
key = os.environ['BATCH_API_KEY']
if 'REPLACE_' in key:
raise ValueError('batch_credentials_required')
return OpenAI(base_url=os.environ['BATCH_BASE_URL'], api_key=key, timeout=60, max_retries=0)
def prepare_batch(items, model):
rows, seen = [], set()
for item in items:
identity = item['id']
if not identity or identity in seen or len(identity) > 256:
raise ValueError('batch_custom_id_invalid')
seen.add(identity)
rows.append({'custom_id': identity, 'method': 'POST', 'url': '/v1/chat/completions',
'body': {'model': model, 'enable_thinking': False,
'messages': [{'role': 'user', 'content': item['input']['question']}]}})
if len(rows) > 50000:
raise ValueError('batch_too_many_rows')
content = '\n'.join(json.dumps(row, ensure_ascii=False) for row in rows) + '\n'
if len(content.encode()) > 500 * 1024 * 1024 or any(len(json.dumps(row).encode()) > 1024 * 1024 for row in rows):
raise ValueError('batch_file_too_large')
return content
def submit_batch(client, input_path, state_path):
state = Path(state_path)
if state.exists():
raise RuntimeError('batch_submission_state_exists')
state.write_text(json.dumps({'state': 'submitting'}))
uploaded = client.files.create(file=Path(input_path), purpose='batch')
state.write_text(json.dumps({'state': 'creating', 'input_file_id': uploaded.id}))
batch = client.batches.create(input_file_id=uploaded.id, endpoint='/v1/chat/completions', completion_window='24h')
state.write_text(json.dumps({'state': 'submitted', 'input_file_id': uploaded.id, 'batch_id': batch.id}))
return batch.id
def reconcile_batch(expected_ids, output_text, error_text):
expected, seen, results = set(expected_ids), set(), {}
for content in [output_text, error_text]:
for line in content.splitlines():
item = json.loads(line)
identity = item['custom_id']
if identity not in expected or identity in seen:
raise ValueError('batch_result_id_invalid')
seen.add(identity)
response = item.get('response')
success = bool(response and response.get('status_code') == 200 and not item.get('error'))
results[identity] = {'state': 'success' if success else 'error', 'response': response, 'error': item.get('error')}
for identity in expected - seen:
results[identity] = {'state': 'missing'}
return results
def collect_batch(client, batch_id, expected_ids):
job = client.batches.retrieve(batch_id)
if job.status not in {'completed', 'failed', 'expired', 'cancelled'}:
return {'state': job.status, 'results': None}
output = client.files.content(job.output_file_id).text if job.output_file_id else ''
errors = client.files.content(job.error_file_id).text if job.error_file_id else ''
return {'state': job.status, 'results': reconcile_batch(expected_ids, output, errors)}
custom_id 在文件内唯一，结果按 ID 合并，不能按输出行顺序关联。示例关闭思考模式并放在 body 顶层。上传用 purpose=batch；创建任务指定 completion_window=24h，保存 file_id 和 batch_id 后按 ID 查询，轮询设定总时限。
完整处理终态与部分成功结果：轮询状态时（需设定总轮询超时上限），validating、in_progress 和 finalizing 代表作业仍在运行；进入 completed、failed、expired 或 cancelled 终态后，由于即便整体状态为 completed 内部仍可能包含单行失败请求（或者在取消/过期前已有部分请求执行完毕），必须同时下载并解析 output_file_id（成功结果文件）与 error_file_id（错误文件），并将既未出现在结果文件也未出现在错误文件中的请求标记为 missing。课程测试仅在本地校验 JSONL 构造与 fixture 对账逻辑，不发起真实的云端批处理计费请求。
SDK 创建客户端时设置 max_retries=0，避免在创建结果不明时自动重复提交。真实使用前检查文件、模型、地域、费用和保留时限。课程只验证本地 JSONL 和 fixture，未提交任何真实批任务。
练习：打乱批处理返回结果的行顺序，并混入成功行、报错行、缺失 custom_id 以及重复 custom_id 的异常样本，验证 reconcile_batch 的对账报告是否准确；同时针对失败批次区分整文件级拒绝错误与单行级执行错误。