GithubBackup/BiliNote
1
0
Fork 0
mirror of https://github.com/JefferyHcool/BiliNote.git synced 2026-06-02 19:59:37 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: 添加硅基流动(SiliconFlow)支持和错误处理优化

Browse Source 
## 主要更新

### 新增功能
- 新增 SiliconFlow_provider.py 专用提供商
- 添加硅基流动 API 集成文档
- 实现 Cherry Studio 风格的连接测试

### 错误处理优化
- 修复前端 Form.tsx 错误显示问题
- 改进 universal_gpt.py 异常处理逻辑
- 统一 URL 格式处理，避免路径重复

### 兼容性改进
- 优化 OpenAI 兼容提供商 URL 处理
- 增强模型列表获取的容错性
- 添加详细的调试日志

### 安全性提升
- 更新 .gitignore 保护敏感信息
- 移除示例配置文件

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
yangyuguang
2025-07-13 09:53:39 +08:00
parent 880f745718
commit ab8cdc416a
15 changed files with 4962 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
backend/app/gpt/gpt_factory.py
View File

@@ -2,6 +2,7 @@ from openai import OpenAI
from app.gpt.base import GPT
from app.gpt.provider.OpenAI_compatible_provider import OpenAICompatibleProvider
from app.gpt.provider.SiliconFlow_provider import SiliconFlowProvider
from app.gpt.universal_gpt import UniversalGPT
from app.models.model_config import ModelConfig
@@ -9,5 +10,18 @@ from app.models.model_config import ModelConfig
class GPTFactory:
@staticmethod
def from_config(config: ModelConfig) -> GPT:
client = OpenAICompatibleProvider(api_key=config.api_key, base_url=config.base_url).get_client
# 检查是否是硅基流动，使用专门的提供商类
if "siliconflow" in config.base_url.lower():
client = SiliconFlowProvider(
api_key=config.api_key,
base_url=config.base_url,
model=config.model_name
).get_client
else:
# 其他提供商使用通用兼容类
client = OpenAICompatibleProvider(
api_key=config.api_key,
base_url=config.base_url
).get_client
return UniversalGPT(client=client, model=config.model_name)

152
backend/app/gpt/provider/OpenAI_compatible_provider.py
View File

@@ -5,6 +5,7 @@ from openai import OpenAI
from app.utils.logger import get_logger
logging= get_logger(__name__)
class OpenAICompatibleProvider:
def __init__(self, api_key: str, base_url: str, model: Union[str, None]=None):
self.client = OpenAI(api_key=api_key, base_url=base_url)
@@ -17,15 +18,148 @@ class OpenAICompatibleProvider:
@staticmethod
def test_connection(api_key: str, base_url: str) -> bool:
try:
client = OpenAI(api_key=api_key, base_url=base_url)
model = client.models.list()
# for segment in model:
# print(segment)
# print(model)
logging.info("连通性测试成功")
return True
# 调试：打印API Key的实际长度和内容
logging.info(f"正在测试连接 - API Key长度: {len(api_key)}, 前8位: {api_key[:8]}, 后4位: {api_key[-4:] if len(api_key) > 4 else 'TOO_SHORT'}")
logging.info(f"Base URL: {base_url}")
# 硅基流动特殊处理：参考Cherry Studio的实现方式
if "siliconflow" in base_url.lower():
logging.info("检测到硅基流动，参考Cherry Studio实现方式")
# 标准化URL处理，避免路径重复
base_url_clean = base_url.rstrip('/')
if base_url_clean.endswith("/v1"):
# 如果用户输入了/v1，直接使用
api_base = base_url_clean
test_url = f"{api_base}/chat/completions"
elif base_url_clean.endswith("/chat/completions"):
# 如果用户直接输入了完整路径，直接使用
test_url = base_url_clean
api_base = base_url_clean.replace("/chat/completions", "")
else:
# Cherry Studio方式：不加/v1后缀，直接加/chat/completions
api_base = base_url_clean
test_url = f"{base_url_clean}/chat/completions"
logging.info(f"使用API基地址: {api_base}")
logging.info(f"测试URL: {test_url}")
# 先用requests验证（Cherry Studio方式）
import requests
import json
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "Qwen/Qwen2.5-7B-Instruct",
"messages": [{"role": "user", "content": "hi"}],
"max_tokens": 1
}
try:
logging.info("Cherry Studio方式：直接HTTP请求测试")
response = requests.post(test_url, headers=headers, json=payload, timeout=15)
logging.info(f"HTTP响应状态码: {response.status_code}")
if response.status_code == 200:
logging.info("硅基流动连接测试成功（Cherry Studio方式）")
result = response.json()
logging.info(f"响应: {json.dumps(result, ensure_ascii=False)[:100]}...")
return True
else:
logging.error(f"HTTP请求失败: {response.status_code} - {response.text}")
# 尝试不同的端点
if response.status_code == 404 and "/v1" in test_url:
# 尝试去掉/v1
alt_url = test_url.replace("/v1", "")
logging.info(f"尝试备用URL: {alt_url}")
alt_response = requests.post(alt_url, headers=headers, json=payload, timeout=15)
if alt_response.status_code == 200:
logging.info("硅基流动连接测试成功（备用URL）")
return True
except Exception as http_error:
logging.error(f"直接HTTP请求异常: {http_error}")
# 标准OpenAI SDK方式作为备选
# 对于硅基流动，需要使用正确的base_url
if "siliconflow" in base_url.lower():
# 确保SDK使用正确的base_url（需要包含/v1）
sdk_base_url = api_base if api_base.endswith('/v1') else f"{api_base}/v1"
client = OpenAI(api_key=api_key, base_url=sdk_base_url)
logging.info(f"尝试OpenAI SDK方式，使用base_url: {sdk_base_url}")
else:
client = OpenAI(api_key=api_key, base_url=base_url)
if "siliconflow" in base_url.lower():
# 硅基流动的免费模型列表
test_models = [
"Qwen/Qwen2.5-7B-Instruct",
"THUDM/glm-4-9b-chat",
"deepseek-ai/DeepSeek-V3"
]
for model in test_models:
try:
logging.info(f"尝试测试模型: {model}")
response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": "hi"}],
max_tokens=1,
timeout=15.0
)
logging.info(f"硅基流动连接测试成功，使用模型: {model}")
return True
except Exception as model_error:
error_msg = str(model_error)
logging.warning(f"模型 {model} 测试失败: {error_msg}")
if "401" in error_msg or "Unauthorized" in error_msg or "Api key is invalid" in error_msg:
raise Exception("API Key 无效或已过期，请检查API Key是否正确")
continue
# 尝试models接口
try:
models = client.models.list()
logging.info("硅基流动连接测试成功（通过models接口）")
return True
except Exception as models_error:
logging.error(f"models接口失败: {models_error}")
raise models_error
else:
# 非硅基流动提供商
model = client.models.list()
logging.info("连通性测试成功")
return True
except Exception as e:
logging.info(f"连通性测试失败：{e}")
error_msg = str(e)
logging.error(f"连通性测试失败：{error_msg}")
# 根据错误类型提供更具体的错误信息
if "401" in error_msg or "Unauthorized" in error_msg or "Api key is invalid" in error_msg:
raise Exception("API Key 无效或已过期，请检查API Key是否正确")
elif "404" in error_msg or "Not Found" in error_msg:
if "siliconflow" in base_url.lower():
raise Exception("API 地址可能不正确。建议尝试: https://api.siliconflow.cn/v1 或 https://api.siliconflow.cn（参考Cherry Studio配置）")
else:
raise Exception("API 地址不正确，请检查 base_url 格式")
elif "timeout" in error_msg.lower():
raise Exception("连接超时，请检查网络连接或 API 地址是否正确")
elif "ssl" in error_msg.lower() or "certificate" in error_msg.lower():
raise Exception("SSL 证书验证失败，请检查 API 地址是否使用 HTTPS")
elif "connection" in error_msg.lower():
if "siliconflow" in base_url.lower():
raise Exception("无法连接到硅基流动服务器，请尝试: https://api.siliconflow.cn/v1 或 https://api.siliconflow.cn")
else:
raise Exception("无法连接到服务器，请检查 API 地址和网络连接")
elif "_set_private_attributes" in error_msg:
raise Exception("OpenAI SDK版本兼容性问题，请尝试重新配置或联系管理员")
else:
raise Exception(f"连接失败（原始错误）: {error_msg}")
# print(f"Error connecting to OpenAI API: {e}")
return False

214
backend/app/gpt/provider/SiliconFlow_provider.py Normal file
View File

@@ -0,0 +1,214 @@
from typing import Optional, Union, List
from openai import OpenAI
from app.utils.logger import get_logger
logger = get_logger(__name__)
class SiliconFlowProvider:
"""
专门为硅基流动(SiliconFlow)优化的提供商类
基于市面上成熟的接入方案设计
"""
# 硅基流动支持的常用模型列表
SUPPORTED_MODELS = [
"deepseek-ai/DeepSeek-V3",
"deepseek-ai/DeepSeek-R1",
"Qwen/Qwen2.5-72B-Instruct",
"Qwen/Qwen2.5-32B-Instruct",
"Qwen/Qwen2.5-14B-Instruct",
"Qwen/Qwen2.5-7B-Instruct",
"meta-llama/Llama-3.1-70B-Instruct",
"meta-llama/Llama-3.1-8B-Instruct",
"THUDM/glm-4-9b-chat",
"01-ai/Yi-1.5-34B-Chat"
]
# 硅基流动API端点
API_ENDPOINTS = [
"https://api.siliconflow.cn/v1", # 国内用户
"https://api-st.siliconflow.cn/v1" # 海外用户
]
def __init__(self, api_key: str, base_url: str = None, model: Union[str, None] = None):
"""
初始化硅基流动提供商
Args:
api_key: API密钥
base_url: API基础URL，默认使用国内端点
model: 模型名称
"""
self.api_key = api_key
# 标准化base_url，确保符合硅基流动API要求
if base_url:
base_url_clean = base_url.rstrip('/')
# 确保使用正确的API端点格式
if not base_url_clean.endswith('/v1'):
if base_url_clean.endswith('/chat/completions'):
# 用户输入了完整的endpoint，提取base部分并添加/v1
base_url_clean = base_url_clean.replace('/chat/completions', '/v1')
elif 'siliconflow' in base_url_clean.lower():
# 硅基流动需要/v1后缀
base_url_clean = f"{base_url_clean}/v1"
self.base_url = base_url_clean
else:
self.base_url = self.API_ENDPOINTS[0]
self.model = model
logger.info(f"硅基流动提供商初始化 - 使用base_url: {self.base_url}")
self.client = OpenAI(api_key=api_key, base_url=self.base_url)
@property
def get_client(self):
return self.client
@classmethod
def test_connection(cls, api_key: str, base_url: str = None) -> bool:
"""
测试硅基流动连接
使用成熟的chat接口测试方法，而非models接口
Args:
api_key: API密钥
base_url: API基础URL
Returns:
bool: 连接是否成功
"""
base_url = base_url or cls.API_ENDPOINTS[0]
try:
logger.info(f"测试硅基流动连接 - API Key: {api_key[:8]}...（已截断） Base URL: {base_url}")
client = OpenAI(api_key=api_key, base_url=base_url)
# 使用轻量级模型进行连接测试
test_models = [
"Qwen/Qwen2.5-7B-Instruct", # 免费模型优先
"deepseek-ai/DeepSeek-V3",
"THUDM/glm-4-9b-chat"
]
for model in test_models:
try:
logger.info(f"尝试测试模型: {model}")
# 发送简单的chat请求测试连接
response = client.chat.completions.create(
model=model,
messages=[
{"role": "user", "content": "hi"}
],
max_tokens=1,
timeout=15.0
)
logger.info(f"硅基流动连接测试成功 - 模型: {model}")
return True
except Exception as model_error:
error_msg = str(model_error)
logger.warning(f"模型 {model} 测试失败: {error_msg}")
# 如果是401错误（API Key问题），不继续尝试其他模型
if "401" in error_msg or "Unauthorized" in error_msg or "Api key is invalid" in error_msg:
raise Exception("API Key 无效或已过期，请检查API Key是否正确")
continue
# 如果所有模型都失败，尝试models接口作为最后手段
logger.info("所有模型测试失败，尝试models接口")
try:
models = client.models.list()
logger.info("硅基流动连接测试成功（通过models接口）")
return True
except Exception as models_error:
logger.error(f"models接口也失败: {models_error}")
raise models_error
except Exception as e:
error_msg = str(e)
logger.error(f"硅基流动连接测试失败：{error_msg}")
# 根据错误类型提供具体的错误信息
if "401" in error_msg or "Unauthorized" in error_msg or "Api key is invalid" in error_msg:
raise Exception("API Key 无效或已过期，请检查API Key是否正确")
elif "404" in error_msg or "Not Found" in error_msg:
raise Exception(f"API地址不正确，请检查URL格式。推荐使用: {cls.API_ENDPOINTS[0]} 或 {cls.API_ENDPOINTS[1]}")
elif "timeout" in error_msg.lower():
raise Exception("连接超时，请检查网络连接或尝试海外端点")
elif "connection" in error_msg.lower():
raise Exception(f"无法连接到硅基流动服务器，请尝试: {cls.API_ENDPOINTS[1]}")
else:
raise Exception(f"连接失败: {error_msg}")
def list_models(self):
"""
获取可用模型列表
优先返回预定义的模型列表，如果API支持则获取实时列表
"""
try:
# 尝试获取实时模型列表
models = self.client.models.list()
logger.info("成功获取硅基流动实时模型列表")
return models
except Exception as e:
logger.warning(f"无法获取实时模型列表，返回预定义列表: {e}")
# 返回预定义的模型列表
from types import SimpleNamespace
model_objects = []
for model_name in self.SUPPORTED_MODELS:
model_obj = SimpleNamespace()
model_obj.id = model_name
model_obj.object = "model"
model_obj.created = 1640995200 # 固定时间戳
model_obj.owned_by = "siliconflow"
# 添加dict方法
def dict_method():
return {
"id": model_name,
"object": "model",
"created": 1640995200,
"owned_by": "siliconflow"
}
model_obj.dict = dict_method
model_objects.append(model_obj)
# 构造兼容的返回对象
result = SimpleNamespace()
result.data = model_objects
return result
def create_chat_completion(self, model: str, messages: list, **kwargs):
"""
创建聊天完成请求
"""
return self.client.chat.completions.create(
model=model,
messages=messages,
**kwargs
)
@classmethod
def get_recommended_config(cls) -> dict:
"""
获取推荐的硅基流动配置
"""
return {
"name": "硅基流动",
"type": "custom",
"base_url": cls.API_ENDPOINTS[0],
"logo": "SiliconFlow",
"supported_models": cls.SUPPORTED_MODELS,
"description": "硅基流动 - 免费高性能AI模型服务",
"features": [
"完全兼容OpenAI API",
"支持多种开源大模型",
"部分模型永久免费",
"国内外双端点支持"
]
}

192
backend/app/gpt/universal_gpt.py
View File

@@ -6,7 +6,14 @@ from app.gpt.utils import fix_markdown
from app.models.transcriber_model import TranscriptSegment
from datetime import timedelta
from typing import List
import json
import logging
import math
logger = logging.getLogger(__name__)
# 设置分段处理的参数
MAX_SEGMENTS_PER_CHUNK = 250 # 每块最多包含的段落数
MAX_CONTENT_LENGTH = 30000 # 字符数
class UniversalGPT(GPT):
def __init__(self, client, model: str, temperature: float = 0.7):
@@ -29,7 +36,6 @@ class UniversalGPT(GPT):
return [TranscriptSegment(**seg) if isinstance(seg, dict) else seg for seg in segments]
def create_messages(self, segments: List[TranscriptSegment], **kwargs):
content_text = generate_base_prompt(
title=kwargs.get('title'),
segment_text=self._build_segment_text(segments),
@@ -38,11 +44,30 @@ class UniversalGPT(GPT):
style=kwargs.get('style'),
extras=kwargs.get('extras'),
)
# ⛳ 组装 content 数组，支持 text + image_url 混合
# 检查文本长度
if len(content_text) > MAX_CONTENT_LENGTH:
# 保留前部分和后部分内容
first_part = int(MAX_CONTENT_LENGTH * 0.3)
second_part = MAX_CONTENT_LENGTH - first_part - 100 # 预留100字符给提示文本
truncated_text = (
content_text[:first_part] +
"\n\n[内容过长，中间部分已省略]\n\n" +
content_text[-second_part:]
)
content_text = truncated_text
print(f"内容已截断，原长度: {len(content_text)}，截断后: {len(truncated_text)}")
# 组装 content 数组，支持 text + image_url 混合
content = [{"type": "text", "text": content_text}]
video_img_urls = kwargs.get('video_img_urls', [])
# 限制图片数量
if len(video_img_urls) > 5:
video_img_urls = video_img_urls[:5]
print("图片数量过多，已限制为5张")
for url in video_img_urls:
content.append({
"type": "image_url",
@@ -52,7 +77,7 @@ class UniversalGPT(GPT):
}
})
# 正确格式：整体包在一个 message 里，role + content array
# 正确格式：整体包在一个 message 里，role + content array
messages = [{
"role": "user",
"content": content
@@ -68,18 +93,159 @@ class UniversalGPT(GPT):
self.link = source.link
source.segment = self.ensure_segments_type(source.segment)
messages = self.create_messages(
source.segment,
# 如果段落数量超过阈值，使用分段处理方法
if len(source.segment) > MAX_SEGMENTS_PER_CHUNK:
print(f"段落过多({len(source.segment)}个)，将使用分段处理方法")
return self._process_long_content_by_chunks(source)
# 正常处理较短的内容
try:
messages = self.create_messages(
source.segment,
title=source.title,
tags=source.tags,
video_img_urls=source.video_img_urls,
_format=source._format,
style=source.style,
extras=source.extras
)
# 检查消息大小
messages_json = json.dumps(messages)
if len(messages_json) > 100000: # API限制
print(f"消息体积过大: {len(messages_json)} 字节，将使用分段处理方法")
return self._process_long_content_by_chunks(source)
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=0.7
)
return response.choices[0].message.content.strip()
except Exception as e:
error_msg = f"总结失败: {str(e)}"
print(error_msg)
# 记录详细错误信息到日志
import logging
logging.error(f"GPT单块处理失败 - {error_msg}")
# 如果处理失败，尝试使用分段处理方法
print("尝试使用分段处理方法")
try:
return self._process_long_content_by_chunks(source)
except Exception as fallback_error:
# 如果分段处理也失败，抛出异常
logging.error(f"GPT分段处理也失败 - {str(fallback_error)}")
raise Exception(f"视频处理完全失败：主处理失败({str(e)})，分段处理也失败({str(fallback_error)})")
def _process_long_content_by_chunks(self, source: GPTSource) -> str:
"""
将长内容分成多个块，分别处理后再整合
"""
segments = source.segment
total_segments = len(segments)
# 计算需要多少块
num_chunks = math.ceil(total_segments / MAX_SEGMENTS_PER_CHUNK)
chunk_size = math.ceil(total_segments / num_chunks)
print(f"将内容分为{num_chunks}块进行处理，每块约{chunk_size}个段落")
chunk_summaries = []
# 处理每个块
for i in range(num_chunks):
start_idx = i * chunk_size
end_idx = min(start_idx + chunk_size, total_segments)
print(f"处理第{i+1}/{num_chunks}块 (段落 {start_idx} 到 {end_idx-1})")
# 创建此块的子源
chunk_source = GPTSource(
title=f"{source.title} - 第{i+1}/{num_chunks}部分",
segment=segments[start_idx:end_idx],
tags=source.tags,
screenshot=False, # 中间块不需要截图
video_img_urls=[], # 中间块不需要图片
link=False, # 中间块不需要链接
_format=[],
style=source.style,
extras=f"这是内容的第{i+1}部分，共{num_chunks}部分。请仅总结这部分内容的要点，无需引言和结论。"
)
try:
# 处理这个块
chunk_messages = self.create_messages(
chunk_source.segment,
title=chunk_source.title,
tags=chunk_source.tags,
video_img_urls=chunk_source.video_img_urls,
_format=chunk_source._format,
style=chunk_source.style,
extras=chunk_source.extras
)
chunk_response = self.client.chat.completions.create(
model=self.model,
messages=chunk_messages,
temperature=0.7
)
chunk_summary = chunk_response.choices[0].message.content.strip()
chunk_summaries.append(f"### 第{i+1}部分内容总结\n\n{chunk_summary}")
print(f"第{i+1}块处理完成")
except Exception as e:
error_msg = f"处理第{i+1}块时出错: {str(e)}"
print(error_msg)
# 记录详细错误信息到日志
import logging
logging.error(f"GPT处理失败 - {error_msg}")
# 如果某块处理失败，抛出异常停止整个处理流程
raise Exception(f"视频处理失败：第{i+1}块GPT调用失败 - {str(e)}")
# 合并所有块的总结
all_summaries = "\n\n".join(chunk_summaries)
# 创建最终总结请求
final_segment = TranscriptSegment(start=0, end=0, text=all_summaries)
final_source = GPTSource(
title=source.title,
segment=[final_segment],
tags=source.tags,
screenshot=source.screenshot,
video_img_urls=source.video_img_urls,
link=source.link,
_format=source._format,
style=source.style,
extras=source.extras
extras="以下是视频各部分的总结，请将它们整合为一篇完整、连贯的笔记。"
)
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=0.7
)
return response.choices[0].message.content.strip()
try:
# 最终合并处理
final_messages = self.create_messages(
final_source.segment,
title=final_source.title,
tags=final_source.tags,
video_img_urls=final_source.video_img_urls,
_format=final_source._format,
style=final_source.style,
extras=final_source.extras
)
final_response = self.client.chat.completions.create(
model=self.model,
messages=final_messages,
temperature=0.7
)
return final_response.choices[0].message.content.strip()
except Exception as e:
error_msg = f"最终合并处理时出错: {str(e)}"
print(error_msg)
# 记录详细错误信息到日志
import logging
logging.error(f"GPT最终合并失败 - {error_msg}")
# 如果最终合并失败，抛出异常
raise Exception(f"视频处理失败：GPT最终合并失败 - {str(e)}")