🧠 Graph Pipeline Phase 3: 지능형 매핑 및 검증 (Intelligent Mapping & Validation)

이 문서는 P&ID Graph Pipeline의 세 번째 단계인 지능형 매핑 및 검증의 상세 구현 계획을 다룹니다. 2단계에서 구축한 위상 그래프(Topology Graph)를 활용하여, 도면 상의 가상 노드들을 실제 Experion 시스템의 **실시간 태그(Real-time Tags)**와 정밀하게 연결하고 그 타당성을 검증하는 것이 목표입니다.

📦 1. 필수 패키지 및 환경 설정

1.1 Python 패키지

패키지	용도	비고
`openai` / `langchain`	LLM API 연동 및 프롬프트 체이닝	매핑 추론 및 검증 핵심
`fuzzywuzzy` / `rapidfuzz`	태그 이름 간의 문자열 유사도 계산	1차 후보군 추출용
`networkx`	그래프 기반 인접 노드(Context) 추출	2단계 그래프 활용
`pydantic`	매핑 결과의 구조화 및 유효성 검사	데이터 정규화
`requests`	ExperionCrawler API (C#)와 통신	실제 태그 리스트 조회

1.2 설치 명령어

pip install openai langchain rapidfuzz networkx pydantic requests

📐 2. 상세 설계 구조

2.1 매핑 파이프라인 (Mapping Pipeline)

단순 이름 매칭의 한계를 극복하기 위해 **[후보 추출 \rightarrow 맥락 분석 \rightarrow LLM 확정]**의 3단계 프로세스를 거칩니다.

1차 후보 추출 (Candidate Generation):
- 도면의 태그 텍스트와 Experion 시스템의 전체 태그 리스트를 RapidFuzz로 비교하여 유사도 상위 N개를 추출합니다.
맥락 정보 수집 (Context Gathering):
- 해당 노드의 그래프 상 인접 노드(1-hop, 2-hop) 정보를 수집합니다.
- 예: "현재 노드는 PT-101이며, 상류에 P-101(Pump)이 있고 하류에 V-101(Valve)이 있음."
LLM 기반 최종 매핑 (LLM-based Resolution):
- 후보 태그 리스트와 위상 맥락을 LLM에게 전달하여 가장 타당한 태그를 선택하게 합니다.

2.2 상호 검증 로직 (Cross-Validation)

매핑된 결과가 실제 공정 데이터와 일치하는지 검증합니다.

위상적 일관성: 도면에서 A $\rightarrow$ B 순서라면, 실제 데이터에서도 A의 변화가 B에 영향을 주는지 상관관계 분석.
속성 일치성: 도면의 심볼 타입(예: Pressure Transmitter)과 실제 태그의 속성(예: Engineering Unit = 'bar' 또는 'psi')이 일치하는지 확인.

💻 3. 실제 구현 코딩 가이드 (Example)

3.1 맥락 기반 매핑 엔진

import networkx as nx
import asyncio
from rapidfuzz import process, fuzz
from openai import AsyncOpenAI # 비동기 클라이언트로 변경

client = AsyncOpenAI(api_key="your-api-key")

class IntelligentMapper:
    def __init__(self, graph, system_tags):
        self.graph = graph          # Phase 2에서 생성된 NetworkX 그래프
        self.system_tags = system_tags # Experion 시스템의 전체 태그 리스트

    def get_node_context(self, node_id):
        """노드의 주변 위상 정보를 텍스트로 변환"""
        neighbors = list(self.graph.neighbors(node_id))
        context = []
        for n in neighbors:
            attr = self.graph.nodes[n]
            context.append(f"Connected to {attr.get('value', n)} (Type: {attr.get('type')})")
        return ", ".join(context)

    async def _resolve_generic(self, node_id, category_prompt):
        """공통 매핑 로직 (비동기)"""
        tag_text = self.graph.nodes[node_id].get('value', '')
        candidates = process.extract(tag_text, self.system_tags, scorer=fuzz.WRatio, limit=5)
        context = self.get_node_context(node_id)
        
        prompt = f"""
        {category_prompt}
        P&ID 도면의 태그 '{tag_text}'를 실제 시스템 태그와 매핑해야 합니다.
        위상 맥락: {context}
        후보 리스트: {candidates}
        
        위 맥락을 고려할 때 가장 적절한 시스템 태그 하나만 반환하세요.
        이유가 불분명하면 'UNKNOWN'을 반환하세요.
        """
        
        response = await client.chat.completions.create(
            model="gpt-4-turbo",
            messages=[{"role": "user", "content": prompt}]
        )
        return response.choices[0].message.content

    # --- 전문화된 Worker 함수들 (Phase 5 병렬 처리 반영) ---
    
    async def extract_transmitters(self, node_ids):
        """전송기(Transmitter) 전문 매핑 Worker"""
        prompt = "당신은 계측기 전문 엔지니어입니다. 특히 Pressure/Flow/Level Transmitter 매핑에 특화되어 있습니다."
        return {nid: await self._resolve_generic(nid, prompt) for nid in node_ids}

    async def extract_valves(self, node_ids):
        """밸브(Valve) 전문 매핑 Worker"""
        prompt = "당신은 밸브 및 액추에이터 전문 엔지니어입니다. 밸브의 개폐 상태 및 제어 태그 매핑에 특화되어 있습니다."
        return {nid: await self._resolve_generic(nid, prompt) for nid in node_ids}

    async def extract_equipment(self, node_ids):
        """주요 설비(Pump, Tank 등) 전문 매핑 Worker"""
        prompt = "당신은 공정 설비 전문 엔지니어입니다. 펌프, 탱크, 열교환기 등의 메인 설비 태그 매핑에 특화되어 있습니다."
        return {nid: await self._resolve_generic(nid, prompt) for nid in node_ids}

# 사용 예시 (Phase 5 Orchestrator 관점)
async def main():
    mapper = IntelligentMapper(graph, ["FIC-101.PV", "PT-101.PV", "P-101.STATUS"])
    
    # 분류별로 노드 그룹화 (예시)
    transmitter_nodes = ["node_1", "node_2"]
    valve_nodes = ["node_3", "node_4"]
    equipment_nodes = ["node_5"]

    # asyncio.gather를 통한 병렬 호출
    results = await asyncio.gather(
        mapper.extract_transmitters(transmitter_nodes),
        mapper.extract_valves(valve_nodes),
        mapper.extract_equipment(equipment_nodes)
    )
    
    # 결과 통합 (flatten)
    final_mapping = {**results[0], **results[1], **results[2]}
    print(f"Parallel Resolved Mapping: {final_mapping}")

asyncio.run(main())

3.2 검증 유틸리티: 속성 일치 확인

def validate_mapping(resolved_tag, symbol_type, tag_metadata):
    """심볼 타입과 실제 태그 메타데이터의 일치 여부 검증"""
    type_map = {
        "Pressure Transmitter": ["pressure", "bar", "psi", "pa"],
        "Flow Meter": ["flow", "m3/h", "lpm"],
        "Temperature Sensor": ["temp", "celsius", "k"]
    }
    
    expected_keywords = type_map.get(symbol_type, [])
    actual_desc = tag_metadata.get('description', '').lower()
    
    # 메타데이터 설명에 기대 키워드가 포함되어 있는지 확인
    is_valid = any(kw in actual_desc for kw in expected_keywords)
    return is_valid

🚀 4. Phase 3 완료 기준 (Definition of Done)

모든 도면 노드에 대해 **1차 후보군(Candidates)**이 자동으로 생성되는가?
NetworkX 그래프를 통해 **인접 노드 맥락(Context)**이 정확히 추출되는가?
LLM이 맥락을 반영하여 최종 태그를 결정하고 그 근거를 제시하는가?
매핑된 태그의 **메타데이터(Unit, Description)**와 도면 심볼 타입 간의 일치성이 검증되는가?
최종 매핑 결과가 (도면노드ID, 시스템태그, 신뢰도, 검증결과) 형태로 저장되는가?

7.4 KiB Raw Blame History