基于 LLM + 四阶段 Pipeline 的知识图谱自然语言查询系统设计与实现

本文介绍了在网络安全知识图谱场景下，如何基于 LLM（大语言模型）实现自然语言查询图数据库的系统设计。系统采用 四阶段 Pipeline 架构，将用户的自然语言问题转化为可执行的 Gremlin 查询语句，并通过 自修正机制 保证查询的准确性。同时实现了 MCP Server 标准协议集成、多轮会话上下文管理等高级能力。

---

一、背景与挑战

知识图谱在网络安全领域被广泛应用于威胁情报、资产关系、漏洞管理等场景。然而 Gremlin 查询语言门槛高，普通分析师难以直接使用。我们面临的挑战是：

1. 查询门槛高：Gremlin 语法复杂，非技术人员无法直接查询图谱
2. Schema 复杂：网络安全图谱包含几十种顶点标签和边标签，LLM 上下文容易溢出
3. 查询准确性：LLM 生成的 Gremlin 经常存在语法错误或执行失败
4. 多轮对话：用户需要追问式查询（“其中高危的有哪些？”“那个漏洞影响了哪些软件？”）
5. 系统集成：需要与现有 AI Agent 生态（如 MCP 协议）无缝对接

---

二、系统架构总览

系统部署在 ai-xx-common-graph 微服务中，通过 Feign 被其他服务调用。前端通过 Vuex Store 与后端 API 交互。

---

三、核心：四阶段 Pipeline 设计

整个查询过程由 PipelineEngine 编排，分为四个阶段，数据通过 PipelineContext 在阶段间传递。

3.1 PipelineContext — 阶段间的数据载体

public class PipelineContext {
    // === 输入 ===
    private String query;          // 用户原始问题
    private String language;       // CN / EN

    // === Phase 1 输出 ===
    private GraphSchema fullSchema;        // 全量图 Schema
    private ParsedIntent parsedIntent;     // 解析后的结构化意图
    private List<ExamplePair> matchedExamples; // 相似查询示例

    // === Phase 2 输出 ===
    private String filteredSchema;              // 精选后的 Schema（JSON）
    private List<ResolvedEntity> resolvedEntities; // 解析后的实体

    // === Phase 3 输出 ===
    private String finalGremlin;              // 最终的 Gremlin 查询
    private List<CorrectionRecord> correctionHistory; // 修正历史

    // === Phase 4 输出 ===
    private Object executionResult;           // 执行结果
    private String naturalLanguageAnswer;     // 自然语言答案

    // === 元数据 ===
    private int llmCallCount;     // LLM 调用次数
    private long startTimeMs;     // 开始时间
}

3.2 Phase 1：并行准备

Phase 1 包含三个并行任务：

意图解析 Prompt 模板

这是核心的 Prompt Engineering 部分。将用户问题 + 图 Schema 作为输入，要求 LLM 输出结构化的 JSON：

你是图查询意图解析器。根据用户问题和图中可用的标签，提取结构化意图。

可用顶点标签: {vertex_labels}
可用边标签: {edge_labels}

用户问题: {query}

输出 JSON（不要 markdown 代码块，直接输出 JSON）:
{
  "entityMentions": [{"text": "实体名", "typeHint": "最可能的顶点标签"}],
  "targetVertexLabels": ["涉及到的顶点标签"],
  "targetEdgeLabels": ["涉及到的边标签"],
  "filters": [{"property": "属性名", "operator": "==|>|<|>=", "value": "值"}],
  "aggregation": null,
  "returnType": "vertices",
  "isComplex": false,
  "complexityReason": null,
  "reasoningSteps": []
}

规则:
1. entityMentions 中的 typeHint 必须是可用顶点标签之一
2. targetVertexLabels 和 targetEdgeLabels 只能包含可用标签
3. 如果查询涉及多步关联（需要两次以上遍历），设置 isComplex=true
4. aggregation 仅在用户明确要求统计时设置（如"有多少个"、"求平均"）

设计要点：

• 将 Schema 信息注入 Prompt，约束 LLM 只生成合法的标签和属性
• typeHint 字段为实体解析提供方向（知道"APT-28"应该去 group 标签下查找）
• isComplex 标志决定是否触发复杂查询拆解（Phase 4）

3.3 Phase 2：Schema 精选 + 实体解析

Schema 精选（纯规则，不调 LLM）

意图解析后，全量 Schema 可能包含几十个标签。为减少 LLM 上下文消耗，用纯规则筛选：

public GraphSchema filter(GraphSchema fullSchema, ParsedIntent parsedIntent) {
    // 1. 收集意图中的目标顶点标签和边标签
    Set<String> targetVLabels = new HashSet<>(parsedIntent.getTargetVertexLabels());

    // 2. 一跳扩展：目标边标签连接的顶点标签也纳入
    for (EdgeLabel edge : fullSchema.getEdgelabels()) {
        if (targetELabels.contains(edge.getName())) {
            targetVLabels.add(edge.getSourceLabel());
            targetVLabels.add(edge.getTargetLabel());
        }
    }

    // 3. 空集降级：如果过滤后为空，返回全量 Schema
    if (filteredVLabels.isEmpty()) {
        return fullSchema;
    }
    // 4. 级联过滤属性键（只保留相关标签用到的属性）
}

实体解析（三级置信度）

将意图中提取的实体提及（如 “APT-28”）映射为图中的真实顶点 ID：

3.4 Phase 3：Gremlin 生成 + 自修正循环

这是系统最关键的创新点。LLM 生成的 Gremlin 经常出错，我们设计了 Try-Correct 循环 自动修正：

public void generate(PipelineContext ctx) {
    int maxRounds = properties.getPipeline().getMaxCorrectionRounds(); // 默认 3
    List<CorrectionRecord> history = new ArrayList<>();

    // 初始生成
    String gremlin = doGenerate(ctx);
    history.add(new CorrectionRecord(0, gremlin, null, null));

    // 自修正循环
    for (int round = 1; round <= maxRounds; round++) {
        String currentGremlin = history.get(history.size() - 1).getGremlin();

        // Step 1: 语法校验
        SyntaxValidationResult syntaxResult = syntaxValidator.validate(currentGremlin);
        if (!syntaxResult.isValid()) {
            String corrected = doCorrect(currentGremlin, syntaxResult.getErrorMessage(), ctx);
            history.add(new CorrectionRecord(round, corrected, "syntax", syntaxResult.getErrorMessage()));
            continue;
        }

        // Step 2: 安全追加 limit
        currentGremlin = GremlinUtils.ensureLimit(currentGremlin, 100);

        // Step 3: 执行 Gremlin
        Object result = graphClient.executeGremlin(currentGremlin);

        // Step 4: 检查空结果
        if (isEmptyResult(result)) {
            String corrected = doCorrect(currentGremlin, "查询返回空结果，请尝试放宽条件", ctx);
            history.add(new CorrectionRecord(round, corrected, "empty_result", "查询返回空结果"));
            continue;
        }

        // 成功 — 保存结果并退出循环
        ctx.setFinalGremlin(currentGremlin);
        ctx.setExecutionResult(result);
        ctx.setCorrectionHistory(history);
        return;
    }

    // 重试耗尽，使用最后一轮结果
}

Gremlin 生成 Prompt

你是 Apache HugeGraph Gremlin 查询专家。根据以下信息生成精确的 Gremlin 查询。

图谱 Schema（已筛选）:
{filtered_schema}

已知实体 ID（直接使用 g.V('id') 引用）:
{resolved_entities}

参考示例:
{examples}

查询意图分析:
{intent_summary}

用户问题: {query}

规则:
1. 有实体 ID 时直接使用 g.V('id') 而非 has() 查找
2. 只使用 Schema 中存在的标签和属性
3. 始终加 .limit(100)
4. 不要返回 g.V().limit(0)，总是尝试生成可执行的查询
5. 只输出 ```gremlin ... ```包裹的查询，不要其他文字

Gremlin 修正 Prompt

当生成失败时，将失败的查询 + 错误信息回传给 LLM 修正：

以下 Gremlin 查询有错误，需要修正。

用户问题: {query}
图谱 Schema: {filtered_schema}
已知实体 ID: {resolved_entities}

失败的查询:
```gremlin
{failed_gremlin}

错误信息: {error_message}

修正规则:

1. 修复上述具体错误
2. 只使用 Schema 中存在的标签和属性
3. 尽量保持原始查询意图
4. 始终加 .limit(100)
5. 只输出 gremlin ... 包裹的修正后查询

**关键设计**：每次修正都会传入完整的 Schema 和实体信息，防止 LLM 在修正过程中引入不存在的标签。

### 3.5 Phase 4：可选后处理

```mermaid
graph TD
    P4["Phase 4（可选后处理）"]
    P4 --> AS["AnswerSynthesizer\n自然语言答案合成"]
    P4 --> QD["QueryDecomposer\n复杂查询拆解"]
    AS --> AS1["查询结果 + 用户问题 → 自然语言回答"]
    QD --> QD1["复杂查询 → 多个子查询顺序执行"]

    style P4 fill:#e8eaf6
    style AS fill:#e3f2fd
    style QD fill:#fce4ec

自然语言答案合成 Prompt：

你是网络安全领域的图谱分析助手。根据用户的原始问题和图数据库的查询结果，用自然语言总结回答。

用户问题: {query}
查询结果:
{execution_result}

规则:
1. 用简洁专业的中文回答
2. 包含关键数据点（数量、名称、评分等）
3. 如果结果为空，说明未找到相关信息
4. 不要编造数据

---

四、多轮会话与上下文管理

单轮查询无法满足实际分析需求。我们实现了完整的多轮会话管理，支持追问式查询。

4.1 上下文分类

首先判断当前输入是独立查询还是追问：

你是一个对话上下文分类器。判断当前用户输入是独立查询还是追问。

规则：
- 包含完整实体名和明确查询意图 -> independent
- 包含代词（它、那个）或过滤词（其中、上述、高危的）-> follow_up
- 无法判断时 -> independent，confidence < 0.6

输出 JSON:
{"type":"independent"或"follow_up","confidence":0.0-1.0}

4.2 指代消解

对于追问式查询，需要将指代替换为完整查询：

你是一个图谱查询指代消解器。根据历史对话上下文，将当前追问中的指代替换为完整的查询。

规则：
- 将"其中"替换为上次查询的具体范围
- 将"那个漏洞"替换为具体的漏洞编号
- 合并上次查询条件与新条件
- 如果无法消解，resolvedQuery 使用原始查询

输出 JSON:
{"resolvedQuery":"消解后的完整查询","inheritedContext":{}}

4.3 上下文压缩与滑动窗口

为防止上下文无限膨胀，实现了 压缩 + 滑动窗口 双重机制：

4.4 零侵入 Pipeline 的上下文注入

上下文管理不修改 PipelineEngine 本身，而是在 SessionService.send() 中构建有效查询：

private String buildEffectiveQuery(String query, SessionContext context) {
    StringBuilder sb = new StringBuilder();
    // 注入压缩摘要
    if (context.getCompressedSummary() != null) {
        sb.append("[历史摘要] ").append(context.getCompressedSummary()).append("\n\n");
    }
    // 注入最近对话
    if (context.getRecentTurns() != null) {
        sb.append("[最近对话]\n");
        for (Object turn : context.getRecentTurns()) {
            sb.append(turn).append("\n");
        }
        sb.append("\n");
    }
    sb.append(query);
    return sb.toString();
}

这种设计保证了 PipelineEngine 的纯净性，上下文信息通过查询前缀注入，对 Pipeline 透明。

---

五、MCP Server 集成

系统实现了 Model Context Protocol (MCP) 标准，允许外部 AI Agent 调用图谱查询能力。

5.1 SSE 连接模式

5.2 注册的 MCP Tools

工具名称	描述	输入参数
query_graph	自然语言查询图谱（完整 Pipeline）	query（必填）、language、enable_answer
parse_intent	仅解析查询意图	query（必填）
get_graph_schema	获取图谱 Schema	无
execute_gremlin	执行只读 Gremlin	gremlin（必填）、limit

MCP Server 的价值在于：任何支持 MCP 协议的 AI Agent（如 Claude Desktop、Cursor 等）都能直接获得图谱查询能力，无需额外开发。

---

六、API 设计

6.1 V1 — 单次查询 API

端点	方法	描述
/api/v1/nl2graph/query	POST	主查询接口（完整 Pipeline）
/api/v1/nl2graph/parse	POST	仅解析意图（调试用）
/api/v1/nl2graph/gremlin	POST	仅生成 Gremlin（不执行）
/api/v1/nl2graph/schema	GET	获取图 Schema
/api/v1/nl2graph/execute	POST	安全执行只读 Gremlin

6.2 V2 — 会话管理 API

端点	方法	描述
/api/v2/nl2graph/session	POST	创建新会话
/api/v2/nl2graph/session/{id}/send	POST	发送消息（带上下文）
/api/v2/nl2graph/session/{id}/history	GET	获取会话历史
/api/v2/nl2graph/session/{id}	DELETE	删除会话
/api/v2/nl2graph/sessions	GET	列出活跃会话

6.3 请求/响应示例

请求：

POST /api/v1/nl2graph/query
{
  "query": "查询 APT-28 使用了哪些恶意软件",
  "language": "CN",
  "enableAnswer": true
}

响应：

{
  "code": 0,
  "data": {
    "query": "查询 APT-28 使用了哪些恶意软件",
    "parsedIntent": {
      "entityMentions": [{"text": "APT-28", "typeHint": "group"}],
      "targetVertexLabels": ["group", "malware"],
      "targetEdgeLabels": ["group_uses_malware"],
      "isComplex": false
    },
    "resolvedEntities": [
      {"text": "APT-28", "vid": "APT-28", "label": "group", "confidence": "CONFIDENT"}
    ],
    "templateGremlin": "g.V('APT-28').out('group_uses_malware').limit(100)",
    "executionResult": [
      {"id": "Cobalt Strike", "label": "malware", "type": "trojan", "name": "Cobalt Strike"}
    ],
    "correctionHistory": [
      {"round": 0, "gremlin": "g.V('APT-28').out('group_uses_malware').limit(100)"}
    ],
    "llmCallCount": 1,
    "elapsedMs": 2350,
    "naturalLanguageAnswer": "APT-28（芬尼熊）使用了 1 款恶意软件：Cobalt Strike（类型：trojan）。"
  }
}

---

七、前端实现

前端采用 Vue 2 + Element UI + AntV G6，实现了完整的对话式图谱查询界面。

7.1 组件架构

pages/nl2graph/
├── index.vue                    # 主页面（双布局：首页输入 / 对话内容）
└── components/
    ├── QueryInput.vue            # 查询输入框（快捷标签 + 高级参数）
    ├── ChatArea.vue              # 对话区域（流式渲染 + 实体高亮）
    ├── MessageBubble.vue         # 消息气泡（流式 token 渲染）
    ├── GraphPanel.vue            # 图谱可视化（G6 力导向布局）
    ├── SessionList.vue           # 会话列表（localStorage 持久化）
    ├── BottomPanel.vue           # 底部面板（结果详情）
    └── NodeDetail.vue            # 节点详情侧边栏

7.2 关键交互

1. 实体可点击：对话中识别到的实体（如 “APT-28”）渲染为可点击标签，点击后在图谱面板中定位并高亮对应节点
2. 流式渲染：通过 SSE 实时渲染 AI 回答，逐 token 展示
3. 图谱追加：每次查询的图谱结果追加到已有图谱中（appendGraphData），而非替换，形成探索式体验
4. 会话持久化：会话列表存储在 localStorage，刷新页面后自动恢复

7.3 Vuex Store 设计

// store/nl2graph.js
state: {
  currentSessionId: null,
  sessions: [],           // 会话列表
  streaming: false,        // 是否正在流式接收
  eventSource: null,        // SSE 连接实例
}

---

八、关键设计决策与经验

8.1 Schema 精选而非全量注入

为什么不全量注入 Schema 给 LLM？

网络安全图谱可能有 50+ 个标签和数百个属性，全部注入会导致：

• Token 消耗过大（成本问题）
• LLM 注意力分散，选择错误标签的概率上升
• 响应变慢

方案：先让 LLM 仅根据标签名列表判断涉及哪些标签，再用规则筛选出相关 Schema 子集，最后用精选 Schema 生成 Gremlin。两步走策略。

8.2 自修正循环

LLM 生成的 Gremlin 错误率约 30-50%，如何处理？

传统的"生成即交付"模式不可行。我们设计了 Try-Correct 循环：

最多修正 3 轮。实践中绝大多数错误在 1-2 轮内可修正。修正 Prompt 包含失败查询和错误信息，LLM 能精准定位问题。

8.3 安全性设计

// 1. 只读模式（默认开启）
if (!GremlinUtils.isReadOnly(gremlin)) {
    return R.fail().code(-1).message("只允许只读查询操作");
}

// 2. 结果数量限制
currentGremlin = GremlinUtils.ensureLimit(currentGremlin, 100);

// 3. 实体解析使用 has() 而非全表扫描
String gremlin = String.format("g.V().hasLabel('%s').has('name','%s').limit(5)", label, text);

8.4 多 LLM 提供商支持

通过 LlmClient 抽象层支持多种 LLM 后端：

nl2graph:
  llm:
    type: openai          # openai / ollama
    baseUrl: https://api.openai.com/v1
    chatModel: gpt-4o-mini
    # Ollama 本地部署
    ollamaHost: 127.0.0.1
    ollamaPort: 11434
    ollamaModel: qwen2.5

用户可选择商业 API（OpenAI 兼容）或本地部署（Ollama），适应不同的安全要求。

8.5 中英文双语支持

所有 Prompt 模板都有 CN/EN 两个版本，根据用户查询语言自动选择：

resources/prompts/
├── intent-parse-cn.txt        # 中文意图解析
├── intent-parse-en.txt        # 英文意图解析
├── gremlin-generate-cn.txt    # 中文 Gremlin 生成
├── gremlin-generate-en.txt    # 英文 Gremlin 生成
├── gremlin-correct-cn.txt     # 中文 Gremlin 修正
├── gremlin-correct-en.txt     # 英文 Gremlin 修正
├── context-classify-cn.txt    # 上下文分类
├── context-resolve-cn.txt     # 指代消解
├── context-compress-cn.txt    # 上下文压缩
├── answer-synthesize-cn.txt   # 答案合成
└── query-decompose-cn.txt    # 复杂查询拆解

---

九、完整流程示例

以一次实际查询为例，展示完整的数据流转：

用户输入："查询 APT-28 使用了哪些恶意软件"

---

十、配置参考

完整的配置项：

nl2graph:
  hugegraph:
    url: "http://localhost:8080"
    graphName: "hugegraph"
  llm:
    type: "openai"              # openai / ollama
    baseUrl: "https://api.openai.com/v1"
    chatModel: "gpt-4o-mini"
    maxTokens: 4096
    timeoutSeconds: 60
  pipeline:
    language: "CN"               # 默认语言
    maxCorrectionRounds: 3       # 最大修正轮次
    maxSubQueries: 4             # 复杂查询最大子查询数
    defaultLimit: 100           # 默认结果限制
    exampleNum: 3               # Few-shot 示例数量
    fuzzyScoreThreshold: 0.7    # 模糊匹配阈值
  security:
    readOnly: true               # 只读模式
    maxResultSize: 1000          # 最大结果集大小
  session:
    ttl: 1800                    # 会话 TTL（秒）
    maxMessages: 50              # 单会话最大消息数
    contextWindow: 10            # 上下文窗口大小
  context:
    maxSummaryTokens: 200        # 摘要最大 Token 数
    maxRecentTurns: 5            # 保留最近对话轮数
    tokenBudget: 3000            # 会话 Token 预算
  mcp:
    sseEnabled: true             # MCP SSE 开关
    sseTimeout: 1800             # SSE 超时（秒）
    maxConnections: 50           # 最大连接数

---

十一、总结与展望

核心价值

1. 降低查询门槛：非技术人员可通过自然语言直接查询知识图谱
2. 自修正保障准确性：Try-Correct 循环将 Gremlin 生成准确率从约 50% 提升到 90%+
3. 多轮对话支持：压缩 + 滑动窗口的上下文管理方案，支持追问式分析
4. 标准协议集成：MCP Server 让图谱查询能力可被任何 AI Agent 调用
5. 多 LLM 后端：支持 OpenAI API 和本地 Ollama，适应不同安全级别

结语

本文分享的是我们在网络安全知识图谱场景下，用 LLM 实现自然语言查询图数据库的一次工程实践。从最初 “让分析师不用写 Gremlin” 这个朴素的想法出发，逐步演进出了四阶段 Pipeline、自修正循环、多轮上下文管理等机制，目前已在实际业务中落地运行。

但坦白说，这套方案仍有不少值得探讨和改进的地方：比如自修正循环增加了延迟和 Token 消耗，在大型图谱上的性能表现还有优化空间；向量化实体索引目前还是半成品；复杂多跳查询的准确率也有待提升。这些既是当前的不足，也是下一步可以探索的方向。

抛砖引玉，希望能给在做类似尝试的同学一些参考。如果你有更好的思路或踩过类似的坑，欢迎随时交流探讨。