diff --git a/services/ai_service.py b/services/ai_service.py index 2581a79..4fa434e 100644 --- a/services/ai_service.py +++ b/services/ai_service.py @@ -3,6 +3,7 @@ AI服务 - 封装大模型调用 仅支持华为大模型 """ import os +import re import requests import json from typing import Dict, List, Optional @@ -122,7 +123,7 @@ class AIService: "messages": [ { "role": "system", - "content": "你是一个专业的数据提取助手,能够从文本中准确提取结构化信息。请严格按照JSON格式返回结果。" + "content": "你是一个专业的数据提取助手,能够从文本中准确提取结构化信息。请严格按照JSON格式返回结果,只返回JSON对象,不要包含任何其他文字说明、思考过程或markdown代码块标记。" }, { "role": "user", @@ -162,23 +163,19 @@ class AIService: if 'choices' in result and len(result['choices']) > 0: content = result['choices'][0]['message']['content'] + # 处理思考过程标签(华为大模型可能返回思考过程) + # 移除思考过程标签之前的内容,只保留实际回答 + # 根据用户提供的示例,华为大模型使用 标签 + if '' in content: + content = content.split('')[-1].strip() + # 尝试解析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) + extracted_data = self._extract_json_from_text(content) + if extracted_data: return extracted_data - except json.JSONDecodeError: - # 如果不是JSON,尝试从文本中提取 - return self._parse_text_response(content, output_fields) + + # 如果无法提取JSON,尝试从文本中提取 + return self._parse_text_response(content, output_fields) else: raise Exception("API返回格式异常") @@ -187,6 +184,82 @@ class AIService: except Exception as e: raise Exception(f"AI服务调用失败: {str(e)}") + def _extract_json_from_text(self, text: str) -> Optional[Dict]: + """ + 从文本中提取JSON对象 + 支持多种格式: + 1. 纯JSON对象 + 2. 包裹在 ```json 代码块中的JSON + 3. 包裹在 ``` 代码块中的JSON + 4. 文本中包含的JSON对象 + """ + # 方法1: 尝试提取代码块中的JSON + if '```json' in text: + json_start = text.find('```json') + 7 + json_end = text.find('```', json_start) + if json_end != -1: + json_str = text[json_start:json_end].strip() + try: + return json.loads(json_str) + except json.JSONDecodeError: + pass + + if '```' in text: + json_start = text.find('```') + 3 + json_end = text.find('```', json_start) + if json_end != -1: + json_str = text[json_start:json_end].strip() + # 如果不是json标记,尝试解析 + try: + return json.loads(json_str) + except json.JSONDecodeError: + pass + + # 方法2: 尝试直接解析整个文本 + try: + return json.loads(text.strip()) + except json.JSONDecodeError: + pass + + # 方法3: 尝试查找文本中的JSON对象(以 { 开始,以 } 结束) + # 使用正则表达式找到最外层的JSON对象 + json_pattern = r'\{[^{}]*(?:\{[^{}]*\}[^{}]*)*\}' + matches = re.finditer(json_pattern, text, re.DOTALL) + + for match in matches: + json_str = match.group(0) + try: + data = json.loads(json_str) + # 验证是否包含预期的字段(至少有一个输出字段的key) + if isinstance(data, dict) and len(data) > 0: + return data + except json.JSONDecodeError: + continue + + # 方法4: 尝试查找嵌套的JSON对象(更复杂的匹配) + # 找到第一个 { 和最后一个匹配的 } + start_idx = text.find('{') + if start_idx != -1: + brace_count = 0 + end_idx = start_idx + for i in range(start_idx, len(text)): + if text[i] == '{': + brace_count += 1 + elif text[i] == '}': + brace_count -= 1 + if brace_count == 0: + end_idx = i + break + + if end_idx > start_idx: + json_str = text[start_idx:end_idx + 1] + try: + return json.loads(json_str) + except json.JSONDecodeError: + pass + + return None + def _parse_text_response(self, text: str, output_fields: List[Dict]) -> Dict: """ 从文本响应中解析字段值(备用方案)