在实际部署和使用大语言模型（LLM）的过程中，单个本地推理机器的算力往往有限，难以应对高并发或复杂模型的推理需求。而完全依赖云端 API 不仅成本高昂，还存在数据隐私和网络延迟的问题。因此，构建一个能够智能调度多个本地推理节点，并在必要时无缝降级到云端服务的负载均衡器，成为许多团队在落地 LLM 应用时的核心需求。

本文将围绕如何构建一个“Free LLM Balancer”展开，它能够聚合多个本地推理机器（例如使用 Ollama、vLLM 或 Transformers 部署的模型），并设置云端服务（如 OpenAI API）作为后备方案。我们将从核心概念入手，逐步讲解其架构设计、环境准备、关键配置、代码实现，并深入探讨如何验证调度逻辑、排查常见问题，以及在生产环境中需要注意的最佳实践。无论你是希望提升本地模型的利用率，还是构建一个高可用的混合推理架构，这篇文章都将提供一条清晰的实现路径。

1. 理解 LLM 负载均衡器的核心价值与工作机制

1.1 为什么需要混合本地与云端的负载均衡？

单纯使用本地模型，可能会因为单机 GPU 内存不足或算力瓶颈导致请求超时或失败。而全部请求云端，虽然稳定性高，但长期来看 API 调用费用不菲，且不适合处理敏感数据。一个智能的负载均衡器能够在两者之间取得平衡：优先使用免费的本地资源，只有当本地资源不可用、过载或无法满足特定请求（如本地未部署的模型）时，才将请求转发至付费的云端服务。这种模式既控制了成本，又保证了服务的可用性。

1.2 负载均衡器的基本调度逻辑

一个典型的 LLM Balancer 的核心调度逻辑可以用以下流程来描述：

  1. 接收请求 ：接收一个标准的聊天补全请求，其中包含了模型名称、消息列表等参数。
  2. 路由决策 ：
    • 模型匹配 ：检查请求指定的模型（如 gpt-3.5-turbo ）是否在本地有部署。如果本地有，则进入本地节点池。
    • 健康检查 ：对匹配的本地节点进行健康检查（如检查进程是否存活、GPU 内存是否充足）。
    • 负载评估 ：从健康的节点中，根据预设的策略（如轮询、最少连接数、基于显存的权重）选择一个负载最轻的节点。
  3. 请求转发与降级 ：
    • 将请求转发给选中的本地节点。
    • 如果本地节点池中所有节点均不可用，或请求的模型本地未部署，则自动将请求转发至配置好的云端 Fallback 端点（如 OpenAI API）。
  4. 响应与容错 ：
    • 接收节点的响应。如果本地节点请求超时或返回错误，应具备重试机制（例如，重试其他本地节点一次），若重试后仍失败，则最终降级到云端。
    • 将最终成功的响应返回给客户端。

整个过程中，对客户端而言，它只与负载均衡器交互，无需关心请求最终是由哪台本地机器还是云端服务处理的。

2. 环境准备与核心组件选型

在开始构建之前，我们需要明确技术选型和准备基础环境。本文将使用 Python 作为实现语言，因为它有丰富的 LLM 生态库。

2.1 本地推理引擎的选择

你需要在一台或多台机器上部署本地推理引擎。常见的选择有：

  • Ollama ：非常适合快速在本地运行开源模型（如 Llama, Mistral），管理简单，提供类 OpenAI 的 API 接口。
  • vLLM ：专为高吞吐量推理设计，支持 PagedAttention，非常适合作为生产环境的推理后端。
  • Transformers ：Hugging Face 的库，灵活性最高，但需要自行编写服务化接口（如使用 FastAPI）。

为了简化示例，我们假设本地节点均使用 Ollama 部署，因为它提供的 API 与 OpenAI 兼容，便于统一处理。

2.2 负载均衡器的技术栈

负载均衡器本身是一个 Web 服务，需要接收 HTTP 请求并转发。我们选择：

  • FastAPI ：现代、高性能的 Python Web 框架，自带 API 文档，非常适合构建此类代理服务。
  • HTTPX ：支持异步的 HTTP 客户端库，用于向本地节点和云端发起请求。
  • Pydantic ：用于数据验证和设置管理。

2.3 项目结构与依赖

创建一个新的项目目录，结构如下：

free-llm-balancer/
├── app/
│   ├── __init__.py
│   ├── main.py           # FastAPI 应用入口
│   ├── config.py         # 配置文件
│   ├── balancer.py       # 核心负载均衡逻辑
│   └── models.py         # Pydantic 数据模型
├── requirements.txt
└── README.md

requirements.txt 文件内容：

fastapi==0.104.1
uvicorn==0.24.0
httpx==0.25.2
pydantic==2.5.0
pydantic-settings==2.1.0

使用 pip install -r requirements.txt 安装依赖。

3. 实现负载均衡器的核心代码

3.1 定义配置模型

首先，在 app/config.py 中定义配置，允许通过环境变量进行配置。

from pydantic_settings import BaseSettings
from typing import List
from pydantic import AnyHttpUrl

class Settings(BaseSettings):
    # 本地 Ollama 节点列表，格式为 "http://host:port"
    local_nodes: List[AnyHttpUrl] = [
        "http://192.168.1.10:11434",
        "http://192.168.1.11:11434"
    ]
    # 云端 Fallback 端点，例如 OpenAI
    cloud_endpoint: AnyHttpUrl = "https://api.openai.com/v1"
    # Cloud API Key，从环境变量读取
    cloud_api_key: str

    # 负载均衡策略，可选 "round_robin", "least_connections"
    lb_strategy: str = "round_robin"
    # 请求超时时间（秒）
    request_timeout: int = 60
    # 健康检查超时时间（秒）
    health_check_timeout: int = 5

    class Config:
        env_file = ".env"

settings = Settings()

对应的 .env 文件示例：

LOCAL_NODES='["http://192.168.1.10:11434", "http://192.168.1.11:11434"]'
CLOUD_ENDPOINT="https://api.openai.com/v1"
CLOUD_API_KEY="your-openai-api-key-here"
LB_STRATEGY="round_robin"

3.2 定义数据模型

在 app/models.py 中定义与 OpenAI API 兼容的请求和响应模型。

from pydantic import BaseModel
from typing import List, Optional, Literal

class Message(BaseModel):
    role: Literal["system", "user", "assistant"]
    content: str

class ChatCompletionRequest(BaseModel):
    model: str
    messages: List[Message]
    temperature: Optional[float] = 0.7
    max_tokens: Optional[int] = None
    stream: Optional[bool] = False

class ChatCompletionResponse(BaseModel):
    id: str
    object: str = "chat.completion"
    created: int
    model: str
    choices: List[dict]
    usage: Optional[dict] = None

3.3 实现负载均衡核心逻辑

这是最核心的部分，在 app/balancer.py 中实现。

import asyncio
import random
from typing import List, Dict
import httpx
from app.config import settings
from app.models import ChatCompletionRequest, ChatCompletionResponse

class LLMBalancer:
    def __init__(self):
        self.local_nodes = settings.local_nodes
        self.cloud_endpoint = settings.cloud_endpoint
        self.cloud_api_key = settings.cloud_api_key
        self.client = httpx.AsyncClient(timeout=settings.request_timeout)
        # 简单记录节点的活跃连接数（用于最少连接数策略）
        self.node_connections: Dict[str, int] = {str(node): 0 for node in self.local_nodes}
        self.lb_strategy = settings.lb_strategy

    async def health_check(self, node_url: str) -> bool:
        """检查本地节点是否健康"""
        try:
            async with httpx.AsyncClient(timeout=settings.health_check_timeout) as client:
                resp = await client.get(f"{node_url}/api/tags") # Ollama 的健康检查端点
                return resp.status_code == 200
        except (httpx.ConnectError, httpx.TimeoutException):
            return False

    async def select_local_node(self, model: str) -> str:
        """根据策略选择一个健康的本地节点"""
        healthy_nodes = []
        # 检查所有节点的健康状态
        for node_url in self.local_nodes:
            if await self.health_check(node_url):
                healthy_nodes.append(node_url)

        if not healthy_nodes:
            raise Exception("No healthy local nodes available")

        # 负载均衡策略
        if self.lb_strategy == "round_robin":
            selected_node = random.choice(healthy_nodes)  # 简单随机模拟轮询
        elif self.lb_strategy == "least_connections":
            # 选择连接数最少的节点
            selected_node = min(healthy_nodes, key=lambda url: self.node_connections[str(url)])
        else:
            selected_node = healthy_nodes[0]

        self.node_connections[str(selected_node)] += 1
        return selected_node

    async def send_to_local(self, node_url: str, request: ChatCompletionRequest) -> ChatCompletionResponse:
        """发送请求到指定的本地节点"""
        try:
            # Ollama 的聊天补全端点
            resp = await self.client.post(
                f"{node_url}/api/chat",
                json=request.dict()
            )
            resp.raise_for_status()
            response_data = resp.json()
            # 将 Ollama 的响应格式转换为类 OpenAI 格式
            return self._format_ollama_response(request.model, response_data)
        finally:
            self.node_connections[node_url] -= 1  # 请求完成，减少连接数

    async def send_to_cloud(self, request: ChatCompletionRequest) -> ChatCompletionResponse:
        """发送请求到云端 Fallback"""
        headers = {
            "Authorization": f"Bearer {self.cloud_api_key}",
            "Content-Type": "application/json"
        }
        resp = await self.client.post(
            f"{self.cloud_endpoint}/chat/completions",
            json=request.dict(),
            headers=headers
        )
        resp.raise_for_status()
        return ChatCompletionResponse(**resp.json())

    def _format_ollama_response(self, model: str, ollama_resp: dict) -> ChatCompletionResponse:
        """将 Ollama 的响应格式转换为与 OpenAI 兼容的格式"""
        # 这是一个简化版的转换，实际需要根据 Ollama 的响应结构仔细映射
        import time
        return ChatCompletionResponse(
            id=f"local-{int(time.time())}",
            created=int(time.time()),
            model=model,
            choices=[{
                "index": 0,
                "message": {
                    "role": "assistant",
                    "content": ollama_resp.get("message", {}).get("content", "")
                },
                "finish_reason": "stop"
            }],
            usage={}  # Ollama 可能不返回 usage，留空
        )

    async def chat_completion(self, request: ChatCompletionRequest) -> ChatCompletionResponse:
        """处理聊天补全请求的核心方法"""
        # 策略：总是先尝试本地
        try:
            selected_node = await self.select_local_node(request.model)
            return await self.send_to_local(selected_node, request)
        except Exception as local_error:
            # 本地节点全部失败或不可用，降级到云端
            print(f"Local inference failed: {local_error}. Falling back to cloud.")
            return await self.send_to_cloud(request)

balancer = LLMBalancer()

3.4 创建 FastAPI 主应用

在 app/main.py 中创建 API 端点。

from fastapi import FastAPI, HTTPException
from app.models import ChatCompletionRequest, ChatCompletionResponse
from app.balancer import balancer

app = FastAPI(title="Free LLM Balancer", version="1.0.0")

@app.post("/v1/chat/completions", response_model=ChatCompletionResponse)
async def chat_completion(request: ChatCompletionRequest):
    """
    提供与 OpenAI Chat Completions API 兼容的端点。
    该端点会智能地将请求路由到本地节点或云端。
    """
    try:
        response = await balancer.chat_completion(request)
        return response
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")

@app.get("/health")
async def health_check():
    """负载均衡器自身的健康检查端点"""
    return {"status": "healthy"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

现在，你可以使用 uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload 启动服务。

4. 运行验证与结果分析

4.1 启动并测试服务

  1. 启动负载均衡器 ：在项目根目录下运行 uvicorn app.main:app --reload 。
  2. 验证端点 ：访问 http://localhost:8000/docs 查看自动生成的 API 文档。
  3. 发送测试请求 ：使用 curl 或 Python 脚本模拟客户端请求。

示例请求（curl）： 

curl -X POST "http://localhost:8000/v1/chat/completions" \
-H "Content-Type: application/json" \
-d '{
  "model": "llama2",
  "messages": [
    {"role": "user", "content": "请用中文介绍一下你自己。"}
  ],
  "temperature": 0.7
}'

预期的成功响应（类 OpenAI 格式）： 

{
  "id": "local-1700000000",
  "object": "chat.completion",
  "created": 1700000000,
  "model": "llama2",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "你好！我是一个由Meta开发的大型语言模型..."
      },
      "finish_reason": "stop"
    }
  ],
  "usage": {}
}

4.2 如何验证调度逻辑

  1. 本地节点正常 ：当所有本地节点健康时，请求应由本地节点处理。你可以查看负载均衡器和 Ollama 节点的日志来确认。
  2. 本地节点故障 ：手动停止一个或多个 Ollama 服务，然后发送请求。负载均衡器应能检测到节点不健康，并将请求路由到其他健康节点或云端。
  3. 模型不匹配 ：发送一个请求，其 model 参数为 gpt-3.5-turbo （假设本地只部署了 llama2 ）。负载均衡器应直接将该请求转发至云端。

5. 常见问题排查与优化策略

在实际运行中，你可能会遇到以下问题。

5.1 请求失败或响应缓慢

问题现象 可能原因 检查方式 处理建议
所有请求都超时 负载均衡器无法连接任何节点或云端。 1. 检查负载均衡器网络。
2. 检查 .env 配置中的 URL 和 API Key 是否正确。
3. 检查本地节点和云端的防火墙/安全组规则。		 确保网络连通性和配置准确性。
只有云端请求成功，本地请求失败 本地节点不健康或模型未加载。 1. 直接访问 http://<node_ip>:11434/api/tags 查看节点状态和已加载模型。
2. 检查 Ollama 日志。		 确保 Ollama 服务正常运行，且请求的模型已通过 ollama pull <model> 拉取。
负载不均，某个节点压力过大 负载均衡策略（如轮询）在节点性能差异大时失效。 监控各节点的 GPU 利用率和内存使用情况。 采用更智能的策略，如基于 GPU 内存使用率的权重轮询。

5.2 响应格式不一致错误

客户端期望严格的 OpenAI 格式，但我们的转换函数 _format_ollama_response 可能不完整。

解决方案 ：更精细地映射 Ollama 响应。例如，正确处理 stream 流式响应，完整映射 usage 字段。

# 改进后的 _format_ollama_response 示例片段
def _format_ollama_response(self, model: str, ollama_resp: dict) -> ChatCompletionResponse:
    # ... 其他代码 ...
    message_content = ollama_resp.get("message", {}).get("content", "")
    # 尝试解析 token 使用情况（如果 Ollama 提供）
    # 注意：Ollama 的响应结构可能变，需适配
    prompt_tokens = ollama_resp.get("prompt_eval_count", 0)
    completion_tokens = ollama_resp.get("eval_count", 0)
    total_tokens = prompt_tokens + completion_tokens

    return ChatCompletionResponse(
        # ... id, created 等 ...
        usage={
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens,
            "total_tokens": total_tokens
        }
    )

5.3 云端 API Key 泄露或配置错误

API Key 硬编码在代码或配置文件中存在安全风险。

最佳实践 ：

  • 始终通过环境变量 ( CLOUD_API_KEY ) 传递敏感信息。
  • 使用专门的密钥管理服务（如 Kubernetes Secrets, HashiCorp Vault）。
  • 在负载均衡器前设置一个 API 网关，由网关负责认证和限流，负载均衡器只处理内部路由。

6. 生产环境最佳实践与扩展方向

6.1 增强可靠性

  • 重试机制 ：在 send_to_local 方法中，当某个节点请求失败时，可以立即重试另一个健康节点，而不是直接降级到云端。
  • 断路器模式 ：对频繁失败的节点实施断路器，暂时将其从健康节点列表中剔除，避免持续请求导致雪崩。
  • 更全面的健康检查 ：不仅检查节点是否存活，还可以检查 GPU 显存余量，只将请求路由到有足够资源的节点。

6.2 提升可观测性

  • 日志记录 ：详细记录每个请求的路由决策（最终由哪个节点处理）、耗时、是否降级等。这有助于排查问题和优化调度策略。
  • 指标监控 ：集成 Prometheus 等监控工具，暴露指标如：请求总量、本地/云端请求比例、各节点错误率、请求延迟等。
  • 分布式追踪 ：为每个请求生成唯一的 Trace ID，并在整个请求链路中传递，便于在复杂系统中定位问题。

6.3 扩展功能

  • 成本控制 ：为云端 Fallback 设置月度预算或速率限制，防止意外费用。
  • 模型映射 ：实现一个模型映射表。例如，当客户端请求 gpt-3.5-turbo 时，可以将其映射到本地部署的 llama2-13b-chat 模型，进一步降低成本。
  • 支持更多后端 ：当前实现针对 Ollama，可以抽象一个 Provider 接口，使其支持 vLLM、TGI（Text Generation Inference）等其他推理后端。

构建一个稳定高效的 LLM 负载均衡器是一个迭代过程。从本文提供的最小可行方案出发，结合具体的业务需求、基础设施和监控体系，逐步完善其功能与可靠性，是将其成功应用于生产环境的关键。

# Ollama
Logo
码道开发者社区

码道开发者社区，聚焦华为云码道 CodeArts 代码智能体，沉淀 Agent、Skill、鸿蒙开发实战内容，供开发者查阅资料、交流技术、分享工程实践

更多推荐

cover

Claude API + Python实战：结构化输出后，如何拦住“看起来正确”的JSON

avatar 码道开发者社区
cover

GLM-5.3-Flash深度解析：3200亿参数只卖Opus四十分之一，国产算力首次跑通前沿模型

avatar 码道开发者社区
cover

AI Agent的应用场景

avatar 码道开发者社区