diff --git a/test/config.yaml b/test/config.yaml index 7ac32f484..a8b30af57 100644 --- a/test/config.yaml +++ b/test/config.yaml @@ -7,15 +7,15 @@ reports: filename: "report.html" title: "UCM Pytest Test Report" -database: - backup: "results/" - enabled: true - host: "127.0.0.1" - port: 3306 - name: "ucm_pytest" - user: "root" - password: "123456" - charset: "utf8mb4" +# database: +# backup: "results/" +# enabled: true +# host: "127.0.0.1" +# port: 3306 +# name: "ucm_pytest" +# user: "root" +# password: "123456" +# charset: "utf8mb4" # LLM Connection Configuration llm_connection: @@ -24,4 +24,8 @@ llm_connection: tokenizer_path: "/home/models/QwQ-32B" stream: true # stream output ignore_eos: true # Ignore the returned terminator - timeout: 180 # request time out \ No newline at end of file + timeout: 180 # request time out + + # Offline inference configuration (for accuracy tests) + model_path: "/home/models/DeepSeek-V2-Lite" + ucm_storage_dir: "/home/share/qyh-test" \ No newline at end of file diff --git a/test/conftest.py b/test/conftest.py index 150257952..ea1dd922c 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -34,6 +34,8 @@ def pytest_collection_modifyitems(config, items): markers = [m.split(":", 1)[0].strip() for m in config.getini("markers")] for name in markers: + if name == "forked": + continue opt = config.getoption(f"--{name}", "").strip() if not opt: continue diff --git a/test/suites/E2E/prompt.json b/test/suites/E2E/prompt.json new file mode 100644 index 000000000..d2eccf01b --- /dev/null +++ b/test/suites/E2E/prompt.json @@ -0,0 +1 @@ +{"input": "全国美国文学研究会的第十八届年会在哪所大学举办的？", "context": "全国美国文学研究会\n受秘书处委托，由我向美文会会员单位的各位代表简单汇报一下全国美国文学研究会自上届（第十七届）年会召开以来所做的工作。美文会秘书处刚刚完成了教育部社团办“关于在教育部主管社会组织中开展调研工作的通知”中要求提交的“全国美国文学研究会调研报告”（2016年11月），主要内容4项：我想就把我们提交的“调研报告”中1、2两部分中的部分内容，作为我向大会汇报的“全国美国文学研究会2014-2016年工作总结”的内容。\n1、研究会现状和基本情况\n美文会现有会员单位127个（不招收个人会员），包括国内主要985与211高校，以及中国社科院等科研单位和知名出版社。会长单位是南京大学，副会长单位是南京大学，中国社会科学院，北京外国语大学，北京大学，复旦大学，山东大学，秘书处设在南京大学外国语学院，秘书长、副秘书长是南京大学赵文书、何宁。美文会正式成立于1979年7月，是我国改革开放后成立最早的高校外国文学研究机构。1992年8月18日在民政部正式注册登记，获颁《中华人民共和国社会团体登记证》。美文会挂靠南京大学，财务由南京大学财务处负责，接受南京大学审计处审计，按民政部要求，每年参加年检，年检结果均为“通过”。\n美文会秘书处聘有专职秘书，工作人员7人，包括会长，副会长，秘书长，副秘书长，常务理事等。美文会设有党小组，隶属外国语学院英语系党支部，由会长任党小组长，成员包括副会长、常务理事、副秘书长、以及参加秘书处工作的青年教师与博士生。美文会发行《全国美国文学研究会通讯》（CASAL Newsletter），现已刊出33期。美文会每年轮流召开年会和专题研讨会，迄今已经举办17届年会和11届专题研讨会。\n2、近两年主要工作和取得的成效\n1）上届年会。美文会第十七届年会于2014年10月24日至26日在中国人民大学苏州校区举行，由中国人民大学外国语学院承办。来自全国28个省市自治区175所高等院校、研究所、出版社的348名正式代表参加了该次年会。年会的主题是“全球化语境中的美国文学研究：理论与实践”，收到论文全文164篇，摘要273篇。会议期间，全国美国文学研究会第七届理事会召开第三次会议。会议讨论通过了增补美文会副会长、常务理事、理事、会员单位事宜。会议再次明确，两期不缴纳会费的单位视为自动退出。理事和常务理事连续两次无故不参加理事会会议自动取消理事和常务理事。\n2、上届专题研讨会。美文会第十一届专题研讨会于2015年10月23至25日在徐州江苏师范大学举行，由江苏师大外国语学院承办。专题研讨会的主题是“美国文学中的城市”。来自全国21个省市自治区125所高等院校、研究所、出版社的88名正式代表参加了本次研讨会。收到论文全文54篇，论文摘要73篇。会议期间，美文会召开第七届理事会第三次会议，讨论通过了美文会秘书处的提议，增补何宁为理事兼任副秘书长，提请本届年会的理事会确认。\n3、业务活动。1）继续举办“全国美国文学研究会学术成果奖”评选。美文会设立此奖项，是为了促进我国美国文学研究的繁荣与发展，每5年评选一次，迄今已经评选3次。第三届成果奖评选出“优秀专著奖”14项，在2015年10月公示，“优秀论文奖”9项，“优秀教材奖”3项，“优秀译作奖”1项。该活动不收取任何费用。2）第十七届美国戏剧研究年会。由南京师范大学外国语学院承办，2015年7月21-22日在南京举办，主题是“20-21世纪之交美国戏剧研究”。\n4、年检情况。美文会接受业务主管单位教育部的业务指导和社团管理机关民政部社团的监督管理，执行《民间非营利组织会计制度》，接受南京大学财务处、审计处的管理和督查，接受“江苏兴瑞会计师事务所有限公司”财务审计，结果报教育部、民政部。2015年3月进行年检，编制2015年度美文会工作报告书。8月民政部“中国社会组织网”公布年检结果：“合格”。\n2016年11月25日晚，全国美国文学研究会召开理事会，讨论了如下事项。\n会议申办：\n1. 19届年会（2018，浙江大学）和第12届专题研讨会（2017，河海大学），主题尚没有最终确定。\n2. 哈尔滨工业大学申办第20届年会（2020）\n新申请理事单位：\n1. 中国矿业大学，推荐王丽明副教授担任理事\n2. 哈尔滨工业大学，推荐刘克东院长任理事\n3. 南京大学，推荐何宁（美文会副秘书长）任理事\n会员单位变更：\n1. 解放军国际关系学院按照相关要求退出全国美国文学研究会，因此不再常务理事单位，方成教授退出常务理事\n2. 对外经贸大学英语学院英美文学研究所长金冰接替孙建秋担任理事\n3. 中央民族大学外国语学院朱小琳接替郭英剑担任理事\n4. 厦门理工大学张跃军担任理事（原为中南大学常务理事）\n5. 上海外语教育出版社孙静接替汪义群担任理事\n6. 黑龙江大学推荐徐文培为常务理事\n美国族裔文学研究：空间拓展与界域重绘\n全国美国文学研究会第十八届年会（2016）纪要\n2016年11月25日至28日，全国美国文学研究会第十八届年会在厦门大学举行。本届年会由厦门大学外文学院承办，来自全国各地180余所高等院校、研究所、出版社的296位正式代表参加了会议。大会组委会共收到论文全文127篇、摘要371篇，与会代表围绕大会主题“美国族裔文学研究：空间拓展与界域重绘”（Ethnic Studies in US: Extending Interspace and Redefining Typology）展开了广泛而深入的研讨。\n11月25日晚，全国美国文学研究会召开常务理事和理事会，共有23位常务理事和理事出席，理事会主要讨论通过了以下议题：\n1. 确认2017年的专题研讨会由河海大学承办，河海大学外国语学院院长蔡斌教授在闭幕式作简单介绍。\n2. 确认2018年第十九届年会由浙江大学承办。\n3. 因政策变化，解放军国际关系学院退出美文会，方成教授不再担任常务理事。\n4. 增补黑龙江大学徐文培教授为常务理事。\n5. 增补哈尔滨工业大学（刘克东教授）、北京航空航天大学（田俊武教授）、中国矿业大学（王丽明副教授）、北京联合大学（黄宗英教授）为理事单位，山东大学李保杰教授、对外经济贸易大学金冰教授（接替孙建秋教授）、中央民族大学朱小琳教授（接替郭英剑教授）、上海外语教育出版社孙静（接替汪义群）为理事。\n6. 重新明确会员单位申请原则。美文会实行单位会议制，欢迎尚未加入协会的单位申请加入。申请方法和申请表格可以从美文会官网上下载。填写后加盖单位公章邮寄到协会秘书处。美文会秘书处收到入会申请并收到会员费之后即通报理事会并确认会员单位资格。\n7. 重新明确理事单位申请条件。第一，理事单位必须是正常缴纳会费的会员单位；第二，原则上需有英语语言文学硕士点；第三，符合以上条件单位可以申请成为美文会理事单位并推荐合适人选担任理事。\n8. 理事会决定，在美文会的年会和专题研讨会上评选会议优秀论文并颁发证书。其中，优秀论文仅在向会议提交的论文全文（未发表）中评选；作者所在单位须为美文会员单位，在向会议提交论文时，注明论文未经发表，并注明申请参加会议优秀论文评选；美文会常务理事以上（含）不参加申请。\n9. 关于本次年会优秀论文评选：已向会议提交未发表论文全文的会员单位参会代表，在12月15日前，向本会秘书处提交修改后的论文申请评选，本会将在寒假组织评选，在2017年3月公布评选结果并颁发证书。\n11月26日上午，本届年会开幕式在厦门大学科艺报告厅举行。厦门大学校长助理张建霖教授，外文学院张龙海院长，全国美国文学研究会会长朱刚教授、副会长盛宁教授、郭继德教授、杨仁敬教授、金莉教授、王守仁教授、张冲教授、申富英教授，秘书长赵文书教授及其他与会代表出席了开幕式。\n开幕式由外文学院副院长李美华教授主持。张建霖校长助理首先代表厦门大学对来自全国各地的与会者表示热烈欢迎，并对全国美国文学研究会第十八届年会的顺利召开表示衷心祝贺。外文学院张龙海院长代表承办方致欢迎辞，向与会者介绍了厦大外文学院的人才培养、学术研究等情况，以及年会的准备情况。全国美国文学研究会朱刚会长代表与会人员感谢厦门大学对本届年会的大力支持。朱会长简要回顾了美文会的历史和现状，并向与会代表汇报了研究会自第十七届年会以来的主要工作。最后，朱刚会长感谢全体参会代表及承办方对美文会工作的大力支持和对共同推动美国文学研究所做出的贡献，并对今后的工作提出了殷切希望。\n本届年会共分为大会发言、小组讨论、专题研讨（panel discussion）及研究生学术论坛四个部分。11月26日上午的大会发言分别由美文会副会长金莉教授和美文会前副会长、南京大学王守仁教授主持，共有5位代表发言。\n中国社科院外国文学研究所盛宁教授的发言题目是《对政治正确的文化批评的再审视》。盛宁教授指出，美国总统大选造成的国内民族分裂愈演愈烈，这一新国情使我国的族裔文学研究更具价值和意义。作为学者我们必须凸显自己的立场和价值判断，对少数族裔文学的审美价值要有清晰的认识。盛教授以第一代华裔作家代表汤婷婷和第二代华裔作家代表哈金为例，评析了两代作家迥异的“政治正确”书写策略。他认为，借助“政治正确”发音的族裔文学的审美价值会很快消失，我们应深刻反思非裔作家的代表――托尼・莫里森――的创作遗产。莫里森不只着眼于描写黑人苦难，更深入探索人性，将黑人作为“人性”的缩影进行刻画，这是她能够进入美国文学传统、流芳传世的重要原因。\n复旦大学外文学院张冲教授以《超越族裔：美国族裔文学研究的几点思考》为题，探讨我国当前族裔文学研究面临的困境及出路。张冲教授指出，国内族裔文学研究仍然面临研究角度单一与模仿、研究方法过于“理论导向”、文本“碎片化”释读等问题。他建议可从“族裔文学发展流变史”、“比较族裔文学史”以及“本土裔与中国文学文化比较”等维度，重新思考我国方兴未艾的族裔文学研究，族裔文学研究应努力超越族裔而回归文学，既要思考族裔文学的“族裔性”也需关注其“文学性”。\n在《再议作家的族裔身份问题：本质主义与自由选择》的发言中，上海外国语大学虞建华教授以斯图亚特・霍尔对“身份”的定义为出发点，对现有族裔身份的归置基准进行拷问。虞教授强调，在讨论族裔作家文化身份时，我们需聚焦常被忽视的身份的表演性和叙事性，应以社会建构理论为指导思路，走出本质主义，作家的族裔身份在全球化大势下的多元社会，应被看作一个动态、临时、杂糅的建构过程。\n南京大学英语系朱雪峰副教授的发言《重组芝加哥：拉图尔ANT理论视阈下的<克莱伯恩公园>》以社会学家布鲁诺・拉图尔的“行动者网络理论”为视角，从“流动的城市”、“行动者网络”、“蚂蚁视角与新现实主义”三个层面审视《克莱伯恩公园》中的芝加哥城市再现。朱教授认为，此剧在美国本土政治正确风潮中的接受悖论正在于它如实近距离描述了芝加哥城市地理在互动中流变的复杂性，其政治相关性在于它没有给出一个关于芝加哥社会的明晰解释或批评，而是通过不断追踪新问题联合来重组社会，以貌似传统的新现实主义风格体现了戏剧价值。\n厦门大学外文学院张龙海教授以《美国少数族裔文学研究在中国》为题，向大家勾勒了我国美国族裔文学研究的历史图景。张教授通过大量的文献研究和详细的数据，从研究的规模、研究队伍的状况、期刊报纸的刊登情况以及研究中出现的不平衡等方面详细探析美国少数族裔文学研究在中国的涌现和繁荣发展。\n11月26日下午，年会设立23个分会场进行小组讨论。代表们围绕“华裔文学研究新视野”、“亚裔文学研究新视野”、“非裔文学研究新视野”、 “犹太裔文学研究新视野”、“拉美裔文学研究新视野”、“印第安裔文学研究新视野”、“族裔文学与性别研究”、“族裔文学批评理论新动向”、“少数族裔与多元文化”、“族裔文学研究中的中国视角”、“美国文学理论研究与教学”、“美国现代派文学研究”、“早期美国文学研究”等议题，对美国族裔文学展开了多层次全方位的探讨。\n第一组（专题讨论：族裔成长小说研究）由方红、芮渝萍主持，发言人有方红（南京大学）“消声、言说与成长：《褐姑娘、褐砖房》研究”；侯金萍（华南农业大学）“华裔美国文学对成长小说的改写与创新”；芮渝萍（宁波大学）“美国华裔成长小说的特点”；谭岸青（暨南大学）“解读任碧莲《世界与小镇》的成长书写”；邹惠玲（江苏师范大学）“《飞逸》：在自省与融合之中成长”。\n第二组（华裔文学研究新视野之一）由刘永杰、戴鸿斌主持，发言人有黄明（商丘师范学院）“严歌苓小说《扶桑》对华人形象的颠覆”；霍盛亚（北京外国语大学）“华裔美国科幻作家刘宇昆小说的“复族裔化”倾向”；刘向辉（许昌学院）“谭恩美小说《喜福会》中的文学地图与民族记忆”；刘永杰（郑州大学）“‘秘密’的真相：《蝴蝶君》主人公断袖之谊探析”；史博（华北科技学院）“解读《折纸》中爱的主题”；孙坚（陕西师范大学）“新历史主义关照下的《中国佬》”；颜碧洪（福建师范大学福清分校）“论汤亭亭《中国佬》的后现代主义书写”。\n第三组（华裔文学研究新视野之二）由郭栖庆、金衡山主持，发言人有黄一畅（南京航空航天大学）“虚构的权威―《谁是爱尔兰人？》中的叙事伦理之辨”；季峥（重庆工商大学）“华裔美国作家入典原因探究”；金衡山（华东师范大学）“The Puzzling and Enlightening Racial Identity in Who’ s Irish？”；苏娉（中山大学）“论李翊云的非母语写作及其意义”；王芳（中央民族大学）“《无声告白》中的华裔精神生存困境探析”；王增红（厦门大学）“种族冒充、冒充叙事与混血族身份政治―威妮弗蕾德•伊顿新解”；姚红艳（武汉大学）“族群记忆、族群认同与身份建构―《接骨师之女》中的仪式书写”；周凌敏（南方医科大学）“以物为导向的本体论下的后人文主义―以《咸鱼女孩》为例”。\n第四组（族裔文学与性别研究之一）由王玉括、田俊武主持，发言人有方小莉（四川大学）“20世纪黑人女性小说叙述策略研究”；李蕊（南京大学）“论《他们眼望上苍》中珍妮的‘生成女性’特质”；毛艳华（浙江大学）“性别‘引用’视域下《秀拉》中女性主体的初现与重构”；隋红升（浙江大学）“汉斯伯里《太阳下的葡萄干》对美国男性气质的反思”；田俊武（北京航空航天大学）“回归之路―托尼•莫里森作品中的旅行叙事”；王玉括（南京邮电大学）“黑人女性主义文学批评述评”；杨艳春（哈尔滨石油学院）“生态女性主义视域下艾丽丝•沃克作品中女性族裔身份的自我认同”；朱海峰（东北师范大学）“父权、女权、后女权―论《钢琴课》中黑人的种族出路”。\n第五组（族裔文学与性别研究之二）由张跃军主持，发言人有董秋芳（广东农工商职业技术学院）“美国华裔女性主体身份流变―以华裔女作家英语创作为例”；刘兮颖（华中师范大学）“《卢布林的魔术师》中雅夏的身份危机与伦理选择”；杨静（广东外语外贸大学）“全球化时代的跨国婚姻：《追寻亚裔女性》”；姚丽梅（佳木斯大学）“论邝丽莎在《雪花秘扇》中的女性主义身份伦理观”；张跃军（厦门理工学院）“‘温和的女性主义’：华裔美国诗人陈美玲诗歌解读”；朱骅（上海海洋大学）“跨国主义的美国族裔文学建构”。\n第六组（犹太裔文学研究新视野）由刘文松主持，发言人有高莉敏（上海立信会计金融学院）“《末世之城》：大屠杀的历史记忆”；胡选恩（陕西师范大学）“E.L.多克托罗《大进军》中的历史阐释模式”；孔伟（北京外国语大学）“俄国犹太人的‘应许之地’―新移民叙事中的‘发声’策略研究”；刘文松（厦门大学）“美国犹太知识分子小说探秘”；孙璐（上海外国语大学）“菲利普•罗斯《美国牧歌》中的美国民族神话及其当代启示”；张国庆（中国人民大学）“《人性的污秽》的后人道主义解读”；赵永健（浙江工商大学）“国外美国犹太戏剧研究评述”。\n第七组（美国后现代派文学研究之一）由陈世丹、刘雪岚主持，发言人有杨仁敬（厦门大学）“略论《时间》与《达洛威夫人》的互文性”；陈世丹（中国人民大学）“后现代文学伦理学批评要义”；曾艳钰（湖南科技大学）“‘流动的爱国主义盛宴’―评美国后现代战争小说”；谷红丽（华南师范大学）“后现代主义历史叙事”；刘雪岚（社会科学院外国文学研究所）“从‘加州三部曲’看托马斯•品钦的后现代城市书写”；方凡（浙江大学）“论威廉•加斯笔下的图像与文字”；王祖友（泰州学院）“后人道主义与人道主义辨析”；陈奔（厦门大学）“美国研究背景下的后现代主义文学研究”；范小玫（厦门大学外）“德里罗小说中的全球化”。\n第八组（美国后现代派文学研究之二）由吴泽庆、陈俊松主持，发言人有陈俊松（华东师范大学）“《地下世界》：冷战阴云的文化记忆与后现代恐怖叙事”；许希夷（南京大学）“福尔‘后9/11’小说《特别响，非常近》中的历史叙事”；史菊鸿（兰州大学）“一个城市，两幅画面――库切和詹姆斯对伦敦的不同文学再现”；吴泽庆（中央民族大学）“‘恶魔的诅咒’―欧茨的《被诅咒的》中历史书写”；姚本标（广西师范学院）“《白噪音》的‘风险社会’表征”；栾天宇（南京大学）“《赛姆勒先生的行星》中的记忆伦理与美国20世纪60年代”。\n第九组（美国后现代派文学研究之三）由甘文平、杨纪平主持，发言人有甘文平（武汉理工大学）“米歇尔•福柯、共同体、美国越战文学”；崔永光（大连海洋大学）“世界文学史视域中的纳博科夫形象及其创作密码”；范湘萍（上海政法学院）“论‘9.11文学’结构主义叙事中的空间与政治”；林莉（东北师范大学）“论小说《恶棍来访》的空间叙事策略”；刘丹（大连外国语大学）“融合与分裂：《地下世界》中的种族冲突与文化政治”；王程辉（湖南科技大学）“纳博科夫《国王、王后和杰克》与福楼拜《包法利夫人》的互文性”；杨纪平、胡燕（北京邮电大学）“《X战警：第一战》中的族裔观”；张芳芳（上海电力学院）“论纳博科夫小说《普宁》中‘坐错车’的隐喻与流亡主题”；张蓝予（中央民族大学）“文明对话与身份认同：评《恐怖分子》的身份观念”。\n第十组（拉美裔文学研究新视野）由李保杰、李毅峰主持，发言人有李保杰（山东大学）“当历史的重负成为过去―《古巴之王》中的‘反流亡’书写”；李毅峰（天津商业大学）“桑德拉•西斯内罗斯对女性原型形象的重新阐释”；乔玲玲（山西大同大学）“芒果街上的奇卡纳游荡者”；涂沙丽（中南民族大学）“论《石化鹿》中的奇卡娜形象”；王绵绵（浙江传媒学院）“加勒比裔美国移民女作家的空间意识及空间策略”。\n第十一组（美国文学理论研究与教学）由郭建辉、刘春芳主持，发言人有陈 Q（中央民族大学）“论当代反本质主义文学理论的发生因缘与中国进程”；郭建辉（四川外国语大学期刊社）“英美文学教学与审美教育”；焦敏（广东外语外贸大学）“人文主义与戏剧教学”；刘春芳（山东工商学院）“美国浪漫主义文学中的平民思想”；马特（中央财经大学）“文学批评的空间转向：空间批评的新动向”；许玉军（集美大学）“东方启蒙：西方的‘东方主义’话语”。\n第十二组（族裔文学批评理论新动向）由胡铁生、郭英剑主持，发言人有胡铁生（吉林大学）“美国少数族裔文学的演进”；郭英剑（中国人民大学）“2015美国文学：种族，还是种族问题”；洪琪（湖北第二师范学院）“美国华裔戏剧的创伤叙事”；任虎军（四川外国语大学）“性别视阈下新世纪中国的美国族裔小说研究”；王斐（集美大学）“追寻都会中的空间正义：美国非裔城市叙事嬗变初探”。\n第十三组（美国现代派文学研究之一）由黄宗英、王跃洪主持，发言人有陈秋红（青岛大学）“亨利•詹姆斯后期小说的进化叙述”；陈喜华（湘潭大学）“菲茨杰拉德的服饰书写与爵士时代美国文化”；黄宗英（北京联合大学）“‘其城/其人，一种身份’：读威廉斯的《帕特森》”；蒋贤萍（西北师范大学）“表演的自我――再读《进入黑夜的漫长旅程》”；李晶（中南财经政法大学）“生存还是生活？：凯瑟《一个迷途的女人》的伦理选择”；陶久胜（南昌大学）“无意识的种族偏见――《上帝的儿女都有翅膀》的心理原型解读”；朱晓萍（贵州大学）“追逐无的欲望――《嘉莉妹妹》的拉康式解读”。\n第十四组（美国现代派文学研究之二）由李建波、朴玉主持，发言人有朴玉（吉林大学）“科伦•麦凯恩在《光明这一面》中的城市创伤叙事”；王晓丹（哈尔滨师范大学）“阶层流动的幻灭：《纹身女孩》中的社会身份”；王跃洪、郝天昕（上海理工大学）“福柯凝视理论视角下的亨利•詹姆斯《德莫福夫人》研究”；薛丽（北京师范大学）“《布拉迪默传奇》中矛盾的女性意识形态”；姚学丽（安徽大学）“映射美国南方的‘隐约轮廓’――析《干旱的九月》碎片化叙事”；张金良（天津外国语大学）“哈贝马斯有效沟通视域下《奥利安娜》中的交流困境分析”；张小平（扬州大学）“旅行•幻梦•混沌――论麦卡锡小说《骏马》中的‘奇异吸引子’”。\n第十五组（早期美国文学研究）由张和龙、金冰主持，发言人有金冰（对外经贸大学）“美国自然主义文学的进化叙事与伦理想像”；李晋（中南财经政法大学）“19世纪美国文学市场研究综述”；李敏（山东工商学院）“《红字》通奸案的法、罚与霍桑的‘疼痛’书写”；李方木（北京外国语大学）“爱的伪装：《献给爱米丽的玫瑰》中的罗曼司及其多义性”；戚涛（安徽大学）“多数暴政下的碎片――梅尔维尔的价值困惑与身份建构”；张和龙（上海外国语大学）“中国杰克•伦敦研究中的话语模式及其历史嬗变”。\n第十六组（非裔文学研究新视野之一）由林元富主持，发言人有陈红（广东外语外贸大学）“‘所有的故事都是真的’―评怀德曼的编史元小说《法农》”；甘婷（集美大学）“第三空间理论视阈下《中间通道》的空间建构”；林元富（福建师范大学）“当代非裔美国涉奴题材小说的历史传承”；刘锦丽（湖北科技学院）“种族歧视下的身份困惑―论切斯纳特小说中的混血儿”；龙跃（湖南师范大学）“兰斯顿•休斯诗歌中的‘黑人性’”；吕春媚（大连外国语大学）“黑白的空间对峙――解读《莱尼大妈的黑臀舞》中的社会空间”；王予霞（集美大学）“美国黑人左翼文学消长的历史启示”；修树新（东北师范大学外国语学院）“论特瑞•麦克米兰小说中爱的主题”；张健然（四川外国语大学）“《他们眼望上苍》中原始性与现代性的背离与融合”。\n第十七组（非裔文学研究新视野之二）由徐文培、杜志卿主持，发言人有杜志卿（华侨大学外国语学院）“从霍妮的精神分析理论看阿契贝笔下的伊祖鲁”；蒯冲（荆楚理工学院）“非裔美国人的身份缺失与身份认同――以《阳光下的葡萄干》为例”；李美芹（浙江工商大学）“论埃里森‘文学爵士乐’美学中表达的种族政治思想”；李云瑾（华北科技学院）“论托妮•莫里森《宣叙》的含混性”；马粉英（西北师范大学）“《最蓝的眼睛》中克劳迪娅拆解行为的后殖民叙事”；唐莹（大连外国语大学）“从‘挪亚的诅咒’到非洲主义―对罗宾逊种族书写的反思”；徐文培（黑龙江大学）“《所罗门之歌》与奴隶叙事文学”；张宏薇（东北师范大学）“莫里森早晚期两部小说中‘儿童创伤’主题的对比分析”；朱小琳（中央民族大学）“悲莫悲兮伤永逝：以墓志为叙事策略的《宠儿》新解”。\n第十八组（亚裔文学研究新视野）由谷红丽、李汝成主持，发言人有李青霜（南京审计大学）“论戏剧《耻辱》中的穆斯林文化定势”；刘喜波（齐齐哈尔大学）“列斐伏尔的身体空间理论下的《灿烂千阳》解读”；李东风（盐城师范学院）“美国印度裔离散文学中的‘家叙事’”。", "answers": ["厦门大学。"], "length": 9593, "dataset": "multifieldqa_zh", "language": "zh", "all_classes": null, "_id": "5b1b8e937b83c3ff9b75ac386fae9c4575c4b9f26a4fbdad"} diff --git a/test/suites/E2E/test_uc_accuracy_offline.py b/test/suites/E2E/test_uc_accuracy_offline.py new file mode 100644 index 000000000..321c63274 --- /dev/null +++ b/test/suites/E2E/test_uc_accuracy_offline.py @@ -0,0 +1,824 @@ +""" +NOTE: Each test case should run with multiprocessing spawn mode to ensure GPU memory +is fully released after each test. This prevents memory accumulation across test cases. +""" + +import pytest +import yaml +import time +import os +import contextlib +import gc +import json +import tempfile +import multiprocessing +import subprocess +import sys +from functools import wraps +from pathlib import Path +from typing import List, Dict, Any, Tuple, Optional +from dataclasses import asdict + +from common.capture_utils import export_vars +from transformers import AutoTokenizer + +_test_functions = {} + + +def _run_test_in_spawn_process(test_id, args, kwargs, result_queue, error_queue): + os.environ["_IN_SPAWN_PROCESS"] = "1" + try: + test_func = _test_functions.get(test_id) + if test_func is None: + raise RuntimeError(f"Test function {test_id} not found") + result = test_func(*args, **kwargs) + result_queue.put(("success", result)) + except Exception as e: + error_queue.put(("error", e)) + + +def run_in_spawn_process(func): + # 注册测试函数到全局字典，用于存储测试函数（避免 pickle 嵌套函数的问题） + test_id = f"{func.__module__}.{func.__name__}" + _test_functions[test_id] = func + + @wraps(func) + def wrapper(*args, **kwargs): + if os.environ.get("_IN_SPAWN_PROCESS") == "1": + return func(*args, **kwargs) + + ctx = multiprocessing.get_context("spawn") + result_queue = ctx.Queue() + error_queue = ctx.Queue() + + process = ctx.Process( + target=_run_test_in_spawn_process, + args=(test_id, args, kwargs, result_queue, error_queue) + ) + process.start() + process.join(timeout=3600) + + if process.is_alive(): + process.terminate() + process.join() + raise RuntimeError(f"Test {func.__name__} timed out after 1 hour") + + if not error_queue.empty(): + status, error = error_queue.get() + raise error + + if not result_queue.empty(): + status, result = result_queue.get() + return result + + if process.exitcode != 0: + raise RuntimeError(f"Test {func.__name__} failed in spawn process with exit code {process.exitcode}") + + return wrapper + + + +try: + from vllm import LLM, SamplingParams + from vllm.config import KVTransferConfig + from vllm.engine.arg_utils import EngineArgs + from ucm.logger import init_logger + import torch + VLLM_AVAILABLE = True +except ImportError: + VLLM_AVAILABLE = False + pytest.skip("vLLM not available", allow_module_level=True) + +logger = init_logger(__name__) + + +@contextlib.contextmanager +def build_llm_with_uc( + model_path: str, + ucm_config: Optional[Dict[str, Any]] = None, + enable_prefix_caching: bool = False, + **llm_kwargs +): + module_path = "ucm.integration.vllm.ucm_connector" + name = "UCMConnector" + + ktc = KVTransferConfig( + kv_connector=name, + kv_connector_module_path=module_path, + kv_role="kv_both", + kv_connector_extra_config=ucm_config, + ) + + if not os.getenv("CUDA_VISIBLE_DEVICES"): + os.environ["CUDA_VISIBLE_DEVICES"] = "4,5" + + tensor_parallel_size = 2 + + default_args = { + "model": model_path, + "kv_transfer_config": ktc, + "max_model_len": 32768, + "gpu_memory_utilization": 0.8, + "max_num_batched_tokens": 30000, + "block_size": 128, + "enforce_eager": True, + "trust_remote_code": True, + "enable_prefix_caching": enable_prefix_caching, + "tensor_parallel_size": tensor_parallel_size, + } + default_args.update(llm_kwargs) + + llm_args = EngineArgs(**default_args) + llm = LLM(**asdict(llm_args)) + + try: + yield llm + finally: + logger.info("LLM engine is exiting") + del llm + gc.collect() + + +def split_prompt_by_tokens( + prompt: str, + tokenizer: AutoTokenizer, + split_ratio: float = 0.5 +) -> Tuple[str, str]: + tokens = tokenizer.encode(prompt) + split_idx = int(len(tokens) * split_ratio) + + first_tokens = tokens[:split_idx] + second_tokens = tokens[split_idx:] + + first_part = tokenizer.decode(first_tokens, skip_special_tokens=False) + second_part = tokenizer.decode(second_tokens, skip_special_tokens=False) + + return first_part, second_part + + +def create_prompt_with_token_count( + base_prompt: str, + tokenizer: AutoTokenizer, + target_token_count: int +) -> str: + """Create a prompt with approximately target token count. + + Args: + base_prompt: Base prompt string to repeat/extend. + tokenizer: Tokenizer to count tokens. + target_token_count: Target number of tokens. + + Returns: + Prompt string with approximately target_token_count tokens. + """ + base_tokens = tokenizer.encode(base_prompt) + base_token_count = len(base_tokens) + + if base_token_count >= target_token_count: + # If base prompt is already long enough, truncate it + tokens = base_tokens[:target_token_count] + return tokenizer.decode(tokens, skip_special_tokens=False) + + # Calculate how many times to repeat + repeat_count = (target_token_count // base_token_count) + 1 + extended_prompt = (base_prompt + " ") * repeat_count + + # Trim to target length + tokens = tokenizer.encode(extended_prompt) + tokens = tokens[:target_token_count] + return tokenizer.decode(tokens, skip_special_tokens=False) + + +def load_prompt_from_file(prompt_file: Optional[Path] = None) -> Tuple[str, List[str]]: + """Load prompt and answers from JSON file (LongBench format). + + LongBench format structure: + { + "input": "任务输入/问题", + "context": "长上下文/文档", + "answers": ["答案列表"], + "length": 总长度, + "dataset": "数据集名称", + "language": "语言", + ... + } + + For LongBench, the typical format is: + - context: 长文档/上下文（放在前面） + - input: 问题/查询（放在后面） + - Combined format: context + "\n\n" + input + + Args: + prompt_file: Path to the prompt JSON file. If None, uses default path. + + Returns: + Tuple of (combined_prompt_string, answers_list). + - combined_prompt_string: Combined prompt (context + input) + - answers_list: List of standard answers from the file + """ + if prompt_file is None: + prompt_file = Path(__file__).parent / "prompt.json" + + if not prompt_file.exists(): + raise FileNotFoundError(f"Prompt file not found: {prompt_file}") + + with open(prompt_file, "r", encoding="utf-8") as f: + content = f.read().strip() + + try: + data = json.loads(content) + except json.JSONDecodeError as e: + raise ValueError(f"Invalid JSON format in {prompt_file}: {e}") + + if isinstance(data, list): + if len(data) == 0: + raise ValueError(f"Empty list in {prompt_file}") + data = data[0] # Use first item, or you can process all items + + # Extract input and context from LongBench format + input_text = data.get("input", "") + context_text = data.get("context", "") + + # LongBench standard format: context (long document) + input (question) + # Combine context and input to form the full prompt + # Format: context + "\n\n" + input + if context_text and input_text: + # Standard LongBench format: context first, then input + full_prompt = f"{context_text}\n\n{input_text}" + elif context_text: + # Only context available + full_prompt = context_text + elif input_text: + # Only input available + full_prompt = input_text + else: + raise ValueError(f"No input or context found in {prompt_file}") + + # Extract answers + answers = data.get("answers", []) + if not isinstance(answers, list): + answers = [answers] if answers else [] + + return full_prompt, answers + + +def run_inference( + llm: LLM, + prompts: List[str], + sampling_params: SamplingParams, + description: str = "", +) -> Tuple[List[str], float]: + """Run inference and return generated texts and elapsed time. + + Args: + llm: LLM instance. + prompts: List of prompt strings. + sampling_params: Sampling parameters. + description: Description for logging. + + Returns: + Tuple of (generated_texts, elapsed_time). + """ + start_time = time.time() + outputs = llm.generate(prompts, sampling_params) + elapsed_time = time.time() - start_time + + generated_texts = [] + for output in outputs: + generated_text = output.outputs[0].text + generated_texts.append(generated_text) + + if description: + print(f"[INFO] {description} completed in {elapsed_time:.2f}s") + + return generated_texts, elapsed_time + + +@pytest.mark.parametrize("model_path", [ + "/home/models/QwQ-32B", + "/home/models/DeepSeek-V2-Lite", +]) +@pytest.mark.parametrize("max_tokens", [100]) +@pytest.mark.feature("uc_accuracy_test_offline") +@export_vars +@run_in_spawn_process +def test_offline_accuracy_ssd_load( + model_path: str, + max_tokens: int, +): + """Test SSD load accuracy (Phase 1). + + Test flow: + 1. Phase 1.1: Disable HBM PC, send full prompt -> KV cache saved to SSD + 2. Phase 1.2: Disable HBM PC, load from SSD, send full prompt -> verify SSD load accuracy + + The prompt is loaded from prompt.json file (LongBench format). + + Args: + model_path: Path to the model. + max_tokens: Maximum tokens to generate. + """ + # Load configuration + config_file = Path(__file__).parent.parent.parent / "config.yaml" + with open(config_file, "r", encoding="utf-8") as f: + config = yaml.safe_load(f) + + # Use model_path from parameter, fallback to config or environment + if not model_path or not os.path.exists(model_path): + model_path = config.get("llm_connection", {}).get("model_path") or os.getenv("MODEL_PATH") + if not model_path: + pytest.skip(f"model_path not configured or not found: {model_path}") + + if not os.path.exists(model_path): + pytest.skip(f"Model path does not exist: {model_path}") + + ucm_storage_dir = config.get("llm_connection", {}).get("ucm_storage_dir") or os.getenv("UCM_STORAGE_DIR", "/tmp/ucm_cache") + + try: + test_prompt, standard_answers = load_prompt_from_file() + print(f"[INFO] Loaded prompt from prompt.json (length: {len(test_prompt)} chars)") + if standard_answers: + print(f"[INFO] Standard answers: {standard_answers}") + else: + print(f"[INFO] No standard answers found in prompt.json") + except Exception as e: + pytest.skip(f"Failed to load prompt from prompt.json: {e}") + + # Setup tokenizer + tokenizer = AutoTokenizer.from_pretrained(model_path, use_chat_template=True) + + try: + messages = [{"role": "user", "content": test_prompt}] + formatted_full_prompt = tokenizer.apply_chat_template( + messages, + tokenize=False, + add_generation_prompt=True, + add_special_tokens=True, + ) + except Exception: + formatted_full_prompt = test_prompt + + ucm_config = { + "ucm_connectors": [ + { + "ucm_connector_name": "UcmNfsStore", + "ucm_connector_config": { + "storage_backends": ucm_storage_dir, + "use_direct": False, + }, + } + ], + "load_only_first_rank": False, + } + + sampling_params = SamplingParams( + temperature=0.0, + top_p=1, + max_tokens=max_tokens, + ) + + print(f"\n[INFO] ===== SSD Load Accuracy Test =====") + print(f"[INFO] Model: {model_path}") + print(f"[INFO] Full prompt length: {len(test_prompt)} chars") + print(f"[INFO] Max tokens: {max_tokens}") + print(f"[INFO] Temperature: 0.0 (deterministic)") + print(f"[INFO] UCM storage: {ucm_storage_dir}") + + # ===== Phase 1.1 and 1.2: Disable HBM PC, save KV cache to SSD and load ===== + print(f"\n[INFO] ===== Phase 1.1 and 1.2: Save KV Cache to SSD And Load =====") + print(f"[INFO] HBM Prefix Caching: DISABLED") + + with build_llm_with_uc( + model_path=model_path, + ucm_config=ucm_config, + enable_prefix_caching=False, # Disable HBM PC + ) as llm: + phase1_outputs, phase1_time = run_inference( + llm, [formatted_full_prompt, formatted_full_prompt], sampling_params, "First save then load" + ) + phase1_1_output = phase1_outputs[0] + phase1_2_output = phase1_outputs[1] + + # ===== Compare outputs ===== + print(f"\n[INFO] ===== Accuracy Test Results =====") + + # Compare Phase 1.1 vs Phase 1.2 (SSD load accuracy) + phase1_identical = phase1_1_output == phase1_2_output + # if not phase1_identical: + print(f"\n[INFO] ===== Phase 1: SSD Load Accuracy Test =====") + print(f"[INFO] Phase 1.1 (SSD save) output differs from Phase 1.2 (SSD load) output!") + print(f"[INFO] Phase 1.1 output:\n{phase1_1_output}") + print(f"[INFO] Phase 1.2 output:\n{phase1_2_output}") + + # Assert outputs are identical - test fails if any difference + assert phase1_identical, ( + f"SSD Load Accuracy Test Failed!\n" + f"See detailed output above for differences." + ) + + print(f"\n[INFO] SSD load accuracy test passed: outputs are identical") + + value_lists = { + "model_path": [model_path], + "model_name": [os.path.basename(model_path)], + "test_prompt_length": [len(test_prompt)], + "max_tokens": [max_tokens], + "phase1_identical": [1 if phase1_identical else 0], + "phase1_time": [phase1_time], + } + + return {"_name": "accuracy_test_offline_ssd_load", "_data": value_lists} + + +@pytest.mark.parametrize("model_path", [ + "/home/models/QwQ-32B", + "/home/models/DeepSeek-V2-Lite", +]) +@pytest.mark.parametrize("max_tokens", [200]) +@pytest.mark.parametrize("prompt_split_ratio", [0.5]) # Split prompt in half +@pytest.mark.feature("uc_accuracy_test_offline") +@export_vars +@run_in_spawn_process +def test_offline_accuracy_hbm_ssd_mixed( + model_path: str, + max_tokens: int, + prompt_split_ratio: float, +): + """Test HBM + SSD mixed hit accuracy (Phase 2). + + This test first runs Phase 1 to generate a baseline output, then tests Phase 2. + Test flow: + 1. Phase 1: Disable HBM PC, send full prompt -> KV cache saved to SSD (baseline) + 2. Phase 2: Enable HBM PC, send partial prompt (warm HBM), then send full prompt (hits both HBM and SSD) -> verify mixed hit accuracy + + The prompt is loaded from prompt.json file (LongBench format). + + Args: + model_path: Path to the model. + max_tokens: Maximum tokens to generate. + prompt_split_ratio: Ratio to split prompt for Phase 2 (0.5 = split in half). + """ + # Load configuration + config_file = Path(__file__).parent.parent.parent / "config.yaml" + with open(config_file, "r", encoding="utf-8") as f: + config = yaml.safe_load(f) + + # Use model_path from parameter, fallback to config or environment + if not model_path or not os.path.exists(model_path): + model_path = config.get("llm_connection", {}).get("model_path") or os.getenv("MODEL_PATH") + if not model_path: + pytest.skip(f"model_path not configured or not found: {model_path}") + + if not os.path.exists(model_path): + pytest.skip(f"Model path does not exist: {model_path}") + + ucm_storage_dir = config.get("llm_connection", {}).get("ucm_storage_dir") or os.getenv("UCM_STORAGE_DIR", "/tmp/ucm_cache") + + # Load prompt and answers from prompt.json file + try: + test_prompt, standard_answers = load_prompt_from_file() + print(f"[INFO] Loaded prompt from prompt.json (length: {len(test_prompt)} chars)") + if standard_answers: + print(f"[INFO] Standard answers: {standard_answers}") + else: + print(f"[INFO] No standard answers found in prompt.json") + except Exception as e: + pytest.skip(f"Failed to load prompt from prompt.json: {e}") + + # Setup tokenizer + tokenizer = AutoTokenizer.from_pretrained(model_path, use_chat_template=True) + + # Format prompt with chat template if available + try: + messages = [{"role": "user", "content": test_prompt}] + formatted_full_prompt = tokenizer.apply_chat_template( + messages, + tokenize=False, + add_generation_prompt=True, + add_special_tokens=True, + ) + except Exception: + formatted_full_prompt = test_prompt + + prompt_first_part, prompt_second_part = split_prompt_by_tokens( + formatted_full_prompt, tokenizer, split_ratio=prompt_split_ratio + ) + + ucm_config = { + "ucm_connectors": [ + { + "ucm_connector_name": "UcmNfsStore", + "ucm_connector_config": { + "storage_backends": ucm_storage_dir, + "use_direct": False, + }, + } + ], + "load_only_first_rank": False, + } + + sampling_params = SamplingParams( + temperature=0.0, + top_p=1, + max_tokens=max_tokens, + ignore_eos=False, + ) + + print(f"\n[INFO] ===== HBM + SSD Mixed Accuracy Test =====") + print(f"[INFO] Model: {model_path}") + print(f"[INFO] Full prompt length: {len(test_prompt)} chars") + print(f"[INFO] Max tokens: {max_tokens}") + print(f"[INFO] Temperature: 0.0 (deterministic)") + print(f"[INFO] UCM storage: {ucm_storage_dir}") + print(f"[INFO] Prompt split ratio: {prompt_split_ratio}") + + # ensure Phase 1 has already run and KV cache is saved to SSD + # ===== Phase 2: Enable HBM PC, test HBM + SSD mixed hit ===== + print(f"\n[INFO] ===== Phase 2: HBM + SSD Mixed Hit Test =====") + print(f"[INFO] HBM Prefix Caching: ENABLED") + print(f"[INFO] Sending partial prompt and full prompt together in one batch...") + print(f"[INFO] - Partial prompt (first {int(prompt_split_ratio*100)}%) will warm up HBM") + print(f"[INFO] - Full prompt will hit HBM for prefix + SSD for suffix") + + with build_llm_with_uc( + model_path=model_path, + ucm_config=ucm_config, + enable_prefix_caching=True, # Enable HBM PC + ) as llm: + # Send both partial and full prompts together in one batch + # This ensures that partial prompt warms up HBM, then full prompt hits both HBM and SSD + phase2_outputs, phase2_time = run_inference( + llm, [prompt_first_part, formatted_full_prompt], sampling_params, "Phase 2 (HBM + SSD mixed)" + ) + phase2_partial_output = phase2_outputs[0] # Output from partial prompt (for reference) + phase2_full_output = phase2_outputs[1] # Output from full prompt (this is what we compare) + + # ===== Compare outputs ===== + print(f"\n[INFO] ===== Accuracy Test Results =====") + + # Compare Phase 1.1 vs Phase 1.2 (SSD load accuracy) + phase1_identical = phase1_1_output == phase1_2_output + if not phase1_identical: + print(f"\n[ERROR] ===== Phase 1: SSD Load Accuracy Test FAILED =====") + print(f"[ERROR] Phase 1.1 (SSD save) output differs from Phase 1.2 (SSD load) output!") + print(f"[ERROR] Phase 1.1 output:\n{phase1_1_output}") + print(f"[ERROR] Phase 1.2 output:\n{phase1_2_output}") + + phase2_identical = phase1_1_output == phase2_full_output + if not phase2_identical: + print(f"\n[ERROR] ===== Phase 2: HBM + SSD Mixed Accuracy Test FAILED =====") + print(f"[ERROR] Phase 1.1 (SSD save) output differs from Phase 2.2 (HBM + SSD mixed) output!") + print(f"[ERROR] Phase 1.1 output:\n{phase1_1_output}") + print(f"[ERROR] Phase 2.2 output:\n{phase2_full_output}") + + # Assert outputs are identical - test fails if any difference + assert phase1_identical, ( + f"SSD Load Accuracy Test Failed!\n" + f"See detailed output above for differences." + ) + + assert phase2_identical, ( + f"HBM + SSD Mixed Accuracy Test Failed!\n" + f"See detailed output above for differences." + ) + + print(f"\n[INFO] ✓ HBM + SSD mixed accuracy test passed: outputs are identical") + + # Prepare data for export + value_lists = { + "model_path": [model_path], + "model_name": [os.path.basename(model_path)], + "test_prompt_length": [len(test_prompt)], + "max_tokens": [max_tokens], + "prompt_split_ratio": [prompt_split_ratio], + "phase2_identical": [1 if phase2_identical else 0], + "phase2_time": [phase2_time], + } + + return {"_name": "accuracy_test_offline_hbm_ssd_mixed", "_data": value_lists} + + +@pytest.mark.parametrize("model_path", [ + "/home/models/QwQ-32B", + "/home/models/DeepSeek-V2-Lite", +]) +@pytest.mark.parametrize("base_prompt", [ + "This is a test prompt for chunk prefill accuracy testing. ", +]) +@pytest.mark.parametrize("max_tokens", [200]) +@pytest.mark.parametrize("test_scenario", [ + # (prompt_token_count, max_num_batched_tokens, block_size, description) + # Scenario 1: prompt < max_num_matched_tokens (no chunk prefill) + (5000, 30000, 128, "small_prompt_no_chunk"), + # Scenario 2: prompt > max_num_batched_tokens, divisible by block_size + (35000, 30000, 128, "large_prompt_chunk_divisible"), + # Scenario 3: prompt > max_num_batched_tokens, not divisible by block_size + (35000, 30001, 128, "large_prompt_chunk_not_divisible"), + # Scenario 4: prompt > max_num_batched_tokens, max_num_batched_tokens divisible by block_size + (40000, 30000, 128, "very_large_prompt_divisible_batch"), + # Scenario 5: prompt > max_num_batched_tokens, max_num_batched_tokens not divisible by block_size + (40000, 30001, 128, "very_large_prompt_not_divisible_batch"), +]) +@pytest.mark.feature("uc_accuracy_test_offline") +@export_vars +@run_in_spawn_process +def test_offline_accuracy_chunk_prefill( + model_path: str, + base_prompt: str, + max_tokens: int, + test_scenario: Tuple[int, int, int, str], +): + """Test accuracy with chunk prefill scenarios. + + This test covers various chunk prefill scenarios: + 1. Prompt < max_num_matched_tokens (no chunk prefill) + 2. Prompt > max_num_batched_tokens (triggers chunk prefill) + 3. max_num_batched_tokens divisible by block_size + 4. max_num_batched_tokens not divisible by block_size + + Args: + base_prompt: Base prompt string to extend. + max_tokens: Maximum tokens to generate. + test_scenario: Tuple of (prompt_token_count, max_num_batched_tokens, block_size, description). + """ + prompt_token_count, max_num_batched_tokens, block_size, scenario_name = test_scenario + + # Load configuration + config_file = Path(__file__).parent.parent.parent / "config.yaml" + with open(config_file, "r", encoding="utf-8") as f: + config = yaml.safe_load(f) + + # Use model_path from parameter, fallback to config or environment + if not model_path or not os.path.exists(model_path): + model_path = config.get("llm_connection", {}).get("model_path") or os.getenv("MODEL_PATH") + if not model_path: + pytest.skip(f"model_path not configured or not found: {model_path}") + + if not os.path.exists(model_path): + pytest.skip(f"Model path does not exist: {model_path}") + + ucm_storage_dir = config.get("llm_connection", {}).get("ucm_storage_dir") or os.getenv("UCM_STORAGE_DIR", "/tmp/ucm_cache") + + # Setup tokenizer + tokenizer = AutoTokenizer.from_pretrained(model_path, use_chat_template=True) + + # Create prompt with target token count + test_prompt = create_prompt_with_token_count(base_prompt, tokenizer, prompt_token_count) + actual_token_count = len(tokenizer.encode(test_prompt)) + + # Format prompt with chat template if available + try: + messages = [{"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": test_prompt}] + formatted_prompt = tokenizer.apply_chat_template( + messages, + tokenize=False, + add_generation_prompt=True, + add_special_tokens=True, + ) + except Exception: + formatted_prompt = test_prompt + + # Setup UCM config + ucm_config = { + "ucm_connectors": [ + { + "ucm_connector_name": "UcmNfsStore", + "ucm_connector_config": { + "storage_backends": ucm_storage_dir, + "use_direct": False, + }, + } + ], + } + + # Create sampling params with temperature=0 for deterministic output + sampling_params = SamplingParams( + temperature=0.0, + top_p=0.95, + max_tokens=max_tokens, + ignore_eos=False, + ) + + # Check if divisible + is_divisible = (max_num_batched_tokens % block_size == 0) + + print(f"\n[INFO] ===== Chunk Prefill Accuracy Test: {scenario_name} =====") + print(f"[INFO] Model: {os.path.basename(model_path)}") + print(f"[INFO] Prompt token count: {actual_token_count} (target: {prompt_token_count})") + print(f"[INFO] Max num batched tokens: {max_num_batched_tokens}") + print(f"[INFO] Block size: {block_size}") + print(f"[INFO] Divisible: {is_divisible}") + print(f"[INFO] Will trigger chunk prefill: {actual_token_count > max_num_batched_tokens}") + print(f"[INFO] Max tokens: {max_tokens}") + print(f"[INFO] Temperature: 0.0 (deterministic)") + print(f"[INFO] UCM storage: {ucm_storage_dir}") + + # ===== Phase 1.1: Save KV cache to SSD (with chunk prefill if needed) ===== + print(f"\n[INFO] ===== Phase 1.1: Save KV Cache to SSD =====") + print(f"[INFO] HBM Prefix Caching: DISABLED") + print(f"[INFO] Sending prompt (may trigger chunk prefill)...") + + with build_llm_with_uc( + model_path=model_path, + ucm_config=ucm_config, + enable_prefix_caching=False, + max_num_batched_tokens=max_num_batched_tokens, + block_size=block_size, + ) as llm: + phase1_1_outputs, phase1_1_time = run_inference( + llm, [formatted_prompt], sampling_params, "Phase 1.1 (SSD save, chunk prefill)" + ) + phase1_1_output = phase1_1_outputs[0] + + time.sleep(1) + + # ===== Phase 1.2: Load from SSD (with chunk prefill if needed) ===== + print(f"\n[INFO] ===== Phase 1.2: Load from SSD (HBM PC disabled) =====") + print(f"[INFO] HBM Prefix Caching: DISABLED") + print(f"[INFO] Loading KV cache from SSD and sending prompt (may trigger chunk prefill)...") + + with build_llm_with_uc( + model_path=model_path, + ucm_config=ucm_config, + enable_prefix_caching=False, + max_num_batched_tokens=max_num_batched_tokens, + block_size=block_size, + ) as llm: + phase1_2_outputs, phase1_2_time = run_inference( + llm, [formatted_prompt], sampling_params, "Phase 1.2 (SSD load, chunk prefill)" + ) + phase1_2_output = phase1_2_outputs[0] + + # ===== Compare outputs ===== + print(f"\n[INFO] ===== Accuracy Test Results =====") + + # Compare Phase 1.1 vs Phase 1.2 (SSD load accuracy with chunk prefill) + phase1_identical = phase1_1_output == phase1_2_output + phase1_1_len = len(phase1_1_output) + phase1_2_len = len(phase1_2_output) + + # Calculate similarity + if phase1_1_len > 0 and phase1_2_len > 0: + min_len = min(phase1_1_len, phase1_2_len) + matching_chars = sum(1 for i in range(min_len) if phase1_1_output[i] == phase1_2_output[i]) + phase1_similarity = matching_chars / max(phase1_1_len, phase1_2_len) if max(phase1_1_len, phase1_2_len) > 0 else 0.0 + else: + phase1_similarity = 1.0 if phase1_identical else 0.0 + + print(f"\n[INFO] --- Chunk Prefill SSD Load Accuracy ---") + print(f"[INFO] Scenario: {scenario_name}") + print(f"[INFO] Phase 1.1 (SSD save) output length: {phase1_1_len}") + print(f"[INFO] Phase 1.2 (SSD load) output length: {phase1_2_len}") + print(f"[INFO] Outputs identical: {phase1_identical}") + print(f"[INFO] Similarity ratio: {phase1_similarity:.4f}") + print(f"[INFO] Phase 1.1 time: {phase1_1_time:.2f}s") + print(f"[INFO] Phase 1.2 time: {phase1_2_time:.2f}s") + + if not phase1_identical: + print(f"\n[ERROR] Outputs differ! Chunk prefill SSD load accuracy issue detected.") + print(f"[INFO] Phase 1.1 output (first 200 chars): {phase1_1_output[:200]}") + print(f"[INFO] Phase 1.2 output (first 200 chars): {phase1_2_output[:200]}") + + diff_pos = next((i for i, (c1, c2) in enumerate(zip(phase1_1_output, phase1_2_output)) if c1 != c2), None) + if diff_pos is not None: + print(f"[INFO] First difference at position: {diff_pos}") + context_start = max(0, diff_pos - 50) + context_end = min(len(phase1_1_output), diff_pos + 50) + print(f"[INFO] Context around difference:") + print(f"[INFO] Phase 1.1: ...{phase1_1_output[context_start:context_end]}...") + print(f"[INFO] Phase 1.2: ...{phase1_2_output[context_start:context_end]}...") + + # Assert outputs are identical + assert phase1_identical, ( + f"Chunk Prefill SSD Load Accuracy Test Failed!\n" + f"Scenario: {scenario_name}\n" + f"Phase 1.1 (SSD save) output differs from Phase 1.2 (SSD load) output.\n" + f"Similarity ratio: {phase1_similarity:.4f}\n" + f"Prompt token count: {actual_token_count}, Max batched tokens: {max_num_batched_tokens}\n" + f"Block size: {block_size}, Divisible: {is_divisible}\n" + f"Phase 1.1 output (first 500 chars): {phase1_1_output[:500]}\n" + f"Phase 1.2 output (first 500 chars): {phase1_2_output[:500]}" + ) + + print(f"[INFO] ✓ Chunk prefill accuracy test passed: outputs are identical") + + # Prepare data for export + value_lists = { + "model_path": [model_path], + "model_name": [os.path.basename(model_path)], + "scenario_name": [scenario_name], + "prompt_token_count": [actual_token_count], + "max_num_batched_tokens": [max_num_batched_tokens], + "block_size": [block_size], + "is_divisible": [1 if is_divisible else 0], + "triggers_chunk_prefill": [1 if actual_token_count > max_num_batched_tokens else 0], + "max_tokens": [max_tokens], + "phase1_identical": [1 if phase1_identical else 0], + "phase1_similarity_ratio": [phase1_similarity], + "phase1_1_output_length": [phase1_1_len], + "phase1_2_output_length": [phase1_2_len], + "phase1_1_time": [phase1_1_time], + "phase1_2_time": [phase1_2_time], + } + + return {"_name": "accuracy_test_offline_chunk_prefill", "_data": value_lists}