diff --git a/deploy/docker-compose-ollama.yml b/deploy/docker-compose-ollama.yml
new file mode 100644
index 0000000..daf89c4
--- /dev/null
+++ b/deploy/docker-compose-ollama.yml
@@ -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
diff --git a/deploy/nginx-kcg.conf b/deploy/nginx-kcg.conf
index 05f0b60..8626455 100644
--- a/deploy/nginx-kcg.conf
+++ b/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/;
diff --git a/frontend/src/components/common/EventLog.tsx b/frontend/src/components/common/EventLog.tsx
index 0caf308..2938e49 100644
--- a/frontend/src/components/common/EventLog.tsx
+++ b/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>
   );
diff --git a/frontend/src/components/korea/AiChatPanel.tsx b/frontend/src/components/korea/AiChatPanel.tsx
index 611f137..28669c7 100644
--- a/frontend/src/components/korea/AiChatPanel.tsx
+++ b/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(),
-      };
-      setMessages(prev => [...prev, assistantMsg]);
+      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,
+              };
+              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, {
-        role: 'assistant',
-        content: `오류: ${err instanceof Error ? err.message : 'AI 서버 연결 실패'}. Ollama가 실행 중인지 확인하세요.`,
-        timestamp: Date.now(),
-      }]);
+      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 서버 연결 실패'}`,
+          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={{
-      borderTop: '1px solid rgba(168,85,247,0.2)',
-      marginTop: 8,
+      ...(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) => (
-              <div
-                key={i}
-                style={{
-                  alignSelf: msg.role === 'user' ? 'flex-end' : 'flex-start',
-                  maxWidth: '85%',
-                  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',
-                  padding: '6px 8px',
-                  fontSize: 10,
-                  color: '#e2e8f0',
-                  lineHeight: 1.5,
-                  whiteSpace: 'pre-wrap',
-                  wordBreak: 'break-word',
-                }}
-              >
-                {msg.content}
-              </div>
-            ))}
-            {isLoading && (
+            {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: 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',
+                  }}>
+                    {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>
+                </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',
diff --git a/frontend/vite.config.ts b/frontend/vite.config.ts
index 1826b7d..c0c411e 100644
--- a/frontend/vite.config.ts
+++ b/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,
diff --git a/prediction/chat/__init__.py b/prediction/chat/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/prediction/chat/cache.py b/prediction/chat/cache.py
new file mode 100644
index 0000000..662594a
--- /dev/null
+++ b/prediction/chat/cache.py
@@ -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)
diff --git a/prediction/chat/context_builder.py b/prediction/chat/context_builder.py
new file mode 100644
index 0000000..953116d
--- /dev/null
+++ b/prediction/chat/context_builder.py
@@ -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)
diff --git a/prediction/chat/domain_knowledge.py b/prediction/chat/domain_knowledge.py
new file mode 100644
index 0000000..f7eb7fc
--- /dev/null
+++ b/prediction/chat/domain_knowledge.py
@@ -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
diff --git a/prediction/chat/router.py b/prediction/chat/router.py
new file mode 100644
index 0000000..93542f6
--- /dev/null
+++ b/prediction/chat/router.py
@@ -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}
diff --git a/prediction/chat/tools.py b/prediction/chat/tools.py
new file mode 100644
index 0000000..766c260
--- /dev/null
+++ b/prediction/chat/tools.py
@@ -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'
diff --git a/prediction/config.py b/prediction/config.py
index 7414022..29d24cf 100644
--- a/prediction/config.py
+++ b/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'
 
diff --git a/prediction/db/kcgdb.py b/prediction/db/kcgdb.py
index ba4282f..90bf5bb 100644
--- a/prediction/db/kcgdb.py
+++ b/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:
diff --git a/prediction/env.example b/prediction/env.example
index c771eac..5a915ed 100644
--- a/prediction/env.example
+++ b/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
diff --git a/prediction/main.py b/prediction/main.py
index 4f717dd..2e1a9dc 100644
--- a/prediction/main.py
+++ b/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():
diff --git a/prediction/requirements.txt b/prediction/requirements.txt
index 46d3abc..7eedee5 100644
--- a/prediction/requirements.txt
+++ b/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
diff --git a/prediction/scheduler.py b/prediction/scheduler.py
index 313ffdf..d463098 100644
--- a/prediction/scheduler.py
+++ b/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)