joungmin/vpd-permission-poc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refs #740: add Few-shot management tab

Browse Source
This commit is contained in:
devmrko
2026-07-28 15:46:53 +09:00
parent 041ab4f287
commit 9d2c75459f
4 changed files with 336 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

165
poc4_active_source_20260714/apps/poc4/mcp_discovery_ui.py
View File

@@ -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:
'<div id="kb-main-tabs-anchor" style="scroll-margin-top: 0.75rem;"></div>',
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

131
poc4_active_source_20260714/src/poc4/qa_history_store.py
View File

@@ -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,

13
poc4_active_source_20260714/tests/test_qa_history.py
View File

@@ -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()