# 用 Gemini 从非结构化文本构建知识图谱：一篇可落地教程

> 来源：HackerNoon《Building Knowledge Graphs with Gemini》及其配套 Google notebook。  
> 原文 URL：https://hackernoon.com/building-knowledge-graphs-with-gemini  
> 配套 notebook：https://github.com/GoogleCloudPlatform/generative-ai/blob/main/gemini/use-cases/knowledge-graph/knowledge_graph_generation.ipynb  
> 本教程不是逐字翻译，而是把原文方法改写成一个可执行的工程落地流程。

## 你将做出什么

完成后，你会得到一个最小知识图谱抽取流水线：

1. 输入一段非结构化文本。
2. 用提示词让 Gemini 抽取实体和关系。
3. 要求 Gemini 输出 TSV，而不是 JSON。
4. 用 Python 解析 TSV。
5. 构建节点和边。
6. 验证输出是否可入库或可视化。

为了降低试错成本，教程先用 **mock 输出**跑通解析和校验，再给出 Gemini API 替换点。这样即使暂时没有 API key，也能先验证工程骨架。

## 适用场景

适合这些任务：

- 从会议纪要中抽取“人、系统、任务、依赖关系”。
- 从运维故障报告中抽取“服务、告警、根因、处置动作”。
- 从合同或制度文档中抽取“主体、义务、期限、约束”。
- 从长篇故事/书籍中抽取“人物、地点、关系网络”。

不适合直接用于这些任务：

- 未经人工复核的生产级事实库。
- 法务、医疗、金融等高风险自动决策。
- 需要强一致实体消歧的大规模图数据库写入。

## 总体架构

```text
raw text
  │
  ▼
Gemini extraction prompt
  │
  ▼
TSV output
  │
  ├── entities.tsv: id, name, label
  │
  └── relations.tsv: source_id, link, target_id
  │
  ▼
Python parser + validator
  │
  ▼
Graph object / NetworkX / Neo4j / SQLite
  │
  ▼
visualization, search, analysis
```

核心设计选择：

| 设计点 | 推荐做法 | 原因 |
|---|---|---|
| 输出格式 | TSV 优先，JSON 备选 | 实体/关系本质是表，TSV 更省 Token |
| 模型参数 | `temperature=0.0`, `top_p=0.0` | 抽取任务要稳定，不要创作性 |
| 提示词约束 | 明确“只使用输入数据” | 降低模型调用训练记忆补全事实的概率 |
| 关系方向 | 明确双向/反向规则 | 避免图谱只抽到单向边 |
| 生产落地 | 先校验，再入库 | LLM 输出不能直接信任 |

## 第 1 步：准备环境

如果你只跑 mock 版本，Python 标准库即可。

如果你要接 Gemini 和画图，安装：

```bash
python3 -m venv .venv
source .venv/bin/activate
pip install "google-genai>=2.6.0" "networkx[default]" tenacity
```

API key 不要写进代码。使用环境变量：

```bash
export GOOGLE_API_KEY="*** Gemini API Key ***"
```

> 说明：原文也支持 Vertex AI / Agent Platform 路线；本教程以 Google AI Studio API key 路线说明，便于最小化落地。

## 第 2 步：先定义输出契约

不要一开始就让模型“自由总结关系”。先固定输出协议。

实体表：

```text
id	name	label
0	Henry Jones	person
1	Sophie Jones	person
```

关系表：

```text
source_id	link	target_id
0	father_of	1
1	child_of	0
```

关系命名建议：

- 用小写英文和下划线：`father_of`, `works_at`, `depends_on`。
- 不要用自然语言长句做关系名。
- 同一类关系只保留一种命名，不要混用 `employee_of` 和 `works_for`。

## 第 3 步：写一个可执行的 mock-first 解析器

先新建文件：

```bash
mkdir -p kg_demo
cd kg_demo
cat > kg_from_tsv.py <<'PY'
from __future__ import annotations

from dataclasses import dataclass
from typing import Iterable


@dataclass(frozen=True)
class Entity:
    id: str
    name: str
    label: str


@dataclass(frozen=True)
class Relation:
    source_id: str
    link: str
    target_id: str


@dataclass(frozen=True)
class KnowledgeGraph:
    entities: list[Entity]
    relations: list[Relation]


def parse_tsv_table(text: str, expected_header: list[str]) -> list[dict[str, str]]:
    lines = [line.strip() for line in text.strip().splitlines() if line.strip()]
    if not lines:
        return []

    header = lines[0].split("\t")
    if header != expected_header:
        raise ValueError(f"Unexpected header: {header}, expected: {expected_header}")

    rows: list[dict[str, str]] = []
    for line_number, line in enumerate(lines[1:], start=2):
        values = line.split("\t")
        if len(values) != len(expected_header):
            raise ValueError(f"Line {line_number} has {len(values)} columns: {line!r}")
        rows.append(dict(zip(expected_header, values, strict=True)))
    return rows


def parse_entities(text: str) -> list[Entity]:
    rows = parse_tsv_table(text, ["id", "name", "label"])
    return [Entity(id=row["id"], name=row["name"], label=row["label"]) for row in rows]


def parse_relations(text: str) -> list[Relation]:
    rows = parse_tsv_table(text, ["source_id", "link", "target_id"])
    return [
        Relation(source_id=row["source_id"], link=row["link"], target_id=row["target_id"])
        for row in rows
    ]


def validate_graph(graph: KnowledgeGraph) -> None:
    entity_ids = {entity.id for entity in graph.entities}
    if len(entity_ids) != len(graph.entities):
        raise ValueError("Duplicate entity ids found")

    for relation in graph.relations:
        if relation.source_id not in entity_ids:
            raise ValueError(f"Missing source entity: {relation.source_id}")
        if relation.target_id not in entity_ids:
            raise ValueError(f"Missing target entity: {relation.target_id}")


def format_edges(graph: KnowledgeGraph) -> list[str]:
    names = {entity.id: entity.name for entity in graph.entities}
    return [
        f"{names[relation.source_id]} --{relation.link}--> {names[relation.target_id]}"
        for relation in graph.relations
    ]


def build_graph_from_tsv(entities_tsv: str, relations_tsv: str) -> KnowledgeGraph:
    graph = KnowledgeGraph(
        entities=parse_entities(entities_tsv),
        relations=parse_relations(relations_tsv),
    )
    validate_graph(graph)
    return graph


def main() -> None:
    # Mock Gemini TSV output for the first validation loop.
    entities_tsv = """id	name	label
0	Henry Jones	person
1	Sophie Jones	person
2	William Smith	person
3	Acme Aerospace	organization
"""

    relations_tsv = """source_id	link	target_id
0	father_of	1
1	child_of	0
1	works_at	3
2	works_at	3
1	friend_of	2
2	friend_of	1
"""

    graph = build_graph_from_tsv(entities_tsv, relations_tsv)
    print(f"entities={len(graph.entities)} relations={len(graph.relations)}")
    for edge in format_edges(graph):
        print(edge)


if __name__ == "__main__":
    main()
PY
python3 kg_from_tsv.py
```

预期输出：

```text
entities=4 relations=6
Henry Jones --father_of--> Sophie Jones
Sophie Jones --child_of--> Henry Jones
Sophie Jones --works_at--> Acme Aerospace
William Smith --works_at--> Acme Aerospace
Sophie Jones --friend_of--> William Smith
William Smith --friend_of--> Sophie Jones
```

这一轮的目标不是接入模型，而是确认：

- TSV 格式能被稳定解析。
- 实体 ID 没有重复。
- 关系引用的实体都存在。
- 对称关系能表达成双向边。

## 第 4 步：设计 Gemini 抽取提示词

下面是可直接复用的提示词模板：

```text
You are an information extraction engine.
Use only the input text. Do not use external knowledge.

Task:
Extract entities and relationships from the input text.

Entity schema:
- id: stable integer string starting from 0
- name: canonical entity name
- label: one of person, organization, place, system, document, concept, event, animal, other

Relationship schema:
- source_id: id of the source entity
- link: lowercase snake_case relationship type
- target_id: id of the target entity

Rules:
1. Output TSV only. No Markdown, no commentary.
2. Do not include tabs or newlines inside field values.
3. Use only facts explicitly present in the input text.
4. If a relationship is symmetric, output both directions.
5. If a relationship is asymmetric and an obvious inverse is needed for graph traversal, output the inverse relation too.
6. Prefer canonical names. Merge aliases when the input clearly says they refer to the same entity.
7. If there is no valid entity or relation, output only the headers.

Output format:
ENTITIES
id	name	label
...

RELATIONS
source_id	link	target_id
...

Input text:
{{INPUT_TEXT}}
```

为什么要写得这么死：

- `TSV only`：减少模型解释性废话。
- `Use only the input text`：降低幻觉。
- `No tabs or newlines inside field values`：保护解析器。
- `symmetric output both directions`：避免朋友关系只出现一条边。
- `canonical names`：减少别名导致的重复节点。

## 第 5 步：接入 Gemini API

把下面文件保存为 `gemini_extract.py`：

```python
from __future__ import annotations

import os
from textwrap import dedent

from google import genai
from google.genai.types import GenerateContentConfig


PROMPT_TEMPLATE = """
You are an information extraction engine.
Use only the input text. Do not use external knowledge.

Task:
Extract entities and relationships from the input text.

Entity schema:
- id: stable integer string starting from 0
- name: canonical entity name
- label: one of person, organization, place, system, document, concept, event, animal, other

Relationship schema:
- source_id: id of the source entity
- link: lowercase snake_case relationship type
- target_id: id of the target entity

Rules:
1. Output TSV only. No Markdown, no commentary.
2. Do not include tabs or newlines inside field values.
3. Use only facts explicitly present in the input text.
4. If a relationship is symmetric, output both directions.
5. If a relationship is asymmetric and an obvious inverse is needed for graph traversal, output the inverse relation too.
6. Prefer canonical names. Merge aliases when the input clearly says they refer to the same entity.
7. If there is no valid entity or relation, output only the headers.

Output format:
ENTITIES
id\tname\tlabel
...

RELATIONS
source_id\tlink\ttarget_id
...

Input text:
{input_text}
"""


def extract_with_gemini(input_text: str, model: str = "gemini-3.1-flash-lite") -> str:
    if not os.environ.get("GOOGLE_API_KEY"):
        raise RuntimeError("GOOGLE_API_KEY is not set")

    client = genai.Client()
    prompt = PROMPT_TEMPLATE.format(input_text=input_text)
    response = client.models.generate_content(
        model=model,
        contents=prompt,
        config=GenerateContentConfig(
            temperature=0.0,
            top_p=0.0,
            seed=42,
        ),
    )
    return response.text or ""


def main() -> None:
    sample_text = dedent(
        """
        Henry Jones is a famous archaeologist.
        Sophie is Henry's daughter and works as a software engineer.
        William Smith is Sophie's lifelong friend.
        Sophie and William both work at Acme Aerospace.
        """
    ).strip()
    print(extract_with_gemini(sample_text))


if __name__ == "__main__":
    main()
```

运行：

```bash
export GOOGLE_API_KEY="你的 Gemini API Key"
python3 gemini_extract.py
```

预期输出形态：

```text
ENTITIES
id	name	label
0	Henry Jones	person
1	Sophie	person
2	William Smith	person
3	Acme Aerospace	organization

RELATIONS
source_id	link	target_id
0	father_of	1
1	child_of	0
1	friend_of	2
2	friend_of	1
1	works_at	3
2	works_at	3
```

注意：具体实体名可能有轻微差异，比如 `Sophie` 是否补成 `Sophie Jones`，取决于输入是否明确提供姓氏。生产环境不要依赖模型自动补全。

## 第 6 步：把 Gemini 输出拆成两个 TSV 表

Gemini 返回的是一个文本块，需要切出 `ENTITIES` 和 `RELATIONS` 两段。

```python
from __future__ import annotations


def split_gemini_output(output: str) -> tuple[str, str]:
    if "ENTITIES" not in output or "RELATIONS" not in output:
        raise ValueError("Gemini output must contain ENTITIES and RELATIONS sections")

    before_relations, relations_part = output.split("RELATIONS", maxsplit=1)
    _, entities_part = before_relations.split("ENTITIES", maxsplit=1)

    entities_tsv = entities_part.strip()
    relations_tsv = relations_part.strip()
    return entities_tsv, relations_tsv
```

接到第 3 步的解析器：

```python
raw_output = extract_with_gemini(sample_text)
entities_tsv, relations_tsv = split_gemini_output(raw_output)
graph = build_graph_from_tsv(entities_tsv, relations_tsv)
```

## 第 7 步：加入 3 个实操用例

### 用例 A：人物关系图谱

输入：

```text
Henry Jones is a famous archaeologist.
Sophie is Henry's daughter and works as a software engineer.
William Smith is Sophie's lifelong friend.
Sophie and William both work at Acme Aerospace.
```

验收标准：

- 至少抽到 4 个实体：Henry、Sophie、William、Acme Aerospace。
- 至少抽到 4 类关系：父女、子女、朋友、工作单位。
- `friend_of` 应该双向出现。

### 用例 B：运维故障报告图谱

输入：

```text
At 10:03, checkout-api returned HTTP 500 errors.
The root cause was a Redis connection timeout.
SRE Alice restarted the checkout-api deployment.
The incident affected the payment-service and order-service.
```

期望关系：

```text
checkout-api --returned--> HTTP 500 errors
Redis connection timeout --root_cause_of--> checkout-api incident
Alice --restarted--> checkout-api deployment
checkout-api incident --affected--> payment-service
checkout-api incident --affected--> order-service
```

验收标准：

- 系统、事件、人、错误类型不要混成同一个 label。
- `root_cause_of` 这类关系必须方向清晰。
- 后续可以扩展为故障复盘知识库。

### 用例 C：合同义务图谱

输入：

```text
Vendor must deliver the monthly security report to Customer by the fifth business day of each month.
Customer must pay the invoice within 30 days after receiving it.
Either party may terminate the agreement with 60 days written notice.
```

期望关系：

```text
Vendor --must_deliver--> monthly security report
monthly security report --delivered_to--> Customer
Customer --must_pay--> invoice
Either party --may_terminate--> agreement
termination --requires_notice_period--> 60 days
```

验收标准：

- 主体、义务、对象、时间约束要分开。
- 不要把“每月第五个工作日”和“30 天付款期限”丢掉。
- 高风险合同场景必须人工复核，不允许直接作为法律结论。

## 第 8 步：可选，用 NetworkX 建图

安装：

```bash
pip install "networkx[default]"
```

示例：

```python
from __future__ import annotations

import networkx as nx


def to_networkx(graph: KnowledgeGraph) -> nx.DiGraph:
    result = nx.DiGraph()
    for entity in graph.entities:
        result.add_node(entity.id, name=entity.name, label=entity.label)
    for relation in graph.relations:
        result.add_edge(relation.source_id, relation.target_id, label=relation.link)
    return result
```

基础查询：

```python
nx_graph = to_networkx(graph)
print(nx_graph.number_of_nodes())
print(nx_graph.number_of_edges())
print(list(nx_graph.out_edges("1", data=True)))
```

你可以先不画图。对工程落地来说，先确认节点和边正确，比先做漂亮可视化更重要。

## 第 9 步：生产化前必须补的校验

LLM 输出不能直接入库，至少加这些 gate：

| Gate | 检查内容 | 失败处理 |
|---|---|---|
| 格式校验 | 是否包含 `ENTITIES` / `RELATIONS`，TSV 列数是否正确 | 重试一次；仍失败则进入人工队列 |
| ID 校验 | 实体 ID 是否唯一，关系引用是否存在 | 拒绝入库 |
| label 白名单 | label 是否属于允许集合 | 映射到 `other` 或拒绝 |
| relation 白名单 | link 是否属于允许集合 | 进入人工确认或归一化 |
| 规模校验 | 节点/边数量是否异常 | 标记为异常样本 |
| 抽样复核 | 随机抽样比对原文 | 记录准确率和常见错误 |

推荐把模型输出保存下来：

```text
runs/
  inputs/2026-xx-xx-sample.txt
  outputs/2026-xx-xx-sample.tsv
  reviews/2026-xx-xx-sample-review.md
```

这样后续提示词改动、模型升级、成本变化都能回放对比。

## 第 10 步：常见失败与修复

### 失败 1：模型输出了 Markdown 表格

现象：

```text
| id | name | label |
|---|---|---|
```

修复：

- 提示词开头加：`Output TSV only. Do not output Markdown tables.`
- 解析前检测 `|---|`，发现就拒绝，不要硬解析。

### 失败 2：字段里出现制表符或换行

修复：

- 提示词明确：`Do not include tabs or newlines inside field values.`
- 入库前对字段做二次检查。
- 发现非法字符时拒绝或重试。

### 失败 3：同一个实体被抽成多个节点

例子：

```text
William Smith
Bill
```

修复：

- 在提示词里要求：`Merge aliases when the input clearly says they refer to the same entity.`
- 后处理阶段按别名表、精确规则或人工确认合并。

### 失败 4：关系方向反了

修复：

- 对关键关系维护方向定义表。
- 例如：`root_cause_of` 必须从根因指向事件，`affected` 必须从事件指向受影响对象。
- 校验时发现不符合方向约定就进入人工复核。

## 一个一周落地计划

| 天数 | 目标 | 产物 |
|---|---|---|
| Day 1 | 选定一个低风险领域，如故障报告或会议纪要 | 20 条样本文本 |
| Day 2 | 跑通 mock 解析器和 TSV 校验 | `kg_from_tsv.py` |
| Day 3 | 接入 Gemini，保存原文和输出 | `runs/inputs` / `runs/outputs` |
| Day 4 | 建立 label/link 白名单 | `schema.yaml` 或 Python 常量 |
| Day 5 | 人工复核 20 条样本 | 错误清单 |
| Day 6 | 修提示词和校验器 | v2 prompt + validator |
| Day 7 | 决定是否接入 NetworkX/Neo4j/SQLite | PoC 报告 |

## 最小验收清单

- [ ] 不带 API key 的 mock 版本能跑通。
- [ ] Gemini 输出能稳定分成实体表和关系表。
- [ ] 解析器能拒绝坏 TSV。
- [ ] 关系引用不存在实体时会失败。
- [ ] 关键关系有方向定义。
- [ ] 至少 20 条样本经过人工复核。
- [ ] 入库前保留原文、提示词版本、模型名、输出和校验结果。

## 结论

这篇文章最值得复用的不是“用 Gemini 生成知识图谱”这句话，而是三个工程判断：

1. 把 LLM 当抽取器，不当事实裁判。
2. 对实体/关系这种表结构输出，TSV 往往比 JSON 更省、更快。
3. 知识图谱落地的关键不是可视化，而是 schema、校验、复核和可回放。

如果只做 PoC，按本文流程一天内能跑通；如果要进生产，先用低风险数据做一周闭环，不要直接把模型输出写入正式图数据库。