入门指南
配置
LLM提供商、检索系统和MCP服务的高级配置选项
配置
MRRA为LLM提供商、检索系统、缓存和MCP服务提供了灵活的配置选项。本指南详细介绍所有配置方面。
LLM提供商配置
支持的提供商
MRRA通过统一接口支持多个LLM提供商：
# OpenAI 配置
openai_cfg = dict(
provider='openai',
model='gpt-4o-mini', # 或 gpt-4, gpt-3.5-turbo
api_key='your-openai-api-key',
temperature=0.2,
max_tokens=1000,
timeout=60
)
# 使用自定义base URL（用于代理）
openai_proxy_cfg = dict(
provider='openai',
model='gpt-4o-mini',
api_key='your-api-key',
base_url='https://your-proxy.com/v1',
temperature=0.2
)# 千问配置（阿里巴巴DashScope）
qwen_cfg = dict(
provider='openai-compatible', # 使用OpenAI兼容接口
model='qwen-plus', # 或 qwen-turbo, qwen-max
base_url='https://dashscope.aliyuncs.com/compatible-mode/v1',
api_key='your-dashscope-api-key',
temperature=0.2,
max_tokens=2000
)
# 替代的直接配置
qwen_direct_cfg = dict(
provider='qwen',
model='qwen-plus',
api_key='your-dashscope-api-key',
temperature=0.2
)# SiliconFlow 配置
siliconflow_cfg = dict(
provider='openai-compatible',
model='deepseek-chat', # 或其他可用模型
base_url='https://api.siliconflow.cn/v1',
api_key='your-siliconflow-api-key',
temperature=0.2,
max_tokens=1500
)# DeepInfra 配置
deepinfra_cfg = dict(
provider='openai-compatible',
model='meta-llama/Meta-Llama-3.1-8B-Instruct',
base_url='https://api.deepinfra.com/v1/openai',
api_key='your-deepinfra-api-key',
temperature=0.2,
max_tokens=2000
)# 自定义提供商配置
custom_cfg = dict(
provider='openai-compatible', # 使用OpenAI兼容接口
model='your-model-name',
base_url='https://your-api-endpoint.com/v1',
api_key='your-api-key',
headers={'Custom-Header': 'value'}, # 可选的自定义头部
temperature=0.2,
max_tokens=1000,
timeout=120
)环境变量配置
设置环境变量以进行安全配置：
# OpenAI
export MRRA_OPENAI_API_KEY="your-openai-key"
export MRRA_OPENAI_MODEL="gpt-4o-mini"
# 千问
export MRRA_QWEN_API_KEY="your-dashscope-key"
export MRRA_QWEN_MODEL="qwen-plus"
# SiliconFlow
export MRRA_SILICONFLOW_API_KEY="your-siliconflow-key"
# 默认提供商
export MRRA_DEFAULT_PROVIDER="openai"
export MRRA_DEFAULT_MODEL="gpt-4o-mini"在配置中使用环境变量：
import os
def create_llm_config_from_env():
"""从环境变量创建LLM配置"""
provider = os.getenv('MRRA_DEFAULT_PROVIDER', 'openai')
if provider == 'openai':
return dict(
provider='openai',
model=os.getenv('MRRA_OPENAI_MODEL', 'gpt-4o-mini'),
api_key=os.getenv('MRRA_OPENAI_API_KEY'),
temperature=float(os.getenv('MRRA_TEMPERATURE', '0.2'))
)
elif provider == 'qwen':
return dict(
provider='openai-compatible',
model=os.getenv('MRRA_QWEN_MODEL', 'qwen-plus'),
base_url='https://dashscope.aliyuncs.com/compatible-mode/v1',
api_key=os.getenv('MRRA_QWEN_API_KEY'),
temperature=float(os.getenv('MRRA_TEMPERATURE', '0.2'))
)
else:
raise ValueError(f"不支持的提供商: {provider}")
# 使用方法
llm_cfg = create_llm_config_from_env()活动提取配置
参数指南
根据数据特征配置活动提取：
# 城市移动性的默认配置
default_cfg = dict(
method="radius", # 'radius' 或 'grid'
radius_m=300, # 检测半径（米）
min_dwell_minutes=30, # 最小停留时间
max_gap_minutes=90, # 点之间的最大间隙
grid_size_m=200 # 网格单元大小（网格方法）
)
# 不同场景的配置
scenarios = {
'urban_detailed': dict(
method="radius",
radius_m=200, # 密集城市区域更小的半径
min_dwell_minutes=20, # 更短的最小停留时间
max_gap_minutes=60, # 更紧的间隙容忍度
grid_size_m=150
),
'suburban': dict(
method="radius",
radius_m=500, # 郊区更大的半径
min_dwell_minutes=45, # 更长的停留时间
max_gap_minutes=120, # 对间隙更宽松
grid_size_m=300
),
'sparse_data': dict(
method="grid", # 网格方法对稀疏数据更稳健
radius_m=800,
min_dwell_minutes=15, # 稀疏数据的较低阈值
max_gap_minutes=180, # 更高容忍度
grid_size_m=500
),
'high_frequency': dict(
method="radius",
radius_m=100, # 高频数据非常细粒度
min_dwell_minutes=10, # 频繁采样的短停留
max_gap_minutes=30, # 紧密间隙
grid_size_m=100
)
}
# 根据数据类型选择
data_type = 'urban_detailed' # 根据需要更改
ext_cfg = scenarios[data_type]
acts = ActivityExtractor(tb, **ext_cfg).extract()自适应配置
基于数据分析自动配置：
def auto_configure_extraction(tb):
"""根据轨迹特征自动配置提取"""
df = tb.df
# 分析数据特征
avg_points_per_user = len(df) / df['user_id'].nunique()
time_span = (df['timestamp_local'].max() - df['timestamp_local'].min()).total_seconds() / 3600 # 小时
# 计算典型采样频率
sampling_freq_minutes = time_span * 60 / len(df) if len(df) > 0 else 60
# 估计典型移动距离
distances = []
for user_id in df['user_id'].unique()[:5]: # 取前5个用户样本
user_df = df[df['user_id'] == user_id].sort_values('timestamp_local')
for i in range(1, min(100, len(user_df))): # 最多取100个点
prev = user_df.iloc[i-1]
curr = user_df.iloc[i]
lat_diff = curr['latitude'] - prev['latitude']
lon_diff = curr['longitude'] - prev['longitude']
distance = ((lat_diff**2 + lon_diff**2)**0.5) * 111000 # 近似米数
distances.append(distance)
median_distance = sorted(distances)[len(distances)//2] if distances else 300
# 基于分析配置
if avg_points_per_user > 2000 and sampling_freq_minutes < 5:
# 高频数据
config = dict(
method="radius",
radius_m=max(100, median_distance * 0.5),
min_dwell_minutes=max(5, sampling_freq_minutes * 2),
max_gap_minutes=max(30, sampling_freq_minutes * 10),
grid_size_m=max(100, median_distance * 0.3)
)
elif avg_points_per_user < 200:
# 稀疏数据
config = dict(
method="grid",
radius_m=max(500, median_distance * 2),
min_dwell_minutes=max(15, sampling_freq_minutes * 0.5),
max_gap_minutes=max(120, sampling_freq_minutes * 5),
grid_size_m=max(300, median_distance * 1.5)
)
else:
# 标准数据
config = dict(
method="radius",
radius_m=max(200, median_distance),
min_dwell_minutes=max(20, sampling_freq_minutes),
max_gap_minutes=max(60, sampling_freq_minutes * 3),
grid_size_m=max(150, median_distance * 0.7)
)
print(f"自动配置提取参数: {config}")
print(f"基于: {avg_points_per_user:.0f}平均点数/用户, {sampling_freq_minutes:.1f}分钟采样, {median_distance:.0f}米中位距离")
return config
# 使用自适应配置
ext_cfg = auto_configure_extraction(tb)
acts = ActivityExtractor(tb, **ext_cfg).extract()图配置
图构建选项
from mrra.graph.mobility_graph import GraphConfig
# 基础配置
basic_cfg = GraphConfig(
grid_size_m=200, # 网格单元大小
min_dwell_minutes=5, # 图节点的最小停留时间
use_activities=True # 使用基于活动的构建
)
# 高级配置
advanced_cfg = GraphConfig(
grid_size_m=200,
min_dwell_minutes=5,
use_activities=True,
include_transitions=True, # 包含位置到位置的边
purpose_transitions=True, # 包含目的到目的的边
temporal_granularity='both', # 'hour', 'timebin', 或 'both'
weight_by_duration=True, # 按花费时间加权边
max_edge_weight=1000, # 边权重上限
normalize_weights=True # 标准化边权重
)
# 大数据集的内存优化配置
memory_optimized_cfg = GraphConfig(
grid_size_m=500, # 更大的网格减少节点数
min_dwell_minutes=10, # 更高阈值减少边数
use_activities=True,
include_transitions=False, # 跳过转换边以节省内存
purpose_transitions=False,
temporal_granularity='hour', # 仅使用小时节点，不使用时间段
max_nodes=10000 # 限制总节点数
)检索配置
配置GraphRAG检索权重和参数：
from mrra.retriever.graph_rag import GraphRAGGenerate
# 使用自定义权重创建检索器
retriever = GraphRAGGenerate(tb=tb, mobility_graph=mg)
# 配置检索权重
retriever.purpose_weight = 0.6 # 目的上下文权重
retriever.hour_weight = 0.5 # 小时上下文权重
retriever.dow_weight = 0.3 # 星期几权重
retriever.recent_weight = 0.2 # 最近位置权重
retriever.user_weight = 1.0 # 用户上下文权重
# 配置检索参数
retriever.max_results = 50 # 考虑的最大位置数
retriever.min_score_threshold = 0.01 # 最小相关性得分
retriever.diversification_factor = 0.3 # 结果多样化（0=无多样化，1=最大多样化）
# 任务特定配置
task_configs = {
'next_position': {
'purpose_weight': 0.4,
'hour_weight': 0.6,
'dow_weight': 0.4,
'recent_weight': 0.8, # 最近上下文高权重
},
'future_position': {
'purpose_weight': 0.6,
'hour_weight': 0.5,
'dow_weight': 0.3,
'recent_weight': 0.3, # 最近上下文低权重
},
'full_day_traj': {
'purpose_weight': 0.7, # 目的模式高权重
'hour_weight': 0.4,
'dow_weight': 0.5,
'recent_weight': 0.2,
}
}
def configure_retriever_for_task(retriever, task):
"""根据预测任务配置检索器权重"""
if task in task_configs:
config = task_configs[task]
for param, value in config.items():
setattr(retriever, param, value)
return retriever缓存配置
缓存目录和设置
from mrra.persist.cache import CacheManager
# 默认缓存配置
default_cache = CacheManager() # 使用 .mrra_cache/
# 自定义缓存目录
custom_cache = CacheManager(base_dir='/path/to/custom/cache')
# 基于环境的缓存配置
import os
cache_dir = os.getenv('MRRA_CACHE_DIR', '.mrra_cache')
cache_manager = CacheManager(base_dir=cache_dir)多智能体反思配置
智能体配置模板
# 不同用例的模板配置
agent_templates = {
'basic': dict(
max_round=1,
subAgents=[
{"name": "temporal", "prompt": "专注于时间模式并从选项中选择。"},
{"name": "spatial", "prompt": "专注于位置模式并从选项中选择。"},
],
aggregator="confidence_weighted_voting"
),
'comprehensive': dict(
max_round=2,
subAgents=[
{"name": "temporal", "prompt": "分析时间模式包括一天中的时间和星期几偏好。"},
{"name": "spatial", "prompt": "分析空间模式包括聚类和可达性。"},
{"name": "purpose", "prompt": "分析活动目的和序列逻辑。"},
{"name": "routing", "prompt": "分析移动效率和转换模式。"},
],
aggregator="confidence_weighted_voting",
consensus_threshold=0.6
),
'fast': dict(
max_round=1,
subAgents=[
{"name": "combined", "prompt": "快速考虑所有因素并从选项中选择。"}
],
aggregator="simple_selection"
)
}
def get_agent_config(template_name, custom_overrides=None):
"""从模板获取智能体配置，可选覆盖"""
if template_name not in agent_templates:
raise ValueError(f"未知模板: {template_name}")
config = agent_templates[template_name].copy()
if custom_overrides:
config.update(custom_overrides)
return config
# 使用方法
config = get_agent_config('comprehensive', {
'max_round': 1, # 覆盖以加速
'consensus_threshold': 0.8 # 需要更高共识
})配置最佳实践：
- 从默认配置开始，根据数据特征进行调整
- 使用环境变量保存敏感信息如API密钥
- 在运行昂贵操作前验证配置
- 监控性能并相应调整参数
- 广泛使用缓存避免重复计算
安全考虑：
- 永远不要将API密钥或敏感数据提交到版本控制
- 对所有秘密使用环境变量
- 为配置文件实施适当的访问控制
- 定期轮换API密钥并更新配置