Function main

internbootcamp/utils/data_postprocess.py:853–903

命令行接口示例

()

# ============= 命令行接口 =============

def main():
    """命令行接口示例"""
    import argparse
    
    parser = argparse.ArgumentParser(description="数据后处理工具")
    parser.add_argument("input", help="输入 jsonl 文件路径")
    parser.add_argument("output", nargs='?', default=None, help="输出 jsonl 文件路径（可选，不提供则自动生成）")
    parser.add_argument("--filter-success", action="store_true", help="只保留成功的样本")
    parser.add_argument("--min-score", type=float, default=0.9, help="最小分数")
    parser.add_argument("--max-score", type=float, default=1.0, help="最大分数")
    parser.add_argument("--data-source", type=str, help="按数据源过滤")
    parser.add_argument("--filter-design-end", action="store_true", help="只保留最后一条消息包含 <Design End> 的样本")
    parser.add_argument("--extract-training", action="store_true", help="提取训练数据格式")
    parser.add_argument("--extract-multimodal-training", action="store_true", help="提取多模态 SFT 训练格式")
    parser.add_argument("--extract-xtuner-strict-training", action="store_true", help="提取 XTuner 严格字段训练格式")
    parser.add_argument("--extract-messages", action="store_true", help="只提取消息")
    parser.add_argument("--expand-messages-prefixes", action="store_true", help="将多轮对话展开为前缀集")
    
    args = parser.parse_args()
    
    # 创建处理器
    processor = DataPostProcessor()
    
    # 添加过滤器
    if args.filter_success:
        processor.add_filter(filter_by_success, name="success")
    
    if args.min_score > 0.0 or args.max_score < 1.0:
        processor.add_filter(filter_by_score(args.min_score, args.max_score), name="score")
    
    if args.data_source:
        processor.add_filter(filter_by_data_source(args.data_source), name="data_source")
    
    if args.filter_design_end:
        processor.add_filter(filter_by_design_end, name="design_end")
    
    # 添加转换器(需要按顺序添加，因为有些转换函数会修改数据)
    if args.expand_messages_prefixes:
        processor.add_transformer(expand_messages_prefixes, name="expand_messages_prefixes")
    if args.extract_training:
        processor.add_transformer(extract_for_training, name="training_format")
    if args.extract_multimodal_training:
        processor.add_transformer(extract_for_multimodal_training, name="multimodal_training_format")
    if args.extract_xtuner_strict_training:
        processor.add_transformer(extract_for_xtuner_strict_training, name="xtuner_strict_training_format")
    if args.extract_messages:
        processor.add_transformer(extract_messages_only, name="messages_only")

    
    # 执行处理
    processor.process(args.input, args.output)


if __name__ == "__main__":