graph_rag_server/docs/development.md

Development Guide

本文档说明如何设置开发环境、运行测试、以及贡献代码。

目录

• 开发环境设置
• 项目结构
• 开发工作流
• 代码规范
• 测试指南
• 调试技巧
• 贡献指南

开发环境设置

前置要求

• Python 3.11 或 3.12
• Git
• pip（Python 包管理器）
• 推荐使用 VS Code 或 PyCharm

1. 克隆仓库

git clone https://github.com/YOUR_USERNAME/rag-system.git
cd rag-system

2. 创建虚拟环境

# 创建虚拟环境
python -m venv venv

# 激活虚拟环境
# Linux/macOS:
source venv/bin/activate

# Windows:
venv\Scripts\activate

3. 安装依赖

# 安装生产依赖
pip install -r requirements.txt

# 安装开发依赖
pip install pytest pytest-asyncio pytest-cov hypothesis httpx
pip install flake8 black isort mypy
pip install safety bandit

4. 配置环境变量

# 复制示例配置
cp .env.example .env

# 编辑配置（使用你喜欢的编辑器）
nano .env

5. 启动依赖服务

使用 Docker Compose 启动开发环境所需的服务：

cd docker
docker-compose up -d postgres infinity redis

6. 初始化数据库

# 创建数据库
createdb rag_system_dev

# 运行迁移
alembic upgrade head

7. 验证安装

# 运行测试
pytest

# 启动应用
uvicorn src.main:app --reload

访问 http://localhost:8000/docs 查看 API 文档。

项目结构

rag_system/
├── src/                      # 源代码
│   ├── domain/              # 领域层（业务逻辑）
│   │   ├── shared/          # 共享领域概念
│   │   ├── vector_search/   # 向量搜索领域
│   │   ├── document_parsing/# 文档解析领域
│   │   └── knowledge_base/  # 知识库领域
│   ├── application/         # 应用层（用例）
│   │   ├── shared/          # 共享应用概念
│   │   ├── vector_search/   # 向量搜索用例
│   │   ├── document_parsing/# 文档解析用例
│   │   └── knowledge_base/  # 知识库用例
│   ├── infrastructure/      # 基础设施层（技术实现）
│   │   ├── database/        # 数据库实现
│   │   ├── vector_db/       # 向量数据库适配器
│   │   ├── parsers/         # 文档解析器
│   │   ├── external_services/# 外部服务集成
│   │   └── file_storage/    # 文件存储
│   ├── presentation/        # 表现层（API）
│   │   ├── api/             # API 路由和中间件
│   │   └── schemas/         # 请求/响应模型
│   ├── config/              # 配置管理
│   └── main.py              # 应用入口
├── tests/                   # 测试代码
│   ├── unit/                # 单元测试
│   ├── integration/         # 集成测试
│   ├── e2e/                 # 端到端测试
│   └── fixtures/            # 测试夹具
├── docs/                    # 文档
├── scripts/                 # 工具脚本
├── docker/                  # Docker 配置
└── .github/                 # GitHub 配置

详细的目录结构说明请参考 Directory Structure。

开发工作流

1. 创建功能分支

# 从 main 分支创建新分支
git checkout main
git pull origin main
git checkout -b feature/your-feature-name

2. 开发功能

遵循 Clean Architecture 原则：

1. 领域层：定义实体、值对象、领域服务
2. 应用层：实现用例、命令/查询处理器
3. 基础设施层：实现仓储、适配器
4. 表现层：实现 API 端点

3. 编写测试

为每个层编写相应的测试：

# 运行测试
pytest

# 运行特定测试
pytest tests/unit/domain/

# 运行并查看覆盖率
pytest --cov=src --cov-report=html

4. 代码质量检查

# 格式化代码
black src tests
isort src tests

# 代码检查
flake8 src tests

# 类型检查
mypy src

# 安全检查
safety check
bandit -r src

5. 提交代码

使用规范的提交信息：

git add .
git commit -m "feat: add new feature"

提交信息格式：

• feat: 新功能
• fix: 修复 bug
• docs: 文档更新
• style: 代码格式（不影响功能）
• refactor: 重构
• test: 测试相关
• chore: 构建/工具相关

6. 推送并创建 PR

git push origin feature/your-feature-name

然后在 GitHub 上创建 Pull Request。

代码规范

Python 代码风格

遵循 PEP 8 规范：

# 好的示例
def calculate_similarity(vector_a: List[float], vector_b: List[float]) -> float:
    """
    Calculate cosine similarity between two vectors.
    
    Args:
        vector_a: First vector
        vector_b: Second vector
        
    Returns:
        Similarity score between 0 and 1
        
    Raises:
        ValueError: If vectors have different dimensions
    """
    if len(vector_a) != len(vector_b):
        raise ValueError("Vectors must have same dimension")
    
    # Implementation
    return similarity_score

类型提示

所有函数都应该有类型提示：

from typing import List, Optional, Dict, Any

def process_document(
    content: str,
    metadata: Optional[Dict[str, Any]] = None
) -> List[str]:
    """Process document and return chunks."""
    pass

文档字符串

使用 Google 风格的文档字符串：

def search_documents(query: str, limit: int = 10) -> List[Document]:
    """
    Search documents using hybrid search.
    
    This function combines vector search and full-text search
    to provide better search results.
    
    Args:
        query: Search query string
        limit: Maximum number of results to return
        
    Returns:
        List of matching documents sorted by relevance
        
    Raises:
        ValueError: If query is empty
        DatabaseError: If database connection fails
        
    Example:
        >>> results = search_documents("machine learning", limit=5)
        >>> len(results)
        5
    """
    pass

命名约定

• 类名: PascalCase (DocumentParser, VectorDatabase)
• 函数名: snake_case (parse_document, search_vectors)
• 常量: UPPER_SNAKE_CASE (MAX_CHUNK_SIZE, DEFAULT_LIMIT)
• 私有成员: 前缀下划线 (_internal_method, _cache)

导入顺序

# 1. 标准库
import os
import sys
from typing import List, Optional

# 2. 第三方库
import numpy as np
from fastapi import FastAPI, HTTPException

# 3. 本地模块
from src.domain.entities import Document
from src.application.handlers import SearchHandler

测试指南

测试结构

tests/
├── unit/                    # 单元测试（快速、隔离）
│   ├── domain/             # 领域层测试
│   ├── application/        # 应用层测试
│   └── infrastructure/     # 基础设施层测试
├── integration/            # 集成测试（组件交互）
│   ├── api/               # API 集成测试
│   └── database/          # 数据库集成测试
└── e2e/                   # 端到端测试（完整流程）

编写单元测试

import pytest
from src.domain.vector_search.entities import Document

def test_document_creation():
    """Test document entity creation."""
    doc = Document(
        id="doc_123",
        content="Test content",
        metadata={"title": "Test"}
    )
    
    assert doc.id == "doc_123"
    assert doc.content == "Test content"
    assert doc.metadata["title"] == "Test"

def test_document_validation():
    """Test document validation."""
    with pytest.raises(ValueError):
        Document(id="", content="Test")  # Empty ID should fail

使用 Fixtures

import pytest
from src.infrastructure.database.session import get_session

@pytest.fixture
def db_session():
    """Provide a database session for testing."""
    session = get_session()
    yield session
    session.rollback()
    session.close()

@pytest.fixture
def sample_document():
    """Provide a sample document for testing."""
    return Document(
        id="test_doc",
        content="Sample content",
        metadata={"title": "Sample"}
    )

def test_save_document(db_session, sample_document):
    """Test saving document to database."""
    repository = DocumentRepository(db_session)
    saved_doc = repository.save(sample_document)
    assert saved_doc.id == sample_document.id

使用 Mock

from unittest.mock import Mock, patch

def test_search_with_mock():
    """Test search with mocked vector database."""
    mock_vector_db = Mock()
    mock_vector_db.search.return_value = [
        {"id": "doc_1", "score": 0.9},
        {"id": "doc_2", "score": 0.8}
    ]
    
    service = SearchService(vector_db=mock_vector_db)
    results = service.search("test query")
    
    assert len(results) == 2
    mock_vector_db.search.assert_called_once()

属性测试（Property-Based Testing）

使用 Hypothesis 进行属性测试：

from hypothesis import given, strategies as st

@given(st.lists(st.floats(min_value=0, max_value=1), min_size=1, max_size=1000))
def test_vector_normalization(vector):
    """Test that vector normalization always produces unit vectors."""
    normalized = normalize_vector(vector)
    magnitude = sum(x**2 for x in normalized) ** 0.5
    assert abs(magnitude - 1.0) < 1e-6

运行测试

# 运行所有测试
pytest

# 运行特定测试文件
pytest tests/unit/domain/test_entities.py

# 运行特定测试函数
pytest tests/unit/domain/test_entities.py::test_document_creation

# 运行特定标记的测试
pytest -m unit          # 只运行单元测试
pytest -m integration   # 只运行集成测试
pytest -m "not slow"    # 跳过慢速测试

# 并行运行测试
pytest -n auto

# 查看覆盖率
pytest --cov=src --cov-report=html
open htmlcov/index.html

调试技巧

使用 Python 调试器

# 在代码中设置断点
import pdb; pdb.set_trace()

# 或使用 breakpoint()（Python 3.7+）
breakpoint()

VS Code 调试配置

创建 .vscode/launch.json：

{
    "version": "0.2.0",
    "configurations": [
        {
            "name": "Python: FastAPI",
            "type": "python",
            "request": "launch",
            "module": "uvicorn",
            "args": [
                "src.main:app",
                "--reload"
            ],
            "jinja": true,
            "justMyCode": false
        },
        {
            "name": "Python: Pytest",
            "type": "python",
            "request": "launch",
            "module": "pytest",
            "args": [
                "-v"
            ],
            "console": "integratedTerminal",
            "justMyCode": false
        }
    ]
}

日志调试

import logging

logger = logging.getLogger(__name__)

def process_document(doc: Document):
    logger.debug(f"Processing document: {doc.id}")
    logger.info(f"Document size: {len(doc.content)}")
    
    try:
        result = parse(doc)
        logger.info(f"Successfully parsed document: {doc.id}")
        return result
    except Exception as e:
        logger.error(f"Failed to parse document: {doc.id}", exc_info=True)
        raise

性能分析

import cProfile
import pstats

# 分析函数性能
profiler = cProfile.Profile()
profiler.enable()

# 运行代码
result = expensive_function()

profiler.disable()
stats = pstats.Stats(profiler)
stats.sort_stats('cumulative')
stats.print_stats(10)  # 显示前 10 个最慢的函数

贡献指南

提交 Pull Request

1. Fork 仓库
2. 创建功能分支
3. 编写代码和测试

4. 运行所有检查:

   python scripts/run_tests.py --all
   

5. 提交代码

6. 推送到你的 fork

7. 创建 Pull Request

PR 检查清单

• 代码遵循项目规范
• 所有测试通过
• 测试覆盖率 ≥ 80%
• 添加了必要的文档
• 更新了 CHANGELOG（如果适用）
• 提交信息清晰明确

Code Review 流程

1. 自动检查: CI/CD 自动运行测试和检查
2. 人工审查: 至少一位维护者审查代码
3. 反馈处理: 根据反馈修改代码
4. 合并: 审查通过后合并到 main 分支

报告 Bug

使用 GitHub Issues 报告 bug，包含：

• 描述: 清晰描述问题
• 重现步骤: 如何重现问题
• 期望行为: 应该发生什么
• 实际行为: 实际发生了什么
• 环境信息: Python 版本、操作系统等
• 日志: 相关的错误日志

提出功能请求

使用 GitHub Issues 提出功能请求，包含：

• 用例: 为什么需要这个功能
• 建议方案: 如何实现
• 替代方案: 其他可能的实现方式
• 影响: 对现有功能的影响

常用命令

开发命令

# 启动开发服务器
uvicorn src.main:app --reload

# 运行测试
pytest

# 代码格式化
black src tests
isort src tests

# 代码检查
flake8 src tests
mypy src

# 生成覆盖率报告
pytest --cov=src --cov-report=html

# 运行所有检查
python scripts/run_tests.py --all

数据库命令

# 创建迁移
alembic revision --autogenerate -m "描述"

# 应用迁移
alembic upgrade head

# 回滚迁移
alembic downgrade -1

# 查看迁移历史
alembic history

Docker 命令

# 启动服务
docker-compose up -d

# 查看日志
docker-compose logs -f

# 停止服务
docker-compose down

# 重建镜像
docker-compose build --no-cache

开发工具推荐

IDE/编辑器

• VS Code: 推荐插件

  • Python
  • Pylance
  • Python Test Explorer
  • GitLens
  • Docker

• PyCharm: 专业的 Python IDE

命令行工具

• httpie: 更友好的 HTTP 客户端

  pip install httpie
  http POST localhost:8000/api/v1/documents/ content="test"
  

• jq: JSON 处理工具

  curl localhost:8000/api/v1/documents/ | jq '.data'
  

数据库工具

• DBeaver: 通用数据库客户端
• pgAdmin: PostgreSQL 管理工具
• TablePlus: 现代数据库客户端

学习资源

Clean Architecture

• Clean Architecture by Robert C. Martin
• Domain-Driven Design

FastAPI

• FastAPI Documentation
• FastAPI Best Practices

Testing

• pytest Documentation
• Hypothesis Documentation

Python

• PEP 8 Style Guide
• Python Type Hints

获取帮助

• 文档: 查看 docs/ 目录
• Issues: GitHub Issues
• Discussions: GitHub Discussions
• Email: dev@example.com

下一步

• 阅读 架构文档 了解系统设计
• 查看 API 文档 了解 API 接口
• 参考 测试指南 编写测试
• 遵循 配置指南 配置环境

祝你开发愉快！🚀