贵州网站公司百度关键词排名联系
环境:
qwen
DeepSeek
问题描述:
如何将本地模型服务不可用,自动切换调用线上模型?
解决方案:
将本地模型调用部分封装到 qw.py
,线上模型调用部分封装到 zx.py
,并且在主程序 zj.py
中优先调用本地模型,若本地模型不可用则自动切换调用线上模型。
下面方案:
1. qw.py
— 本地模型调用模块
import requests
from loguru import logger
import time
from typing import OptionalLOCAL_MODEL_API = "http://192.168.1.9:11434/v1/chat/completions"def summarize_with_local_model(text: str) -> Optional[str]:headers = {"Content-Type": "application/json"}payload = {"model": "qwen2.5-3bnsfwny","messages": [{"role": "user", "content": text}],"max_tokens": 512,"temperature": 0.25,"stream": False}try:logger.info("调用本地大模型接口,准备发送请求")start_time = time.time()r = requests.post(LOCAL_MODEL_API, headers=headers, json=payload, timeout=10)r.raise_for_status()logger.info(f"接口响应状态码:{r.status_code},耗时:{time.time() - start_time:.2f}秒")resp_json = r.json()content = resp_json.get("choices", [{}])[0].get("message", {}).get("content")logger.info(f"模型返回内容长度:{len(content) if content else 0}")logger.info(f"模型返回具体内容:{content}")return contentexcept requests.exceptions.Timeout:logger.error("调用本地大模型失败: 请求超时")return Noneexcept Exception as e:logger.error(f"调用本地大模型失败: {e}")return None
2. zx.py
— 线上模型调用模块(示例)
假设线上 DeepSeek 模型接口类似,您需要根据实际接口调整:
import os
from openai import OpenAI
from loguru import logger
from typing import Optionalclient = OpenAI(api_key="your_real_api_key",base_url="https://ark.cn-beijing.volces.com/api/v3",
)def summarize_with_online_model(text: str) -> Optional[str]:"""调用线上DeepSeek的chat completion接口,非流式调用,返回模型文本内容,异常时返回None。"""try:logger.info("调用线上DeepSeek模型接口,准备发送请求")# 构造消息列表,您可以根据实际需求调整system prompt等messages = [{"role": "system", "content": "你是人工智能助手"},{"role": "user", "content": text},]completion = client.chat.completions.create(model="ep-20250602174215-fw5hx",messages=messages,stream=False)content = completion.choices[0].message.contentlogger.info(f"线上模型返回内容长度:{len(content) if content else 0}")logger.info(f"线上模型返回内容:{content}")return contentexcept Exception as e:logger.error(f"调用线上模型失败: {e}")return None
3. zj.py
— 主逻辑,自动切换调用
from loguru import logger
from typing import Optionalimport qw # 本地模型
import zx # 线上模型def summarize(text: str) -> Optional[str]:# 优先调用本地模型logger.info("优先尝试调用本地模型")result = qw.summarize_with_local_model(text)if result:logger.info("本地模型调用成功,返回结果")return resultelse:logger.warning("本地模型不可用,切换调用线上模型")result_online = zx.summarize_with_online_model(text)if result_online:logger.info("线上模型调用成功,返回结果")return result_onlineelse:logger.error("线上模型调用也失败,无法获取结果")return None# 下面是使用示例
if __name__ == "__main__":test_text = "请告诉我今天北京的天气。"answer = summarize(test_text)if answer:print("模型回答:", answer)else:print("未能获取模型回答,请稍后重试。")
总结
- 本地模型调用代码放
qw.py
里,包含请求和异常处理; - 线上模型调用代码放
zx.py
,类似结构,替换接口地址和认证信息; - 主程序
zj.py
负责调用本地模型,失败时自动切换调用线上模型; - 方便维护,且切换逻辑集中管理。