WeKnora自动化测试方案：保障系统稳定性-平芜编程栈

WeKnora自动化测试方案：保障系统稳定性

1. 引言

当你投入大量时间搭建好WeKnora知识库系统，上传了重要文档，配置了智能问答流程，最不想看到的就是系统更新后出现各种莫名其妙的问题。文档检索不准了、问答响应变慢了、甚至整个系统直接崩溃——这些情况在复杂的AI系统中并不少见。

WeKnora作为一个基于大语言模型的文档理解与语义检索框架，涉及多个复杂模块的协同工作：文档解析、向量化处理、语义检索、大模型推理等。任何一个环节出现问题，都可能影响整个系统的稳定性和用户体验。

这就是为什么我们需要一套完整的自动化测试方案。它就像是给系统请了一位不知疲倦的质量检测员，每次更新后自动检查所有核心功能是否正常工作，确保你的知识库始终稳定可靠。

2. 为什么WeKnora需要自动化测试

WeKnora的架构特点决定了手动测试的局限性。系统包含多个微服务组件，每个都有其特定的功能和行为模式。

系统复杂性的挑战：

多模块协同：文档解析、向量索引、语义检索、大模型交互等多个服务需要无缝协作
外部依赖：依赖大语言模型API、向量数据库、文件存储等外部服务
数据处理流水线：从文档上传到生成答案，涉及多个处理阶段和状态转换

手动测试的不足：

覆盖范围有限：难以测试所有可能的场景和边界条件
效率低下：每次更新都需要重复执行大量测试用例
容易遗漏：复杂的数据流和异常情况容易被忽略

自动化测试能够持续监控系统健康状态，在问题影响用户之前及时发现并修复，为系统稳定性提供坚实保障。

3. 测试环境搭建

3.1 环境准备

开始之前，确保你已经安装了必要的工具：

# 安装测试相关依赖 pip install pytest pytest-asyncio httpx python-dotenv # 或者使用项目提供的开发环境 make dev-env

3.2 测试配置

创建测试配置文件tests/.env.test：

# 测试数据库配置 TEST_DATABASE_URL=postgresql://postgres:postgres@localhost:5432/weknora_test # 测试模型配置（使用轻量级模型提高测试速度） TEST_LLM_MODEL=Qwen2-0.5B TEST_EMBEDDING_MODEL=text-embedding-3-small # 测试文件路径 TEST_DOCUMENTS_DIR=./tests/test_documents

3.3 测试数据准备

准备一些测试用的文档文件：

# tests/conftest.py import pytest import os from pathlib import Path @pytest.fixture(scope="session") def test_documents(): """准备测试文档""" test_dir = Path("./tests/test_documents") test_dir.mkdir(exist_ok=True) # 创建示例文本文件 with open(test_dir / "example.txt", "w") as f: f.write("这是测试文档内容。WeKnora是一个智能知识库系统。") # 创建示例Markdown文件 with open(test_dir / "README.md", "w") as f: f.write("# 测试文档\n\n这是Markdown格式的测试内容。") return test_dir

4. 单元测试：确保基础组件可靠性

单元测试关注单个函数或类的正确性，是测试金字塔的基础。

4.1 文档解析测试

# tests/unit/test_document_parser.py import pytest from weknora.core.parsers import DocumentParser class TestDocumentParser: @pytest.mark.asyncio async def test_text_parsing(self): """测试文本文档解析""" parser = DocumentParser() content = "简单文本内容" result = await parser.parse_text(content) assert result["content"] == content assert result["metadata"]["type"] == "text" @pytest.mark.asyncio async def test_pdf_parsing(self, test_documents): """测试PDF文档解析""" parser = DocumentParser() pdf_path = test_documents / "sample.pdf" # 假设我们已经有一个示例PDF result = await parser.parse_pdf(pdf_path) assert "content" in result assert len(result["content"]) > 0

4.2 向量化服务测试

# tests/unit/test_embedding_service.py import pytest import numpy as np from weknora.core.embedding import EmbeddingService class TestEmbeddingService: @pytest.mark.asyncio async def test_text_embedding(self): """测试文本向量化""" service = EmbeddingService() text = "测试文本" embedding = await service.embed_text(text) assert isinstance(embedding, list) assert len(embedding) == 768 # 假设维度为768 assert all(isinstance(x, float) for x in embedding) @pytest.mark.asyncio async def test_batch_embedding(self): """测试批量向量化""" service = EmbeddingService() texts = ["文本1", "文本2", "文本3"] embeddings = await service.embed_batch(texts) assert len(embeddings) == len(texts) assert all(len(emb) == 768 for emb in embeddings)

5. 集成测试：验证模块间协作

集成测试确保各个模块能够正确协同工作。

5.1 文档处理流水线测试

# tests/integration/test_document_pipeline.py import pytest from weknora.core.pipeline import DocumentProcessingPipeline class TestDocumentPipeline: @pytest.mark.asyncio async def test_full_document_processing(self, test_documents): """测试完整文档处理流程""" pipeline = DocumentProcessingPipeline() txt_path = test_documents / "example.txt" # 处理文档 result = await pipeline.process_document(txt_path) # 验证处理结果 assert result["status"] == "completed" assert "document_id" in result assert "chunks" in result assert len(result["chunks"]) > 0 # 验证每个chunk都有向量 for chunk in result["chunks"]: assert "embedding" in chunk assert len(chunk["embedding"]) == 768 @pytest.mark.asyncio async def test_document_processing_failure(self): """测试文档处理失败情况""" pipeline = DocumentProcessingPipeline() # 处理不存在的文件 result = await pipeline.process_document("nonexistent.txt") assert result["status"] == "failed" assert "error" in result

5.2 检索服务测试

# tests/integration/test_retrieval_service.py import pytest from weknora.core.retrieval import RetrievalService class TestRetrievalService: @pytest.mark.asyncio async def test_semantic_search(self): """测试语义检索""" service = RetrievalService() query = "如何配置知识库" results = await service.semantic_search(query, top_k=5) assert isinstance(results, list) assert len(results) <= 5 if results: assert "score" in results[0] assert "content" in results[0] @pytest.mark.asyncio async def test_hybrid_search(self): """测试混合检索（语义+关键词）""" service = RetrievalService() query = "WeKnora 部署" results = await service.hybrid_search(query, top_k=3) assert len(results) <= 3 # 验证结果包含相关性分数 for result in results: assert 0 <= result["score"] <= 1

6. 端到端测试：模拟真实用户场景

端到端测试从用户角度验证整个系统的功能。

6.1 知识库管理测试

# tests/e2e/test_knowledge_base_management.py import pytest import httpx from weknora.client import WeKnoraClient class TestKnowledgeBaseManagement: @pytest.mark.asyncio async def test_create_and_delete_knowledge_base(self): """测试创建和删除知识库""" client = WeKnoraClient() # 创建知识库 kb_data = { "name": "测试知识库", "description": "用于自动化测试的知识库" } create_response = await client.create_knowledge_base(kb_data) assert create_response.status_code == 201 kb_id = create_response.json()["id"] # 验证知识库创建成功 list_response = await client.list_knowledge_bases() kb_names = [kb["name"] for kb in list_response.json()] assert "测试知识库" in kb_names # 删除知识库 delete_response = await client.delete_knowledge_base(kb_id) assert delete_response.status_code == 200 @pytest.mark.asyncio async def test_document_upload_and_query(self): """测试文档上传和查询""" client = WeKnoraClient() # 创建知识库 kb_response = await client.create_knowledge_base({ "name": "测试查询知识库" }) kb_id = kb_response.json()["id"] try: # 上传文档 with open("tests/test_documents/example.txt", "rb") as f: upload_response = await client.upload_document(kb_id, f) assert upload_response.status_code == 200 # 等待文档处理完成 import time time.sleep(10) # 给系统一些处理时间 # 执行查询 query_response = await client.query_knowledge_base( kb_id, "什么是WeKnora" ) assert query_response.status_code == 200 response_data = query_response.json() assert "answer" in response_data assert "sources" in response_data finally: # 清理 await client.delete_knowledge_base(kb_id)

6.2 问答系统测试

# tests/e2e/test_qa_system.py import pytest from weknora.client import WeKnoraClient class TestQASystem: @pytest.mark.asyncio async def test_basic_question_answering(self): """测试基础问答功能""" client = WeKnoraClient() response = await client.ask_question("WeKnora是什么？") assert response.status_code == 200 data = response.json() # 验证响应结构 assert "answer" in data assert isinstance(data["answer"], str) assert len(data["answer"]) > 0 # 如果有来源，验证来源信息 if "sources" in data: for source in data["sources"]: assert "title" in source assert "content" in source @pytest.mark.asyncio async def test_multi_turn_conversation(self): """测试多轮对话""" client = WeKnoraClient() # 第一轮对话 first_response = await client.ask_question("如何安装WeKnora？") assert first_response.status_code == 200 # 第二轮对话（基于上下文） second_response = await client.follow_up("需要什么前置条件？") assert second_response.status_code == 200 # 验证两轮对话的相关性 first_answer = first_response.json()["answer"] second_answer = second_response.json()["answer"] # 第二轮回答应该与第一轮相关 assert len(second_answer) > 0

7. 性能测试：确保系统响应速度

性能测试帮助识别系统瓶颈，确保用户体验。

7.1 响应时间测试

# tests/performance/test_response_times.py import pytest import time import statistics from weknora.client import WeKnoraClient class TestPerformance: @pytest.mark.asyncio async def test_query_response_time(self): """测试查询响应时间""" client = WeKnoraClient() test_queries = [ "什么是知识库", "如何上传文档", "WeKnora支持哪些格式", "怎么配置大模型", "检索原理是什么" ] response_times = [] for query in test_queries: start_time = time.time() response = await client.ask_question(query) end_time = time.time() assert response.status_code == 200 response_times.append(end_time - start_time) # 输出性能指标 print(f"平均响应时间: {statistics.mean(response_times):.3f}s") print(f"最大响应时间: {max(response_times):.3f}s") print(f"最小响应时间: {min(response_times):.3f}s") # 性能断言：95%的查询应该在3秒内完成 assert statistics.mean(response_times) < 3.0 assert max(response_times) < 10.0 @pytest.mark.asyncio async def test_concurrent_requests(self): """测试并发请求处理能力""" import asyncio client = WeKnoraClient() async def make_request(query): start_time = time.time() response = await client.ask_question(query) end_time = time.time() return end_time - start_time, response.status_code # 并发10个请求 tasks = [make_request(f"测试查询 {i}") for i in range(10)] results = await asyncio.gather(*tasks) response_times = [time for time, status in results] status_codes = [status for time, status in results] # 所有请求都应该成功 assert all(code == 200 for code in status_codes) # 并发性能检查 assert statistics.mean(response_times) < 5.0

7.2 负载测试

# tests/performance/test_load.py import pytest import asyncio import aiohttp from datetime import datetime class TestLoad: @pytest.mark.asyncio async def test_sustained_load(self): """测试持续负载下的性能""" async with aiohttp.ClientSession() as session: start_time = datetime.now() successful_requests = 0 failed_requests = 0 # 持续运行1分钟 while (datetime.now() - start_time).total_seconds() < 60: try: async with session.get( "http://localhost:8080/api/health" ) as response: if response.status == 200: successful_requests += 1 else: failed_requests += 1 except Exception: failed_requests += 1 await asyncio.sleep(0.1) # 每秒约10个请求 # 输出测试结果 total_requests = successful_requests + failed_requests success_rate = (successful_requests / total_requests) * 100 print(f"总请求数: {total_requests}") print(f"成功请求: {successful_requests}") print(f"失败请求: {failed_requests}") print(f"成功率: {success_rate:.2f}%") # 断言：成功率应高于99% assert success_rate >= 99.0

8. 持续集成与自动化执行

8.1 GitHub Actions配置

创建.github/workflows/test.yml：

name: WeKnora Tests on: push: branches: [ main, develop ] pull_request: branches: [ main ] jobs: test: runs-on: ubuntu-latest services: postgres: image: postgres:13 env: POSTGRES_USER: postgres POSTGRES_PASSWORD: postgres POSTGRES_DB: weknora_test options: >- --health-cmd pg_isready --health-interval 10s --health-timeout 5s --health-retries 5 ports: - 5432:5432 steps: - uses: actions/checkout@v3 - name: Set up Python uses: actions/setup-python@v4 with: python-version: '3.10' - name: Install dependencies run: | pip install -r requirements.txt pip install -r requirements-test.txt - name: Run unit tests run: | pytest tests/unit/ -v --cov=weknora.core - name: Run integration tests run: | pytest tests/integration/ -v --cov=weknora.core --cov-append - name: Run performance tests run: | pytest tests/performance/ -v --cov=weknora.core --cov-append - name: Upload coverage reports uses: codecov/codecov-action@v3

8.2 本地测试脚本

创建测试执行脚本run_tests.sh：

#!/bin/bash # WeKnora自动化测试脚本 echo "启动WeKnora自动化测试..." # 设置环境变量 export ENVIRONMENT=test export DATABASE_URL=postgresql://postgres:postgres@localhost:5432/weknora_test # 启动测试数据库 docker-compose -f docker-compose.test.yml up -d # 等待数据库就绪 echo "等待数据库就绪..." sleep 10 # 运行单元测试 echo "运行单元测试..." pytest tests/unit/ -v --cov=weknora.core # 运行集成测试 echo "运行集成测试..." pytest tests/integration/ -v --cov=weknora.core --cov-append # 运行端到端测试（需要启动完整服务） echo "启动测试服务..." docker-compose -f docker-compose.test.yml up -d app echo "等待服务就绪..." sleep 20 echo "运行端到端测试..." pytest tests/e2e/ -v --cov=weknora.core --cov-append # 生成测试报告 echo "生成测试报告..." coverage html coverage report # 清理 echo "清理测试环境..." docker-compose -f docker-compose.test.yml down echo "测试完成！"

9. 常见问题与解决方案

测试环境问题：

数据库连接失败：检查测试数据库配置和网络连接
外部服务不可用：使用mock或stub替代外部依赖
资源不足：优化测试数据量，使用轻量级替代方案

测试稳定性问题：

随机失败：确保测试的独立性和可重复性
性能波动：多次运行取平均值，设置合理的阈值
环境依赖：使用容器化确保环境一致性

测试维护建议：

定期更新测试数据：反映真实使用场景
监控测试执行时间：及时发现性能退化
代码覆盖率分析：识别未测试的代码路径

10. 总结

建立完善的自动化测试体系对于WeKnora这样的复杂系统至关重要。通过单元测试确保基础组件的可靠性，集成测试验证模块间协作，端到端测试模拟真实用户场景，性能测试保障系统响应速度，最终形成一个全面的质量保障体系。

实际使用下来，这套测试方案确实能大大提升系统稳定性，每次更新后运行测试套件，心里就有底了。当然测试不是一劳永逸的，需要随着系统功能演进不断维护和更新测试用例。

建议团队将自动化测试纳入开发流程，每次代码提交都自动运行测试，及时发现和修复问题。对于刚开始的项目，可以从最重要的核心功能开始编写测试，逐步扩大测试覆盖范围。

获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。

WeKnora自动化测试方案：保障系统稳定性