tk-factory-services/ai-service/services/chatbot_service.py

import json
from services.ollama_client import ollama_client


def sanitize_user_input(text: str, max_length: int = 500) -> str:
    """사용자 입력 길이 제한 및 정리"""
    if not text:
        return ""
    return str(text)[:max_length].strip()


ANALYZE_SYSTEM_PROMPT = """당신은 공장 현장 신고 접수를 도와주는 AI 도우미입니다.
사용자가 현장에서 발견한 문제를 설명하면, 아래 카테고리 목록을 참고하여 가장 적합한 신고 유형과 카테고리를 제안해야 합니다.

신고 유형:
- nonconformity (부적합): 제품/작업 품질 관련 문제
- facility (시설설비): 시설, 설비, 장비 관련 문제
- safety (안전): 안전 위험, 위험 요소 관련 문제

반드시 아래 JSON 형식으로만 응답하세요. 다른 텍스트는 포함하지 마세요:
{
  "organized_description": "정리된 설명 (1-2문장)",
  "suggested_type": "nonconformity 또는 facility 또는 safety",
  "suggested_category_id": 카테고리ID(숫자) 또는 null,
  "confidence": 0.0~1.0 사이의 확신도
}"""

SUMMARIZE_SYSTEM_PROMPT = """당신은 공장 현장 신고 내용을 요약하는 AI 도우미입니다.
주어진 신고 정보를 보기 좋게 정리하여 한국어로 요약해주세요.

반드시 아래 JSON 형식으로만 응답하세요:
{
  "summary": "요약 텍스트"
}"""


async def analyze_user_input(user_text: str, categories: dict) -> dict:
    """사용자 초기 입력을 분석하여 유형 제안 + 설명 정리"""
    category_context = ""
    for type_key, cats in categories.items():
        type_label = {"nonconformity": "부적합", "facility": "시설설비", "safety": "안전"}.get(type_key, type_key)
        cat_names = [f"  - ID {c['id']}: {c['name']}" for c in cats]
        category_context += f"\n[{type_label} ({type_key})]\n" + "\n".join(cat_names) + "\n"

    safe_text = sanitize_user_input(user_text)
    prompt = f"""카테고리 목록:
{category_context}

사용자 입력:
<user_input>{safe_text}</user_input>

위 카테고리 목록을 참고하여 JSON으로 응답하세요."""

    raw = await ollama_client.generate_text(prompt, system=ANALYZE_SYSTEM_PROMPT)

    try:
        start = raw.find("{")
        end = raw.rfind("}") + 1
        if start >= 0 and end > start:
            result = json.loads(raw[start:end])
            # Validate required fields
            if "organized_description" not in result:
                result["organized_description"] = user_text
            if "suggested_type" not in result:
                result["suggested_type"] = "nonconformity"
            if "confidence" not in result:
                result["confidence"] = 0.5
            return result
    except json.JSONDecodeError:
        pass

    return {
        "organized_description": user_text,
        "suggested_type": "nonconformity",
        "suggested_category_id": None,
        "confidence": 0.3,
    }


async def summarize_report(data: dict) -> dict:
    """최종 신고 내용을 요약"""
    prompt = f"""신고 정보:
<user_input>
- 설명: {sanitize_user_input(data.get('description', ''))}
- 유형: {sanitize_user_input(data.get('type', ''))}
- 카테고리: {sanitize_user_input(data.get('category', ''))}
- 항목: {sanitize_user_input(data.get('item', ''))}
- 위치: {sanitize_user_input(data.get('location', ''))}
- 프로젝트: {sanitize_user_input(data.get('project', ''))}
</user_input>

위 정보를 보기 좋게 요약하여 JSON으로 응답하세요."""

    raw = await ollama_client.generate_text(prompt, system=SUMMARIZE_SYSTEM_PROMPT)

    try:
        start = raw.find("{")
        end = raw.rfind("}") + 1
        if start >= 0 and end > start:
            result = json.loads(raw[start:end])
            if "summary" in result:
                return result
    except json.JSONDecodeError:
        pass

    # Fallback: construct summary manually
    parts = []
    if data.get("type"):
        parts.append(f"[{data['type']}]")
    if data.get("category"):
        parts.append(data["category"])
    if data.get("item"):
        parts.append(f"- {data['item']}")
    if data.get("location"):
        parts.append(f"\n위치: {data['location']}")
    if data.get("project"):
        parts.append(f"\n프로젝트: {data['project']}")
    if data.get("description"):
        parts.append(f"\n내용: {data['description']}")

    return {"summary": " ".join(parts) if parts else "신고 내용 요약"}