gc/kcg-monitoring
7
0
포크 0
코드 이슈 풀 리퀘스트 1 Actions 위키 활동

feat(chat): Ollama Qwen3 기반 AI 해양분석 채팅 구축

소스 검색 
- Ollama Docker(14b/32b) + Redis 컨텍스트 캐싱 + 대화 히스토리
- Python SSE 채팅 엔드포인트 + 사전 쿼리 + Tool Calling
- 도메인 지식(해양법/어업협정/알고리즘) + DB 스키마 가이드
- Frontend SSE 스트리밍 + 타이머 + thinking 접기 + 확장 UI
This commit is contained in:
htlee 2026-03-26 09:03:05 +09:00
부모 b0bb0fe33d
커밋 e797beaac6
17개의 변경된 파일과 1791개의 추가작업 그리고 124개의 파일을 삭제
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
deploy/docker-compose-ollama.yml Normal file
파일 보기

@ -0,0 +1,21 @@
services:
ollama:
image: ollama/ollama:latest
container_name: kcg-ollama
restart: unless-stopped
ports:
- "11434:11434"
volumes:
- /home/kcg-ollama/data:/root/.ollama
deploy:
resources:
limits:
memory: 64G
reservations:
memory: 40G
environment:
- OLLAMA_NUM_PARALLEL=4
- OLLAMA_MAX_LOADED_MODELS=1
- OLLAMA_KEEP_ALIVE=24h
- OLLAMA_FLASH_ATTENTION=1
- OLLAMA_NUM_THREADS=48

15
deploy/nginx-kcg.conf
파일 보기

@ -20,6 +20,21 @@ server {
try_files $uri $uri/ /index.html;
}
# ── AI Chat (SSE → Python prediction on redis-211) ──
location /api/prediction-chat {
rewrite ^/api/prediction-chat(.*)$ /api/v1/chat$1 break;
proxy_pass http://192.168.1.18:8001;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_buffering off;
proxy_cache off;
proxy_read_timeout 120s;
proxy_set_header Connection '';
chunked_transfer_encoding off;
}
# ── Backend API (direct) ──
location /api/ {
proxy_pass http://127.0.0.1:8080/api/;

7
frontend/src/components/common/EventLog.tsx
파일 보기

@ -885,12 +885,7 @@ export function EventLog({ events, currentTime, totalShipCount: _totalShipCount,
{/* AI 해양분석 챗 — 한국 탭 전용 */}
{dashboardTab === 'korea' && (
<AiChatPanel
ships={ships}
koreanShipCount={koreanShips.length}
chineseShipCount={chineseShips.length}
totalShipCount={ships.length}
/>
<AiChatPanel />
)}
</div>
);

348
frontend/src/components/korea/AiChatPanel.tsx
파일 보기

@ -1,73 +1,76 @@
import { useState, useRef, useEffect, useCallback } from 'react';
import type { Ship } from '../../types';
import { useAuth } from '../../hooks/useAuth';
interface ChatMessage {
role: 'user' | 'assistant' | 'system';
role: 'user' | 'assistant';
content: string;
timestamp: number;
isStreaming?: boolean;
}
interface Props {
ships: Ship[];
koreanShipCount: number;
chineseShipCount: number;
totalShipCount: number;
const AI_CHAT_URL = '/api/prediction-chat';
/** assistant 메시지에서 thinking(JSON tool call, 구분선 등)과 답변을 분리 */
function splitThinking(content: string): { thinking: string; answer: string } {
// 패턴: ```json...``` 블록 + ---\n_데이터 조회 완료..._\n\n 까지가 thinking
const thinkingPattern = /^([\s\S]*?```json[\s\S]*?```[\s\S]*?---\n_[^_]*_\n*)/;
const match = content.match(thinkingPattern);
if (match) {
return { thinking: match[1].trim(), answer: content.slice(match[0].length).trim() };
}
// ```json 블록만 있고 답변이 아직 안 온 경우 (스트리밍 중)
const jsonOnly = /^([\s\S]*```json[\s\S]*?```[\s\S]*)$/;
const m2 = content.match(jsonOnly);
if (m2 && !content.includes('---')) {
return { thinking: m2[1].trim(), answer: '' };
}
return { thinking: '', answer: content };
}
// TODO: Python FastAPI 기반 해양분석 AI API로 전환 예정
const AI_CHAT_URL = '/api/kcg/ai/chat';
export function AiChatPanel() {
const { user } = useAuth();
const userId = user?.email ?? 'anonymous';
function buildSystemPrompt(props: Props): string {
const { ships, koreanShipCount, chineseShipCount, totalShipCount } = props;
// 선박 유형별 통계
const byType: Record<string, number> = {};
const byFlag: Record<string, number> = {};
ships.forEach(s => {
byType[s.category || 'unknown'] = (byType[s.category || 'unknown'] || 0) + 1;
byFlag[s.flag || 'unknown'] = (byFlag[s.flag || 'unknown'] || 0) + 1;
});
// 중국 어선 통계
const cnFishing = ships.filter(s => s.flag === 'CN' && (s.category === 'fishing' || s.typecode === '30'));
const cnFishingOperating = cnFishing.filter(s => s.speed >= 2 && s.speed <= 5);
return `당신은 대한민국 해양경찰청의 해양상황 분석 AI 어시스턴트입니다.
현재 실시간 해양 모니터링 데이터를 기반으로 분석을 제공합니다.
## 현재 해양 상황 요약
- 전체 선박: ${totalShipCount}척
- 한국 선박: ${koreanShipCount}척
- 중국 선박: ${chineseShipCount}척
- 중국 어선: ${cnFishing.length}척 (조업 추정: ${cnFishingOperating.length}척)
## 선박 유형별 현황
${Object.entries(byType).map(([k, v]) => `- ${k}: ${v}척`).join('\n')}
## 국적별 현황 (상위)
${Object.entries(byFlag).sort((a, b) => b[1] - a[1]).slice(0, 10).map(([k, v]) => `- ${k}: ${v}척`).join('\n')}
## 한중어업협정 핵심
- 중국 허가어선 906척 (PT 저인망 323쌍, GN 유자망 200척, PS 위망 16척, OT 1척식 13척, FC 운반선 31척)
- 특정어업수역 I~IV에서만 조업 허가
- 휴어기: PT/OT 4/16~10/15, GN 6/2~8/31
- 다크베셀(AIS 차단) 감시 필수
## 응답 규칙
- 한국어로 답변
- 간결하고 분석적으로
- 데이터 기반 답변 우선
- 불법조업 의심 시 근거 제시
- 필요시 조치 권고 포함`;
}
export function AiChatPanel({ ships, koreanShipCount, chineseShipCount, totalShipCount }: Props) {
const [isOpen, setIsOpen] = useState(true);
const [messages, setMessages] = useState<ChatMessage[]>([]);
const [input, setInput] = useState('');
const [isLoading, setIsLoading] = useState(false);
const [elapsed, setElapsed] = useState(0);
const timerRef = useRef<ReturnType<typeof setInterval> | null>(null);
const [expanded, setExpanded] = useState(false);
const [historyLoaded, setHistoryLoaded] = useState(false);
const messagesEndRef = useRef<HTMLDivElement>(null);
const inputRef = useRef<HTMLInputElement>(null);
const abortRef = useRef<AbortController | null>(null);
// 마운트 시 Redis에서 대화 히스토리 로드
useEffect(() => {
if (historyLoaded) return;
fetch(`${AI_CHAT_URL}/history?user_id=${encodeURIComponent(userId)}`)
.then(res => res.ok ? res.json() : [])
.then((history: { role: string; content: string }[]) => {
if (history.length > 0) {
setMessages(history.map(m => ({
role: m.role as 'user' | 'assistant',
content: m.content,
timestamp: Date.now(),
})));
}
})
.catch(() => { /* Redis 미연결 시 무시 */ })
.finally(() => setHistoryLoaded(true));
}, [userId, historyLoaded]);
useEffect(() => {
if (isLoading) {
setElapsed(0);
timerRef.current = setInterval(() => setElapsed(s => s + 1), 1000);
} else if (timerRef.current) {
clearInterval(timerRef.current);
timerRef.current = null;
}
return () => { if (timerRef.current) clearInterval(timerRef.current); };
}, [isLoading]);
useEffect(() => {
messagesEndRef.current?.scrollIntoView({ behavior: 'smooth' });
@ -85,72 +88,166 @@ export function AiChatPanel({ ships, koreanShipCount, chineseShipCount, totalShi
setInput('');
setIsLoading(true);
try {
const systemPrompt = buildSystemPrompt({ ships, koreanShipCount, chineseShipCount, totalShipCount });
const apiMessages = [
{ role: 'system', content: systemPrompt },
...messages.filter(m => m.role !== 'system').map(m => ({ role: m.role, content: m.content })),
{ role: 'user', content: userMsg.content },
];
// 스트리밍 placeholder 추가
const streamingMsg: ChatMessage = { role: 'assistant', content: '', timestamp: Date.now(), isStreaming: true };
setMessages(prev => [...prev, streamingMsg]);
const controller = new AbortController();
abortRef.current = controller;
try {
const res = await fetch(AI_CHAT_URL, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
model: 'qwen2.5:7b',
messages: apiMessages,
stream: false,
options: { temperature: 0.3, num_predict: 1024 },
message: userMsg.content,
user_id: userId,
stream: true,
}),
signal: controller.signal,
});
if (!res.ok) throw new Error(`Ollama error: ${res.status}`);
const data = await res.json();
const assistantMsg: ChatMessage = {
role: 'assistant',
content: data.message?.content || '응답을 생성할 수 없습니다.',
timestamp: Date.now(),
if (!res.ok) throw new Error(`서버 오류: ${res.status}`);
if (!res.body) throw new Error('스트리밍 미지원');
const reader = res.body.getReader();
const decoder = new TextDecoder();
let accumulated = '';
let buffer = '';
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
const lines = buffer.split('\n');
buffer = lines.pop() ?? '';
for (const line of lines) {
if (!line.startsWith('data: ')) continue;
const data = line.slice(6);
if (data === '[DONE]') break;
try {
const chunk = JSON.parse(data) as { content: string; done: boolean };
accumulated += chunk.content;
setMessages(prev => {
const updated = [...prev];
updated[updated.length - 1] = {
...updated[updated.length - 1],
content: accumulated,
};
setMessages(prev => [...prev, assistantMsg]);
return updated;
});
if (chunk.done) break;
} catch {
// JSON 파싱 실패 무시
}
}
}
// 스트리밍 완료
setMessages(prev => {
const updated = [...prev];
updated[updated.length - 1] = {
...updated[updated.length - 1],
isStreaming: false,
};
return updated;
});
} catch (err) {
setMessages(prev => [...prev, {
if ((err as Error).name === 'AbortError') return;
setMessages(prev => {
const updated = [...prev];
updated[updated.length - 1] = {
role: 'assistant',
content: `오류: ${err instanceof Error ? err.message : 'AI 서버 연결 실패'}. Ollama가 실행 중인지 확인하세요.`,
content: `오류: ${err instanceof Error ? err.message : 'AI 서버 연결 실패'}`,
timestamp: Date.now(),
}]);
isStreaming: false,
};
return updated;
});
} finally {
setIsLoading(false);
abortRef.current = null;
}
}, [input, isLoading, messages, ships, koreanShipCount, chineseShipCount, totalShipCount]);
}, [input, isLoading, userId]);
const clearHistory = useCallback(async () => {
setMessages([]);
try {
await fetch(`${AI_CHAT_URL}/history?user_id=${encodeURIComponent(userId)}`, { method: 'DELETE' });
} catch { /* 무시 */ }
}, [userId]);
const quickQuestions = [
'현재 해양 상황을 요약해줘',
'중국어선 불법조업 의심 분석해줘',
'서해 위험도를 평가해줘',
'위험 선박 상위 10척 알려줘',
'다크베셀 현황 분석해줘',
];
return (
<div style={{
...(expanded ? {
position: 'fixed' as const,
bottom: 16,
right: 16,
width: 520,
height: 600,
zIndex: 9999,
borderRadius: 8,
boxShadow: '0 8px 32px rgba(0,0,0,0.6)',
display: 'flex',
flexDirection: 'column' as const,
background: 'rgba(12,24,37,0.97)',
border: '1px solid rgba(168,85,247,0.3)',
} : {
borderTop: '1px solid rgba(168,85,247,0.2)',
marginTop: 8,
}),
}}>
{/* Toggle header */}
<div
onClick={() => setIsOpen(p => !p)}
onClick={() => {
if (!isOpen) { setIsOpen(true); return; }
if (expanded) { setExpanded(false); return; }
setExpanded(true);
}}
style={{
display: 'flex', alignItems: 'center', gap: 6,
padding: '6px 8px', cursor: 'pointer',
background: isOpen ? 'rgba(168,85,247,0.12)' : 'rgba(168,85,247,0.04)',
borderRadius: 4,
borderLeft: '2px solid rgba(168,85,247,0.5)',
borderRadius: expanded ? '8px 8px 0 0' : 4,
borderLeft: expanded ? 'none' : '2px solid rgba(168,85,247,0.5)',
flexShrink: 0,
}}
>
<span style={{ fontSize: 12 }}>🤖</span>
<span style={{ fontSize: 11, fontWeight: 700, color: '#c4b5fd' }}>AI 해양분석</span>
<span style={{ fontSize: 8, color: '#8b5cf6', marginLeft: 2, background: 'rgba(139,92,246,0.15)', padding: '1px 5px', borderRadius: 3 }}>Qwen 2.5</span>
<span style={{ marginLeft: 'auto', fontSize: 8, color: '#8b5cf6' }}>
{isOpen ? '▼' : '▶'}
<span style={{ fontSize: 8, color: '#8b5cf6', marginLeft: 2, background: 'rgba(139,92,246,0.15)', padding: '1px 5px', borderRadius: 3 }}>
Qwen3 14B
</span>
<span style={{ marginLeft: 'auto', display: 'flex', gap: 4, alignItems: 'center' }}>
{isOpen && (
<button
onClick={e => { e.stopPropagation(); setIsOpen(false); setExpanded(false); }}
title="접기"
style={{
background: 'none', border: 'none', cursor: 'pointer',
fontSize: 8, color: '#8b5cf6',
padding: '8px 10px', margin: '-8px -8px -8px -6px',
lineHeight: 1,
}}
>
{expanded ? '⊖' : '▼'}
</button>
)}
{!isOpen && (
<span style={{ fontSize: 8, color: '#8b5cf6' }}>▶</span>
)}
</span>
</div>
@ -158,10 +255,11 @@ export function AiChatPanel({ ships, koreanShipCount, chineseShipCount, totalShi
{isOpen && (
<div style={{
display: 'flex', flexDirection: 'column',
height: 360, background: 'rgba(88,28,135,0.08)',
...(expanded ? { flex: 1 } : { height: 360 }),
background: expanded ? 'transparent' : 'rgba(88,28,135,0.08)',
borderRadius: '0 0 6px 6px', overflow: 'hidden',
borderLeft: '2px solid rgba(168,85,247,0.3)',
borderBottom: '1px solid rgba(168,85,247,0.15)',
borderLeft: expanded ? 'none' : '2px solid rgba(168,85,247,0.3)',
borderBottom: expanded ? 'none' : '1px solid rgba(168,85,247,0.15)',
}}>
{/* Messages */}
<div style={{
@ -192,34 +290,79 @@ export function AiChatPanel({ ships, koreanShipCount, chineseShipCount, totalShi
</div>
</div>
)}
{messages.map((msg, i) => (
{messages.map((msg, i) => {
const isAssistant = msg.role === 'assistant';
const { thinking, answer } = isAssistant ? splitThinking(msg.content) : { thinking: '', answer: msg.content };
const displayText = isAssistant ? (answer || (thinking && !msg.isStreaming ? '' : msg.content)) : msg.content;
return (
<div
key={i}
style={{
alignSelf: msg.role === 'user' ? 'flex-end' : 'flex-start',
maxWidth: '85%',
}}
>
{/* thinking 접기 블록 */}
{isAssistant && thinking && (
<details style={{
background: 'rgba(100,116,139,0.1)',
borderRadius: '6px 6px 0 0',
padding: '4px 8px',
fontSize: 9,
color: '#64748b',
cursor: 'pointer',
borderLeft: '2px solid rgba(139,92,246,0.3)',
}}>
<summary style={{ userSelect: 'none', outline: 'none' }}>도구 호출</summary>
<pre style={{
margin: '4px 0 0', padding: '4px',
fontSize: 8, color: '#94a3b8',
whiteSpace: 'pre-wrap', wordBreak: 'break-all',
background: 'rgba(0,0,0,0.2)', borderRadius: 3,
maxHeight: 120, overflowY: 'auto',
}}>{thinking}</pre>
</details>
)}
{/* 메시지 본문 */}
<div style={{
background: msg.role === 'user'
? 'rgba(139,92,246,0.25)'
: 'rgba(168,85,247,0.08)',
borderRadius: msg.role === 'user' ? '8px 8px 2px 8px' : '8px 8px 8px 2px',
borderRadius: thinking
? '0 0 8px 8px'
: (msg.role === 'user' ? '8px 8px 2px 8px' : '8px 8px 8px 2px'),
padding: '6px 8px',
fontSize: 10,
color: '#e2e8f0',
lineHeight: 1.5,
whiteSpace: 'pre-wrap',
wordBreak: 'break-word',
}}
>
{msg.content}
}}>
{displayText}
{msg.isStreaming && msg.content && (
<span style={{ color: '#a78bfa' }}>
<span style={{ animation: 'pulse 1s infinite' }}> ▋</span>
<span style={{ fontSize: 8, color: '#64748b', marginLeft: 4, fontVariantNumeric: 'tabular-nums' }}>
{String(Math.floor(elapsed / 60)).padStart(2, '0')}:{String(elapsed % 60).padStart(2, '0')}
</span>
</span>
)}
</div>
))}
{isLoading && (
</div>
);
})}
{isLoading && !messages[messages.length - 1]?.content && (
<div style={{
alignSelf: 'flex-start', padding: '6px 8px',
background: 'rgba(168,85,247,0.08)', borderRadius: 8,
fontSize: 10, color: '#a78bfa',
display: 'flex', alignItems: 'center', gap: 6,
}}>
분석 중...
<span style={{ animation: 'pulse 1.5s ease-in-out infinite' }}>분석 중</span>
<span style={{ color: '#64748b', fontVariantNumeric: 'tabular-nums' }}>
{String(Math.floor(elapsed / 60)).padStart(2, '0')}:{String(elapsed % 60).padStart(2, '0')}
</span>
</div>
)}
<div ref={messagesEndRef} />
@ -231,11 +374,24 @@ export function AiChatPanel({ ships, koreanShipCount, chineseShipCount, totalShi
borderTop: '1px solid rgba(255,255,255,0.06)',
background: 'rgba(0,0,0,0.15)',
}}>
{messages.length > 0 && (
<button
onClick={clearHistory}
title="대화 초기화"
style={{
background: 'none', border: 'none',
color: '#64748b', fontSize: 12, cursor: 'pointer',
padding: '0 4px', flexShrink: 0,
}}
>
↺
</button>
)}
<input
ref={inputRef}
value={input}
onChange={e => setInput(e.target.value)}
onKeyDown={e => { if (e.key === 'Enter' && !e.shiftKey) { e.preventDefault(); sendMessage(); } }}
onKeyDown={e => { if (e.key === 'Enter' && !e.shiftKey) { e.preventDefault(); void sendMessage(); } }}
placeholder="해양 상황 질문..."
disabled={isLoading}
style={{
@ -246,7 +402,7 @@ export function AiChatPanel({ ships, koreanShipCount, chineseShipCount, totalShi
}}
/>
<button
onClick={sendMessage}
onClick={() => { void sendMessage(); }}
disabled={isLoading || !input.trim()}
style={{
background: isLoading || !input.trim() ? '#334155' : '#7c3aed',

5
frontend/vite.config.ts
파일 보기

@ -110,6 +110,11 @@ export default defineConfig(({ mode }): UserConfig => ({
changeOrigin: true,
secure: false,
},
'/api/prediction-chat': {
target: 'https://kcg.gc-si.dev',
changeOrigin: true,
secure: true,
},
'/ollama': {
target: 'http://localhost:11434',
changeOrigin: true,

0
prediction/chat/__init__.py Normal file
파일 보기

90
prediction/chat/cache.py Normal file
파일 보기

@ -0,0 +1,90 @@
"""Redis 캐시 유틸 — 분석 컨텍스트 + 대화 히스토리."""
import json
import logging
from typing import Optional
import redis
from config import settings
logger = logging.getLogger(__name__)
_redis: Optional[redis.Redis] = None
def _get_redis() -> redis.Redis:
global _redis
if _redis is None:
_redis = redis.Redis(
host=settings.REDIS_HOST,
port=settings.REDIS_PORT,
password=settings.REDIS_PASSWORD or None,
decode_responses=True,
socket_connect_timeout=3,
)
return _redis
# ── 분석 컨텍스트 캐시 (전역, 5분 주기 갱신) ──
CONTEXT_KEY = 'kcg:chat:context'
CONTEXT_TTL = 360 # 6분 (5분 주기 + 1분 버퍼)
def cache_analysis_context(context_dict: dict):
"""스케줄러에서 분석 완료 후 호출 — Redis에 요약 데이터 캐싱."""
try:
r = _get_redis()
r.setex(CONTEXT_KEY, CONTEXT_TTL, json.dumps(context_dict, ensure_ascii=False, default=str))
logger.debug('cached analysis context (%d bytes)', len(json.dumps(context_dict)))
except Exception as e:
logger.warning('failed to cache analysis context: %s', e)
def get_cached_context() -> Optional[dict]:
"""Redis에서 캐시된 분석 컨텍스트 조회."""
try:
r = _get_redis()
data = r.get(CONTEXT_KEY)
return json.loads(data) if data else None
except Exception as e:
logger.warning('failed to read cached context: %s', e)
return None
# ── 대화 히스토리 (계정별, 24h TTL) ──
HISTORY_TTL = 86400 # 24시간
MAX_HISTORY = 50
def save_chat_history(user_id: str, messages: list[dict]):
"""대화 히스토리 저장 (최근 50개 메시지만 유지)."""
try:
r = _get_redis()
key = f'kcg:chat:history:{user_id}'
trimmed = messages[-MAX_HISTORY:]
r.setex(key, HISTORY_TTL, json.dumps(trimmed, ensure_ascii=False))
except Exception as e:
logger.warning('failed to save chat history for %s: %s', user_id, e)
def load_chat_history(user_id: str) -> list[dict]:
"""대화 히스토리 로드."""
try:
r = _get_redis()
data = r.get(f'kcg:chat:history:{user_id}')
return json.loads(data) if data else []
except Exception as e:
logger.warning('failed to load chat history for %s: %s', user_id, e)
return []
def clear_chat_history(user_id: str):
"""대화 히스토리 삭제."""
try:
r = _get_redis()
r.delete(f'kcg:chat:history:{user_id}')
except Exception as e:
logger.warning('failed to clear chat history for %s: %s', user_id, e)

140
prediction/chat/context_builder.py Normal file
파일 보기

@ -0,0 +1,140 @@
"""vessel_store + kcgdb 분석 데이터 + 도메인 지식을 기반으로 LLM 시스템 프롬프트를 구성."""
import logging
import re
from datetime import datetime, timezone
from chat.cache import get_cached_context
from chat.domain_knowledge import build_compact_prompt
logger = logging.getLogger(__name__)
def _build_realtime_context(ctx: dict) -> str:
"""Redis 캐시 데이터로 실시간 현황 프롬프트 구성 (간소화)."""
stats = ctx.get('vessel_stats', {})
risk = ctx.get('risk_distribution', {})
now = datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')
return f"""## 현황 ({now})
전체 {stats.get('vessels', 0)}척, 중국 {stats.get('chinese', 0)}척, 분석완료 {stats.get('targets', 0)}척, 허가 {stats.get('permitted', 0)}/906척
CRITICAL {risk.get('CRITICAL', 0)} / HIGH {risk.get('HIGH', 0)} / MEDIUM {risk.get('MEDIUM', 0)} / LOW {risk.get('LOW', 0)}
다크 {ctx.get('dark_count', 0)} / 스푸핑 {ctx.get('spoofing_count', 0)} / 환적 {ctx.get('transship_count', 0)}
영해 {risk.get('TERRITORIAL_SEA', 0)} / 접속 {risk.get('CONTIGUOUS_ZONE', 0)} / I {risk.get('ZONE_I', 0)} / II {risk.get('ZONE_II', 0)} / III {risk.get('ZONE_III', 0)} / IV {risk.get('ZONE_IV', 0)} / EEZ {risk.get('EEZ_OR_BEYOND', 0)}
(상세 데이터는 query_vessels 도구로 조회)"""
def _build_fallback_context() -> str:
"""Redis 캐시가 없을 때 vessel_store + kcgdb에서 직접 구성."""
try:
from cache.vessel_store import vessel_store
stats = vessel_store.stats()
from db import kcgdb
summary = kcgdb.fetch_analysis_summary()
top_risk = kcgdb.fetch_recent_high_risk(10)
polygon_summary = kcgdb.fetch_polygon_summary()
ctx = {
'vessel_stats': stats,
'risk_distribution': summary.get('risk_distribution', {}),
'dark_count': summary.get('dark_count', 0),
'spoofing_count': summary.get('spoofing_count', 0),
'transship_count': summary.get('transship_count', 0),
'top_risk_vessels': top_risk,
'polygon_summary': polygon_summary,
}
from chat.cache import cache_analysis_context
cache_analysis_context(ctx)
return _build_realtime_context(ctx)
except Exception as e:
logger.error('fallback context build failed: %s', e)
return '(실시간 데이터를 불러올 수 없습니다. 일반 해양 감시 지식으로 답변합니다.)'
# ── RAG: 사용자 질문에서 MMSI를 추출하여 선박별 상세 컨텍스트 주입 ──
_MMSI_PATTERN = re.compile(r'\b(\d{9})\b')
def _extract_mmsis(text: str) -> list[str]:
"""사용자 메시지에서 9자리 MMSI 추출."""
return _MMSI_PATTERN.findall(text)
def _build_vessel_detail(mmsi: str) -> str:
"""특정 MMSI의 분석 결과를 상세 컨텍스트로 구성 (RAG)."""
try:
from cache.vessel_store import vessel_store
info = vessel_store.get_vessel_info(mmsi)
positions = vessel_store.get_all_latest_positions()
pos = positions.get(mmsi)
from db import kcgdb
high_risk = kcgdb.fetch_recent_high_risk(100)
vessel_data = next((v for v in high_risk if v['mmsi'] == mmsi), None)
if not vessel_data and not pos:
return f'\n(MMSI {mmsi}: 분석 데이터 없음)\n'
lines = [f'\n## 선박 상세: {mmsi}']
if info:
name = info.get('name', 'N/A')
lines.append(f'- 선명: {name}')
if pos:
lines.append(f"- 위치: {pos.get('lat', 'N/A')}°N, {pos.get('lon', 'N/A')}°E")
lines.append(f"- SOG: {pos.get('sog', 'N/A')} knots, COG: {pos.get('cog', 'N/A')}°")
is_permitted = vessel_store.is_permitted(mmsi)
lines.append(f"- 허가 여부: {'허가어선' if is_permitted else '미허가/미등록'}")
if vessel_data:
lines.append(f"- 위험도: {vessel_data.get('risk_score', 'N/A')}점 ({vessel_data.get('risk_level', 'N/A')})")
lines.append(f"- 수역: {vessel_data.get('zone', 'N/A')}")
lines.append(f"- 활동: {vessel_data.get('activity_state', 'N/A')}")
lines.append(f"- 다크베셀: {'Y' if vessel_data.get('is_dark') else 'N'}")
lines.append(f"- 환적 의심: {'Y' if vessel_data.get('is_transship') else 'N'}")
lines.append(f"- 스푸핑 점수: {vessel_data.get('spoofing_score', 0):.2f}")
return '\n'.join(lines)
except Exception as e:
logger.warning('vessel detail build failed for %s: %s', mmsi, e)
return f'\n(MMSI {mmsi}: 상세 조회 실패)\n'
class MaritimeContextBuilder:
"""도메인 지식 + 실시간 데이터 + 선박별 RAG를 결합하여 시스템 프롬프트 구성."""
def build_system_prompt(self, user_message: str = '') -> str:
"""시스템 프롬프트 구성.
구조:
1) 압축 도메인 지식 (~500토큰: 역할+핵심용어+도구목록)
2) 실시간 현황 (Redis 캐시 → DB fallback)
3) RAG: 사용자 질문에 포함된 MMSI의 선박별 상세 데이터
상세 도메인 지식은 LLM이 get_knowledge 도구로 필요 시 조회.
"""
parts = []
# 1) 압축 도메인 지식 (~500토큰)
parts.append(build_compact_prompt())
# 2) 실시간 현황
cached = get_cached_context()
if cached:
parts.append(_build_realtime_context(cached))
else:
parts.append(_build_fallback_context())
# 3) RAG: MMSI 기반 선박 상세
if user_message:
mmsis = _extract_mmsis(user_message)
for mmsi in mmsis[:3]: # 최대 3척
parts.append(_build_vessel_detail(mmsi))
return '\n\n'.join(parts)

467
prediction/chat/domain_knowledge.py Normal file
파일 보기

@ -0,0 +1,467 @@
"""해양 감시 도메인 전문 지식 — LLM 시스템 프롬프트 보강용.
수집 출처:
- 한중어업협정 (2001.6.30 발효, 한국민족문화대백과사전)
- 해양수산부 한중어업공동위원회 결과 공표
- UNCLOS 해양법협약 (영해/접속수역/EEZ 기준)
- Global Fishing Watch 환적 탐지 기준
- 해양경찰청 불법조업 단속 현황
- MarineTraffic AIS/GNSS 스푸핑 가이드
"""
# ── 역할 정의 ──
ROLE_DEFINITION = """당신은 대한민국 해양경찰청의 **해양상황 분석 AI 어시스턴트**입니다.
Python AI 분석 파이프라인(7단계 + 8개 알고리즘)의 실시간 결과를 기반으로,
해양 감시 전문가 수준의 분석과 조치 권고를 제공합니다.
당신이 접근하는 데이터:
- 14,000척 이상의 AIS 실시간 위치 (24시간 슬라이딩 윈도우)
- 중국 어선(412* MMSI) 대상 AI 분석 결과 (28개 필드, 5분 주기 갱신)
- 선단/어구 그룹 폴리곤 (Shapely 기반, 5분 주기)
- 한중어업협정 허가어선 DB (906척 등록)"""
# ── 해양 수역 법적 체계 ──
MARITIME_ZONES = """## 해양 수역 법적 체계 (UNCLOS + 국내법)
| 수역 | 범위 | 법적 지위 | 단속 권한 |
|------|------|----------|----------|
| **영해** (TERRITORIAL_SEA) | 기선~12해리 | 완전한 주권 | 즉시 나포 가능 |
| **접속수역** (CONTIGUOUS_ZONE) | 12~24해리 | 관세·출입국 통제 | 정선·검색 가능 |
| **EEZ** (EEZ_OR_BEYOND) | 24~200해리 | 자원 주권적 권리 | 어업법 적용 |
- 1해리 = 1,852m, 기선은 서해·남해 직선기선, 동해 통상기선
- 서해는 한중 간 중간선이 200해리 미만이므로 EEZ 경계 미확정
- 독도·울릉도·제주도는 각 섬 해안에서 12해리
### 특정어업수역 (한중어업협정)
- **수역 I~IV**: 한국 EEZ 내 중국 허가어선 조업 가능 구역
- **잠정조치수역**: 약 83,000km², 한중 공동 관리 (북위 37°~32°11')
- **과도수역**: 잠정조치수역 좌우 20해리 (2005.6.30부터 연차 감축)
- 수역 외 조업 = **불법** (무허가 조업)"""
# ── 한중어업협정 상세 ──
FISHING_AGREEMENT = """## 한중어업협정 상세 (2001.6.30 발효)
### 허가어선 현황 (총 906척)
| 어구코드 | 어구명 | 허가 수 | 비고 |
|---------|--------|---------|------|
| PT | 쌍끌이 저인망 | 323쌍 (646척) | 2척 1조 운영 |
| GN | 유자망 (길그물) | 200척 | |
| PS | 위망 (선망) | 16척 | |
| OT | 기선인망 (외끌이) | 13척 | 1척 단독 |
| FC | 운반선 | 31척 | 어획물 운반 전용 |
### 휴어기 (조업 금지 기간)
| 어구 | 기간 | 비고 |
|------|------|------|
| PT (저인망) | 4/16 ~ 10/15 (6개월) | 산란기 보호 |
| OT (외끌이) | 4/16 ~ 10/15 (6개월) | PT와 동일 |
| GN (유자망) | 6/2 ~ 8/31 (3개월) | 하절기 |
### 어구별 조업 속도 기준 (UCAF 판정 참조)
| 어구 | 조업 속도 | 항행 속도 | 판별 기준 |
|------|----------|----------|----------|
| PT/OT (저인망) | 2.5~4.5 knots | 6+ knots | 그물 끌기 중 |
| GN (유자망) | 0.5~2.0 knots | 5+ knots | 그물 투망/양망 |
| PS (위망) | 1.0~3.0 knots | 7+ knots | 그물 투·양망 |
| TRAP (통발) | 0.5~2.0 knots | 5+ knots | 통발 투·양 |
| LONGLINE (연승) | 1.0~3.0 knots | 6+ knots | 줄 투·양승 |
### 2024.5.1 시행 신규 합의사항
- 한국 EEZ 내 모든 중국어선 **AIS 의무 장착·가동**
- 자망어선: 어구마다 부표/깃대 설치 의무 (30×20cm 표지)
- 위반 시: 허가 취소 + 벌금 + 3년 이내 재허가 불가"""
# ── 알고리즘 해석 가이드 ──
ALGORITHM_GUIDE = """## AI 분석 알고리즘 해석 가이드 (8개 알고리즘)
### ALGO 01: 위치 분석 (location)
- `zone`: 선박이 현재 위치한 해양 수역
- TERRITORIAL_SEA (영해): **즉각 주의** — 외국어선 영해 침범
- CONTIGUOUS_ZONE (접속수역): 감시 강화 필요
- ZONE_I~IV (특정어업수역): 허가 여부 확인 필수
- EEZ_OR_BEYOND: 일반 감시
- `dist_to_baseline_nm`: 기선까지 거리 (NM)
- <12NM: 영해 내 — 최고 위험
- 12~24NM: 접속수역 — 높은 경계
- >24NM: EEZ 이원
### ALGO 02: 활동 패턴 (activity)
- `activity_state`: STATIONARY(정박) / FISHING(조업) / SAILING(항행)
- SOG ≤1.0 → STATIONARY
- SOG 1.0~5.0 → FISHING (어구에 따라 다름)
- SOG >5.0 → SAILING
- `ucaf_score` (0~1): 어구별 조업속도 매칭률
- >0.7: 높은 확률로 해당 어구 사용 중
- 0.3~0.7: 불확실
- <0.3: 비매칭 (다른 어구이거나 항행 중)
- `ucft_score` (0~1): 조업-항행 구분 신뢰도
- >0.8: 명확히 조업/항행 구분됨
- <0.5: 패턴 불명확
### ALGO 03: 다크베셀 (dark_vessel)
- `is_dark`: AIS 신호 의도적 차단 의심
- `gap_duration_min`: AIS 최장 공백 시간 (분)
- 30~60분: 경미한 갭 (기술적 원인 가능)
- 60~180분: 의심 수준 — 의도적 차단 가능성
- 180분+: **높은 의심** — 불법조업 은폐 목적 추정
- 참고: 2024.5.1부터 한국 EEZ 내 중국어선 AIS 의무화
- AIS 차단 자체가 **협정 위반**
### ALGO 04: GPS 스푸핑 (gps_spoofing)
- `spoofing_score` (0~1): 종합 스푸핑 의심도
- >0.7: **높은 스푸핑 의심** — 위치 조작 추정
- 0.3~0.7: 중간 의심
- <0.3: 정상
- `bd09_offset_m`: 바이두(BD-09) 좌표계 오프셋 (미터)
- 중국 선박 특유의 GPS 좌표 변환 오차
- 412* MMSI는 기본 제외 (중국 위성항법 특성)
- `speed_jump_count`: 비현실적 속도 점프 횟수
- 0: 정상
- 1~2: 일시적 GPS 오류 가능
- 3+: **스푸핑 강력 의심** — 위치 은폐 목적
### ALGO 05-06: 선단 분석 (fleet/cluster)
- `cluster_id`: 선단 그룹 ID (-1 = 미소속)
- `cluster_size`: 같은 선단 소속 선박 수
- 2~5: 소규모 선단
- 5~15: 중규모 선단 (일반적)
- 15+: 대규모 선단 — 조직적 조업
- `fleet_role`: 선단 내 역할
- LEADER: 선단 지휘선 (이동 경로 결정)
- FOLLOWER: 추종선 (리더 경로 따름)
- PROCESS_VESSEL: 가공선 (어획물 처리)
- FUEL_VESSEL: 급유선
- NOISE: 미분류
### ALGO 07: 위험도 종합 (risk_score)
- 0~100점 종합 점수, 4개 영역 합산:
- **위치** (최대 40점): 영해 내=40, 접속수역=10
- **조업 행위** (최대 30점): 영해 내 조업=20, 기타 조업=5, U-turn 패턴=10
- **AIS 조작** (최대 35점): 순간이동=20, 장시간 갭=15, 단시간 갭=5
- **허가 이력** (최대 20점): 미허가 어선=20
- 등급: CRITICAL(≥70) / HIGH(≥50) / MEDIUM(≥30) / LOW(<30)
- 프론트엔드 표시: WATCH=HIGH, MONITOR=MEDIUM, NORMAL=LOW
### ALGO 08: 환적 의심 (transshipment)
- `is_transship_suspect`: 해상 환적 의심 여부
- `transship_pair_mmsi`: 상대 선박 MMSI
- `transship_duration_min`: 접촉 지속 시간 (분)
- 탐지 기준 (Global Fishing Watch 참조):
- 두 선박 500m 이내 접근
- 속도 2노트 미만
- 2시간 이상 지속
- 정박지에서 10km 이상 떨어진 해상"""
# ── 대응 절차 가이드 ──
RESPONSE_GUIDE = """## 위험도별 대응 절차 권고
### CRITICAL (≥70점) — 즉각 대응
1. 해당 선박 위치·항적 실시간 추적
2. 인근 경비함정 긴급 출동 지시
3. VHF 채널 16 경고방송 (한국어+중국어)
4. 정선명령 → 승선검색 → 나포
5. 상급기관 즉시 보고
### WATCH/HIGH (≥50점) — 강화 감시
1. 감시 우선순위 상향
2. 항적 지속 추적 (15분 간격)
3. 인근 해역 순찰 함정에 정보 공유
4. 위험도 변화 시 CRITICAL 대응 전환 준비
### MONITOR/MEDIUM (≥30점) — 일반 감시
1. 정기 모니터링 대상 등록
2. 1시간 간격 위치·상태 확인
3. 패턴 변화(조업→이동, 군집화 등) 시 알림
### NORMAL/LOW (<30점) — 기본 감시
1. 시스템 자동 모니터링
2. 일일 요약 보고에 포함
### 불법조업 유형별 조치
| 유형 | 해당 알고리즘 | 즉시 조치 |
|------|-------------|----------|
| 영해 침범 | zone=TERRITORIAL_SEA | 나포 (영해법 위반) |
| 무허가 조업 | is_permitted=False + zone=ZONE_* | 정선·검색 |
| AIS 차단 | is_dark=True, gap>60min | 위치 추적 + 출동 |
| GPS 위치조작 | spoofing_score>0.7 | 실제 위치 특정 후 출동 |
| 불법 환적 | is_transship_suspect=True | 쌍방 정선·검색 |
| 휴어기 위반 | 어구+날짜 크로스체크 | 정선·어구 확인 |"""
# ── 응답 규칙 ──
RESPONSE_RULES = """## 응답 규칙
- 한국어로 답변
- 데이터 기반 분석 (추측 최소화, 근거 수치 명시)
- 구체적 MMSI, 좌표, 점수, 수역명 제시
- 불법조업 의심 시 **법적 근거 + 알고리즘 근거 + 조치 권고** 3가지를 함께 제시
- 위험도 등급 언급 시 점수도 함께 표기 (예: "CRITICAL(82점)")
- 마크다운 형식으로 구조화 (표, 목록, 강조 활용)
- "~일 수 있습니다" 대신 데이터에 근거한 단정적 분석 제공
- 선박 특정 질문 시 해당 선박의 모든 알고리즘 결과를 종합 제시"""
# ── DB 스키마 + Tool Calling 가이드 ──
DB_SCHEMA_AND_TOOLS = """## 데이터 조회 도구 (Tool Calling)
사용자 질문에 답하기 위해 실시간 DB 조회가 필요하면, 다음 도구를 호출할 수 있습니다.
도구 호출 시 반드시 아래 형식을 사용하세요:
### 사용 가능한 도구
#### 1. query_vessels — 선박 분석 결과 조회
조건에 맞는 선박 목록을 조회합니다.
```json
{"tool": "query_vessels", "params": {"zone": "ZONE_I", "activity": "FISHING", "risk_level": "CRITICAL", "is_dark": true, "limit": 20}}
```
- 모든 파라미터는 선택적 (조합 가능)
- zone 값: TERRITORIAL_SEA, CONTIGUOUS_ZONE, ZONE_I, ZONE_II, ZONE_III, ZONE_IV, EEZ_OR_BEYOND
- activity 값: STATIONARY, FISHING, SAILING
- risk_level 값: CRITICAL, HIGH, MEDIUM, LOW
- is_dark: true/false
- is_transship: true/false
- vessel_type 값: TRAWL, PURSE, LONGLINE, TRAP, UNKNOWN
- limit: 최대 반환 수 (기본 20)
#### 2. query_vessel_detail — 특정 선박 상세
```json
{"tool": "query_vessel_detail", "params": {"mmsi": "412236758"}}
```
#### 3. query_fleet_group — 선단/어구 그룹 조회
```json
{"tool": "query_fleet_group", "params": {"group_type": "FLEET", "zone_id": "ZONE_I"}}
```
- group_type: FLEET, GEAR_IN_ZONE, GEAR_OUT_ZONE
#### 4. query_vessel_history — 선박 항적 이력 (snpdb daily)
```json
{"tool": "query_vessel_history", "params": {"mmsi": "412236758", "days": 7}}
```
- 일별 이동거리, 평균/최대 속도, AIS 포인트 수
- 최대 30일까지 조회
#### 5. query_vessel_static — 선박 정적정보 + 변경 이력 (snpdb)
```json
{"tool": "query_vessel_static", "params": {"mmsi": "412236758", "limit": 10}}
```
- 최신 선명/선종/제원/목적지/상태 + 변경 이력 감지
- 선명·목적지·상태 변경 시점과 이전/이후 값 표시
### DB 스키마 참조 (쿼리 조합 시 참고)
#### kcg.vessel_analysis_results (5분 주기 갱신, 48시간 보존)
| 컬럼 | 타입 | 값 예시 |
|------|------|---------|
| mmsi | varchar | '412236758' (중국=412*) |
| timestamp | timestamptz | 분석 시점 |
| vessel_type | varchar | TRAWL/PURSE/LONGLINE/TRAP/UNKNOWN |
| zone | varchar | TERRITORIAL_SEA/CONTIGUOUS_ZONE/ZONE_I~IV/EEZ_OR_BEYOND |
| dist_to_baseline_nm | float | 기선까지 거리(NM) |
| activity_state | varchar | STATIONARY/FISHING/SAILING |
| ucaf_score | float | 0~1 (어구 매칭률) |
| is_dark | boolean | AIS 차단 의심 |
| gap_duration_min | int | AIS 최장 공백(분) |
| spoofing_score | float | 0~1 |
| risk_score | int | 0~100 |
| risk_level | varchar | CRITICAL(≥70)/HIGH(≥50)/MEDIUM(≥30)/LOW(<30) |
| cluster_id | int | 선단 ID (-1=미소속) |
| cluster_size | int | 선단 규모 |
| fleet_role | varchar | LEADER/FOLLOWER/PROCESS_VESSEL/FUEL_VESSEL/NOISE |
| is_transship_suspect | boolean | 환적 의심 |
| transship_pair_mmsi | varchar | 상대 선박 |
| analyzed_at | timestamptz | WHERE 조건에 사용 (> NOW() - '1 hour') |
- PK: (mmsi, timestamp), 인덱스: mmsi, timestamp DESC
#### kcg.fleet_vessels (허가어선 등록부)
| 컬럼 | 타입 | 설명 |
|------|------|------|
| mmsi | varchar | 매칭된 MMSI (NULL 가능) |
| permit_no | varchar | 허가번호 |
| name_cn | text | 중국어 선명 |
| gear_code | varchar | PT/GN/PS/OT/FC |
| company_id | int | → fleet_companies.id |
| tonnage | int | 톤수 |
#### kcg.group_polygon_snapshots (선단/어구 폴리곤, 5분 APPEND, 7일 보존)
| 컬럼 | 타입 | 설명 |
|------|------|------|
| group_type | varchar | FLEET/GEAR_IN_ZONE/GEAR_OUT_ZONE |
| group_key | varchar | 그룹 식별자 |
| group_label | text | 표시 라벨 |
| snapshot_time | timestamptz | 스냅샷 시점 |
| member_count | int | 소속 선박 수 |
| zone_id | varchar | 수역 ID |
| members | jsonb | [{mmsi, name, lat, lon, sog, cog, ...}] |
### snpdb 테이블 상세 (signal 스키마, 읽기 전용)
#### signal.t_vessel_tracks_5min — 실시간 항적 (5분 집계)
| 컬럼 | 타입 | 설명 |
|------|------|------|
| mmsi | varchar | 선박 ID |
| time_bucket | timestamp | 5분 버킷 시점 |
| track_geom | LineStringM | 타임스탬프 포함 궤적 |
| distance_nm | numeric | 이동 거리(NM) |
| avg_speed | numeric | 평균 속도(knots) |
| max_speed | numeric | 최대 속도(knots) |
| point_count | int | AIS 포인트 수 |
| start_position | jsonb | {lat, lon, sog, cog, timestamp} |
| end_position | jsonb | {lat, lon, sog, cog, timestamp} |
- PK: (mmsi, time_bucket), 인덱스: mmsi, time_bucket
- **일별 파티셔닝**: t_vessel_tracks_5min_YYMMDD (예: _260326 = 2026-03-26)
- 하루 약 850만 건, vessel_store에 24시간 인메모리 캐시
- **활용**: 최근 수 시간 ~ 24시간 내 세밀한 이동 패턴 분석
#### signal.t_vessel_tracks_hourly — 시간별 항적 집계
| 컬럼 | 타입 | 설명 |
|------|------|------|
| mmsi | varchar | 선박 ID |
| time_bucket | timestamp | 1시간 버킷 |
| track_geom | LineStringM | 시간별 궤적 |
| distance_nm | numeric | 시간당 이동 거리 |
| avg_speed | numeric | 평균 속도 |
| max_speed | numeric | 최대 속도 |
| point_count | int | AIS 포인트 수 |
| start_position | jsonb | 시작 위치 |
| end_position | jsonb | 종료 위치 |
- **월별 파티셔닝**: t_vessel_tracks_hourly_YYYY_MM (예: _2026_03)
- 월 약 1.2억 건
- **활용**: 수일~수주 단위 이동 경로 추적, 패턴 비교
#### signal.t_vessel_tracks_daily — 일별 항적 요약
| 컬럼 | 타입 | 설명 |
|------|------|------|
| mmsi | varchar | 선박 ID |
| time_bucket | date | 날짜 |
| track_geom | LineStringM | 하루 궤적 |
| distance_nm | numeric | 일일 이동 거리(NM) |
| avg_speed | numeric | 일 평균 속도 |
| max_speed | numeric | 일 최대 속도 |
| point_count | int | AIS 포인트 수 |
| operating_hours | numeric | 운항 시간 |
| port_visits | jsonb | 입출항 기록 |
| start_position | jsonb | 일 시작 위치 |
| end_position | jsonb | 일 종료 위치 |
- **월별 파티셔닝**: t_vessel_tracks_daily_YYYY_MM (예: _2026_03)
- 월 약 800만 건, **2015년 8월~현재** 11년+ 이력
- **활용**: 장기 행동 패턴, 계절별 어장 이동, 기간 비교 분석
#### signal.t_vessel_static — 선박 정적정보 (1시간 주기 스냅샷)
| 컬럼 | 타입 | 설명 | 값 예시 |
|------|------|------|---------|
| mmsi | varchar | 선박 ID | '412236758' |
| time_bucket | timestamptz | 스냅샷 시점 (1시간 간격) | |
| imo | bigint | IMO 번호 | |
| name | varchar | 선명 (AIS 브로드캐스트) | 'LU_RONG_YU_55759' |
| callsign | varchar | 호출부호 | |
| vessel_type | varchar | 선종 | Cargo/Tanker/Vessel/Fishing/N/A 등 |
| extra_info | varchar | 추가 정보 | |
| length | int | 선장(m) | |
| width | int | 선폭(m) | |
| draught | float | 흘수(m) | |
| destination | varchar | 목적지 (AIS 입력) | 'PU TIAN' |
| eta | timestamptz | 도착 예정 시각 | |
| status | varchar | 항해 상태 | Under way using engine/Moored/Anchored/Engaged in fishing |
| class_type | varchar | AIS 클래스 | A/B |
- PK: (mmsi, time_bucket)
- **변경 이력 보존**: 동일 MMSI라도 1시간마다 스냅샷 저장. name, destination, status 등이 변경되면 히스토리로 추적 가능
- **활용 예시**:
- 선명 변경 이력 추적 (위장/은폐 탐지)
- 목적지(destination) 변경 패턴 분석
- AIS 상태(status) 시계열 — 'Engaged in fishing' ↔ 'Under way' 전환 빈도
- 선박 제원(length/width/draught) 불일치 탐지
### snpdb 테이블 활용 가이드
| 분석 목적 | 사용 테이블 | 조회 범위 | 쿼리 팁 |
|----------|-----------|----------|---------|
| **실시간 위치 추적** | 5min (오늘 파티션) | 최근 수 시간 | `_YYMMDD` 파티션 직접 지정 |
| **최근 항적 패턴** | 5min | 최근 24h | vessel_store 인메모리 캐시 우선 |
| **수일간 이동 경로** | hourly | 최근 7일 | `_YYYY_MM` 월 파티션 |
| **장기 행동 패턴** | daily | 수개월~수년 | 월 파티션, distance_nm 집계 |
| **선명/목적지 변경** | static | 변경 이력 | mmsi 기준 time_bucket DESC |
| **선박 제원 확인** | static | 최신 1건 | MAX(time_bucket) |
| **AIS 상태 시계열** | static | 최근 수일 | status 변화 패턴 |
| **계절 조업 패턴** | daily | 연 단위 | 월별 distance_nm, avg_speed 비교 |
### 파티션 테이블 쿼리 시 주의
- 5min: `signal.t_vessel_tracks_5min_YYMMDD` (날짜 6자리)
- hourly: `signal.t_vessel_tracks_hourly_YYYY_MM` (연_월)
- daily: `signal.t_vessel_tracks_daily_YYYY_MM` (연_월)
- **부모 테이블 직접 조회 가능** (PostgreSQL이 파티션 프루닝 수행)
- 대량 조회 시 파티션 직접 지정이 성능에 유리
### 데이터 흐름
```
snpdb (AIS 원본 항적) → vessel_store (인메모리 24h) → 7단계 파이프라인
→ kcgdb.vessel_analysis_results (분석 결과, 48h 보존)
→ kcgdb.group_polygon_snapshots (선단/어구 폴리곤, 7일 보존)
→ Redis (채팅 컨텍스트 캐시, 6분 TTL)
```
### 도구 호출 규칙
- 답변에 필요한 구체적 선박 목록이 시스템 프롬프트에 없으면 도구를 호출하세요
- 도구 호출 결과를 받은 후, 그 데이터를 기반으로 답변하세요
- 한 번에 최대 2개 도구 호출 가능
- 집계 데이터(몇 척인지)는 이미 시스템 프롬프트에 있으므로 도구 불필요
- 대부분의 질문은 kcgdb로 충분 — snpdb 직접 조회는 특수한 항적 분석에만 사용"""
# ── 지식 섹션 레지스트리 (키워드 → 상세 텍스트) ──
KNOWLEDGE_SECTIONS: dict[str, str] = {
'maritime_zones': MARITIME_ZONES,
'fishing_agreement': FISHING_AGREEMENT,
'algorithm_guide': ALGORITHM_GUIDE,
'response_guide': RESPONSE_GUIDE,
'db_schema': DB_SCHEMA_AND_TOOLS,
}
def get_knowledge_section(key: str) -> str:
"""키워드로 특정 도메인 지식 섹션을 반환."""
return KNOWLEDGE_SECTIONS.get(key, f'(알 수 없는 지식 키: {key})')
# ── 압축 시스템 프롬프트 (항상 포함, ~500토큰) ──
COMPACT_SYSTEM_PROMPT = """당신은 대한민국 해양경찰청의 해양상황 분석 AI 어시스턴트입니다.
14,000척 AIS 실시간 모니터링 + AI 분석 파이프라인(8개 알고리즘) 결과를 기반으로 답변합니다.
핵심 용어:
- 수역: 영해(TERRITORIAL_SEA, 12NM이내), 접속수역(CONTIGUOUS_ZONE, 12~24NM), 특정어업수역(ZONE_I~IV), EEZ
- 위험도: CRITICAL(≥70) / HIGH/WATCH(≥50) / MEDIUM/MONITOR(≥30) / LOW/NORMAL(<30)
- 다크베셀: AIS 의도적 차단 (gap_duration_min), 2024.5.1부터 AIS 의무화
- 허가어선: 906척 등록 (PT 저인망 323쌍, GN 유자망 200, PS 위망 16, OT 외끌이 13, FC 운반 31)
- 휴어기: PT/OT 4/16~10/15, GN 6/2~8/31
도구를 호출하여 데이터를 조회하거나 상세 지식에 접근할 수 있습니다:
- query_vessels: 조건별 선박 목록 조회 (zone, activity, risk_level, is_dark, vessel_type)
- query_vessel_detail: MMSI별 상세 분석 결과
- query_fleet_group: 선단/어구 그룹 조회
- query_vessel_history: 일별 항적 이력 (snpdb, 최대 30일)
- query_vessel_static: 선박 정적정보 + 변경 이력 (snpdb)
- get_knowledge: 상세 도메인 지식 조회 (키: maritime_zones, fishing_agreement, algorithm_guide, response_guide, db_schema)
도구 호출 형식:
```json
{"tool": "도구명", "params": {"key": "value"}}
```
응답 규칙: 한국어, 데이터 기반, 구체적 수치 명시, 마크다운 형식, 불법 의심 시 근거+조치 권고"""
def build_domain_knowledge() -> str:
"""전체 도메인 지식 반환 (레거시 호환용)."""
return '\n\n'.join([
ROLE_DEFINITION,
MARITIME_ZONES,
FISHING_AGREEMENT,
ALGORITHM_GUIDE,
RESPONSE_GUIDE,
RESPONSE_RULES,
DB_SCHEMA_AND_TOOLS,
])
def build_compact_prompt() -> str:
"""압축 시스템 프롬프트 반환 (~500토큰)."""
return COMPACT_SYSTEM_PROMPT

236
prediction/chat/router.py Normal file
파일 보기

@ -0,0 +1,236 @@
"""AI 해양분석 채팅 엔드포인트 — 사전 쿼리 + SSE 스트리밍 + Tool Calling."""
import json
import logging
import httpx
from fastapi import APIRouter
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from chat.cache import load_chat_history, save_chat_history, clear_chat_history
from chat.context_builder import MaritimeContextBuilder
from chat.tools import detect_prequery, execute_prequery, parse_tool_calls, execute_tool_call
from config import settings
logger = logging.getLogger(__name__)
router = APIRouter(prefix='/api/v1/chat', tags=['chat'])
class ChatRequest(BaseModel):
message: str
user_id: str = 'anonymous'
stream: bool = True
class ChatResponse(BaseModel):
role: str = 'assistant'
content: str
@router.post('')
async def chat(req: ChatRequest):
"""해양분석 채팅 — 사전 쿼리 + 분석 컨텍스트 + Ollama SSE 스트리밍."""
history = load_chat_history(req.user_id)
builder = MaritimeContextBuilder()
system_prompt = builder.build_system_prompt(user_message=req.message)
# ── 사전 쿼리: 키워드 패턴 매칭으로 DB 조회 후 컨텍스트 보강 ──
prequery_params = detect_prequery(req.message)
prequery_result = ''
if prequery_params:
prequery_result = execute_prequery(prequery_params)
logger.info('prequery: params=%s, results=%d chars', prequery_params, len(prequery_result))
# 시스템 프롬프트에 사전 쿼리 결과 추가
if prequery_result:
system_prompt += '\n\n' + prequery_result
messages = [
{'role': 'system', 'content': system_prompt},
*history[-10:],
{'role': 'user', 'content': req.message},
]
ollama_payload = {
'model': settings.OLLAMA_MODEL,
'messages': messages,
'stream': req.stream,
'options': {
'temperature': 0.3,
'num_predict': 1024,
'num_ctx': 2048,
},
}
if req.stream:
return StreamingResponse(
_stream_with_tools(ollama_payload, req.user_id, history, req.message),
media_type='text/event-stream',
headers={
'Cache-Control': 'no-cache',
'Connection': 'keep-alive',
'X-Accel-Buffering': 'no',
},
)
return await _call_with_tools(ollama_payload, req.user_id, history, req.message)
async def _stream_with_tools(payload: dict, user_id: str, history: list[dict], user_message: str):
"""SSE 스트리밍 — 1차 응답 후 Tool Call 감지 시 2차 호출."""
accumulated = ''
try:
async with httpx.AsyncClient(timeout=httpx.Timeout(settings.OLLAMA_TIMEOUT_SEC)) as client:
# 1차 LLM 호출
async with client.stream(
'POST',
f'{settings.OLLAMA_BASE_URL}/api/chat',
json=payload,
) as response:
async for line in response.aiter_lines():
if not line:
continue
try:
chunk = json.loads(line)
content = chunk.get('message', {}).get('content', '')
done = chunk.get('done', False)
accumulated += content
sse_data = json.dumps({
'content': content,
'done': False, # 아직 done 보내지 않음 (tool call 가능)
}, ensure_ascii=False)
yield f'data: {sse_data}\n\n'
if done:
break
except json.JSONDecodeError:
continue
# Tool Call 감지
tool_calls = parse_tool_calls(accumulated)
if tool_calls:
# Tool 실행
tool_results = []
for tc in tool_calls:
result = execute_tool_call(tc)
tool_results.append(result)
logger.info('tool call: %s → %d chars', tc.get('tool'), len(result))
tool_context = '\n'.join(tool_results)
# 2차 LLM 호출 (tool 결과 포함)
payload['messages'].append({'role': 'assistant', 'content': accumulated})
payload['messages'].append({
'role': 'user',
'content': f'도구 조회 결과입니다. 이 데이터를 기반으로 사용자 질문에 답변하세요:\n{tool_context}',
})
# 구분자 전송
separator = json.dumps({'content': '\n\n---\n_데이터 조회 완료. 분석 결과:_\n\n', 'done': False}, ensure_ascii=False)
yield f'data: {separator}\n\n'
accumulated_2 = ''
async with client.stream(
'POST',
f'{settings.OLLAMA_BASE_URL}/api/chat',
json=payload,
) as response2:
async for line in response2.aiter_lines():
if not line:
continue
try:
chunk = json.loads(line)
content = chunk.get('message', {}).get('content', '')
done = chunk.get('done', False)
accumulated_2 += content
sse_data = json.dumps({
'content': content,
'done': done,
}, ensure_ascii=False)
yield f'data: {sse_data}\n\n'
if done:
break
except json.JSONDecodeError:
continue
# 히스토리에는 최종 답변만 저장
accumulated = accumulated_2 or accumulated
except httpx.TimeoutException:
err_msg = json.dumps({'content': '\n\n[응답 시간 초과]', 'done': True})
yield f'data: {err_msg}\n\n'
except Exception as e:
logger.error('ollama stream error: %s', e)
err_msg = json.dumps({'content': f'[오류: {e}]', 'done': True})
yield f'data: {err_msg}\n\n'
if accumulated:
updated = history + [
{'role': 'user', 'content': user_message},
{'role': 'assistant', 'content': accumulated},
]
save_chat_history(user_id, updated)
yield 'data: [DONE]\n\n'
async def _call_with_tools(
payload: dict, user_id: str, history: list[dict], user_message: str,
) -> ChatResponse:
"""비스트리밍 — Tool Calling 포함."""
try:
async with httpx.AsyncClient(timeout=httpx.Timeout(settings.OLLAMA_TIMEOUT_SEC)) as client:
# 1차 호출
response = await client.post(
f'{settings.OLLAMA_BASE_URL}/api/chat',
json=payload,
)
data = response.json()
content = data.get('message', {}).get('content', '')
# Tool Call 감지
tool_calls = parse_tool_calls(content)
if tool_calls:
tool_results = [execute_tool_call(tc) for tc in tool_calls]
tool_context = '\n'.join(tool_results)
payload['messages'].append({'role': 'assistant', 'content': content})
payload['messages'].append({
'role': 'user',
'content': f'도구 조회 결과입니다. 이 데이터를 기반으로 답변하세요:\n{tool_context}',
})
response2 = await client.post(
f'{settings.OLLAMA_BASE_URL}/api/chat',
json=payload,
)
data2 = response2.json()
content = data2.get('message', {}).get('content', content)
updated = history + [
{'role': 'user', 'content': user_message},
{'role': 'assistant', 'content': content},
]
save_chat_history(user_id, updated)
return ChatResponse(content=content)
except Exception as e:
logger.error('ollama sync error: %s', e)
return ChatResponse(content=f'오류: AI 서버 연결 실패 ({e})')
@router.get('/history')
async def get_history(user_id: str = 'anonymous'):
return load_chat_history(user_id)
@router.delete('/history')
async def delete_history(user_id: str = 'anonymous'):
clear_chat_history(user_id)
return {'ok': True}

359
prediction/chat/tools.py Normal file
파일 보기

@ -0,0 +1,359 @@
"""LLM Tool Calling 실행기 — 사전 쿼리 + 동적 DB 조회."""
import json
import logging
import re
from typing import Optional
logger = logging.getLogger(__name__)
# ── 사전 쿼리 패턴 (키워드 기반, 1회 왕복으로 해결) ──
_ZONE_MAP = {
'수역1': 'ZONE_I', '수역 1': 'ZONE_I', '수역I': 'ZONE_I', 'ZONE_I': 'ZONE_I', '수역i': 'ZONE_I',
'수역2': 'ZONE_II', '수역 2': 'ZONE_II', '수역II': 'ZONE_II', 'ZONE_II': 'ZONE_II',
'수역3': 'ZONE_III', '수역 3': 'ZONE_III', '수역III': 'ZONE_III', 'ZONE_III': 'ZONE_III',
'수역4': 'ZONE_IV', '수역 4': 'ZONE_IV', '수역IV': 'ZONE_IV', 'ZONE_IV': 'ZONE_IV',
'영해': 'TERRITORIAL_SEA', '접속수역': 'CONTIGUOUS_ZONE',
}
_ACTIVITY_MAP = {
'조업': 'FISHING', '어로': 'FISHING', '조업중': 'FISHING', '조업활동': 'FISHING',
'정박': 'STATIONARY', '정지': 'STATIONARY', '대기': 'STATIONARY',
'항행': 'SAILING', '이동': 'SAILING', '항해': 'SAILING',
}
_RISK_MAP = {
'크리티컬': 'CRITICAL', 'critical': 'CRITICAL', '긴급': 'CRITICAL',
'워치': 'HIGH', 'watch': 'HIGH', '경고': 'HIGH', '고위험': 'HIGH',
'모니터': 'MEDIUM', 'monitor': 'MEDIUM', '주의': 'MEDIUM',
'위험': None, # 위험 선박 → CRITICAL+HIGH
}
_DARK_KEYWORDS = ['다크', '다크베셀', 'dark', 'ais 차단', 'ais차단', '신호차단']
_TRANSSHIP_KEYWORDS = ['환적', 'transshipment', '전재']
_SPOOF_KEYWORDS = ['스푸핑', 'spoofing', 'gps 조작', 'gps조작', '위치조작']
def detect_prequery(message: str) -> Optional[dict]:
"""사용자 메시지에서 사전 쿼리 패턴을 감지하여 DB 조회 파라미터 반환."""
msg = message.lower().strip()
params: dict = {}
# 수역 감지
for keyword, zone in _ZONE_MAP.items():
if keyword.lower() in msg:
params['zone'] = zone
break
# 활동 감지
for keyword, activity in _ACTIVITY_MAP.items():
if keyword in msg:
params['activity'] = activity
break
# 위험도 감지
for keyword, level in _RISK_MAP.items():
if keyword in msg:
if level:
params['risk_level'] = level
else:
params['risk_levels'] = ['CRITICAL', 'HIGH']
break
# 다크베셀 감지
if any(k in msg for k in _DARK_KEYWORDS):
params['is_dark'] = True
# 환적 감지
if any(k in msg for k in _TRANSSHIP_KEYWORDS):
params['is_transship'] = True
# 스푸핑 감지
if any(k in msg for k in _SPOOF_KEYWORDS):
params['spoofing'] = True
return params if params else None
def execute_prequery(params: dict) -> str:
"""사전 쿼리 패턴에 해당하는 DB 조회를 실행하여 결과를 텍스트로 반환."""
try:
from db import kcgdb
conditions = ["analyzed_at > NOW() - INTERVAL '1 hour'"]
bind_params: list = []
if 'zone' in params:
conditions.append('zone = %s')
bind_params.append(params['zone'])
if 'activity' in params:
conditions.append('activity_state = %s')
bind_params.append(params['activity'])
if 'risk_level' in params:
conditions.append('risk_level = %s')
bind_params.append(params['risk_level'])
elif 'risk_levels' in params:
placeholders = ','.join(['%s'] * len(params['risk_levels']))
conditions.append(f'risk_level IN ({placeholders})')
bind_params.extend(params['risk_levels'])
if params.get('is_dark'):
conditions.append('is_dark = TRUE')
if params.get('is_transship'):
conditions.append('is_transship_suspect = TRUE')
if params.get('spoofing'):
conditions.append('spoofing_score > 0.5')
where = ' AND '.join(conditions)
query = f"""
SELECT v.mmsi, v.risk_score, v.risk_level, v.zone, v.activity_state,
v.vessel_type, v.is_dark, v.gap_duration_min, v.spoofing_score,
v.cluster_id, v.cluster_size, v.dist_to_baseline_nm,
v.is_transship_suspect, v.transship_pair_mmsi,
fv.permit_no, fv.name_cn, fv.gear_code
FROM kcg.vessel_analysis_results v
LEFT JOIN kcg.fleet_vessels fv ON v.mmsi = fv.mmsi
WHERE {where}
ORDER BY v.risk_score DESC
LIMIT 30
"""
with kcgdb.get_conn() as conn:
with conn.cursor() as cur:
cur.execute(query, bind_params)
rows = cur.fetchall()
if not rows:
return '\n## 조회 결과\n해당 조건에 맞는 선박이 없습니다.\n'
# 결과를 간략 테이블로 구성 (토큰 절약)
lines = [f'\n## 조회 결과 ({len(rows)}척)']
lines.append('| MMSI | 점수 | 수역 | 활동 | 허가 | 다크 |')
lines.append('|---|---|---|---|---|---|')
for row in rows[:15]: # 최대 15척
mmsi, risk_score, risk_level, zone, activity, vtype, is_dark, gap, spoof, \
cid, csize, dist_nm, is_trans, trans_pair, permit, name_cn, gear = row
permit_str = 'Y' if permit else 'N'
dark_str = 'Y' if is_dark else '-'
lines.append(f'| {mmsi} | {risk_score} | {zone} | {activity} | {permit_str} | {dark_str} |')
return '\n'.join(lines)
except Exception as e:
logger.error('prequery execution failed: %s', e)
return f'\n(DB 조회 실패: {e})\n'
# ── LLM Tool Calling 응답 파싱 + 실행 ──
_TOOL_CALL_PATTERN = re.compile(
r'\{"tool"\s*:\s*"(\w+)"\s*,\s*"params"\s*:\s*(\{[^}]+\})\}',
)
def parse_tool_calls(llm_response: str) -> list[dict]:
"""LLM 응답에서 tool call JSON을 추출."""
calls = []
for match in _TOOL_CALL_PATTERN.finditer(llm_response):
try:
tool_name = match.group(1)
params = json.loads(match.group(2))
calls.append({'tool': tool_name, 'params': params})
except json.JSONDecodeError:
continue
return calls[:2] # 최대 2개
def execute_tool_call(call: dict) -> str:
"""단일 tool call 실행."""
tool = call.get('tool', '')
params = call.get('params', {})
if tool == 'query_vessels':
return execute_prequery(params)
if tool == 'query_vessel_detail':
mmsi = params.get('mmsi', '')
if mmsi:
from chat.context_builder import _build_vessel_detail
return _build_vessel_detail(mmsi)
return '(MMSI 미지정)'
if tool == 'query_fleet_group':
return _query_fleet_group(params)
if tool == 'query_vessel_history':
return _query_vessel_history(params)
if tool == 'query_vessel_static':
return _query_vessel_static(params)
if tool == 'get_knowledge':
return _get_knowledge(params)
return f'(알 수 없는 도구: {tool})'
def _get_knowledge(params: dict) -> str:
"""도메인 지식 섹션 조회."""
key = params.get('key', '')
if not key:
return '(key 미지정. 사용 가능: maritime_zones, fishing_agreement, algorithm_guide, response_guide, db_schema)'
from chat.domain_knowledge import get_knowledge_section
return get_knowledge_section(key)
def _query_fleet_group(params: dict) -> str:
"""선단/어구 그룹 조회."""
try:
from db import kcgdb
conditions = ["snapshot_time = (SELECT MAX(snapshot_time) FROM kcg.group_polygon_snapshots)"]
bind_params: list = []
if 'group_type' in params:
conditions.append('group_type = %s')
bind_params.append(params['group_type'])
if 'zone_id' in params:
conditions.append('zone_id = %s')
bind_params.append(params['zone_id'])
where = ' AND '.join(conditions)
query = f"""
SELECT group_type, group_key, group_label, member_count, zone_name, members
FROM kcg.group_polygon_snapshots
WHERE {where}
ORDER BY member_count DESC
LIMIT 20
"""
with kcgdb.get_conn() as conn:
with conn.cursor() as cur:
cur.execute(query, bind_params)
rows = cur.fetchall()
if not rows:
return '\n(해당 조건의 그룹 없음)\n'
lines = [f'\n## 그룹 조회 결과 ({len(rows)}건)']
lines.append('| 유형 | 키 | 라벨 | 선박수 | 수역 |')
lines.append('|---|---|---|---|---|')
for row in rows:
gtype, gkey, glabel, mcount, zname, members = row
lines.append(f'| {gtype} | {gkey} | {glabel or "-"} | {mcount} | {zname or "-"} |')
return '\n'.join(lines)
except Exception as e:
logger.error('fleet group query failed: %s', e)
return f'\n(그룹 조회 실패: {e})\n'
def _query_vessel_history(params: dict) -> str:
"""snpdb에서 선박 항적 이력 조회 (daily 집계)."""
try:
from db import snpdb
mmsi = params.get('mmsi', '')
days = min(params.get('days', 7), 30) # 최대 30일
if not mmsi:
return '(MMSI 미지정)'
query = """
SELECT time_bucket, distance_nm, avg_speed, max_speed, point_count,
start_position, end_position
FROM signal.t_vessel_tracks_daily
WHERE mmsi = %s AND time_bucket >= CURRENT_DATE - %s
ORDER BY time_bucket DESC
"""
with snpdb.get_conn() as conn:
with conn.cursor() as cur:
cur.execute(query, (mmsi, days))
rows = cur.fetchall()
if not rows:
return f'\n(MMSI {mmsi}: 최근 {days}일 항적 데이터 없음)\n'
lines = [f'\n## 항적 이력: {mmsi} (최근 {days}일)']
lines.append('| 날짜 | 이동거리(NM) | 평균속도 | 최대속도 | AIS포인트 |')
lines.append('|---|---|---|---|---|')
for row in rows:
dt, dist, avg_spd, max_spd, pts, start_pos, end_pos = row
lines.append(
f"| {dt} | {float(dist or 0):.1f} | {float(avg_spd or 0):.1f}kt "
f"| {float(max_spd or 0):.1f}kt | {pts or 0} |"
)
return '\n'.join(lines)
except Exception as e:
logger.error('vessel history query failed: %s', e)
return f'\n(항적 이력 조회 실패: {e})\n'
def _query_vessel_static(params: dict) -> str:
"""snpdb에서 선박 정적정보 + 변경 이력 조회."""
try:
from db import snpdb
mmsi = params.get('mmsi', '')
limit = min(params.get('limit', 10), 24)
if not mmsi:
return '(MMSI 미지정)'
query = """
SELECT time_bucket, name, vessel_type, length, width, draught,
destination, status, callsign, imo
FROM signal.t_vessel_static
WHERE mmsi = %s
ORDER BY time_bucket DESC
LIMIT %s
"""
with snpdb.get_conn() as conn:
with conn.cursor() as cur:
cur.execute(query, (mmsi, limit))
rows = cur.fetchall()
if not rows:
return f'\n(MMSI {mmsi}: 정적정보 없음)\n'
# 최신 정보
latest = rows[0]
lines = [f'\n## 선박 정적정보: {mmsi}']
lines.append(f'- 선명: {latest[1] or "N/A"}')
lines.append(f'- 선종: {latest[2] or "N/A"}')
lines.append(f'- 제원: L={latest[3] or 0}m × W={latest[4] or 0}m, 흘수={latest[5] or 0}m')
lines.append(f'- 목적지: {latest[6] or "N/A"}')
lines.append(f'- 상태: {latest[7] or "N/A"}')
lines.append(f'- 호출부호: {latest[8] or "N/A"}, IMO: {latest[9] or "N/A"}')
# 변경 이력 감지
changes = []
for i in range(len(rows) - 1):
curr, prev = rows[i], rows[i + 1]
diffs = []
if curr[1] != prev[1]:
diffs.append(f'선명: {prev[1]}→{curr[1]}')
if curr[6] != prev[6]:
diffs.append(f'목적지: {prev[6]}→{curr[6]}')
if curr[7] != prev[7]:
diffs.append(f'상태: {prev[7]}→{curr[7]}')
if diffs:
changes.append(f'- {curr[0].strftime("%m/%d %H:%M")}: {", ".join(diffs)}')
if changes:
lines.append(f'\n### 변경 이력 (최근 {len(changes)}건)')
lines.extend(changes[:10])
return '\n'.join(lines)
except Exception as e:
logger.error('vessel static query failed: %s', e)
return f'\n(정적정보 조회 실패: {e})\n'

10
prediction/config.py
파일 보기

@ -31,6 +31,16 @@ class Settings(BaseSettings):
MMSI_PREFIX: str = '412'
MIN_TRAJ_POINTS: int = 100
# Ollama (LLM)
OLLAMA_BASE_URL: str = 'http://localhost:11434'
OLLAMA_MODEL: str = 'qwen3:14b' # CPU-only: 14b 권장, GPU 있으면 32b
OLLAMA_TIMEOUT_SEC: int = 300
# Redis
REDIS_HOST: str = 'localhost'
REDIS_PORT: int = 6379
REDIS_PASSWORD: str = ''
# 로깅
LOG_LEVEL: str = 'INFO'

114
prediction/db/kcgdb.py
파일 보기

@ -21,7 +21,7 @@ def init_pool():
global _pool
_pool = pool.ThreadedConnectionPool(
minconn=1,
maxconn=3,
maxconn=5,
host=settings.KCGDB_HOST,
port=settings.KCGDB_PORT,
dbname=settings.KCGDB_NAME,
@ -195,6 +195,118 @@ def save_group_snapshots(snapshots: list[dict]) -> int:
return 0
def fetch_analysis_summary() -> dict:
"""최근 1시간 분석 결과 요약 (채팅 컨텍스트용)."""
try:
with get_conn() as conn:
with conn.cursor() as cur:
# 위험도 분포
cur.execute("""
SELECT risk_level, COUNT(*) FROM vessel_analysis_results
WHERE analyzed_at > NOW() - INTERVAL '1 hour'
GROUP BY risk_level
""")
risk_dist = {row[0]: row[1] for row in cur.fetchall()}
# 수역별 분포
cur.execute("""
SELECT zone, COUNT(*) FROM vessel_analysis_results
WHERE analyzed_at > NOW() - INTERVAL '1 hour'
GROUP BY zone
""")
zone_dist = {row[0]: row[1] for row in cur.fetchall()}
# 다크/스푸핑/환적 카운트
cur.execute("""
SELECT
COUNT(*) FILTER (WHERE is_dark = TRUE) AS dark_count,
COUNT(*) FILTER (WHERE spoofing_score > 0.5) AS spoofing_count,
COUNT(*) FILTER (WHERE is_transship_suspect = TRUE) AS transship_count
FROM vessel_analysis_results
WHERE analyzed_at > NOW() - INTERVAL '1 hour'
""")
row = cur.fetchone()
result = {
'risk_distribution': {**risk_dist, **zone_dist},
'dark_count': row[0] if row else 0,
'spoofing_count': row[1] if row else 0,
'transship_count': row[2] if row else 0,
}
return result
except Exception as e:
logger.error('fetch_analysis_summary failed: %s', e)
return {'risk_distribution': {}, 'dark_count': 0, 'spoofing_count': 0, 'transship_count': 0}
def fetch_recent_high_risk(limit: int = 10) -> list[dict]:
"""위험도 상위 N척 선박 상세 (채팅 컨텍스트용)."""
try:
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute("""
SELECT mmsi, risk_score, risk_level, zone, is_dark,
is_transship_suspect, activity_state, spoofing_score
FROM vessel_analysis_results
WHERE analyzed_at > NOW() - INTERVAL '1 hour'
ORDER BY risk_score DESC
LIMIT %s
""", (limit,))
rows = cur.fetchall()
result = []
for row in rows:
result.append({
'mmsi': row[0],
'name': row[0], # vessel_store에서 이름 조회 필요시 보강
'risk_score': row[1],
'risk_level': row[2],
'zone': row[3],
'is_dark': row[4],
'is_transship': row[5],
'activity_state': row[6],
'spoofing_score': float(row[7]) if row[7] else 0.0,
})
return result
except Exception as e:
logger.error('fetch_recent_high_risk failed: %s', e)
return []
def fetch_polygon_summary() -> dict:
"""최신 그룹 폴리곤 요약 (채팅 컨텍스트용)."""
try:
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute("""
SELECT group_type, COUNT(*), SUM(member_count)
FROM kcg.group_polygon_snapshots
WHERE snapshot_time = (
SELECT MAX(snapshot_time) FROM kcg.group_polygon_snapshots
)
GROUP BY group_type
""")
rows = cur.fetchall()
result = {
'fleet_count': 0, 'fleet_members': 0,
'gear_in_zone': 0, 'gear_out_zone': 0,
}
for row in rows:
gtype, count, members = row[0], row[1], row[2] or 0
if gtype == 'FLEET':
result['fleet_count'] = count
result['fleet_members'] = members
elif gtype == 'GEAR_IN_ZONE':
result['gear_in_zone'] = count
elif gtype == 'GEAR_OUT_ZONE':
result['gear_out_zone'] = count
return result
except Exception as e:
logger.error('fetch_polygon_summary failed: %s', e)
return {'fleet_count': 0, 'fleet_members': 0, 'gear_in_zone': 0, 'gear_out_zone': 0}
def cleanup_group_snapshots(days: int = 7) -> int:
"""오래된 그룹 폴리곤 스냅샷 삭제."""
try:

9
prediction/env.example
파일 보기

@ -21,5 +21,14 @@ TRAJECTORY_HOURS=6
MMSI_PREFIX=412
MIN_TRAJ_POINTS=100
# Ollama (LLM)
OLLAMA_BASE_URL=http://localhost:11434
OLLAMA_MODEL=qwen3:32b
# Redis
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_PASSWORD=
# 로깅
LOG_LEVEL=INFO

6
prediction/main.py
파일 보기

@ -39,10 +39,14 @@ async def lifespan(application: FastAPI):
app = FastAPI(
title='KCG Prediction Service',
version='2.0.0',
version='2.1.0',
lifespan=lifespan,
)
# AI 해양분석 채팅 라우터
from chat.router import router as chat_router
app.include_router(chat_router)
@app.get('/health')
def health_check():

2
prediction/requirements.txt
파일 보기

@ -7,3 +7,5 @@ pandas>=2.2
scikit-learn>=1.5
apscheduler>=3.10
shapely>=2.0
httpx>=0.27
redis>=5.0

46
prediction/scheduler.py
파일 보기

@ -258,6 +258,52 @@ def run_analysis_cycle():
upserted = kcgdb.upsert_results(results)
kcgdb.cleanup_old(hours=48)
# 8. Redis에 분석 컨텍스트 캐싱 (채팅용)
try:
from chat.cache import cache_analysis_context
results_map = {r.mmsi: r for r in results}
risk_dist = {}
zone_dist = {}
dark_count = 0
spoofing_count = 0
transship_count = 0
top_risk_list = []
for r in results:
risk_dist[r.risk_level] = risk_dist.get(r.risk_level, 0) + 1
zone_dist[r.zone] = zone_dist.get(r.zone, 0) + 1
if r.is_dark:
dark_count += 1
if r.spoofing_score > 0.5:
spoofing_count += 1
if r.is_transship_suspect:
transship_count += 1
top_risk_list.append({
'mmsi': r.mmsi,
'name': vessel_store.get_vessel_info(r.mmsi).get('name', r.mmsi),
'risk_score': r.risk_score,
'risk_level': r.risk_level,
'zone': r.zone,
'is_dark': r.is_dark,
'is_transship': r.is_transship_suspect,
'activity_state': r.activity_state,
})
top_risk_list.sort(key=lambda x: x['risk_score'], reverse=True)
cache_analysis_context({
'vessel_stats': vessel_store.stats(),
'risk_distribution': {**risk_dist, **zone_dist},
'dark_count': dark_count,
'spoofing_count': spoofing_count,
'transship_count': transship_count,
'top_risk_vessels': top_risk_list[:10],
'polygon_summary': kcgdb.fetch_polygon_summary(),
})
except Exception as e:
logger.warning('failed to cache analysis context for chat: %s', e)
elapsed = round(time.time() - start, 2)
_last_run['duration_sec'] = elapsed
_last_run['vessel_count'] = len(results)