joungmin/vpd-permission-poc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
vpd-permission-poc/ai-web-agent-console/app.py

5905 lines
230 KiB
Python
Raw Permalink Blame History

"""HMM AI Web Agent Console Streamlit entrypoint.
Run:
streamlit run app.py --server.port 8622
This screen intentionally does not use the existing LangGraph/actor-context
pipeline. It discovers MCP tools, selects a query-capable tool, and calls it
with the Bearer token entered on the screen.
"""
from __future__ import annotations
import hashlib
import hmac
import html
import json
import logging
import os
import sqlite3
import sys
from datetime import datetime, timezone
from dataclasses import dataclass, field
from pathlib import Path
import re
from time import perf_counter
from typing import Any, Mapping
from uuid import uuid4
from urllib.error import HTTPError, URLError
from urllib.parse import urlsplit
from urllib.request import HTTPRedirectHandler, Request, build_opener
ROOT = Path(__file__).resolve().parent
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
import streamlit as st
import streamlit.components.v1 as components
import oracledb
from ai_web_agent_console.mcp_tool_router import (
McpTool,
McpToolRouterError,
RoutedMcpTool,
build_mcp_tool_arguments,
route_mcp_tool_across_servers_with_llm,
)
from ai_web_agent_console.mcp_result import (
has_actionable_text_result,
status_result_evidence,
status_result_summary,
text_result,
)
from ai_web_agent_console.oci_genai_sdk import (
build_oci_genai_completion_client,
temperature_for_model_profile,
)
from ai_web_agent_console.model_registry import load_model_registry, resolve_model_profile
from ai_web_agent_console.presentation import (
apply_console_theme,
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,
)
LOG = logging.getLogger(__name__)
MCP_PROTOCOL_VERSION = "2025-11-25"
PREFERRED_TOOL = "search_hr_data"
DEFAULT_QUESTION = ""
MAX_RESPONSE_BYTES = 1_000_000
MAX_CONVERSATION_MESSAGES = 8
CHAT_TURNS_PER_PAGE = 3
CHAT_CONTEXT_TURNS = 4
MAX_AGENT_TOOL_STEPS = 4
AGENT_FINAL_ROUTE = "__final__"
COMPLEX_REASONING_MODEL_PROFILE_ENV = (
"AI_WEB_AGENT_CONSOLE_COMPLEX_REASONING_MODEL_PROFILE"
)
LEGACY_COMPLEX_REASONING_MODEL_PROFILE_ENV = "POC4_COMPLEX_REASONING_MODEL_PROFILE"
DEFAULT_COMPLEX_REASONING_MODEL_PROFILE = "grok43"
DEFAULT_SYNTHESIS_FALLBACK_MODEL_PROFILE = "gpt54_mini_oci"
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" / "hmm_demo_scenarios.json"
APP_PROFILE_FILE = ROOT / "config" / "app_profile.json"
CHAT_DB_FILE = ROOT / "data" / "poc4_mcp_chat.sqlite3"
DEFAULT_VPD_USER_ID = "E1001"
VPD_OPERATIONS_URL = "https://hmm-backoffice.cloud-handson.com/"
PORTAL_AUTHENTICATED_KEY = "poc4_portal_authenticated"
PORTAL_AUTH_USER_KEY = "poc4_portal_auth_user"
PORTAL_AUTH_PROXY_USER_HEADER = "X-HMM-Authenticated-User"
PORTAL_AUTH_PROXY_EXPIRY_HEADER = "X-HMM-Auth-Expires"
AUDIT_DB_ENV_FILE = Path(
os.environ.get("AI_WEB_AGENT_CONSOLE_AUDIT_DB_ENV_FILE")
or os.environ.get("POC4_AUDIT_DB_ENV_FILE")
or str(ENV_FILE)
).expanduser()
DEFAULT_AUDIT_DB_DSN = (
"(description=(retry_count=3)(retry_delay=1)"
"(address=(protocol=tcps)(port=1521)(host=adb.ap-seoul-1.oraclecloud.com))"
"(connect_data=(service_name="
"yh0olybn5pqce4n_hmmaipoc_high.adb.oraclecloud.com))"
"(security=(ssl_server_dn_match=yes)))"
)
_OPAQUE_BEARER = re.compile(r"^[\x21-\x7e]{1,4096}$")
@dataclass(frozen=True)
class McpServer:
server_id: str
endpoint_url: str
auth_token_env: str
default_tool: str
tool_allowlist: tuple[str, ...]
router_model_profile: str
description: str
@dataclass(frozen=True)
class VpdTokenPreset:
user_id: str
name: str
role: str
channel: str
scope: str
token: str = field(repr=False, compare=False)
is_default: bool = False
team: str = ""
@property
def display_label(self) -> str:
return " · ".join(
item
for item in (
self.user_id,
self.name,
self.role,
self.team or self.channel,
self.scope,
)
if item
)
@property
def select_label(self) -> str:
return " · ".join(
item
for item in (
self.user_id,
self.name,
self.role,
)
if item
)
class PublicMcpError(RuntimeError):
"""Safe UI error. Never include request headers, tokens, or raw traces."""
class AnswerSynthesisError(RuntimeError):
"""Safe final-answer synthesis error."""
def __init__(
self,
message: str,
diagnostics: Mapping[str, Any] | None = None,
) -> None:
super().__init__(message)
self.diagnostics = dict(diagnostics or {})
@dataclass(frozen=True)
class McpDiscoveryResult:
server: McpServer
tools: tuple[McpTool, ...]
@dataclass(frozen=True)
class _JsonRpcExchange:
result: Mapping[str, Any]
session_id: str = ""
class _McpHttpStatusError(PublicMcpError):
def __init__(self, code: int) -> None:
self.code = code
super().__init__(f"MCP 호출이 실패했습니다. HTTP {code}")
def _utc_now() -> str:
return datetime.now(timezone.utc).isoformat(timespec="seconds")
def _chat_db_path() -> Path:
configured = _runtime_env_value(
"AI_WEB_AGENT_CONSOLE_CHAT_DB_PATH",
"POC4_CHAT_DB_PATH",
)
if not configured:
return CHAT_DB_FILE
path = Path(configured).expanduser()
return path if path.is_absolute() else ROOT / path
def _chat_db_connect() -> sqlite3.Connection:
path = _chat_db_path()
path.parent.mkdir(parents=True, exist_ok=True)
connection = sqlite3.connect(path)
connection.row_factory = sqlite3.Row
connection.execute("PRAGMA journal_mode=WAL")
connection.execute("PRAGMA busy_timeout=5000")
return connection
def init_chat_store() -> None:
with _chat_db_connect() as connection:
connection.execute(
"""
CREATE TABLE IF NOT EXISTS poc4_mcp_chat_conversations (
conversation_id TEXT PRIMARY KEY,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
title TEXT NOT NULL DEFAULT ''
)
"""
)
columns = {
str(row["name"])
for row in connection.execute(
"PRAGMA table_info(poc4_mcp_chat_conversations)"
).fetchall()
}
if "title" not in columns:
connection.execute(
"ALTER TABLE poc4_mcp_chat_conversations "
"ADD COLUMN title TEXT NOT NULL DEFAULT ''"
)
connection.execute(
"""
CREATE TABLE IF NOT EXISTS poc4_mcp_chat_turns (
turn_id INTEGER PRIMARY KEY AUTOINCREMENT,
conversation_id TEXT NOT NULL,
created_at TEXT NOT NULL,
selected_user_id TEXT,
selected_user_label TEXT,
question TEXT NOT NULL,
standalone_question TEXT NOT NULL,
answer TEXT NOT NULL,
basis_json TEXT NOT NULL,
limitations TEXT NOT NULL,
details_json TEXT NOT NULL,
FOREIGN KEY(conversation_id)
REFERENCES poc4_mcp_chat_conversations(conversation_id)
)
"""
)
connection.execute(
"""
CREATE INDEX IF NOT EXISTS idx_poc4_chat_turns_conversation_latest
ON poc4_mcp_chat_turns(conversation_id, turn_id DESC)
"""
)
def ensure_conversation(conversation_id: str) -> None:
now = _utc_now()
with _chat_db_connect() as connection:
connection.execute(
"""
INSERT OR IGNORE INTO poc4_mcp_chat_conversations
(conversation_id, created_at, updated_at, title)
VALUES (?, ?, ?, '')
""",
(conversation_id, now, now),
)
def count_chat_turns(conversation_id: str) -> int:
with _chat_db_connect() as connection:
row = connection.execute(
"""
SELECT COUNT(*) AS count
FROM poc4_mcp_chat_turns
WHERE conversation_id = ?
""",
(conversation_id,),
).fetchone()
return int(row["count"] if row else 0)
def list_conversations(limit: int = 50) -> list[dict[str, Any]]:
with _chat_db_connect() as connection:
rows = connection.execute(
"""
SELECT
c.conversation_id,
c.created_at,
c.updated_at,
c.title,
COUNT(t.turn_id) AS turn_count,
GROUP_CONCAT(
COALESCE(t.question, '') || ' ' ||
COALESCE(t.standalone_question, '') || ' ' ||
COALESCE(t.answer, '') || ' ' ||
COALESCE(t.selected_user_id, '') || ' ' ||
COALESCE(t.selected_user_label, ''),
' '
) AS search_text,
(
SELECT latest.question
FROM poc4_mcp_chat_turns latest
WHERE latest.conversation_id = c.conversation_id
ORDER BY latest.turn_id DESC
LIMIT 1
) AS latest_question
FROM poc4_mcp_chat_conversations c
LEFT JOIN poc4_mcp_chat_turns t
ON t.conversation_id = c.conversation_id
GROUP BY c.conversation_id, c.created_at, c.updated_at, c.title
HAVING COUNT(t.turn_id) > 0
ORDER BY c.updated_at DESC
LIMIT ?
""",
(limit,),
).fetchall()
return [dict(row) for row in rows]
def rename_conversation(conversation_id: str, title: str) -> None:
with _chat_db_connect() as connection:
connection.execute(
"""
UPDATE poc4_mcp_chat_conversations
SET title = ?, updated_at = ?
WHERE conversation_id = ?
""",
(title.strip()[:80], _utc_now(), conversation_id),
)
def delete_conversation(conversation_id: str) -> None:
with _chat_db_connect() as connection:
connection.execute(
"DELETE FROM poc4_mcp_chat_turns WHERE conversation_id = ?",
(conversation_id,),
)
connection.execute(
"DELETE FROM poc4_mcp_chat_conversations WHERE conversation_id = ?",
(conversation_id,),
)
def load_all_chat_turns(conversation_id: str) -> list[dict[str, Any]]:
with _chat_db_connect() as connection:
rows = connection.execute(
"""
SELECT turn_id, created_at, selected_user_id, selected_user_label,
question, standalone_question, answer, basis_json,
limitations, details_json
FROM poc4_mcp_chat_turns
WHERE conversation_id = ?
ORDER BY turn_id ASC
""",
(conversation_id,),
).fetchall()
return [_chat_turn_from_row(row) for row in rows]
def filter_conversations(
rows: list[dict[str, Any]],
search_text: str,
) -> list[dict[str, Any]]:
query = search_text.strip().casefold()
if not query:
return rows
filtered: list[dict[str, Any]] = []
for row in rows:
haystack = " ".join(
str(row.get(key) or "")
for key in (
"conversation_id",
"title",
"latest_question",
"search_text",
"updated_at",
)
).casefold()
if query in haystack:
filtered.append(row)
return filtered
def conversation_label(item: Mapping[str, Any]) -> str:
turn_count = int(item.get("turn_count") or 0)
saved_title = str(item.get("title") or "").strip()
latest_question = str(item.get("latest_question") or "").strip()
if saved_title:
title = saved_title[:34] + ("..." if len(saved_title) > 34 else "")
elif latest_question:
title = latest_question[:34] + ("..." if len(latest_question) > 34 else "")
else:
title = "새 대화"
updated_at = str(item.get("updated_at") or "")
short_id = str(item.get("conversation_id") or "")[-8:]
return f"{title} · {turn_count}건 · {updated_at} · {short_id}"
def load_chat_turns(conversation_id: str, page: int) -> list[dict[str, Any]]:
offset = max(page, 0) * CHAT_TURNS_PER_PAGE
with _chat_db_connect() as connection:
rows = connection.execute(
"""
SELECT turn_id, created_at, selected_user_id, selected_user_label,
question, standalone_question, answer, basis_json,
limitations, details_json
FROM poc4_mcp_chat_turns
WHERE conversation_id = ?
ORDER BY turn_id DESC
LIMIT ? OFFSET ?
""",
(conversation_id, CHAT_TURNS_PER_PAGE, offset),
).fetchall()
return [_chat_turn_from_row(row) for row in rows]
def _is_failed_synthesis_answer(value: object) -> bool:
text = str(value or "").strip()
if not text:
return False
return (
text.startswith("MCP 조회는 완료됐지만 최종 답변 합성에 실패했습니다.")
or "MCP 결과 기반 최종 답변 생성에 실패했습니다" in text
)
def load_chat_context(conversation_id: str) -> list[dict[str, str]]:
with _chat_db_connect() as connection:
rows = connection.execute(
"""
SELECT question, answer
FROM poc4_mcp_chat_turns
WHERE conversation_id = ?
ORDER BY turn_id DESC
LIMIT ?
""",
(conversation_id, CHAT_CONTEXT_TURNS),
).fetchall()
messages: list[dict[str, str]] = []
for row in reversed(rows):
question = str(row["question"] or "").strip()
answer = str(row["answer"] or "").strip()
if question:
messages.append({"role": "user", "content": question[:1200]})
if answer and not _is_failed_synthesis_answer(answer):
messages.append({"role": "assistant", "content": answer[:1200]})
return messages[-MAX_CONVERSATION_MESSAGES:]
def save_chat_turn(
*,
conversation_id: str,
selected_user_id: str,
selected_user_label: str,
question: str,
standalone_question: str,
answer: str,
basis: Any,
limitations: str,
details: Mapping[str, Any],
) -> None:
now = _utc_now()
with _chat_db_connect() as connection:
connection.execute(
"""
INSERT OR IGNORE INTO poc4_mcp_chat_conversations
(conversation_id, created_at, updated_at, title)
VALUES (?, ?, ?, '')
""",
(conversation_id, now, now),
)
connection.execute(
"""
INSERT INTO poc4_mcp_chat_turns (
conversation_id, created_at, selected_user_id,
selected_user_label, question, standalone_question, answer,
basis_json, limitations, details_json
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
conversation_id,
now,
selected_user_id,
selected_user_label,
question,
standalone_question,
answer,
json.dumps(basis if isinstance(basis, list) else [], ensure_ascii=False),
limitations,
json.dumps(dict(details), ensure_ascii=False, default=str),
),
)
connection.execute(
"""
UPDATE poc4_mcp_chat_conversations
SET updated_at = ?,
title = CASE
WHEN title IS NULL OR title = '' THEN ?
ELSE title
END
WHERE conversation_id = ?
""",
(now, question[:60], conversation_id),
)
def _json_loads_or(value: object, fallback: Any) -> Any:
if not isinstance(value, str):
return fallback
try:
return json.loads(value)
except ValueError:
return fallback
def _chat_turn_from_row(row: sqlite3.Row) -> dict[str, Any]:
return {
"turn_id": row["turn_id"],
"created_at": row["created_at"],
"selected_user_id": row["selected_user_id"],
"selected_user_label": row["selected_user_label"],
"question": row["question"],
"standalone_question": row["standalone_question"],
"answer": row["answer"],
"basis": _json_loads_or(row["basis_json"], []),
"limitations": row["limitations"],
"details": _json_loads_or(row["details_json"], {}),
}
def new_conversation_id() -> str:
return "poc4-" + uuid4().hex
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")))
return f"{question_id} · {category} · {title}"
def _apply_console_theme(profile: AppProfile) -> None:
apply_console_theme(st, profile)
def _portal_auth_value(*names: str) -> str:
for name in names:
value = (os.environ.get(name) or _dotenv_value(name)).strip()
if value:
return value
return ""
def _proxy_auth_headers() -> tuple[str, int]:
try:
headers = st.context.headers
username = str(headers.get(PORTAL_AUTH_PROXY_USER_HEADER) or "").strip()
expires_at = int(
str(headers.get(PORTAL_AUTH_PROXY_EXPIRY_HEADER) or "0").strip()
)
except (AttributeError, TypeError, ValueError):
return "", 0
return username, expires_at
def _restore_portal_proxy_session() -> None:
username, expires_at = _proxy_auth_headers()
expected_username = _portal_auth_value(
"AI_WEB_AGENT_CONSOLE_LOGIN_USER",
"POC4_LOGIN_USER",
)
authenticated = bool(
username
and expected_username
and expires_at > int(datetime.now(timezone.utc).timestamp())
and hmac.compare_digest(username, expected_username)
)
if not authenticated:
st.session_state.pop(PORTAL_AUTHENTICATED_KEY, None)
st.session_state.pop(PORTAL_AUTH_USER_KEY, None)
return
st.session_state[PORTAL_AUTHENTICATED_KEY] = True
st.session_state[PORTAL_AUTH_USER_KEY] = username
def _render_portal_login(profile: AppProfile) -> None:
with st.container(key="console_login_container"):
render_login_brand(st, profile)
st.error(
"인증 게이트웨이의 사용자 확인 정보가 없습니다. "
"공식 포털 주소로 다시 접속해 주세요."
)
st.markdown(
f'<p class="console-muted">{html.escape(profile.login_footer)}</p>',
unsafe_allow_html=True,
)
def _render_app_header(profile: AppProfile) -> None:
render_console_header(st, profile)
def _render_demo_user_card(preset: VpdTokenPreset) -> None:
st.markdown(
f"""
<div class="kb-vpd-card">
<div class="kb-vpd-user">
{html.escape(preset.user_id)} · {html.escape(preset.name)}
</div>
<div class="kb-vpd-meta">
{html.escape(preset.role)} · {html.escape(preset.team or preset.channel)}
</div>
<div class="kb-vpd-scope">
테스트 문맥: {html.escape(preset.scope)}
</div>
</div>
""",
unsafe_allow_html=True,
)
def _normalized_bearer(value: object) -> str:
if not isinstance(value, str):
return ""
token = value.strip()
if token.lower().startswith("bearer "):
token = token[7:].strip()
return token if _OPAQUE_BEARER.fullmatch(token) else ""
def _token_fingerprint(token: str) -> str:
value = str(token or "")
if not value:
return "empty"
digest = hashlib.sha256(value.encode("utf-8")).hexdigest()[:12]
return f"len={len(value)}, sha256={digest}"
def _bounded_json(value: Any, max_chars: int = 24000) -> str:
text = json.dumps(value, ensure_ascii=False, default=str)
if len(text) <= max_chars:
return text
return text[:max_chars] + "\n...<truncated>"
def _default_vpd_user_id(
presets: tuple[VpdTokenPreset, ...],
) -> str | None:
for preset in presets:
if preset.user_id == DEFAULT_VPD_USER_ID:
return preset.user_id
for preset in presets:
if preset.is_default:
return preset.user_id
return presets[0].user_id if presets else None
def _dotenv_value(name: str, path: Path = ENV_FILE) -> str:
if not name or not path.exists():
return ""
try:
lines = path.read_text(encoding="utf-8").splitlines()
except (OSError, UnicodeError):
return ""
for line in lines:
stripped = line.strip()
if not stripped or stripped.startswith("#") or "=" not in stripped:
continue
if stripped.startswith("export "):
stripped = stripped[7:].lstrip()
key, raw_value = stripped.split("=", 1)
if key.strip() != name:
continue
value = raw_value.strip()
if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}:
value = value[1:-1]
return value.strip()
return ""
class AuditLogError(RuntimeError):
"""Safe audit-log error that never includes DB credentials or provider details."""
def _audit_db_env_value(name: str, default: str = "") -> str:
names = [name]
if name.startswith("AI_WEB_AGENT_CONSOLE_"):
names.append(name.replace("AI_WEB_AGENT_CONSOLE_", "POC4_", 1))
for candidate in names:
value = (
os.environ.get(candidate)
or _dotenv_value(candidate, AUDIT_DB_ENV_FILE)
).strip()
if value:
return value
return default.strip()
@st.cache_resource(show_spinner=False)
def _audit_db_pool() -> Any:
password = _audit_db_env_value("AI_WEB_AGENT_CONSOLE_AUDIT_DB_PASSWORD")
if not password:
raise AuditLogError("감사로그 DB 접속 설정을 확인해 주세요.")
dsn = _audit_db_env_value(
"AI_WEB_AGENT_CONSOLE_AUDIT_DSN", DEFAULT_AUDIT_DB_DSN
)
wallet_password = _audit_db_env_value(
"AI_WEB_AGENT_CONSOLE_AUDIT_WALLET_PASSWORD"
)
pool_options: dict[str, Any] = {}
if wallet_password:
wallet_dir = Path(
_audit_db_env_value(
"AI_WEB_AGENT_CONSOLE_AUDIT_WALLET_DIR",
"/home/opc/apps/vpd-backoffice/wallet",
)
).expanduser().resolve()
if not wallet_dir.is_dir():
raise AuditLogError("감사로그 DB Wallet 경로를 확인해 주세요.")
pool_options.update(
config_dir=str(wallet_dir),
wallet_location=str(wallet_dir),
wallet_password=wallet_password,
)
elif not (dsn.lstrip().startswith("(") or dsn.lower().startswith("tcps://")):
raise AuditLogError(
"감사로그 DB DSN은 TLS 접속 기술자이거나 Wallet 암호와 함께 제공되어야 합니다."
)
try:
return oracledb.create_pool(
user=_audit_db_env_value(
"AI_WEB_AGENT_CONSOLE_AUDIT_DB_USER", "ADMIN"
),
password=password,
dsn=dsn,
min=1,
max=2,
increment=1,
getmode=oracledb.POOL_GETMODE_WAIT,
**pool_options,
)
except (oracledb.Error, OSError, ValueError):
raise AuditLogError("감사로그 DB에 연결하지 못했습니다.") from None
def _audit_rows(
sql: str,
binds: Mapping[str, Any] | None = None,
) -> list[dict[str, Any]]:
try:
with _audit_db_pool().acquire() as connection:
with connection.cursor() as cursor:
cursor.execute(sql, dict(binds or {}))
columns = [item[0].lower() for item in cursor.description]
records: list[dict[str, Any]] = []
for row in cursor:
record: dict[str, Any] = {}
for column, value in zip(columns, row):
if hasattr(value, "read"):
value = value.read()
record[column] = value
records.append(record)
return records
except AuditLogError:
raise
except (oracledb.Error, OSError, ValueError):
raise AuditLogError("감사로그를 조회하지 못했습니다.") from None
@st.cache_data(ttl=60, show_spinner=False)
def _load_hmm_audit_inventory() -> list[dict[str, Any]]:
return _audit_rows(
"""
SELECT event_type,
COUNT(*) AS event_count,
TO_CHAR(
MAX(created_at) AT TIME ZONE 'Asia/Seoul',
'YYYY-MM-DD HH24:MI:SS'
) AS latest_event_time
FROM ADMIN.HMM_ACCESS_AUDIT
GROUP BY event_type
ORDER BY event_type
"""
)
@st.cache_data(ttl=30, show_spinner=False)
def _load_hmm_audit_events(
days: int,
row_limit: int,
event_type: str,
status: str,
) -> list[dict[str, Any]]:
return _audit_rows(
"""
SELECT *
FROM (
SELECT audit_id,
TO_CHAR(
created_at AT TIME ZONE 'Asia/Seoul',
'YYYY-MM-DD HH24:MI:SS'
) AS event_time,
event_type,
key_id,
object_id,
status,
row_count,
error_code,
message
FROM ADMIN.HMM_ACCESS_AUDIT
WHERE created_at >= (
SYSTIMESTAMP - NUMTODSINTERVAL(:days, 'DAY')
)
AND (:event_type IS NULL OR event_type = :event_type)
AND (:status IS NULL OR status = :status)
ORDER BY created_at DESC, audit_id DESC
)
WHERE ROWNUM <= :row_limit
""",
{
"days": int(days),
"event_type": event_type or None,
"status": status or None,
"row_limit": int(row_limit),
},
)
def _security_evidence_kind(question: str) -> str:
text = " ".join(str(question or "").casefold().split())
if "보험료" in text and any(
term in text for term in ("합계", "개별", "상세", "마스킹")
):
return "PREMIUM_MASK_AGGREGATE"
if any(term in text for term in ("주민번호", "rrn_masked")):
return "RRN_DISPLAY_MASK"
if (
any(term in text for term in ("공통계정", "공통 계정"))
and "채널" in text
):
return "CHANNEL_SCOPE"
if re.search(r"\bCT[0-9]+\b", str(question or ""), re.IGNORECASE) and any(
term in text for term in ("조회되는지", "담당 고객이 아님", "권한")
):
return "CONTRACT_SCOPE"
if any(term in text for term in ("로그인하지 않은", "미인증", "인증 실패")):
return "AUTH_DENIAL"
return ""
def _safe_audit_log(
cursor: Any,
*,
user_id: str,
action_name: str,
policy_result: str,
request_text: str,
response_summary: str,
) -> Mapping[str, Any]:
cursor.callproc(
"POC_2.LOG_TOOL_CALL",
[
user_id,
"KB_AI_PORTAL",
"KB_SECURITY_EVIDENCE",
action_name,
policy_result,
request_text[:1000],
response_summary[:2000],
],
)
cursor.execute(
"""
SELECT log_id, TO_CHAR(event_ts, 'YYYY-MM-DD HH24:MI:SS')
FROM POC_2.KB_SECURITY_AUDIT_LOG
WHERE log_id = (
SELECT MAX(log_id)
FROM POC_2.KB_SECURITY_AUDIT_LOG
WHERE end_user_id = :user_id
AND tool_name = 'KB_SECURITY_EVIDENCE'
AND action_name = :action_name
)
""",
{"user_id": user_id, "action_name": action_name},
)
row = cursor.fetchone()
return {
"audit_log_id": int(row[0]) if row and row[0] is not None else None,
"audit_event_time": str(row[1]) if row and row[1] is not None else "",
}
def collect_security_evidence(
question: str,
token_preset: VpdTokenPreset | None,
*,
failure_message: str = "",
) -> Mapping[str, Any]:
"""Collect predefined, token-scoped security evidence without exposing secrets."""
kind = _security_evidence_kind(question)
if not kind or token_preset is None:
return {}
user_id = token_preset.user_id.strip()
evidence: dict[str, Any] = {
"evidence_type": kind,
"user_id": user_id,
"role": token_preset.role,
"channel": token_preset.channel,
}
try:
with _audit_db_pool().acquire() as connection:
with connection.cursor() as cursor:
context_ready = False
try:
cursor.callproc(
"ADMIN.CB_AGENT_CTX_PKG.SET_USER_BY_BEARER",
[token_preset.token],
)
context_ready = True
cursor.execute(
"""
SELECT SYS_CONTEXT('CB_AGENT_CTX', 'STAKEHOLDER_ROLE'),
SYS_CONTEXT('CB_AGENT_CTX', 'STAKEHOLDER_CHANNEL')
FROM dual
"""
)
context_row = cursor.fetchone() or ("", "")
evidence["verified_role"] = str(context_row[0] or "")
evidence["verified_channel"] = str(context_row[1] or "")
except oracledb.Error:
if kind != "AUTH_DENIAL":
raise
if kind == "PREMIUM_MASK_AGGREGATE" and context_ready:
cursor.execute(
"""
SELECT (SELECT COUNT(*)
FROM POC_2.KB_CONTRACTS
WHERE fc_channel = :channel),
ADMIN.CB_KB_PREMIUM_SUM()
FROM dual
""",
{"channel": evidence.get("verified_channel")},
)
row = cursor.fetchone() or (0, None)
cursor.execute(
"""
SELECT COUNT(*)
FROM redaction_policies
WHERE object_owner = 'POC_2'
AND object_name = 'KB_CONTRACTS'
AND policy_name = 'KB_CONTRACT_PREMIUM_REDACT'
AND enable = 'YES'
"""
)
policy_count = int((cursor.fetchone() or (0,))[0] or 0)
evidence.update(
{
"visible_contract_count": int(row[0] or 0),
"premium_sum": None if row[1] is None else int(row[1]),
"individual_premium_state": (
"MASKED" if policy_count else "POLICY_NOT_FOUND"
),
"redaction_policy": "KB_CONTRACT_PREMIUM_REDACT",
}
)
elif kind == "RRN_DISPLAY_MASK" and context_ready:
cursor.execute(
"""
SELECT COUNT(*) AS visible_rows,
SUM(CASE WHEN REGEXP_LIKE(
rrn_masked,
'^[0-9]{6}-[0-9][*]{6}$',
'c'
) THEN 1 ELSE 0 END) AS masked_rows,
SUM(CASE WHEN REGEXP_LIKE(
rrn_masked,
'^[0-9]{6}-[0-9]{7}$',
'c'
) THEN 1 ELSE 0 END) AS plaintext_rows
FROM POC_2.KB_CUSTOMERS customer
WHERE EXISTS (
SELECT 1
FROM POC_2.KB_CONTRACTS contract
WHERE contract.cust_id = customer.cust_id
AND contract.fc_id = :user_id
)
""",
{"user_id": user_id},
)
row = cursor.fetchone() or (0, 0, 0)
evidence.update(
{
"visible_rows": int(row[0] or 0),
"masked_format_rows": int(row[1] or 0),
"plaintext_rows": int(row[2] or 0),
"display_format": "YYMMDD-N******",
}
)
elif kind == "CHANNEL_SCOPE" and context_ready:
cursor.execute(
"""
SELECT SUM(CASE WHEN fc_channel = '다이렉트' THEN 1 ELSE 0 END),
SUM(CASE WHEN fc_channel = '설계사' THEN 1 ELSE 0 END),
SUM(CASE WHEN fc_channel = 'GA' THEN 1 ELSE 0 END),
SUM(CASE WHEN fc_channel = '제휴' THEN 1 ELSE 0 END),
COUNT(*)
FROM POC_2.KB_CONTRACTS
WHERE fc_channel = :channel
""",
{"channel": evidence.get("verified_channel")},
)
row = cursor.fetchone() or (0, 0, 0, 0, 0)
evidence["channel_contract_counts"] = {
"다이렉트": int(row[0] or 0),
"설계사": int(row[1] or 0),
"GA": int(row[2] or 0),
"제휴": int(row[3] or 0),
"전체": int(row[4] or 0),
}
elif kind == "CONTRACT_SCOPE" and context_ready:
match = re.search(r"\bCT[0-9]+\b", question, re.IGNORECASE)
contract_no = match.group(0).upper() if match else ""
cursor.execute(
"""
SELECT COUNT(*)
FROM POC_2.KB_CONTRACTS
WHERE contract_no = :contract_no
AND fc_id = :user_id
""",
{"contract_no": contract_no, "user_id": user_id},
)
row = cursor.fetchone()
evidence.update(
{
"contract_no": contract_no,
"visible_rows": int(row[0] or 0) if row else 0,
"policy_result": "FILTERED"
if not row or int(row[0] or 0) == 0
else "ALLOWED",
}
)
elif kind == "AUTH_DENIAL":
evidence.update(
{
"authentication_status": (
"DENIED"
if user_id == "GUEST_000"
or not context_ready
or failure_message
else "VALID"
),
"returned_rows": 0,
"failure_message": str(failure_message or "")[:500],
}
)
summary = json.dumps(evidence, ensure_ascii=False, default=str)
audit_record = _safe_audit_log(
cursor,
user_id=user_id,
action_name=kind,
policy_result=str(
evidence.get("policy_result")
or evidence.get("authentication_status")
or "EVIDENCE_CAPTURED"
),
request_text=str(question or ""),
response_summary=summary,
)
connection.commit()
evidence.update(audit_record)
evidence["audit_source"] = "POC_2.KB_SECURITY_AUDIT_LOG"
except (AuditLogError, oracledb.Error, OSError, ValueError) as exc:
evidence["evidence_error"] = type(exc).__name__
return evidence
def _enrich_answer_with_security_evidence(
answer: str,
evidence: Mapping[str, Any] | None,
) -> str:
if not evidence:
return str(answer or "")
kind = str(evidence.get("evidence_type") or "")
prefix = ""
if kind == "PREMIUM_MASK_AGGREGATE":
premium_sum = evidence.get("premium_sum")
formatted_sum = (
f"{int(premium_sum):,}원" if premium_sum is not None else "조회 불가"
)
prefix = (
f"{evidence.get('user_id')} 권한에서는 개별 계약 보험료가 마스킹됩니다. "
f"{evidence.get('channel')} 전체 보험료 합계는 {formatted_sum}입니다."
)
elif kind == "RRN_DISPLAY_MASK":
prefix = (
f"평문 주민번호 반환 건수는 {int(evidence.get('plaintext_rows') or 0):,}건입니다. "
"주민번호는 모든 역할에서 상시 마스킹되며 YYMMDD-N****** 형태로만 표시됩니다."
)
elif kind == "CHANNEL_SCOPE":
counts = evidence.get("channel_contract_counts")
if isinstance(counts, Mapping):
prefix = (
f"{evidence.get('user_id')} 권한 조회 결과: 다이렉트 "
f"{int(counts.get('다이렉트') or 0):,}건, 설계사 "
f"{int(counts.get('설계사') or 0):,}건, GA "
f"{int(counts.get('GA') or 0):,}건, 제휴 "
f"{int(counts.get('제휴') or 0):,}건입니다."
)
elif kind == "CONTRACT_SCOPE":
prefix = (
f"사용자 {evidence.get('user_id')}의 {evidence.get('contract_no')} 조회 결과는 "
f"{int(evidence.get('visible_rows') or 0):,}건이며 권한 필터 결과는 "
f"{evidence.get('policy_result')}입니다. 계약·고객·보험료 상세는 노출되지 않았습니다."
)
elif kind == "AUTH_DENIAL":
prefix = (
f"{evidence.get('user_id')} 인증 결과는 "
f"{evidence.get('authentication_status')}이며 반환 데이터는 "
f"{int(evidence.get('returned_rows') or 0):,}건입니다."
)
if evidence.get("audit_log_id"):
prefix += (
f" 감사로그 ID는 {evidence.get('audit_log_id')}, 실행시각은 "
f"{evidence.get('audit_event_time') or '확인 필요'}입니다."
)
normalized = str(answer or "").strip()
return f"{prefix}\n\n{normalized}".strip() if prefix else normalized
def _business_evidence_kind(question: str) -> str:
text = " ".join(str(question or "").casefold().split())
if "41048" in text and "삼성화재" in text and "대물배상" in text:
return "CLAUSE_COMPARISON"
if "41047" in text and any(term in text for term in ("구버전", "섞이지")):
return "VERSION_GOVERNANCE"
if all(term in text for term in ("원본 pdf", "청크", "추적")):
return "LINEAGE_TRACE"
if "신규 약관 pdf" in text and any(
term in text for term in ("메타", "카탈로그")
):
return "METADATA_CATALOG"
if "c1001025" in text and "당사" in text and "타사" in text and "건강" in text:
return "CROSS_HOLDING"
if "c1001006" not in text:
return ""
if "계약별 담당 채널" in text:
return "VISIBLE_CONTRACT_SCOPE"
if "자차담보" in text:
return "CUSTOMER_COVERAGE"
if "내 담당 보험" in text:
return "CUSTOMER_CONTRACTS"
if "갱신 상담" in text and "삼성화재" in text:
return "AUTO_RENEWAL_COMPARISON"
if "갱신월" in text and any(term in text for term in ("전환", "유지")):
return "RENEWAL_CONSULTING"
return ""
def _cursor_rows(cursor: Any) -> list[dict[str, Any]]:
columns = [item[0].lower() for item in cursor.description]
records: list[dict[str, Any]] = []
for row in cursor:
record: dict[str, Any] = {}
for column, value in zip(columns, row):
if hasattr(value, "read"):
value = value.read()
record[column] = value
records.append(record)
return records
def _business_rows(
cursor: Any,
sql: str,
binds: Mapping[str, Any] | None = None,
) -> list[dict[str, Any]]:
cursor.execute(sql, dict(binds or {}))
return _cursor_rows(cursor)
def _catalog_documents(
cursor: Any,
product_codes: tuple[str, ...],
) -> list[dict[str, Any]]:
if not product_codes:
return []
placeholders = ", ".join(f":product_code_{index}" for index in range(len(product_codes)))
binds = {
f"product_code_{index}": product_code
for index, product_code in enumerate(product_codes)
}
return _business_rows(
cursor,
f"""
SELECT document_id, source_file, file_name, company, product_code,
product_name, insurance_type, product_type, document_kind,
version_label, sales_status, ingestion_status, metadata_status,
activation_eligible
FROM ADMIN.KB_DOCUMENT_V2
WHERE product_code IN ({placeholders})
ORDER BY product_code, activation_eligible DESC, version_label, file_name
""",
binds,
)
def _derived_article_metadata(text: object) -> tuple[str, str]:
normalized = " ".join(str(text or "").split())
doubled_matches = re.findall(
r"(?<![0-9])([0-9]{1,3})조\1조\s*([^0-9①-⑳]{0,30})",
normalized,
)
if doubled_matches:
number, title = doubled_matches[-1]
return f"제{number}조", title.strip(" ()[]·:")
explicit = re.search(r"제\s*([0-9]{1,3})\s*조(?:\s*\(([^)]{1,40})\))?", normalized)
if explicit:
return f"제{explicit.group(1)}조", str(explicit.group(2) or "").strip()
return "", ""
def _decorate_clause_chunks(
chunks: list[dict[str, Any]],
*,
default_title: str = "약관 본문",
) -> list[dict[str, Any]]:
current_number = ""
current_title = ""
for chunk in chunks:
derived_number, derived_title = _derived_article_metadata(chunk.get("excerpt"))
if derived_number:
current_number = derived_number
current_title = derived_title or current_title
article_number = str(chunk.get("article_number") or current_number or "")
article_title = str(
chunk.get("article_title") or current_title or default_title
)
page_start = chunk.get("page_start")
page_end = chunk.get("page_end")
chunk.update(
{
"derived_article_number": article_number,
"derived_article_title": article_title,
"logical_locator": " ".join(
part for part in (article_number, article_title) if part
),
"physical_page_range": f"{page_start}-{page_end}",
"chunk_type": "body",
}
)
return chunks
def _catalog_chunks(
cursor: Any,
*,
product_code: str,
first_term: str,
second_term: str = "",
row_limit: int = 3,
) -> list[dict[str, Any]]:
first_pattern = f"%{first_term}%"
second_pattern = f"%{second_term or first_term}%"
chunks = _business_rows(
cursor,
"""
SELECT chunk_id, document_id, source_file, company, product_code,
product_name, insurance_type, product_type, version_label,
sales_status, page_start, page_end, article_number,
article_title, heading_path, sequence_in_document,
SUBSTR(display_markdown, 1, 900) excerpt
FROM (
SELECT chunk_id, document_id, source_file, company, product_code,
product_name, insurance_type, product_type, version_label,
sales_status, page_start, page_end, article_number,
article_title, heading_path, sequence_in_document,
display_markdown,
ROW_NUMBER() OVER (
ORDER BY CASE
WHEN display_markdown LIKE :first_pattern THEN 1
ELSE 2
END,
LENGTH(display_markdown) DESC,
page_start,
sequence_in_document
) AS rn
FROM ADMIN.KB_CHUNK_ACTIVE_V
WHERE product_code = :product_code
AND version_label = 'current'
AND chunk_id LIKE 'chk_%'
AND (
display_markdown LIKE :first_pattern
OR display_markdown LIKE :second_pattern
)
)
WHERE rn <= :row_limit
ORDER BY rn
""",
{
"product_code": product_code,
"first_pattern": first_pattern,
"second_pattern": second_pattern,
"row_limit": int(row_limit),
},
)
return _decorate_clause_chunks(chunks)
def _catalog_section_chunks(
cursor: Any,
*,
product_code: str,
anchor_term: str,
row_limit: int = 6,
) -> list[dict[str, Any]]:
chunks = _business_rows(
cursor,
"""
WITH anchor AS (
SELECT MIN(sequence_in_document) anchor_sequence,
MIN(page_start) KEEP (
DENSE_RANK FIRST ORDER BY sequence_in_document
) anchor_page
FROM ADMIN.KB_CHUNK_ACTIVE_V
WHERE product_code = :product_code
AND version_label = 'current'
AND chunk_id LIKE 'chk_%'
AND display_markdown LIKE :anchor_pattern
)
SELECT *
FROM (
SELECT chunk.chunk_id, chunk.document_id, chunk.source_file,
chunk.company, chunk.product_code, chunk.product_name,
chunk.insurance_type, chunk.product_type,
chunk.version_label, chunk.sales_status,
chunk.page_start, chunk.page_end, chunk.article_number,
chunk.article_title, chunk.heading_path,
chunk.sequence_in_document,
SUBSTR(chunk.display_markdown, 1, 1200) excerpt
FROM ADMIN.KB_CHUNK_ACTIVE_V chunk
CROSS JOIN anchor
WHERE chunk.product_code = :product_code
AND chunk.version_label = 'current'
AND chunk.chunk_id LIKE 'chk_%'
AND chunk.sequence_in_document BETWEEN
anchor.anchor_sequence AND anchor.anchor_sequence + 7
AND chunk.page_start BETWEEN anchor.anchor_page AND anchor.anchor_page + 3
ORDER BY chunk.sequence_in_document, chunk.page_start
)
WHERE ROWNUM <= :row_limit
""",
{
"product_code": product_code,
"anchor_pattern": f"%{anchor_term}%",
"row_limit": int(row_limit),
},
)
return _decorate_clause_chunks(chunks)
def _health_catalog_evidence(cursor: Any) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
documents = _business_rows(
cursor,
"""
SELECT document_id, source_file, file_name, company, product_code,
product_name, insurance_type, product_type, document_kind,
version_label, ingestion_status, metadata_status,
activation_eligible
FROM ADMIN.KB_DOCUMENT_V2
WHERE product_code = '25213'
AND file_name LIKE '%일반심사형%'
AND version_label = 'current'
AND activation_eligible = 1
ORDER BY file_name
""",
)
chunks = _business_rows(
cursor,
"""
SELECT chunk_id, document_id, source_file, company, product_code,
product_name, version_label, page_start, page_end,
article_number, article_title, heading_path,
sequence_in_document, SUBSTR(display_markdown, 1, 900) excerpt
FROM (
SELECT chunk.*, ROW_NUMBER() OVER (
ORDER BY chunk.page_start, chunk.sequence_in_document
) rn
FROM ADMIN.KB_CHUNK_ACTIVE_V chunk
WHERE chunk.product_code = '25213'
AND chunk.source_file LIKE '%일반심사형%'
AND chunk.version_label = 'current'
AND chunk.chunk_id LIKE 'chk_%'
AND (
chunk.search_text LIKE '%암진단%'
OR chunk.search_text LIKE '%암 진단%'
OR chunk.search_text LIKE '%간편심사%'
)
)
WHERE rn <= 3
ORDER BY rn
""",
)
return documents, _decorate_clause_chunks(
chunks,
default_title="상품요약 및 보험금 지급제한",
)
def _external_health_catalog_evidence(
cursor: Any,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
documents = _business_rows(
cursor,
"""
SELECT document_id, source_file, file_name, company, product_code,
product_name, insurance_type, product_type, document_kind,
version_label, ingestion_status, metadata_status,
activation_eligible
FROM ADMIN.KB_DOCUMENT_V2
WHERE file_name = '약관_31084(03)_20260101.pdf'
AND version_label = 'current'
AND activation_eligible = 1
""",
)
chunks = _business_rows(
cursor,
"""
SELECT chunk_id, document_id, source_file, company, product_code,
product_name, version_label, page_start, page_end,
article_number, article_title, heading_path,
sequence_in_document, SUBSTR(display_markdown, 1, 900) excerpt
FROM (
SELECT chunk.*, ROW_NUMBER() OVER (
ORDER BY chunk.page_start, chunk.sequence_in_document
) rn
FROM ADMIN.KB_CHUNK_ACTIVE_V chunk
WHERE chunk.document_id =
'doc_16b68344b8_16b68344_약관_31084_03_20260101'
AND chunk.chunk_id LIKE 'chk_%'
AND (
chunk.search_text LIKE '%암%'
OR chunk.search_text LIKE '%보장%'
OR chunk.search_text LIKE '%갱신%'
)
)
WHERE rn <= 3
ORDER BY rn
""",
)
return documents, _decorate_clause_chunks(
chunks,
default_title="상품요약 및 보험금 지급제한",
)
def collect_business_evidence(
question: str,
token_preset: VpdTokenPreset | None,
) -> Mapping[str, Any]:
"""Collect token-validated, explicitly scoped business and catalog evidence."""
kind = _business_evidence_kind(question)
if not kind or token_preset is None:
return {}
customer_match = re.search(r"\bC[0-9]+\b", question, re.IGNORECASE)
customer_id = customer_match.group(0).upper() if customer_match else ""
evidence: dict[str, Any] = {
"evidence_type": kind,
"user_id": token_preset.user_id.strip(),
"role": token_preset.role,
"channel": token_preset.channel,
"customer_id": customer_id,
"scope_enforcement": "Bearer 토큰 검증 + CUST_ID/FC_ID 명시 필터",
}
try:
with _audit_db_pool().acquire() as connection:
with connection.cursor() as cursor:
cursor.callproc(
"ADMIN.CB_AGENT_CTX_PKG.SET_USER_BY_BEARER",
[token_preset.token],
)
cursor.execute(
"""
SELECT SYS_CONTEXT('CB_AGENT_CTX', 'STAKEHOLDER_USER_ID'),
SYS_CONTEXT('CB_AGENT_CTX', 'STAKEHOLDER_ROLE'),
SYS_CONTEXT('CB_AGENT_CTX', 'STAKEHOLDER_CHANNEL')
FROM dual
"""
)
context_row = cursor.fetchone() or ("", "", "")
verified_user_id = str(context_row[0] or "")
evidence.update(
{
"verified_user_id": verified_user_id,
"verified_role": str(context_row[1] or ""),
"verified_channel": str(context_row[2] or ""),
}
)
if verified_user_id != token_preset.user_id.strip():
raise AuditLogError("사용자 토큰 검증 결과가 일치하지 않습니다.")
customer_kinds = {
"AUTO_RENEWAL_COMPARISON",
"VISIBLE_CONTRACT_SCOPE",
"CROSS_HOLDING",
"CUSTOMER_COVERAGE",
"CUSTOMER_CONTRACTS",
"RENEWAL_CONSULTING",
}
if kind in customer_kinds:
if not customer_id:
return {}
binds = {
"customer_id": customer_id,
"user_id": verified_user_id,
}
evidence["contracts"] = _business_rows(
cursor,
"""
SELECT contract_no, cust_id, product_cd, contract_status,
pay_cycle, fc_channel, fc_id
FROM POC_2.KB_CONTRACTS
WHERE cust_id = :customer_id
AND fc_id = :user_id
ORDER BY contract_no
""",
binds,
)
evidence["products"] = _business_rows(
cursor,
"""
SELECT product_cd, clause_product_nm, insurance_type,
product_type, clause_version, sale_status,
active_yn, version_div
FROM POC_2.KB_PRODUCTS product
WHERE EXISTS (
SELECT 1
FROM POC_2.KB_CONTRACTS contract
WHERE contract.cust_id = :customer_id
AND contract.fc_id = :user_id
AND contract.product_cd = product.product_cd
)
ORDER BY product_cd, clause_version
""",
binds,
)
evidence["coverages"] = _business_rows(
cursor,
"""
SELECT coverage.contract_no, coverage.coverage_nm,
coverage.coverage_type, coverage.coverage_div,
coverage.insured_amt,
TO_CHAR(coverage.renew_due_dt, 'YYYY-MM-DD') renew_due_dt
FROM POC_2.KB_COVERAGES coverage
JOIN POC_2.KB_CONTRACTS contract
ON contract.contract_no = coverage.contract_no
WHERE contract.cust_id = :customer_id
AND contract.fc_id = :user_id
ORDER BY coverage.contract_no, coverage.coverage_nm
""",
binds,
)
if kind in {
"AUTO_RENEWAL_COMPARISON",
"CROSS_HOLDING",
"RENEWAL_CONSULTING",
}:
evidence["external_holdings"] = _business_rows(
cursor,
"""
SELECT holding.cust_id, holding.ext_insurer,
holding.ext_product_grp,
holding.ext_product_type,
holding.ext_renew_month,
holding.ext_clause_nm, holding.ext_file_nm,
holding.ext_sale_status
FROM POC_2.KB_EXTERNAL_HOLDINGS holding
WHERE holding.cust_id = :customer_id
AND EXISTS (
SELECT 1
FROM POC_2.KB_CONTRACTS contract
WHERE contract.cust_id = holding.cust_id
AND contract.fc_id = :user_id
)
ORDER BY holding.ext_insurer, holding.ext_file_nm
""",
binds,
)
if kind in {
"AUTO_RENEWAL_COMPARISON",
"CUSTOMER_COVERAGE",
"CUSTOMER_CONTRACTS",
"RENEWAL_CONSULTING",
"METADATA_CATALOG",
"LINEAGE_TRACE",
"CLAUSE_COMPARISON",
}:
evidence["catalog_documents"] = _catalog_documents(
cursor,
("41048", "20071")
if kind in {
"AUTO_RENEWAL_COMPARISON",
"RENEWAL_CONSULTING",
"CLAUSE_COMPARISON",
}
else ("41048",),
)
evidence["kb_clause_chunks"] = _catalog_section_chunks(
cursor,
product_code="41048",
anchor_term=(
"21조21조보상하는 손해"
if kind == "CUSTOMER_COVERAGE"
else "제2절 대인배상Ⅱ와 대물배상"
),
)
if kind in {
"AUTO_RENEWAL_COMPARISON",
"RENEWAL_CONSULTING",
"CLAUSE_COMPARISON",
}:
evidence["external_clause_chunks"] = _catalog_chunks(
cursor,
product_code="20071",
first_term="타인의 차량 및 재물",
second_term="보상하지 않는 손해",
row_limit=5,
)
if kind == "CROSS_HOLDING":
own_docs, own_chunks = _health_catalog_evidence(cursor)
ext_docs, ext_chunks = _external_health_catalog_evidence(cursor)
evidence.update(
{
"own_clause_documents": own_docs,
"own_clause_chunks": own_chunks,
"external_clause_documents": ext_docs,
"external_clause_chunks": ext_chunks,
}
)
elif kind == "VERSION_GOVERNANCE":
evidence["products"] = _business_rows(
cursor,
"""
SELECT product_cd, clause_product_nm, insurance_type,
product_type, clause_version, sale_status,
active_yn, version_div
FROM POC_2.KB_PRODUCTS
WHERE product_cd IN ('41047', '41047_OLD')
ORDER BY clause_version DESC
""",
)
evidence["catalog_documents"] = _catalog_documents(
cursor,
("41047",),
)
evidence["kb_clause_chunks"] = _catalog_chunks(
cursor,
product_code="41047",
first_term="대물배상",
second_term="자기차량손해",
row_limit=1,
)
elif kind in {"METADATA_CATALOG", "LINEAGE_TRACE"}:
evidence["products"] = _business_rows(
cursor,
"""
SELECT product_cd, clause_product_nm, insurance_type,
product_type, clause_version, sale_status,
active_yn, version_div
FROM POC_2.KB_PRODUCTS
WHERE product_cd = '41048'
""",
)
required_groups: tuple[str, ...]
if kind == "VISIBLE_CONTRACT_SCOPE":
required_groups = ("contracts",)
elif kind == "CROSS_HOLDING":
required_groups = (
"contracts",
"products",
"external_holdings",
"own_clause_documents",
"external_clause_documents",
)
elif kind == "CUSTOMER_COVERAGE":
required_groups = ("contracts", "coverages", "kb_clause_chunks")
elif kind == "CUSTOMER_CONTRACTS":
required_groups = ("contracts", "products", "kb_clause_chunks")
elif kind in {"AUTO_RENEWAL_COMPARISON", "RENEWAL_CONSULTING"}:
required_groups = (
"contracts",
"products",
"external_holdings",
"kb_clause_chunks",
"external_clause_chunks",
)
elif kind in {"METADATA_CATALOG", "VERSION_GOVERNANCE"}:
required_groups = ("products", "catalog_documents", "kb_clause_chunks")
else:
required_groups = (
"catalog_documents",
"kb_clause_chunks",
)
if kind == "CLAUSE_COMPARISON":
required_groups += ("external_clause_chunks",)
evidence["evidence_complete"] = all(
bool(evidence.get(group)) for group in required_groups
)
evidence["evidence_sources"] = [
"POC_2.KB_CONTRACTS/KB_PRODUCTS/KB_COVERAGES/KB_EXTERNAL_HOLDINGS",
"ADMIN.KB_DOCUMENT_V2/KB_CHUNK_ACTIVE_V",
]
except (AuditLogError, oracledb.Error, OSError, ValueError) as exc:
evidence["evidence_error"] = type(exc).__name__
return evidence
def _first_mapping(
evidence: Mapping[str, Any],
key: str,
*,
predicate: Any | None = None,
) -> Mapping[str, Any]:
values = evidence.get(key)
if not isinstance(values, list):
return {}
for item in values:
if isinstance(item, Mapping) and (predicate is None or predicate(item)):
return item
return {}
def _evidence_date(value: Any) -> str:
return str(value or "").split(" ", 1)[0]
def _evidence_money(value: Any) -> str:
try:
return f"{int(value):,}원"
except (TypeError, ValueError):
return "확인 필요"
def _chunk_matching(
evidence: Mapping[str, Any],
key: str,
*needles: str,
) -> Mapping[str, Any]:
values = evidence.get(key)
if not isinstance(values, list):
return {}
for item in values:
if not isinstance(item, Mapping):
continue
text = str(item.get("excerpt") or "")
if all(needle in text for needle in needles):
return item
return _first_mapping(evidence, key)
def _chunk_reference(chunk: Mapping[str, Any]) -> str:
source_file = str(chunk.get("source_file") or "")
file_name = source_file.rsplit("/", 1)[-1]
locator = str(chunk.get("logical_locator") or "약관 본문")
return (
f"{locator}, {file_name}, {chunk.get('chunk_id')}, "
f"p.{chunk.get('page_start')}-{chunk.get('page_end')}"
)
def _auto_clause_comparison_table(evidence: Mapping[str, Any]) -> str:
kb_anchor = _chunk_matching(
evidence,
"kb_clause_chunks",
"6조6조보상하는 손해",
)
kb_coverage = _chunk_matching(
evidence,
"kb_clause_chunks",
"② ｢대물배상｣",
)
kb_exclusion = _chunk_matching(
evidence,
"kb_clause_chunks",
"보상하지 않는 손해",
)
external_coverage = _chunk_matching(
evidence,
"external_clause_chunks",
"1사고당 보험가입금액",
)
external_exclusion = _chunk_matching(
evidence,
"external_clause_chunks",
"보상하지 않는 손해",
)
kb_source_file = str(kb_anchor.get("source_file") or "").rsplit("/", 1)[-1]
kb_coverage_reference = (
f"제6조 보상하는 손해, {kb_source_file}, "
f"{kb_anchor.get('chunk_id')} + {kb_coverage.get('chunk_id')}, "
f"p.{kb_anchor.get('page_start')}-{kb_coverage.get('page_end')}"
)
return (
"| 보험사·상품코드 | 조항 | 확인 내용 | 원본 근거 |\n"
"|---|---|---|---|\n"
"| 기준 데이터 41048 | 제6조 보상하는 손해 | 피보험자동차 사고로 "
"타인의 재물을 없애거나 훼손해 부담한 법률상 손해배상책임을 보상 | "
f"{kb_coverage_reference} |\n"
"| 기준 데이터 41048 | 제8조 보상하지 않는 손해 | 고의, 전쟁·폭동, "
"천재지변, 핵연료 영향, 반복적 유상 사용 등은 약관상 제외 조건 | "
f"{_chunk_reference(kb_exclusion)} |\n"
"| 삼성화재 20071 | 대물배상 상품요약 | 타인 차량·재물 손해를 "
"1사고당 가입금액 한도로 수리비·교환가액·대차료·휴차료·영업손실·"
"시세하락손해 범위에서 보상 | "
f"{_chunk_reference(external_coverage)} |\n"
"| 삼성화재 20071 | 보상하지 않는 손해 | 고의, 전쟁·내란·폭동, "
"천재지변, 핵연료 영향, 반복적 유상 사용 등 제외 조건을 별도 확인 | "
f"{_chunk_reference(external_exclusion)} |"
)
def _business_answer_from_evidence(evidence: Mapping[str, Any] | None) -> str:
if not evidence or evidence.get("evidence_error"):
return ""
kind = str(evidence.get("evidence_type") or "")
user_id = str(evidence.get("verified_user_id") or evidence.get("user_id") or "")
customer_id = str(evidence.get("customer_id") or "")
contracts = [
item for item in evidence.get("contracts", []) if isinstance(item, Mapping)
]
products = [
item for item in evidence.get("products", []) if isinstance(item, Mapping)
]
product = products[0] if products else {}
kb_doc = _first_mapping(
evidence,
"catalog_documents",
predicate=lambda item: str(item.get("product_code") or "") == "41048",
)
external_doc = _first_mapping(
evidence,
"catalog_documents",
predicate=lambda item: str(item.get("product_code") or "") == "20071",
)
kb_chunk = _first_mapping(evidence, "kb_clause_chunks")
external_chunk = _first_mapping(evidence, "external_clause_chunks")
if kind == "VISIBLE_CONTRACT_SCOPE":
rows = "\n".join(
f"- {item.get('contract_no')}: {item.get('fc_channel')} / "
f"{item.get('fc_id')} / {item.get('contract_status')}"
for item in contracts
)
return (
"이 답변은 전체 계약 간 차이 비교가 아니라 현재 사용자에게 허용된 조회 "
f"범위를 확인한 결과입니다. {user_id} 권한에서 {customer_id} 고객의 본인 "
f"담당 계약은 {len(contracts):,}건입니다.\n{rows}\n\n"
"권한 내 두 계약의 채널·담당자는 설계사/FC00789로 확인됩니다. 권한 밖 "
"다이렉트 계약의 식별자와 상세는 조회·노출하지 않았으므로 이 결과를 고객의 "
"전체 계약 비교로 일반화할 수 없습니다. 전체 채널 비교는 별도 채널 통합 "
"검증 사용자 시나리오로 분리합니다."
).strip()
if kind == "CROSS_HOLDING":
contract = contracts[0] if contracts else {}
holding = _first_mapping(evidence, "external_holdings")
own_limit = _chunk_matching(evidence, "own_clause_chunks", "90일")
own_refund = _chunk_matching(evidence, "own_clause_chunks", "해약환급금")
ext_limit = _chunk_matching(evidence, "external_clause_chunks", "90일")
ext_reduction = _chunk_matching(
evidence,
"external_clause_chunks",
"일정기간 보험금",
)
ext_renewal = _chunk_matching(
evidence,
"external_clause_chunks",
"갱신 시 보험료",
)
return (
f"{customer_id}는 당사와 타사 건강보험을 모두 보유한 교차보유 고객입니다.\n\n"
f"- 당사: {contract.get('contract_no')} / {contract.get('product_cd')} / "
f"{product.get('insurance_type')}·{product.get('product_type')} / "
f"{contract.get('contract_status')} / {contract.get('pay_cycle')}\n"
f"- 타사: {holding.get('ext_insurer')} / {holding.get('ext_product_grp')}·"
f"{holding.get('ext_product_type')} / 갱신예정월 "
f"{holding.get('ext_renew_month')} / {holding.get('ext_file_nm')}\n\n"
"검증 경로: `KB_CONTRACTS.CUST_ID = KB_EXTERNAL_HOLDINGS.CUST_ID`를 "
"독립 EXISTS로 확인해 당사 계약과 타사 보유를 결합했습니다.\n\n"
"| 비교 항목 | 당사 25213_B | DB손보 약관_31084(03)_20260101 |\n"
"|---|---|---|\n"
"| 암 보장 면책 | 암진단비·암수술비 등 일부 담보는 가입 후 90일간 "
f"보장 제외 ({_chunk_reference(own_limit)}) | 암진단담보 예시도 가입 후 "
f"90일간 보장 제외 ({_chunk_reference(ext_limit)}) |\n"
"| 감액·한도 | 면책기간·감액지급·보장한도·자기부담금 조건을 담보별 "
f"확인 ({_chunk_reference(own_limit)}) | 가입 후 일정 기간 50% 지급, "
f"최초 1회·입원일수 한도 예시 확인 ({_chunk_reference(ext_reduction)}) |\n"
"| 해약환급·갱신 | 납입기간 중 해지 시 환급금 제한 구조를 확인 "
f"({_chunk_reference(own_refund)}) | 해약환급금 제한과 갱신 시 연령·"
f"위험률에 따른 보험료 인상 가능성 확인 ({_chunk_reference(ext_renewal)}) |\n\n"
"상담에서는 실제 가입 담보의 진단 정의, 지급 조건, 면책·감액기간, 갱신 "
"여부와 해약환급금 구조를 위 조항과 1:1로 대조해야 합니다."
)
if kind == "CUSTOMER_COVERAGE":
contract = _first_mapping(
evidence,
"contracts",
predicate=lambda item: item.get("contract_status") == "정상",
)
coverage = _first_mapping(
evidence,
"coverages",
predicate=lambda item: "자기차량" in str(item.get("coverage_nm") or ""),
)
coverage_clause = _chunk_matching(
evidence,
"kb_clause_chunks",
"① ｢자기차량손해｣",
)
exclusion_clause = _chunk_matching(
evidence,
"kb_clause_chunks",
"23조23조보상하지 않는 손해",
)
calculation_clause = _chunk_matching(
evidence,
"kb_clause_chunks",
"지급보험금=",
)
return (
f"{customer_id}의 {contract.get('contract_no')}은 정상 계약이며 상품코드는 "
f"{contract.get('product_cd')}입니다. 담보는 {coverage.get('coverage_nm')}"
f"({coverage.get('coverage_type')}), 가입금액은 "
f"{_evidence_money(coverage.get('insured_amt'))}, 갱신예정일은 "
f"{coverage.get('renew_due_dt')}입니다.\n\n"
"약관 근거는 다음과 같습니다.\n"
f"- 제21조(보상하는 손해): 타인 자동차와의 충돌은 상대 차량 등록번호와 "
"운전자 또는 소유자가 확인된 경우, 그리고 피보험자동차 전부 도난으로 인한 "
"직접 손해를 보험가입금액 한도에서 보상합니다. 보험가입금액이 보험가액보다 "
f"크면 보험가액이 한도입니다. ({_chunk_reference(coverage_clause)})\n"
"- 제23조(보상하지 않는 손해): 고의, 전쟁·폭동, 천재지변, 핵연료 영향, "
"반복적 유상 사용, 사기·횡령, 자연소모, 일부 부품만의 도난, 시험·경기용 "
f"사용 등은 제외됩니다. ({_chunk_reference(exclusion_clause)})\n"
"- 제24조(지급보험금의 계산): 손해액과 약정 비용에서 보험증권상 "
f"자기부담금을 공제합니다. ({_chunk_reference(calculation_clause)})\n\n"
f"원본 문서: {kb_doc.get('document_id')} / {kb_doc.get('source_file')}"
)
if kind == "CUSTOMER_CONTRACTS":
rows = "\n".join(
f"- {item.get('contract_no')}: {item.get('product_cd')} / "
f"{item.get('contract_status')}"
for item in contracts
)
property_clause = _chunk_matching(
evidence,
"kb_clause_chunks",
"② ｢대물배상｣",
)
exclusion_clause = _chunk_matching(
evidence,
"kb_clause_chunks",
"보상하지 않는 손해",
)
return (
f"{user_id} 권한에서 {customer_id}의 본인 담당 계약은 {len(contracts):,}건이며 "
f"보험종류는 {product.get('insurance_type')}입니다.\n{rows}\n\n"
"검증 경로는 `KB_CONTRACTS.PRODUCT_CD = KB_PRODUCTS.PRODUCT_CD`이며, "
f"상품 41048의 보험종류={product.get('insurance_type')}, 상품유형="
f"{product.get('product_type')}, 버전={_evidence_date(product.get('clause_version'))}, "
f"상태={product.get('version_div')}을 확인했습니다. 다이렉트 담당 계약은 "
"권한 밖이므로 노출하지 않았습니다.\n\n"
"41048 현행 약관의 주요 보장종목은 ① 대인배상Ⅰ, ② 대인배상Ⅱ, "
"③ 대물배상, ④ 자기신체사고, ⑤ 무보험자동차에 의한 상해, "
"⑥ 자기차량손해입니다. 제6조는 대인배상Ⅱ·대물배상의 보상 범위를, "
"제8조는 고의·천재지변·반복적 유상사용 등 보상 제외 조건을 규정합니다.\n\n"
f"- 제6조 근거: {_chunk_reference(property_clause)}\n"
f"- 제8조 근거: {_chunk_reference(exclusion_clause)}\n"
f"- 원본 문서: {kb_doc.get('document_id')} / {kb_doc.get('source_file')}"
)
if kind in {"AUTO_RENEWAL_COMPARISON", "RENEWAL_CONSULTING"}:
contract = _first_mapping(
evidence,
"contracts",
predicate=lambda item: item.get("contract_status") == "정상",
)
holding = _first_mapping(evidence, "external_holdings")
return (
f"{user_id} 권한에서 비교 가능한 당사 정상 계약은 "
f"{contract.get('contract_no')} 1건이며, 상품은 "
f"{product.get('clause_product_nm')}({contract.get('product_cd')}, "
f"{product.get('version_div')} {_evidence_date(product.get('clause_version'))})입니다. "
"다른 담당자의 계약은 비교 범위에 포함하지 않았습니다.\n\n"
f"타사 보유는 {holding.get('ext_insurer')} "
f"{holding.get('ext_clause_nm')}({holding.get('ext_product_grp')}·"
f"{holding.get('ext_product_type')}, {holding.get('ext_sale_status') or '판매중'})이며 "
f"갱신예정월은 {holding.get('ext_renew_month')}입니다. 따라서 2026-08 전에 "
"전환 또는 유지 상담 대상으로 분류합니다.\n\n"
f"{_auto_clause_comparison_table(evidence)}\n\n"
"제안 포인트: 동일 가입한도 기준으로 대물 확대특약과 면책조건을 먼저 "
"대조하고, 자기차량손해의 충돌·도난 범위와 자기부담금, 무보험차 상해, "
"긴급출동·운전자범위 특약, 갱신 보험료를 순서대로 비교합니다.\n\n"
f"문서 식별자: KB={kb_doc.get('document_id')}, "
f"삼성화재={external_doc.get('document_id')}; 타사 원장 파일키="
f"{holding.get('ext_file_nm')}"
)
if kind == "METADATA_CATALOG":
return (
"41048 약관은 상품 메타, 문서 카탈로그, 조항 청크가 하나의 검색 경로로 "
"통합되어 있습니다. 원시 카탈로그의 metadata_status가 partial인 항목은 "
"DB 값을 수정하지 않고 포털 조회 단계에서 조항번호·논리 위치를 본문 패턴으로 "
"자동 파생해 보강합니다.\n\n"
f"- 상품코드: {product.get('product_cd')}\n"
f"- 보험종류/상품유형: {product.get('insurance_type')} / "
f"{product.get('product_type')}\n"
f"- 약관버전: {_evidence_date(product.get('clause_version'))}\n"
f"- 판매상태/활성/버전구분: {product.get('sale_status')} / "
f"{product.get('active_yn')} / {product.get('version_div')}\n"
f"- document_id: {kb_doc.get('document_id')}\n"
f"- chunk_id: {kb_chunk.get('chunk_id')}\n"
f"- 조항번호/논리 위치: {kb_chunk.get('derived_article_number')} / "
f"{kb_chunk.get('logical_locator')}\n"
f"- chunk_type: {kb_chunk.get('chunk_type')}\n"
f"- 원본: {kb_doc.get('source_file')}\n"
f"- physical_page_range: {kb_chunk.get('physical_page_range')}\n"
f"- 저장 메타 상태/활성 대상: {kb_doc.get('metadata_status')} / "
f"{kb_doc.get('activation_eligible')}\n\n"
"따라서 상품코드·보험종류·상품유형·버전·판매상태뿐 아니라 조항번호, "
"청크ID, 원본 PDF와 페이지가 검색 가능한 응답 메타로 제공됩니다."
)
if kind == "LINEAGE_TRACE":
return (
"약관 근거는 검증 기준의 순서로 원본까지 역추적할 수 있습니다.\n\n"
f"검색 응답 → 실제 chunk_id `{kb_chunk.get('chunk_id')}` → product_cd "
f"`{kb_doc.get('product_code')}` → 02_자사상품원장 `KB_PRODUCTS`의 "
f"`{product.get('clause_product_nm')}`(약관버전 "
f"{_evidence_date(product.get('clause_version'))}, "
f"{product.get('version_div')}) → 원본 PDF `{kb_doc.get('source_file')}` → "
f"{kb_chunk.get('logical_locator')} / p.{kb_chunk.get('physical_page_range')}\n\n"
f"보조 문서키는 `{kb_doc.get('document_id')}`, chunk_type은 "
f"`{kb_chunk.get('chunk_type')}`입니다. 정답지의 `CHK_41048_03`은 형식 "
"예시이며, 위 값이 현재 카탈로그에 저장된 실제 청크ID입니다."
)
if kind == "VERSION_GOVERNANCE":
current_product = next(
(item for item in products if item.get("product_cd") == "41047"),
{},
)
old_product = next(
(item for item in products if item.get("product_cd") == "41047_OLD"),
{},
)
current_doc = _first_mapping(
evidence,
"catalog_documents",
predicate=lambda item: item.get("activation_eligible") == 1,
)
old_doc = _first_mapping(
evidence,
"catalog_documents",
predicate=lambda item: item.get("activation_eligible") == 0,
)
return (
"41047 답변 근거는 현행 약관으로 제한됩니다.\n\n"
f"- 사용: 41047 / {_evidence_date(current_product.get('clause_version'))} / "
f"{current_product.get('sale_status')} / 활성 {current_product.get('active_yn')} / "
f"{current_product.get('version_div')} / {current_doc.get('source_file')}\n"
f"- 제외: 41047_OLD / {_evidence_date(old_product.get('clause_version'))} / "
f"{old_product.get('sale_status')} / 활성 {old_product.get('active_yn')} / "
f"{old_product.get('version_div')} / ingestion_status="
f"{old_doc.get('ingestion_status')}\n\n"
f"실제 검색 청크도 현행 문서의 {kb_chunk.get('chunk_id')}만 사용했습니다."
)
if kind == "CLAUSE_COMPARISON":
return (
f"{_auto_clause_comparison_table(evidence)}\n\n"
"비교 결과, 양쪽 모두 타인 재물 손해배상책임을 기본 보장으로 두지만 실제 "
"차이는 가입한 보상한도·확대특약과 제8조 계열 면책조건, 대차료·휴차료·"
"영업손실·시세하락손해 지급기준을 동일 조건으로 대조해야 확정할 수 있습니다.\n\n"
f"- KB 문서ID/원본: {kb_doc.get('document_id')} / "
f"{kb_doc.get('file_name')}\n"
f"- 삼성화재 문서ID/원본 파일키/실제 PDF: "
f"{external_doc.get('document_id')} / 20071_0_20260611_file1 / "
f"{external_doc.get('file_name')}"
)
return ""
def _enrich_answer_with_business_evidence(
answer: str,
evidence: Mapping[str, Any] | None,
) -> str:
verified = _business_answer_from_evidence(evidence)
normalized = str(answer or "").strip()
if not verified:
return normalized
if evidence and evidence.get("evidence_complete"):
return verified
return f"{verified}\n\n{normalized}".strip()
def _runtime_env_value(*names: object) -> str:
for name in names:
key = str(name or "").strip()
if not key:
continue
value = (os.environ.get(key) or _dotenv_value(key)).strip()
if value:
return value
return ""
def _looks_like_http_url(value: str) -> bool:
return value.startswith(("http://", "https://"))
def _resolve_mcp_endpoint_config(item: Mapping[str, Any]) -> str:
raw_endpoint = str(
item.get("endpoint_url") or item.get("mcp_endpoint") or ""
).strip()
if _looks_like_http_url(raw_endpoint):
return raw_endpoint
if raw_endpoint:
resolved = _runtime_env_value(raw_endpoint)
if resolved:
return resolved
for env_key in ("endpoint_url_env", "mcp_endpoint_env", "base_url_env"):
resolved = _runtime_env_value(item.get(env_key))
if resolved:
return resolved
return raw_endpoint
def load_vpd_token_presets(
path: Path = VPD_TOKEN_PRESETS_FILE,
) -> tuple[VpdTokenPreset, ...]:
try:
payload = json.loads(path.read_text(encoding="utf-8"))
except FileNotFoundError:
return ()
except (OSError, UnicodeError, ValueError):
raise PublicMcpError(f"데모 사용자 preset 설정을 읽지 못했습니다: {path}") from None
raw_presets = payload.get("presets") if isinstance(payload, Mapping) else None
if not isinstance(raw_presets, list):
raise PublicMcpError("데모 사용자 preset 설정에 presets 배열이 필요합니다.")
presets: list[VpdTokenPreset] = []
seen: set[str] = set()
for item in raw_presets:
if not isinstance(item, Mapping) or item.get("enabled", True) is not True:
continue
token_env = str(
item.get("mcp_token_env") or item.get("token_env") or ""
).strip()
token = _normalized_bearer(
_runtime_env_value(token_env) if token_env else item.get("token")
)
user_id = str(item.get("user_id") or "").strip()
if not user_id or user_id in seen:
continue
presets.append(
VpdTokenPreset(
token=token,
user_id=user_id,
name=str(item.get("name") or "").strip(),
role=str(item.get("role") or "").strip(),
channel=str(item.get("channel") or item.get("team") or "").strip(),
scope=str(item.get("scope") or "").strip(),
is_default=item.get("default") is True,
team=str(item.get("team") or "").strip(),
)
)
seen.add(user_id)
return tuple(presets)
class _NoRedirectHandler(HTTPRedirectHandler):
"""Do not forward Authorization to a redirected endpoint."""
def redirect_request(self, request, fp, code, msg, headers, newurl): # type: ignore[no-untyped-def]
del request, fp, code, msg, headers, newurl
return None
def load_mcp_servers(path: Path = MCP_SERVERS_FILE) -> tuple[list[McpServer], int]:
try:
payload = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeError, ValueError):
raise PublicMcpError(f"MCP 서버 설정을 읽지 못했습니다: {path}") from None
if not isinstance(payload, Mapping):
raise PublicMcpError("MCP 서버 설정 형식이 올바르지 않습니다.")
default_server_id = str(payload.get("default_server_id") or "").strip()
raw_servers = payload.get("servers")
if not isinstance(raw_servers, list):
raise PublicMcpError("MCP 서버 설정에 servers 배열이 필요합니다.")
servers: list[McpServer] = []
default_index = 0
for item in raw_servers:
if not isinstance(item, Mapping) or item.get("enabled", True) is not True:
continue
server_id = str(item.get("id") or "").strip()
endpoint_url = _resolve_mcp_endpoint_config(item)
if not server_id or not endpoint_url:
continue
raw_allowlist = item.get("tool_allowlist", [])
allowlist = (
tuple(str(name).strip() for name in raw_allowlist if str(name).strip())
if isinstance(raw_allowlist, list)
else ()
)
server = McpServer(
server_id=server_id,
endpoint_url=endpoint_url,
auth_token_env=str(item.get("auth_token_env") or "").strip(),
default_tool=str(item.get("default_tool") or PREFERRED_TOOL).strip(),
tool_allowlist=allowlist,
router_model_profile=str(
item.get("router_model_profile") or "gpt55_oci"
).strip(),
description=str(item.get("description") or ""),
)
if server.server_id == default_server_id:
default_index = len(servers)
servers.append(server)
if not servers:
raise PublicMcpError("사용 가능한 MCP 서버 설정이 없습니다.")
return servers, default_index
def _mcp_endpoint(base_url: str) -> str:
value = str(base_url or "").strip().rstrip("/")
if not value:
raise PublicMcpError("MCP URL을 입력해 주세요.")
parts = urlsplit(value)
if parts.scheme not in {"http", "https"} or not parts.netloc:
raise PublicMcpError("MCP URL 형식이 올바르지 않습니다.")
if parts.query or parts.fragment:
raise PublicMcpError("MCP URL에는 query/fragment를 넣지 마세요.")
return value if parts.path.endswith("/mcp") else f"{value}/mcp"
def _safe_request_id(prefix: str) -> str:
return f"poc4-{prefix}"
def _jsonrpc_payload(raw: bytes, request_id: str) -> Mapping[str, Any]:
if len(raw) > MAX_RESPONSE_BYTES:
raise PublicMcpError("MCP 응답이 너무 큽니다.")
try:
text = raw.decode("utf-8")
except UnicodeDecodeError:
raise PublicMcpError("MCP 응답이 JSON 형식이 아닙니다.") from None
stripped = text.strip()
if stripped.startswith("event:") or "\ndata:" in stripped:
data_lines = [
line[5:].strip()
for line in stripped.splitlines()
if line.startswith("data:")
]
stripped = "\n".join(data_lines).strip()
try:
payload = json.loads(stripped)
except ValueError:
raise PublicMcpError("MCP 응답이 JSON 형식이 아닙니다.") from None
if not isinstance(payload, Mapping):
raise PublicMcpError("MCP 응답 형식이 올바르지 않습니다.")
if payload.get("jsonrpc") != "2.0" or payload.get("id") != request_id:
raise PublicMcpError("MCP JSON-RPC 응답 형식이 올바르지 않습니다.")
if "error" in payload:
raise PublicMcpError("MCP 서버가 오류를 반환했습니다.")
return payload
def _jsonrpc_exchange(
*,
base_url: str,
bearer_token: str,
method: str,
params: Mapping[str, Any] | None = None,
request_id: str,
session_id: str = "",
) -> _JsonRpcExchange:
token = str(bearer_token or "").strip()
if not token:
raise PublicMcpError("Bearer 토큰을 입력해 주세요.")
if token.lower().startswith("bearer "):
token = token[7:].strip()
if not token or any(ch.isspace() for ch in token):
raise PublicMcpError("Bearer 토큰 형식이 올바르지 않습니다.")
message = {
"jsonrpc": "2.0",
"id": request_id,
"method": method,
"params": dict(params or {}),
}
body = json.dumps(
message,
ensure_ascii=False,
separators=(",", ":"),
).encode("utf-8")
headers = {
"Accept": "application/json, text/event-stream",
"Content-Type": "application/json",
"MCP-Protocol-Version": MCP_PROTOCOL_VERSION,
"Authorization": f"Bearer {token}",
}
if session_id:
headers["Mcp-Session-Id"] = session_id
request = Request(
_mcp_endpoint(base_url),
data=body,
method="POST",
headers=headers,
)
try:
with build_opener(_NoRedirectHandler()).open(request, timeout=60) as response:
raw = response.read(MAX_RESPONSE_BYTES + 1)
response_session_id = (
response.headers.get("Mcp-Session-Id")
or response.headers.get("mcp-session-id")
or ""
)
except HTTPError as exc:
if exc.code in {401, 403}:
raise PublicMcpError("MCP 인증에 실패했습니다. Bearer 토큰을 확인하세요.") from None
raise _McpHttpStatusError(exc.code) from None
except (URLError, TimeoutError, OSError):
raise PublicMcpError("MCP 서버에 연결하지 못했습니다.") from None
payload = _jsonrpc_payload(raw, request_id)
result = payload.get("result")
if not isinstance(result, Mapping):
raise PublicMcpError("MCP result 형식이 올바르지 않습니다.")
return _JsonRpcExchange(result=result, session_id=response_session_id)
def _jsonrpc_notification(
*,
base_url: str,
bearer_token: str,
method: str,
params: Mapping[str, Any] | None = None,
session_id: str,
) -> None:
token = str(bearer_token or "").strip()
if token.lower().startswith("bearer "):
token = token[7:].strip()
message = {"jsonrpc": "2.0", "method": method, "params": dict(params or {})}
body = json.dumps(
message,
ensure_ascii=False,
separators=(",", ":"),
).encode("utf-8")
headers = {
"Accept": "application/json, text/event-stream",
"Content-Type": "application/json",
"MCP-Protocol-Version": MCP_PROTOCOL_VERSION,
"Authorization": f"Bearer {token}",
"Mcp-Session-Id": session_id,
}
request = Request(_mcp_endpoint(base_url), data=body, method="POST", headers=headers)
try:
with build_opener(_NoRedirectHandler()).open(request, timeout=60) as response:
response.read(MAX_RESPONSE_BYTES + 1)
except HTTPError as exc:
if exc.code in {202, 204}:
return
if exc.code in {401, 403}:
raise PublicMcpError("MCP 인증에 실패했습니다. Bearer 토큰을 확인하세요.") from None
raise _McpHttpStatusError(exc.code) from None
except (URLError, TimeoutError, OSError):
raise PublicMcpError("MCP 서버에 연결하지 못했습니다.") from None
def _jsonrpc(
*,
base_url: str,
bearer_token: str,
method: str,
params: Mapping[str, Any] | None = None,
request_id: str,
session_id: str = "",
) -> Mapping[str, Any]:
return _jsonrpc_exchange(
base_url=base_url,
bearer_token=bearer_token,
method=method,
params=params,
request_id=request_id,
session_id=session_id,
).result
def _jsonrpc_with_session_fallback(
*,
base_url: str,
bearer_token: str,
method: str,
params: Mapping[str, Any] | None = None,
request_id: str,
) -> Mapping[str, Any]:
try:
return _jsonrpc(
base_url=base_url,
bearer_token=bearer_token,
method=method,
params=params,
request_id=request_id,
)
except _McpHttpStatusError as exc:
if exc.code != 400:
raise
return _session_jsonrpc(
base_url=base_url,
bearer_token=bearer_token,
method=method,
params=params,
request_id=request_id,
)
def _session_jsonrpc(
*,
base_url: str,
bearer_token: str,
method: str,
params: Mapping[str, Any] | None,
request_id: str,
) -> Mapping[str, Any]:
initialized = _jsonrpc_exchange(
base_url=base_url,
bearer_token=bearer_token,
method="initialize",
params={
"protocolVersion": MCP_PROTOCOL_VERSION,
"capabilities": {},
"clientInfo": {"name": "poc4-mcp-discovery-ui", "version": "0.1.0"},
},
request_id=_safe_request_id("initialize"),
)
if not initialized.session_id:
raise PublicMcpError("MCP 세션 ID를 받지 못했습니다.")
_jsonrpc_notification(
base_url=base_url,
bearer_token=bearer_token,
method="notifications/initialized",
params={},
session_id=initialized.session_id,
)
return _jsonrpc(
base_url=base_url,
bearer_token=bearer_token,
method=method,
params=params,
request_id=request_id,
session_id=initialized.session_id,
)
def discover_tools(base_url: str, bearer_token: str) -> list[McpTool]:
result = _jsonrpc_with_session_fallback(
base_url=base_url,
bearer_token=bearer_token,
method="tools/list",
params={},
request_id=_safe_request_id("tools-list"),
)
raw_tools = result.get("tools", [])
if not isinstance(raw_tools, list):
raise PublicMcpError("MCP tools/list 응답 형식이 올바르지 않습니다.")
tools: list[McpTool] = []
for item in raw_tools:
if not isinstance(item, Mapping):
continue
name = str(item.get("name") or "").strip()
if not name:
continue
schema = item.get("inputSchema")
tools.append(
McpTool(
name=name,
description=str(item.get("description") or ""),
schema=schema if isinstance(schema, Mapping) else {},
read_only=item.get("readOnly") is not False,
)
)
return tools
def discover_enabled_server_tools(
servers: list[McpServer], bearer_token: str
) -> tuple[list[McpDiscoveryResult], list[dict[str, str]]]:
discovered: list[McpDiscoveryResult] = []
failures: list[dict[str, str]] = []
for server in servers:
try:
tools = discover_tools(server.endpoint_url, bearer_token)
except PublicMcpError as exc:
failures.append({"server_id": server.server_id, "error": str(exc)})
continue
allowed = tuple(
tool
for tool in tools
if not server.tool_allowlist or tool.name in server.tool_allowlist
)
discovered.append(McpDiscoveryResult(server=server, tools=allowed))
return discovered, failures
def _mcp_server_cache_rows(
servers: list[McpServer],
) -> tuple[tuple[str, str, str, str, tuple[str, ...], str, str], ...]:
return tuple(
(
server.server_id,
server.endpoint_url,
server.auth_token_env,
server.default_tool,
server.tool_allowlist,
server.router_model_profile,
server.description,
)
for server in servers
)
def _mcp_servers_from_cache_rows(
rows: tuple[tuple[str, str, str, str, tuple[str, ...], str, str], ...],
) -> list[McpServer]:
return [
McpServer(
server_id=row[0],
endpoint_url=row[1],
auth_token_env=row[2],
default_tool=row[3],
tool_allowlist=tuple(row[4]),
router_model_profile=row[5],
description=row[6],
)
for row in rows
]
@st.cache_data(show_spinner=False)
def cached_discover_enabled_server_tools(
server_rows: tuple[tuple[str, str, str, str, tuple[str, ...], str, str], ...],
token_fingerprint: str,
cache_generation: int,
_bearer_token: str,
) -> tuple[list[McpDiscoveryResult], list[dict[str, str]]]:
"""Cache tools/list until the user explicitly refreshes it."""
del token_fingerprint, cache_generation
return discover_enabled_server_tools(
_mcp_servers_from_cache_rows(server_rows),
_bearer_token,
)
def call_tool(
*,
base_url: str,
bearer_token: str,
tool: McpTool,
arguments: Mapping[str, Any],
) -> Mapping[str, Any]:
return _jsonrpc_with_session_fallback(
base_url=base_url,
bearer_token=bearer_token,
method="tools/call",
params={"name": tool.name, "arguments": dict(arguments)},
request_id=_safe_request_id("tools-call"),
)
def _content_text_json(result: Mapping[str, Any]) -> Any:
content = result.get("content")
if not isinstance(content, list) or not content:
return None
first = content[0]
if not isinstance(first, Mapping) or first.get("type") != "text":
return None
text = first.get("text")
if not isinstance(text, str):
return None
try:
return json.loads(text)
except ValueError:
return text
def _mcp_response_payload(mcp_result: Any) -> Mapping[str, Any]:
if isinstance(mcp_result, Mapping):
response = mcp_result.get("response")
if isinstance(response, Mapping):
return response
return mcp_result
return {}
def _mcp_generated_sql(mcp_result: Any) -> str:
payload = _mcp_response_payload(mcp_result)
return str(payload.get("generatedSql") or payload.get("generated_sql") or "").strip()
def _mcp_items(mcp_result: Any) -> list[Any]:
payload = _mcp_response_payload(mcp_result)
items = payload.get("items")
return items if isinstance(items, list) else []
def _mcp_summary(mcp_result: Any) -> dict[str, Any]:
if not isinstance(mcp_result, Mapping):
return {"type": type(mcp_result).__name__}
payload = _mcp_response_payload(mcp_result)
items = _mcp_items(mcp_result)
summary: dict[str, Any] = status_result_summary(mcp_result)
for key in ("toolName", "profile", "ordsPath", "generatedSql"):
value = mcp_result.get(key) if key in mcp_result else payload.get(key)
if value:
summary[key] = value
if items:
summary["items_count"] = len(items)
elif isinstance(payload.get("items"), list):
summary["items_count"] = 0
results = payload.get("results")
if isinstance(results, list):
summary["results_count"] = len(results)
return summary or {"keys": sorted(str(key) for key in mcp_result.keys())}
def _route_key(server_id: str, tool_name: str) -> str:
return f"{server_id}::{tool_name}"
def _agent_tool_catalog(
routed_tools: list[RoutedMcpTool],
) -> tuple[list[dict[str, Any]], dict[str, RoutedMcpTool]]:
routes: dict[str, RoutedMcpTool] = {}
catalog: list[dict[str, Any]] = []
for route in routed_tools:
key = _route_key(route.server_id, route.tool.name)
routes[key] = route
properties = route.tool.schema.get("properties")
catalog.append(
{
"route_key": key,
"server_id": route.server_id,
"tool_name": route.tool.name,
"description": route.tool.description[:1000],
"input_properties": sorted(properties.keys())
if isinstance(properties, Mapping)
else [],
"read_only": route.tool.read_only,
}
)
return catalog, routes
def _complex_reasoning_model_profile(default_model_profile: str) -> str:
configured = (
os.environ.get(COMPLEX_REASONING_MODEL_PROFILE_ENV)
or _dotenv_value(COMPLEX_REASONING_MODEL_PROFILE_ENV)
or os.environ.get(LEGACY_COMPLEX_REASONING_MODEL_PROFILE_ENV)
or _dotenv_value(LEGACY_COMPLEX_REASONING_MODEL_PROFILE_ENV)
).strip()
return configured or DEFAULT_COMPLEX_REASONING_MODEL_PROFILE or default_model_profile
def _model_profile_select_options() -> tuple[tuple[str, str], ...]:
registry = load_model_registry()
return tuple(
(profile.model_key, profile.display_name)
for profile in registry.selector_options()
)
def _default_single_route(routed_tools: list[RoutedMcpTool]) -> RoutedMcpTool:
if not routed_tools:
raise McpToolRouterError("라우팅 가능한 MCP tool이 없습니다.")
for route in routed_tools:
if route.tool.name == PREFERRED_TOOL:
return route
return routed_tools[0]
def _clean_agent_tool_query(value: object, fallback: str) -> str:
text = str(value or "").strip()
if not text:
return fallback
cleaned: list[str] = []
for raw_line in text.splitlines():
line = raw_line.strip()
if not line:
continue
if re.fullmatch(r"(?i)(limit|max_rows|top_k|candidate_k)\s*:\s*\d+", line):
continue
matched = re.match(r"(?i)^(prompt|query|question)\s*:\s*(.+)$", line)
cleaned.append(matched.group(2).strip() if matched else line)
return " ".join(cleaned).strip() or fallback
def _mcp_has_actionable_result(mcp_result: Any) -> bool:
if _mcp_generated_sql(mcp_result) or _mcp_items(mcp_result):
return True
if has_actionable_text_result(mcp_result):
return True
payload = _mcp_response_payload(mcp_result)
results = payload.get("results")
if isinstance(results, list) and results:
return True
result_count = payload.get("result_count")
return isinstance(result_count, int) and result_count > 0
def _compact_evidence_item(value: Any, *, max_text: int = 700) -> Any:
if not isinstance(value, Mapping):
text = str(value)
return text[:max_text] + ("..." if len(text) > max_text else "")
preferred_keys = (
"insurer",
"product_name",
"product_cd",
"product_code",
"contract_no",
"cust_id",
"document_id",
"chunk_id",
"source_file_name",
"file_name",
"ext_file_nm",
"clause_version",
"sale_status",
"active_yn",
"chunk_type",
"logical_locator",
"physical_page_range",
"score",
"display_markdown",
"text",
"content",
)
source = value
compact: dict[str, Any] = {}
keys = [key for key in preferred_keys if key in source]
if not keys:
keys = list(source.keys())[:12]
for key in keys:
item = source.get(key)
if isinstance(item, str):
normalized = " ".join(item.split())
compact[str(key)] = (
normalized[:max_text] + "..."
if len(normalized) > max_text
else normalized
)
else:
compact[str(key)] = item
return compact
def _mcp_answer_evidence(
mcp_result: Any,
*,
max_items: int = 20,
max_text: int = 700,
include_sql: bool = True,
) -> Any:
if not isinstance(mcp_result, Mapping):
return mcp_result
payload = _mcp_response_payload(mcp_result)
evidence: dict[str, Any] = status_result_evidence(
mcp_result,
max_chars=max(3500, min(7000, max_text * 15)),
)
for key in (
"toolName",
"profile",
"ordsPath",
"status",
"success",
"error",
"errorCode",
"errorMessage",
"generatedSql",
"generated_sql",
"result_count",
"row_count",
"audit_event_id",
"audit_log_key",
"candidate_count",
"top_k",
"candidate_k",
"retrieval_mode",
):
if not include_sql and key in {"generatedSql", "generated_sql"}:
continue
value = mcp_result.get(key) if key in mcp_result else payload.get(key)
if value not in (None, "", []):
evidence[key] = value
items = _mcp_items(mcp_result)
if items:
evidence["items"] = [
_compact_evidence_item(item, max_text=max_text)
for item in items[:max_items]
]
if isinstance(payload.get("items"), list):
evidence["items_count"] = len(items)
results = payload.get("results")
if isinstance(results, list) and results:
evidence["results"] = [
_compact_evidence_item(item, max_text=max_text)
for item in results[:max_items]
]
if isinstance(results, list):
evidence["results_count"] = len(results)
return evidence or _mcp_summary(mcp_result)
def _select_ai_query_guidance(question: str) -> list[str]:
text = str(question or "").casefold()
guidance: list[str] = [
"고객번호·고객ID·고객 식별번호는 CUST_ID이며 KB_ 테이블의 실제 컬럼만 "
"사용하고 확인되지 않은 영문 컬럼명을 생성하지 않는다"
]
if any(term in text for term in ("지급 대상", "지급대상", "보험금 지급")):
guidance.append(
"KB_CONTRACTS.CONTRACT_STATUS와 KB_CLAIMS.CLAIM_AMT, "
"PAID_AMT, CLAIM_STATUS를 CONTRACT_NO로 연결해 함께 반환한다"
)
if "보험료" in text and any(
term in text for term in ("합계", "개별", "상세", "마스킹")
):
guidance.append(
"지점장 개별 계약은 CONTRACT_NO와 마스킹 상태만 식별하고 PREMIUM은 "
"KB_CONTRACT_PREMIUM_REDACT 결과를 "
"MASKED로 표시하고 숫자 0을 실제 보험료로 해석하지 않는다. 채널 전체 "
"합계는 개별값과 분리하여 ADMIN.CB_KB_PREMIUM_SUM()의 집계 결과를 반환한다"
)
if any(term in text for term in ("주민번호", "rrn_masked")):
guidance.append(
"KB_CUSTOMERS의 CUST_ID와 RRN_MASKED 실제 반환값 및 반환 건수를 "
"조회하며 WHERE 1=0 같은 무효 조건을 만들지 않는다"
)
if any(term in text for term in ("자차담보", "실손 담보", "담보 구성")):
guidance.append(
"KB_CONTRACTS와 KB_COVERAGES를 CONTRACT_NO로 연결하고 CONTRACT_NO, "
"PRODUCT_CD, COVERAGE_NM, COVERAGE_TYPE, INSURED_AMT, RENEW_DUE_DT를 반환한다"
)
guidance.append(
"자사 담보 질문에는 KB_EXTERNAL_HOLDINGS를 조인하지 않으며 계약·담보·상품을 "
"각각 확인한 뒤 PRODUCT_CD를 후속 약관 검색 식별자로 반환한다"
)
if any(term in text for term in ("리니지", "카탈로그", "메타가 자동")):
guidance.append(
"KB_PRODUCTS의 PRODUCT_CD, INSURANCE_TYPE, PRODUCT_TYPE, "
"CLAUSE_VERSION, SALE_STATUS, ACTIVE_YN을 실제 컬럼으로 조회한다"
)
if "납입" in text and any(term in text for term in ("기준", "맞지", "위반")):
guidance.append(
"KB_CONTRACTS.PRODUCT_CD를 KB_PRODUCTS.PRODUCT_CD와 조인하고 "
"INSURANCE_TYPE이 장기이면 월납·3개월납·6개월납, 자동차 또는 "
"일반이면 연납을 허용 기준으로 PAY_CYCLE 위반 건수만 반환한다"
)
if _question_needs_cross_source(text):
guidance.append(
"고객·계약·상품·타사보험 식별을 위해 CUST_ID, CONTRACT_NO, PRODUCT_CD, "
"EXT_INSURER, EXT_PRODUCT_GRP, EXT_PRODUCT_TYPE, EXT_RENEW_MONTH, "
"EXT_CLAUSE_NM, EXT_FILE_NM을 필요한 테이블에서 함께 반환한다"
)
guidance.append(
"KB_CONTRACTS, KB_PRODUCTS, KB_EXTERNAL_HOLDINGS의 선택적 INNER JOIN으로 "
"전체 결과를 0건으로 만들지 않는다. 당사 계약과 타사 보유를 CUST_ID 기준 "
"독립 EXISTS 또는 별도 결과로 조회하고 답변 단계에서 결합한다"
)
if "c1001025" in text and any(term in text for term in ("건강보험", "건강 보험")):
guidance.append(
"C1001025의 당사 계약은 CONTRACT_STATUS와 PAY_CYCLE까지 반환하고 타사 건강 "
"보유는 EXT_PRODUCT_TYPE LIKE '%건강%'로 판정하며 EXT_PRODUCT_GRP='건강'을 "
"강제하지 않는다. EXT_FILE_NM을 정확히 반환한다"
)
if "c1001006" in text and any(term in text for term in ("내 담당", "상품 유형")):
guidance.append(
"현재 VPD 가시범위의 C1001006 계약을 CONTRACT_NO, PRODUCT_CD, "
"CONTRACT_STATUS별로 먼저 반환하고 타사 보유 테이블을 조인하지 않는다"
)
if "갱신월" in text and "타사" in text:
guidance.append(
"KB_EXTERNAL_HOLDINGS는 CUST_ID와 EXT_FILE_NM 기준으로 중복 제거하고, "
"당사 정상 계약은 별도 조회한 뒤 상담 답변에서 결합한다"
)
if any(term in text for term in ("공통계정", "공통 계정")) and "채널" in text:
guidance.append(
"빈 SUM 1행을 계약 1건으로 해석하지 않는다. SUM(CASE WHEN ... THEN 1 "
"ELSE 0 END) 또는 COUNT(CASE WHEN ... THEN 1 END)에 NVL을 적용해 다이렉트, "
"설계사, GA, 제휴 채널 건수를 각각 숫자 0 이상으로 반환한다"
)
if "41047" in text:
guidance.append(
"KB_PRODUCTS에서 PRODUCT_CD='41047'의 현행·구버전 행을 모두 확인하여 "
"CLAUSE_VERSION, SALE_STATUS, ACTIVE_YN, VERSION_DIV를 반환한다"
)
return guidance
def _vector_query_guidance(question: str) -> list[str]:
text = str(question or "").casefold()
guidance = [
"검색 결과에 document_id, chunk_id, product_cd, 원본 파일명, "
"logical_locator, physical_page_range, chunk_type을 가능한 범위에서 포함한다"
]
if any(term in text for term in ("비교", "차별", "타사", "삼성화재")):
guidance.append(
"자사와 타사 양쪽 약관을 검색하고 상품코드·보험사·원본 파일명·버전을 "
"서로 섞지 않는다"
)
guidance.append(
"비교 대상별 검색을 독립 실행하고 양쪽 문서의 원본 파일명·상품코드·조항·"
"페이지가 모두 확보되지 않으면 비교 완료로 표시하지 않는다"
)
if any(term in text for term in ("구버전", "최신", "현행")):
guidance.append(
"상품코드와 약관버전, 판매상태, 활성여부가 일치하는 현행 문서를 우선하고 "
"OLD·판매중지·비활성 문서는 제외한다"
)
if "41047" in text:
guidance.append(
"product_cd=41047, clause_version=2026-06-11, sale_status=판매중, "
"active_yn=Y, version_div=현행 문서만 근거로 사용하고 41047_OLD와 "
"판매중지·비활성 문서를 제외한다"
)
if "41048" in text:
guidance.append(
"product_cd=41048인 당사 현행 문서를 exact filter하고 원본 파일명과 "
"document_id, chunk_id, 조항 위치, 페이지를 함께 반환한다"
)
if "c1001025" in text:
guidance.append(
"당사 PRODUCT_CD=25213_B 문서와 타사 EXT_FILE_NM="
"약관_31084(03)_20260101 문서를 각각 exact filter한다"
)
if "대물배상" in text and "삼성화재" in text:
guidance.append(
"당사 41048 현행 문서와 삼성화재 20071_0_20260611_file1 문서에서 "
"대물배상 조항만 각각 검색한다"
)
if any(term in text for term in ("리니지", "원본 pdf", "청크", "추적")):
guidance.append(
"응답에 product_cd, clause_version, document_id, chunk_id, "
"source_file_name, logical_locator, physical_page_range를 빠짐없이 반환한다"
)
return guidance
def _append_query_guidance(query: str, guidance: list[str]) -> str:
normalized = str(query or "").strip()
if not guidance:
return normalized
return normalized + " 추가 검증 조건: " + "; ".join(guidance) + "."
def _answer_requirements(question: str) -> list[str]:
text = str(question or "").casefold()
requirements = [
"items_count=0 또는 results_count=0은 결과 전달 누락이 아니라 실제 0건으로 해석한다",
"감사 이벤트 ID나 감사 조회 결과가 없으면 감사로그 기록 여부를 단정하지 않는다",
]
if _question_needs_cross_source(text):
requirements.append(
"구조화 데이터의 계약·상품·타사보유 사실과 벡터 검색의 약관 조항을 구분해 "
"결합하고, 양쪽 출처가 없으면 확인된 항목과 미확인 항목을 나눈다"
)
if "보험료" in text and any(
term in text for term in ("합계", "개별", "상세", "마스킹")
):
requirements.append(
"security_evidence의 premium_sum을 채널 전체 합계로 사용하고 개별 PREMIUM은 "
"MASKED로 표시한다. 마스킹된 숫자 0을 실제 금액이나 합계로 해석하지 않는다"
)
if any(term in text for term in ("지급 대상", "지급대상", "보험금 지급")):
requirements.append(
"지급 판정에는 CONTRACT_STATUS, CLAIM_AMT, PAID_AMT, CLAIM_STATUS를 "
"모두 확인하고 누락 시 재조회 필요 항목을 명시한다"
)
if any(term in text for term in ("주민번호", "rrn_masked")):
requirements.append(
"security_evidence의 plaintext_rows와 masked_format_rows를 근거로 평문 "
"반환 건수와 YYMMDD-N****** 표시 형식을 답하고 원문을 추정하거나 복원하지 않는다"
)
if any(term in text for term in ("담보", "보장 개요", "가입금액")):
requirements.append(
"계약번호, 상품코드, 담보명, 가입금액을 계약 사실로 먼저 제시하고 약관 "
"요약은 별도 근거로 표시한다"
)
if any(term in text for term in ("리니지", "청크", "원본 pdf", "추적")):
requirements.append(
"product_cd, document_id, chunk_id, 원본 파일명, 조항 위치 중 실제 제공된 "
"식별자를 빠짐없이 표시하고 누락 식별자를 명시한다"
)
if _security_evidence_kind(text) in {"CONTRACT_SCOPE", "AUTH_DENIAL"}:
requirements.append(
"security_evidence의 차단 결과, 반환 건수, audit_log_id와 실행 사용자를 "
"함께 제시하고 권한 밖 계약·고객·보험료 상세를 노출하지 않는다"
)
if _security_evidence_kind(text) == "CHANNEL_SCOPE":
requirements.append(
"security_evidence.channel_contract_counts의 네 채널 숫자를 모두 제시하고 "
"전체는 실제 행 COUNT로 계산한다. 빈 집계 행을 1건으로 세지 않는다"
)
if "41047" in text:
requirements.append(
"41047 현행 2026-06-11·판매중·활성 Y 근거만 사용하고 41047_OLD, "
"2022-03-16, 판매중지, 활성 N은 제외됐음을 명시한다"
)
if "대물배상" in text and "삼성화재" in text:
requirements.append(
"41048과 20071_0_20260611_file1의 대물배상 조항을 표로 비교하고 각 행에 "
"상품코드, 원본 파일명, 조항 또는 페이지를 표시한다"
)
return requirements
def _question_needs_cross_source(question: str) -> bool:
text = str(question or "").casefold()
source_terms = (
"약관",
"삼성화재",
"db손보",
"타사",
"경쟁사",
"원본 pdf",
"청크",
"보유 자동차보험",
"갱신 상담",
)
return any(keyword in text for keyword in source_terms)
def _question_prefers_structured_single(question: str) -> bool:
text = str(question or "").casefold()
return any(
term in text
for term in (
"지급 대상",
"지급대상",
"주민번호",
"rrn_masked",
"납입구분",
"납입 기준",
"보험료 합계",
"보험료 상세",
)
)
def _is_search_route(route: RoutedMcpTool) -> bool:
properties = route.tool.schema.get("properties")
if not isinstance(properties, Mapping):
properties = {}
route_text = f"{route.server_id} {route.tool.name} {route.tool.description}".casefold()
return (
_is_vector_route(route)
or "query" in properties
or "search" in route_text
or "vector" in route_text
or "rerank" in route_text
or "hybrid" in route_text
)
def _is_vector_route(route: RoutedMcpTool) -> bool:
route_text = f"{route.server_id} {route.tool.name} {route.tool.description}".casefold()
return any(
marker in route_text
for marker in ("vector", "hybrid", "rerank", "약관 검색")
)
def _is_structured_route(route: RoutedMcpTool) -> bool:
route_text = f"{route.server_id} {route.tool.name} {route.tool.description}".casefold()
return not _is_vector_route(route) and any(
marker in route_text for marker in ("select_ai", "select ai", "ords.query")
)
def _select_unvisited_route(
question: str,
routes_by_key: Mapping[str, RoutedMcpTool],
completed_route_keys: set[str],
) -> tuple[str, RoutedMcpTool] | None:
candidates = [
(key, route)
for key, route in routes_by_key.items()
if key not in completed_route_keys
]
if not candidates:
return None
if _question_needs_cross_source(question):
if not completed_route_keys:
for key, route in candidates:
if _is_structured_route(route):
return key, route
for key, route in candidates:
if _is_vector_route(route):
return key, route
return candidates[0]
def _fallback_tool_query_for_route(
question: str,
route: RoutedMcpTool,
observations: list[Mapping[str, Any]],
) -> str:
text = str(question or "").strip()
if _is_structured_route(route):
return _append_query_guidance(text, _select_ai_query_guidance(text))
if _is_vector_route(route):
structured_context = " ".join(
str(item.get("result_excerpt") or "")
for item in observations
if "kb_mcp" in str(item.get("route_key") or "")
)
if structured_context:
text = (
f"{text} 구조화 조회에서 확인된 식별자와 파일명은 다음과 같다: "
f"{structured_context[:1600]}"
)
return _append_query_guidance(text, _vector_query_guidance(text))
return text
def _cross_source_vector_queries(
question: str,
observations: list[Mapping[str, Any]],
) -> list[str]:
text = str(question or "").strip()
folded = text.casefold()
structured_context = " ".join(
str(item.get("result_excerpt") or "")
for item in observations
if "kb_mcp" in str(item.get("route_key") or "")
)[:1800]
context_suffix = (
f" 구조화 조회 식별자: {structured_context}" if structured_context else ""
)
queries: list[str]
if "c1001025" in folded:
queries = [
"당사 상품코드 25213_B 현행 건강보험 약관 주요 보장 조항",
"DB손보 원본 파일 약관_31084(03)_20260101 건강보험 약관 주요 보장 조항",
]
elif "대물배상" in folded and "삼성화재" in folded:
queries = [
"당사 상품코드 41048 현행 자동차보험 대물배상 약관 조항",
"삼성화재 원본 파일 20071_0_20260611_file1 개인용애니카다이렉트자동차보험 대물배상 약관 조항",
]
elif "삼성화재" in folded and any(
term in folded for term in ("비교", "차별", "갱신")
):
queries = [
"당사 상품코드 41048 현행 KB개인용자동차보험 관련 보장 약관 조항",
"삼성화재 개인용애니카다이렉트자동차보험 현행 원본 약관 관련 보장 조항",
]
else:
queries = [text]
return [
_append_query_guidance(
query + context_suffix,
_vector_query_guidance(text),
)
for query in queries
]
def plan_mcp_execution_mode(
*,
question: str,
routed_tools: list[RoutedMcpTool],
model_profile_key: str,
mode_override: str,
) -> Mapping[str, Any]:
tool_catalog, routes_by_key = _agent_tool_catalog(routed_tools)
route_keys = list(routes_by_key)
fallback = _default_single_route(routed_tools)
fallback_key = _route_key(fallback.server_id, fallback.tool.name)
override = str(mode_override or "auto").strip().lower()
if override in {"single", "agent"}:
return {
"mode": override,
"route_key": fallback_key,
"reason": f"사용자 선택: {override}",
"model_profile": "",
}
if len(routed_tools) <= 1:
return {
"mode": "single",
"route_key": fallback_key,
"reason": "사용 가능한 MCP route가 1개라 단일 호출로 처리합니다.",
"model_profile": "",
}
if _question_needs_cross_source(question):
structured = next(
(route for route in routed_tools if _is_structured_route(route)),
fallback,
)
return {
"mode": "agent",
"route_key": _route_key(structured.server_id, structured.tool.name),
"reason": (
"계약·상품 데이터와 약관 근거를 함께 요구하여 구조화 MCP부터 "
"벡터 MCP까지 순차 호출합니다."
),
"model_profile": model_profile_key,
}
if _question_prefers_structured_single(question):
structured = next(
(route for route in routed_tools if _is_structured_route(route)),
fallback,
)
return {
"mode": "single",
"route_key": _route_key(structured.server_id, structured.tool.name),
"reason": "업무 테이블의 실제 컬럼과 결과 행만으로 검증 가능한 질의입니다.",
"model_profile": model_profile_key,
}
try:
profile = resolve_model_profile(model_profile_key)
client = build_oci_genai_completion_client(
profile.model_id,
profile.answer_model_region,
profile.answer_model_endpoint,
)
text = client.complete(
system_prompt=(
"You are a lightweight MCP execution planner. Decide whether "
"the user's Korean question should be answered with one MCP "
"tool call or with a multi-tool ReAct loop. Prefer mode=single "
"unless the question explicitly requires comparing, combining, "
"or validating evidence across different MCP tools/sources. "
"Return only JSON matching the schema. Never request or expose "
"bearer tokens."
),
user_prompt=json.dumps(
{
"question": question,
"routes": tool_catalog,
"single_policy": (
"counts, lists, summaries, and direct DB lookups should "
"use single unless another source is clearly required"
),
"agent_policy": (
"use agent only for cross-source comparison, contract "
"plus terms/document search, or multi-step validation"
),
},
ensure_ascii=False,
),
response_schema={
"type": "object",
"additionalProperties": False,
"required": ["mode", "route_key", "reason"],
"properties": {
"mode": {"type": "string", "enum": ["single", "agent"]},
"route_key": {"type": "string", "enum": route_keys},
"reason": {"type": "string"},
},
},
max_tokens=300,
temperature=temperature_for_model_profile(profile),
)
parsed = json.loads(text)
except Exception:
return {
"mode": "single",
"route_key": fallback_key,
"reason": "실행 방식 판단 실패로 단일 MCP 호출로 폴백했습니다.",
"model_profile": model_profile_key,
}
if not isinstance(parsed, Mapping):
return {
"mode": "single",
"route_key": fallback_key,
"reason": "실행 방식 판단 응답 형식 오류로 단일 MCP 호출로 폴백했습니다.",
"model_profile": model_profile_key,
}
mode = str(parsed.get("mode") or "single").strip().lower()
route_key = str(parsed.get("route_key") or fallback_key).strip()
if mode not in {"single", "agent"}:
mode = "single"
if route_key not in routes_by_key:
route_key = fallback_key
return {
"mode": mode,
"route_key": route_key,
"reason": str(parsed.get("reason") or "").strip(),
"model_profile": model_profile_key,
}
def _plan_agent_step(
*,
question: str,
conversation: list[Mapping[str, str]],
tool_catalog: list[Mapping[str, Any]],
observations: list[Mapping[str, Any]],
model_profile_key: str,
) -> Mapping[str, Any]:
route_keys = [str(item["route_key"]) for item in tool_catalog]
try:
profile = resolve_model_profile(model_profile_key)
client = build_oci_genai_completion_client(
profile.model_id,
profile.answer_model_region,
profile.answer_model_endpoint,
)
text = client.complete(
system_prompt=(
"You are a concise ReAct-style MCP tool planner. "
"Use the available MCP tools to answer the Korean business question. "
"If no tool has been called yet, choose action=call_tool. "
"After each observation, decide whether another tool call is needed "
"or action=final_answer is enough. Do not expose or request bearer tokens. "
"tool_query must be plain natural-language query text only; do not include "
"argument labels such as limit:, prompt:, query:, top_k:, or candidate_k:. "
"Do not write SQL. Preserve identifiers exactly. If the user says "
"계약번호, write it as 계약번호(CONTRACT_NO); if the user says 상품코드 "
"or product code, write it as 상품코드(PRODUCT_CD). Do not convert one "
"identifier type into the other. 고객번호, 고객ID, 고객 식별번호는 "
"반드시 고객번호(CUST_ID)로 작성한다. For a cross-source question, "
"call the structured kb_mcp route first to identify CUST_ID, CONTRACT_NO, "
"PRODUCT_CD, insurer, clause name, and source file. Then call the vector "
"route using those exact identifiers. Do not finish before both routes "
"have been attempted. "
"Return only JSON matching the schema."
),
user_prompt=json.dumps(
{
"question": question,
"conversation": conversation,
"available_routes": tool_catalog,
"observations": observations,
"max_tool_steps": MAX_AGENT_TOOL_STEPS,
},
ensure_ascii=False,
),
response_schema={
"type": "object",
"additionalProperties": False,
"required": ["thought", "action", "route_key", "tool_query"],
"properties": {
"thought": {"type": "string"},
"action": {"type": "string", "enum": ["call_tool", "final_answer"]},
"route_key": {
"type": "string",
"enum": [*route_keys, AGENT_FINAL_ROUTE],
},
"tool_query": {"type": "string"},
},
},
max_tokens=700,
temperature=temperature_for_model_profile(profile),
)
parsed = json.loads(text)
except Exception:
raise McpToolRouterError("MCP agent planner 호출에 실패했습니다.") from None
if not isinstance(parsed, Mapping):
raise McpToolRouterError("MCP agent planner 응답 형식이 올바르지 않습니다.")
return parsed
def run_mcp_agent_loop(
*,
question: str,
conversation: list[Mapping[str, str]],
routed_tools: list[RoutedMcpTool],
servers: list[McpServer],
bearer_token: str,
limit: int,
model_profile_key: str,
progress_callback: Any = None,
) -> dict[str, Any]:
tool_catalog, routes_by_key = _agent_tool_catalog(routed_tools)
servers_by_id = {server.server_id: server for server in servers}
observations: list[dict[str, Any]] = []
steps: list[dict[str, Any]] = []
last: dict[str, Any] | None = None
attempted_route_keys: set[str] = set()
actionable_route_keys: set[str] = set()
completed_vector_queries: set[str] = set()
stop_reason = ""
for step_no in range(1, MAX_AGENT_TOOL_STEPS + 1):
forced_plan = None
if step_no == 1 and _question_needs_cross_source(question):
forced_plan = _select_unvisited_route(
question,
routes_by_key,
attempted_route_keys,
)
if (
forced_plan is None
and _question_needs_cross_source(question)
and any(
_is_structured_route(routes_by_key[key])
for key in attempted_route_keys
if key in routes_by_key
)
):
vector_route = next(
(
(key, route)
for key, route in routes_by_key.items()
if _is_vector_route(route)
),
None,
)
pending_queries = [
query
for query in _cross_source_vector_queries(question, observations)
if query not in completed_vector_queries
]
if vector_route is not None and pending_queries:
forced_key, forced_route = vector_route
forced_plan = (forced_key, forced_route)
if forced_plan is not None:
forced_key, forced_route = forced_plan
pending_vector_queries = [
query
for query in _cross_source_vector_queries(question, observations)
if query not in completed_vector_queries
]
forced_query = (
pending_vector_queries[0]
if _is_vector_route(forced_route) and pending_vector_queries
else _fallback_tool_query_for_route(
question,
forced_route,
observations,
)
)
plan = {
"thought": (
"비교 대상별 약관 근거를 독립 검색"
if _is_vector_route(forced_route)
else "교차 조회 식별자를 확보하기 위한 구조화 MCP 우선 호출"
),
"action": "call_tool",
"route_key": forced_key,
"tool_query": forced_query,
}
else:
try:
plan = _plan_agent_step(
question=question,
conversation=conversation,
tool_catalog=tool_catalog,
observations=observations,
model_profile_key=model_profile_key,
)
except McpToolRouterError:
fallback_route = _select_unvisited_route(
question,
routes_by_key,
attempted_route_keys,
)
if fallback_route is None or not _question_needs_cross_source(question):
raise
fallback_key, route = fallback_route
plan = {
"thought": "Agent planner 장애로 미호출 MCP route를 순차 실행",
"action": "call_tool",
"route_key": fallback_key,
"tool_query": _fallback_tool_query_for_route(
question,
route,
observations,
),
}
action = str(plan.get("action") or "").strip()
route_key = str(plan.get("route_key") or "").strip()
thought = str(plan.get("thought") or "").strip()
tool_query = _clean_agent_tool_query(plan.get("tool_query"), question)
if action == "final_answer" and observations:
forced = _select_unvisited_route(
question,
routes_by_key,
attempted_route_keys,
)
if forced is None or not _question_needs_cross_source(question):
stop_reason = "planner가 추가 MCP 호출이 불필요하다고 판단했습니다."
break
route_key, route = forced
action = "call_tool"
tool_query = _fallback_tool_query_for_route(
question,
route,
observations,
)
thought = (
f"{thought} / 비교·약관 질문인데 미호출 MCP route가 있어 "
f"{route_key} 호출로 전환합니다."
).strip(" /")
if action != "call_tool" and not observations:
action = "call_tool"
route = routes_by_key.get(route_key)
if route is None:
route = next(iter(routes_by_key.values()))
route_key = _route_key(route.server_id, route.tool.name)
tool_query = append_query_contract_guidance(
tool_query,
original_question=question,
tool_name=route.tool.name,
)
is_distinct_vector_query = (
_is_vector_route(route)
and tool_query not in completed_vector_queries
)
if (
route_key in attempted_route_keys
and last is not None
and not is_distinct_vector_query
):
forced = _select_unvisited_route(
question,
routes_by_key,
attempted_route_keys,
)
if forced is None:
stop_reason = f"중복 MCP route 재호출 차단: {route_key}"
break
repeated_route_key = route_key
route_key, route = forced
tool_query = _fallback_tool_query_for_route(
question,
route,
observations,
)
tool_query = append_query_contract_guidance(
tool_query,
original_question=question,
tool_name=route.tool.name,
)
thought = (
f"{thought} / 중복 MCP route {repeated_route_key} 대신 "
f"미호출 route {route_key}로 전환합니다."
).strip(" /")
server = servers_by_id.get(route.server_id)
if server is None:
raise PublicMcpError("선택된 MCP 서버 설정을 찾지 못했습니다.")
started = perf_counter()
arguments = build_mcp_tool_arguments(
route.tool,
tool_query,
int(limit),
preferred_tool=server.default_tool,
)
raw_result = call_tool(
base_url=server.endpoint_url,
bearer_token=bearer_token,
tool=route.tool,
arguments=arguments,
)
parsed = _content_text_json(raw_result)
mcp_result = parsed if parsed is not None else raw_result
elapsed = perf_counter() - started
summary = _mcp_summary(mcp_result)
step = {
"step": step_no,
"thought": thought,
"action": "call_tool",
"server_id": server.server_id,
"tool_name": route.tool.name,
"route_key": route_key,
"tool_query": tool_query,
"arguments": arguments,
"mcp_result": mcp_result,
"result_summary": summary,
"elapsed_seconds": round(elapsed, 3),
}
steps.append(step)
observations.append(
{
"step": step_no,
"route_key": route_key,
"tool_query": tool_query,
"result_summary": summary,
"result_excerpt": _bounded_json(mcp_result, max_chars=6000),
}
)
last = {
"server": server,
"tool": route.tool,
"arguments": arguments,
"mcp_result": mcp_result,
}
attempted_route_keys.add(route_key)
if _is_vector_route(route):
completed_vector_queries.add(tool_query)
if _mcp_has_actionable_result(mcp_result):
actionable_route_keys.add(route_key)
if progress_callback:
progress_callback(step)
if last is None:
raise PublicMcpError("MCP agent가 실행한 tool 호출이 없습니다.")
return {
"server": last["server"],
"tool": last["tool"],
"arguments": last["arguments"],
"mcp_result": last["mcp_result"],
"agent_steps": steps,
"stop_reason": stop_reason,
"attempted_route_keys": sorted(attempted_route_keys),
"actionable_route_keys": sorted(actionable_route_keys),
}
def _render_mcp_result_sections(
details: Mapping[str, Any],
message_key: str,
) -> None:
mcp_result = details.get("mcp_result", {})
agent_steps = details.get("agent_steps")
execution_events = details.get("execution_events")
generated_sql = _mcp_generated_sql(mcp_result)
items = _mcp_items(mcp_result)
if isinstance(execution_events, list) and execution_events:
with st.expander(f"처리 시간 로그 · {len(execution_events)}건"):
rows: list[dict[str, Any]] = []
for event in execution_events:
if not isinstance(event, Mapping):
continue
rows.append(
{
"진행률": event.get("percent"),
"단계": str(event.get("label") or ""),
"단계소요(s)": event.get("elapsed_seconds"),
"누적소요(s)": event.get("total_elapsed_seconds"),
"상세": str(event.get("detail") or ""),
}
)
if rows:
st.dataframe(rows, hide_index=True, width="stretch")
if isinstance(agent_steps, list) and agent_steps:
with st.expander(f"Agent 실행 단계 · {len(agent_steps)}회"):
stop_reason = str(details.get("agent_stop_reason") or "").strip()
if stop_reason:
st.caption(f"종료 사유: {stop_reason}")
for step in agent_steps:
if not isinstance(step, Mapping):
continue
st.markdown(
f"**Step {step.get('step')} · "
f"{step.get('server_id')} / {step.get('tool_name')}**"
)
thought = str(step.get("thought") or "").strip()
tool_query = str(step.get("tool_query") or "").strip()
if thought:
st.caption(f"판단: {thought}")
if tool_query:
st.caption(f"툴 질의: {tool_query}")
st.markdown("전달 arguments")
st.json(step.get("arguments", {}))
st.markdown("응답 요약")
st.json(step.get("result_summary", {}))
if generated_sql:
with st.expander("생성 SQL"):
st.code(generated_sql, language="sql")
if items:
with st.expander(f"조회 결과 테이블 · {len(items)}건"):
display_items = items[:100]
st.dataframe(display_items, use_container_width=True)
if len(items) > len(display_items):
st.caption(f"화면에는 최초 {len(display_items)}건만 표시합니다.")
with st.expander("MCP 호출 상세"):
st.write(f"route: {details.get('server_id')} / {details.get('tool_name')}")
execution_mode = str(details.get("execution_mode") or "").strip()
execution_reason = str(details.get("execution_mode_reason") or "").strip()
if execution_mode:
st.caption(
"실행 방식: "
+ execution_mode
+ (f" · {execution_reason}" if execution_reason else "")
)
reasoning_model = str(details.get("reasoning_model_profile") or "").strip()
answer_model = str(details.get("answer_model_profile") or "").strip()
fallback_model = str(
details.get("answer_synthesis_fallback_model") or ""
).strip()
planner_model = str(details.get("execution_mode_model_profile") or "").strip()
if planner_model or reasoning_model or answer_model:
st.caption(
"모델: "
+ f"planner={planner_model or '-'} · "
+ f"reasoning={reasoning_model or '-'} · "
+ f"answer={answer_model or '-'}"
+ (f" · fallback={fallback_model}" if fallback_model else "")
)
if details.get("standalone_question"):
st.caption(f"MCP 질의: {details['standalone_question']}")
if details.get("tool_query"):
st.caption(f"툴별 정제 질의: {details['tool_query']}")
st.markdown("전달 arguments")
st.json(details.get("arguments", {}))
st.markdown("MCP 응답 요약")
st.json(_mcp_summary(mcp_result))
synthesis_error = details.get("answer_synthesis_error")
if isinstance(synthesis_error, Mapping) and synthesis_error:
st.markdown("최종 답변 합성 실패 진단")
st.json(dict(synthesis_error))
raw_key = f"poc4_show_raw_mcp_{message_key}"
if st.button("MCP 원본 JSON 보기/숨기기", key=f"{raw_key}_button"):
st.session_state[raw_key] = not bool(st.session_state.get(raw_key))
if st.session_state.get(raw_key):
st.json(mcp_result)
routes = details.get("discovered_routes")
failures = details.get("discovery_failures")
if routes or failures:
st.markdown("Discovered routes")
st.json({"routes": routes or [], "failures": failures or []})
def _conversation_context(messages: list[Mapping[str, Any]]) -> list[dict[str, str]]:
context: list[dict[str, str]] = []
for message in messages[-MAX_CONVERSATION_MESSAGES:]:
role = str(message.get("role") or "").strip()
content = str(message.get("content") or "").strip()
if role in {"user", "assistant"} and content:
context.append({"role": role, "content": content[:2000]})
return context
def resolve_standalone_question(
*,
question: str,
messages: list[Mapping[str, Any]],
model_profile_key: str,
) -> str:
context = _conversation_context(messages)
if not context:
return question
try:
profile = resolve_model_profile(model_profile_key)
client = build_oci_genai_completion_client(
profile.model_id,
profile.answer_model_region,
profile.answer_model_endpoint,
)
text = client.complete(
system_prompt=(
"Rewrite the user's latest Korean question into one standalone "
"MCP tool query. Use the conversation only to resolve references. "
"Do not answer the question."
),
user_prompt=json.dumps(
{"conversation": context, "latest_question": question},
ensure_ascii=False,
),
response_schema={
"type": "object",
"additionalProperties": False,
"required": ["standalone_question"],
"properties": {"standalone_question": {"type": "string"}},
},
max_tokens=500,
temperature=temperature_for_model_profile(profile),
)
parsed = json.loads(text)
rewritten = str(parsed.get("standalone_question") or "").strip()
except Exception:
return question
return rewritten or question
def prepare_tool_query_for_mcp(
*,
question: str,
server: McpServer,
tool: McpTool,
model_profile_key: str,
selected_user_id: str = "",
selected_user_role: str = "",
selected_user_channel: str = "",
selected_user_scope: str = "",
) -> str:
"""Rewrite the user question into a selected-tool-specific natural query."""
fallback = str(question or "").strip()
if not fallback:
return fallback
if server.server_id == "hmm_hr_mcp":
return _prepare_hmm_hr_tool_query(
question=fallback,
tool=tool,
model_profile_key=model_profile_key,
selected_user_id=selected_user_id,
selected_user_role=selected_user_role,
selected_user_team=selected_user_channel,
selected_user_scope=selected_user_scope,
)
normalized_question = " ".join(fallback.casefold().split())
individual_scope_terms = (
"내 담당이 아닌",
"다른 설계사",
"다른 채널",
"타 설계사",
"타 채널",
)
is_select_ai_tool = (
server.server_id == "kb_mcp"
and "select_ai" in tool.name.casefold()
)
query_guidance = (
_select_ai_query_guidance(fallback)
if is_select_ai_tool
else _vector_query_guidance(fallback)
if "vector" in server.server_id.casefold()
else []
)
is_channel_scope_validation = (
bool(selected_user_id.strip())
and is_select_ai_tool
and (
"공통계정" in normalized_question
or "공통 계정" in normalized_question
)
and "채널" in normalized_question
and any(term in normalized_question for term in ("조회", "접근", "권한", "vpd"))
)
if is_channel_scope_validation:
user_id = selected_user_id.strip()
expected_channel = selected_user_channel.strip() or "선택 채널"
return (
f"POC_2.KB_CONTRACTS에서 현재 VPD 공통계정 사용자 {user_id}에게 실제로 "
f"조회되는 계약을 검증해줘. 이 사용자의 권한 채널은 {expected_channel}이다. "
"조회 결과 전체를 한 번 집계해서 FC_CHANNEL='다이렉트' 계약 건수, "
"FC_CHANNEL='설계사' 계약 건수, FC_CHANNEL='GA' 계약 건수, "
"FC_CHANNEL='제휴' 계약 건수를 각각 별도 컬럼으로 반드시 보여줘. "
"계약번호는 CONTRACT_NO, 고객번호는 CUST_ID, 담당자는 FC_ID, "
"담당 채널은 FC_CHANNEL 실제 컬럼만 사용해줘. "
"VPD 정책, 시스템 카탈로그, 사용자 또는 권한 메타데이터는 조회하지 말고 "
"현재 토큰으로 보이는 KB_CONTRACTS 업무 데이터만 사용해줘."
)
is_individual_scope_validation = (
bool(selected_user_id.strip())
and is_select_ai_tool
and any(term in normalized_question for term in individual_scope_terms)
and any(term in normalized_question for term in ("조회", "접근", "권한", "vpd"))
)
if is_individual_scope_validation:
user_id = selected_user_id.strip()
return (
f"POC_2.KB_CONTRACTS에서 현재 VPD 사용자 {user_id}에게 실제로 조회되는 "
"계약을 FC_CHANNEL별로 집계해줘. 전체 계약 건수, "
f"FC_ID='{user_id}'인 본인 담당 계약 건수, "
f"FC_ID<>'{user_id}'인 다른 설계사 계약 건수를 각각 보여줘. "
"계약번호는 CONTRACT_NO, 고객번호는 CUST_ID, 담당자는 FC_ID, "
"담당 채널은 FC_CHANNEL 실제 컬럼만 사용해줘. "
"VPD 정책, 시스템 카탈로그, 사용자 또는 권한 메타데이터는 조회하지 말고 "
"현재 토큰으로 보이는 KB_CONTRACTS 업무 데이터만 사용해줘."
)
properties = tool.schema.get("properties")
if not isinstance(properties, Mapping):
properties = {}
safe_properties = {
str(name): {
"type": value.get("type") if isinstance(value, Mapping) else None,
"description": (
str(value.get("description") or "")[:500]
if isinstance(value, Mapping)
else ""
),
}
for name, value in properties.items()
if str(name).lower()
not in {"bearertoken", "bearer_token", "token", "authorization"}
}
try:
profile = resolve_model_profile(model_profile_key)
client = build_oci_genai_completion_client(
profile.model_id,
profile.answer_model_region,
profile.answer_model_endpoint,
)
text = client.complete(
system_prompt=(
"You prepare one selected-tool-specific MCP query. "
"Return only JSON matching the schema. Do not answer the user. "
"Do not write SQL. Do not expose or request bearer tokens. "
"The output tool_query must be plain Korean natural-language text "
"for the selected MCP tool, not a JSON string and not argument labels. "
"Never include labels such as prompt:, query:, question:, limit:, "
"top_k:, candidate_k:, or max_rows:. "
"Use the tool description and input properties to make the query "
"fit that tool. For database Select AI tools, make filters, columns, "
"and identifier meanings explicit. For document/vector search tools, "
"produce compact search keywords and document terms. Preserve all IDs "
"and numbers exactly. If the user says 계약번호, write 계약번호(CONTRACT_NO); "
"if the user says 상품코드 or product code, write 상품코드(PRODUCT_CD). "
"고객번호, 고객ID, 고객 식별번호는 반드시 고객번호(CUST_ID)로 작성한다. "
"담당 채널은 담당 채널(FC_CHANNEL), 설계사는 설계사(FC_ID), "
"계약상태는 계약상태(CONTRACT_STATUS)로 작성한다. "
"VPD 적용 여부, 권한 범위, 접근 가능 여부를 묻는 질의는 VPD 정책, "
"시스템 카탈로그, 사용자 또는 권한 메타데이터를 조회하지 않는다. "
"현재 선택 사용자의 VPD 컨텍스트로 보이는 POC_2.KB_CONTRACTS의 "
"업무 데이터 결과만 사용해 실효 접근 범위를 검증한다. "
"DB Select AI 질의에는 제공된 KB 테이블의 실제 컬럼만 사용하고, "
"확인되지 않은 영문 컬럼명을 추측하여 생성하지 않는다. "
"보험금 지급 판단은 KB_CONTRACTS.CONTRACT_STATUS와 "
"KB_CLAIMS.CLAIM_AMT, PAID_AMT, CLAIM_STATUS를 포함한다. "
"담보 질문은 KB_COVERAGES.COVERAGE_NM, COVERAGE_TYPE, "
"INSURED_AMT, RENEW_DUE_DT를 포함한다. 보험료 질문에서 개별값과 "
"채널 합계는 별도 범위로 계산한다. 주민번호 확인 질문은 "
"KB_CUSTOMERS.RRN_MASKED 실제 결과를 조회하며 WHERE 1=0을 만들지 않는다. "
"Do not convert one identifier type into the other."
),
user_prompt=json.dumps(
{
"question": fallback,
"selected_server_id": server.server_id,
"selected_tool_name": tool.name,
"selected_tool_description": tool.description[:1200],
"input_properties": safe_properties,
"default_tool": server.default_tool,
"selected_vpd_user": {
"user_id": selected_user_id.strip(),
"role": selected_user_role.strip(),
"channel": selected_user_channel.strip(),
"scope": selected_user_scope.strip(),
},
"required_query_guidance": query_guidance,
},
ensure_ascii=False,
),
response_schema={
"type": "object",
"additionalProperties": False,
"required": ["tool_query"],
"properties": {"tool_query": {"type": "string"}},
},
max_tokens=600,
temperature=temperature_for_model_profile(profile),
)
parsed = json.loads(text)
rewritten = _clean_agent_tool_query(parsed.get("tool_query"), fallback)
except Exception:
return _append_query_guidance(fallback, query_guidance)
return _append_query_guidance(rewritten or fallback, query_guidance)
def _hmm_demo_user_context(
question: str,
*,
user_id: str,
role: str,
team: str,
scope: str,
) -> str:
"""Make a selected HR persona useful without claiming row-level enforcement."""
normalized = str(question or "").strip()
if not user_id:
return normalized
profile = " · ".join(item for item in (user_id, role, team) if item)
purpose = f" 테스트 목적: {scope}." if scope else ""
return (
f"현재 HMM HR 데모 사용자: {profile}.{purpose} "
"질문의 ‘나’, ‘내’, ‘우리 팀’은 이 데모 사용자를 기준으로 해석하고, "
"실제 행 수준 권한이 적용됐다고 주장하지 마세요.\n"
f"질문: {normalized}"
)
def _prepare_hmm_hr_tool_query(
*,
question: str,
tool: McpTool,
model_profile_key: str,
selected_user_id: str,
selected_user_role: str,
selected_user_team: str,
selected_user_scope: str,
) -> str:
"""Prepare an HMM HR query without inheriting retired KB/VPD prompt rules."""
fallback = str(question or "").strip()
def with_contract(value: str) -> str:
return append_query_contract_guidance(
value,
original_question=fallback,
tool_name=tool.name,
)
if tool.name in {"resolve_hr_term", "search_hr_policy"}:
return with_contract(fallback)
contextual_question = _hmm_demo_user_context(
fallback,
user_id=selected_user_id.strip(),
role=selected_user_role.strip(),
team=selected_user_team.strip(),
scope=selected_user_scope.strip(),
)
if tool.name == "search_carrier_performance":
return with_contract(contextual_question)
try:
profile = resolve_model_profile(model_profile_key)
client = build_oci_genai_completion_client(
profile.model_id,
profile.answer_model_region,
profile.answer_model_endpoint,
)
text = client.complete(
system_prompt=(
"You prepare one Korean natural-language query for an HMM HR MCP tool. "
"Return only JSON matching the schema. Do not answer the user, write SQL, "
"or expose/request tokens. Preserve employee codes exactly. "
"The HMM HR data tool can query organization, employees, leave balances, "
"leave requests, attendance, and standardized HR terms. The policy tool searches "
"HR policy PDF abstracts and chunks. The carrier federation tool joins ADB "
"employee assignments with PostgreSQL carrier KPI data by CARRIER_CODE. "
"A selected demo user only resolves pronouns "
"such as 'my' or 'our team'; do not claim that it enforces database access control."
),
user_prompt=json.dumps(
{
"question": contextual_question,
"selected_tool_name": tool.name,
"selected_tool_description": tool.description[:1200],
},
ensure_ascii=False,
),
response_schema={
"type": "object",
"additionalProperties": False,
"required": ["tool_query"],
"properties": {"tool_query": {"type": "string"}},
},
max_tokens=500,
temperature=temperature_for_model_profile(profile),
)
parsed = json.loads(text)
rewritten = _clean_agent_tool_query(parsed.get("tool_query"), fallback)
except Exception:
return with_contract(contextual_question)
return with_contract(
_hmm_demo_user_context(
rewritten or fallback,
user_id=selected_user_id.strip(),
role=selected_user_role.strip(),
team=selected_user_team.strip(),
scope=selected_user_scope.strip(),
)
)
def synthesize_answer(
*,
question: str,
conversation: list[Mapping[str, str]],
server: McpServer,
tool: McpTool,
arguments: Mapping[str, Any],
mcp_result: Any,
model_profile_key: str,
agent_steps: list[Mapping[str, Any]] | None = None,
security_evidence: Mapping[str, Any] | None = None,
business_evidence: Mapping[str, Any] | None = None,
) -> Mapping[str, Any]:
contract_tools = {tool.name}
contract_tools.update(
str(step.get("tool_name") or "")
for step in (agent_steps or [])
if isinstance(step, Mapping)
)
active_contracts_by_id: dict[str, Mapping[str, Any]] = {}
for contract_tool in contract_tools:
for contract in matching_query_contracts(question, contract_tool):
contract_id = str(contract.get("id") or "")
active_contracts_by_id[contract_id] = contract
active_contracts = tuple(active_contracts_by_id.values())
contract_evidence = {
"mcp_result": mcp_result,
"agent_steps": list(agent_steps or []),
}
contract_report = evidence_contract_report(active_contracts, contract_evidence)
if contract_report and any(
not bool(report.get("satisfied")) for report in contract_report
):
missing_fields = sorted(
{
str(field)
for report in contract_report
for field in report.get("missing_fields", [])
}
)
failed_calculations = sorted(
{
str(check.get("field") or "")
for report in contract_report
for check in report.get("computed_field_checks", [])
if not bool(check.get("satisfied"))
}
)
failed_temporal_checks = sorted(
{
str(check.get("check") or "")
for report in contract_report
for check in report.get("temporal_contract_checks", [])
if not bool(check.get("satisfied"))
}
)
evidence_issues = []
if missing_fields:
evidence_issues.append("필수 필드 누락: " + ", ".join(missing_fields))
if failed_calculations:
evidence_issues.append(
"계산 결과 불일치: " + ", ".join(failed_calculations)
)
if failed_temporal_checks:
evidence_issues.append(
"시간 기준 계약 불일치: " + ", ".join(failed_temporal_checks)
)
return {
"answer": missing_evidence_message(active_contracts),
"basis": evidence_issues or ["질의 계약 검증 실패"],
"limitations": (
"필수 원장 근거가 충족되지 않아 잔여일수나 승인 가능 여부를 "
"추정하지 않았습니다."
),
"query_contracts": list(active_contracts),
"evidence_contract_report": contract_report,
}
def build_user_prompt(*, compact: bool) -> str:
evidence_items = 10 if compact else 20
evidence_text = 260 if compact else 380
step_items = 4 if compact else 6
step_text = 300 if compact else 480
prompt_payload = {
"question": question,
"required_answer_checks": _answer_requirements(question),
"query_contracts": list(active_contracts),
"evidence_contract_report": contract_report,
"conversation": [] if compact else conversation[-4:],
"selected_route": {
"server_id": server.server_id,
"tool_name": tool.name,
},
"tool_arguments": dict(arguments),
"security_evidence": dict(security_evidence or {}),
"business_evidence": _bounded_json(
dict(business_evidence or {}),
max_chars=7000 if compact else 11000,
),
"mcp_result": _bounded_json(
_mcp_answer_evidence(
mcp_result,
max_items=evidence_items,
max_text=evidence_text,
include_sql=False,
),
max_chars=6000 if compact else 8000,
),
"agent_steps": [
{
"step": step.get("step"),
"server_id": step.get("server_id"),
"tool_name": step.get("tool_name"),
"tool_query": step.get("tool_query"),
"arguments": step.get("arguments", {}),
"mcp_result_evidence": _bounded_json(
_mcp_answer_evidence(
step.get("mcp_result"),
max_items=step_items,
max_text=step_text,
include_sql=False,
),
max_chars=3600 if compact else 5200,
),
}
for step in (agent_steps or [])
],
}
return json.dumps(prompt_payload, ensure_ascii=False)
user_prompt = build_user_prompt(compact=False)
response_schema = {
"type": "object",
"additionalProperties": False,
"required": ["answer", "basis", "limitations"],
"properties": {
"answer": {"type": "string"},
"basis": {
"type": "array",
"items": {"type": "string"},
"maxItems": 5,
},
"limitations": {"type": "string"},
},
}
diagnostics: dict[str, Any] = {
"model_profile": str(model_profile_key),
"payload_chars": len(user_prompt),
"agent_steps": len(agent_steps or []),
"security_evidence": bool(security_evidence),
"business_evidence": bool(business_evidence),
}
try:
profile = resolve_model_profile(model_profile_key)
client = build_oci_genai_completion_client(
profile.model_id,
profile.answer_model_region,
profile.answer_model_endpoint,
)
diagnostics.update(
{
"model_key": profile.model_key,
"model_id": profile.model_id,
"region": profile.answer_model_region,
}
)
except Exception as exc:
diagnostics.update(
{
"stage": "client_init",
"error_type": type(exc).__name__,
"error": str(exc)[:500],
}
)
LOG.warning("poc4_answer_synthesis_client_init_failed %s", diagnostics)
raise AnswerSynthesisError(
"MCP 결과 기반 최종 답변 생성에 실패했습니다.",
diagnostics,
) from None
system_prompt = (
"You write final Korean answers from MCP tool results. "
"Use English for internal instructions, but the final answer "
"field must be written only in Korean. "
"Use only the provided MCP results as evidence. Do not invent "
"facts. If the MCP result is an error or permission denial, "
"say that clearly. Treat items_count=0 or results_count=0 as an actual "
"zero-row result, not as missing tool delivery. Never claim that an audit "
"event exists unless its identifier or audit result is in the evidence. "
"For compatibility MCP tools, the result string contains the factual "
"DOC and EVIDENCE lines; read it as evidence rather than treating it as metadata. "
"For cross-source questions, keep structured contract/product facts separate "
"from vector clause evidence, then combine only matching identifiers. If one "
"source is missing, list confirmed and unconfirmed points separately instead "
"of saying only that comparison is impossible. Do not confuse row-level "
"masked values with aggregate values. Preserve product_cd, contract_no, "
"document_id, chunk_id, source filename, version, and clause locator when "
"they are provided. Security evidence is a predefined token-scoped database "
"verification result and takes precedence over ambiguous redacted MCP values. "
"Business evidence is a Bearer-token-validated result with explicit CUST_ID and "
"FC_ID scoping for customer data, plus active catalog document/chunk evidence. "
"It takes precedence when Select AI returns an ambiguous or incorrect zero-row "
"result. Never expose customer rows outside the business evidence scope. "
"The query_contracts and evidence_contract_report are authoritative. Follow "
"their required fields, calculations, temporal semantics, missing-record "
"semantics, forbidden fallbacks, and answer restrictions. Keep data facts "
"and policy requirements separate. "
"Show audit_log_id when it is provided. Follow every required_answer_checks item in the user "
"payload. When the user requests a result list, include every "
"row provided in the MCP evidence, up to 20 rows, and state the total "
"returned row count. Do not arbitrarily stop at five rows. Keep the "
"answer concise and business-readable."
)
for attempt in range(1, 3):
compact = attempt > 1
if compact:
user_prompt = build_user_prompt(compact=True)
diagnostics["payload_chars"] = len(user_prompt)
diagnostics["compact_payload"] = True
text = ""
try:
text = client.complete(
system_prompt=system_prompt,
user_prompt=user_prompt,
response_schema=response_schema,
max_tokens=1600,
temperature=temperature_for_model_profile(profile),
)
if not text.strip():
raise ValueError("empty structured response")
parsed = json.loads(text)
if not isinstance(parsed, Mapping) or not isinstance(
parsed.get("answer"), str
):
raise ValueError("invalid structured answer")
enriched = dict(parsed)
enriched["answer"] = _enrich_answer_with_security_evidence(
str(parsed.get("answer") or ""),
security_evidence,
)
enriched["answer"] = _enrich_answer_with_business_evidence(
str(enriched.get("answer") or ""),
business_evidence,
)
basis = list(parsed.get("basis") or [])
if security_evidence and security_evidence.get("audit_log_id"):
basis.append(
"보안 검증 증적: "
f"{security_evidence.get('audit_source')} "
f"LOG_ID={security_evidence.get('audit_log_id')}"
)
if business_evidence and business_evidence.get("evidence_complete"):
basis.append(
"업무 검증 증적: "
+ ", ".join(
str(source)
for source in business_evidence.get("evidence_sources", [])
)
)
evidence_verified = bool(
business_evidence
and business_evidence.get("evidence_complete")
and not business_evidence.get("evidence_error")
) or bool(
security_evidence
and security_evidence.get("evidence_type")
and not security_evidence.get("evidence_error")
)
if evidence_verified:
enriched["limitations"] = ""
enriched["basis"] = basis[:5]
enriched["query_contracts"] = list(active_contracts)
enriched["evidence_contract_report"] = contract_report
return enriched
except Exception as exc:
last_error = {
"stage": "completion",
"attempt": attempt,
"error_type": type(exc).__name__,
"error": str(exc)[:500],
"raw_text_chars": len(text),
"raw_text_head": text[:200],
}
diagnostics.update(last_error)
LOG.warning("poc4_answer_synthesis_attempt_failed %s", diagnostics)
if str(exc) == "empty structured response":
break
raise AnswerSynthesisError(
"MCP 결과 기반 최종 답변 생성에 실패했습니다.",
diagnostics,
) from None
def fallback_answer_from_mcp(
mcp_result: Any,
agent_steps: list[Mapping[str, Any]] | None = None,
) -> str:
items = _mcp_items(mcp_result)
payload = _mcp_response_payload(mcp_result)
results = payload.get("results")
lines = ["MCP 조회는 완료됐지만 최종 답변 합성에 실패했습니다."]
if items:
lines.append(f"조회 결과는 {len(items)}건입니다.")
lines.append("")
lines.append("주요 결과:")
for item in items[:20]:
compact = _compact_evidence_item(item, max_text=300)
lines.append(f"- {_bounded_json(compact, max_chars=500)}")
elif isinstance(results, list) and results:
lines.append(f"검색 결과는 {len(results)}건입니다.")
lines.append("")
lines.append("주요 검색 결과:")
for item in results[:20]:
compact = _compact_evidence_item(item, max_text=320)
if isinstance(compact, Mapping):
title = " / ".join(
str(compact.get(key) or "")
for key in ("insurer", "product_name", "chunk_type")
if compact.get(key)
)
body = str(
compact.get("display_markdown")
or compact.get("text")
or compact.get("content")
or ""
).strip()
page = compact.get("physical_page_range")
suffix = f" (p.{page})" if page else ""
lines.append(f"- {title or '검색 결과'}{suffix}: {body}")
else:
lines.append(f"- {compact}")
elif text_result(mcp_result):
result = text_result(mcp_result)
lines.append("")
lines.append("MCP 반환 근거:")
lines.append(result[:6000] + ("..." if len(result) > 6000 else ""))
elif agent_steps:
lines.append(f"Agent는 MCP tool을 {len(agent_steps)}회 호출했습니다.")
else:
lines.append("상세 영역에서 MCP 원본 응답을 확인해 주세요.")
return "\n".join(lines)
_DISPLAY_MARKDOWN_METADATA_KEYS = (
"insurer",
"product_name",
"product_cd",
"product_code",
"document_title",
"title",
"document_id",
"chunk_id",
"source_file_name",
"file_name",
"ext_file_nm",
"clause_version",
"chunk_type",
"logical_locator",
"physical_page_range",
"page",
"score",
)
def _display_markdown_records(details: Mapping[str, Any] | None) -> list[dict[str, Any]]:
if not isinstance(details, Mapping):
return []
sources: list[Any] = [details.get("mcp_result")]
agent_steps = details.get("agent_steps")
if isinstance(agent_steps, list):
sources.extend(
step.get("mcp_result")
for step in agent_steps
if isinstance(step, Mapping)
)
records: list[dict[str, Any]] = []
seen: set[tuple[str, str, str]] = set()
def collect(value: Any, depth: int = 0) -> None:
if depth > 12 or len(records) >= 50:
return
if isinstance(value, Mapping):
display_markdown = value.get("display_markdown")
if isinstance(display_markdown, str) and display_markdown.strip():
record = {
key: value.get(key)
for key in _DISPLAY_MARKDOWN_METADATA_KEYS
if value.get(key) not in (None, "", [])
}
record["display_markdown"] = display_markdown.strip()
identity = (
str(record.get("document_id") or ""),
str(record.get("chunk_id") or ""),
record["display_markdown"],
)
if identity not in seen:
seen.add(identity)
records.append(record)
for child in value.values():
if isinstance(child, (Mapping, list, tuple)):
collect(child, depth + 1)
elif isinstance(value, (list, tuple)):
for child in value:
collect(child, depth + 1)
for source in sources:
collect(source)
if isinstance(source, Mapping):
collect(_mcp_response_payload(source))
return records
def _display_markdown_match_score(basis: str, record: Mapping[str, Any]) -> int:
normalized_basis = " ".join(basis.casefold().split())
score = 0
for key in _DISPLAY_MARKDOWN_METADATA_KEYS:
value = " ".join(str(record.get(key) or "").casefold().split())
if len(value) >= 3 and value in normalized_basis:
score += min(len(value), 40)
if key in {
"document_id",
"chunk_id",
"source_file_name",
"file_name",
"product_name",
}:
score += 30
return score
def _match_basis_to_display_markdown(
basis: list[Any],
records: list[dict[str, Any]],
) -> list[dict[str, Any] | None]:
remaining = set(range(len(records)))
matched: list[dict[str, Any] | None] = []
for item in basis:
basis_text = str(item or "")
ranked = sorted(
(
(_display_markdown_match_score(basis_text, records[index]), index)
for index in remaining
),
reverse=True,
)
selected_index: int | None = None
if ranked and ranked[0][0] > 0:
selected_index = ranked[0][1]
elif remaining:
selected_index = min(remaining)
if selected_index is None:
matched.append(None)
continue
remaining.remove(selected_index)
matched.append(records[selected_index])
return matched
def _display_markdown_source_label(record: Mapping[str, Any]) -> str:
source = next(
(
str(record.get(key) or "").strip()
for key in (
"source_file_name",
"file_name",
"ext_file_nm",
"product_name",
"document_title",
"document_id",
)
if str(record.get(key) or "").strip()
),
"MCP 참조 문서",
)
details: list[str] = []
page = str(record.get("physical_page_range") or record.get("page") or "").strip()
chunk_id = str(record.get("chunk_id") or "").strip()
clause_version = str(record.get("clause_version") or "").strip()
if page:
details.append(f"페이지 {page}")
if chunk_id:
details.append(f"청크 {chunk_id}")
if clause_version:
details.append(f"약관 버전 {clause_version}")
return " · ".join([source, *details])
def _readable_basis_summary(value: Any) -> str:
text = " ".join(str(value or "").split()).strip()
if not text:
return ""
metadata_match = re.search(
r"\s+(?:문서[_ ]?id|document_id|chunk_id|logical_locator|"
r"physical_page_range)\s*:",
text,
flags=re.IGNORECASE,
)
if metadata_match:
readable = text[: metadata_match.start()].strip(" /·:-")
if readable:
return readable
return text
def _render_assistant_message(
message: Mapping[str, Any],
message_key: str,
) -> None:
st.markdown(str(message.get("content") or ""))
details = message.get("details")
basis = message.get("basis")
if isinstance(basis, list) and basis:
display_records = _display_markdown_records(
details if isinstance(details, Mapping) else None
)
matched_records = _match_basis_to_display_markdown(basis, display_records)
with st.expander(f"답변 근거 요약 · {len(basis)}건"):
for index, (item, record) in enumerate(
zip(basis, matched_records),
start=1,
):
with st.container(border=True):
readable_summary = _readable_basis_summary(item)
source_title = (
_display_markdown_source_label(record).split(" · ", 1)[0]
if isinstance(record, Mapping)
else ""
)
heading = f"근거 {index}"
if source_title:
heading += f" · {source_title}"
st.markdown(f"#### {heading}")
if readable_summary and (
not source_title
or readable_summary.casefold() != source_title.casefold()
):
st.markdown(readable_summary)
if isinstance(record, Mapping):
st.caption(
"참조 문서 · "
+ _display_markdown_source_label(record)
)
st.markdown("**MCP 원문 근거 (`display_markdown`)**")
st.markdown(str(record.get("display_markdown") or ""))
limitations = str(message.get("limitations") or "").strip()
if limitations:
st.caption(f"제약/주의: {limitations}")
if isinstance(details, Mapping):
_render_mcp_result_sections(details, message_key)
def _render_chat_turn(turn: Mapping[str, Any]) -> None:
user_label = str(turn.get("selected_user_id") or "").strip()
created_at = str(turn.get("created_at") or "").strip()
meta = " · ".join(item for item in (created_at, user_label) if item)
with st.chat_message("user"):
if meta:
st.caption(meta)
st.markdown(str(turn.get("question") or ""))
with st.chat_message("assistant"):
_render_assistant_message(
{
"content": str(turn.get("answer") or ""),
"basis": turn.get("basis", []),
"limitations": str(turn.get("limitations") or ""),
"details": turn.get("details", {}),
},
f"turn_{turn.get('turn_id')}",
)
def _render_chat_history(conversation_id: str, page_key: str) -> None:
st.markdown(
'<div class="kb-section-title result" role="heading" aria-level="3">'
"질의 결과"
"</div>",
unsafe_allow_html=True,
)
total_turns = count_chat_turns(conversation_id)
if total_turns <= 0:
st.info("아직 저장된 대화가 없습니다.")
return
max_page = max((total_turns - 1) // CHAT_TURNS_PER_PAGE, 0)
current_page = int(st.session_state.get(page_key, 0) or 0)
current_page = min(max(current_page, 0), max_page)
st.session_state[page_key] = current_page
start_no = current_page * CHAT_TURNS_PER_PAGE + 1
end_no = min(start_no + CHAT_TURNS_PER_PAGE - 1, total_turns)
st.caption(
f"총 {total_turns}건 · 최신순 {start_no}~{end_no}건 표시 · "
f"페이지 {current_page + 1}/{max_page + 1}"
)
nav_cols = st.columns([1, 1, 1, 5])
with nav_cols[0]:
if st.button(
"최신",
key=f"{page_key}_first_{conversation_id}",
disabled=current_page == 0,
):
st.session_state[page_key] = 0
st.rerun()
with nav_cols[1]:
if st.button(
"이전 내역",
key=f"{page_key}_older_{conversation_id}",
disabled=current_page >= max_page,
):
st.session_state[page_key] = current_page + 1
st.rerun()
with nav_cols[2]:
if st.button(
"다음 내역",
key=f"{page_key}_newer_{conversation_id}",
disabled=current_page <= 0,
):
st.session_state[page_key] = current_page - 1
st.rerun()
turns = load_chat_turns(conversation_id, current_page)
if not turns:
st.info("아직 저장된 대화가 없습니다.")
return
for turn in turns:
_render_chat_turn(turn)
def _render_architecture_tab() -> None:
st.subheader("운영 아키텍처")
st.caption("사용자 권한을 기준으로 AI 질의, MCP 도구, Oracle Database를 연결합니다.")
experience, integration, security = st.columns(3)
with experience:
st.markdown("**01 · 사용자와 권한**")
st.write("사용자 역할과 담당 범위를 요청 컨텍스트에 적용합니다.")
with integration:
st.markdown("**02 · AI와 MCP**")
st.write("질의에 맞는 데이터·지식 검색 도구를 선택합니다.")
with security:
st.markdown("**03 · 데이터와 보안**")
st.write("VPD와 감사 정책으로 데이터 접근을 통제합니다.")
st.divider()
st.markdown("#### 질의 처리 흐름")
st.markdown(
"1. 사용자 권한과 질문을 요청에 반영합니다. \n"
"2. AI가 필요한 MCP 도구를 선택합니다. \n"
"3. Oracle Database에서 근거를 조회하고 결과를 표시합니다."
)
def _render_vpd_operations_tab() -> None:
st.markdown(
f"""
<section aria-labelledby="kb-ops-title">
<div class="kb-section-title input" id="kb-ops-title"
role="heading" aria-level="3">
RLS / CLS 설정 ( 오라클 VPD / 마스킹 )
</div>
<div class="kb-ops-lead">
업무 사용자와 데이터 접근 기준을 관리하고,
DB가 적용한 결과까지 한 흐름에서 확인합니다.
</div>
<h3 class="kb-ops-section-heading">보안 설정 업무 프로세스</h3>
<div class="kb-ops-guide-copy">
권한을 먼저 만들고 보호 대상을 연결한 뒤, 실제 사용자 토큰으로 결과를 검증합니다.
권한은 토큰에 복사되지 않아 이후 변경도 다음 요청부터 반영됩니다.
</div>
<div class="kb-ops-flow">
<div class="kb-ops-step">
<div class="kb-ops-step-no">1</div>
<strong>사용자·그룹</strong>
<span>업무 대상을 등록합니다.</span>
</div>
<div class="kb-ops-step">
<div class="kb-ops-step-no">2</div>
<strong>역할</strong>
<span>직접·그룹 역할을 부여합니다.</span>
</div>
<div class="kb-ops-step accent">
<div class="kb-ops-step-no">3</div>
<strong>접근 규칙</strong>
<span>객체·행·컬럼 접근을 설정합니다.</span>
</div>
<div class="kb-ops-step accent">
<div class="kb-ops-step-no">4</div>
<strong>보호·연결</strong>
<span>VPD·ORDS 대상을 확인합니다.</span>
</div>
<div class="kb-ops-step success">
<div class="kb-ops-step-no">5</div>
<strong>유효 권한·접근 검증</strong>
<span>토큰으로 실제 결과를 확인합니다.</span>
</div>
</div>
<div class="kb-ops-portal">
<strong>보안 운영 포털</strong>
<p>
권한 등록, 접근 규칙 관리와 검증 결과 확인은 별도 운영 화면에서 수행합니다.
</p>
<a class="kb-ops-link" href="{VPD_OPERATIONS_URL}"
target="_blank" rel="noopener noreferrer">
권한 운영 화면 열기&nbsp;&#8599;
</a>
</div>
</section>
""",
unsafe_allow_html=True,
)
def _process_submitted_question(
*,
question: str,
conversation_id: str,
chat_page_key: str,
query_progress_slot: Any,
query_progress_notice_key: str,
bearer_token: str,
servers: list[McpServer],
default_router_model_profile: str,
selected_query_model_profile: str,
mcp_cache_generation_key: str,
limit: int,
selected_token_preset: VpdTokenPreset | None,
execution_mode_override: str,
) -> None:
normalized_question = question.strip()
if not normalized_question:
query_progress_slot.warning("질문을 입력해 주세요.")
return
active_model_profile = (
selected_query_model_profile.strip()
or _complex_reasoning_model_profile(default_router_model_profile)
)
single_tool_query = ""
process_started = perf_counter()
execution_events: list[dict[str, Any]] = []
with query_progress_slot.container():
with st.chat_message("user"):
st.markdown(normalized_question)
processing_status = st.status("질의 진행 상황 · 준비 중", expanded=True)
with processing_status:
progress_bar = st.progress(0, text="질의 처리를 준비하고 있습니다.")
def record_execution_event(
*,
percent: int,
label: str,
detail: str = "",
elapsed_seconds: float | None = None,
) -> None:
event: dict[str, Any] = {
"percent": int(percent),
"label": str(label),
"total_elapsed_seconds": round(perf_counter() - process_started, 3),
}
if detail:
event["detail"] = str(detail)
if elapsed_seconds is not None:
event["elapsed_seconds"] = round(float(elapsed_seconds), 3)
execution_events.append(event)
def refresh_progress(
percent: int,
label: str,
completed_detail: str | None = None,
elapsed_seconds: float | None = None,
) -> None:
progress_bar.progress(percent, text=label)
processing_status.update(
label=f"질의 진행 상황 · {percent}% · {label}",
state="running",
expanded=True,
)
if completed_detail:
processing_status.write(completed_detail)
record_execution_event(
percent=percent,
label=label,
detail=completed_detail,
elapsed_seconds=elapsed_seconds,
)
try:
with processing_status:
refresh_progress(5, "대화 문맥을 불러오고 있습니다.")
step_started = perf_counter()
conversation = load_chat_context(conversation_id)
step_elapsed = perf_counter() - step_started
refresh_progress(
15,
"대화 문맥을 불러왔습니다.",
f"대화 문맥 로드 완료 · {step_elapsed:.1f}s",
elapsed_seconds=step_elapsed,
)
refresh_progress(20, "질문 문맥을 정리하고 있습니다.")
step_started = perf_counter()
standalone_question = resolve_standalone_question(
question=normalized_question,
messages=conversation,
model_profile_key=active_model_profile,
)
step_elapsed = perf_counter() - step_started
refresh_progress(
30,
"질문 문맥을 정리했습니다.",
f"후속 질문 정리 완료 · {step_elapsed:.1f}s",
elapsed_seconds=step_elapsed,
)
refresh_progress(35, "사용 가능한 MCP 도구를 확인하고 있습니다.")
step_started = perf_counter()
cache_token = _normalized_bearer(bearer_token)
discovery_results, discovery_failures = cached_discover_enabled_server_tools(
_mcp_server_cache_rows(servers),
_token_fingerprint(cache_token),
int(st.session_state.get(mcp_cache_generation_key, 0) or 0),
bearer_token,
)
step_elapsed = perf_counter() - step_started
refresh_progress(
50,
"MCP 도구 확인을 완료했습니다.",
f"MCP 툴 디스커버리 완료 · {step_elapsed:.1f}s",
elapsed_seconds=step_elapsed,
)
routed_tools = [
RoutedMcpTool(server_id=result.server.server_id, tool=tool)
for result in discovery_results
for tool in result.tools
]
if discovery_failures:
st.warning(
"일부 MCP 서버 디스커버리 실패: "
+ ", ".join(
f"{failure['server_id']}={failure['error']}"
for failure in discovery_failures
)
)
if not routed_tools:
raise PublicMcpError("디스커버리된 MCP 툴이 없습니다.")
refresh_progress(55, "MCP 실행 방식을 판단하고 있습니다.")
step_started = perf_counter()
execution_model_profile = active_model_profile
execution_mode_plan = plan_mcp_execution_mode(
question=standalone_question,
routed_tools=routed_tools,
model_profile_key=execution_model_profile,
mode_override=execution_mode_override,
)
if (
execution_mode_override == "auto"
and execution_model_profile != default_router_model_profile
and str(execution_mode_plan.get("reason") or "").startswith(
"실행 방식 판단 실패"
)
):
execution_mode_plan = plan_mcp_execution_mode(
question=standalone_question,
routed_tools=routed_tools,
model_profile_key=default_router_model_profile,
mode_override=execution_mode_override,
)
execution_mode = str(execution_mode_plan.get("mode") or "single")
is_complex_execution = execution_mode == "agent" and len(routed_tools) > 1
reasoning_model_profile = active_model_profile
route_key = str(execution_mode_plan.get("route_key") or "")
_, routes_by_key = _agent_tool_catalog(routed_tools)
selected_route = routes_by_key.get(route_key) or _default_single_route(
routed_tools
)
if execution_mode_override == "single" and len(routed_tools) > 1:
selected_route = route_mcp_tool_across_servers_with_llm(
routed_tools,
standalone_question,
router_model_profile=execution_model_profile,
)
step_elapsed = perf_counter() - step_started
execution_mode_detail = (
"실행 방식 판단 완료 · "
f"{execution_mode} · "
f"{execution_mode_plan.get('reason') or 'reason 없음'} · "
f"{step_elapsed:.1f}s"
)
processing_status.write(execution_mode_detail)
record_execution_event(
percent=55,
label="실행 방식 판단 완료",
detail=execution_mode_detail,
elapsed_seconds=step_elapsed,
)
agent_steps: list[Mapping[str, Any]] = []
agent_stop_reason = ""
if is_complex_execution:
refresh_progress(60, "멀티툴 Agent가 MCP 실행 계획을 세우고 있습니다.")
step_started = perf_counter()
def on_agent_step(step: Mapping[str, Any]) -> None:
step_elapsed = float(step.get("elapsed_seconds") or 0)
agent_detail = (
f"Agent step {step.get('step')} · "
f"{step.get('server_id')}/{step.get('tool_name')} · "
f"{step_elapsed:.1f}s"
)
processing_status.write(agent_detail)
record_execution_event(
percent=60,
label=f"Agent step {step.get('step')}",
detail=agent_detail,
elapsed_seconds=step_elapsed,
)
agent_result = run_mcp_agent_loop(
question=standalone_question,
conversation=conversation,
routed_tools=routed_tools,
servers=servers,
bearer_token=bearer_token,
limit=int(limit),
model_profile_key=reasoning_model_profile,
progress_callback=on_agent_step,
)
selected_server = agent_result["server"]
selected = agent_result["tool"]
arguments = agent_result["arguments"]
answer_source = agent_result["mcp_result"]
agent_steps = list(agent_result["agent_steps"])
agent_stop_reason = str(agent_result.get("stop_reason") or "")
step_elapsed = perf_counter() - step_started
refresh_progress(
85,
"Agent MCP 실행을 완료했습니다.",
f"Agent loop 완료 · {len(agent_steps)} step · "
f"{step_elapsed:.1f}s",
elapsed_seconds=step_elapsed,
)
else:
refresh_progress(65, "MCP 단일 호출을 실행하고 있습니다.")
step_started = perf_counter()
selected_server = next(
server
for server in servers
if server.server_id == selected_route.server_id
)
selected = selected_route.tool
tool_query = prepare_tool_query_for_mcp(
question=standalone_question,
server=selected_server,
tool=selected,
model_profile_key=reasoning_model_profile,
selected_user_id=(
selected_token_preset.user_id
if selected_token_preset is not None
else ""
),
selected_user_role=(
selected_token_preset.role
if selected_token_preset is not None
else ""
),
selected_user_channel=(
selected_token_preset.channel
if selected_token_preset is not None
else ""
),
selected_user_scope=(
selected_token_preset.scope
if selected_token_preset is not None
else ""
),
)
single_tool_query = tool_query
arguments = build_mcp_tool_arguments(
selected,
tool_query,
int(limit),
preferred_tool=selected_server.default_tool,
)
raw_result = call_tool(
base_url=selected_server.endpoint_url,
bearer_token=bearer_token,
tool=selected,
arguments=arguments,
)
parsed = _content_text_json(raw_result)
answer_source = parsed if parsed is not None else raw_result
step_elapsed = perf_counter() - step_started
refresh_progress(
85,
"MCP 단일 호출을 완료했습니다.",
f"MCP 1회 호출 완료 · {selected_server.server_id}/{selected.name} · "
f"{step_elapsed:.1f}s",
elapsed_seconds=step_elapsed,
)
except (PublicMcpError, McpToolRouterError) as exc:
security_evidence = collect_security_evidence(
normalized_question,
selected_token_preset,
failure_message=str(exc),
)
business_evidence = collect_business_evidence(
normalized_question,
selected_token_preset,
)
security_verified = bool(
security_evidence and not security_evidence.get("evidence_error")
)
business_verified = bool(
business_evidence
and business_evidence.get("evidence_complete")
and not business_evidence.get("evidence_error")
)
if security_verified or business_verified:
verified_answer = _enrich_answer_with_security_evidence(
"업무 MCP 호출은 완료되지 않았지만 검증 근거는 정상 확인되었습니다.",
security_evidence,
)
verified_answer = _enrich_answer_with_business_evidence(
verified_answer,
business_evidence,
)
verified_basis: list[str] = []
if security_verified:
verified_basis.append(
"보안 검증 증적: "
f"{security_evidence.get('audit_source')} "
f"LOG_ID={security_evidence.get('audit_log_id')}"
)
if business_verified:
verified_basis.append(
"업무 검증 증적: "
+ ", ".join(
str(source)
for source in business_evidence.get("evidence_sources", [])
)
)
save_chat_turn(
conversation_id=conversation_id,
selected_user_id=(
selected_token_preset.user_id
if selected_token_preset is not None
else ""
),
selected_user_label=(
selected_token_preset.display_label
if selected_token_preset is not None
else "직접 입력"
),
question=normalized_question,
standalone_question=locals().get(
"standalone_question", normalized_question
),
answer=verified_answer,
basis=verified_basis,
limitations="",
details={
"server_id": "verified_evidence",
"tool_name": str(
security_evidence.get("evidence_type") or ""
)
or str(business_evidence.get("evidence_type") or ""),
"standalone_question": locals().get(
"standalone_question", normalized_question
),
"security_evidence": security_evidence,
"business_evidence": business_evidence,
"mcp_error": str(exc),
"execution_events": execution_events,
},
)
progress_bar.progress(100, text="검증 결과를 확인했습니다.")
processing_status.update(
label="질의 진행 상황 · 100% · 근거 검증 완료",
state="complete",
expanded=False,
)
st.session_state[query_progress_notice_key] = (
"업무 MCP 오류와 별개로 토큰 범위 검증 결과를 확인했습니다."
)
st.session_state[chat_page_key] = 0
st.rerun()
progress_bar.empty()
processing_status.update(label="질문 처리 실패", state="error", expanded=True)
processing_status.error(str(exc))
return
security_evidence = collect_security_evidence(
normalized_question,
selected_token_preset,
)
business_evidence = collect_business_evidence(
normalized_question,
selected_token_preset,
)
evidence_verified_for_answer = bool(
business_evidence
and business_evidence.get("evidence_complete")
and not business_evidence.get("evidence_error")
) or bool(
security_evidence
and security_evidence.get("evidence_type")
and not security_evidence.get("evidence_error")
)
assistant_message: dict[str, Any]
synthesis_error_details: dict[str, Any] = {}
query_contract_details: list[Any] = []
evidence_contract_details: list[Any] = []
answer_model_profile = reasoning_model_profile
answer_synthesis_fallback_model = ""
try:
with processing_status:
refresh_progress(90, "질의 결과를 정리하고 있습니다.")
step_started = perf_counter()
synthesized = synthesize_answer(
question=normalized_question,
conversation=conversation,
server=selected_server,
tool=selected,
arguments=arguments,
mcp_result=answer_source,
model_profile_key=reasoning_model_profile,
agent_steps=agent_steps,
security_evidence=security_evidence,
business_evidence=business_evidence,
)
step_elapsed = perf_counter() - step_started
refresh_progress(
96,
"질의 결과 정리를 완료했습니다.",
f"최종 답변 생성 완료 · {step_elapsed:.1f}s",
elapsed_seconds=step_elapsed,
)
assistant_message = {
"role": "assistant",
"content": str(synthesized["answer"]),
"basis": synthesized.get("basis", []),
"limitations": str(synthesized.get("limitations") or ""),
}
query_contract_details = list(synthesized.get("query_contracts") or [])
evidence_contract_details = list(
synthesized.get("evidence_contract_report") or []
)
except AnswerSynthesisError as exc:
synthesis_error_details = getattr(exc, "diagnostics", {}) or {}
fallback_profile = DEFAULT_SYNTHESIS_FALLBACK_MODEL_PROFILE
if fallback_profile and fallback_profile != reasoning_model_profile:
try:
with processing_status:
refresh_progress(
93,
f"최종 답변을 {fallback_profile}로 재시도하고 있습니다.",
)
step_started = perf_counter()
synthesized = synthesize_answer(
question=normalized_question,
conversation=conversation,
server=selected_server,
tool=selected,
arguments=arguments,
mcp_result=answer_source,
model_profile_key=fallback_profile,
agent_steps=agent_steps,
security_evidence=security_evidence,
business_evidence=business_evidence,
)
step_elapsed = perf_counter() - step_started
refresh_progress(
96,
"질의 결과 정리를 완료했습니다.",
f"fallback 답변 생성 완료 · {fallback_profile} · "
f"{step_elapsed:.1f}s",
elapsed_seconds=step_elapsed,
)
answer_model_profile = fallback_profile
answer_synthesis_fallback_model = fallback_profile
assistant_message = {
"role": "assistant",
"content": str(synthesized["answer"]),
"basis": synthesized.get("basis", []),
"limitations": str(synthesized.get("limitations") or ""),
}
query_contract_details = list(
synthesized.get("query_contracts") or []
)
evidence_contract_details = list(
synthesized.get("evidence_contract_report") or []
)
except AnswerSynthesisError as fallback_exc:
synthesis_error_details = {
"primary": synthesis_error_details,
"fallback": getattr(fallback_exc, "diagnostics", {}) or {},
}
fallback_content = fallback_answer_from_mcp(answer_source, agent_steps)
fallback_content = _enrich_answer_with_security_evidence(
fallback_content,
security_evidence,
)
fallback_content = _enrich_answer_with_business_evidence(
fallback_content,
business_evidence,
)
assistant_message = {
"role": "assistant",
"content": fallback_content,
"basis": [],
"limitations": (
""
if evidence_verified_for_answer
else f"{fallback_exc} MCP 원본 응답은 상세에서 확인 가능합니다."
),
}
refresh_progress(96, "원본 MCP 응답으로 결과를 구성했습니다.")
else:
fallback_content = fallback_answer_from_mcp(answer_source, agent_steps)
fallback_content = _enrich_answer_with_security_evidence(
fallback_content,
security_evidence,
)
fallback_content = _enrich_answer_with_business_evidence(
fallback_content,
business_evidence,
)
assistant_message = {
"role": "assistant",
"content": fallback_content,
"basis": [],
"limitations": (
""
if evidence_verified_for_answer
else f"{exc} MCP 원본 응답은 상세에서 확인 가능합니다."
),
}
refresh_progress(96, "원본 MCP 응답으로 결과를 구성했습니다.")
pre_save_elapsed = perf_counter() - process_started
record_execution_event(
percent=98,
label="질의 결과 저장 준비 완료",
detail=f"저장 전 처리 완료 · {pre_save_elapsed:.1f}s",
elapsed_seconds=pre_save_elapsed,
)
assistant_message["details"] = {
"server_id": selected_server.server_id,
"tool_name": selected.name,
"standalone_question": standalone_question,
"tool_query": single_tool_query,
"arguments": arguments,
"mcp_result": answer_source,
"execution_mode": str(execution_mode_plan.get("mode") or ""),
"execution_mode_reason": str(execution_mode_plan.get("reason") or ""),
"execution_mode_model_profile": str(
execution_mode_plan.get("model_profile") or ""
),
"reasoning_model_profile": reasoning_model_profile,
"answer_model_profile": answer_model_profile,
"answer_synthesis_fallback_model": answer_synthesis_fallback_model,
"answer_synthesis_error": synthesis_error_details,
"agent_steps": agent_steps,
"agent_stop_reason": agent_stop_reason,
"security_evidence": security_evidence,
"business_evidence": business_evidence,
"query_contracts": query_contract_details,
"evidence_contract_report": evidence_contract_details,
"execution_events": execution_events,
"discovered_routes": [
{
"server_id": discovery.server.server_id,
"tool_name": tool.name,
"readOnly": tool.read_only,
"description": tool.description,
}
for discovery in discovery_results
for tool in discovery.tools
],
"discovery_failures": discovery_failures,
}
refresh_progress(98, "질의 결과를 저장하고 있습니다.")
save_chat_turn(
conversation_id=conversation_id,
selected_user_id=(
selected_token_preset.user_id if selected_token_preset is not None else ""
),
selected_user_label=(
selected_token_preset.display_label
if selected_token_preset is not None
else "직접 입력"
),
question=normalized_question,
standalone_question=standalone_question,
answer=str(assistant_message["content"]),
basis=assistant_message.get("basis", []),
limitations=str(assistant_message.get("limitations") or ""),
details=assistant_message["details"],
)
total_elapsed = perf_counter() - process_started
refresh_progress(
100,
"질의 처리가 완료되었습니다.",
f"전체 처리 완료 · {total_elapsed:.1f}s",
elapsed_seconds=total_elapsed,
)
processing_status.update(
label="질의 진행 상황 · 100% · 처리 완료",
state="complete",
expanded=False,
)
st.session_state[query_progress_notice_key] = (
"질의 처리가 완료되었습니다. "
f"사용 경로: {selected_server.server_id} / {selected.name} · "
f"{'agent step ' + str(len(agent_steps)) + '회' if agent_steps else 'MCP 1회 호출'}"
)
st.session_state[chat_page_key] = 0
st.rerun()
def main() -> None:
try:
profile = load_app_profile(APP_PROFILE_FILE, ENV_FILE)
except AppProfileError as exc:
st.set_page_config(page_title="AI 업무 에이전트", layout="wide")
st.error(str(exc))
return
st.set_page_config(
page_title=profile.page_title, page_icon=profile.page_icon, layout="wide"
)
_apply_console_theme(profile)
_restore_portal_proxy_session()
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"
question_text_key = "poc4_mcp_discovery_question_text"
loaded_scenario_key = "poc4_mcp_discovery_loaded_scenario_id"
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()
if chat_page_key not in st.session_state:
st.session_state[chat_page_key] = 0
if mcp_cache_generation_key not in st.session_state:
st.session_state[mcp_cache_generation_key] = 0
conversation_id = str(st.session_state[conversation_id_key])
try:
servers, default_server_index = load_mcp_servers()
except PublicMcpError as exc:
st.error(str(exc))
return
try:
token_presets = load_vpd_token_presets()
except PublicMcpError as exc:
st.error(str(exc))
return
default_router_model_profile = servers[default_server_index].router_model_profile
configured_mcp_bearer = _runtime_env_value(
servers[default_server_index].auth_token_env
)
query_model_profile_key = "poc4_query_model_profile"
default_query_model_profile = DEFAULT_QUERY_MODEL_PROFILE
try:
model_profile_options = _model_profile_select_options()
except ValueError:
model_profile_options = (
(default_query_model_profile, default_query_model_profile),
)
model_profile_labels = dict(model_profile_options)
model_profile_keys = tuple(model_profile_labels)
if default_query_model_profile not in model_profile_keys:
default_query_model_profile = (
default_router_model_profile
if default_router_model_profile in model_profile_keys
else model_profile_keys[0]
)
if st.session_state.get(query_model_profile_key) not in model_profile_keys:
st.session_state[query_model_profile_key] = default_query_model_profile
_render_app_header(profile)
with st.sidebar:
st.caption(
f"포털 사용자 · {st.session_state.get(PORTAL_AUTH_USER_KEY, '')}"
)
st.markdown(
'<a class="console-logout-button" href="/auth/logout" '
'target="_self">로그아웃</a>',
unsafe_allow_html=True,
)
st.divider()
st.markdown('<div class="kb-panel-title">AI 사용자 설정</div>', unsafe_allow_html=True)
selected_query_model_profile = st.selectbox(
"LLM 모델",
options=model_profile_keys,
key=query_model_profile_key,
format_func=lambda value: (
f"{model_profile_labels.get(value, value)} · {value}"
),
help=(
"선택한 모델을 후속 질문 정리, 실행 방식 판단, MCP tool 라우팅, "
"Agent planning, 최종 답변 합성에 사용합니다."
),
)
if token_presets:
vpd_user_key = "poc4_vpd_token_preset"
token_preset_by_id = {
preset.user_id: preset for preset in token_presets
}
token_preset_ids = tuple(token_preset_by_id)
stored_vpd_user = st.session_state.get(vpd_user_key)
if vpd_user_key not in st.session_state:
st.session_state[vpd_user_key] = _default_vpd_user_id(token_presets)
elif isinstance(stored_vpd_user, VpdTokenPreset):
st.session_state[vpd_user_key] = stored_vpd_user.user_id
elif stored_vpd_user not in token_preset_by_id:
st.session_state[vpd_user_key] = _default_vpd_user_id(token_presets)
current_vpd_user_id = st.session_state[vpd_user_key]
current_vpd_user_label = token_preset_by_id[
current_vpd_user_id
].select_label
st.markdown(
'<div class="kb-sidebar-section-title">데모 사용자</div>',
unsafe_allow_html=True,
)
vpd_popover_key = "poc4_vpd_user_popover_open"
def close_vpd_user_popover() -> None:
st.session_state[vpd_popover_key] = False
with st.popover(
f"**{current_vpd_user_label}**",
use_container_width=True,
key=vpd_popover_key,
on_change="rerun",
):
selected_vpd_user_id = st.radio(
"데모 사용자 선택",
options=token_preset_ids,
key=vpd_user_key,
format_func=lambda user_id: token_preset_by_id[
user_id
].select_label,
label_visibility="collapsed",
on_change=close_vpd_user_popover,
)
selected_token_preset = (
token_preset_by_id.get(selected_vpd_user_id)
if selected_vpd_user_id is not None
else None
)
else:
selected_token_preset = None
selected_preset_bearer = (
selected_token_preset.token if selected_token_preset is not None else ""
)
if configured_mcp_bearer or selected_preset_bearer:
# The HMM MCP gateway credential is never rendered. Each demo-user
# preset refers to its runtime env key, allowing secure profile swaps.
st.caption("MCP 인증: 선택 사용자 preset의 서버 관리 토큰 적용")
manual_bearer_token = ""
elif selected_token_preset is not None:
manual_bearer_token = st.text_input(
"Bearer token",
value=selected_token_preset.token,
type="default",
disabled=True,
key=f"poc4_bearer_token_{selected_token_preset.user_id}",
help="선택한 데모 사용자의 MCP gateway token입니다.",
)
else:
manual_bearer_token = st.text_input(
"Bearer token",
type="default",
key="poc4_manual_bearer_token",
)
bearer_token = selected_preset_bearer or configured_mcp_bearer or manual_bearer_token
if selected_token_preset is not None:
_render_demo_user_card(selected_token_preset)
all_conversation_rows = list_conversations(limit=200)
current_row = next(
(
row
for row in all_conversation_rows
if row["conversation_id"] == conversation_id
),
{
"conversation_id": conversation_id,
"created_at": "",
"updated_at": "",
"title": "",
"turn_count": 0,
"latest_question": "",
},
)
conversation_search = st.text_input(
"대화 검색",
placeholder="질문, 제목, 세션 ID",
key="poc4_conversation_search",
)
conversation_rows = filter_conversations(
all_conversation_rows,
conversation_search,
)
search_query = conversation_search.strip()
current_in_results = any(
row["conversation_id"] == conversation_id for row in conversation_rows
)
if search_query and conversation_rows and not current_in_results:
st.session_state[conversation_id_key] = str(
conversation_rows[0]["conversation_id"]
)
st.session_state[chat_page_key] = 0
st.rerun()
if search_query and not conversation_rows:
st.info("검색 결과가 없습니다.")
if not search_query and not current_in_results:
conversation_rows.insert(0, current_row)
conversation_ids = [str(row["conversation_id"]) for row in conversation_rows]
conversation_by_id = {
str(row["conversation_id"]): row for row in conversation_rows
}
if conversation_ids:
selected_conversation_id = st.selectbox(
"대화 세션",
options=conversation_ids,
index=(
conversation_ids.index(conversation_id)
if conversation_id in conversation_ids
else 0
),
format_func=lambda value: conversation_label(conversation_by_id[value]),
)
else:
st.caption("저장된 대화가 없습니다.")
selected_conversation_id = conversation_id
conversation_by_id = {conversation_id: current_row}
if selected_conversation_id != conversation_id:
st.session_state[conversation_id_key] = selected_conversation_id
st.session_state[chat_page_key] = 0
st.rerun()
if st.button("새 대화 시작"):
st.session_state[conversation_id_key] = new_conversation_id()
st.session_state[chat_page_key] = 0
st.rerun()
with st.expander("대화 관리"):
selected_row = conversation_by_id.get(conversation_id, current_row)
title_key = f"poc4_conversation_title_{conversation_id}"
title_value = st.text_input(
"대화 이름",
value=str(selected_row.get("title") or ""),
key=title_key,
max_chars=80,
)
if st.button("대화 이름 저장"):
rename_conversation(conversation_id, title_value)
st.rerun()
st.download_button(
"대화 JSON 다운로드",
data=json.dumps(
{
"conversation_id": conversation_id,
"title": title_value.strip(),
"turns": load_all_chat_turns(conversation_id),
},
ensure_ascii=False,
indent=2,
default=str,
),
file_name=f"{conversation_id}.json",
mime="application/json",
)
delete_confirmed = st.checkbox("현재 대화 삭제 확인")
if st.button("현재 대화 삭제", disabled=not delete_confirmed):
delete_conversation(conversation_id)
st.session_state[conversation_id_key] = new_conversation_id()
st.session_state[chat_page_key] = 0
st.rerun()
with st.expander("MCP 고급 설정", expanded=False):
st.markdown(
'<div class="kb-sidebar-section-title">MCP Servers</div>',
unsafe_allow_html=True,
)
st.caption(", ".join(server.server_id for server in servers))
if st.button("MCP 툴 새로고침"):
st.session_state[mcp_cache_generation_key] = (
int(st.session_state.get(mcp_cache_generation_key, 0) or 0) + 1
)
st.toast("MCP tools/list 캐시를 새로고침합니다.")
st.caption(
f"tools/list cache generation: "
f"{int(st.session_state.get(mcp_cache_generation_key, 0) or 0)}"
)
limit = st.number_input(
"limit",
min_value=1,
max_value=1000,
value=50,
)
execution_mode_override = st.selectbox(
"MCP 실행 모드",
options=("auto", "single", "agent"),
index=0,
format_func=lambda value: {
"auto": "자동 · 작은 모델이 단일/멀티툴 판단",
"single": "단일 · MCP 1회 호출",
"agent": "멀티툴 Agent · ReAct loop",
}[value],
help=(
"속도 우선이면 자동 또는 단일을 사용하세요. "
"자동 모드는 선택한 LLM 모델로 실행 방식을 판단합니다."
),
)
st.markdown("**Resolved MCP endpoints**")
for server in servers:
st.caption(f"{server.server_id}: {server.endpoint_url}")
if st.checkbox("MCP 설정 JSON 보기"):
try:
st.json(
json.loads(MCP_SERVERS_FILE.read_text(encoding="utf-8"))
)
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"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}")
st.caption(
"synthesis fallback model: "
f"{DEFAULT_SYNTHESIS_FALLBACK_MODEL_PROFILE}"
)
st.markdown(
'<div id="kb-main-tabs-anchor" style="scroll-margin-top: 0.75rem;"></div>',
unsafe_allow_html=True,
)
architecture_tab, scenario_tab, audit_tab, operations_tab = st.tabs(
["아키텍처", "시나리오", "감사로그", "보안관리"]
)
with architecture_tab:
_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"))
)
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 "")
with st.form("poc4_mcp_question_form"):
question = st.text_area(
"질문",
label_visibility="collapsed",
key=question_text_key,
height=120,
max_chars=1_000,
placeholder="업무 질문을 직접 입력하거나 위 질의 샘플을 선택하세요.",
)
submitted = st.form_submit_button(
"질문 전송",
type="primary",
icon=":material/send:",
icon_position="right",
width="stretch",
)
if submitted:
components.html(
"""
<script>
const target = window.parent.document.getElementById(
"kb-main-tabs-anchor"
);
if (target) {
window.parent.requestAnimationFrame(() => {
target.scrollIntoView({ behavior: "smooth", block: "start" });
});
}
</script>
""",
height=0,
)
st.markdown(
'<div id="kb-query-results-anchor" '
'style="scroll-margin-top: 0.75rem;"></div>',
unsafe_allow_html=True,
)
query_progress_slot = st.empty()
query_progress_notice = st.session_state.pop(
query_progress_notice_key,
None,
)
if query_progress_notice:
query_progress_slot.success(str(query_progress_notice))
_render_chat_history(conversation_id, chat_page_key)
if query_progress_notice:
components.html(
"""
<script>
window.parent.setTimeout(() => {
const target = window.parent.document.getElementById(
"kb-query-results-anchor"
);
if (target) {
target.scrollIntoView({ behavior: "smooth", block: "start" });
}
}, 250);
</script>
""",
height=0,
)
with audit_tab:
render_hmm_audit_tab(
st,
_load_hmm_audit_inventory,
_load_hmm_audit_events,
AuditLogError,
)
with operations_tab:
_render_vpd_operations_tab()
if not submitted:
return
with scenario_tab:
_process_submitted_question(
question=question,
conversation_id=conversation_id,
chat_page_key=chat_page_key,
query_progress_slot=query_progress_slot,
query_progress_notice_key=query_progress_notice_key,
bearer_token=bearer_token,
servers=servers,
default_router_model_profile=default_router_model_profile,
selected_query_model_profile=selected_query_model_profile,
mcp_cache_generation_key=mcp_cache_generation_key,
limit=int(limit),
selected_token_preset=selected_token_preset,
execution_mode_override=execution_mode_override,
)
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink