核心概念
移动图
异构移动图的概念、结构和使用
移动图
移动图是使用多类型异构图（networkx.MultiDiGraph）对"人-位置-时间-目的"关系的统一表示。它能实现移动模式的检索、推理和可解释分析。
图结构
节点类型
移动图包含多种类型的节点，每种代表移动性的不同方面：
格式： u_<user_id>
代表系统中的个体用户。
# 用户节点示例
"u_user_123"
"u_user_456" 格式： g_<gy>_<gx> （基于网格）
属性：
gy,gx：网格坐标lat,lon：近似坐标（使用活动构建时）
# 位置节点示例
"g_1234_5678" # y=1234, x=5678处的网格单元
"g_2345_6789" # 另一个网格单元小时节点： h_<0..23>
星期几节点： d_<0..6> （0=周一）
时间段节点： t_<hour>_<dow>
# 时间节点示例
"h_9" # 上午9点
"d_1" # 周二
"t_9_1" # 周二上午9点格式： p_<purpose>
属性：
name：目的描述
# 目的节点示例
"p_work" # 工作活动
"p_dining" # 用餐活动
"p_home" # 家庭/住宅活动边类型和权重
图包含多种类型的边，代表不同的关系：
用户-位置边
- 方向：用户 → 位置
- 权重：访问强度（基于点构建时为计数；基于活动构建时为分钟）
- 属性：
activity_type、activity_purpose
时间共现边
- 方向：双向（loc ↔ hour/dow/timebin）
- 权重：共现强度（分钟/计数）
- 目的：捕捉位置通常被访问的时间
目的-上下文边
- 方向：双向（purpose ↔ loc/hour/dow/timebin）
- 权重：共现强度（分钟）
- 目的：将目的与空间和时间上下文连接
用户-目的边
- 方向：用户 → 目的
- 权重：在该目的上花费的总时间（分钟）
- 目的：用户活动画像
转换边
- 位置转换：loc → loc（相邻活动转换）
- 目的转换：purpose → purpose（相邻活动目的转换）
- 属性：包含用户信息用于个性化分析
图构建
基本构庻
from mrra.graph.mobility_graph import MobilityGraph, GraphConfig
# 配置图参数
cfg = GraphConfig(
grid_size_m=200, # 网格单元大小（米）
min_dwell_minutes=5, # 活动的最小停留时间
use_activities=True # 使用基于活动的构庻
)
# 构庻图
mg = MobilityGraph(tb, cfg, activities=acts, assume_purposes_assigned=True)
G = mg.G # 访问networkx图配置选项
# 详细配置示例
cfg = GraphConfig(
grid_size_m=200, # 网格分辨率
min_dwell_minutes=5, # 活动检测阈值
use_activities=True, # 使用活动 vs 原始点
include_transitions=True, # 包含loc→loc边
purpose_transitions=True, # 包含purpose→purpose边
temporal_granularity='hour', # 'hour'、'timebin'或'both'
weight_by_duration=True # 按花费时间加权边
)使用自定义活动的高级构庻
# 使用带目的的预计算活动
from mrra.data.activity import ActivityExtractor
from mrra.analysis.activity_purpose import ActivityPurposeAssigner
# 提取并分配目的到活动
ext_cfg = dict(method="radius", radius_m=300, min_dwell_minutes=30)
acts = ActivityExtractor(tb, **ext_cfg).extract()
acts = ActivityPurposeAssigner(tb, llm=llm, concurrency=8).assign(acts)
# 使用目的丰富的活动构庻图
cfg = GraphConfig(grid_size_m=200, use_activities=True)
mg = MobilityGraph(tb, cfg, activities=acts, assume_purposes_assigned=True)图分析
基本图统计
# 获取基本图信息
print(f"节点数: {G.number_of_nodes()}")
print(f"边数: {G.number_of_edges()}")
# 分析节点类型
node_types = {}
for node in G.nodes():
node_type = node.split('_')[0]
node_types[node_type] = node_types.get(node_type, 0) + 1
print("节点类型分布:", node_types)中心性分析
import networkx as nx
# 计算中心性指标
degree_centrality = nx.degree_centrality(G)
betweenness_centrality = nx.betweenness_centrality(G, k=1000) # 大图采样
# 找到最中心的位置
location_nodes = [n for n in G.nodes() if n.startswith('g_')]
top_locations = sorted(
[(n, degree_centrality[n]) for n in location_nodes],
key=lambda x: x[1],
reverse=True
)[:10]
print("最中心的位置:")
for loc, centrality in top_locations:
print(f" {loc}: {centrality:.4f}")目的分析
# 分析目的模式
purpose_nodes = [n for n in G.nodes() if n.startswith('p_')]
purpose_stats = {}
for purpose in purpose_nodes:
# 获取连接的位置
connected_locs = [n for n in G.neighbors(purpose) if n.startswith('g_')]
# 获取连接的时间
connected_hours = [n for n in G.neighbors(purpose) if n.startswith('h_')]
purpose_stats[purpose] = {
'locations': len(connected_locs),
'peak_hours': connected_hours,
'total_weight': sum(G[purpose][loc]['weight'] for loc in connected_locs)
}
print("目的分析:")
for purpose, stats in purpose_stats.items():
print(f" {purpose}: {stats}")图检索（GraphRAG）
GraphRAGGenerate类实现了复杂的基于图的检索：
基本检索
from mrra.retriever.graph_rag import GraphRAGGenerate
retriever = GraphRAGGenerate(tb=tb, mobility_graph=mg)
# 为查询检索相关位置
docs = retriever.get_relevant_documents({
"user_id": "user_123",
"t": "2024-09-10 12:30:00", # 可选时间戳
"purpose": ["dining", "work"], # 可选目的过滤
"k": 8 # 结果数量
})
for doc in docs:
print(f"位置: {doc.metadata['node']}, 得分: {doc.metadata['score']:.4f}")检索配置
# 配置检索权重
retriever.purpose_weight = 0.6 # 目的种子重要性
retriever.hour_weight = 0.5 # 小时上下文重要性
retriever.dow_weight = 0.3 # 星期几重要性
retriever.recent_weight = 0.2 # 最近位置偏向
# 使用多个种子的高级检索
docs = retriever.get_relevant_documents({
"user_id": "user_123",
"purpose": "dining", # 单个目的
"hour": 12, # 显式小时
"dow": 1, # 周二
"recent_locations": ["g_1234_5678"], # 最近上下文
"k": 10
})图持久化和缓存
保存和加载图
from mrra.persist.cache import CacheManager
cm = CacheManager()
tb_hash = compute_tb_hash(tb)
# 保存图
cm.save_graph(tb_hash, "mobility_default", mg.G)
# 加载图
cached_graph = cm.load_graph(tb_hash, "mobility_default")
if cached_graph:
mg.G = cached_graph
print("加载了缓存图")图版本管理
# 使用配置特定键保存
config_key = f"mobility_grid{cfg.grid_size_m}_dwell{cfg.min_dwell_minutes}"
cm.save_graph(tb_hash, config_key, mg.G)
# 加载特定版本
specific_graph = cm.load_graph(tb_hash, config_key)GraphRAG原理：使用user/hour/day/purpose节点作为"种子"，通过边权重和种子类型权重传播到位置节点。这提供了更语义化的位置排名，同时结合来自最近位置的新近性偏向。
性能考虑
大图优化
# 对于大数据集，考虑子图提取
def extract_user_subgraph(G, user_id, hops=2):
"""提取特定用户周围的子图"""
import networkx as nx
user_node = f"u_{user_id}"
if user_node not in G:
return nx.MultiDiGraph()
# 获取n跳距离内的节点
subgraph_nodes = set([user_node])
current_nodes = {user_node}
for _ in range(hops):
next_nodes = set()
for node in current_nodes:
next_nodes.update(G.neighbors(node))
subgraph_nodes.update(next_nodes)
current_nodes = next_nodes
return G.subgraph(subgraph_nodes).copy()
# 使用子图进行用户特定分析
user_subgraph = extract_user_subgraph(mg.G, "user_123")内存管理
# 对于内存受限的环境
def compress_graph_weights(G, precision=3):
"""压缩边权重以减少内存使用"""
for src, dst, key, data in G.edges(data=True, keys=True):
if 'weight' in data:
G[src][dst][key]['weight'] = round(data['weight'], precision)
return G
# 应用压缩
mg.G = compress_graph_weights(mg.G)