LLM核心引擎：大模型选型与集成

1
2
3
4
5
6
7
8

# 安装 Ollama（macOS/Linux）
curl -fsSL https://ollama.com/install.sh | sh

# 拉取并运行 Qwen3 模型（8B 版本适合本地开发）
ollama run qwen3:8b

# 拉取 Llama 3.1
ollama run llama3.1:8b

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

# ollama_call.py —— 通过 Ollama 调用本地开源模型
import requests

# Ollama 兼容 OpenAI API 格式，所以请求结构跟调用 OpenAI 一模一样
response = requests.post(
    "http://localhost:11434/v1/chat/completions",  # Ollama 默认服务地址
    json={
        "model": "qwen3:8b",  # 指定要调用的本地模型
        "messages": [{"role": "user", "content": "你好，请介绍一下你自己"}],
    },
)
# 从响应 JSON 中取出模型回复文本
print(response.json()["choices"][0]["message"]["content"])

1
2
3
4
5
6
7
8

# 安装 vLLM
pip install vllm

# 启动 Qwen3 服务，兼容 OpenAI API 格式
python -m vllm.entrypoints.openai.api_server \
    --model Qwen/Qwen3-235B-A22B \
    --tensor-parallel-size 4 \
    --port 8000

1

单次任务成本 = 输入tokens × 输入单价 + 输出tokens × 输出单价

1
2

# 安装依赖
pip install openai anthropic python-dotenv

1
2
3
4

# .env
OPENAI_API_KEY=sk-你的密钥
ANTHROPIC_API_KEY=sk-ant-你的密钥
OLLAMA_BASE_URL=http://localhost:11434

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

# openai_call.py —— OpenAI GPT-5.x 调用示例
from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()  # 从 .env 文件加载环境变量

# 初始化客户端，会自动读取 OPENAI_API_KEY 环境变量
client = OpenAI()

def chat_with_gpt(
    messages: list[dict],
    model: str = "gpt-5.4-mini",
    temperature: float = 0.7,
    tools: list[dict] | None = None,
) -> str:
    """调用 OpenAI 模型进行对话
    
    参数:
        messages: 消息列表，格式同 OpenAI API
        model: 模型名称，如 gpt-5.5 / gpt-5.4 / gpt-5.4-mini
        temperature: 采样温度（0最确定，1最随机）
        tools: 函数调用工具定义
    返回:
        模型回复的文本内容
    """
    response = client.chat.completions.create(
        model=model,                 # 指定模型
        messages=messages,           # 对话历史
        temperature=temperature,     # 控制随机性
        tools=tools,                 # 可选：传入函数定义
    )
    # response.choices[0].message.content 就是模型的文本回复
    return response.choices[0].message.content

# 测试调用
if __name__ == "__main__":
    messages = [
        {"role": "system", "content": "你是一个专业的Python开发助手"},
        {"role": "user", "content": "用一行代码实现列表去重并保持顺序"},
    ]
    reply = chat_with_gpt(messages, model="gpt-5.4-mini")
    print(f"GPT 回复: {reply}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

# anthropic_call.py —— Claude 4.x 调用示例
from dotenv import load_dotenv
from anthropic import Anthropic

load_dotenv()

# 自动读取 ANTHROPIC_API_KEY 环境变量
client = Anthropic()

def chat_with_claude(
    user_message: str,
    system_prompt: str = "你是一个专业的开发助手",
    model: str = "claude-sonnet-4-20250514",
    max_tokens: int = 4096,
) -> str:
    """调用 Claude 模型进行对话
    
    参数:
        user_message: 用户消息
        system_prompt: 系统提示词（Claude 单独传参，不放 messages 里）
        model: 模型名称
        max_tokens: 最大输出长度（Claude 必填）
    返回:
        模型回复的文本内容
    """
    response = client.messages.create(
        model=model,
        max_tokens=max_tokens,
        system=system_prompt,  # 注意：system 是顶层参数，不是 messages 里的一条
        messages=[{"role": "user", "content": user_message}],
    )
    # Claude 的返回结构是 content 列表，取第一块文本
    return response.content[0].text

# 测试调用
if __name__ == "__main__":
    reply = chat_with_claude(
        user_message="解释Python的GIL是什么，如何绕过它",
        model="claude-sonnet-4-20250514",
    )
    print(f"Claude 回复: {reply}")

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189

# llm_router.py —— 统一多模型调用封装
import os
from dataclasses import dataclass
from dotenv import load_dotenv
from openai import OpenAI
from anthropic import Anthropic

load_dotenv()


@dataclass
class LLMResponse:
    """统一的模型响应结构"""
    content: str          # 回复文本
    model: str            # 实际使用的模型
    input_tokens: int     # 输入token数
    output_tokens: int    # 输出token数
    cost_usd: float       # 本次调用估算成本


# 各模型的定价表（$/M tokens）：(输入单价, 输出单价)
PRICING = {
    "gpt-5.5":          (5.00, 30.00),
    "gpt-5.4":          (2.50, 15.00),
    "gpt-5.4-mini":     (0.75, 4.50),
    "gpt-5.4-nano":     (0.15, 0.90),
    "claude-opus-4":    (15.00, 75.00),
    "claude-sonnet-4":  (3.00, 15.00),
    "claude-haiku-4":   (0.80, 4.00),
    "qwen3:8b":         (0.0, 0.0),   # 本地部署，成本为0
}


def _normalize_model_name(model: str) -> str:
    """规范模型名，便于查价。
    
    Anthropic 的模型名带日期后缀（如 claude-sonnet-4-20250514），
    但 PRICING 表里只存了 claude-sonnet-4，所以查价前要去掉后缀。
    """
    if "-20" in model:  # 形如 -20250514 的日期后缀
        return model.rsplit("-20", 1)[0]
    return model


class UnifiedLLM:
    """统一大模型调用接口，支持 OpenAI / Anthropic / Ollama"""

    def __init__(self):
        self.openai_client = OpenAI()
        self.anthropic_client = Anthropic()
        # Ollama 复用 OpenAI SDK，只是把地址指向本地服务
        self.ollama_client = OpenAI(
            base_url=os.getenv("OLLAMA_BASE_URL", "http://localhost:11434") + "/v1",
            api_key="ollama",  # Ollama 不校验密钥，随便填
        )

    def chat(
        self,
        messages: list[dict],
        model: str = "gpt-5.4-mini",
        system: str | None = None,
        temperature: float = 0.7,
        max_tokens: int = 4096,
    ) -> LLMResponse:
        """统一对话接口
        
        参数:
            messages: 消息列表（不含system）
            model: 模型名称，自动路由到对应SDK
            system: 系统提示词（可选）
            temperature: 采样温度
            max_tokens: 最大输出token数
        返回:
            LLMResponse 对象
        """
        if model.startswith("gpt-"):
            return self._call_openai(messages, model, system, temperature, max_tokens)
        elif model.startswith("claude-"):
            return self._call_anthropic(messages, model, system, temperature, max_tokens)
        else:
            # 默认走 Ollama（本地开源模型）
            return self._call_ollama(messages, model, system, temperature, max_tokens)

    def _call_openai(self, messages, model, system, temperature, max_tokens) -> LLMResponse:
        """调用 OpenAI 系列"""
        full_messages = []
        if system:
            full_messages.append({"role": "system", "content": system})
        full_messages.extend(messages)

        resp = self.openai_client.chat.completions.create(
            model=model,
            messages=full_messages,
            temperature=temperature,
            max_tokens=max_tokens,
        )
        usage = resp.usage
        cost = self._calc_cost(model, usage.prompt_tokens, usage.completion_tokens)
        return LLMResponse(
            content=resp.choices[0].message.content,
            model=model,
            input_tokens=usage.prompt_tokens,
            output_tokens=usage.completion_tokens,
            cost_usd=cost,
        )

    def _call_anthropic(self, messages, model, system, temperature, max_tokens) -> LLMResponse:
        """调用 Claude 系列"""
        resp = self.anthropic_client.messages.create(
            model=model,
            max_tokens=max_tokens,
            temperature=temperature,
            system=system or "",  # Claude 的 system 是顶层参数
            messages=messages,
        )
        usage = resp.usage
        # 注意：Claude 的 token 字段叫 input_tokens/output_tokens
        cost = self._calc_cost(model, usage.input_tokens, usage.output_tokens)
        return LLMResponse(
            content=resp.content[0].text,
            model=model,
            input_tokens=usage.input_tokens,
            output_tokens=usage.output_tokens,
            cost_usd=cost,
        )

    def _call_ollama(self, messages, model, system, temperature, max_tokens) -> LLMResponse:
        """调用本地 Ollama 服务（OpenAI兼容格式）"""
        full_messages = []
        if system:
            full_messages.append({"role": "system", "content": system})
        full_messages.extend(messages)

        resp = self.ollama_client.chat.completions.create(
            model=model,
            messages=full_messages,
            temperature=temperature,
            max_tokens=max_tokens,
        )
        usage = resp.usage
        return LLMResponse(
            content=resp.choices[0].message.content,
            model=model,
            input_tokens=usage.prompt_tokens if usage else 0,
            output_tokens=usage.completion_tokens if usage else 0,
            cost_usd=0.0,  # 本地部署无API成本
        )

    @staticmethod
    def _calc_cost(model: str, input_tok: int, output_tok: int) -> float:
        """根据定价表计算调用成本（美元）"""
        normalized = _normalize_model_name(model)  # 去掉日期后缀再查价
        pricing = PRICING.get(normalized, (0, 0))
        return (input_tok / 1_000_000 * pricing[0]) + (output_tok / 1_000_000 * pricing[1])


# 使用示例
if __name__ == "__main__":
    llm = UnifiedLLM()

    # 用同一个接口调用不同模型——上层代码完全一样，只改 model 字符串
    question = "用Python实现快速排序，并解释时间复杂度"

    # 调用 GPT
    r1 = llm.chat(
        [{"role": "user", "content": question}],
        model="gpt-5.4-mini",
        system="你是一个资深算法工程师",
    )
    print(f"[GPT-5.4-mini] 成本: ${r1.cost_usd:.4f}")
    print(f"回复: {r1.content[:100]}...\n")

    # 调用 Claude
    r2 = llm.chat(
        [{"role": "user", "content": question}],
        model="claude-sonnet-4-20250514",
        system="你是一个资深算法工程师",
    )
    print(f"[Claude Sonnet 4] 成本: ${r2.cost_usd:.4f}")
    print(f"回复: {r2.content[:100]}...\n")

    # 调用本地 Qwen
    r3 = llm.chat(
        [{"role": "user", "content": question}],
        model="qwen3:8b",
        system="你是一个资深算法工程师",
    )
    print(f"[Qwen3-8B 本地] 成本: ${r3.cost_usd:.4f}")
    print(f"回复: {r3.content[:100]}...")

1

GPT-5.5 → GPT-5.4 → GPT-5.4-mini → Qwen3（本地）→ 返回缓存/默认回复

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132

# smart_router.py —— 智能路由与降级
import time
import logging
from llm_router import UnifiedLLM, LLMResponse

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 任务复杂度到模型的映射
COMPLEXITY_MODEL_MAP = {
    "simple":   "gpt-5.4-nano",        # 简单任务
    "standard": "gpt-5.4-mini",        # 标准任务
    "complex":  "gpt-5.4",             # 复杂任务
    "extreme":  "gpt-5.5",             # 极限任务
}

# 每个主模型的降级链：主模型挂了就依次试这些备选
FALLBACK_CHAINS = {
    "gpt-5.5":          ["gpt-5.4", "gpt-5.4-mini", "qwen3:8b"],
    "gpt-5.4":          ["gpt-5.4-mini", "qwen3:8b"],
    "gpt-5.4-mini":     ["gpt-5.4-nano", "qwen3:8b"],
    "gpt-5.4-nano":     ["qwen3:8b"],
    "claude-sonnet-4-20250514": ["gpt-5.4-mini", "qwen3:8b"],
}

# 关键词 → 复杂度等级的简单规则路由
COMPLEXITY_KEYWORDS = {
    "extreme":  ["架构设计", "系统设计", "重构整个", "安全审计", "性能优化方案"],
    "complex":  ["编写代码", "调试", "推理", "分析报告", "多步骤"],
    "standard": ["翻译", "总结", "摘要", "问答", "分类"],
    "simple":   ["你好", "谢谢", "再见", "是或否"],
}


class SmartRouter:
    """智能模型路由器：自动选择模型 + 故障降级"""

    def __init__(self):
        self.llm = UnifiedLLM()
        self.total_cost = 0.0   # 累计成本（美元）
        self.call_count = 0     # 累计调用次数

    def classify_complexity(self, user_input: str) -> str:
        """根据关键词判断任务复杂度（规则路由）
        
        在生产中可以用一个小模型替代关键词匹配，
        让模型自己判断复杂度，效果更好但会增加一次调用。
        """
        for level, keywords in COMPLEXITY_KEYWORDS.items():
            if any(kw in user_input for kw in keywords):
                return level
        return "standard"  # 默认标准

    def route(self, user_input: str, system: str = "") -> LLMResponse:
        """智能路由主入口
        
        流程：1.判断复杂度 → 2.选模型 → 3.调用 → 4.失败则降级
        """
        # 第一步：判断复杂度
        complexity = self.classify_complexity(user_input)
        primary_model = COMPLEXITY_MODEL_MAP[complexity]
        logger.info(f"任务复杂度: {complexity} → 主模型: {primary_model}")

        # 第二步：构建降级链（主模型 + 备选列表）
        fallback_models = FALLBACK_CHAINS.get(primary_model, ["qwen3:8b"])
        models_to_try = [primary_model] + fallback_models

        # 第三步：依次尝试调用，成功即返回
        messages = [{"role": "user", "content": user_input}]
        last_error = None

        for model in models_to_try:
            try:
                resp = self.llm.chat(
                    messages=messages,
                    model=model,
                    system=system,
                )
                self.total_cost += resp.cost_usd
                self.call_count += 1  # 记录一次成功调用
                logger.info(
                    f"✅ 调用成功 [{model}] "
                    f"成本=${resp.cost_usd:.4f} 累计=${self.total_cost:.4f}"
                )
                return resp
            except Exception as e:
                logger.warning(f"❌ [{model}] 调用失败: {e}，尝试降级...")
                last_error = e
                time.sleep(0.5)  # 降级前短暂等待，给上游喘口气

        # 所有模型都失败，返回兜底回复
        logger.error(f"所有模型均不可用，最后错误: {last_error}")
        return LLMResponse(
            content="抱歉，服务暂时不可用，请稍后重试。",
            model="fallback",
            input_tokens=0,
            output_tokens=0,
            cost_usd=0.0,
        )

    def get_cost_report(self) -> dict:
        """获取成本报告"""
        return {
            "total_cost_usd": round(self.total_cost, 4),
            "avg_cost_per_call": round(
                self.total_cost / max(self.call_count, 1), 4
            ),
        }


# 使用示例
if __name__ == "__main__":
    router = SmartRouter()

    test_cases = [
        "你好，今天天气怎么样？",                    # → simple → nano
        "请帮我总结这篇文章的要点",                  # → standard → mini
        "帮我编写一个Python并发爬虫，要求支持重试",  # → complex → 5.4
        "请设计一个支持百万并发的微服务架构方案",    # → extreme → 5.5
    ]

    for query in test_cases:
        print(f"\n{'='*60}")
        print(f"用户输入: {query}")
        resp = router.route(query, system="你是一个专业的技术助手")
        print(f"使用模型: {resp.model}")
        print(f"本次成本: ${resp.cost_usd:.4f}")
        print(f"回复预览: {resp.content[:80]}...")

    print(f"\n{'='*60}")
    print(f"总成本: ${router.total_cost:.4f}")
    print(f"成本报告: {router.get_cost_report()}")