python/ai-business-write
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

创建新分支，用于静态交通本地服务器部署，调整默认为华为大模型调用

Browse Source
This commit is contained in:
python 2025-12-09 10:33:41 +08:00
parent 5830de355d
commit 7d50b160c2
4 changed files with 137 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
README.md
View File

@ -6,8 +6,7 @@
- ✅ AI解析接口 (`/api/ai/extract`) - 从输入文本中提取结构化字段
- ✅ 字段配置管理 - 从数据库读取字段配置
- ✅ 支持硅基流动大模型（DeepSeek）
- 🔄 预留华为大模型接口支持
- ✅ 支持华为大模型（DeepSeek-R1-Distill-Llama-70B）
- ✅ Web测试界面 - 可视化测试解析功能
## 项目结构
@ -73,9 +72,10 @@ cp .env.example .env
编辑 `.env` 文件，填入你的API密钥：
```env
# 硅基流动API配置（必需）
SILICONFLOW_API_KEY=your_api_key_here
SILICONFLOW_MODEL=deepseek-ai/DeepSeek-V3.2-Exp
# 华为大模型API配置（必需，已内置默认值，如需修改可调整）
HUAWEI_API_KEY=sk-PoeiV3qwyTIRqcVc84E8E24cD2904872859a87922e0d9186
HUAWEI_API_ENDPOINT=http://10.100.31.26:3001/v1/chat/completions
HUAWEI_MODEL=DeepSeek-R1-Distill-Llama-70B
# 数据库配置（已默认配置，如需修改可调整）
DB_HOST=152.136.177.240
@ -243,7 +243,7 @@ print(response.json())
## 常见问题
**Q: 提示"未配置AI服务"？**
A: 检查 `.env` 文件中的 `SILICONFLOW_API_KEY` 是否已正确配置。
A: 系统仅支持华为大模型（已内置默认配置），请确保 `.env` 文件中正确设置了 `HUAWEI_API_KEY` 和 `HUAWEI_API_ENDPOINT`。如果华为大模型不可用，请检查网络连接和API配置。
**Q: 解析结果为空？**
A: 检查输入文本是否包含足够的信息，可以尝试更详细的输入文本。

104
services/ai_service.py
View File

@ -1,6 +1,6 @@
"""
AI服务 - 封装大模型调用
支持硅基流动和华为大模型（预留）
仅支持华为大模型
"""
import os
import requests
@ -12,22 +12,17 @@ class AIService:
"""AI服务类"""
def __init__(self):
self.siliconflow_api_key = os.getenv('SILICONFLOW_API_KEY')
self.siliconflow_model = os.getenv('SILICONFLOW_MODEL', 'deepseek-ai/DeepSeek-V3.2-Exp')
self.siliconflow_url = "https://api.siliconflow.cn/v1/chat/completions"
# 华为大模型配置（预留）
self.huawei_api_endpoint = os.getenv('HUAWEI_API_ENDPOINT')
self.huawei_api_key = os.getenv('HUAWEI_API_KEY')
# 华为大模型配置（必需）
self.huawei_api_endpoint = os.getenv('HUAWEI_API_ENDPOINT', 'http://10.100.31.26:3001/v1/chat/completions')
self.huawei_api_key = os.getenv('HUAWEI_API_KEY', 'sk-PoeiV3qwyTIRqcVc84E8E24cD2904872859a87922e0d9186')
self.huawei_model = os.getenv('HUAWEI_MODEL', 'DeepSeek-R1-Distill-Llama-70B')
# 确定使用的AI服务
self.ai_provider = self._determine_ai_provider()
def _determine_ai_provider(self) -> str:
"""确定使用的AI服务提供商"""
if self.siliconflow_api_key:
return 'siliconflow'
elif self.huawei_api_endpoint and self.huawei_api_key:
"""确定使用的AI服务提供商（仅支持华为大模型）"""
if self.huawei_api_endpoint and self.huawei_api_key:
return 'huawei'
else:
return 'none'
@ -44,17 +39,18 @@ class AIService:
提取的字段字典，格式: {field_code: field_value}
"""
if self.ai_provider == 'none':
raise Exception("未配置AI服务，请设置SILICONFLOW_API_KEY或华为大模型配置")
raise Exception("未配置华为大模型服务，请设置 HUAWEI_API_KEY 和 HUAWEI_API_ENDPOINT")
if self.ai_provider == 'siliconflow':
return self._extract_with_siliconflow(prompt, output_fields)
elif self.ai_provider == 'huawei':
if self.ai_provider == 'huawei':
return self._extract_with_huawei(prompt, output_fields)
else:
raise Exception(f"未知的AI服务提供商: {self.ai_provider}")
def _extract_with_siliconflow(self, prompt: str, output_fields: List[Dict]) -> Optional[Dict]:
"""使用硅基流动API提取字段"""
"""
使用硅基流动API提取字段（已不再使用，仅保留用于参考）
系统仅支持华为大模型，不再支持自动回退
"""
try:
payload = {
"model": self.siliconflow_model,
@ -119,9 +115,77 @@ class AIService:
raise Exception(f"AI服务调用失败: {str(e)}")
def _extract_with_huawei(self, prompt: str, output_fields: List[Dict]) -> Optional[Dict]:
"""使用华为大模型API提取字段（预留实现）"""
# TODO: 实现华为大模型接口调用
raise Exception("华为大模型接口暂未实现，请使用硅基流动API")
"""使用华为大模型API提取字段"""
try:
payload = {
"model": self.huawei_model,
"messages": [
{
"role": "system",
"content": "你是一个专业的数据提取助手，能够从文本中准确提取结构化信息。请严格按照JSON格式返回结果。"
},
{
"role": "user",
"content": prompt
}
],
"stream": False,
"presence_penalty": 1.03,
"frequency_penalty": 1.0,
"repetition_penalty": 1.0,
"temperature": 0.5,
"top_p": 0.95,
"top_k": 1,
"seed": 1,
"max_tokens": 8192,
"n": 1
}
headers = {
"Authorization": f"Bearer {self.huawei_api_key}",
"Content-Type": "application/json"
}
response = requests.post(
self.huawei_api_endpoint,
json=payload,
headers=headers,
timeout=60
)
if response.status_code != 200:
raise Exception(f"API调用失败: {response.status_code} - {response.text}")
result = response.json()
# 提取AI返回的内容
if 'choices' in result and len(result['choices']) > 0:
content = result['choices'][0]['message']['content']
# 尝试解析JSON
try:
# 如果返回的是代码块，提取JSON部分
if '```json' in content:
json_start = content.find('```json') + 7
json_end = content.find('```', json_start)
content = content[json_start:json_end].strip()
elif '```' in content:
json_start = content.find('```') + 3
json_end = content.find('```', json_start)
content = content[json_start:json_end].strip()
extracted_data = json.loads(content)
return extracted_data
except json.JSONDecodeError:
# 如果不是JSON，尝试从文本中提取
return self._parse_text_response(content, output_fields)
else:
raise Exception("API返回格式异常")
except requests.exceptions.Timeout:
raise Exception("AI服务调用超时")
except Exception as e:
raise Exception(f"AI服务调用失败: {str(e)}")
def _parse_text_response(self, text: str, output_fields: List[Dict]) -> Dict:
"""

35
template_ai_helper.py
View File

@ -15,12 +15,17 @@ class TemplateAIHelper:
"""模板AI辅助类，用于智能分析文档内容"""
def __init__(self):
self.api_key = os.getenv('SILICONFLOW_API_KEY')
self.model = os.getenv('SILICONFLOW_MODEL', 'deepseek-ai/DeepSeek-V3.2-Exp')
self.api_url = "https://api.siliconflow.cn/v1/chat/completions"
# 华为大模型配置（必需）
huawei_key = os.getenv('HUAWEI_API_KEY', 'sk-PoeiV3qwyTIRqcVc84E8E24cD2904872859a87922e0d9186')
huawei_endpoint = os.getenv('HUAWEI_API_ENDPOINT', 'http://10.100.31.26:3001/v1/chat/completions')
if not self.api_key:
raise Exception("未配置 SILICONFLOW_API_KEY，请在 .env 文件中设置")
if not huawei_key or not huawei_endpoint:
raise Exception("未配置华为大模型服务，请设置 HUAWEI_API_KEY 和 HUAWEI_API_ENDPOINT")
# 使用华为大模型
self.api_key = huawei_key
self.model = os.getenv('HUAWEI_MODEL', 'DeepSeek-R1-Distill-Llama-70B')
self.api_url = huawei_endpoint
def test_api_connection(self) -> bool:
"""
@ -30,7 +35,9 @@ class TemplateAIHelper:
是否连接成功
"""
try:
print(" [测试] 正在测试硅基流动API连接...")
print(f" [测试] 正在测试华为大模型API连接...")
# 华为大模型payload
test_payload = {
"model": self.model,
"messages": [
@ -39,6 +46,8 @@ class TemplateAIHelper:
"content": "测试"
}
],
"stream": False,
"temperature": 0.5,
"max_tokens": 10
}
@ -137,7 +146,7 @@ class TemplateAIHelper:
只返回JSON，不要其他说明文字。"""
# 调用AI API
# 调用AI API（华为大模型）
payload = {
"model": self.model,
"messages": [
@ -150,8 +159,16 @@ class TemplateAIHelper:
"content": prompt
}
],
"temperature": 0.2,
"max_tokens": 4000
"stream": False,
"presence_penalty": 1.03,
"frequency_penalty": 1.0,
"repetition_penalty": 1.0,
"temperature": 0.5,
"top_p": 0.95,
"top_k": 1,
"seed": 1,
"max_tokens": 8192,
"n": 1
}
headers = {

33
test_scripts/test_ai_api.py
View File

@ -1,5 +1,5 @@
"""
测试硅基流动API连接
测试大模型API连接（华为大模型默认，硅基流动备用）
"""
import os
from dotenv import load_dotenv
@ -8,23 +8,23 @@ import requests
# 加载环境变量
load_dotenv()
def test_siliconflow_api():
"""测试硅基流动API连接"""
def test_huawei_api():
"""测试华为大模型API连接"""
print("="*80)
print("测试硅基流动API连接")
print("测试华为大模型API连接")
print("="*80)
print()
# 读取配置
api_key = os.getenv('SILICONFLOW_API_KEY')
model = os.getenv('SILICONFLOW_MODEL', 'deepseek-ai/DeepSeek-V3.2-Exp')
api_url = "https://api.siliconflow.cn/v1/chat/completions"
# 读取配置（华为大模型）
api_key = os.getenv('HUAWEI_API_KEY', 'sk-PoeiV3qwyTIRqcVc84E8E24cD2904872859a87922e0d9186')
model = os.getenv('HUAWEI_MODEL', 'DeepSeek-R1-Distill-Llama-70B')
api_url = os.getenv('HUAWEI_API_ENDPOINT', 'http://10.100.31.26:3001/v1/chat/completions')
# 检查配置
print("1. 检查配置...")
if not api_key:
print(" ✗ 错误: 未找到 SILICONFLOW_API_KEY")
print(" 请在 .env 文件中设置: SILICONFLOW_API_KEY=你的API密钥")
print(" ✗ 错误: 未找到 HUAWEI_API_KEY")
print(" 请在 .env 文件中设置: HUAWEI_API_KEY=你的API密钥")
return False
print(f" ✓ API密钥: {api_key[:10]}...{api_key[-5:]}")
@ -43,7 +43,16 @@ def test_siliconflow_api():
"content": "请回复'测试成功'"
}
],
"max_tokens": 20
"stream": False,
"presence_penalty": 1.03,
"frequency_penalty": 1.0,
"repetition_penalty": 1.0,
"temperature": 0.5,
"top_p": 0.95,
"top_k": 1,
"seed": 1,
"max_tokens": 20,
"n": 1
}
headers = {
@ -87,7 +96,7 @@ def test_siliconflow_api():
return False
if __name__ == '__main__':
success = test_siliconflow_api()
success = test_huawei_api()
print()
print("="*80)
if success: