diff --git a/docs/design/740-smilegate-fewshot-management-tab/README.md b/docs/design/740-smilegate-fewshot-management-tab/README.md new file mode 100644 index 0000000..a96e7ee --- /dev/null +++ b/docs/design/740-smilegate-fewshot-management-tab/README.md @@ -0,0 +1,30 @@ +# Smilegate 질문별 Few-shot 관리 탭 (#740) + +## 프로젝트 개요 + +Smilegate Portal은 고객 질답 벤치마크를 실행하고, Oracle Select AI MCP가 승인된 few-shot 예제를 검색해 읽기 전용 SQL을 생성·실행하는 서비스다. 고객 질문, 실행 이력, few-shot 예제는 Oracle ADB의 `SG_AI_QA_QUESTION`, `SG_AI_QA_ANSWER`, `SG_QA_VECTOR_EXAMPLE`에 보관한다. + +## 목표 + +Portal에 **Few-shot 관리** 탭을 추가해 운영자가 질문별 few-shot 내용을 확인하고 수정할 수 있게 한다. + +## 설계 + +- 질문 목록은 기존 고객 QA 기준 파일을 사용하고, 예제는 `source_type=CUSTOMER_QA_BENCHMARK` 및 `source_case_id=질문 코드`로 연결한다. +- 탭은 질문, 예제 SQL, 설명, 참조 종류, 적용 게임 범위, 논리 객체 역할, 검토 메모, 검색 상태를 표시한다. +- 저장은 예제 본문과 메타데이터를 갱신하고 embedding 입력과 vector embedding을 함께 재생성한다. 따라서 수정 후 벡터 검색이 수정 전 내용과 어긋나지 않는다. +- 저장 시 상태를 자동 승인하지 않는다. `APPROVED`는 기존 승인 정책에 맞는 운영자 선택으로만 유지·변경하며, `RETIRED` 예제는 검색에서 제외된다. +- 물리 게임 테이블명은 예제의 정책 값으로 넣지 않고 논리 `object_role`만 보관한다. 게임 범위는 현재 query plan이 권위 있는 원천이다. + +## 완료 기준 + +1. 인증된 Portal에서 기존 아키텍처·시나리오와 함께 Few-shot 관리 탭을 볼 수 있다. +2. 질문을 고르면 해당 few-shot 예제의 전체 내용을 확인할 수 있다. +3. 유효한 값으로 수정 후 저장하면 다시 열었을 때 저장 내용이 보인다. +4. Python 단위 테스트가 저장 입력 검증과 DB 갱신 SQL을 확인한다. + +## 검증 계획 + +- `python -m pytest`로 Portal 테스트를 실행한다. +- Python compile 검사로 Streamlit 모듈 구문을 확인한다. +- DB 접속 설정이 있는 환경에서는 질문별 예제 조회와 수정 후 재조회로 검증한다. diff --git a/poc4_active_source_20260714/apps/poc4/mcp_discovery_ui.py b/poc4_active_source_20260714/apps/poc4/mcp_discovery_ui.py index a0391ec..fef5a03 100644 --- a/poc4_active_source_20260714/apps/poc4/mcp_discovery_ui.py +++ b/poc4_active_source_20260714/apps/poc4/mcp_discovery_ui.py @@ -57,7 +57,14 @@ from src.agent_console.presentation import ( ) from src.agent_console.profile import AppProfile, AppProfileError, load_app_profile from src.poc4.qa_history import QaJudgment, QaQuestion, evaluate_sql, load_benchmark_questions -from src.poc4.qa_history_store import QaHistoryStore, QaHistoryStoreError +from src.poc4.qa_history_store import ( + INSPECTION_STATUSES, + REFERENCE_KINDS, + REFERENCE_STATUSES, + TARGET_TYPES, + QaHistoryStore, + QaHistoryStoreError, +) LOG = logging.getLogger(__name__) @@ -2066,6 +2073,151 @@ def _render_qa_benchmark_panel( return selected +def _render_few_shot_management_tab( + questions: list[QaQuestion], + store: QaHistoryStore | None, + store_error: str, +) -> None: + st.subheader("질문별 Few-shot 관리") + st.caption( + "고객 질답 질문에 연결된 few-shot 예제를 검토하고 수정합니다. 저장하면 검색용 embedding도 함께 다시 생성됩니다." + ) + if store_error or store is None: + st.warning( + "Few-shot 관리 DB에 연결하지 못했습니다. " + + (store_error or "질답 이력 DB 접속 설정을 확인해 주세요.") + ) + return + if not questions: + st.info("표시할 고객 질답 질문이 없습니다.") + return + + by_code = {question.question_code: question for question in questions} + selected_code = st.selectbox( + "질문 선택", + options=tuple(by_code), + format_func=lambda code: _question_label(by_code[code]), + key="poc4_few_shot_question_code", + ) + selected_question = by_code[selected_code] + st.markdown("**선택한 질문**") + st.write(selected_question.question_text) + try: + examples = store.list_vector_examples_by_case(selected_question.question_code) + except QaHistoryStoreError as exc: + st.error(str(exc)) + return + if not examples: + st.info( + "이 질문에는 아직 연결된 few-shot 후보가 없습니다. " + "후보 적재 작업 후 이 화면에서 검토할 수 있습니다." + ) + return + + example_by_id = {str(item["example_id"]): item for item in examples} + selected_example_id = st.selectbox( + "Few-shot 예제", + options=tuple(example_by_id), + format_func=lambda example_id: ( + f"예제 #{example_id} · " + f"{example_by_id[example_id]['reference_status']} · " + f"{example_by_id[example_id]['inspection_status']}" + ), + key=f"poc4_few_shot_example_{selected_question.question_code}", + ) + example = example_by_id[selected_example_id] + st.caption( + "검색 대상은 APPROVED 예제뿐입니다. 게임별 물리 테이블은 여기에 넣지 않고, " + "현재 질문의 query plan이 해석하는 논리 객체 역할만 관리합니다." + ) + if example["verified_at"]: + st.caption( + f"마지막 승인: {example['verified_at']} · {example['verified_by'] or '확인자 미기록'}" + ) + + form_key = f"poc4_few_shot_form_{example['example_id']}" + with st.form(form_key): + question = st.text_area( + "Few-shot 질문", + value=example["question"], + height=100, + max_chars=4_000, + ) + answer_sql = st.text_area( + "Few-shot SQL 템플릿", + value=example["answer_sql"], + height=220, + max_chars=12_000, + help="읽기 전용 SQL 템플릿을 사용하세요. 게임별 물리 객체는 논리 placeholder로 표현합니다.", + ) + answer_text = st.text_area( + "설명 / 답변 보조 문구", + value=example["answer_text"], + height=120, + max_chars=8_000, + ) + left, right = st.columns(2) + with left: + reference_status = st.selectbox( + "검색 상태", + options=REFERENCE_STATUSES, + index=REFERENCE_STATUSES.index(example["reference_status"]), + help="APPROVED만 실제 few-shot 검색에 사용됩니다.", + ) + reference_kind = st.selectbox( + "참조 종류", + options=REFERENCE_KINDS, + index=REFERENCE_KINDS.index(example["reference_kind"]), + ) + with right: + target_type = st.selectbox( + "적용 게임 범위", + options=TARGET_TYPES, + index=TARGET_TYPES.index(example["target_type"]), + ) + inspection_status = st.selectbox( + "검토 상태", + options=INSPECTION_STATUSES, + index=INSPECTION_STATUSES.index(example["inspection_status"]), + ) + object_role = st.text_input( + "논리 객체 역할", + value=example["object_role"], + max_chars=64, + placeholder="예: GAME_USER_MASTER", + ) + inspection_note = st.text_area( + "검토 메모", + value=example["inspection_note"], + height=100, + max_chars=4_000, + ) + submitted = st.form_submit_button("Few-shot 변경 저장", type="primary") + + if not submitted: + return + try: + store.update_vector_example( + example_id=example["example_id"], + source_case_id=selected_question.question_code, + question=question, + answer_sql=answer_sql, + answer_text=answer_text, + reference_status=reference_status, + reference_kind=reference_kind, + target_type=target_type, + object_role=object_role, + inspection_status=inspection_status, + inspection_note=inspection_note, + verified_by=str(st.session_state.get(PORTAL_AUTH_USER_KEY) or "PORTAL_OPERATOR"), + ) + except QaHistoryStoreError as exc: + st.error(str(exc)) + return + st.success("Few-shot 예제와 검색용 embedding을 저장했습니다.") + st.rerun() + + def _apply_console_theme(profile: AppProfile) -> None: apply_console_theme(st, profile) @@ -7515,7 +7667,9 @@ def main() -> None: '
', unsafe_allow_html=True, ) - architecture_tab, scenario_tab = st.tabs(["아키텍처", "시나리오"]) + architecture_tab, scenario_tab, few_shot_tab = st.tabs( + ["아키텍처", "시나리오", "Few-shot 관리"] + ) with architecture_tab: _render_architecture_tab() @@ -7617,6 +7771,13 @@ def main() -> None: height=0, ) + with few_shot_tab: + _render_few_shot_management_tab( + qa_questions, + qa_history_store, + qa_store_error, + ) + if not submitted: return diff --git a/poc4_active_source_20260714/src/poc4/qa_history_store.py b/poc4_active_source_20260714/src/poc4/qa_history_store.py index 5d3b39a..0fac70f 100644 --- a/poc4_active_source_20260714/src/poc4/qa_history_store.py +++ b/poc4_active_source_20260714/src/poc4/qa_history_store.py @@ -21,7 +21,12 @@ class QaHistoryStoreError(RuntimeError): QUESTION_TABLE = "SG_AI_QA_QUESTION" ANSWER_TABLE = "SG_AI_QA_ANSWER" +VECTOR_EXAMPLE_TABLE = "SG_QA_VECTOR_EXAMPLE" HISTORICAL_RUN_KEY = "HISTORICAL:2026-07-21:term-dict-final-v2" +REFERENCE_STATUSES = ("DRAFT", "APPROVED", "RETIRED") +REFERENCE_KINDS = ("SQL_TEMPLATE", "NO_TARGET", "OBJECT_UNAVAILABLE", "METADATA_POLICY") +TARGET_TYPES = ("NONE", "SINGLE", "MULTI", "ALL", "ANY") +INSPECTION_STATUSES = ("PENDING", "REVIEW", "VERIFIED", "RETIRED") def _env_value(name: str, env_file: Path | None = None) -> str: @@ -110,6 +115,32 @@ def _answer_record(row: Mapping[str, Any]) -> dict[str, Any]: } +def _vector_example_record(row: Mapping[str, Any]) -> dict[str, Any]: + return { + "example_id": int(row["example_id"]), + "question": str(row.get("question") or ""), + "answer_sql": str(row.get("answer_sql") or ""), + "answer_text": str(row.get("answer_text") or ""), + "reference_status": str(row.get("reference_status") or "DRAFT"), + "reference_kind": str(row.get("reference_kind") or "SQL_TEMPLATE"), + "target_type": str(row.get("target_type") or "ANY"), + "object_role": str(row.get("object_role") or ""), + "inspection_status": str(row.get("inspection_status") or "PENDING"), + "inspection_note": str(row.get("inspection_note") or ""), + "verified_at": str(row.get("verified_at") or ""), + "verified_by": str(row.get("verified_by") or ""), + "source_case_id": str(row.get("source_case_id") or ""), + "source_type": str(row.get("source_type") or ""), + } + + +def _choice(value: object, choices: tuple[str, ...], field: str) -> str: + normalized = str(value or "").strip().upper() + if normalized not in choices: + raise QaHistoryStoreError(f"{field} 값을 확인해 주세요.") + return normalized + + class QaHistoryStore: def __init__(self, *, env_file: Path | None = None) -> None: self._env_file = env_file @@ -225,6 +256,106 @@ class QaHistoryStore: row = cursor.fetchone() return question_from_record(_record_from_cursor(cursor, row)) if row else None + def list_vector_examples_by_case(self, question_code: str) -> list[dict[str, Any]]: + """Return governed few-shot examples linked to one customer QA case.""" + case_id = str(question_code or "").strip().upper() + if not case_id: + raise QaHistoryStoreError("질문 코드를 확인해 주세요.") + sql = f""" + SELECT example_id, question, answer_sql, answer_text, + reference_status, reference_kind, target_type, object_role, + inspection_status, inspection_note, + TO_CHAR(verified_at AT TIME ZONE 'Asia/Seoul', + 'YYYY-MM-DD HH24:MI:SS TZH:TZM') AS verified_at, + verified_by, source_case_id, source_type + FROM {VECTOR_EXAMPLE_TABLE} + WHERE source_type = 'CUSTOMER_QA_BENCHMARK' + AND source_case_id = :source_case_id + ORDER BY example_id + """ + with self._connection() as connection: + with connection.cursor() as cursor: + cursor.execute(sql, {"source_case_id": case_id}) + rows = [_record_from_cursor(cursor, row) for row in cursor] + return [_vector_example_record(row) for row in rows] + + def update_vector_example( + self, + *, + example_id: int, + source_case_id: str, + question: str, + answer_sql: str, + answer_text: str, + reference_status: str, + reference_kind: str, + target_type: str, + object_role: str, + inspection_status: str, + inspection_note: str, + verified_by: str, + ) -> None: + """Update one customer example and rebuild its vector document atomically.""" + case_id = str(source_case_id or "").strip().upper() + normalized_question = str(question or "").strip() + normalized_sql = str(answer_sql or "").strip() + if not case_id or not normalized_question or not normalized_sql: + raise QaHistoryStoreError("질문과 예제 SQL은 비워둘 수 없습니다.") + status = _choice(reference_status, REFERENCE_STATUSES, "검색 상태") + kind = _choice(reference_kind, REFERENCE_KINDS, "참조 종류") + target = _choice(target_type, TARGET_TYPES, "적용 범위") + inspection = _choice(inspection_status, INSPECTION_STATUSES, "검토 상태") + sql = f""" + UPDATE {VECTOR_EXAMPLE_TABLE} + SET question = :question, + answer_sql = :answer_sql, + answer_text = :answer_text, + embedding_input = TO_CLOB('Customer QA question: ') || :question + || TO_CLOB(CHR(10) || 'Few-shot SQL template: ') || :answer_sql + || CASE WHEN :answer_text IS NULL THEN NULL + ELSE TO_CLOB(CHR(10) || 'Few-shot answer: ') || :answer_text END, + embedding = DBMS_VECTOR.UTL_TO_EMBEDDING( + TO_CLOB('Customer QA question: ') || :question + || TO_CLOB(CHR(10) || 'Few-shot SQL template: ') || :answer_sql + || CASE WHEN :answer_text IS NULL THEN NULL + ELSE TO_CLOB(CHR(10) || 'Few-shot answer: ') || :answer_text END, + JSON(sg_qa_vector_params('search_document')) + ), + reference_status = :reference_status, + reference_kind = :reference_kind, + target_type = :target_type, + object_role = :object_role, + inspection_status = :inspection_status, + inspection_note = :inspection_note, + verified_at = CASE WHEN :reference_status = 'APPROVED' + THEN SYSTIMESTAMP ELSE NULL END, + verified_by = CASE WHEN :reference_status = 'APPROVED' + THEN :verified_by ELSE NULL END + WHERE example_id = :example_id + AND source_type = 'CUSTOMER_QA_BENCHMARK' + AND source_case_id = :source_case_id + """ + binds = { + "example_id": int(example_id), + "source_case_id": case_id, + "question": normalized_question, + "answer_sql": normalized_sql, + "answer_text": str(answer_text or "").strip() or None, + "reference_status": status, + "reference_kind": kind, + "target_type": target, + "object_role": str(object_role or "").strip()[:64] or None, + "inspection_status": inspection, + "inspection_note": str(inspection_note or "").strip() or None, + "verified_by": str(verified_by or "").strip()[:128] or "PORTAL_OPERATOR", + } + with self._connection() as connection: + with connection.cursor() as cursor: + cursor.execute(sql, binds) + if cursor.rowcount != 1: + raise QaHistoryStoreError("수정할 질문별 few-shot 예제를 찾지 못했습니다.") + connection.commit() + def list_answers(self, question_id: int, *, limit: int = 30) -> list[dict[str, Any]]: sql = f""" SELECT answer_seq, question_id, answer_kind, run_key, conversation_id, diff --git a/poc4_active_source_20260714/tests/test_qa_history.py b/poc4_active_source_20260714/tests/test_qa_history.py index a0cc8b8..d1dbae5 100644 --- a/poc4_active_source_20260714/tests/test_qa_history.py +++ b/poc4_active_source_20260714/tests/test_qa_history.py @@ -4,7 +4,13 @@ from pathlib import Path import unittest from src.poc4.qa_history import evaluate_sql, load_benchmark_questions -from src.poc4.qa_history_store import _normalize_oracle_dsn, schema_statements +from src.poc4.qa_history_store import ( + REFERENCE_STATUSES, + QaHistoryStoreError, + _choice, + _normalize_oracle_dsn, + schema_statements, +) class QaHistoryTest(unittest.TestCase): @@ -75,6 +81,11 @@ class QaHistoryTest(unittest.TestCase): self.assertIn("CREATE TABLE SG_AI_QA_ANSWER", statements) self.assertIn("answer_seq NUMBER GENERATED ALWAYS AS IDENTITY", statements) + def test_few_shot_status_is_normalized_and_unknown_status_is_rejected(self) -> None: + self.assertEqual("APPROVED", _choice(" approved ", REFERENCE_STATUSES, "검색 상태")) + with self.assertRaises(QaHistoryStoreError): + _choice("LIVE", REFERENCE_STATUSES, "검색 상태") + if __name__ == "__main__": unittest.main()