AI Engineering Academy · 课时

批处理、模型路由与成本仪表板

将简单请求路由到 GPT-4o-mini 等成本更低的模型，将复杂请求路由到 GPT-4o，批处理非紧急请求，并构建按功能跟踪支出的成本仪表板。

第 4 / 4 课13 个步骤

批处理、模型路由与成本仪表板 是 CoddyKit 上的免费 AI Engineering Academy 课时。这是第 4 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 AI Engineering Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Engineering Academy 课程共包含 4 节课。

成本优化的另外三个着力点

除了缓存之外，还有三种策略可以大幅降低 LLM 的运行成本：批处理（延后非紧急请求，以较低的 API 费率批量提交）、模型路由（将简单查询路由到低成本模型，将复杂查询路由到功能强大的模型）以及成本仪表板（跟踪各项功能的支出，以确定优化收益最高的部分）。综合使用这些策略，在缓存之外还可以将成本进一步降低 40%–60%。

OpenAI 批处理 API：异步工作负载五折优惠

OpenAI 的批处理 API接受一个包含最多 50,000 个请求的 JSONL 文件，并在 24 小时内异步处理这些请求，价格为标准价格的 50%。它非常适合非交互式工作负载，例如为大型文档语料库生成嵌入、生成产品描述、运行夜间评估或预处理训练数据。代价是延迟：结果需要数小时后才能获得，而不是立即返回。

import json
from openai import OpenAI

client = OpenAI()

# Prepare batch file
requests = [
    {
        'custom_id': f'req_{i}',
        'method': 'POST',
        'url': '/v1/chat/completions',
        'body': {
            'model': 'gpt-4o-mini',
            'messages': [
                {'role': 'user', 'content': f'Summarize: {document}'}
            ],
            'max_tokens': 150,
        }
    }
    for i, document in enumerate(documents_to_process)
]

# Write JSONL batch file
with open('/tmp/batch_requests.jsonl', 'w') as f:
    for req in requests:
        f.write(json.dumps(req) + '\n')

# Upload and submit batch
with open('/tmp/batch_requests.jsonl', 'rb') as f:
    batch_file = client.files.create(file=f, purpose='batch')

batch = client.batches.create(
    input_file_id=batch_file.id,
    endpoint='/v1/chat/completions',
    completion_window='24h',
)
print(f'Batch {batch.id} submitted, status: {batch.status}')

轮询批处理结果

提交批处理后，请持续轮询其状态，直到处理完成（状态会从 in_progress 变为 completed）。完成后，下载包含所有请求结果的输出文件。输出文件的每一行都是一个 JSON 对象，其中包含请求中的 custom_id，以及 response 或 error 字段之一；请始终处理这两种情况，因为批处理中的各个请求可能独立失败。

import time

def wait_for_batch(batch_id: str, poll_interval: int = 60) -> str:
    while True:
        batch = client.batches.retrieve(batch_id)
        print(f'Status: {batch.status}, completed: {batch.request_counts.completed}')

        if batch.status == 'completed':
            return batch.output_file_id
        elif batch.status == 'failed':
            raise RuntimeError(f'Batch failed: {batch.errors}')

        time.sleep(poll_interval)

def download_batch_results(output_file_id: str) -> list[dict]:
    content = client.files.content(output_file_id)
    results = []
    for line in content.text.strip().split('\n'):
        results.append(json.loads(line))
    return results

output_file_id = wait_for_batch(batch.id)
results = download_batch_results(output_file_id)
for result in results[:3]:
    print(result['custom_id'], result.get('response', {}).get('body', {}).get('choices', [{}])[0])

模型路由：根据复杂度匹配模型规模

模型路由会将每个请求分配给能够妥善处理它的最低成本模型。GPT-4o-mini 的成本大约比 GPT-4o 低 30 倍，但在处理简单的分类、提取和简短问答任务时同样出色。将简单的结构化任务路由到低价的小型模型，将复杂推理、长上下文综合和细致入微的生成任务路由到强大的大型模型。即使将 60% 的流量路由到低价模型，也能节省大量成本。

CHEAP_MODEL = 'gpt-4o-mini'
POWERFUL_MODEL = 'gpt-4o'

def classify_query_complexity(query: str) -> str:
    # Heuristic-based routing (replace with ML classifier for production)
    words = query.split()
    has_code = any(c in query for c in ['```', 'def ', 'class ', 'SELECT ', 'function '])
    is_multi_step = any(w in query.lower() for w in ['compare', 'analyze', 'explain why', 'evaluate'])
    is_long = len(words) > 50

    if has_code or is_multi_step or is_long:
        return POWERFUL_MODEL
    return CHEAP_MODEL

def routed_completion(messages: list[dict]) -> str:
    user_query = messages[-1].get('content', '')
    model = classify_query_complexity(user_query)
    print(f'Routing to: {model}')
    response = client.chat.completions.create(model=model, messages=messages)
    return response.choices[0].message.content

基于 LLM 的路由，提升准确率

启发式路由速度快，但不够稳健。更准确的方法是使用低价的小型分类模型来决定应将请求路由到哪个模型。您可以使用所在领域中简单查询与复杂查询的示例对小型模型进行微调，也可以直接使用 GPT-4o-mini 通过少样本提示来完成分类。分类器调用只需花费几百个输入令牌，远低于将复杂查询错误地路由到低价模型所产生的代价，因为后者可能会生成错误答案。

CLASSIFIER_SYSTEM = '''You are a query complexity classifier.
Classify the user query as SIMPLE or COMPLEX.
SIMPLE: factual lookup, extraction, classification with clear answer.
COMPLEX: multi-step reasoning, synthesis, comparison, code generation, long-form writing.
Reply with just SIMPLE or COMPLEX.'''

def llm_classify_complexity(query: str) -> str:
    response = client.chat.completions.create(
        model='gpt-4o-mini',  # use cheap model for routing
        messages=[
            {'role': 'system', 'content': CLASSIFIER_SYSTEM},
            {'role': 'user', 'content': query},
        ],
        max_tokens=10,
        temperature=0,
    )
    label = response.choices[0].message.content.strip()
    return POWERFUL_MODEL if label == 'COMPLEX' else CHEAP_MODEL

跟踪每项功能的成本

要知道应将优化工作重点放在哪里，您需要跟踪每项应用功能的成本，而不只是总支出。为每次 LLM 调用附加功能标签，并按标签累计令牌成本。search_summarization 可能消耗 40% 的预算，却只服务 5% 的流量，因此应成为高优先级的优化目标。user_onboarding 可能成本较高，但服务的是高价值流程，您不希望降低其质量。

from collections import defaultdict

cost_tracker = defaultdict(lambda: {'prompt_tokens': 0, 'completion_tokens': 0, 'cost_usd': 0.0})

MODEL_PRICING = {
    'gpt-4o-mini': {'input': 0.15 / 1e6, 'output': 0.60 / 1e6},
    'gpt-4o':      {'input': 2.50 / 1e6, 'output': 10.00 / 1e6},
}

def tracked_completion(feature: str, messages: list[dict], model: str = 'gpt-4o-mini') -> str:
    response = client.chat.completions.create(model=model, messages=messages)
    usage = response.usage
    pricing = MODEL_PRICING.get(model, {'input': 0, 'output': 0})
    cost = usage.prompt_tokens * pricing['input'] + usage.completion_tokens * pricing['output']

    cost_tracker[feature]['prompt_tokens'] += usage.prompt_tokens
    cost_tracker[feature]['completion_tokens'] += usage.completion_tokens
    cost_tracker[feature]['cost_usd'] += cost

    return response.choices[0].message.content

def print_cost_report():
    print(f'{"Feature":<30} {"Prompt":<10} {"Completion":<12} {"Cost USD":<12}')
    for feature, stats in sorted(cost_tracker.items(), key=lambda x: -x[1]['cost_usd']):
        print(f'{feature:<30} {stats["prompt_tokens"]:<10} {stats["completion_tokens"]:<12} ${stats["cost_usd"]:.4f}')

构建简单的成本仪表板

实用的成本仪表板会汇总功能级别的支出数据，并通过简单的 HTTP 端点对外提供。将累计成本存储在 Redis 中，并使用按天汇总的键，以便随时间了解支出趋势。将此仪表板加入内部开发者工具，让团队几乎实时地看到功能发布对成本的影响，并在失控的支出变成巨额账单之前及时发现。

from fastapi import FastAPI
import datetime

app = FastAPI()

async def record_cost(feature: str, model: str, prompt_tokens: int, completion_tokens: int):
    pricing = MODEL_PRICING.get(model, {'input': 0, 'output': 0})
    cost = prompt_tokens * pricing['input'] + completion_tokens * pricing['output']
    today = datetime.date.today().isoformat()
    key = f'cost:{today}:{feature}:{model}'
    await async_r.incrbyfloat(key, cost)
    await async_r.expire(key, 86400 * 30)  # keep 30 days

@app.get('/dashboard/costs')
async def cost_dashboard():
    today = datetime.date.today().isoformat()
    pattern = f'cost:{today}:*'
    costs = {}
    async for key in async_r.scan_iter(match=pattern):
        value = await async_r.get(key)
        parts = key.split(':')
        feature_model = ':'.join(parts[2:])
        costs[feature_model] = float(value or 0)
    return {'date': today, 'costs': costs, 'total': sum(costs.values())}

每月预算提醒

设置每月预算提醒，以便在意外的成本激增变成巨额账单之前及时发现。根据成本跟踪器计算滚动的每日支出，将其推算到月末，并在预测值超过预算阈值时发送 Slack 提醒。一个简单的预测公式——每日支出 * 剩余天数——即使在实际模式并非线性的情况下，也能及早发现失控的请求。

import datetime
import httpx

SLACK_WEBHOOK = 'https://hooks.slack.com/services/YOUR/WEBHOOK'
MONTHLY_BUDGET_USD = 500.0

async def check_budget_alert():
    today = datetime.date.today()
    days_in_month = 30
    day_of_month = today.day
    days_remaining = days_in_month - day_of_month

    # Sum today's costs
    today_total = sum(cost_tracker[f]['cost_usd'] for f in cost_tracker)
    avg_daily = today_total  # simplified: just today's spend
    projected_month = avg_daily * days_in_month

    if projected_month > MONTHLY_BUDGET_USD:
        message = (
            f'LLM Budget Alert: Projected monthly spend ${projected_month:.2f} '
            f'exceeds budget ${MONTHLY_BUDGET_USD:.2f}. '
            f'Today spend: ${today_total:.2f}'
        )
        async with httpx.AsyncClient() as client:
            await client.post(SLACK_WEBHOOK, json={'text': message})

用于速率限制管理的请求队列

流量激增时，请求会触及 OpenAI 的速率限制，并因 429 Too Many Requests 而失败。请求队列会缓冲传入的请求，并以受控速率提交，从而平滑流量峰值。在生产环境中，请使用由 Redis 或 RabbitMQ 等消息代理支持的异步队列，并为临时性的 429 错误实现指数退避重试逻辑。

import asyncio
from asyncio import Queue

class RateLimitedLLMClient:
    def __init__(self, requests_per_minute: int = 500):
        self.rpm = requests_per_minute
        self.queue: Queue = Queue(maxsize=1000)
        self.interval = 60.0 / requests_per_minute

    async def start(self):
        asyncio.create_task(self._worker())

    async def _worker(self):
        while True:
            request_fn, future = await self.queue.get()
            try:
                result = await request_fn()
                future.set_result(result)
            except Exception as e:
                future.set_exception(e)
            await asyncio.sleep(self.interval)

    async def submit(self, request_fn) -> str:
        loop = asyncio.get_event_loop()
        future = loop.create_future()
        await self.queue.put((request_fn, future))
        return await future

整合应用：成本优化技术栈

完整的 LLM 成本优化技术栈分为多个层次：精确缓存消除重复相同查询的调用，语义缓存消除相似查询的调用，前缀缓存降低其余所有调用的输入成本，模型路由为简单查询使用低价模型，批处理将非紧急工作延后以享受五折优惠，而仪表板和提醒则让成本保持可见并处于受控范围内。请根据具体应用的影响程度，按优先级逐步实施这些措施。

# Decision framework for cost optimization priority:
#
# 1. Enable prefix caching (free, zero effort, automatic)
# 2. Add exact caching (high hit rate for FAQ/support bots)
# 3. Add model routing (simple heuristics first, ML classifier later)
# 4. Add semantic caching (complex, high ROI for paraphrase-heavy use cases)
# 5. Enable batch API (only for non-real-time pipelines)
# 6. Build cost dashboard (essential for ongoing monitoring)
#
# Typical combined result in a customer support bot:
# Before: $1,000/month
# After step 1-2: $400/month (-60%)
# After step 3-4: $200/month (-50% of remaining)
# After step 5-6: $150/month and visible

低价模型失败时的级联回退

将请求路由到低价模型时，您必须处理模型生成不满意答案的情况。请对低价模型的输出实施质量检查——检查响应长度、必需字段是否存在，或快速运行 LLM 评审评分——并在质量不足时自动回退到强大的模型。这个安全机制让您可以大胆地将请求路由到低价模型，同时避免用户体验下降。

async def routing_with_fallback(messages: list[dict], min_length: int = 50) -> str:
    # Try cheap model first
    cheap_response = await async_client.chat.completions.create(
        model=CHEAP_MODEL, messages=messages, temperature=0.0
    )
    answer = cheap_response.choices[0].message.content

    # Quality check: response too short indicates poor answer
    if len(answer.strip()) < min_length:
        print(f'Cheap model answer too short ({len(answer)} chars), escalating...')
        powerful_response = await async_client.chat.completions.create(
            model=POWERFUL_MODEL, messages=messages, temperature=0.0
        )
        return powerful_response.choices[0].message.content

    return answer

快速检查

测试您对本课中批处理、模型路由和成本仪表板的理解。

在交互式课程中回答

课程回顾

在本课中，您学到了：OpenAI Batch API 为异步非实时工作负载提供五折优惠；模型路由会为简单任务使用 GPT-4o-mini 等低价模型，为复杂任务使用高价模型；按功能跟踪成本可以揭示应用中哪些部分消耗了最多预算，从而帮助您有效地确定优化优先级。结合前几课介绍的缓存策略，这些技术可以将 LLM 基础设施成本降低 60%–80%。现在，您已完成 LLM 缓存与成本优化课程。

免费开始

用 AI 导师学习 Python — 免费

在浏览器中编写并运行真实代码，获得全天候 AI 导师的即时帮助，并在网页或应用中继续学习。

课程: 30
课程: 120

免费开始学习查看完整课程

常见问题解答

「批处理、模型路由与成本仪表板」课时是免费的吗？

是的 — 「批处理、模型路由与成本仪表板」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Engineering Academy 课程的其余内容，请升级到 CoddyKit PRO。 AI Engineering Academy 课程共包含 4 节课。

「批处理、模型路由与成本仪表板」这节课中我会学到什么？

将简单请求路由到 GPT-4o-mini 等成本更低的模型，将复杂请求路由到 GPT-4o，批处理非紧急请求，并构建按功能跟踪支出的成本仪表板。你通过在浏览器中直接运行的动手代码来练习 AI Engineering Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Engineering Academy 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Engineering Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 4 节课，共 4 节。

「批处理、模型路由与成本仪表板」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Engineering Academy 课中编写并运行代码吗？

能。每节 AI Engineering Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 AI Engineering Academy