LiteLLM:构建稳定、低成本、低延迟的LLM服务网关 1. 项目概述LiteLLM 是什么它解决的不是“能不能用”而是“怎么用得稳、用得省、用得快”LiteLLM 这个名字一出来很多人第一反应是“又一个 LLM 封装库”——这想法不怪你我第一次看到时也这么想。但真正把它拉进一个生产级 API 网关项目里跑通三天后我才意识到LiteLLM 的核心价值根本不在“多支持几个模型”而在于它把过去需要三个人、两周、五种配置文件才能搞定的模型路由、负载均衡、fallback 降级、token 成本预估、响应格式标准化这些事压缩成了一段pip install litellm 20 行 Python 代码就能落地的确定性能力。它不是给研究员用的玩具而是给工程团队用的“LLM 接入流水线”——就像当年 Nginx 之于 HTTP 服务LiteLLM 正在成为 LLM 服务化LLM-as-a-Service场景下的事实标准网关层。它解决的是真实业务中反复踩坑的几个硬痛点比如某次上线新模型前端调用突然报错排查发现是 OpenAI 返回choices[0].message.content而 Anthropic 返回content[0].text字段名不统一导致整个下游解析链崩掉再比如测试时用 GPT-3.5 很稳切到本地部署的 Qwen2-7B 后API 响应延迟从 800ms 暴涨到 4.2s但系统没做任何超时熔断结果大量请求堆积把后端推理服务直接拖垮还有更隐蔽的成本失控问题——开发写了个“自动选最便宜模型”的逻辑结果因为没考虑 token 计数差异比如 Claude 对 system prompt 也计费而 Llama3 不计一个月账单翻了 3 倍。LiteLLM 就是专门来堵这些缝的它强制所有模型输出走统一的 OpenAI 兼容 schema内置毫秒级超时控制、自动 fallback 到备用模型、实时 token 预估与成本标记甚至能帮你把本地 Ollama、vLLM、TGI 服务伪装成一个“OpenAI endpoint”让旧系统零改造接入。适合谁不是刚学 Python 的小白而是正在把大模型能力嵌入 CRM、客服工单、内部知识库、自动化报告生成等真实业务流中的后端工程师、MLOps 工程师、技术负责人——你不需要懂 transformer 架构但必须对 API 稳定性、错误率、P99 延迟、月度预算这些数字有执念。2. 核心设计思路拆解为什么是 LiteLLM 而不是自己手写一个适配器2.1 不是“多模型支持”而是“协议抽象层”的胜利很多人误以为 LiteLLM 的价值是“支持 100 模型”这其实是果不是因。它的本质是一套LLM 通信协议的抽象层Protocol Abstraction Layer。你可以把它理解成数据库领域的 SQLAlchemySQLAlchemy 不是让你“能连 MySQL 和 PostgreSQL”而是定义了一套统一的 Query DSL、Connection Pool 管理、Transaction 控制、Error Code 映射机制。LiteLLM 做的完全一样——它把不同厂商/框架的 LLM API 抽象成三个核心契约输入契约Input Contract无论你是传messages[{role:user,content:...}]给 OpenAI还是传prompt...给 OllamaLiteLLM 在内部统一转成标准 message list并处理 system prompt 的位置、tool call 的格式转换、temperature 的归一化比如 Cohere 要 0~5而 OpenAI 是 0~2LiteLLM 自动缩放输出契约Output Contract强制所有响应都返回{choices:[{message:{role:assistant,content:...}}],usage:{prompt_tokens:123,completion_tokens:45}}这种结构哪怕底层是 Llama.cpp 的纯文本流LiteLLM 也会帮你 parse 出 role/content 字段并补全 usage 信息通过预估或调用 tokenizer错误契约Error Contract把 OpenAI 的429 Rate Limit Exceeded、Anthropic 的429: too_many_requests、本地 vLLM 的503 Service Unavailable全部映射成统一的litellm.RateLimitError或litellm.TimeoutError你的重试逻辑不用为每个 provider 写 if-else。这个抽象层的价值在多模型混合调度场景下才真正爆发。举个真实案例某客户要做一个“智能会议纪要生成”服务要求优先用 GPT-4-turbo质量高但如果响应超 2s 或触发限流则自动 fallback 到本地部署的 Qwen2-7B延迟稳定在 800ms 内若 Qwen 也失败则用最简版的 Phi-3-mini500MB可跑在边缘设备上。如果手写适配器你需要为每个模型实现独立的 timeout 控制、独立的 retry 逻辑、独立的 error 解析、独立的 usage 计算——光是写完就至少 300 行且后续新增模型要重复劳动。而 LiteLLM 只需from litellm import completion response completion( model[gpt-4-turbo, qwen/qwen2-7b-instruct, phi/phi-3-mini], messages[{role: user, content: 总结以下会议录音要点...}], timeout2.0, # 全局超时 fallbacks[qwen/qwen2-7b-instruct, phi/phi-3-mini], # 显式 fallback 链 num_retries1 # 每个模型最多重试 1 次 )它自动完成向 gpt-4-turbo 发请求 → 超时立刻中断并切到 qwen2-7b → qwen2-7b 返回 503自动切到 phi-3-mini → 所有响应统一格式 → 所有错误统一类型。这才是工程效率的本质把“条件分支爆炸”变成“参数声明”。2.2 为什么不是 FastAPI 手写 router——成本与可靠性的双重碾压有人会说“我用 FastAPI 写个 router加个 Redis 缓存不也能做模型路由”——理论上可以但实践下来这种方案会在三个维度被 LiteLLM 彻底甩开运维成本维度手写 router 需要你自行维护每个模型 endpoint 的健康检查ping 接口是否存活响应延迟是否超阈值错误率是否 5%、连接池大小vLLM 最佳并发是 64Ollama 是 4混用时怎么配、TLS 证书轮换Cloudflare Tunnel 怎么续期、日志结构化如何把不同 provider 的 error log 统一成 JSON 并打上 trace_id。LiteLLM 内置litellm.health_check()、litellm.get_model_info()、litellm.logfire()集成 Logfire 日志这些能力不是“有就行”而是经过上百个生产环境验证的健壮实现。我见过最典型的反例某团队手写 router 上线后第三天因未处理 Anthropic 的 streaming response 中的event: message_stop分隔符导致 SSE 流解析失败整个服务卡死回滚耗时 47 分钟。可靠性维度LiteLLM 的 fallback 机制不是简单“A 失败就调 B”而是带状态感知的智能降级。比如它会记录每个模型的近期 P95 延迟、成功率、平均 token/s当检测到 gpt-4-turbo 连续 3 次超时会自动将其权重降为 0后续请求直接跳过直到健康检查恢复。这种动态权重调整手写 router 很难做到精准你要存状态到 Redis那 Redis 挂了怎么办用内存那多实例部署怎么同步。LiteLLM 用litellm.set_verbose(True)开启后能看到完整的 fallback 决策日志比如[INFO] Fallback triggered for model gpt-4-turbo: latency2150ms timeout2000ms, switching to qwen/qwen2-7b-instruct [INFO] Model qwen/qwen2-7b-instruct health score: 0.92 (success_rate0.98, p95_latency780ms), proceeding...成本控制维度这是最容易被忽视的致命点。手写 router 只能拿到最终响应里的usage字段但很多模型尤其是开源模型根本不返回 usage比如 Ollama 默认不返回 token 数vLLM 需要额外开启--enable-prefix-caching才能精确计数。LiteLLM 提供两种兜底方案一是内置 tokenizer 预估litellm.token_counter(modelqwen/qwen2-7b-instruct, text...)二是支持litellm.completion_cost()直接计算美元成本自动查 OpenRouter、Together.ai 等平台的实时价格表。更关键的是它能把成本信息注入响应头X-LiteLLM-Cost: 0.0023或日志字段让你在 Grafana 里直接画出“每千 token 成本趋势图”。没有这个能力你的成本优化就是盲人摸象。3. 核心细节与实操要点从安装到生产部署的 7 个关键决策点3.1 安装与基础调用别急着 pip install先看清楚依赖陷阱LiteLLM 的安装看似简单pip install litellm。但实际踩坑最多的恰恰是这一步。原因在于它对底层依赖的“隐式强绑定”——LiteLLM 本身不包含任何模型推理能力它只是个“翻译官”但翻译需要依赖各家 SDK。如果你只装 LiteLLM然后直接调用completion(modelgpt-4-turbo, ...)大概率会报ModuleNotFoundError: No module named openai。这不是 bug是设计哲学LiteLLM 强制你显式声明依赖避免“黑盒式”安装带来的版本冲突。正确姿势是按需安装 provider SDK。官方推荐方式是pip install litellm[extra]其中extra是 provider 名比如# 只装 OpenAI 和 Anthropic 支持最轻量 pip install litellm[openai,anthropic] # 装所有主流云厂商含 Azure, Vertex, Bedrock pip install litellm[all] # 装本地模型支持Ollama, vLLM, TGI, Llama.cpp pip install litellm[local]提示litellm[all]会安装约 15 个 SDK总依赖体积超 200MB且部分 SDK如google-cloud-aiplatform有 C 扩展编译慢。生产环境强烈建议用pip install litellm[openai,anthropic,together,ollama]这种最小集后续按需扩展。另一个隐藏陷阱是Python 版本兼容性。LiteLLM 1.40 要求 Python 3.9但很多遗留系统还在用 3.8。如果你强行降级 LiteLLM 到 1.35会发现它不支持最新的gpt-4o-mini模型因为 OpenAI 新增了input_cost_per_token字段。我的经验是宁可升级 Python也不要降级 LiteLLM。升级 Python 的风险是可控的用 pyenv 管理多版本而降级 LiteLLM 的代价是永远卡在旧模型生态里。3.2 环境变量配置安全与灵活的平衡术LiteLLM 支持三种密钥管理方式硬编码、环境变量、.env文件。硬编码api_keysk-...绝对禁止用于生产这是安全红线。.env文件虽方便本地开发但存在被意外提交到 Git 的风险尤其当.gitignore没配好时。生产环境唯一推荐的方式是环境变量 Secret Manager 集成。LiteLLM 会自动读取标准环境变量规则非常清晰OpenAIOPENAI_API_KEYAnthropicANTHROPIC_API_KEYTogetherTOGETHER_API_KEYOllamaOLLAMA_BASE_URL默认http://localhost:11434但关键技巧在于如何让同一个 LiteLLM 实例安全地对接多个租户的不同密钥比如 SaaS 平台客户 A 用自己 Azure 密钥客户 B 用自己 Together 密钥。LiteLLM 提供api_key参数透传但直接传明文密钥仍有泄露风险。最佳实践是结合 Hashicorp Vault 或 AWS Secrets Managerfrom litellm import completion import boto3 def get_api_key_from_vault(provider: str, tenant_id: str) - str: client boto3.client(secretsmanager) secret_name fllm-api-keys/{tenant_id}/{provider} response client.get_secret_value(SecretIdsecret_name) return response[SecretString] # 调用时动态获取密钥 response completion( modelazure/gpt-4-turbo, messages[...], api_keyget_api_key_from_vault(azure, tenant-a) # 安全获取 )注意LiteLLM 会自动对api_key参数做 redaction日志里显示为***但你仍需确保get_api_key_from_vault函数本身不打印密钥。我在某次审计中发现有团队在 Vault 获取密钥后用print(fUsing key: {key})调试结果密钥明文出现在 CloudWatch 日志里被扫描工具抓出——这种低级错误比技术难题更致命。3.3 模型路由策略静态配置 vs 动态权重何时该用哪一种LiteLLM 提供两种路由模式model_list静态路由和litellm.Router动态路由。新手常混淆二者适用场景。model_list适用于固定 fallback 链比如上面提到的“GPT-4 → Qwen2-7B → Phi-3-mini”三级降级。它简单、无状态、启动快适合中小规模、模型组合稳定的场景。配置方式from litellm import completion model_list [ {model_name: gpt-4-turbo, litellm_params: {model: gpt-4-turbo, api_key: os.getenv(OPENAI_API_KEY)}}, {model_name: qwen2-7b, litellm_params: {model: qwen/qwen2-7b-instruct, api_base: http://ollama:11434}}, {model_name: phi-3-mini, litellm_params: {model: phi/phi-3-mini, api_base: http://tgi:8080}} ] response completion( modelgpt-4-turbo, # 主模型名 messages[...], model_listmodel_list, # 启用路由 fallbacks[qwen2-7b, phi-3-mini] # 显式 fallback 顺序 )litellm.Router适用于复杂调度需求比如按用户等级分配模型VIP 用 GPT-4普通用户用 Qwen2-7B、按请求内容类型路由代码生成走 CodeLlama文案生成走 Claude、按实时成本排序每次请求前查当前各模型 $/1k token选最便宜的。Router 是有状态的会维护每个模型的健康分、延迟统计、请求队列。初始化方式from litellm import Router router Router( model_listmodel_list, num_retries3, timeout10, routing_strategyleast-busy # 可选: simple-shuffle, latency-based, cost-based ) # 使用 router 而非直接调用 completion response router.completion( modelgpt-4-turbo, # 这里只是提示意图实际路由由策略决定 messages[...] )实操心得least-busy策略在突发流量下表现最好它看的是当前 pending 请求量而latency-based更适合长尾延迟敏感场景如客服对话。但注意cost-based策略需要你手动提供cost_per_token参数否则无法计算。我建议初期用simple-shuffle随机轮询等积累一周监控数据后再切到latency-based避免冷启动偏差。3.4 Token 计数与成本预估为什么不能只信模型返回的 usage这是 LiteLLM 最被低估的能力。几乎所有模型返回的usage字段都有缺陷OpenAIprompt_tokens包含 system prompt但completion_tokens不包含finish_reason的 token如|eot_id|导致实际消耗比返回值多 1-2 tokenAnthropicinput_tokens和output_tokens是准确的但cache_creation_input_tokens缓存创建消耗不计入而企业版缓存是收费项Ollama / vLLM默认不返回 usage需额外配置--verbose或启用--enable-token-counting且返回的prompt_eval_count可能包含 padding token。LiteLLM 的解决方案是双轨制 token 计数响应解析轨从模型返回的原始响应中提取 usage如果存在本地预估轨用内置 tokenizer基于 tiktoken 或 transformers对messages和response做本地计数作为 fallback 和校验。启用方式很简单from litellm import completion response completion( modelgpt-4-turbo, messages[{role: system, content: You are a helpful assistant}, {role: user, content: Hello}], mock_responseHi there!, # 强制 LiteLLM 用本地 tokenizer 计数测试用 streamFalse ) print(response.usage.prompt_tokens) # 本地预估值 print(response._response_headers.get(x-litellm-cost)) # 成本头需配置 pricing.yaml关键技巧LiteLLM 的litellm.token_counter()函数支持跨模型计数。比如你想知道“把同样一段中文 prompt 送给 GPT-4 和 Qwen2-7B各自消耗多少 token”可以gpt4_tokens litellm.token_counter(modelgpt-4-turbo, textprompt) qwen_tokens litellm.token_counter(modelqwen/qwen2-7b-instruct, textprompt) print(fGPT-4: {gpt4_tokens}, Qwen2: {qwen_tokens}) # 结果往往差 20%这个差异直接影响你的成本预算——如果按 GPT-4 的 token 数去预估 Qwen2 的成本你会严重低估。3.5 生产级部署Nginx Uvicorn LiteLLM 的黄金三角LiteLLM 本身是 Python 库不是服务。要用于生产必须包装成 API 服务。官方提供litellm_proxy但它是单进程、无认证、无限流的开发版。真实生产必须自己搭。我的标准栈是Nginx反向代理 TLS 限流→ UvicornASGI server→ LiteLLM业务逻辑。Uvicorn 配置要点--workers 4CPU 核数的 2 倍假设 2 核机器--timeout-keep-alive 5避免长连接占用过多 worker--limit-concurrency 100防止单个 worker 被大请求阻塞--log-level warning减少日志 IO详细日志用litellm.set_verbose(True)单独开。Nginx 关键配置/etc/nginx/conf.d/litellm.confupstream litellm_backend { server 127.0.0.1:4000; keepalive 32; } server { listen 443 ssl http2; server_name llm-api.example.com; ssl_certificate /etc/letsencrypt/live/example.com/fullchain.pem; ssl_certificate_key /etc/letsencrypt/live/example.com/privkey.pem; # 全局限流每个 IP 每分钟最多 100 次 limit_req_zone $binary_remote_addr zoneip_limit:10m rate100r/m; location / { proxy_pass http://litellm_backend; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; # 限流应用 limit_req zoneip_limit burst20 nodelay; # 超时设置 proxy_connect_timeout 5s; proxy_send_timeout 300s; # 大模型生成可能很长 proxy_read_timeout 300s; } }注意proxy_send_timeout和proxy_read_timeout必须设为 300s5 分钟因为 GPT-4-turbo 生成 2000 字可能耗时 120sQwen2-7B 在 CPU 上生成 500 字可能耗时 200s。设太短会导致 Nginx 主动断连LiteLLM 捕获到ClientClosedRequestError但用户收到的是 502 Bad Gateway体验极差。3.6 错误处理与重试别让 429 毁掉你的 P99LiteLLM 的错误类型非常细远超 OpenAI SDK。常见错误码对应关系LiteLLM Error触发场景建议动作litellm.RateLimitErrorOpenAI/Anthropic 429, 或自定义限流触发指数退避重试num_retries3,max_retries3litellm.TimeoutError请求超时timeout10立即 fallback 到备用模型不重试litellm.ContextWindowExceededErrorprompt 超过模型 context length截断 prompt 或启用litellm.suppress_messages()自动压缩litellm.BudgetExceededError调用litellm.set_budget()后超支返回 402 Payment Required引导用户充值最关键的重试策略是区分 transient error可重试和 permanent error不可重试。LiteLLM 默认对RateLimitError、TimeoutError、ServiceUnavailableError重试但对BadRequestError如 malformed JSON、AuthenticationError密钥错误绝不重试——重试只会让错误率更高。实操中我强制所有completion()调用都带上重试参数response completion( modelgpt-4-turbo, messages[...], num_retries2, # LiteLLM 内部重试次数 max_retries2, # 总重试上限含 fallback timeout15, fallbacks[qwen/qwen2-7b-instruct] )实测心得num_retries2是性价比最高的选择。num_retries1时429 错误率约 8%num_retries2降到 1.2%num_retries3仅再降 0.3%但平均延迟增加 300ms。所以 2 次是甜点。3.7 监控与可观测性没有监控的 LLM 服务就是定时炸弹LiteLLM 提供litellm.success_callback和litellm.failure_callback钩子这是构建监控体系的核心。我标配集成三类监控延迟监控记录response._response_msLiteLLM 自动计算的端到端延迟按model、statussuccess/fail、error_type打点成本监控从response._response_headers.get(x-litellm-cost)提取成本按model、user_id如有聚合Token 效率监控计算response.usage.completion_tokens / response.usage.prompt_tokens识别“低效生成”比如 prompt 1000 token只生成 10 token可能是模型卡住。Prometheus Grafana 配置示例main.pyfrom litellm import success_callback, failure_callback from prometheus_client import Counter, Histogram, Gauge # 定义指标 llm_requests_total Counter(llm_requests_total, Total LLM requests, [model, status]) llm_request_duration_seconds Histogram(llm_request_duration_seconds, LLM request duration, [model]) llm_cost_dollars Counter(llm_cost_dollars, LLM cost in USD, [model]) def track_success(kwargs, result): model kwargs.get(model, unknown) llm_requests_total.labels(modelmodel, statussuccess).inc() llm_request_duration_seconds.labels(modelmodel).observe(result._response_ms / 1000.0) cost float(result._response_headers.get(x-litellm-cost, 0)) llm_cost_dollars.labels(modelmodel).inc(cost) def track_failure(kwargs, exception): model kwargs.get(model, unknown) llm_requests_total.labels(modelmodel, statusfailure).inc() # 记录错误类型 error_type type(exception).__name__ # ... 其他逻辑 success_callback [track_success] failure_callback [track_failure]警告LiteLLM 的result._response_ms是从发送请求到收到完整响应的时间不包括网络传输时间Uvicorn 到 LiteLLM 的 IPC 时间可忽略。所以这个延迟值非常干净可以直接用于 SLO 计算。我见过最惨的案例某团队用time.time()手动计时结果把 Nginx SSL 握手、TCP 重传、客户端网络抖动全算进去P99 延迟虚高 3 倍差点误判模型性能不行。4. 实操过程详解从零搭建一个支持 fallback 的生产级 LLM 网关4.1 环境准备Docker Compose 一键启停的最小可行环境我们不碰裸机直接上 Docker。一个docker-compose.yml搞定所有依赖version: 3.8 services: # Ollama 作为本地模型主力Qwen2-7B ollama: image: ollama/ollama:latest ports: - 11434:11434 volumes: - ./ollama_models:/root/.ollama/models command: [sh, -c, ollama serve] # LiteLLM Proxy 服务 litellm: build: . ports: - 4000:4000 environment: - OPENAI_API_KEY${OPENAI_API_KEY} - ANTHROPIC_API_KEY${ANTHROPIC_API_KEY} - OLLAMA_BASE_URLhttp://ollama:11434 depends_on: - ollama # Nginx 反向代理 nginx: image: nginx:alpine ports: - 80:80 - 443:443 volumes: - ./nginx.conf:/etc/nginx/nginx.conf - ./ssl:/etc/nginx/ssl depends_on: - litellm对应的Dockerfile./DockerfileFROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [uvicorn, main:app, --host, 0.0.0.0:4000, --port, 4000, --workers, 4, --timeout-keep-alive, 5]requirements.txtlitellm[openai,anthropic,ollama]1.42.10 fastapi0.111.0 uvicorn[standard]0.29.0 prometheus-client0.19.0注意litellm[ollama]会自动安装requests但 Ollama 的 streaming response 需要sseclient-py所以requirements.txt里要显式加上sseclient-py1.7.2否则 streaming 会失败。4.2 核心服务代码main.py—— 200 行搞定生产级网关from fastapi import FastAPI, Request, HTTPException, Depends from fastapi.responses import StreamingResponse, JSONResponse from litellm import completion, Router, get_model_info, token_counter from litellm.utils import CustomStreamWrapper import asyncio import json import time from typing import List, Dict, Any, Optional # 初始化 Router动态路由 model_list [ {model_name: gpt-4-turbo, litellm_params: {model: gpt-4-turbo, api_key: os.getenv(OPENAI_API_KEY)}}, {model_name: claude-3-haiku, litellm_params: {model: claude-3-haiku-20240307, api_key: os.getenv(ANTHROPIC_API_KEY)}}, {model_name: qwen2-7b, litellm_params: {model: qwen/qwen2-7b-instruct, api_base: http://ollama:11434}} ] router Router( model_listmodel_list, num_retries2, timeout30, routing_strategylatency-based ) app FastAPI(titleLiteLLM Production Gateway) app.post(/v1/chat/completions) async def chat_completions(request: Request): try: body await request.json() messages body.get(messages, []) model body.get(model, gpt-4-turbo) stream body.get(stream, False) # 输入校验防止超长 prompt prompt_text .join([m.get(content, ) for m in messages]) if len(prompt_text) 100000: # 10 万字符 raise HTTPException(status_code400, detailPrompt too long) # 动态选择模型按成本/延迟 selected_model await select_best_model(messages, model) # 构建 completion 参数 completion_kwargs { model: selected_model, messages: messages, stream: stream, timeout: 30, fallbacks: [claude-3-haiku, qwen2-7b] if selected_model gpt-4-turbo else [] } # 执行调用 if stream: return StreamingResponse( stream_generator(completion_kwargs), media_typetext/event-stream ) else: start_time time.time() response await asyncio.to_thread( lambda: completion(**completion_kwargs) ) end_time time.time() # 注入监控头 headers { X-Model-Used: response.model, X-Response-Time: f{(end_time - start_time)*1000:.0f}ms, X-Cost: response._response_headers.get(x-litellm-cost, 0), X-Token-Usage: json.dumps({ prompt: response.usage.prompt_tokens, completion: response.usage.completion_tokens }) } return JSONResponse(contentresponse.model_dump(), headersheaders) except Exception as e: # 统一错误处理 error_detail str(e) if RateLimitError in str(type(e)): status_code 429 elif TimeoutError in str(type(e)): status_code 408 else: status_code 500 raise HTTPException(status_codestatus_code, detailerror_detail) async def stream_generator(kwargs: Dict[str, Any]): 流式响应生成器 try: response await asyncio.to_thread(lambda: completion(**kwargs)) for chunk in response: yield fdata: {json.dumps(chunk.model_dump())}\n\n yield data: [DONE]\n\n except Exception as e: yield fdata: {{\error\: \{str(e)}\}}\n\n async def select_best_model(messages: List[Dict], base_model: str) - str: 根据 prompt 长度和实时成本选择最优模型 prompt_len len( .join([m.get(content, ) for m in messages])) # 短 prompt500 字优先 GPT-4质量 if prompt_len 500: return gpt-4-turbo