홈랩 AI 서버 가이드 4편: n8n 자동화 아키텍처 및 실전 워크플로우 20선

services:
  n8n:
    image: n8nio/n8n:latest
    container_name: n8n
    restart: unless-stopped
    ports:
      - "5678:5678"
    environment:
      - N8N_BASIC_AUTH_ACTIVE=true
      - N8N_BASIC_AUTH_USER=admin
      - N8N_BASIC_AUTH_PASSWORD=your_secure_password_here
      - GENERIC_TIMEZONE=Asia/Seoul
      - N8N_HOST=n8n.yourdomain.com       # Cloudflare 도메인
      - N8N_PROTOCOL=https
      - WEBHOOK_URL=https://n8n.yourdomain.com
      - DB_TYPE=postgresdb
      - DB_POSTGRESDB_HOST=postgres
      - DB_POSTGRESDB_PORT=5432
      - DB_POSTGRESDB_DATABASE=n8n
      - DB_POSTGRESDB_USER=n8n
      - DB_POSTGRESDB_PASSWORD=n8n_password
    volumes:
      - n8n_data:/home/node/.n8n
    depends_on:
      - postgres
    networks:
      - ai_net

  postgres:
    image: postgres:16-alpine
    container_name: n8n_postgres
    restart: unless-stopped
    environment:
      - POSTGRES_DB=n8n
      - POSTGRES_USER=n8n
      - POSTGRES_PASSWORD=n8n_password
    volumes:
      - postgres_data:/var/lib/postgresql/data
    networks:
      - ai_net

volumes:
  n8n_data:
  postgres_data:

networks:
  ai_net:
    external: true
    name: ai_network

// n8n Code 노드에서 Ollama API 직접 호출
const response = await $http.request({
  method: 'POST',
  url: 'http://ollama:11434/api/generate',
  headers: { 'Content-Type': 'application/json' },
  body: {
    model: 'qwen2.5:14b',
    prompt: `다음 이메일을 한국어로 요약하고 중요도를 High/Medium/Low로 분류해줘:\n\n${$json.emailBody}`,
    stream: false,
    options: {
      temperature: 0.3,
      num_predict: 500
    }
  }
});

return {
  summary: response.response,
  originalEmail: $json.emailBody,
  processedAt: new Date().toISOString()
};

# Dify 공식 docker-compose 클론
cd ~/ai-server/automation
git clone https://github.com/langgenius/dify.git
cd dify/docker

# 환경 설정
cp .env.example .env

# 핵심 환경변수 수정 (.env 파일)
sed -i "s/SECRET_KEY=.*/SECRET_KEY=$(openssl rand -hex 32)/" .env
sed -i "s/CONSOLE_API_URL=.*/CONSOLE_API_URL=https:\/\/dify.yourdomain.com/" .env

# 실행 (포트 3001에서 서비스)
docker compose up -d

# 초기 관리자 계정 생성 (최초 1회)
docker compose exec api flask db upgrade
docker compose exec api flask reset-email-and-password

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
import httpx
import asyncio

app = Server("ai-server-mcp")

N8N_BASE_URL = "http://192.168.1.253:5678"
OLLAMA_URL   = "http://192.168.1.253:11434"

@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="generate_image",
            description="AI 이미지 생성 (ComfyUI/A1111 사용)",
            inputSchema={
                "type": "object",
                "properties": {
                    "prompt": {"type": "string", "description": "이미지 생성 프롬프트"},
                    "width":  {"type": "integer", "default": 1024},
                    "height": {"type": "integer", "default": 1024},
                },
                "required": ["prompt"]
            }
        ),
        types.Tool(
            name="trigger_n8n_workflow",
            description="n8n 워크플로우 트리거",
            inputSchema={
                "type": "object",
                "properties": {
                    "workflow_id": {"type": "string"},
                    "data": {"type": "object"}
                },
                "required": ["workflow_id"]
            }
        ),
        types.Tool(
            name="query_ollama",
            description="로컬 Ollama LLM에 직접 질문",
            inputSchema={
                "type": "object",
                "properties": {
                    "model":  {"type": "string", "default": "qwen2.5:14b"},
                    "prompt": {"type": "string"}
                },
                "required": ["prompt"]
            }
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    async with httpx.AsyncClient(timeout=120) as client:

        if name == "generate_image":
            resp = await client.post(
                "http://192.168.1.253:7860/sdapi/v1/txt2img",
                json={
                    "prompt": arguments["prompt"],
                    "width":  arguments.get("width", 1024),
                    "height": arguments.get("height", 1024),
                    "steps":  20,
                }
            )
            return [types.TextContent(type="text",
                text=f"이미지 생성 완료. 결과: {resp.json()['images'][0][:50]}...")]

        elif name == "trigger_n8n_workflow":
            resp = await client.post(
                f"{N8N_BASE_URL}/webhook/{arguments['workflow_id']}",
                json=arguments.get("data", {})
            )
            return [types.TextContent(type="text",
                text=f"워크플로우 실행: {resp.status_code} - {resp.text[:200]}")]

        elif name == "query_ollama":
            resp = await client.post(
                f"{OLLAMA_URL}/api/generate",
                json={
                    "model":  arguments.get("model", "qwen2.5:14b"),
                    "prompt": arguments["prompt"],
                    "stream": False
                }
            )
            return [types.TextContent(type="text",
                text=resp.json()["response"])]

async def main():
    async with stdio_server() as streams:
        await app.run(*streams, app.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

{
  "mcpServers": {
    "ai-server": {
      "command": "python",
      "args": ["/path/to/ai_server_mcp.py"],
      "env": {
        "AI_SERVER_IP": "192.168.1.253"
      }
    },
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem",
               "/Volumes/data", "/Volumes/storage"]
    },
    "n8n": {
      "command": "npx",
      "args": ["-y", "n8n-mcp"],
      "env": {
        "N8N_API_URL": "http://192.168.1.253:5678",
        "N8N_API_KEY": "your_n8n_api_key"
      }
    }
  }
}

pip install langchain langchain-community langgraph \
            langchain-ollama chromadb \
            tavily-python duckduckgo-search

from langchain_ollama import ChatOllama
from langchain.agents import create_react_agent, AgentExecutor
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_community.tools.python.tool import PythonREPLTool
from langchain_community.vectorstores import Chroma
from langchain_ollama import OllamaEmbeddings
from langchain.tools.retriever import create_retriever_tool
from langchain import hub

# 로컬 Ollama LLM 설정
llm = ChatOllama(
    base_url="http://192.168.1.253:11434",
    model="qwen2.5:14b",
    temperature=0.1
)

# 임베딩 모델
embeddings = OllamaEmbeddings(
    base_url="http://192.168.1.253:11434",
    model="nomic-embed-text"
)

# 지식 베이스 (ChromaDB)
vectorstore = Chroma(
    persist_directory="./knowledge_base",
    embedding_function=embeddings
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# 에이전트 도구 정의
tools = [
    DuckDuckGoSearchRun(name="web_search",
        description="최신 정보나 사실 확인이 필요할 때 사용"),
    PythonREPLTool(name="python_repl",
        description="Python 코드 실행, 계산, 데이터 분석에 사용"),
    create_retriever_tool(
        retriever,
        "knowledge_search",
        "내부 지식베이스에서 관련 문서 검색"
    ),
]

# ReAct 에이전트 생성
prompt  = hub.pull("hwchase17/react")
agent   = create_react_agent(llm, tools, prompt)
executor = AgentExecutor(
    agent=agent, tools=tools,
    verbose=True,
    max_iterations=10,
    handle_parsing_errors=True
)

# 실행 예시
result = executor.invoke({
    "input": "2026년 최신 AI 모델 트렌드를 조사하고, "
             "우리 지식베이스와 비교해서 차이점을 Python으로 분석해줘"
})
print(result["output"])

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator

class ResearchState(TypedDict):
    topic:    str
    searches: Annotated[List[str], operator.add]
    findings: Annotated[List[str], operator.add]
    report:   str
    step:     int

def search_node(state: ResearchState) -> dict:
    """웹 검색 수행"""
    search = DuckDuckGoSearchRun()
    query  = f"{state['topic']} 최신 동향 2026"
    result = search.run(query)
    return {"searches": [result], "step": state["step"] + 1}

def analyze_node(state: ResearchState) -> dict:
    """검색 결과 분석"""
    prompt = f"""다음 검색 결과를 분석하고 핵심 인사이트를 추출해:
    {' '.join(state['searches'][-3:])}"""
    result = llm.invoke(prompt)
    return {"findings": [result.content], "step": state["step"] + 1}

def report_node(state: ResearchState) -> dict:
    """최종 리포트 작성"""
    prompt = f"""다음 분석 내용을 바탕으로 전문적인 리포트를 작성해:
    주제: {state['topic']}
    분석: {' '.join(state['findings'])}"""
    result = llm.invoke(prompt)
    return {"report": result.content}

def should_continue(state: ResearchState) -> str:
    """검색 3회 후 분석으로 전환"""
    return "analyze" if state["step"] >= 3 else "search"

# 그래프 구성
graph = StateGraph(ResearchState)
graph.add_node("search",  search_node)
graph.add_node("analyze", analyze_node)
graph.add_node("report",  report_node)

graph.set_entry_point("search")
graph.add_conditional_edges("search", should_continue, {
    "search":  "search",
    "analyze": "analyze"
})
graph.add_edge("analyze", "report")
graph.add_edge("report",  END)

app = graph.compile()

result = app.invoke({
    "topic": "AI 서버 구축 최신 기술 트렌드",
    "searches": [], "findings": [], "report": "", "step": 0
})
print(result["report"])

import autogen

# Ollama 로컬 모델 설정
config_list = [{
    "model":    "qwen2.5:14b",
    "base_url": "http://192.168.1.253:11434/v1",
    "api_key":  "ollama",
    "api_type": "openai"
}]

llm_config = {
    "config_list": config_list,
    "temperature": 0.1,
    "timeout": 180
}

# 에이전트 역할 정의
user_proxy = autogen.UserProxyAgent(
    name="ProductManager",
    system_message="제품 요구사항을 정의하고 팀을 조율하는 PM입니다.",
    human_input_mode="TERMINATE",
    code_execution_config={"work_dir": "/tmp/autogen"}
)

architect = autogen.AssistantAgent(
    name="SoftwareArchitect",
    llm_config=llm_config,
    system_message="""당신은 시니어 소프트웨어 아키텍트입니다.
    요구사항을 분석하고 최적의 시스템 설계를 제안합니다.
    코드를 직접 작성하지 않고 설계 방향을 제시합니다."""
)

developer = autogen.AssistantAgent(
    name="Developer",
    llm_config=llm_config,
    system_message="""당신은 풀스택 개발자입니다.
    아키텍트의 설계를 바탕으로 실제 Python 코드를 작성합니다.
    코드는 반드시 ```python 블록으로 감싸서 제공하세요."""
)

reviewer = autogen.AssistantAgent(
    name="CodeReviewer",
    llm_config=llm_config,
    system_message="""당신은 시니어 코드 리뷰어입니다.
    작성된 코드의 품질, 보안, 성능을 검토하고 개선점을 제시합니다."""
)

# 그룹 채팅으로 협업
groupchat = autogen.GroupChat(
    agents=[user_proxy, architect, developer, reviewer],
    messages=[],
    max_round=15,
    speaker_selection_method="round_robin"
)
manager = autogen.GroupChatManager(groupchat=groupchat, llm_config=llm_config)

# 작업 시작
user_proxy.initiate_chat(
    manager,
    message="""
    다음 기능을 구현해주세요:
    - Ollama API에서 사용 가능한 모델 목록을 가져오는 함수
    - 각 모델의 크기와 최근 사용 시간을 포함
    - 결과를 pandas DataFrame으로 반환
    - 단위 테스트 포함
    """
)

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, WebsiteSearchTool
from langchain_ollama import ChatOllama

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

# AI 팀 구성
researcher = Agent(
    role="수석 리서처",
    goal="정확하고 최신의 기술 정보를 수집·검증한다",
    backstory="10년 경력의 기술 리서처. 허위 정보를 절대 허용하지 않는다.",
    llm=ollama_llm,
    tools=[SerperDevTool()],
    verbose=True
)

writer = Agent(
    role="기술 블로그 작가",
    goal="복잡한 기술 내용을 쉽고 흥미롭게 설명하는 글을 쓴다",
    backstory="개발자 출신 테크 라이터. SEO와 가독성을 동시에 추구한다.",
    llm=ollama_llm,
    verbose=True
)

editor = Agent(
    role="편집장",
    goal="글의 품질, 정확성, 일관성을 최종 검토한다",
    backstory="20년 경력 편집장. 오탈자, 논리 오류, 사실 오류를 절대 놓치지 않는다.",
    llm=ollama_llm,
    verbose=True
)

# 작업 정의
research_task = Task(
    description="2026년 로컬 AI 서버 구축 트렌드를 조사. GPU 선택, 모델 트렌드, 비용 분석 포함.",
    agent=researcher,
    expected_output="5개 이상의 핵심 인사이트가 담긴 리서치 리포트"
)

writing_task = Task(
    description="리서치 결과를 바탕으로 3000자 이상의 기술 블로그 포스트 작성. 실전 코드 예시 포함.",
    agent=writer,
    expected_output="Markdown 형식의 완성된 블로그 포스트"
)

editing_task = Task(
    description="작성된 블로그 포스트를 검토하고 최종 편집. SEO 키워드 최적화 포함.",
    agent=editor,
    expected_output="최종 편집 완료된 블로그 포스트"
)

# 크루 실행
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    process=Process.sequential,
    verbose=True
)
result = crew.kickoff()
print(result)

## 논문 요약 프롬프트
다음 논문을 분석해줘. 구성:
1. 핵심 기여(Contribution) — 3문장
2. 방법론 요약 — 5~7문장
3. 실험 결과 핵심 수치
4. 한계점 및 향후 연구 방향
5. 실무 적용 가능성 (내 분야: [분야 입력])
[논문 텍스트 붙여넣기]

## 회의 준비 프롬프트
내일 [주제] 회의에 참석합니다.
배경 정보: [컨텍스트]
회의 목적: [목적]
다음을 준비해줘:
- 예상 의제 5가지
- 각 의제별 내가 준비해야 할 질문 2개
- 주의해야 할 반대 의견과 대응 방법
- 회의 전 확인해야 할 사항 체크리스트

## 학습 자료 생성 프롬프트
[주제]를 처음 배우는 사람을 위한 학습 계획을 만들어줘.
내 배경: [현재 수준]
목표: [달성하고 싶은 것]
가용 시간: 하루 [X]시간, 총 [N]주
출력 형식:
- 주차별 학습 목표
- 각 주제별 추천 학습 순서
- 실습 프로젝트 아이디어 3개
- 학습 체크포인트 (스스로 평가 방법)

from fastapi import FastAPI
from pydantic import BaseModel
import httpx, asyncio

app = FastAPI()

class ProductRequest(BaseModel):
    product_name: str
    category:     str
    price:        int
    features:     list[str]
    target:       str = "일반 소비자"

@app.post("/generate-description")
async def generate_product_description(req: ProductRequest):
    """상품 설명 5가지 플랫폼별 자동 생성"""

    prompt = f"""다음 상품의 판매 설명을 5개 플랫폼별로 작성해줘.

상품명: {req.product_name}
카테고리: {req.category}
가격: {req.price:,}원
주요 특징: {', '.join(req.features)}
타겟 고객: {req.target}

JSON 형식으로 출력:
{{
  "naver_smart_store": "네이버 스마트스토어용 (500자 내외, SEO 키워드 포함)",
  "coupang": "쿠팡 상세페이지용 (300자 내외, 혜택 강조)",
  "instagram": "인스타그램 캡션 (150자 내외, 해시태그 10개 포함, 이모지 활용)",
  "kakao_channel": "카카오채널 메시지 (100자 내외, 친근한 말투)",
  "blog_intro": "블로그 포스팅 도입부 (200자 내외, 스토리텔링)",
  "seo_keywords": ["핵심 키워드 10개"]
}}"""

    async with httpx.AsyncClient(timeout=60) as client:
        resp = await client.post(
            "http://ollama:11434/api/generate",
            json={"model": "qwen2.5:14b", "prompt": prompt, "stream": False,
                  "options": {"temperature": 0.8}}
        )

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

    return {"product": req.product_name, "descriptions": result}

# FastAPI 실행: uvicorn main:app --host 0.0.0.0 --port 8500
# n8n HTTP Request 노드: POST http://ai-server:8500/generate-description

당신은 10년 경력의 시니어 풀스택 개발자입니다.
주요 스택: Python, Docker, Kubernetes, PostgreSQL, React

답변 원칙:
1. 코드는 항상 실행 가능한 완전한 형태로 제공
2. 보안 취약점과 엣지 케이스를 먼저 언급
3. 성능 최적화 관점에서 개선점 제안
4. 대안적인 접근 방법 1~2가지 함께 제시
5. 코드에는 반드시 한국어 주석 포함
6. 복잡한 로직은 단계별로 설명

응답 형식:
- 핵심 답변 (3~5문장)
- 코드 (완전한 실행 가능 형태)
- 주의사항 및 개선 포인트
- 참고 사항 (관련 라이브러리, 문서 링크)

from fastapi import FastAPI, Request
import httpx, asyncio

app = FastAPI()

@app.post("/github-webhook")
async def handle_pr_webhook(request: Request):
    """GitHub PR 생성/업데이트 시 자동 AI 코드 리뷰"""
    payload = await request.json()

    if payload.get("action") not in ["opened", "synchronize"]:
        return {"status": "skipped"}

    pr       = payload["pull_request"]
    pr_title = pr["title"]
    pr_body  = pr.get("body", "")
    diff_url = pr["diff_url"]
    comments_url = pr["comments_url"]

    # PR diff 가져오기
    async with httpx.AsyncClient() as client:
        diff_resp = await client.get(
            diff_url,
            headers={"Authorization": f"token {GITHUB_TOKEN}"}
        )
        diff_text = diff_resp.text[:8000]  # 토큰 한도

    # AI 코드 리뷰 생성
    review_prompt = f"""다음 Pull Request를 시니어 개발자 관점에서 리뷰해줘.

PR 제목: {pr_title}
PR 설명: {pr_body}

변경 코드 (diff):
{diff_text}

리뷰 항목:
1. 🐛 잠재적 버그 및 오류 (심각도: 높음/중간/낮음)
2. 🔒 보안 취약점 (SQL Injection, XSS, 인증 우회 등)
3. ⚡ 성능 이슈 (N+1 쿼리, 불필요한 루프, 메모리 누수)
4. 📖 코드 가독성 및 유지보수성
5. ✅ 잘된 점 (긍정적 피드백 필수)
6. 💡 개선 제안 (선택적)

각 항목은 구체적인 라인 번호나 코드 예시와 함께."""

    async with httpx.AsyncClient(timeout=120) as client:
        ai_resp = await client.post(
            "http://ollama:11434/api/generate",
            json={"model": "qwen2.5:14b", "prompt": review_prompt, "stream": False}
        )
        review = ai_resp.json()["response"]

        # GitHub에 리뷰 코멘트 게시
        await client.post(
            comments_url,
            headers={"Authorization": f"token {GITHUB_TOKEN}",
                     "Content-Type": "application/json"},
            json={"body": f"## 🤖 AI 코드 리뷰\n\n{review}\n\n---\n*AI 리뷰는 참고용입니다.*"}
        )

    return {"status": "review_posted"}

GITHUB_TOKEN = "your_github_token"

## 📋 구조화된 분석 프롬프트 (Chain-of-Thought + 구조화 출력)
[역할]
당신은 비즈니스 전략 컨설턴트입니다.

[작업]
다음 상황을 분석하고 전략을 제안해줘.

[상황]
{상황 설명}

[분석 단계 — 반드시 순서대로 수행]
Step 1: 핵심 문제 정의 (1~2문장)
Step 2: 주요 원인 분석 (3가지)
Step 3: 선택 가능한 전략 (3가지, 각각 장단점)
Step 4: 최적 전략 추천 및 근거
Step 5: 실행 계획 (30일/90일/1년)

[제약 조건]
- 실행 예산: {예산}
- 팀 규모: {규모}
- 주어진 기간: {기간}

[출력 형식]
마크다운 형식으로 각 Step을 헤더로 구분

---

## 🔄 Few-shot 예시 — 감정 분석 + 응대 초안
다음은 고객 메시지를 분석하는 예시야:

입력: "배송이 3일이나 늦었어요. 너무 실망입니다."
출력: {{"감정": "실망/불만", "긴급도": "높음", "카테고리": "배송문제", "추천응대": "먼저 사과 후 보상 제안"}}

입력: "상품 색상이 사진과 달라요."  
출력: {{"감정": "불만", "긴급도": "중간", "카테고리": "상품불일치", "추천응대": "환불 또는 교환 안내"}}

입력: "{실제 고객 메시지}"
출력:

open-webui:
  image: ghcr.io/open-webui/open-webui:main
  environment:
    ## ── 핵심 설정 ────────────────────────────────
    - OLLAMA_BASE_URL=http://ollama:11434
    - WEBUI_SECRET_KEY=생성한_32자_랜덤_키
    - WEBUI_NAME=My Private AI            # 브랜딩명
    - WEBUI_URL=https://ai.yourdomain.com

    ## ── 데이터베이스 (기본은 SQLite, 프로덕션 권장) ─
    - DATABASE_URL=postgresql://user:pass@postgres/openwebui

    ## ── 사용자 인증 ──────────────────────────────
    - ENABLE_SIGNUP=false                 # 초기 설정 후 차단
    - DEFAULT_USER_ROLE=pending
    - JWT_EXPIRES_IN=86400               # 24시간 (초)
    - ENABLE_OAUTH_SIGNUP=true

    ## ── OAuth / Google 로그인 ───────────────────
    - GOOGLE_CLIENT_ID=your_google_client_id
    - GOOGLE_CLIENT_SECRET=your_google_secret
    - OAUTH_MERGE_ACCOUNTS_BY_EMAIL=true

    ## ── RAG 설정 ─────────────────────────────────
    - VECTOR_DB=qdrant
    - QDRANT_URI=http://qdrant:6333
    - QDRANT_API_KEY=your_qdrant_key
    - RAG_EMBEDDING_MODEL=BAAI/bge-m3    # 한국어 최강
    - RAG_EMBEDDING_ENGINE=ollama
    - RAG_OLLAMA_BASE_URL=http://ollama:11434
    - CHUNK_SIZE=1000
    - CHUNK_OVERLAP=100
    - RAG_TOP_K=5
    - RELEVANCE_THRESHOLD=0.35           # 낮을수록 많이 검색
    - ENABLE_RAG_HYBRID_SEARCH=true      # 하이브리드 검색 ON
    - RAG_RERANKING_MODEL=cross-encoder/ms-marco-MiniLM-L-6-v2

    ## ── 이미지 생성 ──────────────────────────────
    - IMAGE_GENERATION_ENGINE=comfyui
    - COMFYUI_BASE_URL=http://comfyui:8188
    - IMAGE_SIZE=1024x1024
    - IMAGE_STEPS=20

    ## ── 음성 ─────────────────────────────────────
    - AUDIO_STT_ENGINE=openai
    - AUDIO_STT_OPENAI_API_BASE_URL=http://whisper-api:8000/v1
    - AUDIO_STT_OPENAI_API_KEY=any
    - AUDIO_STT_MODEL=large-v3-turbo
    - AUDIO_TTS_ENGINE=openai
    - AUDIO_TTS_OPENAI_API_BASE_URL=http://kokoro-tts:8880/v1
    - AUDIO_TTS_OPENAI_API_KEY=any
    - AUDIO_TTS_MODEL=kokoro
    - AUDIO_TTS_VOICE=af_sarah

    ## ── 웹 검색 ──────────────────────────────────
    - ENABLE_RAG_WEB_SEARCH=true
    - RAG_WEB_SEARCH_ENGINE=searxng
    - SEARXNG_QUERY_URL=http://searxng:8080/search?q=<query>&format=json
    - RAG_WEB_SEARCH_RESULT_COUNT=5

    ## ── 외부 API 연결 ────────────────────────────
    - OPENAI_API_BASE_URL=https://api.openai.com/v1
    - OPENAI_API_KEY=sk-your-key          # 클라우드 폴백용

    ## ── 성능 & 로깅 ──────────────────────────────
    - WEBUI_AUTH_TRUSTED_EMAIL_HEADER=X-Forwarded-Email
    - ENABLE_ADMIN_CHAT_ACCESS=true       # Admin이 모든 대화 확인
    - ENABLE_ADMIN_EXPORT=true

services:
  pipelines:
    image: ghcr.io/open-webui/pipelines:main
    container_name: owui_pipelines
    restart: unless-stopped
    ports:
      - "9099:9099"
    volumes:
      - pipelines_data:/app/pipelines
    environment:
      - PIPELINES_API_KEY=your_pipeline_secret_key
    networks:
      - ai_net

volumes:
  pipelines_data:

networks:
  ai_net:
    external: true
    name: ai_network

"""
파일명: korean_context_filter.py
기능: 메시지 언어 감지 → 한국어 시 시스템 프롬프트 강화
     + 개인정보(전화번호·이메일·주민번호) 자동 마스킹
"""
from typing import Optional
from pydantic import BaseModel
import re


class Filter:
    class Valves(BaseModel):
        priority: int = 0
        enable_pii_masking: bool = True
        enable_language_detection: bool = True
        korean_system_suffix: str = "답변은 반드시 한국어로 작성하세요."

    def __init__(self):
        self.valves = self.Valves()

    def inlet(self, body: dict, user: Optional[dict] = None) -> dict:
        """메시지가 LLM에 도달하기 전 처리 (입력 필터)"""
        messages = body.get("messages", [])

        for msg in messages:
            if msg.get("role") == "user":
                content = msg.get("content", "")

                # 1. PII 마스킹
                if self.valves.enable_pii_masking:
                    content = self._mask_pii(content)

                # 2. 한국어 감지 → 시스템 프롬프트 강화
                if self.valves.enable_language_detection:
                    if self._is_korean(content):
                        self._inject_korean_context(body)

                msg["content"] = content

        return body

    def outlet(self, body: dict, user: Optional[dict] = None) -> dict:
        """LLM 응답 후처리 (출력 필터)"""
        # 응답에서 불필요한 패턴 정리
        messages = body.get("messages", [])
        for msg in messages:
            if msg.get("role") == "assistant":
                content = msg.get("content", "")
                # 과도한 이모지 정리, 특수문자 정규화 등
                content = self._clean_response(content)
                msg["content"] = content
        return body

    def _mask_pii(self, text: str) -> str:
        """개인정보 마스킹"""
        # 전화번호
        text = re.sub(r'01[0-9]-?\d{3,4}-?\d{4}', '010-****-****', text)
        # 이메일
        text = re.sub(r'[\w\.-]+@[\w\.-]+\.\w+',
                      lambda m: m.group()[0] + '***@***.***', text)
        # 주민번호
        text = re.sub(r'\d{6}-[1-4]\d{6}', '******-*******', text)
        return text

    def _is_korean(self, text: str) -> bool:
        korean_chars = sum(1 for c in text if '\uAC00' <= c <= '\uD7A3')
        return korean_chars / max(len(text), 1) > 0.3

    def _inject_korean_context(self, body: dict):
        """시스템 메시지에 한국어 지시 추가"""
        messages = body.get("messages", [])
        for msg in messages:
            if msg.get("role") == "system":
                if self.valves.korean_system_suffix not in msg["content"]:
                    msg["content"] += f"\n\n{self.valves.korean_system_suffix}"
                return
        # 시스템 메시지 없으면 추가
        body["messages"].insert(0, {
            "role": "system",
            "content": self.valves.korean_system_suffix
        })

    def _clean_response(self, text: str) -> str:
        # 연속 빈 줄 3개 이상 → 2개로 정리
        text = re.sub(r'\n{3,}', '\n\n', text)
        return text.strip()

"""
파일명: claude_manifold.py
기능: Anthropic Claude API를 Open WebUI 모델 목록에 추가
     → 로컬 Ollama 모델과 함께 선택 가능
"""
from typing import List, Union, Generator, Iterator
from pydantic import BaseModel
import anthropic


class Pipe:
    class Valves(BaseModel):
        ANTHROPIC_API_KEY: str = ""

    def __init__(self):
        self.valves = self.Valves()
        self.client = None

    def pipes(self) -> List[dict]:
        """Open WebUI 모델 목록에 추가할 모델 정의"""
        return [
            {"id": "claude-opus-4-5",    "name": "Claude Opus 4.5 (Cloud)"},
            {"id": "claude-sonnet-4-5",  "name": "Claude Sonnet 4.5 (Cloud)"},
            {"id": "claude-haiku-4-5-20251001", "name": "Claude Haiku 4.5 (Cloud, 빠름)"},
        ]

    def pipe(self, body: dict) -> Union[str, Generator, Iterator]:
        """실제 API 호출"""
        if not self.valves.ANTHROPIC_API_KEY:
            return "오류: Anthropic API Key가 설정되지 않았습니다."

        client = anthropic.Anthropic(api_key=self.valves.ANTHROPIC_API_KEY)

        # 모델 ID 추출
        model_id = body["model"].split(".", 1)[-1]

        # 시스템 프롬프트 분리
        system_msg = ""
        messages   = []
        for msg in body.get("messages", []):
            if msg["role"] == "system":
                system_msg = msg["content"]
            else:
                messages.append(msg)

        # 스트리밍 응답
        with client.messages.stream(
            model=model_id,
            max_tokens=body.get("max_tokens", 2048),
            system=system_msg,
            messages=messages,
        ) as stream:
            for text in stream.text_stream:
                yield text

"""
파일명: ai_server_tools.py
Admin Panel → Tools → + New Tool 에서 코드 붙여넣기
"""
import subprocess
import httpx
from datetime import datetime
from pydantic import BaseModel


class Tools:

    def get_gpu_status(self) -> str:
        """
        현재 GPU 상태와 메모리 사용량을 확인합니다.
        :return: GPU 상태 정보 문자열
        """
        try:
            result = subprocess.run(
                ["nvidia-smi",
                 "--query-gpu=name,temperature.gpu,utilization.gpu,memory.used,memory.total,power.draw",
                 "--format=csv,noheader,nounits"],
                capture_output=True, text=True, timeout=10
            )
            if result.returncode != 0:
                return "GPU 정보를 가져올 수 없습니다."

            lines = result.stdout.strip().split('\n')
            statuses = []
            for i, line in enumerate(lines):
                parts = [p.strip() for p in line.split(',')]
                if len(parts) >= 6:
                    statuses.append(
                        f"GPU {i}: {parts[0]}\n"
                        f"  온도: {parts[1]}°C | 사용률: {parts[2]}%\n"
                        f"  VRAM: {parts[3]}MB / {parts[4]}MB\n"
                        f"  전력: {parts[5]}W"
                    )
            return "\n".join(statuses)
        except Exception as e:
            return f"GPU 상태 조회 실패: {e}"

    def get_ollama_models(self) -> str:
        """
        현재 Ollama에 설치된 AI 모델 목록을 조회합니다.
        :return: 모델 목록과 크기 정보
        """
        try:
            resp = httpx.get("http://ollama:11434/api/tags", timeout=10)
            models = resp.json().get("models", [])
            if not models:
                return "설치된 모델이 없습니다."
            lines = [f"설치된 모델 ({len(models)}개):"]
            for m in models:
                size_gb = m.get("size", 0) / 1e9
                lines.append(f"  - {m['name']} ({size_gb:.1f}GB)")
            return "\n".join(lines)
        except Exception as e:
            return f"모델 목록 조회 실패: {e}"

    def get_service_health(self) -> str:
        """
        AI 서버의 핵심 서비스 상태를 확인합니다.
        :return: 각 서비스의 헬스 상태
        """
        services = {
            "Ollama":    "http://ollama:11434/api/tags",
            "Qdrant":    "http://qdrant:6333/healthz",
            "n8n":       "http://n8n:5678/healthz",
            "ComfyUI":   "http://comfyui:8188/",
        }
        results = [f"서비스 상태 ({datetime.now().strftime('%H:%M:%S')})"]
        for name, url in services.items():
            try:
                resp = httpx.get(url, timeout=3)
                status = "✅ 정상" if resp.status_code < 400 else f"⚠️ {resp.status_code}"
            except:
                status = "❌ 연결 실패"
            results.append(f"  {name}: {status}")
        return "\n".join(results)

    def calculate(self, expression: str) -> str:
        """
        수학 계산을 수행합니다.
        :param expression: 계산할 수식 (예: '2 ** 10', 'sum([1,2,3,4,5])')
        :return: 계산 결과
        """
        try:
            allowed = {
                "__builtins__": {},
                "sum": sum, "max": max, "min": min,
                "abs": abs, "round": round, "len": len,
                "range": range, "list": list,
            }
            result = eval(expression, allowed)
            return f"{expression} = {result}"
        except Exception as e:
            return f"계산 오류: {e}"

    def get_current_time(self) -> str:
        """
        현재 한국 시간과 날짜를 반환합니다.
        :return: 현재 날짜시간 정보
        """
        from datetime import timezone, timedelta
        kst = timezone(timedelta(hours=9))
        now = datetime.now(kst)
        return (f"현재 시간: {now.strftime('%Y년 %m월 %d일 %H시 %M분 %S초')}\n"
                f"요일: {['월','화','수','목','금','토','일'][now.weekday()]}요일\n"
                f"Unix Timestamp: {int(now.timestamp())}")

# Google Cloud Console에서 OAuth 2.0 자격증명 생성 후
# Authorized redirect URI: https://ai.yourdomain.com/oauth/google/callback

# docker-compose 환경변수 추가
- ENABLE_OAUTH_SIGNUP=true
- OAUTH_MERGE_ACCOUNTS_BY_EMAIL=true
- GOOGLE_CLIENT_ID=123456789-abcdefg.apps.googleusercontent.com
- GOOGLE_CLIENT_SECRET=GOCSPX-xxxxxxxxxx

# 특정 도메인 계정만 허용 (예: company.com 직원만)
# Admin Panel → Auth → OAuth 설정에서 Allowed Email Domain: company.com

# 처음 OAuth 로그인 시 자동으로 pending 상태로 생성됨
# Admin이 Admin Panel → Users에서 승인 (Role을 user로 변경)

import httpx
from collections import defaultdict

WEBUI_URL = "http://localhost:3000"
ADMIN_KEY  = "your_admin_api_key"  # Admin Panel → Account → API Key

def get_usage_stats() -> dict:
    """사용자별 채팅 사용량 통계"""
    headers = {"Authorization": f"Bearer {ADMIN_KEY}"}

    # 전체 채팅 목록 (관리자 권한)
    resp = httpx.get(f"{WEBUI_URL}/api/v1/chats/all/db",
                     headers=headers, timeout=30)
    chats = resp.json()

    stats = defaultdict(lambda: {"chats": 0, "messages": 0, "models": set()})
    for chat in chats:
        uid   = chat.get("user_id", "unknown")
        msgs  = chat.get("chat", {}).get("messages", [])
        model = chat.get("chat", {}).get("models", ["?"])[0]

        stats[uid]["chats"]    += 1
        stats[uid]["messages"] += len([m for m in msgs if m.get("role") == "user"])
        stats[uid]["models"].add(model)

    return dict(stats)

stats = get_usage_stats()
for user_id, data in sorted(stats.items(),
                            key=lambda x: x[1]["messages"], reverse=True)[:10]:
    models = ", ".join(list(data["models"])[:3])
    print(f"사용자: {user_id[:8]}..."
          f" | 채팅: {data['chats']}개"
          f" | 메시지: {data['messages']}개"
          f" | 모델: {models}")

"""
파일명: smart_router_pipe.py
기능: 쿼리 복잡도·유형에 따라 최적 모델 자동 선택
     단순 질문 → qwen2.5:7b (빠름)
     코딩 문제 → deepseek-coder:6.7b (전문)
     복잡한 추론 → qwen2.5:14b (고품질)
     이미지 관련 → gemma3:12b (Vision)
"""
from typing import Union, Generator, Iterator
from pydantic import BaseModel
import httpx, re


class Pipe:
    class Valves(BaseModel):
        OLLAMA_URL:     str = "http://ollama:11434"
        SIMPLE_MODEL:   str = "qwen2.5:7b"
        CODING_MODEL:   str = "deepseek-r1:8b"
        COMPLEX_MODEL:  str = "qwen2.5:14b"
        VISION_MODEL:   str = "gemma3:12b"
        MIN_TOKENS_FOR_COMPLEX: int = 200

    def __init__(self):
        self.valves = self.Valves()

    def pipes(self):
        return [{"id": "smart-router", "name": "🧠 Smart Router (자동 모델 선택)"}]

    def _detect_intent(self, text: str) -> str:
        """쿼리 의도 분류"""
        text_lower = text.lower()
        code_keywords = ["코드","함수","클래스","python","javascript","sql",
                         "def ","class ","import ","error","bug","디버그","구현"]
        complex_keywords = ["분석","비교","설계","아키텍처","전략","최적화",
                            "장단점","차이점","원인","메커니즘","심층"]

        # 이미지 포함 여부 (Vision)
        has_image = any(isinstance(c, dict) and c.get("type") == "image_url"
                        for c in ([text] if isinstance(text, str) else text))
        if has_image:
            return "vision"

        if any(kw in text_lower for kw in code_keywords):
            return "coding"

        if (any(kw in text_lower for kw in complex_keywords)
                or len(text) > self.valves.MIN_TOKENS_FOR_COMPLEX):
            return "complex"

        return "simple"

    def pipe(self, body: dict) -> Union[str, Generator, Iterator]:
        messages = body.get("messages", [])
        last_user = next((m["content"] for m in reversed(messages)
                          if m["role"] == "user"), "")

        intent = self._detect_intent(
            last_user if isinstance(last_user, str)
            else str(last_user)
        )
        model_map = {
            "simple":  self.valves.SIMPLE_MODEL,
            "coding":  self.valves.CODING_MODEL,
            "complex": self.valves.COMPLEX_MODEL,
            "vision":  self.valves.VISION_MODEL,
        }
        selected_model = model_map[intent]
        print(f"[SmartRouter] intent={intent} → model={selected_model}")

        # 선택된 모델로 요청 전달
        body["model"] = selected_model
        with httpx.stream(
            "POST",
            f"{self.valves.OLLAMA_URL}/api/chat",
            json=body,
            timeout=180
        ) as response:
            for line in response.iter_lines():
                if line:
                    import json
                    data = json.loads(line)
                    if content := data.get("message", {}).get("content"):
                        yield content
                    if data.get("done"):
                        break

from youtube_transcript_api import YouTubeTranscriptApi  # pip install youtube-transcript-api
from qdrant_client import QdrantClient
from qdrant_client.models import PointStruct
from sentence_transformers import SentenceTransformer
import uuid, re

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

def youtube_to_rag(video_url: str, collection: str = "ai_knowledge_base"):
    """유튜브 영상 자막을 RAG 지식베이스로 자동 변환"""
    # 영상 ID 추출
    video_id = re.search(r'(?:v=|/)([A-Za-z0-9_-]{11})', video_url)
    if not video_id:
        raise ValueError("올바른 유튜브 URL을 입력하세요")
    vid_id = video_id.group(1)

    # 자막 가져오기 (한국어 우선, 영어 폴백)
    try:
        transcript = YouTubeTranscriptApi.get_transcript(vid_id, languages=['ko', 'en'])
    except Exception as e:
        print(f"자막 없음: {e}")
        return

    # 자막을 30초 단위로 청킹
    CHUNK_SEC = 30
    chunks    = []
    current   = {"text": "", "start": 0}

    for entry in transcript:
        current["text"] += " " + entry["text"]
        if entry["start"] - current["start"] >= CHUNK_SEC:
            if current["text"].strip():
                chunks.append({
                    "content":    current["text"].strip(),
                    "start_time": current["start"],
                    "source":     video_url,
                    "type":       "youtube_transcript"
                })
            current = {"text": "", "start": entry["start"]}

    if current["text"].strip():
        chunks.append({"content": current["text"].strip(),
                       "start_time": current["start"],
                       "source": video_url, "type": "youtube_transcript"})

    # 임베딩 & Qdrant 삽입
    texts   = [c["content"] for c in chunks]
    vectors = embedder.encode(texts, normalize_embeddings=True).tolist()

    points = [
        PointStruct(
            id=str(uuid.uuid4()),
            vector=vectors[i],
            payload={
                "content":    chunks[i]["content"],
                "source":     chunks[i]["source"],
                "start_time": f"{int(chunks[i]['start_time']//60)}:{int(chunks[i]['start_time']%60):02d}",
                "type":       "youtube_transcript",
                "category":   "video"
            }
        )
        for i in range(len(chunks))
    ]
    qdrant.upsert(collection_name=collection, points=points)
    print(f"✅ 유튜브 자막 RAG 완료: {len(chunks)}개 청크 인덱싱")

# 사용 예시
youtube_to_rag("https://www.youtube.com/watch?v=example_video_id")
# → Open WebUI에서 "영상에서 RAG 설정 방법 설명해줘" 로 검색 가능

흐름: Telegram Trigger → Ollama Chat → Telegram 답장
용도: 어디서나 로컬 LLM에 질문
설정:
  - Telegram Bot Token: @BotFather에서 발급
  - Ollama URL: http://ollama:11434
  - 모델: qwen2.5:14b (한국어 최적)
  - System Prompt: "당신은 친절한 AI 어시스턴트입니다. 한국어로 답하세요."
핵심 노드:
  [Telegram Trigger] → [Ollama: /api/chat POST] → [Telegram: sendMessage]

흐름: Telegram "요약 https://..." → Firecrawl 크롤링 → Ollama 요약 → Obsidian REST API 저장
용도: 읽고 싶은 아티클을 AI가 요약해 내 지식 베이스에 자동 저장
설정:
  - Firecrawl URL: http://firecrawl:3002
  - Obsidian Local REST API: http://192.168.1.100:27123
  - 저장 경로: /vault/.raw/web/{{날짜}}_{{제목}}.md
프롬프트:
  "다음 웹페이지 내용을 한국어로 요약하세요.
  형식: YAML 프론트매터(title, url, date, tags) + 핵심 요점 5개 + 상세 요약"

흐름: 파일 드롭(mp3/wav) → Whisper STT → Ollama 회의록 작성 → Notion API 저장
용도: 회의 후 자동으로 회의록·액션아이템 정리
Whisper 설정:
  - URL: http://whisper:9000/asr
  - 언어: ko (한국어)
  - 모델: large-v3
Ollama 프롬프트:
  "다음 회의 녹취록에서 추출하세요:
  1. 회의 제목 및 일시
  2. 참석자 목록
  3. 주요 논의 사항 (번호 목록)
  4. 결정된 사항
  5. 액션 아이템 (담당자, 기한 포함)
  6. 다음 회의 안건"

흐름: Cron(매일 07:00) → RSS/API 뉴스 수집 → Ollama 감성분석 → Telegram 리포트
용도: 관심 종목 뉴스를 매일 아침 AI가 분석해서 알림
뉴스 소스:
  - 한국경제TV RSS: https://www.wowtv.co.kr/NewsCenter/RSS
  - 네이버 금융 API
  - KIS Developers (주가 데이터)
Ollama 프롬프트:
  "다음 뉴스들에서 {{종목명}} 관련 내용을 분석하세요.
  출력: 긍정/부정/중립 판단, 핵심 이슈, 투자 시사점 (3줄 이내)"

흐름: Gmail Trigger(새 메일) → Ollama 분류/요약 → 카테고리별 레이블 → 답장 초안 → Telegram 알림
용도: 중요 이메일 놓치지 않고, 답장 초안 자동 생성
분류 카테고리: [긴급/업무/광고/구독/개인]
프롬프트:
  "다음 이메일을 분석하세요:
  1. 카테고리: [긴급/업무/광고/구독/개인] 중 선택
  2. 한 줄 요약
  3. 필요한 경우 한국어 답장 초안 (200자 이내)
  4. 긴급 여부 (true/false)"