进化篇：超越检索、迈向统一知识图谱

引言：从"专家小队"到"企业大脑"

• 为结构化数据建立精准的元数据索引。

• 为半结构化文档（如Markdown）保留其内在的层级关系。

• 为非结构化文本捕捉其隐藏的语义边界。

• 为代码知识保留其严谨的语法结构。

"上季度财报提到的新产品策略，在代码层面是如何实现的，并引发了哪些主要的客服问题？"

Part 1: 孤立检索的局限性——我们为何必须进化？

• 知识孤岛问题 (The "Knowledge Silo" Problem)：我们的SQL检索器、文档检索器、代码检索器生活在各自的平行宇宙中。它们无法进行"跨库查询"，无法回答任何需要整合多个领域知识的问题。

• 上下文肤浅问题 (The "Shallow Context" Problem)：向量相似性擅长发现"是什么"，但难以揭示"为什么"和"如何关联"。它可以找到相关的文本块，但无法告诉你一个代码块和一个财报块之间存在"实现"关系。这种关系的缺失，是通向深度分析的最大障碍。

Part 2: 解决方案：用LLM自动构建知识图谱

• 节点 (Node)：代表现实世界中的"实体"，例如一个产品（"新功能A"）、一个用户（"User_123"）、一个代码类（"FileManager"）。

• 关系 (Edge)：代表节点之间的联系，例如 MENTIONS (提及), REPORTED_BY (由...报告), IMPLEMENTS (实现)。

实战：从文本块到知识三元组

1. 提取 (Extraction)：用LLM从非结构化文本中识别出节点和关系。

2. 加载 (Loading)：将提取出的三元组存入图数据库（如Neo4j）。

# 准备工作
# pip install -q langchain_openai neo4j langchain_neo4j
import os
from typing import List, Optional
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_neo4j import Neo4jGraph

# 建议使用环境变量来管理你的API密钥
# os.environ["OPENAI_API_KEY"] = "sk-..."
# os.environ["NEO4J_URI"] = "bolt://localhost:7687"
# os.environ["NEO4J_USERNAME"] = "neo4j"
# os.environ["NEO4J_PASSWORD"] = "your_password"


# 1. 定义我们希望LLM抽取的图谱结构 (Pydantic模型)
class Node(BaseModel):
    """图中的一个节点，代表一个实体。"""
    id: str = Field(description="节点的唯一标识符，通常是实体的名称。")
    type: str = Field(description="节点的类型，例如 'Product', 'User', 'Feature', 'Issue'。")

class Relationship(BaseModel):
    """图中的一条关系（边），连接两个节点。"""
    source: Node = Field(description="关系的起始节点。")
    target: Node = Field(description="关系的目标节点。")
    type: str = Field(description="关系的类型，用大写和下划线表示，例如 'IMPLEMENTS', 'REPORTED_BY'。")

class Graph(BaseModel):
    """从一段文本中抽取的完整知识图谱。"""
    nodes: List[Node] = Field(description="文本中识别出的所有节点列表。")
    relationships: List[Relationship] = Field(description="文本中识别出的所有关系列表。")

# 2. 初始化一个支持结构化输出的LLM
llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)
llm_with_graph_output = llm.with_structured_output(Graph)

# 3. 定义一个函数来处理文本块
def extract_graph_from_text(text: str) -> Optional[Graph]:
    """使用LLM从文本块中提取知识图谱。"""
    prompt = f"""
    你是一个顶级的图谱构建专家。你的任务是从以下文本中识别出核心实体（节点）和它们之间的关系（边），并以指定的结构化格式返回。
    重点关注实体间的直接关系，例如 "X实现了Y", "A报告了B", "C提到了D"。

    请分析以下文本：
    ---
    {text}
    ---
    """
    try:
        return llm_with_graph_output.invoke(prompt)
    except Exception as e:
        print(f"Error processing text: {e}")
        return None

# 4. 使用不同类型的文本块进行测试
chunks = [
    "在第三季度的报告中，我们宣布了'天狼星计划'，该计划旨在通过重构支付模块来提升用户体验。",
    "用户'alex123'报告说，新上线的'即时分享'功能在低配安卓设备上会导致应用崩溃。",
    "class PaymentManager: '该类实现了整个订单的支付流程，并调用了风险控制API。'"
]
graphs = [extract_graph_from_text(chunk) for chunk in chunks]

# 5. 将提取的图谱加载到Neo4j数据库中
# (需要先安装并运行Neo4j: <https://neo4j.com/download/>)
graph_db = Neo4jGraph() # 使用环境变量自动连接
graph_db.add_graph_documents([g for g in graphs if g is not None])

print("--- 图谱数据已成功存入Neo4j ---")
# 清理并获取最新的Schema
graph_db.refresh_schema()
print(graph_db.get_schema())

Part 2.5: 真实世界中的挑战

1. 实体标准化/归一化 (Node Normalization):
• 问题: LLM可能会从不同文本中提取出指向同一实体的不同名称，例如 "支付模块" vs "Payment Module" vs "支付系统"。如果不进行归一化，图谱中会充满大量冗余的孤立节点。
• 解决方案:
• LLM辅助归一化: 在提取后，增加一个额外的LLM调用步骤，要求它将新提取的实体与图谱中已存在的实体进行匹配，判断是否为同义词。
• Fuzzy Matching + 向量相似度: 结合字符串模糊匹配算法（如 Levenshtein distance）和节点名称的向量嵌入相似度，来自动发现并合并相似节点。

2. 关系类型的定义与演化 (Schema Design):
• 问题: 我们应该使用哪些关系类型？是 IMPLEMENTS 还是 IS_IMPLEMENTATION_OF？如果不对关系类型进行预定义和约束，LLM可能会生成五花八门的动词，导致图谱关系混乱，难以查询。
• 解决方案:
• 预定义Schema: 在项目初期，就应该定义一个清晰的、有限的关系类型集合（例如，{MENTIONS, IMPLEMENTS, REPORTED_BY, FIXES, HAS_FEATURE}），并在Prompt中明确指示LLM只能使用这些预定义的关系类型。
• Schema演化: 随着业务发展，可以定期分析LLM试图生成的新关系类型，并由人工审核决定是否将其添加到官方Schema中。

3. 上下文依赖 (Context Dependency):
• 问题: 一个实体或关系的具体含义，往往取决于它所在的文本块。例如，在技术文档中，"Apple" 指的是公司；在生鲜订单中，它指的是水果。
• 解决方案: 在创建图谱节点时，将原始的文本块ID或其摘要作为节点的元数据 (Metadata) 存储起来。这样，在查询和分析时，我们不仅知道节点是什么，还能快速追溯到它被提取的原始上下文。

Part 3: 图谱的存储与查询

• 为什么选择图数据库？ 相较于关系型数据库（如MySQL），图数据库（如Neo4j）专门为存储和查询节点与关系而设计，进行多跳（multi-hop）关联查询的性能是其数百甚至数千倍。

• LangChain集成：LangChain通过langchain_community.graphs.Neo4jGraph提供了与Neo4j的无缝集成。

用自然语言查询图谱

1. 用户提问（自然语言）："哪个用户报告了即时分享功能的什么问题？"

2. LLM翻译：GraphCypherQAChain将问题发送给LLM，LLM生成Cypher查询语句：

MATCH (u:User)-[r:REPORTED_BY]->(i:Issue)<-[:HAS_ISSUE]-(f:Feature {id: '即时分享'})
RETURN u.id, i.id

3. 图数据库执行：该Cypher语句在Neo4j中执行。

4. 返回结构化结果：数据库返回查询结果，例如 [("alex123", "应用崩溃")]。

5. LLM合成答案：GraphCypherQAChain将这个结构化结果再次发给LLM，让它用自然语言总结成最终答案："用户'alex123'报告了'即时分享'功能会导致应用崩溃的问题。"

# 概念代码，需要连接到真实的Neo4j实例
from langchain_neo4j import GraphCypherQAChain
from langchain_community.graphs import Neo4jGraph
from langchain_openai import ChatOpenAI

# 确保LLM和Graph DB已初始化
# llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)
# graph_db = Neo4jGraph()

# 在运行查询之前，刷新图数据库的Schema信息，这对于Cypher语句的生成至关重要
graph_db.refresh_schema()

# 构建问答链
chain = GraphCypherQAChain.from_llm(
    llm, graph=graph_db, verbose=True
)

# 提出一个需要进行关系推理的问题
question = "哪个用户报告了关于'即时分享'功能的问题？"

# 运行链
result = chain.invoke({"query": question})
print(result["result"])

# >> Expected Output:
# 用户'alex123'报告了'即时分享'功能在低配安卓设备上会导致应用崩溃的问题。

Part 4: 终极形态：混合系统 (Graph RAG + Vector RAG)

• Graph RAG：最适合回答关于**"关系"**的问题。例如："谁...什么...？"，"X和Y是如何关联的？"，"A导致了B吗？"

• Vector RAG：最适合回答关于实体**"自身属性"**的问题。例如："详细描述一下'天狼星计划'？"，"PaymentManager类的完整代码是什么？"

实战：构建智能路由 (Intent-Based Routing)

1. 定义工具 (Tools)：我们将Graph RAG和Vector RAG分别定义为两个独立的"工具"。

2. 创建路由链 (Router Chain)：LLM会分析用户问题，并决定调用哪个工具（或者都不调用）。

import os
from typing import Literal
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_neo4j import GraphCypherQAChain

# 假设我们已经有了之前文章中构建好的向量检索器
# from our_vector_rag import vector_retriever

# 假设的向量检索器函数
def vector_search(query: str):
    """用于回答关于实体具体属性、定义或详细描述的问题。"""
    # In a real scenario, this would call our vector_retriever
    return f"Vector search result for: '{query}' - provides detailed text."

# 之前定义的Graph RAG链
# chain = GraphCypherQAChain.from_llm(...)

def graph_search(query: str):
    """用于回答关于多个实体之间关系的问题。"""
    # 在真实场景中，这将调用我们的GraphCypherQAChain
    # chain = GraphCypherQAChain.from_llm(...)
    # For demonstration, we'll simulate the chain's response
    # In a real implementation, you would have the chain object available here
    if "天狼星计划" in query and "支付模块" in query:
        return "天狼星计划'旨在'通过重构'支付模块'来提升用户体验，因此它们之间是'旨在实现'的关系。"
    return "抱歉，我无法找到它们之间的直接关系。"

# 1. 定义路由逻辑的数据结构
class RouteQuery(BaseModel):
    """根据用户问题，决定路由到graph_search或vector_search。"""
    route: Literal["graph", "vector"] = Field(
        ...,
        description=(
            "给定用户问题，判断它更适合由'vector'还是'graph'路径来回答。"
            "'vector'用于查询实体的具体属性、定义或描述。"
            "'graph'用于查询实体之间的关系。"
        ),
    )

# 2. 构建路由链
llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)
structured_llm = llm.with_structured_output(RouteQuery)
router_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are an expert at routing a user query to either a vector search or a graph search."),
        ("human", "Based on the query, which route should I take?\\n\\nQuery: {query}"),
    ]
)
router = router_prompt | structured_llm

# 3. 定义完整的图谱/向量RAG链
def query_router(query):
    result = router.invoke({"query": query})
    if result.route == 'vector':
        return vector_search(query)
    elif result.route == 'graph':
        return graph_search(query)

# 4. 测试路由
query1 = "详细描述一下'天狼星计划'是什么？"
print(f"Query: {query1}\\nResult: {query_router(query1)}\\n")
# Expected: Routes to 'vector'

query2 = "天狼星计划和支付模块有什么关系？"
print(f"Query: {query2}\\nResult: {query_router(query2)}\\n")
# Expected: Routes to 'graph'

Part 4.5: The Trade-offs: When to Use Graph RAG?

结论：企业RAG的星辰大海

1. 基础RAG：简单的文本切分与向量检索。

2. 高级RAG：针对不同知识形态的"专家小队"式精细化处理。

3. 图谱RAG：将所有知识连接成网，实现跨领域推理的"企业大脑"。