유틸리티 및 통신 분야를 위한 LLM 엔지니어링
요약
유틸리티 및 통신 분야의 현장 보고서를 구조화된 JSON 데이터로 변환하는 경량 분류 에이전트 구축 방법을 소개합니다. Oxlo.ai를 활용하여 긴 사고 로그를 비용 효율적으로 처리하고 Pydantic으로 데이터 검증을 수행하는 과정을 다룹니다.
핵심 포인트
- 비정형 현장 보고서를 심각도 및 안전 플래그가 포함된 JSON으로 변환
- Oxlo.ai의 요청 단위 과금 방식을 통한 긴 컨텍스트 처리 비용 절감
- Pydantic을 사용한 엄격한 출력 스키마 정의 및 데이터 검증
- 데이터 정규화 과정을 통한 LLM 입력 품질 향상
유틸리티(Utilities) 및 통신(Telecom) 분야의 현장 운영 센터는 여전히 가공되지 않은 정전 보고서를 수동으로 배차 티켓(dispatch tickets)으로 변환하고 있습니다. 저는 알람 피드(alarm feeds), 작업 팀 노트(crew notes), 고객 불만 사항을 읽고 심각도 점수(severity scores), 안전 플래그(safety flags), 작업 팀 권장 사항(crew recommendations)이 포함된 검증된 JSON을 반환하는 경량 분류 에이전트(triage agent)를 구축했습니다. 이 에이전트는 Oxlo.ai에서 실행되며, 요청당 고정 가격(flat per-request pricing) 방식 덕분에 긴 사고 로그(incident log)를 처리하는 비용이 짧은 핑(ping)을 보내는 비용과 동일합니다.
필요한 사항
- Python 3.10+
- OpenAI SDK:
pip install openai - 검증을 위한 Pydantic:
pip install pydantic - https://portal.oxlo.ai에서 발급받은 Oxlo.ai API 키
Oxlo.ai는 토큰(token) 단위가 아닌 요청(request) 단위로 과금되므로, 모델에 10페이지 분량의 사고 이력을 입력하더라도 비용이 상승하지 않습니다. 현재 요금제에 대한 자세한 내용은 https://oxlo.ai/pricing을 참조하세요.
1단계: Oxlo.ai 엔드포인트 연결 및 테스트
실제 피드(feeds)를 입력하기 전에, SDK가 Oxlo.ai에 도달하고 올바르게 인증되는지 확인하십시오. 이 코드 스니펫(snippet)은 이후의 모든 단계에서 재사용하게 될 정확한 클라이언트 패턴을 사용합니다.
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
...
2단계: 분류 스키마(triage schema) 정의
에이전트는 하위 배차 시스템(downstream dispatch systems)이 소비할 수 있도록 산문(prose)이 아닌 구조화된 데이터(structured data)를 출력해야 합니다. 저는 타입(types)을 강제하고 시스템 프롬프트(system prompt)에서 LLM에게 명확한 목표를 제공하기 위해 Pydantic을 사용합니다.
from pydantic import BaseModel, Field
from typing import Literal, List
...
3단계: 시스템 프롬프트 작성
시스템 프롬프트는 모델에게 스키마(schema)와 도메인 규칙(domain rules)을 가르치는 유일한 장소입니다. 엄격하게 유지하십시오: 유효한 JSON을 요구하고, 모든 열거형(enum)을 정의하며, 마크다운 래퍼(markdown wrappers)를 금지하십시오.
SYSTEM_PROMPT = """당신은 유틸리티 및 통신 정전 분류 엔지니어입니다.
당신의 업무는 가공되지 않은 현장 보고서, SCADA 알람, 고객 불만 사항을 읽고 구조화된 분류 평가를 생성하는 것입니다.
...
4단계: 유입되는 보고서 정규화
4단계: 유입되는 보고서 정규화
실제 피드(feeds)는 일관되지 않은 간격과 혼합된 소스를 가진 다중 행 블롭(multiline blobs) 형태로 도착합니다. 가벼운 정규화 도구(normalizer)가 빈 줄을 제거하고 일관된 헤더를 추가하여 LLM이 균일한 문서를 볼 수 있도록 합니다.
def normalize_report(raw: str) -> str:
lines = raw.strip().splitlines()
cleaned = [line.strip() for line in lines if line.strip()]
...
Step 5: LLM 호출 및 구조화된 출력 파싱
여기에서 정규화된 보고서를 Oxlo.ai로 라우팅하고 응답을 우리의 Pydantic 모델로 강제 변환(coerce)합니다. 작은 새니타이저(sanitizer)가 파싱 전에 실수로 포함된 마크다운 펜스(markdown fences)를 제거합니다.
import json
from pydantic import ValidationError
...
```"):
content = content.split("```\n\njson")[-1].split("\n\n```")[0].strip()
data = json.loads(content)
return OutageTriage(**data)
Step 6: 배차 규칙 추가
...
def dispatch_recommendation(triage: OutageTriage) -> dict:
sla_minutes = {
"low": 1440,
"medium": 240,
"high": 90,
"critical": 30,
}
return {
"dispatch_within_minutes": sla_minutes[triage.severity],
"crew": triage.required_crew_type,
"assets": triage.affected_assets,
"notes": triage.dispatch_notes,
"safety_flag": len(triage.safety_hazards) > 0,
}
실행하기
...
if __name__ == "__main__":
raw = """
Substation 7 alarm at 14:03. Breaker BR-702 tripped.
Field crew reports loud bang from transformer T-7A.
1200 homes out in sector 4.
Tree branch on primary line near Maple St.
Weather clear. No injuries reported but wire is down across road.
"""
result = triage_outage(raw)
print(result.model_dump_json(indent=2))
dispatch = dispatch_recommendation(result)
print(json.dumps(dispatch, indent=2))
출력 예시:
{
"severity": "critical",
"probable_cause": "식생 (vegetation)",
"affected_assets": [
"Substation 7 (변전소 7)",
"Breaker BR-702 (차단기 BR-702)",
"Transformer T-7A (변압기 T-7A)",
"Maple St 인근 주선 (Primary line)"
],
"estimated_customers_impacted": 1200,
"required_crew_type": "lineman (선로 작업자)",
"safety_hazards": [
"도로를 가로지르는 단선 (downed line)"
],
"dispatch_notes": "Substation 7으로 선로 작업자 팀을 즉시 파견하십시오. 단선된 주선이 공공 안전 위험을 초래하고 있습니다."
}
{
"dispatch_within_minutes": 30,
"crew": "lineman (선로 작업자)",
"assets": [
"Substation 7 (변전소 7)",
"Breaker BR-702 (차단기 BR-702)",
"Transformer T-7A (변압기 T-7A)",
"Maple St 인근 주선 (Primary line)"
],
"notes": "Substation 7으로 선로 작업자 팀을 즉시 파견하십시오. 단선된 주선이 공공 안전 위험을 초래하고 있습니다.",
"safety_flag": true
}
## Wrap-up (마무리)
이 에이전트는 비정형화된 정전 노이즈(unstructured outage noise)를 구조화된 파견 데이터(structured dispatch data)로 변환합니다. 구체적인 다음 단계 두 가지는 다음과 같습니다: 이를 FastAPI 서비스로 배포하고 SCADA 또는 NOC 웹훅(webhook)을 엔드포인트로 지정하거나, `qwen-3-32b`를 사용하는 두 번째 Oxlo.ai 단계를 추가하여 긴 보고서에 대한 토큰 비용 걱정 없이 이중 언어 사용 작업자를 위한 스페인어 파견 노트를 생성하는 것입니다. 두 경로 모두 동일한 OpenAI 호환 클라이언트를 유지하므로, 변경 사항은 모델 문자열뿐입니다.
AI 자동 생성 콘텐츠
본 콘텐츠는 Dev.to AI tag의 원문을 AI가 자동으로 요약·번역·분석한 것입니다. 원 저작권은 원저작자에게 있으며, 정확한 내용은 반드시 원문을 확인해 주세요.
원문 바로가기