我要投稿

如何用“图增强 RAG”提升中文问答体验

发布日期：2025-07-01 18:08:04 浏览次数： 1743

作者：用AI写AI应用的小李

微信搜一搜，关注“用AI写AI应用的小李”

想象一下，你的问答机器人不再是“关键词搬运工”，而是个真正会“思考”的小伙伴：它能把相关知识串成思路，再用轻松的口吻给你讲清楚。以下这套基于 Graph RAG（Graph-Enhanced Retrieval-Augmented Generation）的方案，既保留了检索效率，又增强了逻辑可解释性，适合中文内容库。

一、为什么要把“图”搬进检索里

摆脱关键词束缚
传统 RAG 单纯靠向量或关键词匹配，容易对多跳、跨文档问题无能为力。
捕捉知识关系
把“概念”和“文档片段”当成节点，用边标注它们的关联程度，检索时就能沿路径找答案。
可视化检索链路
结果不光给你答案，还能展示“为什么这么找”，便于审计和优化。

二、技术流程概览

1. 文本分片

def chunk_text(text, chunk_size=800, overlap=150):
    chunks = []
    for i in range(0, len(text), chunk_size - overlap):
        part = text[i : i + chunk_size]
        if part:
            chunks.append({
                "id": len(chunks),
                "text": part
            })
    return chunks

将长文本分成大约 800 字的小块，重叠一部分保证上下文连贯。

2. 概念抽取

def extract_concepts(text):
    prompt = f"请列出下面这段中文的 5–8 个核心关键词或实体：\n{text[:200]}…"
    resp = client.chat.completions.create(
        model="chatglm-6b",
        messages=[{"role":"user","content":prompt}],
        temperature=0
    )
    return [c.strip() for c in resp.choices[0].message.content.split(",")]

给每个片段贴上“标签”，让不同内容靠“共同标签”连接。

3. 构建知识图谱

import networkx as nx
from scipy.spatial.distance import cosine

def build_graph(chunks, embeddings, concept_lists):
    G = nx.Graph()
    for idx, chunk in enumerate(chunks):
        G.add_node(idx, text=chunk["text"],
                   emb=embeddings[idx],
                   concepts=concept_lists[idx])
    for i in range(len(chunks)):
        for j in range(i+1, len(chunks)):
            shared = set(concept_lists[i]) & set(concept_lists[j])
            if not shared: continue
            sim = 1 - cosine(embeddings[i], embeddings[j])
            conc_score = len(shared) / min(len(concept_lists[i]), len(concept_lists[j]))
            weight = 0.7 * sim + 0.3 * conc_score
            if weight > 0.6:
                G.add_edge(i, j, weight=weight)
    return G

边的权重由“向量相似度”与“概念重叠度”共同决定，过滤掉微弱关联。

4. 图遍历检索

import heapq

def traverse_graph(query_emb, G, top_k=3, max_depth=2):
    sims = [(1 - cosine(query_emb, G.nodes[n]["emb"]), n) for n in G.nodes]
    sims.sort(reverse=True)
    heap = [(-s, n, 0) for s, n in sims[:top_k]]
    heapq.heapify(heap)

    seen, result = set(), []
    while heap:
        score, node, depth = heapq.heappop(heap)
        if node in seen or depth > max_depth: continue
        seen.add(node)
        result.append(( -score, node ))
        for nbr, data in G[node].items():
            next_score = -score * data["weight"]
            heapq.heappush(heap, (next_score, nbr, depth + 1))
    return [n for _, n in sorted(result, reverse=True)][: top_k * 2]

先选几个相关度最高的“起点”，再沿着图里的边“多跳”扩展，平衡深度和效率。

5. 生成回答

def generate_reply(query, selected_chunks):
    context = "\n\n---\n\n".join(c["text"] for c in selected_chunks)
    prompt = f"以下是检索到的上下文：\n{context}\n\n请用轻松自然的口吻回答：{query}"
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role":"user","content":prompt}],
        temperature=0.3
    )
    return resp.choices[0].message.content