# knowledge-center-server

**Repository Path**: pandaoknight/knowledge-center-server

## Basic Information

- **Project Name**: knowledge-center-server
- **Description**: No description available
- **Primary Language**: Python
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2025-05-28
- **Last Updated**: 2025-06-04

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# knowledge-center-server

## 项目状态概览

🚀 **当前状态**: 阶段二已完成，准备开始阶段三

### 📊 开发进度
- ✅ **阶段一**: 项目基础设施搭建 (100% 完成)
  - ✅ 环境准备 (conda环境、依赖安装)
  - ✅ 项目结构搭建 (目录结构、基础文件)
  - ✅ 配置管理 (YAML配置、环境变量、日志系统)
- ✅ **阶段二**: 核心服务实现 (100% 完成)
  - ✅ RAGFlow HTTP 客户端 (已完成)
  - ✅ 基础路由框架 (已完成)
  - ✅ 请求日志中间件 (已完成)
- ⏳ **阶段三**: API 路由实现 (0% 完成)

### 🎯 当前可用功能
- ✅ 服务启动在 18084 端口
- ✅ 健康检查端点 `/health`
- ✅ 根路径端点 `/`
- ✅ 配置信息端点 `/config`
- ✅ 详细健康检查端点 `/health/detailed`
- ✅ 版本信息端点 `/version`
- ✅ CORS 跨域支持
- ✅ FastAPI 自动文档 `/docs`
- ✅ 完整的配置管理系统
- ✅ 结构化日志记录
- ✅ API Key 自动加载验证
- ✅ RAGFlow HTTP 客户端 (支持所有API)
- ✅ 请求日志中间件 (请求ID追踪、响应时间统计)
- ✅ 统一错误处理 (RAGFlow错误返回"服务不可用")
- ✅ 路由模块化架构

### 🔧 下一步计划
1. 开始阶段三：API路由实现
2. 实现Pydantic数据模型（基于RAGFlow实际API响应）
3. 实现具体的API端点功能

---

## 项目简介

knowledge-center-server 是一个基于 FastAPI 的 RAGFlow 风格的知识中台服务，用于转发和管理 RAGFlow 的 Datasets、Documents、Chunks 和 Retrieval 相关的 API 请求。

**重要特性**: 本项目通过 Pydantic 模型严格定义所有 API 的入参和出参数据结构，确保 API 接口的明确性和可预测性，为后续基于 RAGFlow 的 API 字段裁剪和定制化提供坚实基础。

## 功能特性

- 🔄 **API 转发**: 将客户端请求转发到 RAGFlow 服务
- 🔐 **认证管理**: 自动处理 RAGFlow API Key 认证
- 📝 **日志记录**: 详细记录请求/响应日志
- ⚙️ **配置管理**: 支持灵活的配置文件管理
- 🌐 **CORS 支持**: 支持跨域请求
- 🚀 **高性能**: 基于 FastAPI 异步框架
- 📋 **数据结构明确**: 使用 Pydantic 严格定义所有 API 的入参和出参，支持 JSON 格式输出
- ✂️ **字段裁剪支持**: 为后续 API 字段的定制化裁剪提供基础架构
- 🎯 **RAGFlow 兼容**: 严格基于 RAGFlow 实际 API 规范实现

## API 数据结构设计原则

### 核心设计理念
本项目的一个重要目标是确保所有 API 接口的**入参和出参数据结构完全明确**，具体实现方式：

1. **基于 RAGFlow 实际 API**: 
   - 所有 API 结构严格参考 RAGFlow 的实际 API 规范 (apispec.json)
   - 通过调用 RAGFlow 实际端点验证响应格式
   - 确保与 RAGFlow 的完全兼容性

2. **Pydantic 模型约束**: 
   - 所有 API 入参使用 Pydantic 模型进行严格验证
   - 所有 API 出参使用 Pydantic 模型进行结构化定义
   - 确保数据类型、字段名称、必填/可选属性完全明确

3. **分页格式统一**:
   - 严格遵循 RAGFlow 的分页格式和风格
   - 保持 page、page_size、total 等字段的一致性
   - 支持 orderby、desc 等排序参数

4. **参数验证详细**:
   - 参数验证错误返回详细的错误信息
   - 提供清晰的字段要求和格式说明
   - 支持多语言错误消息

5. **JSON 标准化输出**:
   - 统一使用 JSON 格式进行数据交换
   - 响应结构标准化，便于前端解析和处理
   - 支持嵌套对象和数组的复杂数据结构

6. **字段裁剪预备**:
   - 通过明确的数据模型，为后续 API 字段裁剪提供基础
   - 支持根据业务需求动态调整入参和出参字段
   - 保持与 RAGFlow 原始 API 的兼容性

### 实现指导原则

#### 入参处理
- 为每个 API 端点创建专门的 `Request` 模型
- 使用 Pydantic 的字段验证功能确保数据完整性
- 支持可选字段和默认值设置
- 提供清晰的字段描述和示例

#### 出参处理
- 为每个 API 端点创建专门的 `Response` 模型
- 严格映射 RAGFlow 返回的数据结构
- 支持嵌套模型和复杂数据类型
- 确保所有可能的响应字段都有明确定义

#### 模型组织
- 在 `app/models/` 目录下按功能模块组织模型文件
- 创建基础模型类供其他模型继承
- 使用类型注解确保代码可读性和 IDE 支持
- 提供模型示例和文档字符串

#### 代码生成指导
在后续的 API 实现过程中，需要特别注意：
- 每个 API 端点都必须有对应的入参和出参 Pydantic 模型
- 模型字段名称应与 RAGFlow API 文档保持一致
- 当需要字段裁剪时，创建简化版本的模型
- 保持原始完整模型作为参考和兼容性保证

## 支持的 API (基于 RAGFlow 实际规范)

### 1. Datasets (数据集管理) - 4个端点
- `GET /api/v1/datasets` - 列出数据集 (支持分页、筛选、排序)
- `POST /api/v1/datasets` - 创建数据集
- `PUT /api/v1/datasets/{dataset_id}` - 更新数据集
- `DELETE /api/v1/datasets` - 批量删除数据集

### 2. Documents (文档管理) - 6个端点
- `GET /api/v1/datasets/{dataset_id}/documents` - 列出文档 (支持分页、筛选、排序)
- `POST /api/v1/datasets/{dataset_id}/documents` - 上传文档 (文件上传)
- `GET /api/v1/datasets/{dataset_id}/documents/{document_id}` - 下载文档 (文件流)
- `PUT /api/v1/datasets/{dataset_id}/documents/{document_id}` - 更新文档
- `DELETE /api/v1/datasets/{dataset_id}/documents` - 批量删除文档
- `POST /api/v1/datasets/{dataset_id}/chunks` - 开始解析文档为块
- `DELETE /api/v1/datasets/{dataset_id}/chunks` - 停止解析文档

### 3. Chunks (块管理) - 4个端点
- `GET /api/v1/datasets/{dataset_id}/documents/{document_id}/chunks` - 列出文档的块 (支持分页)
- `POST /api/v1/datasets/{dataset_id}/documents/{document_id}/chunks` - 添加块到文档
- `PUT /api/v1/datasets/{dataset_id}/documents/{document_id}/chunks/{chunk_id}` - 更新块
- `DELETE /api/v1/datasets/{dataset_id}/documents/{document_id}/chunks` - 批量删除块

### 4. Retrieval (检索) - 1个端点
- `POST /api/v1/retrieval` - 基于查询检索块

**总计**: 15个核心API端点

## 开发阶段规划

### ✅ 阶段一：项目基础设施搭建 (已完成)
**目标**: 建立项目基本结构和开发环境

#### ✅ 步骤 1.1: 环境准备 (已完成)
- [x] 创建 conda 虚拟环境 `knowledge-center-server`
- [x] 安装基础依赖包 (FastAPI, uvicorn, httpx, pydantic 等)
- [x] 创建 requirements.txt 和 environment.yml

#### ✅ 步骤 1.2: 项目结构搭建 (已完成)
- [x] 创建标准的 FastAPI 项目目录结构
- [x] 设置基础的 main.py 入口文件
- [x] 创建所有必要的 __init__.py 文件
- [x] 配置 CORS 中间件
- [x] 实现基础健康检查端点
- [x] 验证服务在 18084 端口正常启动

**阶段一完成状态**:
- ✅ 服务成功启动在 18084 端口
- ✅ `/health` 端点正常响应
- ✅ `/` 根路径端点正常响应
- ✅ 项目目录结构完整搭建
- ✅ 基础依赖环境配置完成

#### ✅ 步骤 1.3: 配置管理 (已完成)
- [x] 实现配置文件读取 (支持 API Key 文件路径配置)
- [x] 实现 RAGFlow 端点配置
- [x] 实现服务端口配置 (默认 18084)
- [x] 实现日志配置
- [x] 支持环境变量覆盖
- [x] 集成Pydantic配置验证
- [x] 实现loguru日志系统

**步骤1.3完成状态**:
- ✅ YAML配置文件系统建立
- ✅ 环境变量支持完成
- ✅ API Key自动加载验证
- ✅ 结构化日志记录实现
- ✅ 配置信息API端点可用

### ✅ 阶段二：核心服务实现 (已完成)
**目标**: 实现 RAGFlow 客户端和基础转发功能

#### ✅ 步骤 2.1: RAGFlow 客户端 (已完成)
- [x] 实现 RAGFlow HTTP 客户端类
- [x] 实现 API Key 自动注入
- [x] 实现请求/响应日志记录
- [x] 实现错误处理 (500 错误返回)
- [x] 完成单元测试 (15个测试用例全部通过)

**步骤2.1完成状态**:
- ✅ RAGFlow HTTP客户端类实现完成
- ✅ 支持所有API模块 (Datasets, Documents, Chunks, Retrieval)
- ✅ 采用fail fast策略，无重试机制
- ✅ 详细的请求/响应日志记录
- ✅ 完整的错误处理和异常分类
- ✅ 全局客户端实例管理
- ✅ 15个单元测试用例全部通过

#### ✅ 步骤 2.2: 基础路由框架 (已完成)
- [x] 实现请求日志中间件 (核心任务)
- [x] 重构健康检查端点到独立路由模块
- [x] 创建各API模块的路由框架
- [x] 优化错误处理中间件

**步骤2.2完成状态**:
- ✅ 请求日志中间件实现完成，支持请求ID追踪、响应时间统计、敏感信息过滤
- ✅ 健康检查端点重构到独立模块，新增详细健康检查和版本信息端点
- ✅ 数据集和检索API路由框架创建完成，为阶段三做好准备
- ✅ 统一错误处理机制，RAGFlow错误统一返回"服务不可用"
- ✅ 路由模块化架构建立，支持灵活的API组织和扩展
- ✅ 中间件堆栈优化，确保正确的执行顺序

### ⏳ 阶段三：API 路由实现 (准备开始)
**目标**: 基于 RAGFlow 实际 API 规范逐步实现各个 API 模块

#### ✅ 步骤 3.1: 核心 Pydantic 模型设计 (已完成)
**目标**: 建立完整的数据模型基础架构
- [x] 调用 RAGFlow 实际 API 验证响应格式
- [x] 创建基础 Pydantic 模型类 (base.py)
- [x] 实现数据集相关 Pydantic 模型 (datasets.py)
- [x] 实现文档相关 Pydantic 模型 (documents.py)
- [x] 实现块相关 Pydantic 模型 (chunks.py)
- [x] 实现检索相关 Pydantic 模型 (retrieval.py)
- [x] 创建模型单元测试

**步骤3.1完成状态**:
- ✅ 基于RAGFlow实际API响应设计的完整Pydantic模型体系
- ✅ 5个模型文件全部实现（base.py, datasets.py, documents.py, chunks.py, retrieval.py）
- ✅ 涵盖15个API端点的完整入参/出参模型定义
- ✅ 11个单元测试全部通过，验证模型正确性
- ✅ 支持JSON序列化/反序列化和严格数据验证
- ✅ 为API字段裁剪和定制化提供坚实基础

#### ⏳ 步骤 3.2: Datasets API (4个端点) (准备开始)
**目标**: 使用Pydantic模型实现完整的数据集管理API

##### 子步骤 3.2.1: 路由模型集成
- [x] 导入datasets相关Pydantic模型到路由模块
- [x] 替换`Dict[str, Any]`类型为具体的Request/Response模型
- [x] 更新路由函数签名，使用类型注解
- [x] 验证模型自动验证功能正常工作

**子步骤3.2.1完成状态**:
- ✅ 成功导入所有datasets相关Pydantic模型
- ✅ 所有4个端点的Dict类型已替换为具体模型
- ✅ 路由函数使用了正确的类型注解和response_model
- ✅ FastAPI自动验证功能正常工作（422错误响应）
- ✅ API正常返回RAGFlow数据，格式符合预期
- ✅ 创建了datasets API单元测试文件

##### 子步骤 3.2.2: GET /api/v1/datasets 端点增强
- [x] 使用`DatasetListRequest`模型替换查询参数
- [x] 使用`DatasetListResponse`模型包装返回数据
- [x] 实现分页、筛选、排序逻辑的严格验证
- [x] 添加详细的API文档字符串和示例

##### 子步骤 3.2.3: POST /api/v1/datasets 端点增强
- [x] 使用`DatasetCreateRequest`模型验证创建请求
- [x] 使用`DatasetCreateResponse`模型包装返回数据
- [x] 实现字段验证（必填字段、格式验证等）
- [x] 添加创建失败的详细错误处理

##### 子步骤 3.2.4: PUT /api/v1/datasets/{dataset_id} 端点增强
- [x] 使用`DatasetUpdateRequest`模型验证更新请求
- [x] 使用`DatasetUpdateResponse`模型包装返回数据
- [x] 实现部分更新（PATCH语义）支持
- [x] 添加数据集不存在的错误处理

##### 子步骤 3.2.5: DELETE /api/v1/datasets 端点增强
- [x] 使用`DatasetDeleteRequest`模型验证删除请求
- [x] 使用`DatasetDeleteResponse`模型包装返回数据
- [x] 实现批量删除的事务性处理
- [x] 添加删除结果的详细反馈

##### 子步骤 3.2.6: 错误处理和响应标准化
- [x] 实现统一的HTTP状态码处理
- [x] 标准化所有错误响应格式
- [x] 添加参数验证错误的详细描述
- [x] 确保与RAGFlow错误格式的兼容性

##### 子步骤 3.2.7: API文档和测试
- [x] 完善FastAPI自动文档的示例和描述
- [x] 创建datasets API的集成测试 ⭐ **已完成**
- [ ] 验证API与OpenAPI规范的兼容性

**子步骤3.2.7详细规划**: 数据集API集成测试设计

**🎯 测试设计目标**:
- 基于真实RAGFlow服务的端到端集成测试
- 按序执行的完整CRUD生命周期测试
- 共享测试数据机制，避免重复创建/删除开销
- 参考`test_ragflow_client.py`的成熟测试风格

**📋 测试架构设计**:

1. **共享数据机制**:
   ```python
   class SharedData:
       """共享测试数据的强类型封装类"""
       def __init__(self, test_client, created_dataset_id, created_dataset_info):
           self.test_client = test_client           # FastAPI测试客户端
           self.created_dataset_id = created_dataset_id   # 创建的测试数据集ID
           self.created_dataset_info = created_dataset_info  # 数据集完整信息
           self.updated_dataset_info = None         # 更新后的数据集信息（留待更新测试填充）
   
   # 模块级变量用于持久化共享测试数据
   _shared_data: Optional[SharedData] = None
   _test_client: Optional[TestClient] = None
   ```

2. **测试依赖关系**:
   - **独立前置检查**: RAGFlow服务可用性检测
   - **顺序执行CRUD**: 创建 → 列表验证 → 更新 → 列表验证 → 删除
   - **共享数据传递**: 每个测试通过SharedData获取前序测试的结果

**🔄 测试用例设计** (按执行顺序):

1. **test_ragflow_service_available**: 
   - 检查RAGFlow服务可用性
   - 如失败，后续所有测试跳过

2. **test_create_dataset_integration**:
   - 测试`POST /api/v1/datasets`端点
   - 创建测试数据集并保存到SharedData
   - 验证返回的数据集信息结构完整性

3. **test_list_datasets_integration**:
   - 测试`GET /api/v1/datasets`端点
   - 验证分页、排序、筛选参数
   - 确认创建的数据集出现在列表中

4. **test_list_datasets_with_filters**:
   - 测试各种筛选条件（按ID、按名称等）
   - 验证分页边界情况
   - 测试无效参数的错误处理

5. **test_update_dataset_integration**:
   - 测试`PUT /api/v1/datasets/{dataset_id}`端点
   - 更新SharedData中的数据集
   - 验证部分更新逻辑和字段验证

6. **test_list_datasets_verify_update**:
   - 重新获取数据集列表
   - 验证更新操作确实生效
   - 对比更新前后的字段变化

7. **test_update_dataset_error_scenarios**:
   - 测试更新不存在的数据集
   - 测试无效的更新数据
   - 验证错误响应格式和状态码

8. **test_delete_datasets_integration**:
   - 测试`DELETE /api/v1/datasets`端点
   - 删除SharedData中的测试数据集
   - 验证批量删除功能和响应格式

9. **test_delete_datasets_error_scenarios**:
   - 测试删除不存在的数据集
   - 测试空ID列表等边界情况
   - 验证错误处理机制

**🔧 测试工具函数**:

```python
async def check_ragflow_availability() -> tuple[bool, Optional[str]]:
    """检查RAGFlow服务是否可用"""
    # 复用test_ragflow_client.py中的实现

@pytest_asyncio.fixture(scope="function")
async def fastapi_test_client():
    """创建FastAPI测试客户端"""
    from app.main import app
    from fastapi.testclient import TestClient
    return TestClient(app)

@pytest_asyncio.fixture(scope="function") 
async def shared_test_data():
    """创建或复用共享的测试数据"""
    # 类似test_ragflow_client.py的shared_test_dataset fixture
```

**📊 测试分类标记**:
- `@pytest.mark.integration`: 所有测试都是集成测试
- `@pytest.mark.requires_ragflow`: 需要RAGFlow服务运行
- `@pytest.mark.timeout(30)`: 合理的超时时间

**🔍 验证点设计**:
- **API响应格式**: 验证Pydantic模型序列化正确性
- **HTTP状态码**: 确保错误场景返回正确状态码
- **数据一致性**: 创建→查询→更新→查询的数据一致性
- **错误处理**: RAGFlow错误的正确传播和包装
- **参数验证**: FastAPI自动验证功能正常工作

**⚠️ 注意事项**:
- 测试用例必须按顺序执行，不支持并发或乱序
- SharedData确保测试间的数据传递，但也意味着测试间有依赖
- 如果中间某个测试失败，后续测试可能会因为缺少预期数据而失败
- 每次运行测试会创建新的测试数据集，确保测试环境的干净

**🚀 期望测试覆盖率**:
- 4个API端点的基本功能: 100%
- 常见错误场景: 80%+
- 参数验证和边界情况: 70%+
- 性能和超时场景: 基础覆盖

**📝 测试输出示例**:
```
test_datasets.py::TestDatasetAPIIntegration::test_ragflow_service_available PASSED
test_datasets.py::TestDatasetAPIIntegration::test_create_dataset_integration PASSED  
test_datasets.py::TestDatasetAPIIntegration::test_list_datasets_integration PASSED
test_datasets.py::TestDatasetAPIIntegration::test_update_dataset_integration PASSED
test_datasets.py::TestDatasetAPIIntegration::test_delete_datasets_integration PASSED
```

这种设计既保证了测试的全面性，又通过共享数据机制提高了测试效率，同时保持了与现有测试风格的一致性。

##### 子步骤 3.2.8: 性能优化和监控
- [x] 添加API响应时间监控
- [x] 实现请求/响应大小限制
- [ ] 添加缓存策略（如果需要）
- [x] 优化数据库查询性能（通过RAGFlow）

**子步骤3.2.2-3.2.6完成状态**:
- ✅ 4个数据集API端点功能全面增强
- ✅ 严格的参数验证和错误处理机制
- ✅ 详细的API文档和使用示例
- ✅ 统一的响应格式和错误码映射
- ✅ RAGFlow原始错误透传机制
- ✅ 轻量级性能监控集成
- ✅ 慢请求检测和性能统计API

#### 步骤 3.3: Documents API (6个端点)
- [ ] 实现 documents.py 路由模块
- [ ] 实现 GET /api/v1/datasets/{dataset_id}/documents (列出文档，支持分页)
- [ ] 实现 POST /api/v1/datasets/{dataset_id}/documents (上传文档，文件处理)
- [ ] 实现 GET /api/v1/datasets/{dataset_id}/documents/{document_id} (下载文档，文件流)
- [ ] 实现 PUT /api/v1/datasets/{dataset_id}/documents/{document_id} (更新文档)
- [ ] 实现 DELETE /api/v1/datasets/{dataset_id}/documents (批量删除文档)
- [ ] 实现 POST /api/v1/datasets/{dataset_id}/chunks (开始解析文档)
- [ ] 实现 DELETE /api/v1/datasets/{dataset_id}/chunks (停止解析文档)
- [ ] 创建 documents API 单元测试

#### 步骤 3.4: Chunks API (4个端点)
- [ ] 实现 chunks.py 路由模块
- [ ] 实现 GET /api/v1/datasets/{dataset_id}/documents/{document_id}/chunks (列出块，支持分页)
- [ ] 实现 POST /api/v1/datasets/{dataset_id}/documents/{document_id}/chunks (添加块)
- [ ] 实现 PUT /api/v1/datasets/{dataset_id}/documents/{document_id}/chunks/{chunk_id} (更新块)
- [ ] 实现 DELETE /api/v1/datasets/{dataset_id}/documents/{document_id}/chunks (批量删除块)
- [ ] 创建 chunks API 单元测试

#### 步骤 3.5: Retrieval API (1个端点)
- [ ] 实现 retrieval.py 路由模块
- [ ] 实现 POST /api/v1/retrieval (检索块，支持多种参数)
- [ ] 创建 retrieval API 单元测试

## 项目结构

```
knowledge-center-server/
├── app/
│   ├── __init__.py
│   ├── main.py                 # FastAPI 应用入口
│   ├── config.py              # 配置管理
│   ├── models/               # Pydantic 数据模型
│   │   ├── __init__.py
│   │   ├── base.py          # 基础模型类
│   │   ├── datasets.py      # 数据集相关模型
│   │   ├── documents.py     # 文档相关模型
│   │   ├── chunks.py        # 块相关模型
│   │   └── retrieval.py     # 检索相关模型
│   ├── middleware/            # 中间件
│   │   ├── __init__.py
│   │   ├── cors.py           # CORS 中间件
│   │   └── logging.py        # 日志中间件
│   ├── routers/              # API 路由
│   │   ├── __init__.py
│   │   ├── health.py         # 健康检查
│   │   ├── datasets.py       # 数据集 API
│   │   ├── documents.py      # 文档 API
│   │   ├── chunks.py         # 块 API
│   │   └── retrieval.py      # 检索 API
│   ├── services/             # 业务服务
│   │   ├── __init__.py
│   │   └── ragflow_client.py # RAGFlow 客户端
│   └── utils/                # 工具函数
│       ├── __init__.py
│       └── logger.py         # 日志工具
├── config/                   # 配置文件
│   ├── config.yaml          # 主配置文件
│   └── logging.yaml         # 日志配置
├── keys/                     # 密钥文件
│   └── ragflow_local.txt    # RAGFlow API Key
├── data/                     # 数据文件
│   └── apispec.json         # RAGFlow API 规范文件
├── tests/                    # 测试文件
│   ├── __init__.py
│   ├── conftest.py          # pytest 配置
│   ├── models/              # 模型层测试
│   │   ├── __init__.py
│   │   ├── test_datasets.py    # 数据集模型测试
│   │   ├── test_documents.py   # 文档模型测试
│   │   ├── test_chunks.py      # 块模型测试
│   │   └── test_retrieval.py   # 检索模型测试
│   ├── services/            # 服务层测试
│   │   ├── __init__.py
│   │   └── test_ragflow_client.py  # RAGFlow客户端测试
│   ├── routers/             # 路由层测试
│   │   ├── __init__.py
│   │   ├── test_health.py      # 健康检查API测试
│   │   ├── test_datasets.py    # 数据集API测试
│   │   ├── test_documents.py   # 文档API测试
│   │   ├── test_chunks.py      # 块API测试
│   │   └── test_retrieval.py   # 检索API测试
│   ├── middleware/          # 中间件测试
│   │   ├── __init__.py
│   │   └── test_logging.py     # 日志中间件测试
│   └── utils/               # 工具函数测试
│       ├── __init__.py
│       └── test_logger.py      # 日志工具测试
├── logs/                     # 日志目录
├── requirements.txt          # Python 依赖
├── environment.yml          # Conda 环境配置
├── .gitignore              # Git 忽略文件
└── README.md               # 项目说明
```

## 技术栈

- **框架**: FastAPI 0.104+
- **HTTP 客户端**: httpx
- **配置管理**: PyYAML
- **日志**: loguru
- **验证**: Pydantic
- **ASGI 服务器**: uvicorn
- **测试**: pytest, pytest-timeout

## 测试说明

### 测试架构设计
项目采用**模块导向**的测试组织策略，测试目录结构与应用代码目录一一对应：

- **模型层测试** (`tests/models/`): 测试Pydantic数据模型
- **服务层测试** (`tests/services/`): 测试业务逻辑和外部服务集成
- **路由层测试** (`tests/routers/`): 测试API端点和HTTP接口
- **中间件测试** (`tests/middleware/`): 测试请求处理中间件
- **工具函数测试** (`tests/utils/`): 测试通用工具函数

### 测试分类标记
使用pytest标记系统区分测试类型和运行要求：

- `@pytest.mark.unit`: 单元测试，快速，无外部依赖
- `@pytest.mark.integration`: 集成测试，需要外部服务
- `@pytest.mark.requires_ragflow`: 需要RAGFlow服务运行
- `@pytest.mark.slow`: 运行时间较长的测试

### 🚀 重大突破：RAGFlow客户端测试重构成功

**问题解决**: 修复了关键的pytest异步fixture scope冲突问题，实现了**从0%可运行到100%通过率**的重大突破！

#### 修复的核心问题
- ❌ **原问题**: `ScopeMismatch: function scoped fixture _function_event_loop with session scoped request`
- ✅ **解决方案**: 重构为function scoped异步fixture + 智能RAGFlow可用性检查
- ✅ **成果**: 7个RAGFlow客户端测试全部通过，支持按标记分类运行

#### 测试执行验证
```bash
# 完整RAGFlow客户端测试
python -m pytest tests/services/test_ragflow_client.py -v
# 结果: 7 passed in 3.05s ✅

# 集成测试 (需要RAGFlow服务)
python -m pytest tests/services/test_ragflow_client.py -m integration -v  
# 结果: 6 passed, 1 deselected ✅

# 单元测试 (无外部依赖)
python -m pytest tests/services/test_ragflow_client.py -m unit -v
# 结果: 1 passed, 6 deselected ✅
```

### 运行测试

### ⚠️ 重要提醒: 测试前环境激活
在运行任何测试之前，请务必激活正确的conda环境：
```bash
conda activate knowledge-center-server
```

```bash
# 按测试类型运行
python -m pytest -m unit -v              # 只运行单元测试（快速）
python -m pytest -m integration -v       # 只运行集成测试
python -m pytest -m requires_ragflow -v  # 只运行需要RAGFlow的测试

# 按模块运行
python -m pytest tests/models/ -v        # 只测试数据模型
python -m pytest tests/services/ -v      # 只测试服务层
python -m pytest tests/routers/ -v       # 只测试API路由

# 运行特定测试文件
python -m pytest tests/models/test_datasets.py -v
python -m pytest tests/services/test_ragflow_client.py -v

# 运行所有测试
python -m pytest tests/ -v

# 排除慢测试
python -m pytest -m "not slow" -v

# 只运行模型层单元测试（推荐用于快速验证）
python -m pytest tests/models/ -m unit -v
```

### pytest配置注意事项

⚠️ **重要**: 确保 `pytest.ini` 中使用正确的配置段标题：

```ini
# ✅ 正确
[pytest]
markers = 
    unit: 单元测试，快速运行，无外部依赖
    integration: 集成测试，需要外部服务
    requires_ragflow: 需要RAGFlow服务运行
    slow: 运行时间较长的测试

# ❌ 错误 - 会导致标记警告
[tool:pytest]
markers = ...
```

### 测试环境要求

#### 单元测试环境
- Python 3.11+ 
- 项目依赖包
- 无需外部服务

#### 集成测试环境
- 单元测试环境 + 
- **RAGFlow服务**: 运行在配置的URL（默认 `http://118.113.164.107:18083`）
- **API密钥**: 有效的RAGFlow API Key文件
- **网络连接**: 能够访问RAGFlow服务

### 测试数据管理
- **隔离性**: 每个测试使用独立的测试数据，避免测试间相互影响
- **清理性**: 集成测试自动清理创建的测试数据
- **标准化**: 使用统一的测试数据命名规范

### 测试覆盖范围

#### 📊 当前覆盖情况
- ✅ **Pydantic模型验证** - 完整覆盖 (tests/models/)
- ✅ **RAGFlow客户端集成** - 完整覆盖 (tests/services/)  
- ⚠️ **数据集API端点** - 部分覆盖 (tests/routers/)
- ❌ **文档API端点** - 待实现
- ❌ **块API端点** - 待实现
- ❌ **检索API端点** - 待实现
- ❌ **中间件功能** - 待实现

#### 🎯 测试目标
项目将实现**15个核心API端点**的完整测试覆盖：
- 4个数据集管理端点
- 6个文档管理端点  
- 4个块管理端点
- 1个检索端点

如果测试无法运行，请检查：
1. RAGFlow服务是否正常运行
2. API密钥文件是否存在且有效
3. 网络连接是否正常
4. 依赖包是否正确安装

## 配置说明

### config/config.yaml
```yaml
server:
  host: "0.0.0.0"
  port: 18084
  reload: false

ragflow:
  base_url: "http://localhost:9380"
  api_key_file: "keys/ragflow_local.txt"
  timeout: 30

cors:
  allow_origins: ["*"]
  allow_credentials: true
  allow_methods: ["*"]
  allow_headers: ["*"]

logging:
  level: "INFO"
  file: "logs/app.log"
  max_size: "10MB"
  backup_count: 5
```

## 快速开始

### ⚠️ 重要提醒: 环境激活
在执行任何操作之前，请务必激活正确的conda环境：
```bash
conda activate knowledge-center-server
```

### 1. 创建环境
```bash
conda create -n knowledge-center-server python=3.11
conda activate knowledge-center-server
```

### 2. 安装依赖
```bash
# 确保已激活环境
conda activate knowledge-center-server
pip install -r requirements.txt
```

### 3. 配置 API Key
确保 `keys/ragflow_local.txt` 文件包含有效的 RAGFlow API Key。

### 4. 启动服务
```bash
# 确保已激活环境
conda activate knowledge-center-server
python -m app.main
```

### 5. 访问文档
打开浏览器访问: http://localhost:18084/docs

## 开发指南

### 添加新的 API 端点
1. 在对应的路由文件中添加端点函数
2. 在 RAGFlow 客户端中添加对应的方法
3. 创建对应的 Pydantic 模型
4. 添加相应的测试用例
5. 更新 API 文档

### 基于 RAGFlow 实际 API 开发流程
1. **调用验证**: 首先调用 RAGFlow 实际 API 验证响应格式
2. **模型设计**: 基于实际响应设计 Pydantic 模型
3. **路由实现**: 实现路由逻辑，确保与 RAGFlow 兼容
4. **测试验证**: 创建全面的单元测试
5. **文档更新**: 更新 API 文档和使用说明

### 错误处理规范
- RAGFlow 服务不可用时返回 500 错误
- 参数验证错误返回详细的 422 错误信息
- 错误信息只在日志中记录详细信息
- 前端返回清晰的错误提示

### 日志记录规范
- 记录所有请求/响应
- 记录错误详情
- 记录性能指标
- 敏感信息脱敏处理

## 部署说明

### Docker 部署
```bash
# 构建镜像
docker build -t knowledge-center-server .

# 运行容器
docker run -p 18084:18084 -v ./config:/app/config -v ./keys:/app/keys knowledge-center-server
```

### 生产环境部署
```bash
# 使用 gunicorn 部署
gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:18084
```

## 故障排除

### 常见问题
1. **RAGFlow 连接失败**: 检查 RAGFlow 服务状态和配置
2. **API Key 无效**: 确认 API Key 文件路径和内容
3. **端口占用**: 修改配置文件中的端口设置
4. **CORS 错误**: 检查 CORS 配置是否正确
5. **API 响应不匹配**: 确认是否基于最新的 RAGFlow API 规范

## 贡献指南

1. Fork 项目
2. 创建功能分支
3. 提交更改
4. 推送到分支
5. 创建 Pull Request

## 许可证

MIT License