资源
开发
参与MRRA贡献和开发指南
开发
本指南涵盖开发工作流程、贡献指南和参与MRRA贡献的最佳实践。
开发环境设置
环境要求
- Python 3.10+（推荐Python 3.11）
- Git 版本控制
- 虚拟环境 管理（venv、conda或poetry）
克隆和设置
# 克隆仓库
git clone https://github.com/your-org/mrra.git
cd mrra
# 创建虚拟环境
python -m venv mrra-dev
source mrra-dev/bin/activate # Windows: mrra-dev\Scripts\activate
# 开发模式安装
pip install -e .[dev]
# 安装pre-commit钩子
pre-commit install
# 验证安装
python -c "import mrra; print(f'MRRA版本: {mrra.__version__}')"
pytest tests/ -v开发依赖
开发安装包含额外工具：
# 开发依赖（包含在pip install -e .[dev]中）
dev_dependencies = [
# 测试
'pytest>=7.0.0',
'pytest-cov>=4.0.0',
'pytest-asyncio>=0.21.0',
'pytest-mock>=3.10.0',
# 代码质量
'black>=23.0.0',
'isort>=5.12.0',
'flake8>=6.0.0',
'mypy>=1.0.0',
# 文档
'sphinx>=6.0.0',
'sphinx-rtd-theme>=1.2.0',
'myst-parser>=1.0.0',
# 开发工具
'pre-commit>=3.0.0',
'jupyter>=1.0.0',
'ipdb>=0.13.0',
]项目结构
理解MRRA项目结构：
mrra/
├── src/mrra/ # 主包
│ ├── core/ # 核心类型和接口
│ │ ├── types.py # 数据类型和协议
│ │ ├── config.py # 配置类
│ │ └── exceptions.py # 自定义异常
│ ├── data/ # 数据处理模块
│ │ ├── trajectory.py # TrajectoryBatch类
│ │ └── activity.py # 活动提取
│ ├── analysis/ # 分析模块
│ │ └── activity_purpose.py # 目的分配
│ ├── graph/ # 图相关模块
│ │ ├── mobility_graph.py # MobilityGraph类
│ │ └── pattern.py # 模式生成
│ ├── retriever/ # 检索系统
│ │ └── graph_rag.py # GraphRAG实现
│ ├── agents/ # 智能体系统
│ │ ├── builder.py # 智能体构建器
│ │ └── subagents.py # 子智能体实现
│ ├── persist/ # 持久化和缓存
│ │ └── cache.py # 缓存管理器
│ └── tools/ # MCP工具和实用工具
│ ├── weather.py # 天气工具
│ └── maps.py # 地图工具
├── tests/ # 测试套件
│ ├── unit/ # 单元测试
│ ├── integration/ # 集成测试
│ └── fixtures/ # 测试固件
├── docs/ # 文档
├── scripts/ # 实用脚本
├── examples/ # 示例代码
└── pyproject.toml # 项目配置代码风格和标准
代码格式化
MRRA使用标准化代码格式化工具：
Black 用于代码格式化：
# 格式化所有代码
black src/ tests/ scripts/
# 检查格式化而不更改
black --check src/ tests/ scripts/pyproject.toml中的配置：
[tool.black]
line-length = 88
target-version = ['py310']
include = '\.pyi?$'
extend-exclude = '''
/(
\.eggs
| \.git
| \.hg
| \.mypy_cache
| \.tox
| \.venv
| _build
| buck-out
| build
| dist
)/
'''isort 用于import排序：
# 排序imports
isort src/ tests/ scripts/
# 检查import排序
isort --check-only src/ tests/ scripts/pyproject.toml中的配置：
[tool.isort]
profile = "black"
multi_line_output = 3
line_length = 88
known_first_party = ["mrra"]flake8 用于代码检查：
# 运行检查
flake8 src/ tests/ scripts/.flake8中的配置：
[flake8]
max-line-length = 88
extend-ignore = E203, W503, E501
exclude =
.git,
__pycache__,
.venv,
build,
distmypy 用于类型检查：
# 类型检查
mypy src/mrra/pyproject.toml中的配置：
[tool.mypy]
python_version = "3.10"
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true
disallow_incomplete_defs = true
check_untyped_defs = true
disallow_untyped_decorators = true
no_implicit_optional = true
warn_redundant_casts = true
warn_unused_ignores = true
warn_no_return = true
warn_unreachable = true
strict_equality = true测试框架
测试结构
MRRA使用pytest进行全面的测试覆盖：
# tests/unit/test_trajectory.py
import pytest
import pandas as pd
from mrra.data.trajectory import TrajectoryBatch
from mrra.core.exceptions import ValidationError
class TestTrajectoryBatch:
"""TrajectoryBatch类的单元测试"""
@pytest.fixture
def sample_data(self):
"""测试用的样本轨迹数据"""
return pd.DataFrame({
'user_id': ['user_1', 'user_1', 'user_2', 'user_2'],
'timestamp': [
'2024-01-01 08:00:00', '2024-01-01 12:00:00',
'2024-01-01 09:00:00', '2024-01-01 13:00:00'
],
'latitude': [31.2304, 31.2404, 31.2354, 31.2454],
'longitude': [121.4737, 121.4837, 121.4787, 121.4887]
})
def test_init_valid_data(self, sample_data):
"""使用有效数据测试TrajectoryBatch初始化"""
tb = TrajectoryBatch(sample_data)
assert len(tb.df) == 4
assert len(tb.users()) == 2
assert 'timestamp_local' in tb.df.columns
assert 'hour' in tb.df.columns
assert 'dow' in tb.df.columns
def test_init_missing_columns(self):
"""使用缺少必需列测试TrajectoryBatch"""
invalid_data = pd.DataFrame({
'user_id': ['user_1'],
'latitude': [31.2304],
# 缺少'longitude'和'timestamp'
})
with pytest.raises(ValidationError, match="Missing required columns"):
TrajectoryBatch(invalid_data)运行测试
# 运行所有测试
pytest
# 运行特定测试类别
pytest tests/unit/ # 仅单元测试
pytest tests/integration/ # 仅集成测试
# 带覆盖率运行
pytest --cov=mrra --cov-report=html
# 使用LLM运行集成测试（需要API密钥）
MRRA_INTEGRATION_WITH_LLM=1 OPENAI_API_KEY=your_key pytest tests/integration/
# 运行特定测试
pytest tests/unit/test_trajectory.py::TestTrajectoryBatch::test_init_valid_data
# 详细输出运行测试
pytest -v -s
# 并行运行测试（使用pytest-xdist）
pip install pytest-xdist
pytest -n auto贡献指南
开发工作流程
-
Fork和克隆
git clone https://github.com/yourusername/mrra.git cd mrra git remote add upstream https://github.com/original-org/mrra.git -
创建功能分支
git checkout -b feature/your-feature-name -
开发和测试
# 进行更改 # 添加测试 pytest tests/ # 检查代码质量 black src/ tests/ isort src/ tests/ flake8 src/ tests/ mypy src/mrra/ -
提交更改
git add . git commit -m "feat: 添加新的移动性预测算法 - 实现下一位置预测的新算法 - 添加全面的测试和文档 - 在测试数据集上提高15%的准确性 " -
提交Pull Request
git push origin feature/your-feature-name # 通过GitHub界面创建PR
提交信息约定
使用约定式提交格式：
type(scope): description
body (optional)
footer (optional)类型：
feat: 新功能fix: 错误修复docs: 文档更改style: 代码风格更改（格式化等）refactor: 代码重构test: 添加或更新测试chore: 维护任务
示例：
feat(agents): 添加多轮反思能力
fix(cache): 解决缓存失效问题
docs(api): 更新TrajectoryBatch的文档字符串
test(integration): 添加全面的工作流程测试
refactor(graph): 优化图构建性能代码审查流程
审查清单：
- 代码遵循风格指南（black、isort、flake8）
- 所有测试通过（单元和集成）
- 新功能包含全面的测试
- API更改的文档已更新
- 为公共API提供类型提示
- 考虑了性能影响
- 保持向后兼容性
添加新功能
功能开发模板
添加新功能时，遵循此模板：
# src/mrra/new_module/feature.py
"""新功能实现。
此模块提供[功能描述]。
"""
from typing import Any, Dict, List, Optional, Protocol
import logging
from dataclasses import dataclass
from mrra.core.types import TrajectoryBatch
from mrra.core.exceptions import MRRAError
logger = logging.getLogger(__name__)
@dataclass
class FeatureConfig:
"""新功能的配置。
Attributes:
param1: 参数1的描述
param2: 参数2的描述
"""
param1: str
param2: int = 10
enable_advanced: bool = False
class FeatureError(MRRAError):
"""功能操作失败时抛出。"""
pass
class NewFeature:
"""新功能的实现。
此类提供[详细描述]。
Example:
>>> feature = NewFeature(config)
>>> result = feature.process(data)
"""
def __init__(self, config: FeatureConfig):
"""使用配置初始化功能。
Args:
config: 功能配置对象
"""
self.config = config
self._validate_config()
def _validate_config(self) -> None:
"""验证配置参数。"""
if self.config.param2 <= 0:
raise FeatureError("param2必须为正数")
def process(self, data: TrajectoryBatch) -> Dict[str, Any]:
"""使用新功能处理轨迹数据。
Args:
data: 要处理的轨迹数据
Returns:
包含处理结果的字典
Raises:
FeatureError: 如果处理失败
"""
try:
logger.info(f"使用{len(data.df)}个点处理数据")
# 实现在这里
results = self._internal_process(data)
logger.info("处理成功完成")
return results
except Exception as e:
logger.error(f"处理失败: {e}")
raise FeatureError(f"功能处理失败: {e}") from e
def _internal_process(self, data: TrajectoryBatch) -> Dict[str, Any]:
"""内部处理逻辑。"""
# 实现细节
return {"processed": True}性能优化
性能分析和基准测试
# scripts/benchmark.py
"""MRRA性能基准测试脚本。"""
import time
import cProfile
import pstats
from functools import wraps
from typing import Callable, Any
def profile_function(func: Callable) -> Callable:
"""分析函数性能的装饰器。"""
@wraps(func)
def wrapper(*args, **kwargs) -> Any:
profiler = cProfile.Profile()
start_time = time.time()
profiler.enable()
try:
result = func(*args, **kwargs)
finally:
profiler.disable()
end_time = time.time()
# 打印时间信息
print(f"{func.__name__} 耗时 {end_time - start_time:.2f} 秒")
# 打印分析统计
stats = pstats.Stats(profiler)
stats.sort_stats('cumulative')
stats.print_stats(10) # 前10个函数
return result
return wrapper性能指南：
- 在优化之前先分析 - 测量实际瓶颈
- 广泛使用缓存进行昂贵的操作
- 考虑大数据集的内存使用
- 通过批处理和并发优化LLM调用
- 开发期间监控API成本
- 使用适当的数据结构（pandas vs lists vs sets）
发布流程
版本管理
MRRA使用语义版本控制（semver）：
# src/mrra/__init__.py
__version__ = "0.2.1"
# Major.Minor.Patch
# Major: 破坏性更改
# Minor: 新功能，向后兼容
# Patch: 错误修复，向后兼容