新手必看Neo4j图数据库跨库链路检索文档代码桥接完整实战指南

跨库链路检索：Neo4j 图数据库桥接文档与代码

前言

企业级研发场景中，知识从来不会局限于单一载体——需求文档散落在语雀或 Confluence，代码实现托管在 Gitee 或 GitHub。开发者抛出一个问题，比如“登录功能的设计方案与代码实现”，传统向量检索只能在各集合内独立检索：文档查文档的，代码查代码的，无法打通“设计文档 → 代码实现”这条关键链路。

O-RAG 系统引入的跨库链路检索（Cross-KB Search），正是为解决这一痛点而设计。它依托 Neo4j 图数据库构建跨知识源的语义关系图谱，实现“文档检索 → 关系桥接 → 代码取回”两阶段精准检索。接下来，我们从架构设计、图谱构建、检索策略、关系回写四个维度，拆解该特性的实现细节。

一、问题背景

1.1 单集合方案的局限

初期，O-RAG 曾尝试将所有知识源塞入同一个 Qdrant 集合，仅靠 source_type 字段过滤。实际测试后发现行不通，暴露了多项缺陷：

1.2 分集合 + 图谱桥接方案

最终 O-RAG 选择了 Qdrant 分集合 + Neo4j 跨库关系桥接的双模共存方案。核心思路：让不同类型的知识各住各的“房子”（实现语义空间隔离、生命周期独立），再通过 Neo4j 这座“桥梁工程师”将它们连通。

┌─────────────────────────────────────────────────────────────────┐
│                        Qdrant 向量库                           │
│  ┌──────────┐   ┌──────────┐   ┌──────────┐                  │
│  │ kb_docs  │   │ kb_yuque │   │ kb_code  │  语义空间隔离    │
│  │ 本地文档 │   │ 语雀文档 │   │ 代码仓库 │  生命周期独立    │
│  └──────────┘   └──────────┘   └──────────┘                  │
└─────────────────────────────────────────────────────────────────┘
↕
┌─────────────────────────────────────────────────────────────────┐
│                        Neo4j 图谱库                           │
│  ┌────────────────────────────────────────────────────────┐  │
│  │ DocChunk ──[IMPLEMENTS]──> CodeFunction                │  │
│  │ YuqueChunk ──[MENTIONS]──> CodeClass                   │  │
│  │ CodeFile ──[CONTAINS]──> CodeFunction                  │  │
│  │ CodeClass ──[INHERITS]──> CodeClass                    │  │
│  └────────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────────┘

核心设计理念可概括为三点：

1. 向量库分集合——不同来源数据各得其所，互不干扰。
2. 图数据库桥接——用关系边串联跨集合节点，形成语义关联。
3. 双通道检索——优先走 Neo4j 关系桥接，关系缺失时回退到 LLM 符号抽取 + 向量检索，保证兜底可用。

二、代码图谱构建

跨库检索的基础是一张高质量的代码图谱。O-RAG 通过 AST 解析提取代码符号，再将其导入 Neo4j 完成图谱构建。

2.1 AST 多语言解析

目前 O-RAG 支持 Python、Java、JavaScript、Go、TypeScript 五种语言的 AST 解析。以 Python 为例，解析器遍历语法树，提取函数、类、方法、参数、docstring、装饰器等关键信息，最终组装成 CodeSymbol 列表。

# app/import_process/processors/ast_parser.py
class CodeSymbol:
    """代码符号：函数/类/方法"""
    def __init__(self, name: str, symbol_type: str, language: str, 
                 start_line: int = 0, end_line: int = 0, 
                 parameters: str = "", docstring: str = "", 
                 body: str = "", parent: str = "", 
                 decorators: List[str] = None):
        self.name = name
        self.symbol_type = symbol_type  # "function", "class", "method"
        self.language = language
        self.start_line = start_line
        self.end_line = end_line
        self.parameters = parameters
        self.docstring = docstring
        self.body = body
        self.parent = parent  # 所属类名（方法才有）
        self.decorators = decorators

class ASTParser:
    """多语言 AST 解析器"""
    def parse(self, code: str, language: str, file_path: str) -> List[CodeSymbol]:
        if language == "python":
            return self._parse_python(code, file_path)  # 使用 ast 模块
        elif language in ("javascript", "typescript"):
            return self._parse_js(code, language, file_path)  # 正则解析
        elif language == "java":
            return self._parse_java(code, file_path)
        elif language == "go":
            return self._parse_go(code, file_path)
        else:
            return self._parse_regex(code, language, file_path)  # 通用正则兜底

Python AST 解析细节如下，遍历语法树节点，区分函数定义、异步函数、类定义，并对类中的方法做递归提取：

def _parse_python(self, code: str, file_path: str) -> List[CodeSymbol]:
    """使用 Python ast 模块解析"""
    tree = ast.parse(code)
    symbols = []
    for node in ast.walk(tree):
        if isinstance(node, ast.FunctionDef) or isinstance(node, ast.AsyncFunctionDef):
            # 提取参数
            args = [a.arg for a in node.args.args]
            params = ", ".join(args)
            # 提取 docstring
            docstring = ast.get_docstring(node) or ""
            # 提取装饰器
            decorators = []
            for dec in node.decorator_list:
                if isinstance(dec, ast.Name):
                    decorators.append(dec.id)
                elif isinstance(dec, ast.Attribute):
                    decorators.append(ast.unparse(dec))
            # 获取代码体
            body_lines = code.split("\n")[node.lineno - 1:node.end_lineno]
            body = "\n".join(body_lines)
            symbols.append(CodeSymbol(
                name=node.name,
                symbol_type="function",
                language="python",
                start_line=node.lineno,
                end_line=node.end_lineno or node.lineno,
                parameters=params,
                docstring=docstring,
                body=body,
                decorators=decorators
            ))
        elif isinstance(node, ast.ClassDef):
            # 类解析...
            # 提取类方法
            for item in node.body:
                if isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef)):
                    # 方法解析...
                    symbols.append(CodeSymbol(
                        name=item.name,
                        symbol_type="method",
                        language="python",
                        parent=node.name  # 所属类名
                    ))
    return symbols

2.2 Neo4j 图谱构建

CodeGraphBuilder 是负责将 AST 解析结果写入 Neo4j 的核心类。工作流程清晰：先创建文件节点，再创建类/函数节点，随后建立文件与符号之间的 CONTAINS 关系，接着处理类继承（INHERITS）和函数调用（CALLS）关系。

# app/import_process/processors/code_graph_builder.py
class CodeGraphBuilder:
    """代码图谱构建器"""
    def __init__(self, repo_slug: str, branch: str = "main"):
        self.repo_slug = repo_slug
        self.branch = branch

    def build_graph(self, symbols: List[CodeSymbol], file_path: str) -> Dict[str, int]:
        """根据解析出的符号构建代码图谱"""
        stats = {"files": 0, "classes": 0, "functions": 0, "relationships": 0}
        driver = get_neo4j_driver()
        with driver.session() as session:
            # 1. 创建文件节点
            self._create_file_node(session, file_path)
            stats["files"] += 1
            # 2. 创建类/函数节点
            for symbol in symbols:
                if symbol.symbol_type == "class":
                    self._create_class_node(session, symbol, file_path)
                    stats["classes"] += 1
                elif symbol.symbol_type in ("function", "method"):
                    self._create_function_node(session, symbol, file_path)
                    stats["functions"] += 1
            # 3. 建立 CONTAINS 关系（文件包含类/函数）
            for symbol in symbols:
                self._create_contains_rel(session, file_path, symbol)
                stats["relationships"] += 1
            # 4. 建立 INHERITS 关系（类继承）
            for symbol in symbols:
                if symbol.symbol_type == "class" and symbol.parameters:
                    for base in symbol.parameters.split(","):
                        base = base.strip()
                        if base and base not in ("object", "Object"):
                            self._create_inherits_rel(session, symbol.name, base)
                            stats["relationships"] += 1
            # 5. 建立 CALLS 关系（函数调用）
            calls = self._extract_calls(symbols)
            for caller, callee in calls:
                self._create_calls_rel(session, caller, callee)
                stats["relationships"] += 1
        return stats

2.3 图谱节点与关系类型

节点类型一览：

关系类型定义了不同实体间的语义连接：

对应的 Cypher 语句示例：

// 创建文件节点
MERGE (f:CodeFile {path: $path, repo: $repo})
ON CREATE SET
  f.name = $name,
  f.branch = $branch,
  f.created_at = timestamp()
ON MATCH SET
  f.updated_at = timestamp()

// 创建函数节点
MERGE (fn:CodeFunction {node_id: $node_id})
ON CREATE SET
  fn.name = $name,
  fn.qualified_name = $qualified_name,
  fn.file_path = $file_path,
  fn.repo = $repo,
  fn.language = $language,
  fn.parameters = $parameters,
  fn.docstring = $docstring,
  fn.start_line = $start_line,
  fn.end_line = $end_line,
  fn.created_at = timestamp()

// 创建 CONTAINS 关系
MATCH (f:CodeFile {path: $file_path, repo: $repo})
MATCH (s:CodeFunction {node_id: $node_id})
MERGE (f)-[r:CONTAINS]->(s)
ON CREATE SET r.created_at = timestamp()

// 创建跨库 IMPLEMENTS 关系
MATCH (d:DocChunk {chunk_id: $doc_chunk_id})
MATCH (c:CodeFunction {node_id: $code_node_id})
MERGE (d)-[r:IMPLEMENTS]->(c)
ON CREATE SET r.created_at = timestamp()

三、跨库检索策略

跨库检索是 O-RAG 的核心杀手锏，采用两阶段检索 + 回退通道设计。整体流程可通过下图清晰展示：

┌─────────────────────────────────────────────────────────────────┐
│                      跨库检索流程                              │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│ 第一阶段：文档检索                                             │
│  ┌────────────────────────────────────────────────────────┐    │
│  │ Query → 向量检索 kb_docs + kb_yuque → doc_chunks      │    │
│  └────────────────────────────────────────────────────────┘    │
│  ↓                                                            │
│ 第二阶段：Neo4j 关系桥接（优先）                              │
│  ┌────────────────────────────────────────────────────────┐    │
│  │ doc_chunks → Neo4j 查询 IMPLEMENTS/MENTIONS 关系      │    │
│  │ → code_chunks                                         │    │
│  └────────────────────────────────────────────────────────┘    │
│  ↓                                                            │
│ 回退通道：无关系边时 LLM 符号抽取                            │
│  ┌────────────────────────────────────────────────────────┐    │
│  │ doc_content → LLM 抽取代码符号                        │    │
│  │ → 向量检索 kb_code                                    │    │
│  │ → code_chunks                                         │    │
│  └────────────────────────────────────────────────────────┘    │
│  ↓                                                            │
│ 关系回写：积累跨库关系                                        │
│  ┌────────────────────────────────────────────────────────┐    │
│  │ doc_chunks + code_chunks → 创建 IMPLEMENTS 关系边     │    │
│  └────────────────────────────────────────────────────────┘    │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

3.1 第一阶段：文档检索

从文档和语雀集合中检索与设计相关的内容。这一步实质上是常规的混合检索——同时使用稠密向量和稀疏向量，在 Qdrant 的 kb_docs 和 kb_yuque 两个集合中分别搜索，按分数排序取 top-k。

# app/query_process/agent/nodes/node_cross_kb_search.py
def _search_doc_collections(query: str, limit: int = 5) -> List[Dict[str, Any]]:
    """从文档/语雀集合中检索设计相关内容"""
    doc_collections = [
        qdrant_config.kb_docs_collection,
        qdrant_config.kb_yuque_collection
    ]
    embeddings = generate_embeddings([query])
    qdrant_client = get_qdrant_client()
    all_results = []
    for collection in doc_collections:
        try:
            results = hybrid_search(
                client=qdrant_client,
                collection_name=collection,
                dense_vector=embeddings["dense"][0],
                sparse_vector=embeddings["sparse"][0],
                limit=limit,
                with_payload=True
            )
            if results:
                for hit in results:
                    chunk = {
                        "chunk_id": hit.id,
                        "score": hit.score,
                        "collection": collection,
                        **(hit.payload or {})
                    }
                    all_results.append(chunk)
        except Exception as e:
            logger.error(f"文档集合 [{collection}] 检索失败: {e}")
            continue
    # 按分数排序取 top-k
    all_results.sort(key=lambda x: x.get("score", 0), reverse=True)
    return all_results[:limit]

3.2 第二阶段：Neo4j 关系桥接

拿到文档节点之后，利用它们的 chunk_id 到 Neo4j 中查找关联的代码块。分别查询 DocChunk 和 YuqueChunk 标签下 IMPLEMENTS 和 MENTIONS 两种关系，返回关联的代码函数节点。

def _neo4j_bridge(doc_chunks: List[Dict]) -> List[Dict[str, Any]]:
    """Neo4j 关系桥接：通过文档节点查找关联的代码块"""
    if not doc_chunks:
        return []
    # 提取文档节点的 chunk_id
    chunk_ids = [c["chunk_id"] for c in doc_chunks if c.get("chunk_id")]
    code_results = []
    # 分别查询 DocChunk 和 YuqueChunk 标签的关系
    for label in ["DocChunk", "YuqueChunk"]:
        try:
            results = query_cross_kb_code_chunks(
                source_label=label,
                source_chunk_ids=chunk_ids,
                rel_types=["IMPLEMENTS", "MENTIONS"],
                limit=10
            )
            code_results.extend(results)
        except Exception as e:
            logger.error(f"Neo4j 桥接查询 [{label}] 失败: {e}")
            continue
    logger.info(f"Neo4j 桥接结果: {len(doc_chunks)} 个文档节点 -> {len(code_results)} 个代码块")
    return code_results

底层的 Neo4j Cypher 查询如下：

// 查询文档节点关联的代码块
MATCH (d:DocChunk)-[r:IMPLEMENTS|MENTIONS]->(c:CodeFunction)
WHERE d.chunk_id IN $chunk_ids
RETURN c.node_id AS chunk_id, c.name AS name, c.docstring AS docstring, 
       c.file_path AS file_path, c.language AS language, type(r) AS rel_type
ORDER BY c.start_line
LIMIT $limit

3.3 回退通道：LLM 符号抽取

若 Neo4j 中不存在关系边——例如文档刚导入，尚未建立关联——则回退到 LLM 符号抽取。将文档内容送入 LLM，让其提取可能的函数名、类名、模块名，然后拿着这些符号名称到代码集合做向量检索。

def _llm_extract_symbols(doc_content: str) -> Dict[str, Any]:
    """回退通道：LLM 从文档内容中抽取代码符号"""
    prompt = load_prompt("extract_code_symbols", context=doc_content[:3000])
    llm = get_llm_client(json_mode=True)
    response = llm.invoke([HumanMessage(content=prompt)])
    content = response.content
    if content.startswith("```json"):
        content = content.replace("```json", "").replace("```", "").strip()
    symbols = json.loads(content)
    return symbols  # {"functions": [...], "classes": [...], "modules": [...]}

使用的 Prompt 如下：

你是一个代码符号抽取专家。请从以下文档内容中提取可能涉及的代码符号：
{context}
请以 JSON 格式返回，包含以下字段：
- functions: 函数名列表
- classes: 类名列表
- modules: 模块名列表
- files: 文件名列表

获取符号后，将其与原始查询拼接，再对代码集合执行向量检索：

def _search_code_by_symbols(symbols: Dict[str, Any], query: str, limit: int = 5) -> List[Dict[str, Any]]:
    """回退通道：用抽取的符号名称去代码集合做向量检索"""
    keywords = []
    for key in ("functions", "classes", "modules", "files"):
        items = symbols.get(key, [])
        if items:
            keywords.extend(items)
    if not keywords:
        return []
    # 拼接搜索文本：符号名 + 原始查询
    search_text = f"{query} {' '.join(keywords)}"
    embeddings = generate_embeddings([search_text])
    qdrant_client = get_qdrant_client()
    results = hybrid_search(
        client=qdrant_client,
        collection_name=qdrant_config.kb_code_collection,
        dense_vector=embeddings["dense"][0],
        sparse_vector=embeddings["sparse"][0],
        limit=limit,
        with_payload=True
    )
    code_chunks = []
    if results:
        for hit in results:
            code_chunks.append({
                "chunk_id": hit.id,
                "score": hit.score,
                "source": "symbol_search",
                **(hit.payload or {})
            })
    return code_chunks

3.4 关系回写：逐步积累

检索完成后，O-RAG 会尝试回写 IMPLEMENTS 关系边。这是一种渐进式积累设计——每次检索都有可能发现新的跨库关系，图谱越用越完善。为控制图谱膨胀，每次最多回写 3 个文档节点和 3 个代码节点，形成最多 9 条关系。回写失败不会影响主流程，静默处理即可。

def _write_back_relationships(doc_chunks: List[Dict], code_chunks: List[Dict]):
    """回写 IMPLEMENTS 关系边（逐步积累跨库关系）"""
    if not doc_chunks or not code_chunks:
        return
    for doc in doc_chunks[:3]:   # 最多回写 3 个文档节点
        for code in code_chunks[:3]:  # 最多回写 3 个代码节点
            doc_source_type = doc.get("source_type", "local_doc")
            source_label = "YuqueChunk" if doc_source_type == "yuque_doc" else "DocChunk"
            try:
                create_cross_kb_relationship(
                    source_label=source_label,
                    source_id=doc["chunk_id"],
                    target_label="CodeFunction",
                    target_id=code["chunk_id"],
                    rel_type="IMPLEMENTS"
                )
            except Exception as e:
                logger.debug(f"回写关系失败: {e}")

四、完整检索节点

将上述策略整合到 node_cross_kb_search 节点中，形成完整的处理流程：

# app/query_process/agent/nodes/node_cross_kb_search.py
def node_cross_kb_search(state: QueryGraphState) -> Dict[str, Any]:
    """跨库链路检索节点"""
    logger.info(f">>> [node_cross_kb_search] 开始跨库链路检索")
    try:
        query = state.get("rewritten_query") or state.get("original_query", "")
        # 第一阶段：文档检索
        doc_results = _search_doc_collections(query, limit=5)
        logger.info(f"文档检索结果: {len(doc_results)} 条")
        if not doc_results:
            logger.info("文档检索无结果，跳过跨库检索")
            return {"cross_kb_docs": [], "extracted_symbols": {}}

        # 第二阶段：Neo4j 关系桥接（优先）
        code_results = _neo4j_bridge(doc_results)
        extracted_symbols = {}

        # 回退通道：无关系边时 LLM 抽取符号
        if not code_results:
            logger.info("Neo4j 无关系边，回退到 LLM 符号抽取")
            doc_content = "\n\n".join([d.get("content", "") for d in doc_results[:3]])
            extracted_symbols = _llm_extract_symbols(doc_content)
            if extracted_symbols:
                code_results = _search_code_by_symbols(extracted_symbols, query, limit=5)

        # 回写关系边（积累跨库关系）
        if code_results:
            _write_back_relationships(doc_results, code_results)

        # 组装跨库结果：文档 + 代码
        cross_kb_docs = []
        # 添加文档结果
        for doc in doc_results:
            cross_kb_docs.append({
                "chunk_id": doc.get("chunk_id"),
                "content": doc.get("content", ""),
                "title": doc.get("title", ""),
                "source_type": doc.get("source_type", "local_doc"),
                "score": doc.get("score", 0),
                "role": "design_doc"  # 标记为设计文档
            })
        # 添加代码结果
        for code in code_results:
            cross_kb_docs.append({
                "chunk_id": code.get("chunk_id"),
                "content": code.get("content", code.get("docstring", "")),
                "title": code.get("name", code.get("title", "")),
                "source_type": "code_repo",
                "source": code.get("file_path", ""),
                "score": code.get("score", 0.9),
                "role": "code_block",  # 标记为代码块
                "language": code.get("language", ""),
                "rel_type": code.get("rel_type", "")  # IMPLEMENTS/MENTIONS
            })
        logger.info(f"跨库检索完成: {len(doc_results)} 文档 + {len(code_results)} 代码")
        return {"cross_kb_docs": cross_kb_docs, "extracted_symbols": extracted_symbols}
    except Exception as e:
        logger.error(f"[node_cross_kb_search] 跨库检索失败: {e}", exc_info=True)
        return {"cross_kb_docs": [], "extracted_symbols": {}, "error": str(e)}

五、检索路由与融合

5.1 条件路由

跨库检索并非每次都会触发——系统根据 search_type 字段决定走普通多路检索还是跨库链路检索。该路由逻辑通过条件边实现：

# app/query_process/agent/main_graph.py
def route_after_confirm(state: QueryGraphState) -> str:
    """根据 search_type 路由"""
    search_type = state.get("search_type", "all")
    if search_type == "cross_kb":
        return "cross_kb"  # 跨库链路检索
    return "normal"  # 普通多路检索

builder.add_conditional_edges(
    "node_item_name_confirm",
    route_after_confirm,
    {
        "cross_kb": "node_cross_kb_search",
        "normal": "node_search_embedding"
    }
)

5.2 RRF 融合

跨库检索的结果在 RRF 融合阶段会获得最高权重 1.2，这意味着系统优先信任通过关系桥接找到的代码结果：

# app/query_process/agent/nodes/node_rrf.py
def node_rrf(state: QueryGraphState) -> QueryGraphState:
    """RRF 融合节点"""
    embedding_chunks = state.get("embedding_chunks") or []
    hyde_embedding_chunks = state.get("hyde_embedding_chunks") or []
    web_search_docs = state.get("web_search_docs") or []
    cross_kb_docs = state.get("cross_kb_docs") or []

    # 构建带权重的来源列表
    source_with_weight = []
    if embedding_chunks:
        source_with_weight.append((embedding_chunks, 1.0))
    if hyde_embedding_chunks:
        source_with_weight.append((hyde_embedding_chunks, 0.9))
    if web_search_docs:
        source_with_weight.append((web_search_docs, 0.5))
    if cross_kb_docs:
        source_with_weight.append((cross_kb_docs, 1.2))  # 跨库权重最高

    # 执行 RRF 融合
    rrf_response = step_3_reciprocal_rank_fusion(source_with_weight, top_k=5)
    state["rrf_chunks"] = rrf_response
    return state

六、实战场景

6.1 场景：从需求文档到代码实现

假设用户提问：“登录功能的设计方案和对应代码实现”。整个检索链路如下：

1. 第一阶段：文档检索
   Query → 向量检索 kb_docs + kb_yuque
   结果: 
   - doc_001: "登录功能需求文档 - 用户输入账号密码..."
   - doc_002: "认证模块设计文档 - JWT Token 验证..."

2. 第二阶段：Neo4j 关系桥接
   doc_001 (chunk_id=xxx) → Neo4j 查询 IMPLEMENTS 关系
   结果:
   - code_001: "LoginService.authenticate()" ← IMPLEMENTS
   - code_002: "AuthController.login()" ← IMPLEMENTS

3. 关系回写
   创建/更新 IMPLEMENTS 关系边

4. RRF 融合
   cross_kb_docs (权重 1.2) + 其他来源 → RRF 排序

5. Rerank 精排
   文档 + 代码混合精排

6. 答案生成
   LLM 基于文档和代码生成综合答案

6.2 场景：新文档无关系边

如果用户问的是“新导入的支付模块设计”，此时文档刚入库，Neo4j 中尚未建立任何关系边。回退通道便发挥作用：

1. 第一阶段：文档检索
   结果: doc_003 "支付模块设计文档..."

2. 第二阶段：Neo4j 关系桥接
   doc_003 → Neo4j 查询 → 无 IMPLEMENTS 关系（新文档）

3. 回退通道：LLM 符号抽取
   doc_003 内容 → LLM 抽取
   结果: {"functions": ["process_payment", "validate_order"], "classes": ["PaymentService"]}

4. 符号向量检索
   "支付模块设计 process_payment validate_order PaymentService" → 向量检索 kb_code
   结果: code_003 "PaymentService.process_payment()"

5. 关系回写
   创建 IMPLEMENTS 关系边（下次检索可直接桥接）

七、性能优化

7.1 批量查询

Neo4j 查询使用批量 chunk_ids，一次查询即可获取多个文档的关联代码，大幅减少网络往返：

# 批量查询，而非逐个查询
results = query_cross_kb_code_chunks(
    source_label=label,
    source_chunk_ids=chunk_ids,  # 批量传递
    rel_types=["IMPLEMENTS", "MENTIONS"],
    limit=10
)

7.2 索引优化

为 Neo4j 节点创建索引，加速属性查询：

// 创建索引
CREATE INDEX doc_chunk_id IF NOT EXISTS FOR (d:DocChunk) ON (d.chunk_id);
CREATE INDEX yuque_chunk_id IF NOT EXISTS FOR (y:YuqueChunk) ON (y.chunk_id);
CREATE INDEX code_function_node_id IF NOT EXISTS FOR (f:CodeFunction) ON (f.node_id);
CREATE INDEX code_file_path IF NOT EXISTS FOR (f:CodeFile) ON (f.path);

7.3 限制结果数量

每一层都施加了数量限制，避免结果爆炸：文档检索只取 5 条，代码检索限制 10 条，关系回写最多 3×3 条。精打细算才能跑得稳。

# 文档检索限制
doc_results = _search_doc_collections(query, limit=5)
# 代码检索限制
code_results = _neo4j_bridge(doc_results)  # limit=10 in query
# 关系回写限制
for doc in doc_results[:3]:   # 最多 3 个文档
    for code in code_chunks[:3]:  # 最多 3 个代码
        _write_back_relationship(...)

八、图谱可视化

在 Neo4j Browser 中，可以直观查看跨库关系。例如查询某个文档的跨库关联：

// 查询某个文档的跨库关系
MATCH (d:DocChunk {chunk_id: "xxx"})-[r:IMPLEMENTS]->(c:CodeFunction)
RETURN d, r, c

或者追踪代码调用链：

// 查询代码调用链
MATCH (f1:CodeFunction)-[r:CALLS]->(f2:CodeFunction)
WHERE f1.name = "process_payment"
RETURN f1, r, f2

以及查看类继承关系：

// 查询类继承关系
MATCH (c1:CodeClass)-[r:INHERITS]->(c2:CodeClass)
RETURN c1, r, c2

九、总结

跨库链路检索是 O-RAG 系统最核心的差异化特性。设计要点可归纳为五点：

1. 双模共存：Qdrant 分集合存储 + Neo4j 关系桥接，兼顾语义检索与关系推理，各取所长。
2. 两阶段检索：文档检索 → Neo4j 桥接 → 代码取回，精准建立跨源关联，无需全量扫描。
3. 回退通道：无关系边时由 LLM 抽取符号 + 向量检索兜底，确保新导入内容仍可被检索。
4. 关系回写：检索过程中逐步积累跨库关系，图谱越用越完善，形成正向循环。
5. 权重设计：跨库结果在 RRF 融合阶段权重为 1.2，优先被采纳，体现了“设计文档驱动代码检索”的设计哲学。

这套方案解决了传统 RAG 系统无法处理跨源关系的核心痛点，为“从需求到代码”的全链路知识检索提供了坚实的技术支撑。

附录：完整代码结构

app/
├── clients/
│   ├── neo4j_utils.py        # Neo4j 客户端工具
│   └── qdrant_utils.py       # Qdrant 客户端工具
├── import_process/
│   ├── agent/
│   │   ├── main_graph.py     # 导入工作流
│   │   └── nodes/
│   │       ├── node_import_kg.py   # 知识图谱导入
│   │       └── ...
│   └── processors/
│       ├── ast_parser.py     # AST 多语言解析
│       ├── code_graph_builder.py   # 代码图谱构建
│       ├── git_fetcher.py    # Gitee 仓库抓取
│       └── yuque_sync.py     # 语雀文档同步
├── query_process/
│   ├── agent/
│   │   ├── main_graph.py     # 查询工作流
│   │   └── nodes/
│   │       ├── node_cross_kb_search.py   # 跨库检索节点
│   │       ├── node_rrf.py   # RRF 融合节点
│   │       └── ...
└── utils/
    └── source_type.py        # 数据源类型枚举

系列文章导航：

1. O-RAG 系统架构设计：LangGraph 状态机与多源异构 RAG
2. 多路召回与融合排序：RRF + Rerank + 动态 TopK 工程实践
3. 跨库链路检索：Neo4j 图数据库桥接文档与代码（本文）