Function test_react_mode_trajectory

tests/test_trajectory.py:94–145

测试 ReAct 模式轨迹

()

    log.info(f"✓ 已导出: {filepath}")
    
def test_react_mode_trajectory():
    """测试 ReAct 模式轨迹"""
    log.info("=" * 60)
    log.info("测试 2: ReAct 模式轨迹")
    log.info("=" * 60)
    
    manager = TrajectoryManager()
    manager.start_recording(inputs={"query": "计算 123 + 456"})
    
    collector = manager.get_collector()
    
    # ReAct 循环
    for i in range(2):
        collector.on_node_start(f"agent_step_{i}", StepRole.AGENT.value)
        
        # 思考
        collector.on_thought(f"我需要使用计算器工具来计算 123 + 456")
        
        # 动作
        collector.on_action(
            action_type=ActionType.TOOL_CALL.value,
            action_payload={
                "tool_name": "calculator",
                "tool_args": {"expression": "123 + 456"}
            }
        )
        
        # 观察
        collector.on_observation("计算结果: 579")
        
        collector.on_node_end(output={"result": 579})
    
    # 创建状态
    request = DFRequest(target="计算 123 + 456")
    final_state = DFState(request=request)
    final_state.agent_results = {
        "agent": {"results": {"answer": "579"}}
    }
    
    trajectory = manager.stop_recording(final_state, "react_workflow")
    
    # 验证 ReAct 特征
    assert any(step.thought for step in trajectory.steps)
    assert any(step.observation for step in trajectory.steps)
    assert trajectory.mode == "react"
    
    log.info(f"✓ 模式: {trajectory.mode}")
    log.info(f"✓ 包含思考: {any(step.thought for step in trajectory.steps)}")
    
    # 导出 SFT 格式
    filepath = manager.export(trajectory, format="sft")
    log.info(f"✓ 已导出 SFT 格式: {filepath}")
    
def test_batch_export():
    """测试批量导出"""