#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 测试优化的辩论流程控制系统 验证阶段转换和发言权争夺逻辑的改进 """ import sys import os import time import threading from datetime import datetime, timedelta # 添加项目路径 sys.path.append('/home/ben/liurenchaxin/src') from jixia.debates.optimized_debate_flow import ( OptimizedDebateFlowController, FlowControlConfig, FlowControlMode, TransitionTrigger, SpeakerSelectionStrategy, DebateStage ) def test_basic_flow_control(): """测试基础流程控制""" print("🧪 测试基础流程控制") print("-" * 30) controller = OptimizedDebateFlowController() # 测试获取当前发言者 speaker = controller.get_current_speaker() print(f"✅ 当前发言者: {speaker}") assert speaker is not None, "应该能获取到发言者" # 测试记录发言 controller.record_speech(speaker, "这是一个测试发言") print(f"✅ 发言记录成功，当前进度: {controller.stage_progress}") # 测试流程状态 status = controller.get_flow_status() print(f"✅ 流程状态: {status['current_stage']} 阶段") assert status['current_stage'] == '起', "应该在起阶段" return True def test_stage_transition(): """测试阶段转换""" print("\n🧪 测试阶段转换") print("-" * 30) config = FlowControlConfig( mode=FlowControlMode.STRICT, transition_triggers=[TransitionTrigger.PROGRESS_BASED] ) controller = OptimizedDebateFlowController(config) initial_stage = controller.current_stage print(f"初始阶段: {initial_stage.value}") # 模拟完成一个阶段的所有发言 stage_config = controller.stage_configs[initial_stage] max_progress = stage_config["max_progress"] for i in range(max_progress): speaker = controller.get_current_speaker() controller.record_speech(speaker, f"第{i+1}次发言") if i == max_progress - 1: # 最后一次发言后应该自动转换阶段 if controller._should_advance_stage(): success = controller.advance_stage() print(f"✅ 阶段转换成功: {success}") print(f"新阶段: {controller.current_stage.value}") assert controller.current_stage != initial_stage, "阶段应该已经改变" break return True def test_speaker_selection_strategies(): """测试发言者选择策略""" print("\n🧪 测试发言者选择策略") print("-" * 30) strategies = [ SpeakerSelectionStrategy.ROUND_ROBIN, SpeakerSelectionStrategy.CONTEXT_AWARE, SpeakerSelectionStrategy.COMPETITIVE ] for strategy in strategies: print(f"\n测试策略: {strategy.value}") config = FlowControlConfig(speaker_selection_strategy=strategy) controller = OptimizedDebateFlowController(config) # 获取几个发言者 speakers = [] for i in range(3): speaker = controller.get_current_speaker() speakers.append(speaker) controller.record_speech(speaker, f"策略测试发言 {i+1}") print(f"发言者序列: {speakers}") assert len(set(speakers)) > 0, f"策略 {strategy.value} 应该能选择发言者" print("✅ 所有发言者选择策略测试通过") return True def test_speaker_request_system(): """测试发言请求系统""" print("\n🧪 测试发言请求系统") print("-" * 30) config = FlowControlConfig( speaker_selection_strategy=SpeakerSelectionStrategy.COMPETITIVE ) controller = OptimizedDebateFlowController(config) # 提交发言请求 controller.request_speaking_turn("正1", "紧急反驳", urgency=5, topic_relevance=0.9) controller.request_speaking_turn("反2", "补充论据", urgency=2, topic_relevance=0.7) controller.request_speaking_turn("正3", "重要澄清", urgency=4, topic_relevance=0.8) print(f"待处理请求数量: {len(controller.pending_requests)}") assert len(controller.pending_requests) == 3, "应该有3个待处理请求" # 获取下一个发言者（应该是最高优先级的） next_speaker = controller.get_current_speaker() print(f"✅ 高优先级发言者: {next_speaker}") # 记录发言后，请求应该被移除 controller.record_speech(next_speaker, "响应紧急请求的发言") print(f"发言后待处理请求数量: {len(controller.pending_requests)}") return True def test_context_aware_selection(): """测试上下文感知选择""" print("\n🧪 测试上下文感知选择") print("-" * 30) config = FlowControlConfig( speaker_selection_strategy=SpeakerSelectionStrategy.CONTEXT_AWARE ) controller = OptimizedDebateFlowController(config) # 模拟一些发言历史 test_speeches = [ ("正1", "我支持AI投资"), ("正1", "理由是技术发展迅速"), # 连续发言 ("反1", "但风险很高"), ("正2", "我们有风控措施") ] for speaker, message in test_speeches: controller.record_speech(speaker, message) # 分析当前上下文 context = controller._analyze_current_context() print(f"当前上下文: {context}") # 获取下一个发言者（应该避免连续发言） next_speaker = controller.get_current_speaker() print(f"✅ 上下文感知选择的发言者: {next_speaker}") # 验证不是最近的发言者 recent_speakers = [speech[0] for speech in test_speeches[-2:]] print(f"最近发言者: {recent_speakers}") return True def test_stage_metrics(): """测试阶段指标""" print("\n🧪 测试阶段指标") print("-" * 30) controller = OptimizedDebateFlowController() # 模拟一些发言 test_speeches = [ ("吕洞宾", "AI投资是未来趋势，我们应该积极参与。数据显示这个领域的增长潜力巨大。"), ("何仙姑", "但是我们也要考虑风险因素。"), ("铁拐李", "我同意吕洞宾的观点，因为技术发展确实很快。"), ("汉钟离", "然而市场波动性不容忽视。") ] for speaker, message in test_speeches: controller.record_speech(speaker, message) # 检查阶段指标 metrics = controller.current_stage_metrics print(f"发言数量: {metrics.speech_count}") print(f"质量分数: {metrics.quality_score:.3f}") print(f"参与平衡: {metrics.participation_balance:.3f}") print(f"转换准备度: {metrics.transition_readiness:.3f}") print(f"发言者分布: {metrics.speaker_distribution}") assert metrics.speech_count == 4, "发言数量应该是4" assert 0 <= metrics.quality_score <= 1, "质量分数应该在0-1之间" assert 0 <= metrics.participation_balance <= 1, "参与平衡应该在0-1之间" print("✅ 阶段指标计算正确") return True def test_adaptive_mode(): """测试自适应模式""" print("\n🧪 测试自适应模式") print("-" * 30) config = FlowControlConfig( mode=FlowControlMode.ADAPTIVE, transition_triggers=[TransitionTrigger.QUALITY_BASED, TransitionTrigger.PROGRESS_BASED], quality_threshold=0.7 ) controller = OptimizedDebateFlowController(config) # 模拟高质量发言 high_quality_speeches = [ ("吕洞宾", "根据最新的市场分析数据，AI投资领域在过去三年中显示出了显著的增长趋势。我们需要仔细分析这些数据背后的原因。"), ("何仙姑", "虽然数据显示增长，但是我们也必须考虑到技术泡沫的可能性。历史上类似的技术热潮往往伴随着高风险。"), ("铁拐李", "我认为关键在于风险管理。如果我们能够建立完善的风控体系，就能够在享受收益的同时控制风险。") ] for speaker, message in high_quality_speeches: controller.record_speech(speaker, message) # 检查是否达到质量阈值 if controller.current_stage_metrics.quality_score >= config.quality_threshold: print(f"✅ 达到质量阈值: {controller.current_stage_metrics.quality_score:.3f}") # 检查是否应该转换阶段 should_advance = controller._should_advance_stage() print(f"是否应该推进阶段: {should_advance}") break return True def test_event_system(): """测试事件系统""" print("\n🧪 测试事件系统") print("-" * 30) controller = OptimizedDebateFlowController() # 记录事件 events_received = [] def event_handler(event): events_received.append(event.event_type) print(f"📢 收到事件: {event.event_type}") # 注册事件处理器 controller.add_event_handler("speech_recorded", event_handler) controller.add_event_handler("speaker_request", event_handler) # 触发事件 controller.record_speech("测试发言者", "测试消息") controller.request_speaking_turn("正1", "测试请求") # 等待事件处理 time.sleep(0.1) print(f"收到的事件: {events_received}") assert "speech_recorded" in events_received, "应该收到发言记录事件" assert "speaker_request" in events_received, "应该收到发言请求事件" print("✅ 事件系统工作正常") return True def test_data_persistence(): """测试数据持久化""" print("\n🧪 测试数据持久化") print("-" * 30) controller = OptimizedDebateFlowController() # 模拟一些活动 controller.record_speech("吕洞宾", "测试发言1") controller.record_speech("何仙姑", "测试发言2") controller.request_speaking_turn("正1", "测试请求") # 保存数据 filename = "test_flow_data.json" controller.save_flow_data(filename) # 检查文件是否存在 assert os.path.exists(filename), "数据文件应该被创建" # 读取并验证数据 import json with open(filename, 'r', encoding='utf-8') as f: data = json.load(f) assert "config" in data, "数据应该包含配置信息" assert "current_state" in data, "数据应该包含当前状态" assert "debate_history" in data, "数据应该包含辩论历史" assert len(data["debate_history"]) == 2, "应该有2条发言记录" print(f"✅ 数据持久化成功，文件大小: {os.path.getsize(filename)} 字节") # 清理测试文件 os.remove(filename) return True def test_performance(): """测试性能""" print("\n🧪 测试性能") print("-" * 30) controller = OptimizedDebateFlowController() # 测试发言者选择性能 start_time = time.time() for i in range(100): speaker = controller.get_current_speaker() controller.record_speech(speaker, f"性能测试发言 {i}") end_time = time.time() duration = end_time - start_time print(f"100次发言处理耗时: {duration:.3f} 秒") print(f"平均每次处理时间: {duration/100*1000:.2f} 毫秒") print(f"处理速度: {100/duration:.1f} 次/秒") assert duration < 5.0, "100次处理应该在5秒内完成" # 测试并发性能 def concurrent_speech_recording(): for i in range(10): speaker = controller.get_current_speaker() controller.record_speech(speaker, f"并发测试发言 {threading.current_thread().name}-{i}") start_time = time.time() threads = [] for i in range(5): thread = threading.Thread(target=concurrent_speech_recording, name=f"Thread-{i}") threads.append(thread) thread.start() for thread in threads: thread.join() end_time = time.time() concurrent_duration = end_time - start_time print(f"并发处理耗时: {concurrent_duration:.3f} 秒") print(f"总发言数: {len(controller.debate_history)}") print("✅ 性能测试通过") return True def run_comprehensive_test(): """运行综合测试""" print("🎭 优化的辩论流程控制系统 - 综合测试") print("=" * 60) test_functions = [ ("基础流程控制", test_basic_flow_control), ("阶段转换", test_stage_transition), ("发言者选择策略", test_speaker_selection_strategies), ("发言请求系统", test_speaker_request_system), ("上下文感知选择", test_context_aware_selection), ("阶段指标", test_stage_metrics), ("自适应模式", test_adaptive_mode), ("事件系统", test_event_system), ("数据持久化", test_data_persistence), ("性能测试", test_performance) ] passed = 0 failed = 0 for test_name, test_func in test_functions: try: print(f"\n{'='*20} {test_name} {'='*20}") result = test_func() if result: print(f"✅ {test_name} - 通过") passed += 1 else: print(f"❌ {test_name} - 失败") failed += 1 except Exception as e: print(f"❌ {test_name} - 错误: {str(e)}") failed += 1 print("\n" + "=" * 60) print(f"📊 测试结果统计") print(f"通过: {passed}/{len(test_functions)} ({passed/len(test_functions)*100:.1f}%)") print(f"失败: {failed}/{len(test_functions)} ({failed/len(test_functions)*100:.1f}%)") if failed == 0: print("🎉 所有测试通过！优化的辩论流程控制系统运行正常。") else: print(f"⚠️ 有 {failed} 个测试失败，需要进一步优化。") return passed, failed if __name__ == "__main__": run_comprehensive_test()