feat(rag/ui/eval): sync local workflow, app, parser, prompts and eval scripts

Author: panda

app/components.py

@@ -11,24 +11,13 @@
 import streamlit as st
 
 
-_COMPONENTS_DIR = Path(__file__).parent
-_DATA_CSV_PATH = _COMPONENTS_DIR.parent / "data_index" / "data" / "data_list.csv"
-
-FAQ_EVAL_ITEMS: list[dict] = [
-    {
-        "label": "정부기관 사업조회",
-        "question": "등록된 정부기관의 사업 목록을 전체 조회해 주세요.",
-        "institution": "",
-        "keyword": "",
-        "answer_type": "project_list",
-    },
-    {
-        "label": "sw관련 사업 조회",
-        "question": "SW(소프트웨어) 관련 사업의 목록과 예산을 조회해 주세요.",
-        "institution": "",
-        "keyword": "SW",
-        "answer_type": "project_list",
-    },
+FAQ_QUESTIONS = [
+    "예산 규모와 집행 조건을 요약해줘",
+    "사업 기간과 주요 마일스톤을 정리해줘",
+    "참가 자격 요건을 알려줘",
+    "필수 제출서류 체크리스트를 만들어줘",
+    "평가 항목과 배점을 정리해줘",
+    "마감일과 일정 리스크를 알려줘",
 ]
 
 CARD_META: dict[str, dict[str, str]] = {
@@ -41,109 +30,6 @@
 }
 
 
-def _build_csv_answer(df: pd.DataFrame, answer_type: str) -> str:
-    """CSV 데이터프레임에서 answer_type에 맞는 답변 텍스트를 생성합니다."""
-    def _fmt_amount(val: Any) -> str:
-        try:
-            return f"{int(float(val)):,}원" if pd.notna(val) and str(val).strip() else "정보 없음"
-        except (ValueError, TypeError):
-            return str(val)
-
-    def _clean(val: Any) -> str:
-        s = str(val).strip()
-        return "" if s.lower() in ("nan", "none", "") else s
-
-    if answer_type == "budget":
-        row = df.iloc[0]
-        proj = _clean(row.get("사업명", ""))
-        org = _clean(row.get("발주 기관", ""))
-        amount_fmt = _fmt_amount(row.get("사업 금액", ""))
-        return f"**{proj}** ({org})\n\n사업 금액: **{amount_fmt}** (VAT 포함)"
-
-    if answer_type == "project_list":
-        org = _clean(df.iloc[0].get("발주 기관", ""))
-        projects = [_clean(p) for p in df["사업명"].dropna().tolist() if _clean(p)]
-        lines = [f"**{org}** 진행 사업 — 총 {len(projects)}개\n"]
-        lines += [f"- {p}" for p in projects]
-        return "\n".join(lines)
-
-    if answer_type == "schedule":
-        row = df.iloc[0]
-        proj = _clean(row.get("사업명", ""))
-        start = _clean(row.get("입찰 참여 시작일", "")) or "정보 없음"
-        end = _clean(row.get("입찰 참여 마감일", "")) or "정보 없음"
-        return (
-            f"**{proj}**\n\n"
-            f"- 입찰 참여 시작일: **{start}**\n"
-            f"- 입찰 참여 마감일: **{end}**"
-        )
-
-    # summary (default)
-    row = df.iloc[0]
-    proj = _clean(row.get("사업명", ""))
-    org = _clean(row.get("발주 기관", ""))
-    amount_fmt = _fmt_amount(row.get("사업 금액", ""))
-    start = _clean(row.get("입찰 참여 시작일", ""))
-    end = _clean(row.get("입찰 참여 마감일", ""))
-    summary_text = _clean(row.get("사업 요약", ""))
-
-    parts = [f"**{proj}** ({org})\n", f"- 사업 금액: **{amount_fmt}**"]
-    if start:
-        parts.append(f"- 입찰 참여 시작일: {start}")
-    if end:
-        parts.append(f"- 입찰 참여 마감일: {end}")
-    if summary_text:
-        short = summary_text[:400] + ("…" if len(summary_text) > 400 else "")
-        parts.append(f"\n**사업 요약:**\n{short}")
-    return "\n".join(parts)
-
-
-def answer_from_csv(faq_item: dict) -> dict[str, Any]:
-    """FAQ 항목을 data_list.csv에서 직접 조회하여 답변을 생성합니다."""
-    try:
-        df = pd.read_csv(_DATA_CSV_PATH, encoding="utf-8-sig")
-    except UnicodeDecodeError:
-        try:
-            df = pd.read_csv(_DATA_CSV_PATH, encoding="utf-8")
-        except Exception as exc:
-            return {"answer": f"CSV 파일 로드 실패: {exc}", "evidence": []}
-    except Exception as exc:
-        return {"answer": f"CSV 파일 로드 실패: {exc}", "evidence": []}
-
-    institution = faq_item.get("institution", "")
-    keyword = faq_item.get("keyword", "")
-    answer_type = faq_item.get("answer_type", "summary")
-
-    mask = pd.Series([True] * len(df), index=df.index)
-    if institution:
-        mask &= df["발주 기관"].str.contains(institution, na=False, case=False)
-    if keyword:
-        mask &= df["사업명"].str.contains(keyword, na=False, case=False)
-    filtered = df[mask].copy()
-
-    # keyword 조건 미충족 시 institution만으로 재시도
-    if filtered.empty and institution:
-        inst_mask = df["발주 기관"].str.contains(institution, na=False, case=False)
-        filtered = df[inst_mask].copy()
-
-    if filtered.empty:
-        return {
-            "answer": f"'{institution}' 관련 데이터를 CSV에서 찾을 수 없습니다.",
-            "evidence": [],
-        }
-
-    answer_text = _build_csv_answer(filtered, answer_type)
-    evidence = [
-        {
-            "source": str(row.get("파일명", "") or "").strip() or "data_list.csv",
-            "page": None,
-            "text": f"{row.get('사업명', '')} | {row.get('발주 기관', '')}",
-        }
-        for _, row in filtered.head(5).iterrows()
-    ]
-    return {"answer": answer_text, "evidence": evidence}
-
-
 def inject_styles() -> None:
     st.markdown(
         """
@@ -302,32 +188,16 @@ def inject_styles() -> None:
         .section-title:first-child { margin-top: 0; }
         .answer-summary {
             color: #0F172A;
-            font-size: 0.9rem;
+            font-size: 0.95rem;
             line-height: 1.7;
         }
-        .answer-bullets {
-            margin: 4px 0 0;
-            padding-left: 20px;
-        }
-        .answer-bullet {
-            font-size: 0.9rem;
-            line-height: 1.6;
-            color: #0F172A;
-            margin-bottom: 4px;
-        }
-        .answer-caption {
-            font-size: 0.85rem;
-            color: #64748B;
-            margin-top: 4px;
-            line-height: 1.5;
-        }
         .answer-evidence {
             background: #F8FAFC;
             border-left: 3px solid #10A37F;
             border-radius: 0 8px 8px 0;
             padding: 8px 14px;
             margin-top: 6px;
-            font-size: 0.9rem;
+            font-size: 0.82rem;
             color: #475569;
             line-height: 1.5;
         }
@@ -472,20 +342,6 @@ def inject_styles() -> None:
             color: #0F172A !important;
             letter-spacing: -0.01em;
         }
-
-        /* ── Heading normalization inside content areas ── */
-        .answer-box h1, .answer-box h2, .answer-box h3,
-        .answer-box h4, .answer-box h5, .answer-box h6,
-        .answer-summary h1, .answer-summary h2, .answer-summary h3,
-        .answer-summary h4, .answer-summary h5, .answer-summary h6,
-        .answer-bullet h1, .answer-bullet h2, .answer-bullet h3 {
-            font-size: 0.9rem !important;
-            font-weight: 600 !important;
-            color: #0F172A !important;
-            margin: 4px 0 !important;
-            padding: 0 !important;
-            border: none !important;
-        }
         </style>
         """,
         unsafe_allow_html=True,
@@ -548,14 +404,9 @@ def _sentence_split(text: str) -> list[str]:
     return [p.strip() for p in parts if p.strip()]
 
 
-def _strip_heading_markers(text: str) -> str:
-    """마크다운 헤딩 기호(#~######)를 제거합니다."""
-    return re.sub(r"^#{1,6}\s+", "", text.strip())
-
-
 def parse_answer_payload(result: dict[str, Any]) -> tuple[str, list[str], list[dict[str, Any]]]:
     answer = str(result.get("answer", "") or "").strip()
-    sentences = [_strip_heading_markers(s) for s in _sentence_split(answer)]
+    sentences = _sentence_split(answer)
     summary = " ".join(sentences[:2]) if sentences else "답변이 생성되지 않았습니다."
 
     bullets: list[str] = []
@@ -564,14 +415,10 @@ def parse_answer_payload(result: dict[str, Any]) -> tuple[str, list[str], list[d
         if not stripped:
             continue
         if stripped.startswith("-") or stripped.startswith("*"):
-            bullets.append(_strip_heading_markers(stripped.lstrip("-* ")))
+            bullets.append(stripped.lstrip("-* "))
         elif stripped.startswith("|"):
             continue
-        elif stripped.startswith("#"):
-            clean = _strip_heading_markers(stripped)
-            if len(bullets) < 5 and len(clean) > 2:
-                bullets.append(clean)
-        elif len(bullets) < 5 and len(stripped) > 8:
+        elif len(bullets) < 5 and len(stripped) > 8 and not stripped.startswith("###"):
             bullets.append(stripped)
         if len(bullets) >= 5:
             break
@@ -583,21 +430,18 @@ def parse_answer_payload(result: dict[str, Any]) -> tuple[str, list[str], list[d
 
 
 def render_answer_sections(result: dict[str, Any]) -> None:
-    # raw_answer: RAG/LLM이 생성한 원본 답변 (post-processing 이전)
-    # answer: 후처리 완료 답변 (raw_answer가 없을 때 fallback)
-    raw = str(result.get("raw_answer", "") or "").strip()
-    answer = raw if raw else str(result.get("answer", "") or "").strip()
-    evidence: list[dict[str, Any]] = result.get("evidence", [])
-    if not isinstance(evidence, list):
-        evidence = []
-
+    summary, bullets, evidence = parse_answer_payload(result)
     st.markdown('<div class="answer-box">', unsafe_allow_html=True)
 
-    st.markdown("<div class='section-title'>📝 답변</div>", unsafe_allow_html=True)
-    if answer:
-        st.markdown(answer)
+    st.markdown("<div class='section-title'>📝 요약</div>", unsafe_allow_html=True)
+    st.markdown(f"<div class='answer-summary'>{summary}</div>", unsafe_allow_html=True)
+
+    st.markdown("<div class='section-title'>💡 핵심 포인트</div>", unsafe_allow_html=True)
+    if bullets:
+        for item in bullets:
+            st.markdown(f"- {item}")
     else:
-        st.markdown("<div class='answer-caption'>답변이 생성되지 않았습니다.</div>", unsafe_allow_html=True)
+        st.caption("핵심 포인트를 자동 추출하지 못했습니다.")
 
     st.markdown("<div class='section-title'>📎 근거 (페이지/섹션)</div>", unsafe_allow_html=True)
     if evidence:
@@ -608,7 +452,7 @@ def render_answer_sections(result: dict[str, Any]) -> None:
 
             page_val = str(page).strip() if page not in (None, "", "-", "None") else ""
             page_badge = (
-                f" <span style='font-size:0.75rem;background:#ECFDF5;border:1px solid #10A37F;"
+                f" <span style='font-size:0.7rem;background:#ECFDF5;border:1px solid #10A37F;"
                 f"border-radius:4px;padding:1px 6px;color:#10A37F;font-weight:600;"
                 f"vertical-align:middle;'>p.{page_val}</span>"
                 if page_val else ""
@@ -620,9 +464,10 @@ def render_answer_sections(result: dict[str, Any]) -> None:
                 unsafe_allow_html=True,
             )
             if text:
-                st.markdown(f"<div class='answer-caption'>{text}</div>", unsafe_allow_html=True)
+                preview = text[:200] + ("…" if len(text) > 200 else "")
+                st.caption(preview)
     else:
-        st.markdown("<div class='answer-caption'>근거 정보가 제공되지 않았습니다.</div>", unsafe_allow_html=True)
+        st.caption("근거 정보가 제공되지 않았습니다.")
     st.markdown("</div>", unsafe_allow_html=True)
 
 
@@ -668,28 +513,16 @@ def _split_filename(filename: str) -> tuple[str, str]:
     return org, project
 
 
-def parse_budget_input(val_str: str, default: int) -> int:
-    """쉼표가 포함된 예산 문자열에서 정수를 파싱합니다."""
-    cleaned = re.sub(r"[,\s원]", "", str(val_str))
-    try:
-        result = int(float(cleaned))
-        return result if result >= 0 else default
-    except (ValueError, TypeError):
-        return default
-
-
 def real_bids_dataframe(chatbot: Any) -> pd.DataFrame:
-    """벡터스토어 org_registry, CSV, 또는 파일명 파싱으로 실제 입찰 목록을 생성합니다.
+    """벡터스토어 org_registry 또는 파일명 파싱으로 실제 입찰 목록을 생성합니다.
 
     우선순위:
     1. chatbot.vector_store.org_registry (기관명·예산·사업명 추출 완료)
-    2. data/data_list.csv (예산·마감일 포함 완전 데이터)
-    3. data/files/ 파일명 파싱 (예산·마감일 없음)
-    4. 빈 DataFrame 반환 (데이터 없음 안내)
+    2. data/files/ 파일명 파싱 (예산·마감일 없음)
+    3. 빈 DataFrame 반환 (데이터 없음 안내)
     """
     rows: list[dict] = []
 
-    # Priority 1: org_registry
     try:
         registry = chatbot.vector_store.org_registry
         if registry:
@@ -710,60 +543,25 @@ def real_bids_dataframe(chatbot: Any) -> pd.DataFrame:
     except Exception:
         pass
 
-    if rows and any(r.get("예산") is not None or r.get("마감일") is not None for r in rows):
-        return pd.DataFrame(rows)
-
-    rows = []  # 유효 데이터 없으면 CSV로 재시도
-
-    # Priority 2: CSV data (예산·마감일 포함)
-    try:
-        df_csv = pd.read_csv(_DATA_CSV_PATH, encoding="utf-8-sig")
-        for _, row in df_csv.iterrows():
-            budget_val: int | None = None
-            amount = row.get("사업 금액")
-            if pd.notna(amount):
-                try:
-                    budget_val = int(float(amount))
-                except (ValueError, TypeError):
-                    pass
-            deadline = _parse_date_safe(str(row.get("입찰 참여 마감일", "") or ""))
-            공고번호_raw = str(row.get("공고 번호", "")).strip()
-            공고번호 = (
-                공고번호_raw
-                if 공고번호_raw not in ("", "nan", "None")
-                else f"RFP-{len(rows) + 1:03d}"
-            )
-            rows.append({
-                "공고번호": 공고번호,
-                "사업명": str(row.get("사업명", "") or "").strip(),
-                "기관": str(row.get("발주 기관", "") or "").strip(),
-                "예산": budget_val,
-                "마감일": deadline,
-            })
-        if rows:
-            return pd.DataFrame(rows)
-    except Exception:
-        pass
-
-    # Priority 3: files directory (파일명 파싱만)
-    try:
-        from src.utils.config import get_data_dir
-        files_dir = get_data_dir() / "files"
-        if files_dir.exists():
-            all_files: list[Path] = []
-            for pat in ("*.hwp", "*.hwpx", "*.pdf"):
-                all_files.extend(sorted(files_dir.glob(pat)))
-            for idx, f in enumerate(all_files, start=1):
-                org, project = _split_filename(f.name)
-                rows.append({
-                    "공고번호": f"RFP-{idx:03d}",
-                    "사업명": project,
-                    "기관": org,
-                    "예산": None,
-                    "마감일": None,
-                })
-    except Exception:
-        pass
+    if not rows:
+        try:
+            from src.utils.config import get_data_dir
+            files_dir = get_data_dir() / "files"
+            if files_dir.exists():
+                all_files: list[Path] = []
+                for pat in ("*.hwp", "*.hwpx", "*.pdf"):
+                    all_files.extend(sorted(files_dir.glob(pat)))
+                for idx, f in enumerate(all_files, start=1):
+                    org, project = _split_filename(f.name)
+                    rows.append({
+                        "공고번호": f"RFP-{idx:03d}",
+                        "사업명": project,
+                        "기관": org,
+                        "예산": None,
+                        "마감일": None,
+                    })
+        except Exception:
+            pass
 
     if not rows:
         return pd.DataFrame(columns=["공고번호", "사업명", "기관", "예산", "마감일"])