import logging from fastapi import APIRouter, HTTPException, Query # 진철 추가 (Query) from fastapi.responses import RedirectResponse import openai from openai import OpenAI import boto3 import os import time from bs4 import BeautifulSoup from schemas import SpeechText, UserQuestion from config import logger, openai_api_key from datetime import datetime import chromadb from chromadb.config import Settings from sqlalchemy import create_engine, text # 진철 추가 from pydantic import BaseModel # 진철 추가 import re from datetime import datetime import ast from fastapi import BackgroundTasks import trafilatura from db import database_date, database_num, database_add_q, database_chat from collections import defaultdict from fastapi import Body import asyncio from typing import Tuple import pandas as pd # 🔧 ChromaDB 설정 chroma_client = chromadb.HttpClient( host="localhost", port=8001, settings=Settings(chroma_api_impl="rest") ) # 🔧 MySQL 연결 설정 # 진철 추가 # DB_CONFIG = { # 'host': '3.112.250.101', # 'user': 'user123', # 'password': 'PassWord12!!', # 'database': 'add_questions', # 'charset': 'utf8mb4' # } # engine = create_engine( # f"mysql+pymysql://{DB_CONFIG['user']}:{DB_CONFIG['password']}@{DB_CONFIG['host']}/{DB_CONFIG['database']}?charset=utf8mb4", # echo=True # ) EXCEL_FILE_PATH = "segue_saved_qna_master.xlsx" # 마스터 엑셀 파일 경로 COLLECTION_NAME = "segue_saved_qna13" # 사용할 ChromaDB 컬렉션 이름 collection_saved_qna = chroma_client.get_collection("segue_saved_qna13") collection_qna = chroma_client.get_collection("segue_qna_by_question4") collection_profile = chroma_client.get_collection("segue_profile_split6") # 🔧 OpenAI 설정 openai.api_key = openai_api_key client = OpenAI(api_key=openai_api_key) # 🔧 S3 설정 s3_client = boto3.client('s3') bucket_name = 'shanri-ai-chatbot-for-text-to-speech' router = APIRouter() # ✅ 음성 합성 async def synthesize_speech(text, user_id): response = client.audio.speech.create( model="tts-1", voice="nova", input=text, ) audio_file = f"tmp/audio-{user_id}-{time.time()}.mp3" with open(audio_file, 'wb') as f: for chunk in response.iter_bytes(): f.write(chunk) s3_key = f"{user_id}-{time.time()}.mp3" s3_client.upload_file(audio_file, bucket_name, s3_key) os.remove(audio_file) return f"https://{bucket_name}.s3.amazonaws.com/{s3_key}" # ✅ 저장된 QNA 검색 async def search_saved_qna_answer(question: str) -> tuple[str, float]: collection = chroma_client.get_collection("segue_saved_qna13") response = client.embeddings.create( model="text-embedding-3-small", input=question ) query_embedding = response.data[0].embedding results = collection.query( query_embeddings=[query_embedding], n_results=1 ) if not results or not results['documents'] or not results['documents'][0]: return "情報がありません。", 1.0 answer = results['metadatas'][0][0].get("answer", "") distance = results['distances'][0][0] return answer, distance # ✅ 재무재표 검색 async def search_qna_answer(question: str) -> tuple[str, float]: collection_qna = chroma_client.get_collection("segue_qna_by_question4") response = client.embeddings.create( model="text-embedding-3-large", input=question ) query_embedding = response.data[0].embedding results = collection_qna.query( query_embeddings=[query_embedding], n_results=1 ) if not results or not results['documents'] or not results['documents'][0]: return "情報がありません。", 1.0 answer = results['metadatas'][0][0].get("answer", "") distance = results['distances'][0][0] return answer, distance async def choose_hompage_url(question): """homepage_url을 선택하는 함수""" messages = [ {"role": "system", "content": "質問を分析し、回答を作成する際に必要な情報を持つURLを返す必要があります。質問を見てURLを選択してください。"}, {"role": "system", "content": "最も情報がある可能性が高いURLを配列で返す必要があります。"}, {"role": "system", "content": "関数の戻り値のように、配列のみを返す必要があります。 例) ['https://segue-g.jp/company/boardmember/index.html','https://segue-g.jp/ir/results/settle.html']"}, {"role": "system", "content": "企業理念 : https://segue-g.jp/company/sdgs/index.html"}, {"role": "system", "content": "IR 情報 : https://segue-g.jp/ir/results/settle.html"}, {"role": "system", "content": "経営成績、売上高、営業利益、経常利益、親会社株主に帰属する当期純利益 : https://segue-g.jp/ir/results/index.html"}, {"role": "system", "content": "財政状況、総資産、純資産、自己資本比率、１株当たり純資産 : https://segue-g.jp/ir/results/finance.html"}, {"role": "system", "content": "株式情報 : https://segue-g.jp/ir/stock/index.html"}, {"role": "system", "content": "当社の強み : https://segue-g.jp/ir/investor/strong_point/index.html"}, {"role": "system", "content": "成長戦略 : https://segue-g.jp/ir/investor/strategy/index.html"}, {"role": "system", "content": "会社概要 : https://segue-g.jp/company/basic/index.html"}, {"role": "system", "content": "企業理念 : https://segue-g.jp/company/brand/index.html"}, {"role": "system", "content": "沿革 : https://segue-g.jp/company/history/index.html"}, {"role": "system", "content": "役員一覧 : https://segue-g.jp/company/boardmember/index.html"}, {"role": "system", "content": "事業紹介 : https://segue-g.jp/business/index.html"}, {"role": "user", "content": "質問 : " + question} ] response = client.chat.completions.create( model="gpt-4o", messages=messages ) return response.choices[0].message.content # ✅ 뉴 크롤링 함수 async def crawl_with_urls(url_list: list[str]) -> str: all_text = "" for url in url_list: try: logger.info(f"🌐 Crawling with Trafilatura: {url}") downloaded = trafilatura.fetch_url(url) if downloaded: result = trafilatura.extract(downloaded, include_comments=False, include_tables=True) if result: all_text += ( f"\n\n========== [URL] {url} ==========\n" f"{result.strip()}\n" f"========== [END] {url} ==========\n" ) except Exception as e: logger.warning(f"Trafialtura 크롤링 실패 - {url}: {e}") return all_text if all_text else "サイト情報が取得できませんでした。" async def classification_qustion(question: str) -> str: prompt = f""" 以下は、会社に関する２つの情報カテゴリ（Table1とTable2）です。 [Table1] 売上高 売上 売り上げ うりあげ 営業利益 営業利益率 経常利益 親会社株主に帰属する当期純利益 1株当たり当期純利益 自己資本当期純利益率 総資産 純資産 総負債 自己資本比率 １株当たり純資産 営業活動によるキャッシュ・フロー 投資活動によるキャッシュ・フロー 財務活動によるキャッシュ・フロー 現金及び現金同等物の残高 [Table2] 沿革 決算短信 決算説明資料 有価証券報告書 株主総会関連資料 株主通信 その他IR情報 【追加の厳格一致ルール】 - Table1/2 のキーワードは、質問文にその語が「文字通りに出現」する場合に限り採用する（類義語・言い換え・推測は禁止）。 - 質問文に該当語が一つも出現しない場合は、意味的に近くても必ず「なし」とする。 - 「Table1 - 該当項目なし」や「Table2 - 該当項目なし」という形式は出力してはならない。該当がないときは必ず「なし - 推奨リンク: [...]」の形式で返す。 - 半角/全角や空白の差は正規化して照合するが、語尾変化・類義語置換は行わない。 ユーザーの質問に対して、次の3つの情報を特定してください： 1. 該当するテーブル名（Table1 または Table2。どちらにも該当しない場合は「なし」） 2. 該当するキーワード（例：「売上高」「営業利益率」「有価証券報告書」など。該当しない場合は「該当項目なし」） 3. 質問文に明示された西暦年、または以下の表現から自然に推定できる西暦年の配列： ※ 年が明示されている場合（例：「2023年」「2024年度」など）は、必ずその年を優先して使用してください。 ※ 明示された年と、「過去3年」「昨年」「前年同月比」「改善」「推移」などが両方含まれる場合は、明示された年を中心に考え、その他の年は補足的に扱って構いません。 - 「過去3年」「過去5年」など → 現在を含む直近 {datetime.now().year - 3}〜{datetime.now().year - 1} 年を推定 - 「昨年」「前年同月比」 → {datetime.now().year - 1} 年 - 「一昨年」 → {datetime.now().year - 2} 年 - 質問文に「改善」「推移」「変化」「動向」「変遷」「増減」などの言葉が含まれており、かつ年が明示されていない場合は、直近3年間（{datetime.now().year - 2}, {datetime.now().year - 1}, {datetime.now().year}）を推定して構いません。 - 質問文に「比較」「前年比」「前年同月比」「比べて」「差」などの言葉が含まれている場合は、今年（{datetime.now().year}年）とその前年（{datetime.now().year - 1}年）を両方含めてください。 - Table2に該当し、かつ質問文に年の記載がない場合は、2014年から現在（{datetime.now().year}年）までのすべての年を対象として構いません。 --- さらに、Table1にもTable2にも該当しない場合（"なし"と判定した場合）は、以下の候補リンクの中から「質問内容に関係がありそうなページURL」を配列形式で推薦してください。複数選んでも構いません。 [候補リンク一覧] 「会社の設立年・所在地・資本金・従業員数・事業内容など、基本的な企業情報を知りたい場合」に選択すべきURLです。 - https://segue-g.jp/company/basic/index.html 「経営トップからのメッセージや企業のビジョン・経営方針を知りたい場合」に選択すべきURLです。 - https://segue-g.jp/company/message/index.html 「企業理念・ブランドスローガン・存在意義（パーパス）を知りたい場合」に選択すべきURLです。 - https://segue-g.jp/company/brand/index.html 「代表取締役や取締役など、経営陣の氏名・役職を知りたい場合」に選択すべきURLです。 - https://segue-g.jp/company/boardmember/index.html 「グループ会社の名称・所在地・事業内容など、グループ全体の構成を知りたい場合」に選択すべきURLです。 - https://segue-g.jp/company/group/index.html 「当社のSDGs（持続可能な開発目標）への取り組み内容や社会貢献活動を知りたい場合」に選択すべきURLです。 - https://segue-g.jp/company/sdgs/index.html 「当社の主要事業領域や提供サービス、自社開発製品（RevoWorks等）の概要を知りたい場合」に選択すべきURLです。 - https://segue-g.jp/business/index.html 「セグエグループの中期経営計画や成長戦略、将来ビジョンに関する質問」の場合、このURLを選択すべきです。 - https://segue-g.jp/ir/investor/strategy/index.html 「当社の競争優位性や強み（技術力、顧客基盤、製品特徴など）を知りたい場合」に選択すべきURLです。 - https://segue-g.jp/ir/investor/strong_point/index.html 「取締役会の構成・監査体制・コンプライアンス方針など、企業統治（コーポレート・ガバナンス）に関する情報を知りたい場合」に選択すべきURLです。 - https://segue-g.jp/ir/governance/index.html 「社債の発行状況や格付機関による信用格付情報を知りたい場合」に選択すべきURLです。 - https://segue-g.jp/ir/stock/rating.html 「「当社の主要事業領域や提供サービス、自社開発製品（RevoWorks等）の概要」に選択すべきURLです。 - https://revoworks.jp/about/ 「RevoWorks製品の具体的な導入事例や活用実績、顧客の声を知りたい場合」に選択すべきURLです。 - https://revoworks.jp/media/case-study --- 【出力形式】 1. Tableに該当する場合: Table名 - キーワード - [西暦年リスト] 2. 該当しない場合（なし）: なし - 推奨リンク: [リンク1, リンク2, ...] --- 【出力例】 Table1 - 売上高 - [2022, 2023, 2024] Table1 - 売上高, 営業利益率 - [2023] Table2 - 有価証券報告書 - [2023] なし - 推奨リンク: [https://segue-g.jp/company/basic/index.html, https://segue-g.jp/business/index.html] --- 質問文: {question} """ response = client.chat.completions.create( model="gpt-4o", messages=[ {"role": "system", "content": "あなたは質問文を分類するAIアシスタントです。質問がどのテーブルに該当するか判断してください。"}, {"role": "user", "content": prompt} ] ) return response.choices[0].message.content.strip() def parse_classified_result(result: str) -> list[tuple[str, str, list[int]]]: """ 例: "Table1 - 売上高 - [2022, 2023]" "Table1 - 売上 - [該当項目なし]" "Table1 - 売上高 - 該当項目なし" ← 대괄호 없어도 처리 → [("Table1","売上高",[...]), ...] """ try: # 대괄호가 있든 없든 모두 매칭 m = re.match(r"^(Table[12])\s*-\s*(.+?)\s*-\s*(?:\[(.*?)\]|(.*))$", result) if not m: return [] table_name = m.group(1) # 지표 목록에서 '該当項目なし'는 제거 indicators = [ i.strip() for i in m.group(2).split(",") if i.strip() and i.strip() != "該当項目なし" ] # years 문자열: 대괄호 그룹(3) 우선, 없으면 일반 그룹(4) years_str = (m.group(3) or m.group(4) or "").strip() # 4자리 숫자만 연도로 채택 years = [int(y) for y in re.findall(r"\d{4}", years_str)] # 지표가 하나도 남지 않으면 실패 취급 → 폴백 흐름으로 if not indicators: return [] return [(table_name, indicator, years) for indicator in indicators] except Exception as e: logger.warning(f"分類結果の解析エラー: {e}") return [] async def fetch_table1_indicator_markdown(indicator: str, years: list[int]) -> str: # ✅ indicator 별칭 정규화 alias_map = { "売上": "売上高", "売り上げ": "売上高", "売上額": "売上高", "うりあげ": "売上高", } indicator = alias_map.get(indicator, indicator) # ✅ 연도 비어 있으면 직전 3년 기본값 if not years: y = datetime.now().year years = [y-2, y-1, y] table_map = { "売上高": ("sales_revenue", "https://segue-g.jp/ir/results/index.html"), "営業利益": ("operating_profit", "https://segue-g.jp/ir/results/index.html"), "営業利益率": ("operating_profit_margin", "https://segue-g.jp/ir/results/index.html"), "経常利益": ("ordinary_profit", "https://segue-g.jp/ir/results/index.html"), "親会社株主に帰属する当期純利益": ("net_income_attributable_to_owners", "https://segue-g.jp/ir/results/index.html"), "1株当たり当期純利益": ("earnings_per_share", "https://segue-g.jp/ir/results/index.html"), "自己資本当期純利益率": ("return_on_equity", "https://segue-g.jp/ir/results/index.html"), "総資産": ("total_assets", "https://segue-g.jp/ir/results/finance.html"), "純資産": ("net_assets", "https://segue-g.jp/ir/results/finance.html"), "総負債": ("total_liabilities", "https://segue-g.jp/ir/results/finance.html"), "自己資本比率": ("equity_ratio", "https://segue-g.jp/ir/results/finance.html"), "１株当たり純資産": ("net_assets_per_share", "https://segue-g.jp/ir/results/finance.html"), "営業活動によるキャッシュ・フロー": ("cash_flow_operating", "https://segue-g.jp/ir/results/cf.html"), "投資活動によるキャッシュ・フロー": ("cash_flow_investing", "https://segue-g.jp/ir/results/cf.html"), "財務活動によるキャッシュ・フロー": ("cash_flow_financing", "https://segue-g.jp/ir/results/cf.html"), "現金及び現金同等物の残高": ("cash_and_equivalents", "https://segue-g.jp/ir/results/cf.html"), } unit_map = { "売上高": "百万円", "営業利益": "百万円", "営業利益率": "%", "経常利益": "百万円", "親会社株主に帰属する当期純利益": "百万円", "1株当たり当期純利益": "円", "自己資本当期純利益率": "%", "総資産": "百万円", "純資産": "百万円", "総負債": "百万円", "自己資本比率": "%", "１株当たり純資産": "円", "営業活動によるキャッシュ・フロー": "百万円", "投資活動によるキャッシュ・フロー": "百万円", "財務活動によるキャッシュ・フロー": "百万円", "現金及び現金同等物の残高": "百万円" } table_entry = table_map.get(indicator) if not table_entry: return f"「{indicator}」に対応するデータテーブルが見つかりませんでした。" table_name, reference_url = table_entry unit = unit_map.get(indicator, "金額") query = f""" SELECT year, quarter, amount FROM {table_name} WHERE year IN :years ORDER BY year, FIELD(quarter, '1Q', '2Q', '3Q', '4Q', '合計') """ rows = await database_num.fetch_all(query=query, values={"years": years}) if not rows: return f"「{indicator}」に関するデータが見つかりませんでした。" from collections import defaultdict grouped = defaultdict(list) for row in rows: grouped[row["year"]].append((row["quarter"], row["amount"])) md = f"\n### {indicator}\n" note = ( "弊社の2025年12月期第1四半期の売上高は47億8,800万円。" "第2四半期累計では前年同期比18.9%増の100億円に達しました。" "通期業績予想は売上高248億円（前期比＋32.5%）へ上方修正されています。" ) md += f"\n{note}\n" md += f"| 年度 | 四半期 | 金額 ({unit}) |\n|------|--------|----------------|\n" for year in sorted(grouped.keys()): for idx, (quarter, amount) in enumerate(grouped[year]): year_label = f"{year}年" if idx == 0 else "" if unit == "%": md += f"| {year_label} | {quarter} | {amount:.2f} |\n" else: md += f"| {year_label} | {quarter} | {int(amount):,} |\n" md += f"\n参考リンク: [{reference_url}]\n" return md def format_table2_output(indicator: str, rows: list[dict]) -> str: grouped = defaultdict(list) url_set = set() for row in rows: output = row["output"] url = row["URL"] if "URL" in row and row["URL"] else "" match = re.search(r"(\d{4})年", output) if match: year = int(match.group(1)) grouped[year].append(output) # HTML 태그 포함 여부 확인 if url and not any(tag in url for tag in ["", "href="]): url_set.add(url) md = f"

{indicator}

\n" for year in sorted(grouped): md += f"

{year}年

\n" for output in grouped[year]: for line in output.strip().split("\n"): line = line.strip() if line: md += f"・{line}
\n" md += "

\n" md += "\n参考リンク一覧:\n" for u in sorted(url_set): md += f"- {u}\n" return md async def fetch_table2_indicator_markdown(indicator: str, years: list[int]) -> str: # ✅ 연도 비어 있으면 2014년부터 올해까지 if not years: y = datetime.now().year years = list(range(2014, y+1)) TABLE2_MAP = { "沿革": "Company_History", "決算短信": "financial_summary_reports", "決算説明資料": "financial_presentation_materials", "有価証券報告書": "securities_reports", "株主総会関連資料": "shareholder_meeting_documents", "株主通信": "shareholder_newsletters", "その他IR情報": "other_ir_information", } if indicator not in TABLE2_MAP: return f"「{indicator}」に対応するTable2のテーブルが見つかりませんでした。" table_name = TABLE2_MAP[indicator] query = f""" SELECT output, URL, date FROM {table_name} ORDER BY id """ try: rows = await database_date.fetch_all(query=query) # ✅ 연도 포함 여부만으로 필터링 filtered_rows = [ row for row in rows if any(str(year) in str(row["date"]) for year in years) ] if not filtered_rows: return f"「{indicator}」に関するデータが見つかりませんでした。" return format_table2_output(indicator, filtered_rows) except Exception as e: logger.exception(f"Table2データ取得エラー ({indicator})") return f"「{indicator}」に関するデータの取得中にエラーが発生しました。" ###################################### async def web_search_context(question: str) -> Tuple[str, str]: """ Responses API + web_search 툴을 사용해 - bullet 요약(최대 10줄) - JSON 배열 형태의 출처 [{title, url}] 생성 를 한 번에 받아온 뒤, (summary, sources) 튜플로 리턴 """ import re, json # ← ①에서 사용 # ① 회사명 자동 부착: 질문에 회사명이 없으면 붙여서 검색 company_kw = "セグエグループ株式会社" if not re.search(r"(セグエグループ|Segue\s?Group|セグエ)", question, flags=re.I): question_for_search = f"{company_kw} {question}" else: question_for_search = question # Blocking 방지: sync SDK 호출을 스레드로 def _call(): return client.responses.create( model="gpt-4o", # gpt-4o / gpt-4o-mini 등 web_search 지원 모델 tools=[{"type": "web_search"}], # ② 프롬프트 강화: 공식 도메인 우선 + site: 필터 권장 input=f""" You are a research assistant. Search the web and return: 1) A concise bullet-point summary (<= 10 lines) answering the user's question using the latest info. 2) Then output a JSON array named SOURCES with objects of shape: {{"title": "...", "url": "..."}}, 5 items max. Requirements: - The topic is "セグエグループ株式会社" (Segue Group). If the question text lacks the company name, include it in your search query. - Prefer official and primary sources. Prioritize domains: segue-g.jp, revoworks.jp. If you must use others, prefer EDINET, JPX, Nikkei, and government/regulatory sites. - When possible, use search operators like: (site:segue-g.jp OR site:revoworks.jp). - SOURCES must be high-quality, non-duplicate, and include at least one official page when available. Question: {question_for_search} """.strip(), max_output_tokens=4096, ) resp = await asyncio.to_thread(_call) # Python SDK는 보조 프로퍼티로 합쳐진 텍스트를 제공 (없으면 fallback) text = getattr(resp, "output_text", None) or (resp.output[0].content[0].text if getattr(resp, "output", None) else "") # 간단 파싱: SOURCES JSON 블록만 분리(실패시 전체 텍스트를 sources로) summary = text sources_str = "[]" # ```json ... ``` 또는 SOURCES: [...] 패턴 탐색 m = re.search(r"SOURCES\s*:\s*(\[[\s\S]*?\])", text) if m: sources_str = m.group(1) # summary에서 SOURCES 블록 제거 summary = text.replace(m.group(0), "").strip() else: # fenced json 추출 시도 m2 = re.search(r"```json\s*([\s\S]*?)```", text) if m2: try: _obj = json.loads(m2.group(1)) if isinstance(_obj, list): sources_str = m2.group(1) summary = text.replace(m2.group(0), "").strip() except: pass return summary.strip(), sources_str.strip() async def handle_specific_queries(question: str) -> str: target_question = "最近3年の売上高推移を教えてください" target_question2 = "売上高を教えてください" if question == target_question2: return "弊社の2025年12月期第1四半期の売上高は47億8,800万円。第2四半期累計では前年同期比18.9%増の100億円に達しました。通期業績予想は売上高248億円（前期比＋32.5%）へ上方修正されています。" if question != target_question: # 질문이 다르면 아무것도 하지 않고 None을 반환합니다. return None indicator = "売上高" indicator_info = { "db_table": "sales_revenue", "reference_url": "https://segue-g.jp/ir/results/index.html", "unit": "百万円" } # 3. 최근 3년치 연도를 계산합니다. y = datetime.now().year years = [y - 2, y - 1, y] # 4. 데이터베이스에서 '売上高' 데이터를 조회합니다. query = f""" SELECT year, quarter, amount FROM {indicator_info['db_table']} WHERE year IN :years ORDER BY year, FIELD(quarter, '1Q', '2Q', '3Q', '4Q', '合計') """ rows = await database_num.fetch_all(query=query, values={"years": years}) if not rows: return f"「{indicator}」 관련 데이터를 찾을 수 없습니다." # 5. 조회된 데이터를 바탕으로 Markdown 형식의 표를 생성합니다. grouped = defaultdict(list) for row in rows: grouped[row["year"]].append((row["quarter"], row["amount"])) md = f"\n### {indicator}\n" md += f"| 年度 | 四半期 | 金額 ({indicator_info['unit']}) |\n|------|--------|----------------|\n" for year in sorted(grouped.keys()): for idx, (quarter, amount) in enumerate(grouped[year]): year_label = f"{year}年" if idx == 0 else "" md += f"| {year_label} | {quarter} | {int(amount):,} |\n" md += f"\n参考リンク: [{indicator_info['reference_url']}]\n" return md # ✅ GPT 대답 생성 로직 async def generate_gpt_answer(question: str) -> str: special_answer = await handle_specific_queries(question) if special_answer: # 핸들러가 답변을 생성했으면, 즉시 반환하고 모든 과정을 종료 return special_answer # ✅ Step 0: GPTによる分類 classified_result = await classification_qustion(question) print("분석결과 : " + classified_result) # # Table1, Table2 키워드 맵 (厳密一致용) # TABLE1_KEYWORDS = [ # "売上高", "売上", "売り上げ", "売上額", "営業利益", "営業利益率", "経常利益", # "親会社株主に帰属する当期純利益", "1株当たり当期純利益", "自己資本当期純利益率", # "総資産", "純資産", "総負債", "自己資本比率", "１株当たり純資産", # "営業活動によるキャッシュ・フロー", "投資活動によるキャッシュ・フロー", # "財務活動によるキャッシュ・フロー", "現金及び現金同等物の残高" # ] # TABLE2_KEYWORDS = [ # "沿革", "決算短信", "決算説明資料", "有価証券報告書", # "株主総会関連資料", "株主通信", "その他IR情報" # ] # # ✅ GPT가 "なし"라고 해도 실제 질문에 키워드가 있으면 강제로 보정 # if classified_result.startswith("なし"): # # Table1 키워드 확인 # for kw in TABLE1_KEYWORDS: # if kw in question: # logger.info(f"⚡ 방어発動: 질문内にTable1キーワード {kw} を検出") # classified_result = f"Table1 - {kw} - []" # break # # Table2 키워드 확인 (Table1에서 못 걸렸을 때만) # if classified_result.startswith("なし"): # for kw in TABLE2_KEYWORDS: # if kw in question: # logger.info(f"⚡ 방어発動: 질문内にTable2キーワード {kw} を検出") # classified_result = f"Table2 - {kw} - []" # break OVERRIDE_QUESTIONS = { "株主総会はいつですか", "株主総会はいつ開催されますか", "株主総会はいつ開かれますか", "株主総会はいつありますか", "売上高を教えてください" } if question in OVERRIDE_QUESTIONS: classified_result = "なし" # 결과를 'なし'로 강제 덮어쓰기 # ✅ Step 1: 「なし」で始まる場合 → Fallback 흐름 if classified_result.startswith("なし"): # 🔍 Step 1-a: 財務QnAから検索 qna_answer, qna_distance = await search_qna_answer(question) logger.info(f"📊 財務類似質問 - distance: {qna_distance}, answer: {repr(qna_answer)}") if qna_distance < 0.4: return qna_answer # 🔍 Step 1-b: 保存済みQnAから検索 saved_answer, saved_distance = await search_saved_qna_answer(question) logger.info(f"🎯 保存済みQnA - distance: {saved_distance}, answer: {repr(saved_answer)}") if saved_distance < 0.4: return saved_answer # 🔍 Step 1-c: URL 파싱하여クロール 및 GPT 생성 url_match = re.search(r"なし\s*-\s*推奨リンク:\s*\[(.*?)\]", classified_result) if url_match: url_list_str = url_match.group(1) logger.info(f"📦 추출된 URL 리스트 문자열 (원본): {url_list_str}") try: # ✅ URL들을 안전하게 리스트로 변환: 따옴표가 없으면 추가 urls = [u.strip() for u in url_list_str.split(',')] quoted_urls = [f'"{u}"' if not (u.startswith('"') or u.startswith("'")) else u for u in urls] safe_str = f"[{', '.join(quoted_urls)}]" logger.info(f"📦 변환된 URL 리스트 문자열 (quoted): {safe_str}") url_list = ast.literal_eval(safe_str) logger.info(f"🧾 최종 파싱된 URL 리스트: {url_list}") except Exception as e: logger.warning(f"URL 파싱 실패: {e}") return "情報がありません。" website_data = await crawl_with_urls(url_list) web_summary, web_sources_json = await web_search_context(question) # 확인용 출력 print("=== 🕸 Web Search Summary ===") print(web_summary) print("=== 🕸 Web Search Sources(JSON) ===") print(web_sources_json) messages = [ {"role": "system", "content": "あなたは「セグエグループ」の会社紹介および就職希望者向けのチャットボットです。"}, {"role": "system", "content": "以下のルールに従って、ユーザーと会話してください。"}, {"role": "system", "content": "1.会話スタイル：丁寧で信頼感のあるビジネス口調で応答してください。"}, {"role": "system", "content": "2.トーン：敬語を用いながらも、丁寧で親しみやすい表現を心がけてください。"}, {"role": "system", "content": "3.回答制限：外部サイトへのリンクは絶対に出力してはいけません。リンクを案内する場合は必ず https://segue-g.jp/index.html のみを使用してください。"}, # {"role": "system", "content": "3.情報の優先順位：返答は提供された公式サイト情報を最優先とし、不足する場合は一般知識を補足的に用いてください。その際は公式サイトの内容を優先し、推測や不確実な情報は避けてください。"}, {"role": "system", "content": "4.数値データ形式：数値がある場合は、マークダウン形式の表で表示してください。"}, {"role": "system", "content": "5.リンクを表記する際はマークダウン形式([テキスト]URL)を使わず、必ずURLのみをそのまま記載してください。末尾に括弧や句読点を付けないでください。"}, # {"role": "system", "content": "5.回答制限：電話番号やメールアドレスなどの直接的な連絡先情報は提供しないでください。代わりに、必要に応じて公式ウェブサイトのURL（https://segue-g.jp/index.html）を案内してください。"}, {"role": "system", "content": f"企業サイト情報:\n{website_data}"}, {"role": "system", "content": f"ウェブ検索結果の要約:\n{web_summary}\n\n参考ソース(JSON):\n{web_sources_json}"}, {"role": "user", "content": question} ] logger.info("🕸 GPTにクロール結果を元に回答生成開始") response = client.chat.completions.create( model="gpt-4o", messages=messages ) return response.choices[0].message.content.strip() return "情報がありません。" # ✅ Step 2: Table1 / Table2 데이터를 조회 parsed = parse_classified_result(classified_result) if not parsed: qna_answer, qna_distance = await search_qna_answer(question) if qna_distance < 0.4: return qna_answer saved_answer, saved_distance = await search_saved_qna_answer(question) if saved_distance < 0.4: return saved_answer return "情報がありません。" answers = [] for table_name, indicator, years in parsed: if table_name == "Table1": answer = await fetch_table1_indicator_markdown(indicator, years) elif table_name == "Table2": answer = await fetch_table2_indicator_markdown(indicator, years) else: answer = f"不明なテーブルタイプ: {table_name}" answers.append(answer) return "\n\n".join(answers) #✅ 마스터 엑셀 파일을 읽어 ChromaDB 컬렉션을 통째로 새로고침하는 함수. async def reload_chroma_from_excel(): try: logger.info(f"'{EXCEL_FILE_PATH}' 파일에서 ChromaDB 새로고침을 시작합니다...") df = pd.read_excel(EXCEL_FILE_PATH) df.dropna(subset=['question', 'answer'], inplace=True) questions = df['question'].tolist() answers = df['answer'].tolist() if not questions: logger.warning("엑셀에 데이터가 없어 ChromaDB를 비웁니다.") chroma_client.delete_collection(name=COLLECTION_NAME) chroma_client.create_collection(name=COLLECTION_NAME) return # 모든 질문을 한번에 임베딩 response = client.embeddings.create( model="text-embedding-3-small", input=questions ) embeddings = [data.embedding for data in response.data] # 기존 컬렉션 삭제 후 새로 생성 try: chroma_client.delete_collection(name=COLLECTION_NAME) except Exception: pass collection = chroma_client.create_collection(name=COLLECTION_NAME) # 모든 데이터를 한 번에 추가 collection.add( ids=[str(i) for i in range(len(questions))], embeddings=embeddings, documents=questions, metadatas=[{'answer': ans} for ans in answers] ) logger.info(f"✅ ChromaDB '{COLLECTION_NAME}' 컬렉션을 {len(questions)}개 데이터로 새로고침했습니다.") except FileNotFoundError: logger.error(f"🚨 마스터 엑셀 파일을 찾을 수 없습니다: {EXCEL_FILE_PATH}") except Exception as e: logger.error(f"🚨 ChromaDB 새로고침 중 에러 발생: {e}", exc_info=True) #✅ 새로운 Q&A를 엑셀에 추가하고, ChromaDB 새로고침을 실행하는 함수. async def add_qna_to_excel_and_reload(question: str, answer: str): if any(x in answer for x in ["申し訳", "情報がありません", "わかりかねます"]): return # 저장할 가치가 없는 답변은 무시 try: # ✨ 1. 저장하기 전에, 먼저 ChromaDB에서 유사한 질문이 있는지 확인합니다. logger.info(f"'{question}'과 유사한 질문이 있는지 확인합니다...") response = client.embeddings.create( model="text-embedding-3-small", input=question ) embedding = response.data[0].embedding collection = chroma_client.get_collection(name=COLLECTION_NAME) results = collection.query( query_embeddings=[embedding], n_results=1 ) # ✨ 2. 거리(distance)를 기준으로 중복 여부를 판단합니다. # results['distances']가 비어있지 않고, 첫 번째 거리 값이 0.15보다 작으면 중복으로 간주합니다. if results and results.get('distances') and results['distances'][0] and results['distances'][0][0] < 0.15: logger.info(f"유사한 질문이 이미 존재하여 저장하지 않습니다. (Distance: {results['distances'][0][0]:.4f})") return # 여기서 함수를 종료하여 저장을 막습니다. # ✨ 3. 유사한 질문이 없을 경우에만 엑셀에 추가하고 DB를 새로고침합니다. logger.info(f"새로운 Q&A를 '{EXCEL_FILE_PATH}'에 추가합니다...") new_data = pd.DataFrame([{"question": question, "answer": answer}]) try: df = pd.read_excel(EXCEL_FILE_PATH) updated_df = pd.concat([df, new_data], ignore_index=True) except FileNotFoundError: updated_df = new_data updated_df.to_excel(EXCEL_FILE_PATH, index=False) logger.info("✅ 엑셀 파일에 저장을 완료했습니다.") await reload_chroma_from_excel() except Exception as e: logger.error(f"🚨 엑셀에 Q&A 저장 중 에러 발생: {e}", exc_info=True) # ✅ 건강 체크 @router.get("/health") async def health_check(): return {"status": "healthy"} # ✅ 음성 요청 처리 @router.post("/apige/speech") async def speech(speech_text: SpeechText): text = speech_text.text chat_token = speech_text.chat_token if not text: raise HTTPException(status_code=400, detail="Text is required") audio_file = await synthesize_speech(text, chat_token) return {"audio_file": audio_file} # ✅ 질문 API @router.post("/apige/ask_question") async def gemini_question(user_question: UserQuestion, background_tasks: BackgroundTasks): question_text = user_question.question.strip() chat_token = user_question.chat_token.strip() if not question_text or not chat_token: raise HTTPException(status_code=400, detail="question と chat_token が必要です") # ✅ 1. 기존 chat_token이 존재하는지 확인 query = "SELECT id FROM chats WHERE chat_token = :token" chat_id = await database_chat.fetch_val(query, values={"token": chat_token}) # ✅ 2. 없으면 새로 chats 생성 if not chat_id: insert_chat_query = """ INSERT INTO chats (title, chat_token) VALUES (:title, :token) """ chat_id = await database_chat.execute( query=insert_chat_query, values={"title": question_text[:30], "token": chat_token} ) # ✅ 3. GPT 응답 생성 answer = await generate_gpt_answer(question_text) # ✅ 4. sort_order 계산 sort_query = "SELECT MAX(sort_order) FROM chat_messages WHERE chat_id = :chat_id" current_sort = await database_chat.fetch_val(sort_query, values={"chat_id": chat_id}) or 0 # ✅ 5. 질문 + 응답 저장 insert_query = """ INSERT INTO chat_messages (chat_id, role, message, sort_order) VALUES (:chat_id, :role, :message, :sort_order) """ await database_chat.execute_many( query=insert_query, values=[ {"chat_id": chat_id, "role": "user", "message": question_text, "sort_order": current_sort + 1}, {"chat_id": chat_id, "role": "assistant", "message": answer, "sort_order": current_sort + 2} ] ) background_tasks.add_task(add_qna_to_excel_and_reload, question_text, answer) return {"answer": answer} ############################# #진철 작업 # ===================== add_question.html ======================== # 질문 목록 가져오기 @router.get("/apige/get_questions") async def get_questions(): try: query = "SELECT id, question FROM `list` ORDER BY id" rows = await database_add_q.fetch_all(query) return [{"id": row["id"], "question": row["question"]} for row in rows] except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # 답변 가져오기 @router.get("/apige/get_answer") async def get_answer(id: int): try: query = "SELECT question, answer FROM `list` WHERE id = :id" row = await database_add_q.fetch_one(query, values={"id": id}) if row: return {"question": row["question"], "answer": row["answer"] or ""} return {"question": "", "answer": ""} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # 유사 질문 저장 @router.post("/apige/save_question") async def save_question(data: dict): table = data.get("table") question = data.get("question") # if not table or not question or not table.startswith("Q") or not table[1:].isdigit() or int(table[1:]) not in range(1, 19): # raise HTTPException(status_code=400, detail="無効なテーブル名または質問です") try: query = f"INSERT INTO `{table}` (Q) VALUES (:question)" await database_add_q.execute(query, values={"question": question}) return {"success": True, "message": "データが保存されました"} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # Q1부터 Q18 테이블의 Q 데이터 가져오기 @router.get("/apige/questions/{id}") async def get_table_data(id: int): # if id not in range(1, 19): # ID가 1부터 18까지 허용 # raise HTTPException(status_code=400, detail="無効なIDです") table_name = f"Q{id}" try: query = f"SELECT id, Q FROM `{table_name}` ORDER BY id ASC" rows = await database_add_q.fetch_all(query) return [{"id": row["id"], "Q": row["Q"]} for row in rows] except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # 유사 질문 삭제 @router.delete("/apige/delete_question") async def delete_question(table: str, id: int): # if not table.startswith("Q") or not table[1:].isdigit() or int(table[1:]) not in range(1, 19): # raise HTTPException(status_code=400, detail="無効なテーブル名です") try: query = f"DELETE FROM `{table}` WHERE id = :id LIMIT 1" result = await database_add_q.execute(query, values={"id": id}) return {"success": True, "message": "削除しました"} except Exception as e: return {"success": False, "message": str(e)} #=========================== date.html =============================== class InsertData(BaseModel): table: str date: str content: str URL: str class UpdateData(InsertData): id: int # 테이블 목록 가져오기 @router.get("/apige/get_tables") async def get_tables(): try: rows = await database_date.fetch_all("SHOW TABLES") return [list(row.values())[0] for row in rows] except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # 선택한 테이블 목록 조회 @router.get("/apige/get_rows") async def get_rows(table: str = Query(...)): try: query = f"SELECT * FROM {table}" rows = await database_date.fetch_all(query) return [dict(row) for row in rows] except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # 데이터 추가 @router.post("/apige/insert_row") async def insert_row(data: InsertData): print(data) try: query = f""" INSERT INTO `{data.table}` (`date`, `content`, `URL`, `output`, `{data.table}5`, `{data.table}6`) VALUES (:date, :content, :URL, :output, :col5, :col6) """ values = { "date": data.date, "content": data.content, "URL": data.URL, "output": "", # TEXT 컬럼 "col5": 0, # INT "col6": "あ" # VARCHAR(10) } await database_date.execute(query, values=values) print("실행 쿼리:", query) print("실행 데이터:", values) return {"message": f"{data.table} 테이블에 데이터가 추가되었습니다."} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # 데이터 수정 @router.put("/apige/update_row") async def update_row(data: UpdateData): try: query = f""" UPDATE `{data.table}` SET `date` = :date, `content` = :content, `URL` = :URL, `output` = :output, `{data.table}5` = :col5, `{data.table}6` = :col6 WHERE id = :id """ values = { "date": data.date, "content": data.content, "URL": data.URL, "output": "", # TEXT "col5": 0, # INT "col6": "あ", # VARCHAR(10) "id": data.id } print("실행쿼리:", query) print("실행데이터:", values) await database_date.execute(query, values=values) return {"message": f"{data.table} 테이블의 데이터가 수정되었습니다."} except Exception as e: print("Update Error:", str(e)) raise HTTPException(status_code=500, detail=str(e)) ############################# ############################ # 리스트페이지 @router.get("/apige/chat_list") async def get_chat_list(): query = """ SELECT id, title, chat_token, created_at FROM chats ORDER BY id DESC """ try: rows = await database_chat.fetch_all(query) return [{"id": row["id"], "title": row["title"], "chat_token": row["chat_token"], "created_at": row["created_at"]} for row in rows] except Exception as e: raise HTTPException(status_code=500, detail=f"チャット一覧の取得に失敗しました: {str(e)}") @router.get("/apige/chat_messages/{chat_id}") async def get_chat_messages(chat_id: int): query = """ SELECT role, message, sort_order, created_at FROM chat_messages WHERE chat_id = :chat_id ORDER BY sort_order ASC """ try: rows = await database_chat.fetch_all(query, values={"chat_id": chat_id}) return [{"role": row["role"], "message": row["message"], "sort_order": row["sort_order"], "created_at": row["created_at"]} for row in rows] except Exception as e: raise HTTPException(status_code=500, detail=f"チャットメッセージの取得に失敗しました: {str(e)}") ############################### @router.post("/apige/saveVector") async def save_vector_to_chroma(table: str = Body(...)): if not table or not table.startswith("Q") or not table[1:].isdigit(): raise HTTPException(status_code=400, detail="無効なテーブル名です") table_number = int(table[1:]) # if table_number < 1 or table_number > 18: # raise HTTPException(status_code=400, detail="Q1〜Q18の範囲で指定してください") try: # ✅ 해당 테이블의 공통 answer → list.id = n answer_row = await database_add_q.fetch_one( "SELECT answer FROM list WHERE id = :id", values={"id": table_number} ) if not answer_row or not answer_row["answer"]: return {"message": f"listテーブルに id={table_number} の共通answerが見つかりません。"} common_answer = answer_row["answer"].strip() # ✅ 저장되지 않은 질문 가져오기 query = f"SELECT id, Q FROM `{table}` WHERE is_saved = 0" rows = await database_add_q.fetch_all(query) if not rows: return {"message": f"{table} に保存すべき質問がありません。"} success_count = 0 for row in rows: q_id = row["id"] question_text = row["Q"].strip() if not question_text: continue # ✅ 임베딩 생성 response = client.embeddings.create( model="text-embedding-3-large", input=question_text ) embedding = response.data[0].embedding # ✅ ChromaDB 저장 collection_qna.add( ids=[f"{table}_{q_id}"], documents=[question_text], embeddings=[embedding], metadatas=[{ "answer": common_answer, "table": table }] ) # ✅ 저장된 것으로 표시 await database_add_q.execute( f"UPDATE `{table}` SET is_saved = 1 WHERE id = :id", values={"id": q_id} ) success_count += 1 return { "message": f"{table} の質問 {success_count} 件を保存しました（共通answer id={table_number} 使用）" } except Exception as e: logger.exception("💥 ベクトル保存中にエラー発生") raise HTTPException(status_code=500, detail=str(e)) @router.on_event("startup") async def startup_event(): """서버 시작 시 엑셀 파일 내용으로 ChromaDB를 동기화합니다.""" await reload_chroma_from_excel() # ============================================================================== # ✨ 키워드 검색 기능 테스트를 위한 추가 API (기존 시스템과 분리) # ============================================================================== # 1. --- 테스트 환경 설정 --- # 기존 시스템에 영향을 주지 않도록 별도의 파일과 컬렉션 이름을 사용합니다. TEST_EXCEL_PATH = "test_qna_with_keywords.xlsx" TEST_COLLECTION_NAME = "segue_qna_keywords_test" # 2. --- 테스트용 Pydantic 모델 --- class KeywordQnaItem(BaseModel): question: str keywords: str answer: str class SearchQueryItem(BaseModel): query: str # 3. --- 테스트용 핵심 함수 --- def sync_keyword_excel_to_chromadb(): """ ✨ [최종 수정] 엑셀 파일을 기준으로 ChromaDB를 동기화합니다. ✨ 각 항목에 'type' 메타데이터('question' 또는 'keyword')를 추가하여 분리 검색이 가능하도록 합니다. """ try: logging.info(f"'{TEST_EXCEL_PATH}' 파일을 읽어 테스트 DB를 동기화합니다...") df = pd.read_excel(TEST_EXCEL_PATH) df.dropna(subset=['question', 'keywords', 'answer'], inplace=True, how='any') all_documents_to_embed = [] all_metadatas = [] all_ids = [] for idx, row in df.iterrows(): question_text = str(row['question']).strip(' "') keywords_text = str(row['keywords']).strip(' "') answer_text = str(row['answer']).strip(' "') # 1. '질문' 데이터를 저장 목록에 추가 (type: 'question' 꼬리표 추가) all_documents_to_embed.append(question_text) all_metadatas.append({ 'type': 'question', # ✨ 이 항목이 '질문'임을 명시 'question': question_text, 'keywords': keywords_text, 'answer': answer_text }) all_ids.append(f"qna_test_{idx}_question") # 2. '키워드' 데이터를 저장 목록에 추가 (type: 'keyword' 꼬리표 추가) all_documents_to_embed.append(keywords_text) all_metadatas.append({ 'type': 'keyword', # ✨ 이 항목이 '키워드'임을 명시 'question': question_text, 'keywords': keywords_text, 'answer': answer_text }) all_ids.append(f"qna_test_{idx}_keywords") if not all_documents_to_embed: logging.warning("테스트용 엑셀에 데이터가 없습니다.") return response = client.embeddings.create(model="text-embedding-3-small", input=all_documents_to_embed) embeddings = [data.embedding for data in response.data] try: chroma_client.delete_collection(name=TEST_COLLECTION_NAME) except Exception: pass collection = chroma_client.create_collection(name=TEST_COLLECTION_NAME) collection.add( ids=all_ids, embeddings=embeddings, documents=all_documents_to_embed, metadatas=all_metadatas ) logging.info(f"🎉 테스트 DB 동기화 완료! {len(df)}개 행에서 {len(all_documents_to_embed)}개 검색 항목이 로드되었습니다.") except FileNotFoundError: logging.warning(f"'{TEST_EXCEL_PATH}' 파일이 없어 빈 샘플 데이터로 시작합니다.") pd.DataFrame(columns=['question', 'keywords', 'answer']).to_excel(TEST_EXCEL_PATH, index=False) except Exception as e: logging.error(f"🚨 테스트 DB 동기화 중 오류 발생: {e}", exc_info=True) # 4. --- 테스트용 API 엔드포인트 --- @router.on_event("startup") def run_keyword_test_sync_on_startup(): """서버 시작 시, 테스트용 DB도 함께 동기화합니다.""" sync_keyword_excel_to_chromadb() @router.get("/apige/keyword-test/data", tags=["Keyword Search Test"]) def get_all_test_data(): """테스트용 엑셀 파일의 모든 데이터를 반환합니다.""" try: df = pd.read_excel(TEST_EXCEL_PATH) return df.to_dict(orient='records') except FileNotFoundError: return [] @router.post("/apige/keyword-test/add", tags=["Keyword Search Test"]) def add_test_data(item: KeywordQnaItem): """새로운 Q&A를 테스트용 엑셀에 추가하고 DB를 새로고침합니다.""" try: new_data = pd.DataFrame([item.dict()]) try: df = pd.read_excel(TEST_EXCEL_PATH) updated_df = pd.concat([df, new_data], ignore_index=True) except FileNotFoundError: updated_df = new_data updated_df.to_excel(TEST_EXCEL_PATH, index=False) sync_keyword_excel_to_chromadb() # DB 즉시 새로고침 return {"message": "테스트 데이터가 성공적으로 추가 및 반영되었습니다."} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.post("/apige/keyword-test/search", tags=["Keyword Search Test"]) def search_test_data(query_item: SearchQueryItem): """ [최종 버전] 사용자 질문을 키워드로 변환하여, '질문'과 '키워드' DB를 분리 검색 후, 두 거리 값을 모두 포함한 최적의 결과를 반환합니다. """ if not query_item.query: raise HTTPException(status_code=400, detail="검색어가 없습니다.") try: user_query = query_item.query collection = chroma_client.get_collection(name=TEST_COLLECTION_NAME) # 1. GPT로 검색용 키워드 변환 prompt_messages = [ { "role": "system", "content": """あなたは、キーワードを抽出するテキスト分析の専門家です。 以下に提示されるユーザーの質問文から、重要なキーワード「のみ」を抽出することがあなたのタスクです。 # 厳格なルール 1. ユーザーの質問文に「実際に含まれている単語」のみを使用してください。 2. 質問文に存在しない類義語や関連語を絶対に追加しないでください。 3. 名詞と固有名詞の抽出に集中してください。 4. 抽出したキーワードは、カンマ区切りのリスト形式で出力してください。 # 例 ユーザーの質問: "製品について教えてください。" 正しい出力: 製品, 教えて """ }, {"role": "user", "content": user_query} ] keyword_response = client.chat.completions.create(model="gpt-4o-mini", messages=prompt_messages, temperature=0.0) transformed_keywords = keyword_response.choices[0].message.content.strip() logging.info(f"변환된 검색 키워드: {transformed_keywords}") # 2. '원본 질문'과 '변환된 키워드'로 각각 임베딩 생성 response = client.embeddings.create(model="text-embedding-3-small", input=[user_query, transformed_keywords]) query_embedding_for_question = response.data[0].embedding query_embedding_for_keywords = response.data[1].embedding # 3. 분리된 검색 실행 question_results = collection.query(query_embeddings=[query_embedding_for_question], n_results=1, where={"type": "question"}) keyword_results = collection.query(query_embeddings=[query_embedding_for_keywords], n_results=1, where={"type": "keyword"}) q_dist = question_results['distances'][0][0] if question_results['ids'][0] else float('inf') k_dist = keyword_results['distances'][0][0] if keyword_results['ids'][0] else float('inf') logging.info(f"질문 검색 Distance: {q_dist:.4f} | 키워드 검색 Distance: {k_dist:.4f}") # 4. 최종 평가 로직 QUESTION_THRESHOLD = 0.40 KEYWORD_THRESHOLD = 0.60 winner = None final_results = None if q_dist <= QUESTION_THRESHOLD: winner = "질문 검색" final_results = question_results logging.info(f"승자: {winner} (기준점 {QUESTION_THRESHOLD} 통과)") elif k_dist <= KEYWORD_THRESHOLD: winner = "키워드 검색" final_results = keyword_results logging.info(f"승자: {winner} (기준점 {KEYWORD_THRESHOLD} 통과)") else: logging.info("두 검색 모두 기준점을 통과하지 못했습니다. 더 가까운 쪽을 선택합니다.") if q_dist < k_dist: winner = "질문 검색 (기준점 미달)" final_results = question_results else: winner = "키워드 검색 (기준점 미달)" final_results = keyword_results if not final_results or not final_results['ids'][0]: return {"message": "검색 결과가 없습니다."} # 5. 프론트엔드에 보낼 최종 데이터 조립 best_match = { "winner": winner, "question_distance": q_dist if q_dist != float('inf') else -1, # 무한대 값은 -1로 변환 "keyword_distance": k_dist if k_dist != float('inf') else -1, # 무한대 값은 -1로 변환 "retrieved_document": final_results['documents'][0][0], "transformed_keywords": transformed_keywords, "question": final_results['metadatas'][0][0].get('question'), "keywords": final_results['metadatas'][0][0].get('keywords'), "answer": final_results['metadatas'][0][0].get('answer') } return best_match except Exception as e: logging.error(f"🚨 검색 중 오류 발생: {e}", exc_info=True) raise HTTPException(status_code=500, detail=str(e)) @router.post("/apige/keyword-test/reload", tags=["Keyword Search Test"]) def reload_test_database(): """관리자가 수동으로 테스트 DB 새로고침을 명령합니다.""" try: sync_keyword_excel_to_chromadb() return {"message": "테스트 데이터베이스가 엑셀 파일의 최신 내용으로 새로고침되었습니다."} except Exception as e: raise HTTPException(status_code=500, detail=str(e))