为华为大模型API调用添加重试机制，增强了错误处理逻辑，确保在请求失败时能够自动重试并提供详细的错误信息。同时，将API调用逻辑分离到单独的方法中，以提高代码可读性和可维护性。
This commit is contained in:
python
parent
8461725a13
commit
14ff607b52
@ -4,6 +4,7 @@ AI服务 - 封装大模型调用
|
||||
"""
|
||||
import os
|
||||
import re
|
||||
import time
|
||||
import requests
|
||||
import json
|
||||
from typing import Dict, List, Optional
|
||||
@ -116,74 +117,139 @@ class AIService:
|
||||
raise Exception(f"AI服务调用失败: {str(e)}")
|
||||
|
||||
def _extract_with_huawei(self, prompt: str, output_fields: List[Dict]) -> Optional[Dict]:
|
||||
"""使用华为大模型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.3,
|
||||
"top_p": 0.95,
|
||||
"top_k": 1,
|
||||
"seed": 1,
|
||||
"max_tokens": 8192,
|
||||
"n": 1,
|
||||
"enable_thinking": True
|
||||
}
|
||||
"""
|
||||
使用华为大模型API提取字段（带重试机制）
|
||||
至少重试3次，总共最多尝试4次
|
||||
"""
|
||||
max_retries = 3 # 最多重试3次，总共4次尝试
|
||||
retry_delay = 1 # 重试延迟（秒），每次重试延迟递增
|
||||
|
||||
headers = {
|
||||
"Authorization": f"Bearer {self.huawei_api_key}",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
last_exception = None
|
||||
|
||||
response = requests.post(
|
||||
self.huawei_api_endpoint,
|
||||
json=payload,
|
||||
headers=headers,
|
||||
timeout=60
|
||||
)
|
||||
for attempt in range(max_retries + 1): # 0, 1, 2, 3 (总共4次)
|
||||
try:
|
||||
if attempt > 0:
|
||||
# 重试前等待，延迟时间递增（1秒、2秒、3秒）
|
||||
wait_time = retry_delay * attempt
|
||||
print(f"[AI服务] 第 {attempt} 次重试，等待 {wait_time} 秒后重试...")
|
||||
time.sleep(wait_time)
|
||||
|
||||
if response.status_code != 200:
|
||||
raise Exception(f"API调用失败: {response.status_code} - {response.text}")
|
||||
print(f"[AI服务] 正在调用华为大模型API (尝试 {attempt + 1}/{max_retries + 1})...")
|
||||
result = self._call_huawei_api_once(prompt, output_fields)
|
||||
|
||||
result = response.json()
|
||||
if result is not None:
|
||||
if attempt > 0:
|
||||
print(f"[AI服务] 重试成功！")
|
||||
return result
|
||||
|
||||
# 提取AI返回的内容
|
||||
if 'choices' in result and len(result['choices']) > 0:
|
||||
content = result['choices'][0]['message']['content']
|
||||
except requests.exceptions.Timeout as e:
|
||||
last_exception = e
|
||||
error_msg = f"AI服务调用超时 (尝试 {attempt + 1}/{max_retries + 1})"
|
||||
print(f"[AI服务] {error_msg}")
|
||||
if attempt < max_retries:
|
||||
continue
|
||||
else:
|
||||
raise Exception(f"{error_msg}: {str(e)}")
|
||||
|
||||
# 处理思考过程标签（华为大模型可能返回思考过程）
|
||||
# 移除思考过程标签之前的内容，只保留实际回答
|
||||
# 根据用户提供的示例，华为大模型使用 </think> 标签
|
||||
if '</think>' in content:
|
||||
content = content.split('</think>')[-1].strip()
|
||||
except requests.exceptions.ConnectionError as e:
|
||||
last_exception = e
|
||||
error_msg = f"连接错误 (尝试 {attempt + 1}/{max_retries + 1})"
|
||||
print(f"[AI服务] {error_msg}: {str(e)}")
|
||||
if attempt < max_retries:
|
||||
continue
|
||||
else:
|
||||
raise Exception(f"{error_msg}: {str(e)}")
|
||||
|
||||
# 尝试解析JSON
|
||||
extracted_data = self._extract_json_from_text(content)
|
||||
if extracted_data:
|
||||
return extracted_data
|
||||
except requests.exceptions.RequestException as e:
|
||||
last_exception = e
|
||||
error_msg = f"请求异常 (尝试 {attempt + 1}/{max_retries + 1})"
|
||||
print(f"[AI服务] {error_msg}: {str(e)}")
|
||||
if attempt < max_retries:
|
||||
continue
|
||||
else:
|
||||
raise Exception(f"{error_msg}: {str(e)}")
|
||||
|
||||
# 如果无法提取JSON，尝试从文本中提取
|
||||
return self._parse_text_response(content, output_fields)
|
||||
else:
|
||||
raise Exception("API返回格式异常")
|
||||
except Exception as e:
|
||||
last_exception = e
|
||||
error_msg = f"AI服务调用失败 (尝试 {attempt + 1}/{max_retries + 1})"
|
||||
print(f"[AI服务] {error_msg}: {str(e)}")
|
||||
# 对于其他类型的错误，也进行重试
|
||||
if attempt < max_retries:
|
||||
continue
|
||||
else:
|
||||
raise Exception(f"{error_msg}: {str(e)}")
|
||||
|
||||
except requests.exceptions.Timeout:
|
||||
raise Exception("AI服务调用超时")
|
||||
except Exception as e:
|
||||
raise Exception(f"AI服务调用失败: {str(e)}")
|
||||
# 如果所有重试都失败了
|
||||
if last_exception:
|
||||
raise Exception(f"AI服务调用失败，已重试 {max_retries} 次: {str(last_exception)}")
|
||||
else:
|
||||
raise Exception(f"AI服务调用失败，已重试 {max_retries} 次")
|
||||
|
||||
def _call_huawei_api_once(self, prompt: str, output_fields: List[Dict]) -> Optional[Dict]:
|
||||
"""
|
||||
单次调用华为大模型API（不包含重试逻辑）
|
||||
"""
|
||||
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.3,
|
||||
"top_p": 0.95,
|
||||
"top_k": 1,
|
||||
"seed": 1,
|
||||
"max_tokens": 8192,
|
||||
"n": 1,
|
||||
"enable_thinking": True
|
||||
}
|
||||
|
||||
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']
|
||||
|
||||
# 处理思考过程标签（华为大模型可能返回思考过程）
|
||||
# 移除思考过程标签之前的内容，只保留实际回答
|
||||
# 根据用户提供的示例，华为大模型使用 </think> 标签
|
||||
if '</think>' in content:
|
||||
content = content.split('</think>')[-1].strip()
|
||||
|
||||
# 尝试解析JSON
|
||||
extracted_data = self._extract_json_from_text(content)
|
||||
if extracted_data:
|
||||
return extracted_data
|
||||
|
||||
# 如果无法提取JSON，尝试从文本中提取
|
||||
return self._parse_text_response(content, output_fields)
|
||||
else:
|
||||
raise Exception("API返回格式异常：未找到choices字段或choices为空")
|
||||
|
||||
def _extract_json_from_text(self, text: str) -> Optional[Dict]:
|
||||
"""
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user