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

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

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

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

id	name	label
0	Henry Jones	person
1	Sophie Jones	person

source_id	link	target_id
0	father_of	1
1	child_of	0

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

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

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}}

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()

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

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

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

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)

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.

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.

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

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.

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

pip install "networkx[default]"

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

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)))

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

| id | name | label |
|---|---|---|

William Smith
Bill