传统 RAG：  问题 → 向量相似度搜索 → 扁平文本块列表GraphRAG：  问题 → 实体识别 → 图上多跳遍历 → 结构化子图 + 文本块

【阶段1：图谱构建（离线，一次性）】文档 → LLM 实体/关系抽取 → 三元组(S,P,O) → 写入图数据库(Neo4j/NebulaGraph)     → 同时保留原始文本块（用于最终生成）【阶段2：图检索（在线）】用户问题 → LLM 识别问题中的实体 → 在图上做 N-hop 遍历         → 获取相关子图（节点+边+关联文本块）【阶段3：增强生成（在线）】子图结构化信息 + 关联文本块 + 用户问题 → 组装 Prompt → LLM 生成回答

internal/rag-service/├── graphrag/│   ├── extractor.go# LLM 实体/关系抽取（核心！）│   ├── builder.go# 图谱构建编排│   ├── retriever.go# 图检索（多跳遍历）│   └── prompt/# 抽取和检索用的 Prompt 模板│       ├── extract.txt# 实体关系抽取 Prompt│       └── retrieve.txt# 问题实体识别 Prompt├── repository/│   └── graph.go# 图数据库操作封装└── service/    └── graphrag.go# GraphRAG 对外服务入口

// 文件：internal/rag-service/graphrag/extractor.go// ExtractTriplets 从一段文本中提取实体和关系// 输入：原始文本块// 输出：三元组列表 [(主体, 关系, 客体), ...]func (e *Extractor) ExtractTriplets(ctx context.Context, text string) ([]Triplet, error) {    // Step 1: 加载抽取 Prompt 模板    // 这个 Prompt 是关键工程，决定了抽取质量    promptTemplate := e.loadPrompt(”extract.txt”)    // Step 2: 填充模板    prompt := strings.Replace(promptTemplate, ”{{TEXT}}”, text, 1)    // Step 3: 调用 LLM（要求返回 JSON 格式）    resp, err := e.llm.Complete(ctx, []Message{        {Role: ”system”, Content: ”你是一个知识图谱抽取专家。严格按JSON格式输出。”},        {Role: ”user”, Content: prompt},    })    if err != nil {        return nil, fmt.Errorf(”llm extraction failed: %w”, err)    }    // Step 4: 解析 LLM 返回的 JSON    // 预期格式：{”triplets”: [{”subject”:”张三”,”relation”:”属于”,”object”:”研发部”}]}    var result ExtractionResult    if err := json.Unmarshal([]byte(resp.Content), &result); err != nil {        // LLM 有时返回格式不规范，这里做容错处理        return nil, fmt.Errorf(”parse extraction result failed: %w”, err)    }    // Step 5: 后处理（去重、实体对齐、置信度过滤）    triplets := e.postProcess(result.Triplets)    return triplets, nil}

// 文件：internal/rag-service/graphrag/builder.go// BuildGraph 构建知识图谱的主入口func (b *Builder) BuildGraph(ctx context.Context, kbID string) error {    // Step 1: 获取知识库下所有已解析的文本块    chunks, err := b.chunkRepo.ListByKB(ctx, kbID)    if err != nil {        return err    }    // Step 2: 批量抽取三元组（并发控制，避免 LLM API 限流）    var allTriplets []TripletWithSource    semaphore := make(chan struct{}, 5) // 最多5个并发    for _, chunk := range chunks {        semaphore <- struct{}{}        go func(c Chunk) {            defer func() { <-semaphore }()            triplets, err := b.extractor.ExtractTriplets(ctx, c.Text)            if err != nil {                log.Warnf(”extract failed for chunk %s: %v”, c.ID, err)                return            }            // 记录每个三元组来自哪个文本块（溯源用）            for _, t := range triplets {                allTriplets = append(allTriplets, TripletWithSource{                    Triplet: t,                    ChunkID: c.ID,                })            }        }(chunk)    }    // Step 3: 实体对齐（同一个实体可能有不同表述）    // 例如：”中国联通” 和 ”China Unicom” 应该是同一个节点    aligned := b.entityAligner.Align(allTriplets)    // Step 4: 写入图数据库    err = b.graphRepo.BulkUpsert(ctx, aligned)    if err != nil {        return fmt.Errorf(”write graph failed: %w”, err)    }    // Step 5: 更新知识库状态    b.kbRepo.UpdateGraphStatus(ctx, kbID, GraphReady)    return nil}

// 文件：internal/rag-service/graphrag/retriever.go// Retrieve 根据用户问题在图上检索相关信息func (r *Retriever) Retrieve(ctx context.Context, query string, maxHops int) (*SubGraph, error) {    // Step 1: 用 LLM 识别问题中的实体    // ”A公司CEO和B公司CTO的关系” → [”A公司CEO”, ”B公司CTO”]    entities, err := r.identifyEntities(ctx, query)    if err != nil {        return nil, err    }    // Step 2: 在图上做多跳遍历    // 从识别出的实体出发，沿边走 maxHops 步    subGraph, err := r.graphRepo.MultiHopSearch(ctx, MultiHopParams{        StartEntities: entities,        MaxHops:       maxHops,     // 默认 2 跳        MaxNodes:      50,          // 限制子图大小，防止爆炸        EdgeTypes:     nil,         // nil 表示不限关系类型    })    if err != nil {        return nil, fmt.Errorf(”graph traversal failed: %w”, err)    }    // Step 3: 获取子图中每个节点/边关联的原始文本块    // 这一步让 LLM 生成时有”原文依据”    chunkIDs := subGraph.GetAssociatedChunkIDs()    chunks, _ := r.chunkRepo.BatchGet(ctx, chunkIDs)    subGraph.Chunks = chunks    return subGraph, nil}

// 文件：internal/rag-service/service/search.gofunc (s *Service) HybridSearch(ctx context.Context, req SearchRequest) (*SearchResponse, error) {    var results []ChunkScore    // 路线1：传统 RAG（始终执行）    ragResults, _ := s.ragSearch(ctx, req)    results = append(results, ragResults...)    // 路线2：GraphRAG（仅在知识库启用了图谱时执行）    if req.KnowledgeBase.GraphEnabled {        graphResults, err := s.graphRetriever.Retrieve(ctx, req.Query, 2)        if err == nil && len(graphResults.Chunks) > 0 {            // 将图检索结果也加入候选集            results = append(results, graphResults.Chunks...)        }    }    // 统一 RRF 融合排序（和第 3 篇的逻辑一致）    merged := rrfMerge(results, req.TopK)    return &SearchResponse{Chunks: merged}, nil}

# mini_graphrag.py# 依赖：pip install openai networkximport openai, json, networkx as nxclient = openai.OpenAI(api_key=”sk-your-key”)# ===== 1. 图谱构建 =====docs = [    ”张三是研发部总监，负责AI平台项目。”,    ”李四是研发部工程师，向张三汇报。”,    ”AI平台项目使用了元景大模型技术。”,    ”王五是市场部负责人，与张三共同推进AI营销方案。”,]EXTRACT_PROMPT = ”””从以下文本中提取实体和关系，返回JSON格式：{”triplets”: [{”subject”:”...”, ”relation”:”...”, ”object”:”...”}]}文本：{{TEXT}}”””G = nx.DiGraph()for doc in docs:    prompt = EXTRACT_PROMPT.replace(”{{TEXT}}”, doc)    resp = client.chat.completions.create(        model=”gpt-4o-mini”,        messages=[{”role”: ”user”, ”content”: prompt}],        response_format={”type”: ”json_object”}    )    data = json.loads(resp.choices[0].message.content)    for t in data.get(”triplets”, []):        G.add_edge(t[”subject”], t[”object”], relation=t[”relation”])# 存储原始文本作为节点属性（简化版）        G.nodes[t[”subject”]][”source”] = doc        G.nodes[t[”object”]][”source”] = docprint(f”图谱构建完成：{G.number_of_nodes()} 节点，{G.number_of_edges()} 边”)# ===== 2. 图检索 =====def graph_retrieve(entity, max_hops=2):    ”””从指定实体出发，做多跳遍历”””    visited = set()    queue =    subgraph_nodes = set()    while queue:        node, depth = queue.pop(0)        if depth > max_hops or node in visited:            continue        visited.add(node)        subgraph_nodes.add(node)        for neighbor in G.neighbors(node):            queue.append((neighbor, depth + 1))    return G.subgraph(subgraph_nodes)# ===== 3. 增强生成 =====def graph_answer(question, start_entity):    sub = graph_retrieve(start_entity)# 将子图转为文本描述    edges_desc = ”\n”.join(        f”- {u} --[{d['relation']}]--> {v}”        for u, v, d in sub.edges(data=True)    )    sources = ”\n”.join(set(        sub.nodes[n].get(”source”, ””) for n in sub.nodes    ))    prompt = f”””基于以下知识图谱信息和原始文档回答问题。## 知识图谱{edges_desc}## 原始文档{sources}## 问题{question}请标注信息来源。”””    resp = client.chat.completions.create(        model=”gpt-4o-mini”,        messages=[{”role”: ”user”, ”content”: prompt}]    )    return resp.choices[0].message.content# 测试print(graph_answer(”张三和李四是什么关系？他们共同参与了什么？”, ”张三”))# 预期：张三和李四是上下级关系（李四向张三汇报）。#       张三负责AI平台项目，李四作为研发部工程师可能参与该项目。