홈랩 AI 서버 가이드 5편: Qdrant 벡터DB와 Airflow 및 GPU 모니터링

services:

  # ── Qdrant 벡터 데이터베이스 ──────────────────────
  qdrant:
    image: qdrant/qdrant:latest
    container_name: qdrant
    restart: unless-stopped
    ports:
      - "6333:6333"   # HTTP REST API
      - "6334:6334"   # gRPC (고성능 조회용)
    volumes:
      - qdrant_storage:/qdrant/storage
      - ./qdrant_config.yaml:/qdrant/config/production.yaml
    environment:
      - QDRANT__SERVICE__API_KEY=your_very_secure_api_key_here
      - QDRANT__SERVICE__ENABLE_TLS=false
      - QDRANT__LOG_LEVEL=INFO
    networks:
      - ai_net

  # ── PostgreSQL (n8n + Dify 공유 DB) ──────────────
  postgresql:
    image: pgvector/pgvector:pg16  # pgvector 확장 포함
    container_name: postgresql
    restart: unless-stopped
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data
      - ./init.sql:/docker-entrypoint-initdb.d/init.sql
    environment:
      - POSTGRES_USER=aiserver
      - POSTGRES_PASSWORD=secure_password_here
      - POSTGRES_DB=aiserver_main
    networks:
      - ai_net

  # ── Redis (캐시 + 세션) ───────────────────────────
  redis:
    image: redis:7-alpine
    container_name: redis
    restart: unless-stopped
    ports:
      - "6379:6379"
    command: redis-server --requirepass your_redis_password
    volumes:
      - redis_data:/data
    networks:
      - ai_net

volumes:
  qdrant_storage:
  postgres_data:
  redis_data:

networks:
  ai_net:
    external: true
    name: ai_network

storage:
  storage_path: /qdrant/storage
  snapshots_path: /qdrant/snapshots

  # HNSW 인덱스 전역 기본값
  hnsw_index:
    m: 16                  # 연결 수 (높을수록 정확, 메모리↑)
    ef_construct: 200      # 빌드 시 탐색 범위 (높을수록 정확)
    full_scan_threshold: 10000  # 이 수 이하면 FLAT 검색

  # 성능 최적화
  on_disk_payload: false   # 페이로드 메모리 유지 (검색 속도↑)

service:
  host: 0.0.0.0
  http_port: 6333
  grpc_port: 6334
  max_request_size_mb: 64

  # 요청 제한 (DoS 방어)
  max_workers: 0           # 0 = CPU 코어 수 자동

telemetry_disabled: true   # 원격 측정 비활성화 (프라이버시)

from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance, VectorParams, PointStruct,
    Filter, FieldCondition, MatchValue,
    HnswConfigDiff, OptimizersConfigDiff,
    PayloadSchemaType
)
from sentence_transformers import SentenceTransformer
import uuid

# 클라이언트 초기화
client = QdrantClient(
    host="192.168.1.253",
    port=6333,
    api_key="your_very_secure_api_key_here",
    timeout=60
)
embedder = SentenceTransformer("BAAI/bge-m3")

COLLECTION = "ai_knowledge_base"
VECTOR_DIM  = 1024  # bge-m3의 출력 차원

## ── 컬렉션 생성 (HNSW 최적화) ─────────────────────
client.recreate_collection(
    collection_name=COLLECTION,
    vectors_config=VectorParams(
        size=VECTOR_DIM,
        distance=Distance.COSINE,    # 코사인 유사도
        on_disk=False,               # 메모리 상주 (빠름)
    ),
    hnsw_config=HnswConfigDiff(
        m=16,
        ef_construct=200,
        full_scan_threshold=10_000,
        on_disk=False,
    ),
    optimizers_config=OptimizersConfigDiff(
        indexing_threshold=20_000,   # 이 수 이하면 인덱스 지연
        memmap_threshold=50_000,     # 메모리맵 전환 임계값
    ),
)

# 페이로드 인덱스 생성 (필터 성능 극대화)
client.create_payload_index(
    collection_name=COLLECTION,
    field_name="category",
    field_schema=PayloadSchemaType.KEYWORD,  # 정확히 일치 검색
)
client.create_payload_index(
    collection_name=COLLECTION,
    field_name="created_at",
    field_schema=PayloadSchemaType.DATETIME,
)

## ── 벡터 삽입 (배치 처리) ──────────────────────────
def upsert_documents(docs: list[dict], batch_size: int = 100):
    """문서 목록을 배치로 임베딩 후 Qdrant 삽입"""
    for i in range(0, len(docs), batch_size):
        batch    = docs[i:i + batch_size]
        texts    = [d["content"] for d in batch]
        vectors  = embedder.encode(texts, normalize_embeddings=True).tolist()

        points = [
            PointStruct(
                id=str(uuid.uuid4()),
                vector=vectors[j],
                payload={
                    "title":      batch[j].get("title", ""),
                    "content":    batch[j]["content"][:1000],  # 페이로드 크기 제한
                    "source":     batch[j].get("source", ""),
                    "category":   batch[j].get("category", "general"),
                    "created_at": batch[j].get("created_at", ""),
                    "chunk_index": batch[j].get("chunk_index", 0),
                }
            )
            for j in range(len(batch))
        ]
        client.upsert(collection_name=COLLECTION, points=points)
        print(f"✅ 삽입 완료: {i+len(batch)}/{len(docs)}")

## ── 벡터 검색 (필터 + 유사도 하이브리드) ──────────
def search(query: str, category: str = None, top_k: int = 5) -> list:
    query_vec = embedder.encode([query], normalize_embeddings=True).tolist()[0]

    # 카테고리 필터 조건
    search_filter = None
    if category:
        search_filter = Filter(
            must=[FieldCondition(
                key="category",
                match=MatchValue(value=category)
            )]
        )

    results = client.search(
        collection_name=COLLECTION,
        query_vector=query_vec,
        query_filter=search_filter,
        limit=top_k,
        with_payload=True,
        score_threshold=0.5,    # 이 점수 이하 결과 제외
    )
    return [
        {
            "score":   round(r.score, 4),
            "title":   r.payload.get("title"),
            "content": r.payload.get("content"),
            "source":  r.payload.get("source"),
        }
        for r in results
    ]

# 사용 예시
results = search("Docker 컨테이너 네트워크 설정", category="tech", top_k=5)
for r in results:
    print(f"[{r['score']}] {r['title']}\n{r['content'][:100]}...")

## ── 스냅샷 백업 ─────────────────────────────────────
snapshot_info = client.create_snapshot(collection_name=COLLECTION)
print(f"스냅샷 생성: {snapshot_info.name}")

from qdrant_client.models import (
    NamedVector, NamedSparseVector,
    SparseVector, SparseVectorParams,
    VectorParams, Distance
)
from fastembed import SparseTextEmbedding  # pip install fastembed

# Hybrid 컬렉션 생성 (Dense + Sparse 동시 저장)
client.recreate_collection(
    collection_name="hybrid_knowledge",
    vectors_config={
        "dense": VectorParams(size=1024, distance=Distance.COSINE),
    },
    sparse_vectors_config={
        "sparse": SparseVectorParams(),  # SPLADE/BM25용 희소 벡터
    }
)

sparse_model = SparseTextEmbedding(model_name="prithivida/Splade_PP_en_v1")

def upsert_hybrid(docs: list[dict]):
    """Dense + Sparse 동시 임베딩 후 삽입"""
    texts = [d["content"] for d in docs]

    dense_vecs  = embedder.encode(texts, normalize_embeddings=True)
    sparse_vecs = list(sparse_model.embed(texts))

    points = []
    for i, doc in enumerate(docs):
        sv = sparse_vecs[i]
        points.append(PointStruct(
            id=str(uuid.uuid4()),
            vector={
                "dense":  dense_vecs[i].tolist(),
                "sparse": SparseVector(
                    indices=sv.indices.tolist(),
                    values=sv.values.tolist()
                )
            },
            payload={"content": doc["content"], "title": doc.get("title", "")}
        ))
    client.upsert(collection_name="hybrid_knowledge", points=points)

def hybrid_search(query: str, top_k: int = 5):
    """Dense + Sparse 결합 검색 (RRF 랭킹)"""
    dense_vec   = embedder.encode([query], normalize_embeddings=True)[0].tolist()
    sparse_vecs = list(sparse_model.embed([query]))
    sv = sparse_vecs[0]

    from qdrant_client.models import Prefetch, FusionQuery, Fusion

    results = client.query_points(
        collection_name="hybrid_knowledge",
        prefetch=[
            Prefetch(query=dense_vec, using="dense",  limit=top_k * 2),
            Prefetch(
                query=SparseVector(
                    indices=sv.indices.tolist(),
                    values=sv.values.tolist()
                ),
                using="sparse", limit=top_k * 2
            ),
        ],
        query=FusionQuery(fusion=Fusion.RRF),  # Reciprocal Rank Fusion
        limit=top_k,
        with_payload=True
    )
    return results.points

from langchain_text_splitters import (
    RecursiveCharacterTextSplitter,
    MarkdownHeaderTextSplitter
)
from langchain_ollama import ChatOllama, OllamaEmbeddings
from langchain_community.vectorstores import Qdrant as LCQdrant
from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain.schema import Document
import re

## ══ 1단계: 문서 처리 — Semantic 청킹 ══════════════

def smart_chunk_markdown(text: str, source: str) -> list[Document]:
    """마크다운 구조 기반 계층적 청킹"""

    # 1차: 마크다운 헤더 기준 분할 (대분류)
    md_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=[
        ("#",  "h1"),
        ("##", "h2"),
        ("###","h3"),
    ])
    sections = md_splitter.split_text(text)

    # 2차: 섹션별 세밀 청킹 (실제 벡터화 단위)
    char_splitter = RecursiveCharacterTextSplitter(
        chunk_size=800,
        chunk_overlap=100,
        separators=["\n\n", "\n", "。", ". ", " ", ""],
    )
    chunks = []
    for sec in sections:
        sub_chunks = char_splitter.split_documents([sec])
        for i, chunk in enumerate(sub_chunks):
            chunk.metadata.update({
                "source":      source,
                "chunk_index": i,
                "parent_id":   sec.metadata.get("h1", "") + sec.metadata.get("h2", ""),
            })
            chunks.append(chunk)
    return chunks

## ══ 2단계: Parent-Child 리트리버 ══════════════════

# 부모 청크: 큰 맥락 보존용 (1500 토큰)
parent_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1500, chunk_overlap=200
)
# 자식 청크: 정밀 검색용 (300 토큰)
child_splitter = RecursiveCharacterTextSplitter(
    chunk_size=300, chunk_overlap=50
)

embeddings = OllamaEmbeddings(
    base_url="http://192.168.1.253:11434",
    model="nomic-embed-text"
)

# 자식 청크는 벡터 DB에, 부모 청크는 로컬 스토어에
vectorstore    = LCQdrant.from_existing_collection(
    embedding=embeddings,
    collection_name="rag_children",
    url="http://192.168.1.253:6333",
    api_key="your_api_key"
)
docstore = InMemoryStore()   # 프로덕션에서는 Redis로 교체

retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=docstore,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
    search_kwargs={"k": 10},  # 후보를 충분히 가져온 뒤 재랭킹
)

## ══ 3단계: 쿼리 확장 + 재랭킹 ════════════════════

llm = ChatOllama(
    base_url="http://192.168.1.253:11434",
    model="qwen2.5:14b",
    temperature=0
)

def expand_query(query: str) -> list[str]:
    """LLM으로 쿼리 재작성 — 다양한 관점에서 검색 강화"""
    prompt = f"""다음 질문을 서로 다른 3가지 방식으로 재작성해줘. 
    같은 의미지만 다른 표현과 관점을 사용해.
    각 질문을 줄바꿈으로 구분해서 3개만 출력해.
    
    원본 질문: {query}"""
    result   = llm.invoke(prompt)
    rewrites = [q.strip() for q in result.content.split('\n') if q.strip()]
    return [query] + rewrites[:3]  # 원본 + 3가지 재작성

def cross_encoder_rerank(query: str, docs: list, top_n: int = 5) -> list:
    """Cross-Encoder로 최종 재랭킹 (더 정밀한 관련성 평가)"""
    from sentence_transformers import CrossEncoder
    reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

    pairs  = [(query, doc.page_content) for doc in docs]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(docs, scores), key=lambda x: x[1], reverse=True)
    return [doc for doc, _ in ranked[:top_n]]

## ══ 4단계: 최종 RAG Chain ═════════════════════════

def rag_answer(question: str) -> dict:
    """완전한 Advanced RAG 실행"""

    # 쿼리 확장
    queries     = expand_query(question)

    # 다중 쿼리로 검색 (더 많은 관련 문서 수집)
    all_docs = []
    seen_ids = set()
    for q in queries:
        docs = retriever.get_relevant_documents(q)
        for doc in docs:
            doc_id = hash(doc.page_content)
            if doc_id not in seen_ids:
                all_docs.append(doc)
                seen_ids.add(doc_id)

    # 재랭킹으로 최종 상위 5개 선택
    top_docs = cross_encoder_rerank(question, all_docs, top_n=5)

    # 컨텍스트 압축 (LLM으로 관련 부분만 추출)
    context = "\n\n---\n\n".join([d.page_content for d in top_docs])

    # 최종 답변 생성
    prompt = f"""당신은 정확하고 신뢰할 수 있는 AI 어시스턴트입니다.
아래 제공된 컨텍스트만을 사용해서 질문에 답하세요.
컨텍스트에 없는 내용은 절대 추측하거나 지어내지 마세요.
모르는 경우 "제공된 문서에서 해당 정보를 찾을 수 없습니다."라고 말하세요.

컨텍스트:
{context}

질문: {question}

답변:"""

    answer = llm.invoke(prompt)
    return {
        "answer":    answer.content,
        "sources":   [d.metadata.get("source", "") for d in top_docs],
        "num_docs":  len(top_docs),
        "queries":   queries,
    }

# 실행
result = rag_answer("Docker 컨테이너에서 GPU를 사용하려면 어떻게 설정해야 하나요?")
print(result["answer"])
print(f"\n참고 출처: {result['sources']}")

from ragas import evaluate
from ragas.metrics import (
    faithfulness, answer_relevancy,
    context_precision, context_recall
)
from datasets import Dataset

# 평가 데이터셋 (질문 + 정답 + RAG 답변 + 검색 컨텍스트)
eval_data = {
    "question": [
        "Docker에서 GPU를 사용하려면 무엇이 필요한가?",
        "Qdrant의 HNSW 파라미터 m의 역할은?",
    ],
    "answer": [
        "NVIDIA Container Toolkit을 설치하고 docker run --gpus all 옵션을 사용해야 합니다.",
        "m 파라미터는 HNSW 그래프에서 각 노드의 연결 수를 결정하며, 높을수록 검색 정확도가 올라가지만 메모리를 더 많이 사용합니다.",
    ],
    "contexts": [
        ["NVIDIA Container Toolkit은 Docker 컨테이너에서 GPU를 사용하기 위한 런타임입니다..."],
        ["HNSW 알고리즘의 m 파라미터는 각 노드가 유지하는 최대 연결 수입니다..."],
    ],
    "ground_truth": [
        "NVIDIA Container Toolkit 설치 필요",
        "m은 각 노드의 연결 수로 정확도와 메모리 트레이드오프",
    ],
}

dataset = Dataset.from_dict(eval_data)
result  = evaluate(
    dataset,
    metrics=[
        faithfulness,
        answer_relevancy,
        context_precision,
        context_recall,
    ],
)
print(result)
# 출력 예시:
# {'faithfulness': 0.89, 'answer_relevancy': 0.84,
#  'context_precision': 0.76, 'context_recall': 0.81}

services:

  prometheus:
    image: prom/prometheus:latest
    container_name: prometheus
    restart: unless-stopped
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
      - prometheus_data:/prometheus
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'
      - '--storage.tsdb.retention.time=30d'
      - '--web.enable-lifecycle'
    networks:
      - ai_net

  grafana:
    image: grafana/grafana:latest
    container_name: grafana
    restart: unless-stopped
    ports:
      - "3002:3000"
    volumes:
      - grafana_data:/var/lib/grafana
      - ./grafana/dashboards:/etc/grafana/provisioning/dashboards
      - ./grafana/datasources:/etc/grafana/provisioning/datasources
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=secure_grafana_password
      - GF_USERS_ALLOW_SIGN_UP=false
      - GF_SMTP_ENABLED=true                    # 이메일 알림
      - GF_SMTP_HOST=smtp.gmail.com:587
    networks:
      - ai_net

  # NVIDIA GPU 메트릭 수집기
  nvidia-exporter:
    image: utkuozdemir/nvidia_gpu_exporter:1.2.0
    container_name: nvidia_exporter
    restart: unless-stopped
    ports:
      - "9835:9835"
    devices:
      - /dev/nvidiactl
      - /dev/nvidia0
    volumes:
      - /usr/lib/x86_64-linux-gnu/libnvidia-ml.so.1:/usr/lib/x86_64-linux-gnu/libnvidia-ml.so.1
    networks:
      - ai_net

  # 시스템 메트릭 수집기 (CPU/RAM/Disk/Network)
  node-exporter:
    image: prom/node-exporter:latest
    container_name: node_exporter
    restart: unless-stopped
    ports:
      - "9100:9100"
    volumes:
      - /proc:/host/proc:ro
      - /sys:/host/sys:ro
      - /:/rootfs:ro
    command:
      - '--path.procfs=/host/proc'
      - '--path.sysfs=/host/sys'
      - '--collector.filesystem.ignored-mount-points=^/(sys|proc|dev|host|etc)($$|/)'
    networks:
      - ai_net

  # Docker 컨테이너 메트릭
  cadvisor:
    image: gcr.io/cadvisor/cadvisor:latest
    container_name: cadvisor
    restart: unless-stopped
    ports:
      - "8080:8080"
    volumes:
      - /:/rootfs:ro
      - /var/run:/var/run:ro
      - /sys:/sys:ro
      - /var/lib/docker/:/var/lib/docker:ro
    networks:
      - ai_net

volumes:
  prometheus_data:
  grafana_data:

networks:
  ai_net:
    external: true
    name: ai_network

global:
  scrape_interval:     15s
  evaluation_interval: 15s

rule_files:
  - "alert_rules.yml"

alerting:
  alertmanagers:
    - static_configs:
        - targets: ['alertmanager:9093']

scrape_configs:
  - job_name: 'nvidia-gpu'
    static_configs:
      - targets: ['nvidia-exporter:9835']

  - job_name: 'node'
    static_configs:
      - targets: ['node-exporter:9100']

  - job_name: 'docker-containers'
    static_configs:
      - targets: ['cadvisor:8080']

  - job_name: 'ollama'
    metrics_path: '/api/metrics'
    static_configs:
      - targets: ['ollama:11434']

groups:
  - name: ai_server_alerts
    rules:

      # GPU 온도 경고 (85도 이상)
      - alert: GPUTemperatureHigh
        expr: nvidia_smi_temperature_gpu > 85
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "GPU 온도 위험 수준"
          description: "GPU 온도가 {{ $value }}°C에 도달했습니다. 냉각 확인 필요."

      # VRAM 사용률 95% 이상
      - alert: GPUMemoryCritical
        expr: (nvidia_smi_memory_used_bytes / nvidia_smi_memory_total_bytes) * 100 > 95
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "VRAM 거의 소진"
          description: "VRAM 사용률이 {{ $value }}%입니다."

      # RAM 사용률 90% 이상
      - alert: HighMemoryUsage
        expr: (1 - (node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes)) * 100 > 90
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "시스템 RAM 부족"
          description: "메모리 사용률이 {{ $value }}%입니다."

      # 디스크 사용률 85% 이상
      - alert: DiskSpaceLow
        expr: (1 - (node_filesystem_free_bytes{mountpoint="/"} / node_filesystem_size_bytes{mountpoint="/"})) * 100 > 85
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "디스크 공간 부족"
          description: "루트 파티션 사용률이 {{ $value }}%입니다."

      # 컨테이너 재시작 감지
      - alert: ContainerRestarting
        expr: rate(container_last_seen{name!=""}[5m]) > 0 AND changes(container_last_seen{name!=""}[10m]) > 2
        labels:
          severity: warning
        annotations:
          summary: "컨테이너 {{ $labels.name }} 반복 재시작 감지"

#!/bin/bash
# AI 서버 GPU 진단 스크립트
echo "=== GPU 드라이버 확인 ==="
nvidia-smi && echo "✅ 드라이버 정상" || echo "❌ 드라이버 없음"

echo "=== NVIDIA Container Toolkit 확인 ==="
nvidia-ctk --version && echo "✅ Toolkit 설치됨" || echo "❌ Toolkit 미설치"

echo "=== Docker 런타임 설정 확인 ==="
cat /etc/docker/daemon.json | python3 -m json.tool

echo "=== Docker GPU 직접 테스트 ==="
docker run --rm --gpus all ubuntu nvidia-smi && echo "✅ GPU 정상" || echo "❌ GPU 실패"

echo "=== 해결 시도 ==="
sudo nvidia-ctk runtime configure --runtime=docker
sudo systemctl restart docker
echo "Docker 재시작 완료. 다시 테스트하세요."

#!/bin/bash
# ~/ai-server/scripts/health_check.sh

echo "╔═══════════════════════════════════════╗"
echo "║       AI 서버 헬스체크 리포트          ║"
echo "╚═══════════════════════════════════════╝"
echo ""

# GPU 상태
echo "🎮 GPU 상태"
nvidia-smi --query-gpu=name,temperature.gpu,utilization.gpu,memory.used,memory.total \
  --format=csv,noheader | awk -F',' '{
    printf "  모델: %s | 온도: %s°C | GPU: %s | VRAM: %s / %s\n",
    $1, $2, $3, $4, $5
}'
echo ""

# 컨테이너 상태
echo "🐳 컨테이너 상태"
docker ps --format "  {{.Names}}\t{{.Status}}\t{{.Ports}}" | column -t
echo ""

# 디스크 사용량
echo "💾 디스크 사용량"
df -h / /home 2>/dev/null | grep -v Filesystem | awk '{printf "  %s: %s / %s (%s)\n", $6, $3, $2, $5}'
echo ""

# 메모리
echo "🧠 메모리"
free -h | awk '/^Mem/{printf "  사용: %s / 전체: %s (여유: %s)\n", $3, $2, $7}'
echo ""

# 핵심 서비스 응답 확인
echo "🌐 서비스 응답 확인"
services=(
  "Ollama:http://localhost:11434/api/tags"
  "OpenWebUI:http://localhost:3000/health"
  "Qdrant:http://localhost:6333/healthz"
  "n8n:http://localhost:5678/healthz"
)
for svc in "${services[@]}"; do
  name="${svc%%:*}"
  url="${svc#*:}"
  code=$(curl -s -o /dev/null -w "%{http_code}" --connect-timeout 3 "$url")
  [ "$code" = "200" ] && status="✅" || status="❌"
  echo "  $status $name ($code)"
done

echo ""
echo "완료: $(date '+%Y-%m-%d %H:%M:%S')"

import asyncio, httpx, feedparser
from datetime import datetime, timezone
from typing import AsyncGenerator
import psycopg2

DB_CONFIG = {
    "host": "192.168.1.253", "port": 5432,
    "dbname": "aiserver_main",
    "user": "aiserver", "password": "secure_password"
}

async def collect_rss_feeds(feeds: list[str]) -> list[dict]:
    """RSS/Atom 피드 병렬 수집"""
    async def fetch_feed(url: str) -> list[dict]:
        try:
            async with httpx.AsyncClient(timeout=15) as client:
                resp = await client.get(url)
            parsed = feedparser.parse(resp.text)
            items = []
            for entry in parsed.entries[:20]:
                items.append({
                    "title":       entry.get("title", ""),
                    "content":     entry.get("summary", entry.get("description", "")),
                    "url":         entry.get("link", ""),
                    "published":   entry.get("published", datetime.now(timezone.utc).isoformat()),
                    "source":      parsed.feed.get("title", url),
                    "source_type": "rss",
                })
            return items
        except Exception as e:
            print(f"RSS 수집 실패 {url}: {e}")
            return []

    results = await asyncio.gather(*[fetch_feed(url) for url in feeds])
    return [item for sublist in results for item in sublist]

async def collect_api_data(endpoints: list[dict]) -> list[dict]:
    """REST API 엔드포인트 데이터 수집"""
    results = []
    async with httpx.AsyncClient(timeout=30) as client:
        for ep in endpoints:
            try:
                resp = await client.get(
                    ep["url"],
                    headers=ep.get("headers", {}),
                    params=ep.get("params", {})
                )
                data = resp.json()
                # 각 API별 데이터 정규화
                items = ep.get("parser", lambda x: x)(data)
                for item in items:
                    item["source_type"] = "api"
                    item["source"]      = ep.get("name", ep["url"])
                results.extend(items)
            except Exception as e:
                print(f"API 수집 실패 {ep['url']}: {e}")
    return results

async def collect_web_content(urls: list[str]) -> list[dict]:
    """웹페이지 본문 추출 (Playwright)"""
    try:
        from playwright.async_api import async_playwright
    except ImportError:
        print("pip install playwright && playwright install chromium")
        return []

    results = []
    async with async_playwright() as pw:
        browser = await pw.chromium.launch(headless=True)
        context = await browser.new_context(
            user_agent="Mozilla/5.0 (compatible; AIServerBot/1.0)"
        )
        for url in urls:
            try:
                page = await context.new_page()
                await page.goto(url, wait_until="domcontentloaded", timeout=30000)
                title   = await page.title()
                content = await page.inner_text("article, main, .content, body")
                results.append({
                    "title": title, "content": content[:5000],
                    "url": url, "source_type": "web",
                    "collected_at": datetime.now(timezone.utc).isoformat()
                })
                await page.close()
            except Exception as e:
                print(f"웹 수집 실패 {url}: {e}")
        await browser.close()
    return results

def save_to_postgres(items: list[dict], table: str = "raw_data"):
    """수집 데이터 PostgreSQL 저장 (중복 URL 무시)"""
    conn = psycopg2.connect(**DB_CONFIG)
    cur  = conn.cursor()

    # 테이블 자동 생성
    cur.execute(f"""
        CREATE TABLE IF NOT EXISTS {table} (
            id          SERIAL PRIMARY KEY,
            title       TEXT,
            content     TEXT,
            url         TEXT UNIQUE,
            source      TEXT,
            source_type TEXT,
            published   TIMESTAMPTZ,
            collected_at TIMESTAMPTZ DEFAULT NOW(),
            processed   BOOLEAN DEFAULT FALSE,
            ai_category TEXT,
            ai_sentiment TEXT,
            ai_summary  TEXT
        )
    """)

    inserted = 0
    for item in items:
        try:
            cur.execute(f"""
                INSERT INTO {table} (title, content, url, source, source_type, published)
                VALUES (%s, %s, %s, %s, %s, %s)
                ON CONFLICT (url) DO NOTHING
            """, (
                item.get("title", ""),
                item.get("content", "")[:10000],
                item.get("url", ""),
                item.get("source", ""),
                item.get("source_type", ""),
                item.get("published", None),
            ))
            if cur.rowcount > 0:
                inserted += 1
        except Exception as e:
            print(f"저장 오류: {e}")

    conn.commit()
    cur.close()
    conn.close()
    print(f"✅ {inserted}개 신규 데이터 저장 (총 {len(items)}개 중)")
    return inserted

# 실행 예시
async def main():
    RSS_FEEDS = [
        "https://feeds.feedburner.com/TechCrunch",
        "https://www.technologyreview.com/feed/",
        "https://rss.arxiv.org/rss/cs.AI",
    ]
    print("📡 RSS 수집 중...")
    rss_items = await collect_rss_feeds(RSS_FEEDS)
    print(f"  수집: {len(rss_items)}건")
    save_to_postgres(rss_items, "raw_data")

asyncio.run(main())

import httpx, json, asyncio, psycopg2
from psycopg2.extras import RealDictCursor

OLLAMA_URL = "http://192.168.1.253:11434"
DB_CONFIG  = {"host":"192.168.1.253","port":5432,"dbname":"aiserver_main",
               "user":"aiserver","password":"secure_password"}

async def ai_classify_batch(items: list[dict]) -> list[dict]:
    """
    LLM 배치 분류: 카테고리·감성·요약·키워드·중요도 동시 추출
    배치로 처리해 API 호출 횟수 최소화
    """
    if not items:
        return []

    # 배치 프롬프트 구성 (한 번에 최대 10개)
    batch_text = "\n\n".join([
        f"[{i+1}] 제목: {item['title'][:200]}\n내용: {item['content'][:500]}"
        for i, item in enumerate(items[:10])
    ])

    prompt = f"""다음 {len(items[:10])}개의 데이터를 분석하고 JSON 배열로 출력해.
각 항목에 대해 정확히 다음 형식으로:

{batch_text}

출력 형식 (JSON 배열만, 다른 텍스트 없이):
[
  {{
    "index": 1,
    "category": "기술/비즈니스/정치/경제/스포츠/문화/과학/기타 중 하나",
    "subcategory": "세부 카테고리",
    "sentiment": "positive/negative/neutral",
    "sentiment_score": 0.0~1.0 사이 숫자,
    "importance": "high/medium/low",
    "summary": "한국어 1~2문장 요약",
    "keywords": ["키워드1", "키워드2", "키워드3"],
    "entities": {{"companies": [], "people": [], "locations": []}},
    "is_duplicate_risk": true/false
  }}
]"""

    async with httpx.AsyncClient(timeout=120) as client:
        resp = await client.post(
            f"{OLLAMA_URL}/api/generate",
            json={
                "model":  "qwen2.5:14b",
                "prompt": prompt,
                "stream": False,
                "options": {"temperature": 0.1}
            }
        )

    raw = resp.json()["response"]
    try:
        import re
        json_match = re.search(r'\[.*\]', raw, re.DOTALL)
        results = json.loads(json_match.group()) if json_match else []
        return results
    except Exception as e:
        print(f"JSON 파싱 오류: {e}")
        return []

def update_processed_data(results: list[dict], ids: list[int]):
    """분류 결과를 DB에 업데이트"""
    conn = psycopg2.connect(**DB_CONFIG)
    cur  = conn.cursor()
    for r in results:
        idx = r.get("index", 1) - 1
        if idx < len(ids):
            cur.execute("""
                UPDATE raw_data SET
                    processed    = TRUE,
                    ai_category  = %s,
                    ai_sentiment = %s,
                    ai_summary   = %s
                WHERE id = %s
            """, (
                f"{r.get('category','')}/{r.get('subcategory','')}",
                r.get("sentiment", "neutral"),
                r.get("summary", ""),
                ids[idx]
            ))
    conn.commit(); cur.close(); conn.close()

async def run_etl_pipeline(batch_size: int = 10):
    """미처리 데이터를 배치로 AI 분류"""
    conn = psycopg2.connect(**DB_CONFIG)
    cur  = conn.cursor(cursor_factory=RealDictCursor)
    cur.execute("""
        SELECT id, title, content FROM raw_data
        WHERE processed = FALSE
        ORDER BY collected_at DESC
        LIMIT %s
    """, (batch_size,))
    rows = cur.fetchall()
    cur.close(); conn.close()

    if not rows:
        print("✅ 처리할 데이터 없음")
        return

    print(f"🤖 AI 분류 시작: {len(rows)}건")
    ids   = [r["id"] for r in rows]
    items = [{"title": r["title"], "content": r["content"]} for r in rows]

    results = await ai_classify_batch(items)
    if results:
        update_processed_data(results, ids)
        print(f"✅ {len(results)}건 분류 완료")

    # 벡터 DB 동시 인덱싱 (RAG 검색용)
    await index_to_qdrant(rows)

async def index_to_qdrant(rows: list[dict]):
    """처리된 데이터를 Qdrant에 동시 인덱싱"""
    from qdrant_client import QdrantClient
    from qdrant_client.models import PointStruct
    from sentence_transformers import SentenceTransformer
    import uuid

    embedder = SentenceTransformer("BAAI/bge-m3")
    qdrant   = QdrantClient(host="192.168.1.253", port=6333, api_key="your_key")

    texts   = [f"{r['title']} {r['content'][:500]}" for r in rows]
    vectors = embedder.encode(texts, normalize_embeddings=True).tolist()

    points = [
        PointStruct(
            id=str(uuid.uuid4()),
            vector=vectors[i],
            payload={
                "title":   rows[i]["title"],
                "content": rows[i]["content"][:500],
                "db_id":   rows[i]["id"],
                "type":    "pipeline_data"
            }
        )
        for i in range(len(rows))
    ]
    qdrant.upsert(collection_name="ai_knowledge_base", points=points)
    print(f"  📌 Qdrant 인덱싱: {len(points)}건")

asyncio.run(run_etl_pipeline(batch_size=50))

services:
  airflow-webserver:
    image: apache/airflow:2.10.0
    container_name: airflow_web
    restart: unless-stopped
    ports:
      - "8085:8080"
    environment:
      - AIRFLOW__CORE__EXECUTOR=LocalExecutor
      - AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:airflow@airflow_postgres/airflow
      - AIRFLOW__CORE__FERNET_KEY=your_fernet_key_here
      - AIRFLOW__WEBSERVER__SECRET_KEY=your_secret_key
      - AIRFLOW__CORE__LOAD_EXAMPLES=False
      - _AIRFLOW_WWW_USER_CREATE=true
      - _AIRFLOW_WWW_USER_USERNAME=admin
      - _AIRFLOW_WWW_USER_PASSWORD=admin_password
    volumes:
      - ./dags:/opt/airflow/dags         # DAG 파일 위치
      - ./logs:/opt/airflow/logs
      - ./plugins:/opt/airflow/plugins
    command: webserver
    depends_on:
      - airflow_postgres
    networks:
      - ai_net

  airflow-scheduler:
    image: apache/airflow:2.10.0
    container_name: airflow_scheduler
    restart: unless-stopped
    environment:
      - AIRFLOW__CORE__EXECUTOR=LocalExecutor
      - AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:airflow@airflow_postgres/airflow
      - AIRFLOW__CORE__FERNET_KEY=your_fernet_key_here
    volumes:
      - ./dags:/opt/airflow/dags
      - ./logs:/opt/airflow/logs
    command: scheduler
    depends_on:
      - airflow_postgres
    networks:
      - ai_net

  airflow_postgres:
    image: postgres:16-alpine
    container_name: airflow_postgres
    environment:
      - POSTGRES_DB=airflow
      - POSTGRES_USER=airflow
      - POSTGRES_PASSWORD=airflow
    volumes:
      - airflow_pg_data:/var/lib/postgresql/data
    networks:
      - ai_net

volumes:
  airflow_pg_data:

networks:
  ai_net:
    external: true
    name: ai_network

-- 패널 1: 시간대별 수집 건수 (Time Series)
SELECT
    date_trunc('hour', collected_at) AS time,
    source,
    COUNT(*) AS count
FROM raw_data
WHERE collected_at >= NOW() - INTERVAL '7 days'
GROUP BY 1, 2
ORDER BY 1;

-- 패널 2: 카테고리별 분포 (Pie Chart)
SELECT
    ai_category,
    COUNT(*) AS count,
    ROUND(COUNT(*) * 100.0 / SUM(COUNT(*)) OVER (), 1) AS percentage
FROM raw_data
WHERE processed = TRUE
  AND collected_at >= NOW() - INTERVAL '24 hours'
GROUP BY ai_category
ORDER BY count DESC;

-- 패널 3: 감성 분석 트렌드 (Time Series)
SELECT
    date_trunc('hour', collected_at) AS time,
    ai_sentiment,
    COUNT(*) AS count
FROM raw_data
WHERE processed = TRUE
  AND collected_at >= NOW() - INTERVAL '48 hours'
GROUP BY 1, 2
ORDER BY 1;

-- 패널 4: 최근 중요 뉴스 (Table)
SELECT
    title,
    source,
    ai_category,
    ai_sentiment,
    LEFT(ai_summary, 100) || '...' AS summary,
    collected_at
FROM raw_data
WHERE processed = TRUE
  AND ai_category ILIKE '%기술%'
ORDER BY collected_at DESC
LIMIT 20;

-- 패널 5: 소스별 성과 (Bar Chart)
SELECT
    source,
    COUNT(*) AS total,
    COUNT(*) FILTER (WHERE ai_sentiment = 'positive') AS positive,
    COUNT(*) FILTER (WHERE ai_sentiment = 'negative') AS negative,
    ROUND(AVG(CASE WHEN ai_sentiment='positive' THEN 1.0
                   WHEN ai_sentiment='negative' THEN -1.0
                   ELSE 0 END), 3) AS avg_sentiment
FROM raw_data
WHERE processed = TRUE
  AND collected_at >= NOW() - INTERVAL '7 days'
GROUP BY source
ORDER BY total DESC
LIMIT 10;

import psycopg2, httpx, asyncio
from psycopg2.extras import RealDictCursor
from datetime import datetime, timezone

DB_CONFIG  = {"host":"192.168.1.253","port":5432,"dbname":"aiserver_main",
               "user":"aiserver","password":"secure_password"}
OLLAMA_URL = "http://192.168.1.253:11434"
SLACK_URL  = "https://hooks.slack.com/services/YOUR/WEBHOOK"

async def detect_anomalies():
    """
    최근 1시간 데이터에서 이상 패턴 AI 감지
    - 급격한 감성 변화
    - 특정 키워드 급증
    - 데이터 수집 이상
    """
    conn = psycopg2.connect(**DB_CONFIG)
    cur  = conn.cursor(cursor_factory=RealDictCursor)

    # 최근 1시간 vs 이전 24시간 평균 비교
    cur.execute("""
        SELECT
            COUNT(*) FILTER (WHERE collected_at >= NOW() - INTERVAL '1 hour') AS last_hour,
            COUNT(*) FILTER (WHERE collected_at >= NOW() - INTERVAL '25 hours'
                              AND collected_at < NOW() - INTERVAL '1 hour') / 24.0 AS hourly_avg,
            COUNT(*) FILTER (WHERE ai_sentiment = 'negative'
                              AND collected_at >= NOW() - INTERVAL '1 hour') AS negative_count,
            COUNT(*) FILTER (WHERE collected_at >= NOW() - INTERVAL '1 hour') AS total_recent
        FROM raw_data
        WHERE processed = TRUE
    """)
    stats = cur.fetchone()
    cur.close(); conn.close()

    anomalies = []

    # 수집량 급감 (평균 대비 50% 이하)
    if stats["hourly_avg"] > 0:
        ratio = stats["last_hour"] / stats["hourly_avg"]
        if ratio < 0.5:
            anomalies.append(f"⚠️ 데이터 수집량 급감: 평균의 {ratio*100:.0f}% ({stats['last_hour']}건)")

    # 부정 감성 급증 (70% 초과)
    if stats["total_recent"] > 10:
        neg_ratio = stats["negative_count"] / stats["total_recent"]
        if neg_ratio > 0.7:
            anomalies.append(f"🚨 부정 감성 급증: {neg_ratio*100:.0f}% ({stats['negative_count']}건)")

    if not anomalies:
        return

    # LLM으로 이상 패턴 분석 리포트 생성
    anomaly_text = "\n".join(anomalies)
    async with httpx.AsyncClient(timeout=60) as client:
        resp = await client.post(
            f"{OLLAMA_URL}/api/generate",
            json={
                "model": "qwen2.5:7b",
                "prompt": f"""다음 데이터 이상 패턴을 분석하고 가능한 원인과 권장 조치를 제안해줘 (3~5문장):

이상 항목:
{anomaly_text}

현재 시간: {datetime.now(timezone.utc).isoformat()}""",
                "stream": False,
                "options": {"temperature": 0.3}
            }
        )
    analysis = resp.json()["response"]

    # Slack 알림 전송
    slack_msg = {
        "text": "🔍 *AI 서버 데이터 파이프라인 이상 감지*",
        "blocks": [
            {"type": "header", "text": {"type": "plain_text", "text": "🔍 데이터 이상 감지"}},
            {"type": "section", "text": {"type": "mrkdwn", "text": f"*감지 항목:*\n{anomaly_text}"}},
            {"type": "section", "text": {"type": "mrkdwn", "text": f"*AI 분석:*\n{analysis}"}},
        ]
    }
    await client.post(SLACK_URL, json=slack_msg)
    print(f"✅ 이상 감지 알림 발송: {len(anomalies)}건")

# cron 또는 n8n에서 30분마다 실행
asyncio.run(detect_anomalies())

import asyncio, httpx
from datetime import datetime

COMPETITORS = {
    "경쟁사A": ["https://competitor-a.com/blog/feed", "https://competitor-a.com/news/feed"],
    "경쟁사B": ["https://competitor-b.com/rss"],
}
OLLAMA_URL = "http://192.168.1.253:11434"

async def weekly_competitor_report():
    """매주 월요일 오전 8시 경쟁사 주간 동향 리포트"""
    from collectors import collect_rss_feeds, save_to_postgres

    all_items = []
    for company, feeds in COMPETITORS.items():
        items = await collect_rss_feeds(feeds)
        for item in items:
            item["competitor"] = company
        all_items.extend(items)

    save_to_postgres(all_items, "competitor_data")

    # 주간 요약 생성
    summaries = "\n\n".join([
        f"제목: {item['title']}\n내용: {item['content'][:300]}"
        for item in all_items[:20]
    ])

    async with httpx.AsyncClient(timeout=120) as client:
        resp = await client.post(
            f"{OLLAMA_URL}/api/generate",
            json={
                "model": "qwen2.5:14b",
                "prompt": f"""경쟁사 주간 동향을 분석해서 경영진 보고용 리포트를 작성해줘.

수집 데이터:
{summaries}

리포트 형식:
1. 핵심 요약 (3문장)
2. 경쟁사별 주요 동향
3. 우리가 주목해야 할 포인트
4. 권장 대응 방안""",
                "stream": False
            }
        )
    report = resp.json()["response"]
    print(f"주간 경쟁사 리포트:\n{report}")
    return report

asyncio.run(weekly_competitor_report())

import asyncio, httpx, psycopg2
from datetime import datetime, timezone

async def collect_market_data(symbols: list[str]) -> list[dict]:
    """무료 API로 시장 데이터 수집 (Yahoo Finance 비공식 API)"""
    results = []
    async with httpx.AsyncClient(timeout=15) as client:
        for symbol in symbols:
            try:
                resp = await client.get(
                    f"https://query1.finance.yahoo.com/v8/finance/chart/{symbol}",
                    params={"interval": "1d", "range": "1mo"},
                    headers={"User-Agent": "Mozilla/5.0"}
                )
                data   = resp.json()
                quotes = data["chart"]["result"][0]
                meta   = quotes["meta"]
                results.append({
                    "symbol":        symbol,
                    "price":         meta.get("regularMarketPrice", 0),
                    "change_pct":    meta.get("regularMarketChangePercent", 0),
                    "volume":        meta.get("regularMarketVolume", 0),
                    "currency":      meta.get("currency", "USD"),
                    "collected_at":  datetime.now(timezone.utc).isoformat(),
                })
            except Exception as e:
                print(f"시장 데이터 수집 실패 {symbol}: {e}")
    return results

async def ai_market_analysis(market_data: list[dict], news_sentiment: str) -> str:
    """시장 데이터 + 뉴스 감성 통합 AI 분석"""
    market_text = "\n".join([
        f"{d['symbol']}: ${d['price']:.2f} ({d['change_pct']:+.2f}%)"
        for d in market_data
    ])

    async with httpx.AsyncClient(timeout=60) as client:
        resp = await client.post(
            "http://192.168.1.253:11434/api/generate",
            json={
                "model": "qwen2.5:7b",
                "prompt": f"""현재 시장 데이터와 뉴스 감성을 종합 분석해줘:

시장 데이터:
{market_text}

최근 뉴스 감성: {news_sentiment}

분석 내용:
1. 현재 시장 분위기
2. 주목할 종목/섹터
3. 주의 신호
(투자 조언 아님, 정보 제공 목적)""",
                "stream": False,
                "options": {"temperature": 0.3}
            }
        )
    return resp.json()["response"]

async def run_market_pipeline():
    SYMBOLS = ["NVDA", "MSFT", "GOOGL", "META", "AMZN"]
    market_data = await collect_market_data(SYMBOLS)
    analysis    = await ai_market_analysis(market_data, "중립적")
    print(f"📈 시장 분석 완료:\n{analysis}")

asyncio.run(run_market_pipeline())

import asyncio, httpx, psycopg2

async def analyze_review_batch(reviews: list[str]) -> list[dict]:
    """리뷰 배치를 LLM으로 감성·이슈 분류"""
    batch = "\n\n".join([f"[{i+1}] {r[:300]}" for i, r in enumerate(reviews[:15])])

    async with httpx.AsyncClient(timeout=90) as client:
        resp = await client.post(
            "http://192.168.1.253:11434/api/generate",
            json={
                "model":  "qwen2.5:7b",
                "prompt": f"""다음 고객 리뷰들을 분석해서 JSON 배열로 출력해:

{batch}

JSON 배열 형식 (다른 텍스트 없이):
[
  {{
    "index": 1,
    "sentiment": "positive/negative/neutral",
    "score": 0.0~1.0,
    "issue_category": "배송/품질/가격/CS/사용성/기타",
    "key_complaint": "핵심 불만사항 또는 null",
    "key_praise": "핵심 칭찬 또는 null",
    "urgency": "high/medium/low",
    "requires_response": true/false
  }}
]""",
                "stream": False,
                "options": {"temperature": 0.1}
            }
        )

    import json, re
    raw = resp.json()["response"]
    try:
        json_match = re.search(r'\[.*\]', raw, re.DOTALL)
        return json.loads(json_match.group()) if json_match else []
    except:
        return []

async def run_review_pipeline():
    """
    실제 리뷰 수집은 스토어 API 또는 크롤링으로 구현
    여기서는 예시 데이터로 시연
    """
    sample_reviews = [
        "배송이 너무 빨라서 깜짝 놀랐어요! 제품 품질도 정말 좋습니다.",
        "포장이 너무 부실해서 제품이 파손되어 왔어요. 실망입니다.",
        "가격 대비 성능이 나쁘지 않네요. 그냥 평범합니다.",
        "고객센터 연결이 너무 어려워요. 30분 기다렸는데 연결이 안 됩니다.",
    ]

    results = await analyze_review_batch(sample_reviews)

    # 긴급 대응 필요 리뷰 필터링
    urgent = [r for r in results if r.get("requires_response") and r.get("urgency") == "high"]
    if urgent:
        print(f"🚨 긴급 대응 필요 리뷰: {len(urgent)}건")
        for r in urgent:
            print(f"  [{r['issue_category']}] {r.get('key_complaint', '')}")

    # 일별 감성 통계 DB 저장
    positive = sum(1 for r in results if r.get("sentiment") == "positive")
    negative = sum(1 for r in results if r.get("sentiment") == "negative")
    print(f"✅ 리뷰 분석: 긍정 {positive}건 / 부정 {negative}건 / 총 {len(results)}건")

asyncio.run(run_review_pipeline())