176 大模型统一调度:LiteLLM + LangGraph 如何让 OpenAI/Claude/Qwen/DeepSeek 一键切换

凌晨 3 点,你的 AI 应用挂了。原因:OpenAI 调整了 gpt-4o 的响应字段,你的代码写死了 data.choices[0].message.content,现在偶发返回 refusal。你想临时切到 Claude 顶一下,结果发现 Claude 的 SDK 完全不一样,要改 17 处代码。运维同学哭出声。

这是 LLM 应用最经典的痛:模型厂商 API 各搞一套,切换成本极高。本文讲 IHUI AI 如何用 LiteLLM + LangGraph 把 176 个模型统一成一个接口,实现一键切换、自动降级、成本控制。

---

一、痛点:LLM 厂商 API 的「七国八制」

主流大模型厂商,每家 API 都不一样:

厂商	SDK	字段名	流式协议	Function Call	多模态
OpenAI	openai	choices[0].message.content	SSE	tool_calls	URL/Bus64
Anthropic	@anthropic-ai/sdk	content[0].text	SSE(不同结构)	tool_use block	base64 only
通义千问	dashscope	output.text	自定义事件	tools	URL
DeepSeek	兼容 OpenAI	同 OpenAI	同 OpenAI	同 OpenAI	暂不支持
智谱 GLM	zhipuai	choices[0].message.content	SSE	tools	URL/base64
文心一言	@baidubce	result	流式 JSON	functions	URL
豆包	volcengine-python-sdk	choices[0].message.content	SSE	tools	URL
Ollama	ollama	message.content	NDJSON	tools	base64

真实场景里你会遇到的雷:

1. OpenAI 返回 content: null + tool_calls(纯工具调用)
2. Claude 把思考过程放在 thinking block,内容在 text block
3. DeepSeek-Reasoner 有 reasoning_content 单独字段
4. 通义的 qwen-vl 多模态字段名又跟其他模型不一样
5. 国产模型限流策略各不相同,重试逻辑要分别写

如果你想对接 8 家厂商,就要写 8 套调用代码、8 套错误处理、8 套流式解析。对接 176 个模型呢? 不可能。

---

二、方案:LiteLLM 统一接口 + LangGraph 编排

2.1 LiteLLM 是什么

LiteLLM 是一个 Python 库,把 100+ LLM provider 统一成 OpenAI 格式。你只需要换一个 model 字符串,就能切换底层模型:

from litellm import completion

# OpenAI
response = completion(model="gpt-4o", messages=[...])

# Claude(同一个 API,只换 model)
response = completion(model="claude-3-5-sonnet-20241022", messages=[...])

# 通义千问
response = completion(model="qwen-turbo", messages=[...])

# DeepSeek
response = completion(model="deepseek-chat", messages=[...])

# 本地 Ollama
response = completion(model="ollama/llama3", messages=[...])

返回结构全部是 OpenAI 格式,你的业务代码 0 改动。

2.2 LangGraph 是什么

LangGraph 是 LangChain 出的有状态多步 Agent 编排框架。比 LangChain Agent 更可控:你可以明确定义状态机、节点、边,而不是黑盒 ReAct。

IHUI AI 把 LiteLLM 当「模型访问层」,把 LangGraph 当「业务编排层」:

用户请求 → LangGraph 编排(规划/检索/调用/反思)
                ↓
            LiteLLM 统一接口
                ↓
       176 个模型(OpenAI/Claude/Qwen/...)

---

三、技术细节

3.1 模型字典化

IHUI AI 维护一个模型字典(apps/ai-service/config/model_dict.yaml),每个模型有完整元数据:

models:
  - id: gpt-4o
    provider: openai
    display_name: GPT-4o
    context_window: 128000
    max_output: 16384
    input_price_per_1k: 0.0025    # USD
    output_price_per_1k: 0.01
    supports_streaming: true
    supports_tools: true
    supports_vision: true
    capabilities: [chat, reasoning, code, vision]
    fallback: [claude-3-5-sonnet, gpt-4o-mini]

  - id: claude-3-5-sonnet-20241022
    provider: anthropic
    display_name: Claude 3.5 Sonnet
    context_window: 200000
    max_output: 8192
    input_price_per_1k: 0.003
    output_price_per_1k: 0.015
    supports_streaming: true
    supports_tools: true
    supports_vision: true
    capabilities: [chat, reasoning, code, vision, long_context]
    fallback: [gpt-4o, qwen-max]

  - id: deepseek-reasoner
    provider: deepseek
    display_name: DeepSeek R1
    context_window: 64000
    max_output: 8192
    input_price_per_1k: 0.00055
    output_price_per_1k: 0.0022
    supports_streaming: true
    supports_tools: false
    supports_vision: false
    capabilities: [chat, reasoning, math]
    fallback: [qwen-max, gpt-4o]

  - id: qwen-max
    provider: dashscope
    display_name: 通义千问 Max
    context_window: 32000
    max_output: 8192
    input_price_per_1k: 0.0024
    output_price_per_1k: 0.0096
    supports_streaming: true
    supports_tools: true
    supports_vision: true
    capabilities: [chat, code, vision, chinese_optimized]
    fallback: [gpt-4o, glm-4-plus]

字典是「单一真相」,前端模型选择器、后端调用层、成本统计、降级链路全部从这里读。

3.2 统一调用封装

apps/ai-service/src/llm/dispatcher.py:

from litellm import acompletion
from litellm.utils import get_model_info
from typing import AsyncIterator
import yaml

with open("config/model_dict.yaml") as f:
    MODEL_DICT = {m["id"]: m for m in yaml.safe_load(f)["models"]}

class LLMDispatcher:
    """统一调度 176 个模型"""

    async def chat(
        self,
        model_id: str,
        messages: list[dict],
        stream: bool = False,
        tools: list[dict] | None = None,
        temperature: float = 0.7,
        max_tokens: int | None = None,
    ):
        if model_id not in MODEL_DICT:
            raise ValueError(f"未知模型: {model_id}")

        meta = MODEL_DICT[model_id]
        if tools and not meta["supports_tools"]:
            raise ValueError(f"模型 {model_id} 不支持工具调用")

        # LiteLLM 自动处理 provider 路由
        response = await acompletion(
            model=meta["provider"] + "/" + model_id if meta["provider"] != "openai" else model_id,
            messages=messages,
            stream=stream,
            tools=tools,
            temperature=temperature,
            max_tokens=max_tokens or meta["max_output"],
        )
        return response

    async def chat_with_fallback(
        self,
        model_id: str,
        messages: list[dict],
        **kwargs,
    ):
        """主模型失败时按 fallback 链路降级"""
        chain = [model_id] + MODEL_DICT[model_id].get("fallback", [])
        last_error = None
        for mid in chain:
            try:
                return await self.chat(mid, messages, **kwargs)
            except Exception as e:
                last_error = e
                print(f"模型 {mid} 失败,降级到下一个: {e}")
        raise last_error

业务代码只跟 dispatcher 打交道,不关心是 OpenAI 还是通义。

3.3 自动路由:用对的模型做对的事

不是所有任务都需要 GPT-4o。简单分类用 gpt-4o-mini 即可,数学推理用 deepseek-reasoner,中文长文用 qwen-max,代码生成用 claude-3-5-sonnet。IHUI AI 用 LangGraph 在编排层做路由:

from langgraph.graph import StateGraph, END
from typing import TypedDict, Literal

class AgentState(TypedDict):
    messages: list[dict]
    task_type: Literal["chat", "code", "math", "vision", "long_context"]
    model_id: str
    response: str

def route_task(state: AgentState) -> str:
    """根据消息内容路由到合适的模型"""
    last_msg = state["messages"][-1]["content"]

    if any(kw in last_msg for kw in ["代码", "code", "bug", "重构"]):
        return "claude-3-5-sonnet-20241022"
    if any(kw in last_msg for kw in ["数学", "证明", "方程"]):
        return "deepseek-reasoner"
    if len(state["messages"]) > 20:
        return "claude-3-5-sonnet-20241022"  # 长上下文
    if any(kw in last_msg for kw in ["翻译", "中文", "成语"]):
        return "qwen-max"
    return "gpt-4o-mini"  # 默认便宜模型

def call_model(state: AgentState) -> AgentState:
    return {
        **state,
        "response": dispatcher.chat(state["model_id"], state["messages"]),
    }

graph = StateGraph(AgentState)
graph.add_node("route", lambda s: {**s, "model_id": route_task(s)})
graph.add_node("call", call_model)
graph.add_edge("route", "call")
graph.add_edge("call", END)
graph.set_entry_point("route")
app = graph.compile()

实战收益:简单任务路由到 gpt-4o-mini($0.15/1M token)而不是 gpt-4o($2.5/1M token),成本降 94%,体验几乎没差。

3.4 流式统一

不同模型的流式协议差异最大。LiteLLM 把所有流式响应归一成 OpenAI ChatCompletionChunk 格式:

async def stream_chat(model_id: str, messages: list[dict]):
    async for chunk in await dispatcher.chat(model_id, messages, stream=True):
        # 所有模型统一 delta.content
        if chunk.choices[0].delta.content:
            yield chunk.choices[0].delta.content

前端 SSE 接收的就是统一格式,不再为每家厂商写流式解析器。

3.5 成本控制

模型字典里的 input_price_per_1k / output_price_per_1k 是成本核算基础。每次调用记录 token 消耗:

from datetime import datetime
from packages.database import db

async def record_usage(user_id: str, model_id: str, response):
    meta = MODEL_DICT[model_id]
    usage = response.usage  # LiteLLM 统一字段
    cost = (
        usage.prompt_tokens * meta["input_price_per_1k"] / 1000
        + usage.completion_tokens * meta["output_price_per_1k"] / 1000
    )
    await db.usage_logs.insert({
        "user_id": user_id,
        "model_id": model_id,
        "input_tokens": usage.prompt_tokens,
        "output_tokens": usage.completion_tokens,
        "cost_usd": cost,
        "created_at": datetime.utcnow(),
    })

    # 触发套餐限制检查
    await check_plan_limits(user_id)

实战数据:用户月度成本 = Σ(每次调用 input × 单价 + output × 单价),实时统计,超额自动降级到便宜模型。

3.6 降级策略

IHUI AI 的降级链路有 3 层:

1. 模型级降级:主模型 5xx/超时 → 自动切 fallback 链路(gpt-4o → claude-3-5-sonnet → qwen-max)。
2. 能力级降级:用户要 vision 但所有 vision 模型都挂了 → 返回「当前不可用」而不是抛错。
3. 套餐级降级:Free 用户超额 → 自动切 gpt-4o-mini 而不是直接拒绝,保证基础体验。

async def chat_with_full_fallback(user_id: str, model_id: str, messages: list[dict]):
    try:
        return await dispatcher.chat_with_fallback(model_id, messages)
    except AllModelsFailedError:
        # 套餐级降级
        if await is_free_user(user_id):
            return await dispatcher.chat("gpt-4o-mini", messages)
        raise

---

四、IHUI AI 实战数据

指标	数值
支持模型数量	176 个
覆盖厂商	OpenAI / Anthropic / Google / 通义 / DeepSeek / 智谱 / 文心 / 豆包 / Kimi / Ollama 等 12 家
统一接口字段	OpenAI ChatCompletion 格式
自动故障转移	平均 1.2 秒切到 fallback
平均成本节省(路由后)	78%(对比全部用 GPT-4o)
模型字典字段	13 个元数据(provider/价格/能力/fallback...)

真实案例:某次 OpenAI 全球性故障,我们的 fallback 链路 8 秒内把全部流量切到 Claude + 通义,用户几乎无感。事后日志显示 96% 的请求成功降级,只有 4% 因为同时触发了 Anthropic 限流而失败。

---

五、踩坑总结

坑 1:LiteLLM 的 model 前缀规则

LiteLLM 的 model 参数有前缀规则:claude-3-5-sonnet 不带前缀默认走 OpenAI 兼容,要走原生 Anthropic 必须写 anthropic/claude-3-5-sonnet。我们的 dispatcher 里统一加了前缀逻辑,避免业务代码踩坑。

坑 2:Claude 的 thinking block

Claude 3.5 Sonnet 开启 extended thinking 后,响应里会有 thinking block 和 text block,直接取 content[0].text 会丢思考过程。我们在 dispatcher 层把 thinking 单独抽出来,业务层可选消费。

坑 3:国产模型的 reasoning_content

DeepSeek-Reasoner 和 QwQ 把推理过程放在 reasoning_content 字段,不是 content。LiteLLM 较新版本才统一,旧版本要自己 patch。

坑 4:工具调用字段差异

OpenAI 用 tool_calls,Claude 用 tool_use block,通义用 tools。LiteLLM 帮你抹平了入参,但返回的结构差异在某些边界情况还会漏,要写测试覆盖。

---

六、什么时候不要上 LiteLLM + LangGraph

1. 只用一家厂商:直接用厂商 SDK 更轻量。
2. 模型数量 < 5 个:LiteLLM 收益不明显,自己封装反而更可控。
3. 不需要多步推理:LangGraph 是有状态编排,简单 chat 用它有点重,直接 dispatcher.chat() 就够。

我们之所以上这套,是因为 IHUI AI 定位「176 模型统一平台」,从第一天就要支持 10+ 厂商,统一接口是刚需。

---

七、结语

176 模型统一调度的核心是:

• 模型字典化:13 个元数据字段,前端/后端/统计/降级全读这一份。
• 接口统一:LiteLLM 抹平厂商差异,业务代码只换 model 字符串。
• 智能路由:LangGraph 在编排层按任务类型选模型,成本降 78%。
• 三级降级:模型级 / 能力级 / 套餐级,保证可用性。
• 成本控制:实时记录 token + 价格,超额自动降级。

这套架构让 IHUI AI 在 OpenAI 全球故障时 8 秒切流,用户无感;让 Free 用户也能用上 GPT-4o 级别能力(智能路由到便宜模型);让企业用户一个 API key 切换 176 个模型,不重写一行代码。

如果你也在做 LLM 应用,强烈建议从一开始就用 LiteLLM 统一接口——晚了再迁移,业务代码要改 100+ 处。

---

关于 IHUI AI

IHUI AI 是一站式 8 端全栈 AI 操作系统,Apache 2.0 开源。

• 🌐 官网:https://ihui.ai
• 💻 GitHub:https://github.com/IHUI-INF-AI/IHUI-AI(Star 支持一下 ⭐)
• 📦 8 端同源:Web / API / CLI / Desktop / Extension / Mobile / Miniapp
• 🤖 176 模型:OpenAI / Claude / Gemini / 通义 / DeepSeek / 智谱 / 文心 / 豆包 / Kimi / Ollama
• 💰 定价:Free / Pro ¥49/月 / Team ¥199/人/月 / Enterprise ¥2999/月起

5 分钟 Fork 到上线,替代 ChatGPT Team + Claude Code + Notion AI,月省 $60+。

---

This article was originally published on the IHUI AI Blog. Follow us on GitHub for more AI engineering content.