AI开发进阶③:大模型推理加速与成本控制——从API到自部署的全链路优化

发布时间:2026/5/24 16:14:24

AI开发进阶③:大模型推理加速与成本控制——从API到自部署的全链路优化 AI 开发进阶第3篇大模型推理加速与成本控制——从 API 到自部署的全链路优化适合读者已读完基础9篇 第①②篇想让 Agent 更快、更便宜预计阅读时间40分钟作者AI小渔村前言成本是 Agent 落地的生死线前两篇我们讨论了评估体系和可观测性。但即使你知道了 Agent 好不好、能不能跑**成本」仍然是一个致命问题一个复杂的 Agent 请求可能消耗 $0.5 - $2.0日活 1 万用户每月成本轻松超过 $15,000老板问能不能再便宜点这篇讲的是如何让 Agent 更快、更便宜覆盖从 API 调优到自部署的全链路优化。一、成本到底花在哪了1.1 成本的三个组成部分总成本 Token 成本 推理延迟成本 基础设施成本 Token 成本 (输入 Token 数 × 输入单价) (输出 Token 数 × 输出单价) 推理延迟成本 等待时间 × 服务器/算力成本 基础设施成本 服务器、存储、网络等固定开销以 GPT-4o 为例2026年5月价格模型输入缓存未命中输出GPT-4o$2.50 / 1M$10.00 / 1MGPT-4o-mini$0.15 / 1M$0.60 / 1M优化方向减少 Token 消耗输入压缩、输出精简减少调用次数结果复用、缓存模型路由简单任务用小模型二、输入优化减少 Token 消耗2.1 上下文压缩长对话的痛点历史消息占了一大半 token但模型真正需要的是最近的信息。解法摘要压缩from dataclasses import dataclass from typing import List, Dict import json dataclass class Message: role: str content: str timestamp: str class ContextCompressor: 上下文压缩器 def __init__(self, llm, max_tokens: int 6000, keep_recent: int 5): self.llm llm self.max_tokens max_tokens self.keep_recent keep_recent def compress(self, messages: List[Message]) - List[Message]: 压缩对话历史 # 保留最近 N 条消息完整保留 recent messages[-self.keep_recent:] # 更早的消息需要压缩 older messages[:-self.keep_recent] if not older: return recent # 估算当前 token 数 current_tokens sum(self.estimate_tokens(m.content) for m in recent) if current_tokens self.max_tokens: # 最近的消息也已经太长直接截断 return self._truncate(recent, self.max_tokens) # 对更早的消息做摘要 available_tokens self.max_tokens - current_tokens summary await self._summarize(older, available_tokens) return [Message( rolesystem, contentf[对话历史摘要] {summary} )] recent async def _summarize(self, messages: List[Message], max_tokens: int) - str: 用 LLM 生成摘要 conversation_text \n.join([ f{m.role}: {m.content[:200]} # 每条取前200字符 for m in messages ]) prompt f请用 100 字以内概括以下对话的核心内容 {conversation_text} 摘要 response await self.llm.chat([ {role: user, content: prompt} ]) return response.content def estimate_tokens(self, text: str) - int: 估算 token 数简单实现 return len(text) // 4 # 约等于 def _truncate(self, messages: List[Message], max_tokens: int) - List[Message]: 截断消息 truncated [] total_tokens 0 for m in reversed(messages): tokens self.estimate_tokens(m.content) if total_tokens tokens max_tokens: break truncated.insert(0, m) total_tokens tokens return truncated2.2 System Prompt 精简System Prompt 往往很长但很多是车轱辘话。解法提取核心指令用 Few-shot 示例替代冗长说明# 优化前800 tokens SYSTEM_PROMPT_LONG 你是一个专业的客服助手。 你的职责是帮助用户解决问题。 你必须遵循以下原则 1. 始终保持礼貌 2. 仔细阅读用户的问题 3. 提供准确、完整的答案 4. 如果不确定请明确告知用户 5. 不要编造信息 ... [此处省略 500 字] # 优化后200 tokens SYSTEM_PROMPT_SHORT [角色] 专业客服简洁准确地回答用户问题。 [原则] 不知道就说不知道不要编造。 [格式] 优先用列表控制在 3 点以内。2.3 用户输入预处理用户输入可能包含重复信息无关背景格式混乱解法输入预处理class InputPreprocessor: 用户输入预处理器 def process(self, user_input: str) - str: # 1. 去除重复空格、换行 text re.sub(r\s, , user_input) # 2. 去除明显的车轱辘话 text self._remove_redundancy(text) # 3. 提取核心问题简单实现 text self._extract_core(text) return text.strip() def _remove_redundancy(self, text: str) - str: 去除冗余表达 patterns [ r我之前已经说过.*?。, r麻烦.*?一下, # 麻烦看一下、麻烦处理一下 r谢谢.*?谢谢, # 重复感谢 ] for p in patterns: text re.sub(p, , text) return text def _extract_core(self, text: str) - str: 提取核心问题 # 简单实现取第一段作为核心 sentences text.split(。) if len(sentences) 3: # 如果超过 3 句只保留前 2 句 return 。.join(sentences[:2]) 。 return text三、输出优化减少无效 Token3.1 输出长度限制class OutputLimiter: 输出长度限制器 def __init__(self, max_output_tokens: int 500): self.max_output_tokens max_output_tokens def wrap_prompt(self, original_prompt: str) - str: 在 prompt 中限制输出长度 return f{original_prompt} [重要] 请将回答控制在 {self.max_output_tokens} Token 以内。 - 优先给出核心答案 - 如需详细说明另起一段 - 不要重复问题或解释你是 AI3.2 结构化输出让模型直接输出 JSON/结构化内容减少解释性文字。from pydantic import BaseModel from typing import Optional class WeatherResponse(BaseModel): city: str date: str weather: str temperature: str suggestion: Optional[str] None # 在 prompt 中指定格式 PROMPT_WITH_FORMAT 查询天气后请按以下 JSON 格式输出 json {{ city: 城市名, date: 日期, weather: 天气状况, temperature: 温度范围, suggestion: 出行建议可选 }} # 使用 LangChain 的 Pydantic 输出解析器 from langchain.output_parsers import PydanticOutputParser parser PydanticOutputParser(pydantic_objectWeatherResponse) chain prompt | llm | parser result chain.invoke({query: 北京明天天气}) # result 类型是 WeatherResponse直接可用四、模型路由让合适的任务用合适的模型4.1 什么时候用小模型场景推荐模型理由简单问答GPT-4o-mini成本只有 4o 的 6%意图分类GPT-4o-mini不需要推理能力信息提取GPT-4o-mini结构化提取很简单复杂推理GPT-4o需要强推理能力创意写作GPT-4o需要更好的表达能力代码生成GPT-4o逻辑复杂容易出错4.2 路由实现from enum import Enum from typing import Optional import asyncio class ModelType(Enum): FAST gpt-4o-mini # 快速、便宜 BALANCED gpt-4o # 平衡 POWER gpt-4o-2024-05 # 强力 class ModelRouter: 模型路由器 def __init__(self, llm_factory): self.llm_factory llm_factory self.router_prompt 请根据任务复杂度选择合适的模型 - 简单任务问答、分类、提取选择 fast - 中等任务需要一定推理选择 balanced - 复杂任务深度推理、创意、代码选择 power 只返回一个词fast / balanced / power 任务{task} async def route(self, task: str) - ModelType: 路由任务到合适的模型 # 判断任务复杂度 response await self.llm_factory.create().achat([ {role: user, content: self.router_prompt.format(tasktask)} ]) choice response.content.strip().lower() if fast in choice: return ModelType.FAST elif power in choice: return ModelType.POWER else: return ModelType.BALANCED async def route_batch(self, tasks: list[str]) - list[ModelType]: 批量路由 return await asyncio.gather(*[self.route(t) for t in tasks]) def get_llm(self, model_type: ModelType): 获取对应模型 return self.llm_factory.create(model_type.value) # 使用示例 async def handle_request(task: str): router ModelRouter(llm_factory) # 自动路由 model_type await router.route(task) llm router.get_llm(model_type) # 根据模型不同成本也不同 cost_map { ModelType.FAST: 0.001, ModelType.BALANCED: 0.01, ModelType.POWER: 0.03 } print(f使用模型: {model_type.value}, 预估成本: ${cost_map[model_type]}) return await llm.achat([{role: user, content: task}])4.3 路由效果任务类型路由前路由后成本降低简单问答$0.01$0.000694%意图分类$0.01$0.000694%复杂推理$0.01$0.010%平均$0.01$0.00370%五、自部署终极成本优化5.1 什么时候需要自部署日均 API 调用量 100,000 次对数据安全有严格要求不能发送给第三方需要深度定制微调、特殊优化有自己的 GPU 资源5.2 自部署架构┌─────────────────────────────────────────────┐ │ 用户请求 │ └─────────────────┬───────────────────────────┘ ↓ ┌─────────────────────────────────────────────┐ │ 负载均衡 │ │ (Nginx) │ └─────────────────┬───────────────────────────┘ ↓ ┌─────────────────────────────────────────────┐ │ API Gateway │ │ (FastAPI) │ └─────────────────┬───────────────────────────┘ ↓ ┌─────────────────────────────────────────────┐ │ vLLM 推理集群 │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │ │ GPU 1 │ │ GPU 2 │ │ GPU 3 │ │ │ │(A100 80G)│ │(A100 80G)│ │(A100 80G)│ │ │ └─────────┘ └─────────┘ └─────────┘ │ └─────────────────┬───────────────────────────┘ ↓ ┌─────────────────────────────────────────────┐ │ 模型权重 │ │ (Qwen3-32B / LLaMA3-70B) │ └─────────────────────────────────────────────┘5.3 vLLM 部署实战# 1. 安装 vLLM pip install vllm # 2. 启动 vLLM 服务 vllm serve Qwen/Qwen2.5-32B-Instruct \ --dtype half \ # 使用半精度 --gpu-memory-utilization 0.95 \ # GPU 显存利用率 --max-model-len 8192 \ # 最大上下文长度 --tensor-parallel-size 2 # 张量并行2 卡 --port 8000 # 3. 测试 curl http://localhost:8000/v1/chat/completions \ -H Content-Type: application/json \ -d { model: Qwen/Qwen2.5-32B-Instruct, messages: [{role: user, content: 你好}], max_tokens: 100 }5.4 量化进一步降低成本量化方法精度损失显存减少推荐场景FP16几乎无50%首选INT82-3%75%显存不足时INT45-10%87%极致压缩# INT8 量化部署 vllm serve Qwen/Qwen2.5-32B-Instruct \ --dtype int8 \ --quantization_method awq \ --max-model-len 81925.5 自部署成本对比方案月成本单次请求成本GPT-4o API$15,000$0.01GPT-4o-mini API$2,000$0.001自部署 Qwen3-32B$3,000$0.0005自部署 LLaMA3-70B$5,000$0.001结论日均请求量超过 100k 时自部署开始划算。六、缓存让重复请求不花钱6.1 什么可以缓存相同问题的回答相同文档的摘要相同搜索的结果相同工具的返回值6.2 缓存实现import hashlib import json from typing import Optional import redis class ResponseCache: 响应缓存 def __init__(self, redis_client: redis.Redis, ttl_seconds: int 3600): self.redis redis_client self.ttl ttl_seconds def _make_key(self, prompt: str, model: str) - str: 生成缓存 key content f{model}:{prompt} return fcache:{hashlib.sha256(content.encode()).hexdigest()} def get(self, prompt: str, model: str) - Optional[str]: 获取缓存 key self._make_key(prompt, model) cached self.redis.get(key) return cached.decode() if cached else None def set(self, prompt: str, model: str, response: str): 设置缓存 key self._make_key(prompt, model) self.redis.setex(key, self.ttl, response) # 使用示例 async def chat_with_cache(llm, cache: ResponseCache, prompt: str): # 先查缓存 cached cache.get(prompt, llm.model_name) if cached: print([CACHE HIT]) return cached # 没有缓存调用 LLM response await llm.chat(prompt) # 存入缓存 cache.set(prompt, llm.model_name, response) return response6.3 缓存命中率场景缓存命中率成本降低FAQ 客服60-80%60-80%文档问答30-50%30-50%开放对话5-10%5-10%七、总结成本优化路线图阶段1输入优化现在 ↓ 上下文压缩、Prompt 精简、输入预处理 阶段2输出优化下周 ↓ 长度限制、结构化输出 阶段3模型路由下个月 ↓ 简单任务用小模型复杂任务用大模型 阶段4自部署3个月后 ↓ vLLM 量化高频场景自部署 阶段5智能缓存半年后 → 基于语义相似度的缓存核心思想成本优化是一个系统工程从输入、输出、模型、部署、缓存五个层面同时发力。踩坑经验汇总上下文压缩要谨慎——不要压缩掉关键信息尤其是 Agent 需要知道的历史决策模型路由不是 100% 准确——路由模型也可能选错准备兜底方案自部署的 GPU 成本不容忽视——显卡折旧、电费、运维都是钱缓存不是万能的——需要考虑缓存失效、更新一致性量化有精度损失——INT4 在某些任务上效果明显下降先测试再上线本篇代码https://github.com/dazhuang-zs/run_little_donkey/blob/master/docs/articles/ai-dev-advanced-03-inference-cost-optimization.md篇④预告Context Engineering 深入——长上下文的真相讲百万 token 上下文的坑、压缩策略、什么时候真需要长上下文。

相关新闻