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

refs #706: add Smilegate QA history benchmark

Browse Source
This commit is contained in:
devmrko
2026-07-23 10:12:36 +09:00
parent b21d7ad01b
commit a7ea010f5c
8 changed files with 3270 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

381
ai-web-agent-console/app.py
View File

@@ -60,15 +60,9 @@ from ai_web_agent_console.presentation import (
render_console_header,
render_login_brand,
)
from ai_web_agent_console.audit import render_hmm_audit_tab
from ai_web_agent_console.profile import AppProfile, AppProfileError, load_app_profile
from ai_web_agent_console.scenarios import ScenarioConfigError, load_demo_scenarios
from ai_web_agent_console.query_contracts import (
append_query_contract_guidance,
evidence_contract_report,
matching_query_contracts,
missing_evidence_message,
)
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
LOG = logging.getLogger(__name__)
@@ -91,7 +85,7 @@ DEFAULT_QUERY_MODEL_PROFILE = "gpt54_mini_oci"
ENV_FILE = ROOT / ".env"
MCP_SERVERS_FILE = ROOT / "config" / "mcp_servers.json"
VPD_TOKEN_PRESETS_FILE = ROOT / "config" / "vpd_token_presets.json"
DEMO_SCENARIOS_FILE = ROOT / "config" / "smilegate_demo_scenarios.json"
QA_BENCHMARK_FILE = ROOT / "config" / "smilegate_qa_benchmark.json"
APP_PROFILE_FILE = ROOT / "config" / "app_profile.json"
CHAT_DB_FILE = ROOT / "data" / "poc4_mcp_chat.sqlite3"
DEFAULT_VPD_USER_ID = "1001"
@@ -559,12 +553,180 @@ def new_conversation_id() -> str:
def _question_label(question: object) -> str:
question_id = str(getattr(question, "question_id"))
category = str(getattr(question, "category"))
title = str(getattr(question, "title", getattr(question, "text")))
question_id = str(
getattr(question, "question_code", getattr(question, "question_id", ""))
)
category = str(getattr(question, "category", ""))
title = str(
getattr(question, "title", getattr(question, "question_text", ""))
)
return f"{question_id} · {category} · {title}"
@st.cache_resource(show_spinner=False)
def _qa_history_store() -> QaHistoryStore:
return QaHistoryStore()
def _load_qa_questions() -> tuple[list[QaQuestion], QaHistoryStore | None, str]:
try:
store = _qa_history_store()
questions = store.list_questions(limit=200)
if questions:
return questions, store, ""
return [], store, "질답 이력 DB에 아직 적재된 기준 질문이 없습니다."
except QaHistoryStoreError as exc:
try:
fallback = list(load_benchmark_questions(QA_BENCHMARK_FILE))
except Exception:
fallback = []
return fallback, None, str(exc)
def _judgment_label(status: str) -> str:
labels = {
"PASS": "통과",
"WARN": "검토",
"FAIL": "실패",
"REVIEW": "수동 검토",
}
return labels.get(str(status or "").upper(), "미실행")
def _qa_answer_preview(value: object, limit: int = 180) -> str:
text = str(value or "").strip().replace("\n", " ")
return text if len(text) <= limit else text[:limit] + "..."
def _render_qa_benchmark_panel(
questions: list[QaQuestion],
store: QaHistoryStore | None,
store_error: str,
) -> QaQuestion | None:
st.markdown(
'<div class="kb-section-title input" role="heading" aria-level="3">'
"고객 질답 검증 시나리오"
"</div>",
unsafe_allow_html=True,
)
st.caption(
"고객 제공 Excel에서 정리한 47개 후보입니다. 후보를 선택하면 기준 답변과 과거 검증 이력을 확인하고 바로 실행할 수 있습니다."
)
if store_error:
st.warning(
"현재는 파일 기준 후보만 표시합니다. ADB 이력 저장은 연결 복구 후 사용할 수 있습니다. "
+ store_error
)
if not questions:
st.error("표시할 고객 질답 기준이 없습니다.")
return None
category_options = tuple(sorted({question.category for question in questions}))
category = st.selectbox(
"시나리오 구분",
options=("ALL", *category_options),
format_func=lambda item: "전체 47개" if item == "ALL" else item,
key="poc4_qa_category_filter",
)
visible_questions = [
question for question in questions if category == "ALL" or question.category == category
]
st.dataframe(
[
{
"케이스": question.question_code,
"구분": question.category,
"제목": question.title,
"질문": question.question_text,
"기대 기준": _qa_answer_preview(question.expected_focus, 150),
"최근 판정": _judgment_label(question.last_judgment_status),
"최근 실행": question.last_evaluated_at or "과거 기준",
}
for question in visible_questions
],
hide_index=True,
width="stretch",
height=min(520, 110 + 36 * len(visible_questions)),
column_config={
"질문": st.column_config.TextColumn(width="large"),
"기대 기준": st.column_config.TextColumn(width="large"),
"최근 실행": st.column_config.TextColumn(width="medium"),
},
)
by_code = {question.question_code: question for question in visible_questions}
selected_code = st.selectbox(
"실행할 검증 후보 선택",
options=("", *by_code),
format_func=lambda code: "선택 안 함 · 자유 질의" if not code else _question_label(by_code[code]),
key="poc4_qa_question_code",
)
selected = by_code.get(selected_code)
if selected is None:
st.info("자유 텍스트 질문도 실행하고 이력으로 남길 수 있습니다. 자유 질의는 정답 기준이 없어 수동 검토로 표시됩니다.")
return None
with st.expander("선택한 질문의 정답 기준·원본·과거 답변", expanded=True):
left, right = st.columns(2)
with left:
st.markdown("**정답 기준**")
st.write(selected.expected_focus or "정답 기준 설명이 없습니다.")
st.markdown("**기준 답변**")
st.code(selected.baseline_answer or "기준 실행 결과가 없습니다.", language=None)
with right:
st.markdown("**고객 원본 출처**")
st.write(
f"{selected.source_document} / {selected.source_sheet} / 행 {selected.source_row or '-'}"
)
if selected.source_scenario:
st.caption(selected.source_scenario)
st.markdown("**기준 SQL**")
st.code(selected.baseline_sql or selected.sample_sql or "기준 SQL이 없습니다.", language="sql")
if store is not None and selected.question_id is not None:
try:
answers = store.list_answers(selected.question_id, limit=30)
except QaHistoryStoreError as exc:
st.warning(str(exc))
answers = []
if answers:
st.markdown("**과거 답변 이력**")
st.dataframe(
[
{
"순번": item["answer_seq"],
"구분": "과거 기준" if item["answer_kind"] == "HISTORICAL" else "이번 실행",
"판정": _judgment_label(item["judgment_status"]),
"실행 상태": item["execution_status"],
"실행 시각": item["requested_at"],
"답변": _qa_answer_preview(item["answer_text"]),
}
for item in answers
],
hide_index=True,
width="stretch",
)
answer_by_seq = {str(item["answer_seq"]): item for item in answers}
detail_seq = st.selectbox(
"답변 이력 상세",
options=tuple(answer_by_seq),
format_func=lambda value: (
f"#{value} · {_judgment_label(answer_by_seq[value]['judgment_status'])} · "
f"{answer_by_seq[value]['requested_at']}"
),
key=f"poc4_qa_answer_detail_{selected.question_id}",
)
detail = answer_by_seq[detail_seq]
st.markdown("**판정 근거**")
st.write(detail["judgment_reason"] or "판정 근거가 없습니다.")
st.markdown("**생성 SQL**")
st.code(detail["generated_sql"] or "생성 SQL이 없습니다.", language="sql")
st.markdown("**실행 답변**")
st.write(detail["answer_text"] or "답변이 없습니다.")
else:
st.info("아직 저장된 답변 이력이 없습니다.")
return selected
def _apply_console_theme(profile: AppProfile) -> None:
apply_console_theme(st, profile)
@@ -4824,6 +4986,89 @@ def _render_vpd_operations_tab() -> None:
)
def _select_ai_execution_summary(payload: Any) -> dict[str, Any]:
if not isinstance(payload, Mapping):
return {
"generated_sql": "",
"execution_status": "UNKNOWN",
"execution_succeeded": False,
"result": {},
}
generated_sql = str(
payload.get("generatedSql") or payload.get("generated_sql") or ""
).strip()
status = str(payload.get("status") or "").strip().upper()
execution = str(payload.get("execution") or "").strip().upper()
execution_succeeded = (
status == "SHOWSQL_AND_EXECUTED" and execution == "READ_ONLY_EXECUTED"
)
result = {
key: payload.get(key)
for key in (
"status",
"execution",
"rowCount",
"truncated",
"columns",
"items",
"generatedSql",
)
if key in payload
}
return {
"generated_sql": generated_sql,
"execution_status": status or execution or "UNKNOWN",
"execution_succeeded": execution_succeeded,
"result": result,
}
def _record_qa_history(
*,
store: QaHistoryStore | None,
benchmark_question: QaQuestion | None,
question_text: str,
conversation_id: str,
requested_by: str,
model_profile: str,
answer_text: str,
mcp_result: Any,
duration_ms: int,
) -> tuple[QaQuestion | None, QaJudgment | None]:
if store is None:
return benchmark_question, None
try:
history_question = benchmark_question or store.find_or_create_free_text_question(
question_text
)
execution = _select_ai_execution_summary(mcp_result)
judgment = evaluate_sql(
benchmark_question,
execution["generated_sql"],
execution_succeeded=bool(execution["execution_succeeded"]),
error_text=_bounded_json(mcp_result, max_chars=8000),
)
store.record_answer(
question_id=int(history_question.question_id or 0),
answer_kind="LIVE",
conversation_id=conversation_id,
requested_by=requested_by,
model_profile=model_profile,
generated_sql=execution["generated_sql"],
answer_text=answer_text,
result=execution["result"],
execution_output=_bounded_json(mcp_result, max_chars=24_000),
execution_status=str(execution["execution_status"]),
judgment_status=judgment.status,
judgment_reason=judgment.reason,
duration_ms=duration_ms,
)
return history_question, judgment
except QaHistoryStoreError as exc:
st.warning(f"질답 이력 저장에 실패했습니다: {exc}")
return benchmark_question, None
def _process_submitted_question(
*,
question: str,
@@ -4839,6 +5084,8 @@ def _process_submitted_question(
limit: int,
selected_token_preset: VpdTokenPreset | None,
execution_mode_override: str,
benchmark_question: QaQuestion | None,
qa_history_store: QaHistoryStore | None,
) -> None:
normalized_question = question.strip()
if not normalized_question:
@@ -5395,6 +5642,32 @@ def _process_submitted_question(
],
"discovery_failures": discovery_failures,
}
history_question, history_judgment = _record_qa_history(
store=qa_history_store,
benchmark_question=benchmark_question,
question_text=normalized_question,
conversation_id=conversation_id,
requested_by=(
selected_token_preset.user_id
if selected_token_preset is not None
else "portal-user"
),
model_profile=answer_model_profile,
answer_text=str(assistant_message["content"]),
mcp_result=answer_source,
duration_ms=round((perf_counter() - process_started) * 1000),
)
if history_judgment is not None:
assistant_message["details"]["qa_history"] = {
"question_id": history_question.question_id if history_question else None,
"question_code": history_question.question_code if history_question else "",
"judgment_status": history_judgment.status,
"judgment_reason": history_judgment.reason,
}
st.info(
"질답 검증 판정: "
f"{_judgment_label(history_judgment.status)} · {history_judgment.reason}"
)
refresh_progress(98, "질의 결과를 저장하고 있습니다.")
save_chat_turn(
conversation_id=conversation_id,
@@ -5449,27 +5722,13 @@ def main() -> None:
if not st.session_state.get(PORTAL_AUTHENTICATED_KEY, False):
_render_portal_login(profile)
return
try:
questions = load_demo_scenarios(DEMO_SCENARIOS_FILE)
except ScenarioConfigError as exc:
st.error(str(exc))
return
scenario_key = "poc4_mcp_discovery_scenario"
qa_questions, qa_history_store, qa_store_error = _load_qa_questions()
question_text_key = "poc4_mcp_discovery_question_text"
loaded_scenario_key = "poc4_mcp_discovery_loaded_scenario_id"
loaded_qa_question_key = "poc4_mcp_discovery_loaded_qa_question_code"
conversation_id_key = "poc4_mcp_discovery_conversation_id"
chat_page_key = "poc4_mcp_discovery_page"
query_progress_notice_key = "poc4_query_progress_notice"
mcp_cache_generation_key = "poc4_mcp_tools_cache_generation"
selected_scenario_state = st.session_state.get(scenario_key)
scenario_ids = {question.question_id for question in questions}
if str(getattr(selected_scenario_state, "question_id", "")).strip().upper() not in (
"",
*scenario_ids,
):
st.session_state.pop(scenario_key, None)
st.session_state.pop(loaded_scenario_key, None)
st.session_state[question_text_key] = DEFAULT_QUESTION
init_chat_store()
if conversation_id_key not in st.session_state:
st.session_state[conversation_id_key] = new_conversation_id()
@@ -5757,7 +6016,7 @@ def main() -> None:
except (OSError, UnicodeError, ValueError):
st.warning("MCP 설정 JSON을 읽지 못했습니다.")
st.caption(f"config: {MCP_SERVERS_FILE}")
st.caption(f"scenario config: {DEMO_SCENARIOS_FILE}")
st.caption(f"QA benchmark: {QA_BENCHMARK_FILE}")
st.caption(f"token presets: {VPD_TOKEN_PRESETS_FILE}")
st.caption(f"selected LLM model: {selected_query_model_profile}")
st.caption(f"default model: {default_query_model_profile}")
@@ -5775,35 +6034,36 @@ def main() -> None:
_render_architecture_tab()
with scenario_tab:
st.markdown(
'<div class="kb-section-title input" role="heading" aria-level="3">'
"질문 입력"
"</div>",
unsafe_allow_html=True,
)
selected_scenario = st.selectbox(
"업무 데모 질의 샘플",
options=(None, *questions),
format_func=lambda item: (
"선택 안 함 · 직접 질문" if item is None else _question_label(item)
),
key=scenario_key,
)
selected_scenario_id = (
None
if selected_scenario is None
else str(getattr(selected_scenario, "question_id"))
selected_qa_question = _render_qa_benchmark_panel(
qa_questions,
qa_history_store,
qa_store_error,
)
if question_text_key not in st.session_state:
st.session_state[question_text_key] = DEFAULT_QUESTION
if st.session_state.get(loaded_scenario_key, "") != str(
selected_scenario_id or ""
):
if selected_scenario is not None:
st.session_state[question_text_key] = str(
getattr(selected_scenario, "text")
)
st.session_state[loaded_scenario_key] = str(selected_scenario_id or "")
selected_qa_code = (
selected_qa_question.question_code if selected_qa_question is not None else ""
)
if st.session_state.get(loaded_qa_question_key, "") != selected_qa_code:
st.session_state[question_text_key] = (
selected_qa_question.question_text
if selected_qa_question is not None
else DEFAULT_QUESTION
)
st.session_state[loaded_qa_question_key] = selected_qa_code
st.markdown(
'<div class="kb-section-title input" role="heading" aria-level="3">'
"질문 실행"
"</div>",
unsafe_allow_html=True,
)
if selected_qa_question is not None:
st.caption(
"선택한 후보 문장을 수정하지 않고 실행하면 자동 판정합니다. 문장을 수정하면 자유 질의로 저장되어 수동 검토 대상이 됩니다."
)
else:
st.caption("자유 질의는 실행 이력으로 남기며, 고객 기준 정답과 자동 비교하지 않습니다.")
with st.form("poc4_mcp_question_form"):
question = st.text_area(
@@ -5812,7 +6072,7 @@ def main() -> None:
key=question_text_key,
height=120,
max_chars=1_000,
placeholder="업무 질문을 직접 입력하거나 위 질의 샘플을 선택하세요.",
placeholder="고객 검증 후보를 선택하거나 업무 질문을 직접 입력하세요.",
)
submitted = st.form_submit_button(
"질문 전송",
@@ -5889,6 +6149,13 @@ def main() -> None:
limit=int(limit),
selected_token_preset=selected_token_preset,
execution_mode_override=execution_mode_override,
benchmark_question=(
selected_qa_question
if selected_qa_question is not None
and question.strip() == selected_qa_question.question_text.strip()
else None
),
qa_history_store=qa_history_store,
)