别让所有请求都挤向大模型：用开放模型构建低成本混合 AI 架构

ollama serve
ollama list

export LOCAL_MODEL='替换为本机的-gemma-模型标签'
export FRONTIER_URL='https://your-provider.example/v1/chat/completions'
export FRONTIER_MODEL='your-frontier-model'
export FRONTIER_TOKEN='your-api-token'

import argparse
import json
import os
import urllib.request

LOCAL_URL = os.getenv("LOCAL_URL", "http://127.0.0.1:11434/api/generate")
LOCAL_MODEL = os.environ["LOCAL_MODEL"]
FRONTIER_URL = os.getenv("FRONTIER_URL")
FRONTIER_MODEL = os.getenv("FRONTIER_MODEL")
FRONTIER_TOKEN = os.getenv("FRONTIER_TOKEN")

LOCAL_TASKS = {"classify", "extract", "summarize"}


def post_json(url, payload, headers=None, timeout=60):
    request = urllib.request.Request(
        url,
        data=json.dumps(payload).encode("utf-8"),
        headers={"Content-Type": "application/json", **(headers or {})},
        method="POST",
    )
    with urllib.request.urlopen(request, timeout=timeout) as response:
        return json.loads(response.read().decode("utf-8"))


def local_call(prompt):
    result = post_json(
        LOCAL_URL,
        {"model": LOCAL_MODEL, "prompt": prompt, "stream": False},
    )
    return result["response"].strip()


def frontier_call(prompt):
    if not all([FRONTIER_URL, FRONTIER_MODEL, FRONTIER_TOKEN]):
        raise RuntimeError("未配置前沿模型端点，无法执行升级请求")

    result = post_json(
        FRONTIER_URL,
        {
            "model": FRONTIER_MODEL,
            "messages": [{"role": "user", "content": prompt}],
            "temperature": 0,
        },
        headers={"Authorization": f"Bearer {FRONTIER_TOKEN}"},
    )
    return result["choices"][0]["message"]["content"].strip()


def build_prompt(task, text):
    if task == "classify":
        return (
            "把文本分类为 support、sales 或 other。"
            "只返回严格 JSON，例如：{\"label\":\"support\",\"confidence\":0.92}。\n"
            f"文本：{text}"
        )
    if task == "extract":
        return (
            "从文本中提取 name、email、company。只返回严格 JSON；"
            "缺失字段使用 null。\n"
            f"文本：{text}"
        )
    if task == "summarize":
        return f"用不超过三句话总结以下文本：\n{text}"
    return f"分析以下问题，给出有依据的解决方案：\n{text}"


def run(task, text):
    prompt = build_prompt(task, text)

    # 边界明确且输入较短的任务优先在本地运行。
    if task in LOCAL_TASKS and len(text) <= 8000:
        answer = local_call(prompt)

        # 结构化任务必须通过机器校验，否则升级到前沿模型。
        if task in {"classify", "extract"}:
            try:
                json.loads(answer)
                return {"route": "local", "answer": answer}
            except json.JSONDecodeError:
                pass
        else:
            return {"route": "local", "answer": answer}

    return {"route": "frontier", "answer": frontier_call(prompt)}


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("task", choices=["classify", "extract", "summarize", "reason"])
    parser.add_argument("text")
    args = parser.parse_args()
    print(json.dumps(run(args.task, args.text), ensure_ascii=False, indent=2))

python router.py classify '客户说账单重复扣款，希望今天退款。'

python router.py reason '比较三种跨区域数据架构，并给出故障恢复与合规方面的取舍。'

每成功任务成本 = 推理费用 + GPU 摊销 + 运维成本 + 失败重试成本 + 升级调用成本