LiteLLM+LangGraph多模型统一调度

痛点：LLM 厂商 API 的「七国八制」

OpenAI 返回 content: null + tool_calls（纯工具调用）

Claude 把思考过程放在 thinking block，内容在 text block

DeepSeek-Reasoner 有 reasoning_content 单独字段

通义的 qwen-vl 多模态字段名又跟其他模型不一样

国产模型限流策略各不相同，重试逻辑要分别写

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

方案：LiteLLM 统一接口 + LangGraph 编排

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 改动。

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

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

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

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]

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

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 还是通义。

三、智能路由：按任务类型选模型

不是所有任务都需要 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%，体验几乎没差。

四、统一流式响应

不同模型的流式协议差异最大。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 接收的就是统一格式，不再为每家厂商写流式解析器。

五、成本控制与自动降级

模型字典里的 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 × 单价)，实时统计，超额自动降级到便宜模型。

三级降级链路

模型级降级：主模型 5xx/超时 → 自动切 fallback 链路（gpt-4o → claude-3-5-sonnet → qwen-max）。

能力级降级：用户要 vision 但所有 vision 模型都挂了 → 返回「当前不可用」而不是抛错。

套餐级降级：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

真实案例：某次 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

只用一家厂商：直接用厂商 SDK 更轻量。

模型数量 < 5 个：LiteLLM 收益不明显，自己封装反而更可控。

不需要多步推理：LangGraph 是有状态编排，简单 chat 用它有点重，直接 dispatcher.chat() 就够。

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

八、最佳实践总结

模型字典化：13 个元数据字段，前端/后端/统计/降级全读这一份。

接口统一：LiteLLM 抹平厂商差异，业务代码只换 model 字符串。

智能路由：LangGraph 在编排层按任务类型选模型，成本降 78%。

三级降级：模型级 / 能力级 / 套餐级，保证可用性。

成本控制：实时记录 token + 价格，超额自动降级。

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

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

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

🌐 官网：https://aizhs.top

💻 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.

For further actions, you may consider blocking this person and/or reporting abuse