root/Gitcloned-BettaFish
1
1
Fork 0
Code Issues Pull Requests Projects Releases Activity

1. 修复论坛通信问题，基于日志块增加容错

Browse Source 
2. 使用ERROR层级来避免json解析错误导致的连环问题
This commit is contained in:
Doiiars
2025-11-06 18:53:23 +08:00
parent 7d5105e563
commit e4d075cf56
13 changed files with 92 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
ForumEngine/monitor.py
View File

@@ -70,6 +70,7 @@ class LogMonitor:
self.capturing_json = {} # 每个app的JSON捕获状态
self.json_buffer = {} # 每个app的JSON缓冲区
self.json_start_line = {} # 每个app的JSON开始行
self.in_error_block = {} # 每个app是否在ERROR块中
# 确保logs目录存在
self.log_dir.mkdir(exist_ok=True)
@@ -93,6 +94,7 @@ class LogMonitor:
self.capturing_json = {}
self.json_buffer = {}
self.json_start_line = {}
self.in_error_block = {}
# 重置主持人相关状态
self.agent_speeches_buffer = []
@@ -118,6 +120,21 @@ class LogMonitor:
except Exception as e:
logger.exception(f"ForumEngine: 写入forum.log失败: {e}")
def get_log_level(self, line: str) -> Optional[str]:
"""检测日志行的级别（INFO/ERROR/WARNING/DEBUG等）
支持loguru格式：YYYY-MM-DD HH:mm:ss.SSS | LEVEL | ...
Returns:
'INFO', 'ERROR', 'WARNING', 'DEBUG' 或 None（无法识别）
"""
# 检查loguru格式：YYYY-MM-DD HH:mm:ss.SSS | LEVEL | ...
# 匹配模式：| LEVEL | 或 | LEVEL |
match = re.search(r'\|\s*(INFO|ERROR|WARNING|DEBUG|TRACE|CRITICAL)\s*\|', line)
if match:
return match.group(1)
return None
def is_target_log_line(self, line: str) -> bool:
"""检查是否是目标日志行（SummaryNode）
@@ -132,6 +149,11 @@ class LogMonitor:
- 包含错误关键词的日志（JSON解析失败、JSON修复失败等）
"""
# 排除 ERROR 级别的日志
log_level = self.get_log_level(line)
if log_level == 'ERROR':
return False
# 兼容旧检查方式
if "| ERROR" in line or "| ERROR |" in line:
return False
@@ -381,6 +403,7 @@ class LogMonitor:
# 重置JSON捕获状态
self.capturing_json[app_name] = False
self.json_buffer[app_name] = []
self.in_error_block[app_name] = False
if current_size > last_position:
with open(file_path, 'r', encoding='utf-8') as f:
@@ -400,17 +423,47 @@ class LogMonitor:
return new_lines
def process_lines_for_json(self, lines: List[str], app_name: str) -> List[str]:
"""处理行以捕获多行JSON内容"""
"""处理行以捕获多行JSON内容
实现ERROR块过滤：如果遇到ERROR级别的日志，拒绝处理直到遇到下一个INFO级别的日志
"""
captured_contents = []
# 初始化状态
if app_name not in self.capturing_json:
self.capturing_json[app_name] = False
self.json_buffer[app_name] = []
if app_name not in self.in_error_block:
self.in_error_block[app_name] = False
for line in lines:
if not line.strip():
continue
# 首先检查日志级别，更新ERROR块状态
log_level = self.get_log_level(line)
if log_level == 'ERROR':
# 遇到ERROR，进入ERROR块状态
self.in_error_block[app_name] = True
# 如果正在捕获JSON，立即停止并清空缓冲区
if self.capturing_json[app_name]:
self.capturing_json[app_name] = False
self.json_buffer[app_name] = []
# 跳过当前行，不处理
continue
elif log_level == 'INFO':
# 遇到INFO，退出ERROR块状态
self.in_error_block[app_name] = False
# 其他级别（WARNING、DEBUG等）保持当前状态
# 如果在ERROR块中，拒绝处理所有内容
if self.in_error_block[app_name]:
# 如果正在捕获JSON，立即停止并清空缓冲区
if self.capturing_json[app_name]:
self.capturing_json[app_name] = False
self.json_buffer[app_name] = []
# 跳过当前行，不处理
continue
# 检查是否是目标节点行和JSON开始标记
is_target = self.is_target_log_line(line)
@@ -538,6 +591,7 @@ class LogMonitor:
self.file_positions[app_name] = self.get_file_size(log_file)
self.capturing_json[app_name] = False
self.json_buffer[app_name] = []
self.in_error_block[app_name] = False
# logger.info(f"ForumEngine: {app_name} 基线行数: {self.file_line_counts[app_name]}")
while self.is_monitoring:
@@ -601,6 +655,7 @@ class LogMonitor:
# 重置JSON捕获状态
self.capturing_json[app_name] = False
self.json_buffer[app_name] = []
self.in_error_block[app_name] = False
# 更新行数记录
self.file_line_counts[app_name] = current_lines

2
InsightEngine/agent.py
View File

@@ -779,5 +779,5 @@ def create_agent(config_file: Optional[str] = None) -> DeepSearchAgent:
Returns:
DeepSearchAgent实例
"""
config = settings
config = Settings() # 以空配置初始化，而从从环境变量初始化
return DeepSearchAgent(config)

8
InsightEngine/nodes/report_structure_node.py
View File

@@ -87,11 +87,11 @@ class ReportStructureNode(StateMutationNode):
report_structure = json.loads(cleaned_output)
logger.info("JSON解析成功")
except JSONDecodeError as e:
logger.exception(f"JSON解析失败: {str(e)}")
logger.error(f"JSON解析失败: {str(e)}")
# 使用更强大的提取方法
report_structure = extract_clean_response(cleaned_output)
if "error" in report_structure:
logger.exception("JSON解析失败，尝试修复...")
logger.error("JSON解析失败，尝试修复...")
# 尝试修复JSON
fixed_json = fix_incomplete_json(cleaned_output)
if fixed_json:
@@ -99,11 +99,11 @@ class ReportStructureNode(StateMutationNode):
report_structure = json.loads(fixed_json)
logger.info("JSON修复成功")
except JSONDecodeError:
logger.exception("JSON修复失败")
logger.error("JSON修复失败")
# 返回默认结构
return self._generate_default_structure()
else:
logger.exception("无法修复JSON，使用默认结构")
logger.error("无法修复JSON，使用默认结构")
return self._generate_default_structure()
# 验证结构

4
InsightEngine/nodes/search_node.py
View File

@@ -101,7 +101,7 @@ class FirstSearchNode(BaseNode):
result = json.loads(cleaned_output)
logger.info("JSON解析成功")
except JSONDecodeError as e:
logger.exception(f"JSON解析失败: {str(e)}")
logger.error(f"JSON解析失败: {str(e)}")
# 使用更强大的提取方法
result = extract_clean_response(cleaned_output)
if "error" in result:
@@ -236,7 +236,7 @@ class ReflectionNode(BaseNode):
result = json.loads(cleaned_output)
logger.info("JSON解析成功")
except JSONDecodeError as e:
logger.exception(f"JSON解析失败: {str(e)}")
logger.error(f"JSON解析失败: {str(e)}")
# 使用更强大的提取方法
result = extract_clean_response(cleaned_output)
if "error" in result:

8
InsightEngine/nodes/summary_node.py
View File

@@ -135,7 +135,7 @@ class FirstSummaryNode(StateMutationNode):
result = json.loads(cleaned_output)
logger.info("JSON解析成功")
except JSONDecodeError as e:
logger.exception(f"JSON解析失败: {str(e)}")
logger.error(f"JSON解析失败: {str(e)}")
# 尝试修复JSON
fixed_json = fix_incomplete_json(cleaned_output)
if fixed_json:
@@ -300,7 +300,7 @@ class ReflectionSummaryNode(StateMutationNode):
result = json.loads(cleaned_output)
logger.info("JSON解析成功")
except JSONDecodeError as e:
logger.exception(f"JSON解析失败: {str(e)}")
logger.error(f"JSON解析失败: {str(e)}")
# 尝试修复JSON
fixed_json = fix_incomplete_json(cleaned_output)
if fixed_json:
@@ -308,11 +308,11 @@ class ReflectionSummaryNode(StateMutationNode):
result = json.loads(fixed_json)
logger.info("JSON修复成功")
except JSONDecodeError:
logger.info("JSON修复失败，直接使用清理后的文本")
logger.error("JSON修复失败，直接使用清理后的文本")
# 如果不是JSON格式，直接返回清理后的文本
return cleaned_output
else:
logger.info("无法修复JSON，直接使用清理后的文本")
logger.error("无法修复JSON，直接使用清理后的文本")
# 如果不是JSON格式，直接返回清理后的文本
return cleaned_output

2
MediaEngine/nodes/report_structure_node.py
View File

@@ -87,7 +87,7 @@ class ReportStructureNode(StateMutationNode):
report_structure = json.loads(cleaned_output)
logger.info("JSON解析成功")
except JSONDecodeError as e:
logger.exception(f"JSON解析失败: {str(e)}")
logger.error(f"JSON解析失败: {str(e)}")
# 使用更强大的提取方法
report_structure = extract_clean_response(cleaned_output)
if "error" in report_structure:

4
MediaEngine/nodes/search_node.py
View File

@@ -101,7 +101,7 @@ class FirstSearchNode(BaseNode):
result = json.loads(cleaned_output)
logger.info("JSON解析成功")
except JSONDecodeError as e:
logger.exception(f"JSON解析失败: {str(e)}")
logger.error(f"JSON解析失败: {str(e)}")
# 使用更强大的提取方法
result = extract_clean_response(cleaned_output)
if "error" in result:
@@ -236,7 +236,7 @@ class ReflectionNode(BaseNode):
result = json.loads(cleaned_output)
logger.info("JSON解析成功")
except JSONDecodeError as e:
logger.exception(f"JSON解析失败: {str(e)}")
logger.error(f"JSON解析失败: {str(e)}")
# 使用更强大的提取方法
result = extract_clean_response(cleaned_output)
if "error" in result:

8
MediaEngine/nodes/summary_node.py
View File

@@ -138,7 +138,7 @@ class FirstSummaryNode(StateMutationNode):
result = json.loads(cleaned_output)
logger.info("JSON解析成功")
except JSONDecodeError as e:
logger.exception(f"JSON解析失败: {str(e)}")
logger.error(f"JSON解析失败: {str(e)}")
# 尝试修复JSON
fixed_json = fix_incomplete_json(cleaned_output)
if fixed_json:
@@ -306,7 +306,7 @@ class ReflectionSummaryNode(StateMutationNode):
result = json.loads(cleaned_output)
logger.info("JSON解析成功")
except JSONDecodeError as e:
logger.exception(f"JSON解析失败: {str(e)}")
logger.error(f"JSON解析失败: {str(e)}")
# 尝试修复JSON
fixed_json = fix_incomplete_json(cleaned_output)
if fixed_json:
@@ -314,11 +314,11 @@ class ReflectionSummaryNode(StateMutationNode):
result = json.loads(fixed_json)
logger.info("JSON修复成功")
except JSONDecodeError:
logger.exception("JSON修复失败，直接使用清理后的文本")
logger.error("JSON修复失败，直接使用清理后的文本")
# 如果不是JSON格式，直接返回清理后的文本
return cleaned_output
else:
logger.exception("无法修复JSON，直接使用清理后的文本")
logger.error("无法修复JSON，直接使用清理后的文本")
# 如果不是JSON格式，直接返回清理后的文本
return cleaned_output

2
QueryEngine/nodes/report_structure_node.py
View File

@@ -87,7 +87,7 @@ class ReportStructureNode(StateMutationNode):
report_structure = json.loads(cleaned_output)
logger.info("JSON解析成功")
except JSONDecodeError as e:
logger.exception(f"JSON解析失败: {str(e)}")
logger.error(f"JSON解析失败: {str(e)}")
# 使用更强大的提取方法
report_structure = extract_clean_response(cleaned_output)
if "error" in report_structure:

4
QueryEngine/nodes/search_node.py
View File

@@ -101,7 +101,7 @@ class FirstSearchNode(BaseNode):
result = json.loads(cleaned_output)
logger.info("JSON解析成功")
except JSONDecodeError as e:
logger.exception(f"JSON解析失败: {str(e)}")
logger.error(f"JSON解析失败: {str(e)}")
# 使用更强大的提取方法
result = extract_clean_response(cleaned_output)
if "error" in result:
@@ -236,7 +236,7 @@ class ReflectionNode(BaseNode):
result = json.loads(cleaned_output)
logger.info("JSON解析成功")
except JSONDecodeError as e:
logger.exception(f"JSON解析失败: {str(e)}")
logger.error(f"JSON解析失败: {str(e)}")
# 使用更强大的提取方法
result = extract_clean_response(cleaned_output)
if "error" in result:

12
QueryEngine/nodes/summary_node.py
View File

@@ -138,7 +138,7 @@ class FirstSummaryNode(StateMutationNode):
result = json.loads(cleaned_output)
logger.info("JSON解析成功")
except JSONDecodeError as e:
logger.exception(f"JSON解析失败: {str(e)}")
logger.error(f"JSON解析失败: {str(e)}")
# 尝试修复JSON
fixed_json = fix_incomplete_json(cleaned_output)
if fixed_json:
@@ -146,11 +146,11 @@ class FirstSummaryNode(StateMutationNode):
result = json.loads(fixed_json)
logger.info("JSON修复成功")
except JSONDecodeError:
logger.exception("JSON修复失败，直接使用清理后的文本")
logger.error("JSON修复失败，直接使用清理后的文本")
# 如果不是JSON格式，直接返回清理后的文本
return cleaned_output
else:
logger.exception("无法修复JSON，直接使用清理后的文本")
logger.error("无法修复JSON，直接使用清理后的文本")
# 如果不是JSON格式，直接返回清理后的文本
return cleaned_output
@@ -306,7 +306,7 @@ class ReflectionSummaryNode(StateMutationNode):
result = json.loads(cleaned_output)
logger.info("JSON解析成功")
except JSONDecodeError as e:
logger.exception(f"JSON解析失败: {str(e)}")
logger.error(f"JSON解析失败: {str(e)}")
# 尝试修复JSON
fixed_json = fix_incomplete_json(cleaned_output)
if fixed_json:
@@ -314,11 +314,11 @@ class ReflectionSummaryNode(StateMutationNode):
result = json.loads(fixed_json)
logger.info("JSON修复成功")
except JSONDecodeError:
logger.exception("JSON修复失败，直接使用清理后的文本")
logger.error("JSON修复失败，直接使用清理后的文本")
# 如果不是JSON格式，直接返回清理后的文本
return cleaned_output
else:
logger.exception("无法修复JSON，直接使用清理后的文本")
logger.error("无法修复JSON，直接使用清理后的文本")
# 如果不是JSON格式，直接返回清理后的文本
return cleaned_output

16
ReportEngine/agent.py
View File

@@ -138,7 +138,7 @@ class ReportAgent:
self.state = ReportState()
# 确保输出目录存在
os.makedirs(settings.OUTPUT_DIR, exist_ok=True)
os.makedirs(self.config.OUTPUT_DIR, exist_ok=True)
logger.info("Report Agent已初始化")
logger.info(f"使用LLM: {self.llm_client.get_model_info()}")
@@ -146,11 +146,11 @@ class ReportAgent:
def _setup_logging(self):
"""设置日志"""
# 确保日志目录存在
log_dir = os.path.dirname(settings.LOG_FILE)
log_dir = os.path.dirname(self.config.LOG_FILE)
os.makedirs(log_dir, exist_ok=True)
# 创建专用的logger，避免与其他模块冲突
logger.add(settings.LOG_FILE, level="INFO")
logger.add(self.config.LOG_FILE, level="INFO")
def _initialize_file_baseline(self):
"""初始化文件数量基准"""
@@ -164,9 +164,9 @@ class ReportAgent:
def _initialize_llm(self) -> LLMClient:
"""初始化LLM客户端"""
return LLMClient(
api_key=settings.REPORT_ENGINE_API_KEY,
model_name=settings.REPORT_ENGINE_MODEL_NAME,
base_url=settings.REPORT_ENGINE_BASE_URL,
api_key=self.config.REPORT_ENGINE_API_KEY,
model_name=self.config.REPORT_ENGINE_MODEL_NAME,
base_url=self.config.REPORT_ENGINE_BASE_URL,
)
def _initialize_nodes(self):
@@ -351,7 +351,7 @@ class ReportAgent:
query_safe = query_safe.replace(' ', '_')[:30]
filename = f"final_report_{query_safe}_{timestamp}.html"
filepath = os.path.join(settings.OUTPUT_DIR, filename)
filepath = os.path.join(self.config.OUTPUT_DIR, filename)
# 保存HTML报告
with open(filepath, 'w', encoding='utf-8') as f:
@@ -361,7 +361,7 @@ class ReportAgent:
# 保存状态
state_filename = f"report_state_{query_safe}_{timestamp}.json"
state_filepath = os.path.join(settings.OUTPUT_DIR, state_filename)
state_filepath = os.path.join(self.config.OUTPUT_DIR, state_filename)
self.state.save_to_file(state_filepath)
logger.info(f"状态已保存到: {state_filepath}")

2
ReportEngine/nodes/template_selection_node.py
View File

@@ -145,7 +145,7 @@ class TemplateSelectionNode(BaseNode):
return None
except json.JSONDecodeError as e:
logger.exception(f"JSON解析失败: {str(e)}")
logger.error(f"JSON解析失败: {str(e)}")
# 尝试从文本响应中提取模板信息
return self._extract_template_from_text(response, available_templates)