用 WorkBuddy 搭多后端 LLM 网关：5 个坑、一套降级链，从本地 Ollama 到云端全打通

用户请求
   │
   ▼
┌──────────────┐   失败(超时/校验不通过/报错)
│  local       │ ─────────────────────┐
│  Ollama      │                      │
└──────────────┘                      ▼
                            ┌──────────────────┐
                            │  siliconflow     │
                            └──────────────────┘
                                      │ 失败
                                      ▼
                            ┌──────────────────┐
                            │  bailian         │
                            └──────────────────┘
                                      │ 全失败
                                      ▼
                              抛出聚合异常

uv init ai-arch-lab
cd ai-arch-lab
uv add openai httpx pydantic pyyaml

ai-arch-lab/
├── config.yaml          # 后端配置（解耦）
├── gateway.py           # 核心网关
├── schemas.py           # Pydantic 输出模型
└── demo.py              # 调用示例

backends:
  - name: local
    base_url: "http://localhost:11434/v1"
    api_key: "ollama"
    model: "qwen2.5-coder:7b"
    timeout: 30
  - name: siliconflow
    base_url: "https://api.siliconflow.cn/v1"
    api_key: "${SILICONFLOW_API_KEY}"
    model: "Qwen/Qwen2.5-7B-Instruct"
    timeout: 60
  - name: bailian
    base_url: "https://dashscope.aliyuncs.com/compatible-mode/v1"
    api_key: "${DASHSCOPE_API_KEY}"
    model: "qwen-plus"
    timeout: 60

from pydantic import BaseModel, Field

class ExtractedTask(BaseModel):
    """从自然语言中抽取出的结构化任务"""
    title: str = Field(description="任务标题，不超过 20 字")
    priority: int = Field(ge=1, le=5, description="优先级 1-5，5 最高")
    tags: list[str] = Field(default_factory=list, description="标签列表")
    deadline: str | None = Field(default=None, description="截止时间，ISO 8601")

import json
import os
import yaml
from openai import OpenAI
from pydantic import ValidationError

class AllBackendsFailed(Exception):
    """所有后端均失败时抛出，携带每个后端的失败原因"""
    def __init__(self, errors: dict[str, str]):
        self.errors = errors
        detail = "; ".join(f"{k}: {v}" for k, v in errors.items())
        super().__init__(f"全部后端失败 -> {detail}")


def _resolve_env(value: str) -> str:
    """把 ${VAR} 替换成环境变量实际值"""
    if isinstance(value, str) and value.startswith("${") and value.endswith("}"):
        return os.environ.get(value[2:-1], "")
    return value


class Gateway:
    def __init__(self, config_path: str = "config.yaml"):
        with open(config_path, encoding="utf-8") as f:
            cfg = yaml.safe_load(f)
        self.backends = cfg["backends"]

    def chat_structured(self, prompt: str, schema: type) -> object:
        """
        按顺序尝试每个后端，返回第一个通过 Pydantic 校验的结果。
        全部失败则抛出 AllBackendsFailed。
        """
        errors: dict[str, str] = {}

        for backend in self.backends:
            name = backend["name"]
            try:
                client = OpenAI(
                    base_url=backend["base_url"],
                    api_key=_resolve_env(backend["api_key"]),
                    timeout=backend["timeout"],
                )

                resp = client.chat.completions.create(
                    model=backend["model"],
                    messages=[{"role": "user", "content": prompt}],
                    response_format={
                        "type": "json_schema",
                        "json_schema": {
                            "name": schema.__name__,
                            "schema": schema.model_json_schema(),
                        },
                    },
                )

                raw = resp.choices[0].message.content
                # 双重保险：即使模型说它遵守了 schema，也要自己校验一遍
                result = schema.model_validate_json(raw)
                print(f"[OK] 后端 {name} 命中")
                return result

            except (ValidationError, json.JSONDecodeError) as e:
                # 返回内容不合法 -> 换下一个后端
                errors[name] = f"schema 校验失败: {e}"
                continue
            except Exception as e:
                # 网络/超时/鉴权等 -> 换下一个后端
                errors[name] = f"{type(e).__name__}: {e}"
                continue

        raise AllBackendsFailed(errors)

from gateway import Gateway, AllBackendsFailed
from schemas import ExtractedTask

gw = Gateway("config.yaml")

prompt = "帮我记一下：下周三之前务必搞定数据库迁移方案，优先级最高，涉及 devops 和 backend。"

try:
    task: ExtractedTask = gw.chat_structured(prompt, ExtractedTask)
    print(task.model_dump_json(indent=2))
except AllBackendsFailed as e:
    print("全部后端挂了：", e.errors)

{
  "title": "搞定数据库迁移方案",
  "priority": 5,
  "tags": ["devops", "backend"],
  "deadline": "2026-10-07T00:00:00"
}

```json
{"title": "..."}
```

def strip_code_fence(text: str) -> str:
    text = text.strip()
    if text.startswith("```"):
        lines = text.splitlines()
        lines = lines[1:]                      # 去掉 ```json 那一行
        if lines and lines[-1].strip() == "```":
            lines = lines[:-1]                 # 去掉结尾 ```
        return "\n".join(lines).strip()
    return text

def _resolve_env(value: str) -> str:
    if isinstance(value, str) and value.startswith("${") and value.endswith("}"):
        var = value[2:-1]
        if var not in os.environ:
            raise RuntimeError(f"环境变量 {var} 未设置，请先 export/set")
        return os.environ[var]
    return value

# 临时把 local 的 base_url 改成 http://localhost:9999/v1 跑一次
# 期望看到：local 报 ConnectionError -> siliconflow 命中