问题背景与适用场景

在构建企业级知识图谱问答系统时，传统 RAG（检索增强生成）常面临多跳推理能力不足、跨文档关联缺失的问题。GraphRAG 通过将文本实体与关系建模为图结构，显著提升了复杂查询的准确性。本指南详解如何基于 HolySheep AI API 实现完整的 GraphRAG 流程，涵盖实体抽取、图谱构建与智能查询路由三大核心模块。

前置条件

配置步骤详解

步骤 1：安装依赖并配置 API 客户端

首先初始化项目结构，配置与 HolySheep 的连接参数。使用环境变量或配置文件管理 API Key 避免硬编码风险。

步骤 2：实现实体抽取模块

利用 LLM 的零样本能力进行实体识别，通过 prompt 工程引导模型输出结构化 JSON 格式的实体与关系列表。

步骤 3：构建知识图谱

将抽取的实体与关系存入图数据库（使用 NetworkX 作为内存图），支持后续的图遍历与社区检测。

步骤 4：实现查询路由

根据用户 query 的复杂度自动选择检索策略：简单问题走向量检索，复杂多跳问题走图遍历。


import os
import json
import networkx as nx
from openai import OpenAI
from typing import List, Dict, Tuple


配置 HolySheep API

client = OpenAI(
    api_key=os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1"
)

class EntityExtractor:
    """基于 HolySheep LLM 的实体抽取器"""
    
    SYSTEM_PROMPT = """你是一个专业的知识图谱工程师，负责从文本中抽取实体和关系。
    输出格式为严格的 JSON，包含 'entities' 和 'relations' 两个数组。
    实体包含: id, name, type (PERSON/ORG/LOCATION/EVENT/CONCEPT)
    关系包含: source, target, relation_type"""
    
    EXTRACT_PROMPT = """从以下文本中抽取所有实体和关系：

    {text}

    只输出 JSON，不要包含任何解释。"""
    
    def __init__(self, model: str = "gpt-4o"):
        self.client = client
        self.model = model
    
    def extract(self, text: str) -> Dict:
        """执行实体抽取"""
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": self.SYSTEM_PROMPT},
                {"role": "user", "content": self.EXTRACT_PROMPT.format(text=text)}
            ],
            temperature=0.1,
            response_format={"type": "json_object"}
        )
        return json.loads(response.choices[0].message.content)

class KnowledgeGraph:
    """基于 NetworkX 的知识图谱构建器"""
    
    def __init__(self):
        self.graph = nx.MultiDiGraph()
    
    def add_entities(self, entities: List[Dict]):
        """添加实体到图谱"""
        for entity in entities:
            self.graph.add_node(
                entity["id"],
                name=entity["name"],
                entity_type=entity["type"]
            )
    
    def add_relations(self, relations: List[Dict]):
        """添加关系到图谱"""
        for rel in relations:
            self.graph.add_edge(
                rel["source"],
                rel["target"],
                relation_type=rel["relation_type"]
            )
    
    def get_subgraph(self, center_node: str, depth: int = 2) -> nx.MultiDiGraph:
        """获取指定节点周围深度的子图"""
        return nx.ego_graph(self.graph, center_node, radius=depth)
    
    def community_detection(self) -> List[List[str]]:
        """基于 Louvain 算法的社区检测"""
        undirected = self.graph.to_undirected()
        communities = nx.community.louvain_communities(undirected)
        return [list(c) for c in communities]

完整代码示例

以下示例展示完整的 GraphRAG 流程：从文档处理到图谱查询的端到端实现。



安装依赖

pip install openai networkx langchain-community python-dotenv


设置环境变量

export HOLYSHEEP_API_KEY="YOUR_HOLYSHEEP_API_KEY"


运行完整示例（保存为 run_graphrag.py）

python run_graphrag.py --documents ./data/*.txt --query "公司的核心技术团队有哪些成员？"


import argparse
from entity_extractor import EntityExtractor, KnowledgeGraph

class QueryRouter:
    """智能查询路由：根据查询复杂度选择检索策略"""
    
    ROUTING_PROMPT = """分析以下查询的复杂度：

    "{query}"

    如果是多跳推理、比较分析或需要综合多个实体的问题，返回 "graph"
    如果是简单的事实查询，返回 "vector"
    
    只输出一个词：graph 或 vector。"""
    
    def __init__(self):
        self.client = client
        self.model = "gpt-4o"
    
    def route(self, query: str) -> str:
        """决定检索策略"""
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "user", "content": self.ROUTING_PROMPT.format(query=query)}
            ],
            temperature=0
        )
        return response.choices[0].message.content.strip()
    
    def graph_retrieval(self, kg: KnowledgeGraph, query: str, top_k: int = 5):
        """图谱检索：提取相关实体并获取子图"""
        extractor = EntityExtractor()
        result = extractor.extract(query)
        
        relevant_nodes = []
        for entity in result.get("entities", []):
            if entity["name"] in kg.graph:
                relevant_nodes.append(entity["id"])
        
        # 聚合多跳范围内的所有节点
        subgraph_nodes = set()
        for node in relevant_nodes:
            ego_graph = kg.get_subgraph(node, depth=2)
            subgraph_nodes.update(ego_graph.nodes())
        
        subgraph = kg.graph.subgraph(subgraph_nodes)
        return self._graph_to_context(subgraph)
    
    def _graph_to_context(self, subgraph: nx.MultiDiGraph) -> str:
        """将子图转换为文本上下文"""