AI应用知识图谱推理与多跳问答深度实战：从路径查找到规则推理与常识推理的全解析

引言

知识图谱不仅存储实体关系，还能进行推理——从已知关系推导未知关系。多跳推理让AI回答"张三的导师的合作者是谁"这类需要跨多个实体关系链的问题。本文从知识图谱推理的工程体系讲起，覆盖路径查找（BFS/DFS遍历）、多跳推理（k-hop邻居）、规则推理（IF-THEN规则）、常识推理（LLM+图谱）、推理可解释性（路径展示）、推理评估、图谱补全（预测缺失关系），构建AI应用知识图谱推理工程体系。

一、路径查找

# graph_reasoning/path.py
class GraphPathFinder:
    """图路径查找：多跳关系推理"""
    
    async def find_path(self, graph_store, source: str,
                         target: str, max_hops: int = 3) -> list[dict]:
        """查找两实体间的路径（BFS）"""
        async with graph_store.driver.session() as session:
            result = await session.run(
                """MATCH (s:Entity {name: $source}), (t:Entity {name: $target}),
                      p = shortestPath((s)-[:RELATES*1..3]->(t))
                RETURN [node IN nodes(p) | node.name] AS entities,
                       [rel IN relationships(p) | rel.type] AS relations,
                       length(p) AS hops
                LIMIT 5""",
                source=source, target=target,
            )
            return [r.data() async for r in result]
    
    async def k_hop_neighbors(self, graph_store, entity: str,
                               k: int = 2) -> dict:
        """k跳邻居子图"""
        async with graph_store.driver.session() as session:
            result = await session.run(
                """MATCH (e:Entity {name: $name})-[:RELATES*1..2]-(neighbor)
                RETURN collect(DISTINCT {
                    name: neighbor.name,
                    type: neighbor.type,
                    description: neighbor.description
                }) AS neighbors""",
                name=entity,
            )
            record = await result.single()
            return record.data() if record else {"neighbors": []}

二、多跳问答

# graph_reasoning/multi_hop_qa.py
class MultiHopQA:
    """多跳问答：基于图谱的关系链推理"""
    
    async def answer(self, question: str, graph_store,
                      llm_client) -> dict:
        """回答多跳问题"""
        # 1. 实体抽取
        entities = await self._extract_entities(question, llm_client)
        if len(entities) < 2:
            # 单实体：查邻居
            if entities:
                neighbors = await GraphPathFinder().k_hop_neighbors(
                    graph_store, entities[0], k=2,
                )
                return {
                    "answer": f"{entities[0]}的相关实体：{neighbors}",
                    "reasoning": "单跳邻居查询",
                    "path": [],
                }
            return {"answer": "无法识别问题中的实体", "reasoning": "no entities"}
        
        # 2. 路径查找
        paths = []
        for i in range(len(entities) - 1):
            path = await GraphPathFinder().find_path(
                graph_store, entities[i], entities[i+1], max_hops=3,
            )
            paths.extend(path)
        
        # 3. 基于路径生成回答
        if paths:
            path_text = self._format_paths(paths)
            prompt = f"""基于以下知识图谱路径回答问题。

问题：{question}
图谱路径：
{path_text}

请基于路径中的关系链推理出答案，并展示推理过程。"""
            answer = await llm_client.complete(prompt, temperature=0.2)
            return {
                "answer": answer,
                "reasoning": path_text,
                "path": paths,
            }
        
        return {"answer": "图谱中未找到相关路径", "reasoning": "no path found"}
    
    async def _extract_entities(self, question: str,
                                 llm_client) -> list[str]:
        """从问题中抽取实体"""
        prompt = f"""提取以下问题中的实体名称。

问题：{question}

输出JSON：{{"entities": ["实体1", "实体2"]}}"""
        raw = await llm_client.complete(
            prompt, response_format={"type": "json_object"}, temperature=0,
        )
        import json
        return json.loads(raw).get("entities", [])
    
    def _format_paths(self, paths: list[dict]) -> str:
        """格式化路径为可读文本"""
        lines = []
        for p in paths:
            entities = p.get("entities", [])
            relations = p.get("relations", [])
            chain = " → ".join(
                f"{entities[i]} -[{relations[i]}]-> " if i < len(relations)
                else f"{entities[i]}"
                for i in range(len(entities))
            )
            lines.append(f"路径（{p.get('hops', 0)}跳）：{chain}")
        return "\n".join(lines)

三、规则推理

# graph_reasoning/rules.py
class RuleBasedReasoner:
    """规则推理：IF-THEN规则"""
    
    RULES = [
        {
            "name": "导师-学生传递",
            "pattern": {"relation": "advisor_of", "transitive": True},
            "inference": "如果A是B的导师，B是C的导师，则A是C的学术前辈",
        },
        {
            "name": "组织归属传递",
            "pattern": {"relation": "member_of", "transitive": True},
            "inference": "如果A属于B，B属于C，则A间接属于C",
        },
        {
            "name": "合作者推断",
            "pattern": {
                "if": [{"relation": "collaborated_on", "via": "project"}],
                "then": {"relation": "co_author"},
            },
            "inference": "在同一项目合作的人是合作者",
        },
    ]
    
    async def apply_rules(self, graph_store) -> dict:
        """应用推理规则补全图谱"""
        inferred = 0
        for rule in self.RULES:
            if rule["pattern"].get("transitive"):
                # 传递闭包
                async with graph_store.driver.session() as session:
                    result = await session.run(
                        """MATCH (a)-[:RELATES {type: $rel}]->(b),
                               (b)-[:RELATES {type: $rel}]->(c)
                        WHERE NOT (a)-[:RELATES {type: $rel}]->(c)
                        MERGE (a)-[:RELATES {type: $rel, inferred: true}]->(c)
                        RETURN count(*) AS created""",
                        rel=rule["pattern"]["relation"],
                    )
                    record = await result.single()
                    if record:
                        inferred += record["created"]
        return {"inferred_relations": inferred}

总结

知识图谱推理以"路径查找-多跳问答-规则推理"三层展开：路径查找以Cypher的shortestPath查询在图谱中找到两实体间的最短关系链（如"张三-[同事]->李四-[导师]->王五"），多跳问答先抽取问题中的实体再查找路径最后让LLM基于路径推理生成答案并展示推理过程，规则推理以传递闭包（advisor_of的传递性）和模式匹配（同一项目合作→合作者）自动补全图谱中缺失的关系。当RAG从"纯向量检索"补充了"图谱推理"能力，AI能回答"A和B是什么关系""A间接属于哪个组织"这类需要关系链推理的复杂问题——这是向量检索无法触及而图谱推理的独特价值。