Nanbeige4.1-3B测试驱动开发pytest用例集覆盖推理、流式、错误处理全场景1. 引言为什么需要为AI模型写测试当你部署好一个像Nanbeige4.1-3B这样的文本生成模型用Chainlit前端简单聊了几句感觉一切正常是不是就万事大吉了先别急着庆祝。你有没有想过这些问题模型真的能稳定处理各种复杂的推理问题吗流式输出功能会不会在某个关键时刻卡住如果用户输入了乱七八糟的内容服务会不会直接崩溃今天能跑通的代码明天更新了依赖库还能跑吗这就是为什么我们需要测试驱动开发TDD。今天我就带你用pytest为Nanbeige4.1-3B模型搭建一套完整的测试用例集覆盖从基础推理到错误处理的全部场景。这不是为了应付检查而是为了让你晚上能睡个安稳觉知道自己的服务有多可靠。2. 测试环境搭建与准备2.1 理解我们的测试对象在开始写测试之前我们先明确要测试什么。根据你的描述我们的系统架构是这样的后端模型使用vLLM部署的Nanbeige4.1-3B文本生成模型前端界面Chainlit提供的Web交互界面通信方式通过HTTP API进行模型调用对于测试来说我们主要关注的是模型服务的API接口而不是Chainlit的前端界面。因为只要API工作正常前端自然就能正常调用。2.2 安装测试所需的工具包打开你的项目目录创建一个requirements-test.txt文件pytest7.0.0 pytest-asyncio0.21.0 httpx0.25.0 pytest-html4.0.0 pytest-cov4.1.0然后安装它们pip install -r requirements-test.txt这些工具各有各的用处pytest我们的主力测试框架pytest-asyncio因为vLLM的API可能是异步的httpx用来模拟客户端发送HTTP请求pytest-html生成漂亮的测试报告pytest-cov检查我们的测试覆盖了多少代码2.3 创建测试目录结构好的项目结构能让测试工作事半功倍。我建议这样组织nanbeige-tests/ ├── conftest.py # pytest的配置文件 ├── tests/ # 所有测试用例 │ ├── __init__.py │ ├── test_basic.py # 基础功能测试 │ ├── test_streaming.py # 流式输出测试 │ ├── test_error.py # 错误处理测试 │ └── test_integration.py # 集成测试 ├── fixtures/ # 测试数据 │ └── test_prompts.json └── reports/ # 测试报告自动生成3. 编写基础推理功能测试3.1 测试模型的基本问答能力我们先从最简单的开始测试模型能不能正确回答基础问题。创建tests/test_basic.pyimport pytest import httpx import json import time class TestBasicInference: 测试Nanbeige4.1-3B的基础推理功能 # 这是pytest的fixture相当于测试前的准备工作 pytest.fixture def api_client(self): 创建一个HTTP客户端连接到我们的模型服务 # 假设你的vLLM服务运行在本地8080端口 base_url http://localhost:8080 client httpx.AsyncClient(base_urlbase_url, timeout30.0) yield client # 把客户端交给测试用例使用 client.close() # 测试结束后关闭连接 pytest.mark.asyncio async def test_simple_math(self, api_client): 测试简单的数学推理9.11和9.8哪个大 # 准备请求数据 request_data { prompt: Which number is bigger, 9.11 or 9.8?, max_tokens: 100, temperature: 0.1 # 温度设低一点让输出更确定 } # 发送请求到模型 response await api_client.post(/v1/completions, jsonrequest_data) # 验证响应状态 assert response.status_code 200, f请求失败: {response.text} # 解析响应内容 result response.json() generated_text result[choices][0][text].strip().lower() # 验证模型回答是否正确 # 注意模型可能用不同方式表达9.11更大 assert 9.11 in generated_text or nine point one one in generated_text assert any(word in generated_text for word in [bigger, greater, larger]) print(f✓ 数学推理测试通过模型回答: {generated_text}) pytest.mark.asyncio async def test_logical_reasoning(self, api_client): 测试逻辑推理能力 test_cases [ { prompt: 如果所有猫都怕水而Tom是一只猫那么Tom怕水吗, expected_keywords: [怕水, 是的, 正确] }, { prompt: 北京是中国的首都。上海是中国的首都吗, expected_keywords: [不是, 北京, 上海不是] } ] for i, test_case in enumerate(test_cases): request_data { prompt: test_case[prompt], max_tokens: 50, temperature: 0.1 } response await api_client.post(/v1/completions, jsonrequest_data) assert response.status_code 200 result response.json() answer result[choices][0][text].strip() # 检查回答中是否包含预期的关键词 has_expected any(keyword in answer for keyword in test_case[expected_keywords]) assert has_expected, f测试用例{i1}失败。模型回答: {answer} print(f✓ 逻辑推理测试{i1}通过: {answer[:50]}...) pytest.mark.asyncio async def test_chinese_understanding(self, api_client): 测试中文理解能力 request_data { prompt: 请用中文解释什么是人工智能, max_tokens: 150, temperature: 0.7 } response await api_client.post(/v1/completions, jsonrequest_data) assert response.status_code 200 result response.json() answer result[choices][0][text].strip() # 检查回答是否包含AI相关关键词 ai_keywords [人工智能, AI, 机器学习, 智能, 算法] has_ai_content any(keyword in answer for keyword in ai_keywords) assert has_ai_content, 模型似乎没有理解人工智能这个问题 assert len(answer) 20, 回答太简短可能有问题 print(f✓ 中文理解测试通过回答长度: {len(answer)}字符)3.2 运行基础测试并查看结果在项目根目录下运行pytest tests/test_basic.py -v你会看到类似这样的输出test_basic.py::TestBasicInference::test_simple_math PASSED test_basic.py::TestBasicInference::test_logical_reasoning PASSED test_basic.py::TestBasicInference::test_chinese_understanding PASSED 3 passed in 5.12 seconds 如果测试失败pytest会告诉你具体哪里出了问题方便你快速定位。4. 实现流式输出测试4.1 什么是流式输出为什么要测试它流式输出Streaming是让模型一个字一个字地返回结果而不是等全部生成完再一次性返回。这能提升用户体验感觉响应更快节省服务器内存不用缓存完整结果允许用户中途停止生成但流式输出也更复杂容易出问题。所以我们需要专门测试。创建tests/test_streaming.pyimport pytest import httpx import json import asyncio class TestStreamingOutput: 测试Nanbeige4.1-3B的流式输出功能 pytest.fixture def api_client(self): base_url http://localhost:8080 client httpx.AsyncClient(base_urlbase_url, timeout60.0) yield client client.close() pytest.mark.asyncio async def test_basic_streaming(self, api_client): 测试基础的流式输出功能 request_data { prompt: 请写一个关于秋天的短诗, max_tokens: 100, temperature: 0.8, stream: True # 关键参数开启流式输出 } # 对于流式请求我们需要用流式方式读取响应 async with api_client.stream(POST, /v1/completions, jsonrequest_data) as response: assert response.status_code 200 chunks [] async for chunk in response.aiter_lines(): if chunk: # 跳过空行 # 流式响应通常是SSE格式data: {...} if chunk.startswith(data: ): json_str chunk[6:] # 去掉data: 前缀 if json_str ! [DONE]: try: data json.loads(json_str) text data[choices][0][text] chunks.append(text) print(f收到流式块: {text}) except json.JSONDecodeError: print(fJSON解析错误: {json_str}) # 验证我们收到了多个流式块 assert len(chunks) 1, 应该收到多个流式块但只收到了一个或没有 # 把所有块拼接成完整文本 full_text .join(chunks) assert len(full_text) 20, 生成的文本太短 assert 秋天 in full_text or 秋 in full_text, 生成内容似乎不相关 print(f✓ 流式输出测试通过共收到{len(chunks)}个数据块) print(f完整生成内容: {full_text}) pytest.mark.asyncio async def test_streaming_interruption(self, api_client): 测试流式输出能否被正确中断 # 这个测试模拟用户中途取消请求的场景 request_data { prompt: 请详细解释量子计算的基本原理包括量子比特、叠加态和量子纠缠的概念。, max_tokens: 500, # 故意设置较长的生成 temperature: 0.7, stream: True } try: # 设置较短的超时时间模拟用户中途取消 async with api_client.stream(POST, /v1/completions, jsonrequest_data, timeout5.0) as response: chunks_received 0 start_time asyncio.get_event_loop().time() async for chunk in response.aiter_lines(): if chunk and chunk.startswith(data: ) and chunk[6:] ! [DONE]: chunks_received 1 print(f收到第{chunks_received}个数据块) # 收到3个块后就取消请求 if chunks_received 3: print(模拟用户取消请求...) break # 中断读取流 elapsed asyncio.get_event_loop().time() - start_time print(f流式请求在{elapsed:.2f}秒后被中断共收到{chunks_received}个数据块) # 验证我们确实收到了部分数据 assert chunks_received 0, 在中断前应该至少收到一些数据 assert chunks_received 3, 应该在收到3个数据块后中断 except httpx.ReadTimeout: print(✓ 流式请求按预期超时中断) # 超时也是可接受的中断方式 pass pytest.mark.asyncio async def test_streaming_large_output(self, api_client): 测试生成大量内容时的流式输出稳定性 request_data { prompt: 请写一篇关于人工智能未来发展的短文至少500字。, max_tokens: 1000, temperature: 0.7, stream: True } total_chars 0 chunk_count 0 async with api_client.stream(POST, /v1/completions, jsonrequest_data) as response: assert response.status_code 200 async for chunk in response.aiter_lines(): if chunk and chunk.startswith(data: ): json_str chunk[6:] if json_str ! [DONE]: try: data json.loads(json_str) text data[choices][0][text] total_chars len(text) chunk_count 1 # 每10个块打印一次进度 if chunk_count % 10 0: print(f进度: 已收到{chunk_count}个块共{total_chars}字符) except Exception as e: print(f处理数据块时出错: {e}) # 记录错误但继续测试 continue # 验证生成的内容量 assert total_chars 300, f生成内容太少只有{total_chars}字符 assert chunk_count 5, f流式块数量太少: {chunk_count} print(f✓ 大内容流式测试通过共生成{total_chars}字符分为{chunk_count}个数据块)4.2 流式测试的关键要点写流式输出测试时要特别注意超时处理流式请求可能持续很久要设置合理的超时错误恢复某个数据块解析失败不应该导致整个测试失败完整性验证最后要验证所有数据块能拼成完整、合理的内容中断测试模拟用户中途取消的情况确保服务能正确处理5. 构建全面的错误处理测试5.1 测试各种异常输入模型服务最怕的就是用户输入乱七八糟的东西。好的服务应该能优雅地处理错误而不是直接崩溃。创建tests/test_error.pyimport pytest import httpx import json class TestErrorHandling: 测试Nanbeige4.1-3B的错误处理能力 pytest.fixture def api_client(self): base_url http://localhost:8080 client httpx.AsyncClient(base_urlbase_url, timeout10.0) yield client client.close() pytest.mark.asyncio async def test_empty_prompt(self, api_client): 测试空输入 request_data { prompt: , # 空字符串 max_tokens: 50 } response await api_client.post(/v1/completions, jsonrequest_data) # 空输入应该返回错误而不是崩溃 # 不同的API设计可能不同这里我们检查是否返回了合理的响应 assert response.status_code in [200, 400, 422], f非预期的状态码: {response.status_code} if response.status_code 200: # 如果能处理空输入至少响应应该是有效的JSON result response.json() assert choices in result print(✓ 服务能处理空输入返回了有效响应) else: # 如果返回错误错误信息应该是清晰的 error_data response.json() assert error in error_data or message in error_data print(f✓ 服务正确拒绝了空输入: {error_data.get(error, error_data.get(message, 未知错误))}) pytest.mark.asyncio async def test_very_long_prompt(self, api_client): 测试超长输入可能超过模型限制 # 创建一个很长的提示词约5000字符 long_prompt 请解释人工智能。 * 200 request_data { prompt: long_prompt, max_tokens: 50, temperature: 0.7 } response await api_client.post(/v1/completions, jsonrequest_data) # 超长输入可能被截断或返回错误 if response.status_code 200: result response.json() # 应该能生成一些内容 assert choices in result assert len(result[choices]) 0 print(✓ 服务能处理超长输入可能进行了截断) elif response.status_code 400: # 返回客户端错误是合理的 error_data response.json() assert error in error_data or message in error_data print(f✓ 服务正确拒绝了超长输入: {error_data.get(error, 输入过长)}) else: # 其他状态码可能有问题 pytest.fail(f处理超长输入时返回了非预期状态码: {response.status_code}) pytest.mark.asyncio async def test_invalid_json(self, api_client): 测试发送无效的JSON数据 # 直接发送无效的JSON字符串 invalid_json {这不是有效的JSON response await api_client.post(/v1/completions, contentinvalid_json, headers{Content-Type: application/json}) # 应该返回400 Bad Request assert response.status_code 400, f无效JSON应该返回400但得到了{response.status_code} print(✓ 服务能正确处理无效JSON输入) pytest.mark.asyncio async def test_missing_required_fields(self, api_client): 测试缺少必填字段 test_cases [ {max_tokens: 50}, # 缺少prompt {prompt: 你好}, # 缺少max_tokens {} # 全部缺少 ] for i, data in enumerate(test_cases): response await api_client.post(/v1/completions, jsondata) # 缺少必填字段应该返回错误 assert response.status_code in [400, 422], f测试用例{i}应该返回400/422但得到了{response.status_code} error_data response.json() assert error in error_data or message in error_data print(f✓ 测试用例{i1}通过: 缺少字段时返回了正确错误) pytest.mark.asyncio async def test_invalid_parameters(self, api_client): 测试无效参数值 invalid_cases [ { prompt: 正常提示词, max_tokens: -10, # 负数token数 expected_error: max_tokens }, { prompt: 正常提示词, max_tokens: 100, temperature: 2.5, # 温度超出合理范围 expected_error: temperature }, { prompt: 正常提示词, max_tokens: 100, top_p: 1.5, # top_p超出范围 expected_error: top_p } ] for case in invalid_cases: # 复制用例移除expected_error字段它不是请求参数 request_data case.copy() expected_error_field request_data.pop(expected_error) response await api_client.post(/v1/completions, jsonrequest_data) # 应该返回错误 assert response.status_code in [400, 422], f参数{expected_error_field}无效时应返回错误 if response.status_code ! 200: error_data response.json() error_msg str(error_data).lower() # 错误信息应该提到有问题的字段 assert expected_error_field in error_msg or parameter in error_msg or invalid in error_msg print(f✓ 无效参数测试通过: {expected_error_field}) pytest.mark.asyncio async def test_special_characters(self, api_client): 测试特殊字符和边缘情况 special_prompts [ Hello\tWorld\nNew Line, # 包含制表符和换行 Emoji test: , # 包含emoji SQL注入测试: OR 11, # 看起来像SQL注入 XSS测试: scriptalert(test)/script, # 看起来像XSS 非常长的单词 a * 1000, # 超长单词 前后有空格 , # 前后空格 ] for i, prompt in enumerate(special_prompts): request_data { prompt: prompt, max_tokens: 30, temperature: 0.1 } response await api_client.post(/v1/completions, jsonrequest_data) # 服务不应该崩溃 assert response.status_code ! 500, f特殊输入导致服务器错误: {prompt[:50]}... if response.status_code 200: result response.json() assert choices in result print(f✓ 特殊字符测试{i1}通过: 服务正常响应) else: # 返回400/422也是可以接受的拒绝了异常输入 print(f✓ 特殊字符测试{i1}通过: 服务拒绝了异常输入)5.2 错误测试的设计思路好的错误测试应该覆盖边界情况空输入、超长输入、极端参数值异常格式无效JSON、错误的数据类型安全相关看起来像攻击的输入SQL注入、XSS等服务稳定性连续错误请求后服务是否还能正常工作6. 创建集成测试与性能测试6.1 模拟真实使用场景的集成测试创建tests/test_integration.py模拟真实用户的使用模式import pytest import httpx import json import asyncio from datetime import datetime class TestIntegration: 集成测试模拟真实使用场景 pytest.fixture def api_client(self): base_url http://localhost:8080 client httpx.AsyncClient(base_urlbase_url, timeout30.0) yield client client.close() pytest.mark.asyncio async def test_concurrent_requests(self, api_client): 测试并发请求处理能力 # 模拟5个用户同时请求 prompts [ 用一句话介绍人工智能, 写一个简短的天气预报, 解释什么是机器学习, 写一首关于春天的诗, 回答地球是圆的吗 ] async def make_request(prompt): 单个请求函数 request_data { prompt: prompt, max_tokens: 50, temperature: 0.7 } try: start_time datetime.now() response await api_client.post(/v1/completions, jsonrequest_data) end_time datetime.now() response_time (end_time - start_time).total_seconds() if response.status_code 200: result response.json() return { success: True, response_time: response_time, text: result[choices][0][text][:50] # 只取前50字符 } else: return { success: False, response_time: response_time, error: f状态码: {response.status_code} } except Exception as e: return { success: False, response_time: 0, error: str(e) } # 并发执行所有请求 tasks [make_request(prompt) for prompt in prompts] results await asyncio.gather(*tasks) # 分析结果 successful sum(1 for r in results if r[success]) response_times [r[response_time] for r in results if r[success]] print(f\n并发测试结果:) print(f- 总请求数: {len(prompts)}) print(f- 成功数: {successful}) print(f- 成功率: {successful/len(prompts)*100:.1f}%) if response_times: avg_time sum(response_times) / len(response_times) max_time max(response_times) min_time min(response_times) print(f- 平均响应时间: {avg_time:.2f}秒) print(f- 最快响应: {min_time:.2f}秒) print(f- 最慢响应: {max_time:.2f}秒) # 验证至少大部分请求应该成功 assert successful len(prompts) * 0.8, f并发请求成功率太低: {successful}/{len(prompts)} if response_times: # 平均响应时间应该在合理范围内 assert avg_time 10.0, f平均响应时间过长: {avg_time:.2f}秒 print(✓ 并发请求测试通过) pytest.mark.asyncio async def test_extended_conversation(self, api_client): 测试多轮对话能力 conversation [ 你好请介绍下你自己, 你能做什么, 写一个简短的Python函数计算斐波那契数列, 用中文解释一下这个函数 ] full_conversation for i, user_input in enumerate(conversation): # 将之前的对话历史也包含在提示中 prompt f{full_conversation}用户: {user_input}\n助手: request_data { prompt: prompt, max_tokens: 150, temperature: 0.7 } response await api_client.post(/v1/completions, jsonrequest_data) assert response.status_code 200, f第{i1}轮对话失败 result response.json() assistant_response result[choices][0][text].strip() # 更新对话历史 full_conversation f用户: {user_input}\n助手: {assistant_response}\n print(f第{i1}轮对话 - 用户: {user_input}) print(f助手: {assistant_response[:100]}...) print(- * 50) # 验证响应不是空的 assert len(assistant_response) 10, f第{i1}轮响应太短 # 给模型一点时间休息避免过热 if i len(conversation) - 1: await asyncio.sleep(1) print(✓ 多轮对话测试通过) pytest.mark.asyncio async def test_mixed_request_types(self, api_client): 混合测试普通请求和流式请求交替 test_scenarios [ {type: normal, prompt: 第一个普通请求, stream: False}, {type: stream, prompt: 第一个流式请求, stream: True}, {type: normal, prompt: 第二个普通请求, stream: False}, {type: stream, prompt: 第二个流式请求, stream: True}, ] for i, scenario in enumerate(test_scenarios): request_data { prompt: scenario[prompt], max_tokens: 50, temperature: 0.7, stream: scenario[stream] } print(f执行{i1}: {scenario[type]}请求 - {scenario[prompt]}) if scenario[stream]: # 流式请求 async with api_client.stream(POST, /v1/completions, jsonrequest_data) as response: assert response.status_code 200 chunks [] async for chunk in response.aiter_lines(): if chunk and chunk.startswith(data: ): json_str chunk[6:] if json_str ! [DONE]: try: data json.loads(json_str) text data[choices][0][text] chunks.append(text) except: pass full_text .join(chunks) assert len(full_text) 0 print(f 流式响应: 收到{len(chunks)}个块内容: {full_text[:50]}...) else: # 普通请求 response await api_client.post(/v1/completions, jsonrequest_data) assert response.status_code 200 result response.json() text result[choices][0][text] print(f 普通响应: {text[:50]}...) # 请求之间稍作间隔 if i len(test_scenarios) - 1: await asyncio.sleep(0.5) print(✓ 混合请求类型测试通过)6.2 运行完整的测试套件现在我们可以运行所有测试了# 运行所有测试 pytest tests/ -v # 运行测试并生成HTML报告 pytest tests/ --htmlreports/test_report.html --self-contained-html # 运行测试并检查代码覆盖率 pytest tests/ --cov. --cov-reporthtml:reports/coverage # 只运行错误处理测试 pytest tests/test_error.py -v # 运行特定测试类 pytest tests/test_basic.py::TestBasicInference -v7. 总结建立持续测试的实践7.1 测试的价值与收获通过这一整套测试用例我们为Nanbeige4.1-3B模型服务建立了质量保障体系功能验证确保模型能正确回答各种问题稳定性保障流式输出不会中途崩溃健壮性检查异常输入不会导致服务宕机性能监控了解服务在并发下的表现7.2 下一步的测试扩展建议如果你想让测试更完善可以考虑添加压力测试模拟成百上千的并发用户实现自动化测试流水线每次代码更新自动运行测试添加监控测试定期检查服务健康状态创建回归测试集确保新功能不破坏旧功能7.3 让测试成为开发的一部分最好的测试不是事后补的而是在开发过程中就写好的。我建议先写测试再写代码特别是修复bug时先写一个重现bug的测试测试要快单个测试最好在几秒内完成测试要独立一个测试失败不应该影响其他测试测试要可读好的测试本身就是文档记住测试不是负担而是你的安全网。有了这套测试下次更新模型版本、调整参数或者增加新功能时你就能更有信心了。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。