온톨로지부터 지식 그래프 RAG까지: RDF·OWL·SPARQL로 작은 운영 문서 그래프 만들기

1@prefix ex: <https://example.org/ops#> .
2@prefix owl: <http://www.w3.org/2002/07/owl#> .
3@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#> .
4
5ex:Service a owl:Class .
6ex:Incident a owl:Class .
7ex:Runbook a owl:Class .
8
9ex:runbookOf a owl:ObjectProperty ;
10    rdfs:domain ex:Runbook ;
11    rdfs:range ex:Service ;
12    owl:inverseOf ex:hasRunbook .
13ex:hasRunbook a owl:ObjectProperty .
14
15ex:addresses a owl:ObjectProperty ;
16    rdfs:domain ex:Runbook ;
17    rdfs:range ex:Incident .
18ex:title a owl:DatatypeProperty .
19ex:sourceId a owl:DatatypeProperty .
20ex:instruction a owl:DatatypeProperty .
21
22ex:paymentApi a ex:Service ;
23    ex:title "결제 API"@ko .
24
25ex:paymentTimeout a ex:Incident ;
26    ex:title "결제 타임아웃"@ko .
27
28ex:paymentRunbook a ex:Runbook ;
29    ex:title "결제 오류 대응 절차"@ko ;
30    ex:runbookOf ex:paymentApi ;
31    ex:addresses ex:paymentTimeout ;
32    ex:sourceId "RUNBOOK-001" ;
33    ex:instruction "타임아웃 비율과 결제사 응답 지연을 확인하고, 중복 결제 여부를 대조한다."@ko .

1from rdflib import Graph, Namespace
2
3EX = Namespace("https://example.org/ops#")
4graph = Graph()
5graph.parse("ontology.ttl", format="turtle")
6
7query = """
8PREFIX ex: <https://example.org/ops#>
9SELECT ?runbook ?title ?incidentName ?sourceId ?instruction
10WHERE {
11  ?service ^ex:runbookOf ?runbook .
12  ?runbook a ex:Runbook ;
13           ex:title ?title ;
14           ex:addresses ?incident ;
15           ex:sourceId ?sourceId ;
16           ex:instruction ?instruction .
17  ?incident ex:title ?incidentName .
18}
19"""
20
21rows = list(graph.query(query, initBindings={"service": EX.paymentApi}))
22for row in rows:
23    print(f"문서: {row.title} [{row.sourceId}]")
24    print(f"대상 장애: {row.incidentName}")
25    print(f"절차: {row.instruction}")
26
27direct = graph.query("""
28PREFIX ex: <https://example.org/ops#>
29ASK { ex:paymentApi ex:hasRunbook ex:paymentRunbook }
30""").askAnswer
31path = graph.query("""
32PREFIX ex: <https://example.org/ops#>
33ASK { ex:paymentApi ^ex:runbookOf ex:paymentRunbook }
34""").askAnswer
35print(f"직접 저장된 hasRunbook: {direct}")
36print(f"역방향 경로로 찾기: {path}")

1@prefix ex: <https://example.org/ops#> .
2@prefix sh: <http://www.w3.org/ns/shacl#> .
3
4ex:RunbookShape a sh:NodeShape ;
5    sh:targetClass ex:Runbook ;
6    sh:property [
7        sh:path ex:sourceId ;
8        sh:minCount 1 ;
9        sh:maxCount 1
10    ] ;
11    sh:property [
12        sh:path ex:runbookOf ;
13        sh:minCount 1 ;
14        sh:class ex:Service
15    ] .

1{
2  "question": "결제 API 타임아웃이 나면 무엇부터 확인하나요?",
3  "evidence": [
4    {
5      "sourceId": "RUNBOOK-001",
6      "service": "결제 API",
7      "incident": "결제 타임아웃",
8      "instruction": "타임아웃 비율과 결제사 응답 지연을 확인하고, 중복 결제 여부를 대조한다."
9    }
10  ]
11}

1type Evidence = {
2  sourceId: string;
3  incident: string;
4  instruction: string;
5};
6
7async function answer(question: string, serviceId: string, userId: string) {
8  const serviceIri = allowedServices.get(serviceId);
9  if (!serviceIri) return { answer: "알 수 없는 서비스입니다.", sources: [] };
10
11  const rows: Evidence[] = await findRunbooks(serviceIri); // 고정 SPARQL
12  const evidence = await filterAsync(rows, row => canRead(userId, row.sourceId));
13  if (evidence.length === 0) {
14    return { answer: "확인된 근거가 없습니다.", sources: [] };
15  }
16
17  const result = await model.generate({
18    system: "근거에 있는 사실만 답하고, 사용한 sourceId를 별도 배열로 반환하세요.",
19    user: JSON.stringify({ question, evidence }),
20  });
21
22  const allowedIds = new Set(evidence.map(row => row.sourceId));
23  if (result.sourceIds.length === 0 ||
24      result.sourceIds.some((id: string) => !allowedIds.has(id))) {
25    return { answer: "답변의 출처를 검증할 수 없습니다.", sources: [] };
26  }
27  return { answer: result.answer, sources: result.sourceIds };
28}

1질문 수신
2  → 허용된 serviceId인지 확인
3  → 권한 범위 안에서 고정 SPARQL 조회
4  → 결과 0건이면 "확인된 근거가 없습니다" 반환
5  → 결과를 모델 문맥에 전달
6  → 모델이 반환한 sourceId가 조회 결과에 있는지 검사
7  → 검증된 문서 링크와 함께 답변 반환