从原文抽取到社区证据查询

前两章已经区分实体身份和遍历。现在让边来自真实原文，而不是手填 edges；以 SQLite 保存关系与出处，以 NetworkX 执行遍历和社区划分。原文采用可解析的受限英文语法，这是一种规则抽取器；面对自由文档，可替换成结构化模型抽取，但模型返回的 quote 必须在 source 原文内，实体 ID 必须来自当前实体表，关系类型必须在白名单。

每条边的有效期是半开区间 start≤year<end。先按租户与时间得到允许的子图，再遍历、分社区和形成报告；否则过期的中间边仍可能把两个有效节点串起来。名字相同的 A:amy 与 B:amy 不能合并。

首次完整实验

Python 3.12 新虚拟环境安装 networkx==3.6.1；保存为 graph_lab.py，运行 python graph_lab.py。SQLite 为标准库，无外部图服务。

import re, sqlite3, json
import networkx as nx

texts = {
 's1': 'A:amy mentors A:bo from 2024 to 2028.',
 's2': 'A:bo founded A:cedar from 2023 to 2025.',
 's3': 'A:bo works_for A:delta from 2025 to 2028.',
 's4': 'B:amy founded B:cedar from 2023 to 2030.',
 's5': 'A:eva founded A:elm from 2025 to 2030.'}
pattern = re.compile(r'([AB]:\w+) (mentors|founded|works_for) ([AB]:\w+) from (\d+) to (\d+)\.')
# Restricted grammar is a real extractor for this fixture, not a general NLP model.
def extract(source,text):
    match = pattern.fullmatch(text)
    if not match:raise ValueError('manual extraction required')
    a,relation,b,start,end = match.groups()
    return (a,b,relation,int(start),int(end),source,text)
conn = sqlite3.connect(':memory:')
conn.execute('CREATE TABLE edge(a TEXT,b TEXT,relation TEXT,start INT,end INT,source TEXT,quote TEXT,PRIMARY KEY(source,a,b,relation,start,end))')
for source,text in texts.items():conn.execute('INSERT OR IGNORE INTO edge VALUES(?,?,?,?,?,?,?)',extract(source,text))

def graph(tenant, year):
    g = nx.MultiDiGraph()
    for a,b,rel,start,end,source,quote in conn.execute('SELECT * FROM edge WHERE start<=? AND end>?',(year,year)):
        if a.split(':')[0] == b.split(':')[0] == tenant:
            assert texts[source] == quote
            g.add_edge(a,b,key=(source,rel,start,end),relation=rel,source=source,quote=quote)
    return g

def local(tenant,year,start,hops):
    g=graph(tenant,year)
    if start not in g:return []
    paths=nx.single_source_shortest_path(g,start,cutoff=hops)
    return [(node,[(a,b,sorted({data['source'] for data in g[a][b].values()})) for a,b in zip(path,path[1:])]) for node,path in paths.items() if len(path)>1]
assert any(node=='A:cedar' for node,path in local('A',2024,'A:amy',2))
assert all(node!='A:cedar' for node,path in local('A',2026,'A:amy',2))
assert local('A',2024,'B:amy',2)==[]
# The partition is computed from the permitted, time-valid graph, not supplied reports.
g=graph('A',2026)
# Community topology uses one link per endpoint pair; evidence stays in g.
communities=list(nx.community.greedy_modularity_communities(nx.Graph(g)))
reports=[]
for group in communities:
    edges=[data for a,b,data in g.edges(data=True) if a in group and b in group]
    reports.append({'members':sorted(group),'claims':[e['quote'] for e in edges],'sources':[e['source'] for e in edges]})
# Global map/reduce retains citations and deduplicates repeated evidence.
global_sources=sorted({s for report in reports for s in report['sources']})
assert global_sources==['s1','s3','s5']
assert len(communities)==2
def global_answer(term):
    points=[]
    for report in reports:
        for quote,source in zip(report['claims'],report['sources']):
            if term in quote.split():points.append({'quote':quote,'source':source})
    return list({p['source']:p for p in points}.values())
assert global_answer('founded')==[{'quote':texts['s5'],'source':'s5'}]
assert global_answer('purchased')==[]
print('local',local('A',2026,'A:amy',2),'communities',reports,'global',global_sources)

2024 年 Amy→Bo→Cedar 的两跳链存在；2026 年 Bo→Cedar 到期，即使 Cedar 实体还在历史库里也不再参与回答。社区是按允许图的结构计算出来的；模块度贪心法倾向把内部连接密集的节点放一起，不保证语义主题正确，分辨率变化也会改变分组。

同一对实体可能有不同关系，每个关系也可能有多个来源。MultiDiGraph用(source,relation,start,end)区分断言；SQLite唯一键使同一来源同一断言重复摄取不增加边。社区算法只读投影后的简单无向图，重复来源不会增加结构权重；报告和路径仍回到多重图收集全部出处。当前每份原文仅包含一个受限语法断言，真实多断言文档还需加入片段/claim ID。

local 查询从实体出发，返回路径及每条边的来源；global 查询对所有相关社区报告做 map，再合并来源集合。这里报告采用抽取式原句，避免“摘要说了什么”与“边实际支持什么”混淆。使用生成式摘要时保留 claim→source 的映射，并将无来源 claim 拒绝入库。跨社区关系应另建边界报告；不能只保留社区内部边就声称全图关系都覆盖。

混合检索先以文本向量找到入口实体，再用局部遍历找依赖；返回的边证据与原始文本候选按来源 ID 去重。若问题只是退款天数，向量/关键词检索已足够；若问“创办某公司的人由谁指导”，路径结构才直接对应关系链。图抽错边会传播错误，因此原文引用验证比增加跳数更重要。

练习

将 s1 的结束年份改为2026，查询2026时 Amy 不应再到达 Delta；保留 Bo→Delta，验证直接以 Bo 为入口仍可到达。再加入语法不合法的原文，验证进入人工抽取分支。为Amy→Bo新增不同关系及同关系第二来源，重复摄取同一来源，确认路径/社区保留所有来源而不重复加权。

完成后查看本章答案。

社区层级决定全局查询读多细的报告：粗层摘要数量少，细层更易保留例外但增加map调用与上下文成本。这里用两块不连通分量得到两个社区，并以founded筛出全局原句；生成式map可输出带分数的claim，reduce先过滤和去重再合成。无论哪种方式，purchased查询无证据就必须为空，不能从全局汇总凭空补常识。

返回完整学习路线。