shinalok
fb438864b1
- app.py: styled HTML header with robot icon + subtitle (D2-11) - app.py: gr.themes.Soft(blue/indigo/slate) applied to gr.Blocks (D2-12) - app.py: user_selector moved to header Row, right-aligned scale=0 (D2-13) - app.py: .message-wrap .user/.bot custom background + border-radius CSS (D2-14) - app.py: streaming-indicator blink animation on _live_html (D2-15) - app.py: @media max-width:768px responsive CSS (D2-16) - ROADMAP.md: mark all D2 items complete Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
463 lines
16 KiB
Python
463 lines
16 KiB
Python
"""율봇 WebUI — youlbot REST API를 호출하는 Gradio 프론트엔드.
|
|
|
|
실행:
|
|
python app.py
|
|
|
|
환경변수 (.env):
|
|
YOULBOT_API_URL=http://localhost:8000
|
|
YOULBOT_API_TOKEN= ← api.py에 API_TOKEN 설정 시 동일 값
|
|
"""
|
|
import html as _html
|
|
import logging
|
|
import os
|
|
|
|
import gradio as gr
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
logging.basicConfig(
|
|
level=os.getenv("LOG_LEVEL", "INFO").upper(),
|
|
format="%(asctime)s %(levelname)-8s %(name)s — %(message)s",
|
|
datefmt="%Y-%m-%d %H:%M:%S",
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
from container import Container
|
|
|
|
container = Container()
|
|
|
|
USER_LABELS = ["아록", "근혜", "도율", "하율"]
|
|
DEFAULT_USER = "아록"
|
|
|
|
# ── STT (Whisper) — 로컬 실행 유지 ──────────────────────────────
|
|
_whisper_model = None
|
|
|
|
|
|
def _get_whisper():
|
|
global _whisper_model
|
|
if _whisper_model is None:
|
|
import whisper
|
|
_whisper_model = whisper.load_model(container.config().whisper_model_size)
|
|
return _whisper_model
|
|
|
|
|
|
def transcribe_audio(filepath: str) -> str:
|
|
if not filepath:
|
|
return ""
|
|
model = _get_whisper()
|
|
result = model.transcribe(filepath, language="ko")
|
|
return result["text"].strip()
|
|
|
|
|
|
# ── 채팅 ─────────────────────────────────────────────────────────
|
|
|
|
async def respond(message, history, show_thinking, user_id, use_tts, run_ids, image_path):
|
|
if not message.strip() and not image_path:
|
|
yield history, "", None, run_ids, "", "", None
|
|
return
|
|
|
|
history = list(history)
|
|
run_ids = list(run_ids)
|
|
display_msg = message
|
|
if image_path:
|
|
display_msg = f"🖼️ [이미지 첨부]\n{message}" if message.strip() else "🖼️ [이미지 첨부]"
|
|
history.append({"role": "user", "content": display_msg})
|
|
history.append({"role": "assistant", "content": ""})
|
|
yield history, "", None, run_ids, "", "", None # boxes 초기화 + 이미지 초기화
|
|
|
|
collected_run_id: str | None = None
|
|
tts_text = ""
|
|
thinking_acc = ""
|
|
thinking_text = ""
|
|
thinking_finalized = False
|
|
source_box_html = ""
|
|
|
|
try:
|
|
async for token, run_id in container.chat_service().chat(
|
|
message or "이 이미지를 분석해줘.", user_id, show_thinking, image_path=image_path
|
|
):
|
|
if run_id is not None:
|
|
collected_run_id = run_id
|
|
break
|
|
|
|
# 즉시 상태 — thinking_acc에 누적 안 함
|
|
if isinstance(token, dict) and "__status" in token:
|
|
if not thinking_acc:
|
|
yield history, "", None, run_ids, _status_html(token["__status"]), gr.update(), gr.update()
|
|
continue
|
|
|
|
# 사고 과정(LLM thinking) — 현재 줄만 live_html로 표시
|
|
if isinstance(token, dict) and "__thinking" in token:
|
|
thinking_text += token["__thinking"]
|
|
thinking_acc += token["__thinking"]
|
|
yield history, "", None, run_ids, _live_html(_last_line(thinking_text)), gr.update(), gr.update()
|
|
continue
|
|
|
|
# 진행 로그(LangGraph, 검색 등) — 메시지 전체를 live_html로 표시
|
|
if isinstance(token, dict) and "__meta" in token:
|
|
thinking_acc += token["__meta"]
|
|
live = token["__meta"].strip()
|
|
if live:
|
|
yield history, "", None, run_ids, _live_html(live), gr.update(), gr.update()
|
|
continue
|
|
|
|
# RAG 출처 — 별도 source_box로 표시
|
|
if isinstance(token, dict) and "__sources" in token:
|
|
source_box_html = _sources_html(token["__sources"])
|
|
yield history, "", None, run_ids, gr.update(), source_box_html, gr.update()
|
|
continue
|
|
|
|
# 첫 답변 토큰 도착 — 전체를 details로 전환 (접힌 상태)
|
|
if thinking_acc and not thinking_finalized:
|
|
thinking_finalized = True
|
|
yield history, "", None, run_ids, _thinking_html(thinking_acc), gr.update(), gr.update()
|
|
|
|
tts_text += token
|
|
history[-1]["content"] += token
|
|
yield history, "", None, run_ids, gr.update(), gr.update(), gr.update()
|
|
|
|
except Exception as e:
|
|
history[-1]["content"] += f"\n\n[오류: {e}]"
|
|
yield history, "", None, run_ids, gr.update(), gr.update(), gr.update()
|
|
return
|
|
|
|
run_ids.append(collected_run_id)
|
|
|
|
if use_tts:
|
|
audio_path = await container.tts_service().speak(tts_text)
|
|
yield history, "", audio_path, run_ids, gr.update(), gr.update(), gr.update()
|
|
else:
|
|
yield history, "", None, run_ids, gr.update(), gr.update(), gr.update()
|
|
|
|
|
|
async def handle_feedback(like_data: gr.LikeData, history, run_ids, user_id):
|
|
idx = like_data.index
|
|
if isinstance(idx, (list, tuple)):
|
|
idx = idx[0]
|
|
if not isinstance(idx, int) or idx < 0 or idx >= len(history):
|
|
return
|
|
if history[idx].get("role") != "assistant":
|
|
return
|
|
asst_turn = sum(1 for m in history[:idx] if m.get("role") == "assistant")
|
|
run_id = run_ids[asst_turn] if run_ids and asst_turn < len(run_ids) else None
|
|
|
|
user_msg = str(history[idx - 1]["content"]) if idx > 0 else ""
|
|
asst_msg = str(history[idx]["content"])
|
|
rating = 1 if like_data.liked else -1
|
|
|
|
try:
|
|
await container.chat_service().save_feedback(user_id, user_msg, asst_msg, rating, run_id)
|
|
except Exception as e:
|
|
logger.error("피드백 저장 실패: %s", e)
|
|
|
|
|
|
def switch_user(user_id):
|
|
return [], []
|
|
|
|
|
|
async def reset_chat(user_id):
|
|
try:
|
|
await container.chat_service().reset(user_id)
|
|
except Exception as e:
|
|
logger.error("대화 초기화 실패: %s", e)
|
|
return [], []
|
|
|
|
|
|
# ── 문서 관리 ─────────────────────────────────────────────────────
|
|
|
|
async def ingest_files(files):
|
|
if not files:
|
|
return gr.update(value="파일을 선택해주세요.", visible=True)
|
|
paths = [f if isinstance(f, str) else f.name for f in files]
|
|
results = []
|
|
for path in paths:
|
|
try:
|
|
result = await container.document_service().ingest(path)
|
|
name = os.path.basename(path)
|
|
results.append(f"{name} → {result.get('chunks', '?')}개 청크")
|
|
except Exception as e:
|
|
results.append(f"{os.path.basename(path)} 오류: {e}")
|
|
return gr.update(value="\n".join(results), visible=True)
|
|
|
|
|
|
async def list_docs():
|
|
try:
|
|
sources = await container.document_service().list_documents()
|
|
return [[os.path.basename(s), s] for s in sources]
|
|
except Exception as e:
|
|
return [[f"오류: {e}", ""]]
|
|
|
|
|
|
def select_doc_row(evt: gr.SelectData, doc_data):
|
|
row = evt.index[0]
|
|
try:
|
|
if hasattr(doc_data, "iloc"):
|
|
return str(doc_data.iloc[row, 1])
|
|
return str(doc_data[row][1])
|
|
except Exception:
|
|
return gr.update()
|
|
|
|
|
|
async def delete_doc(source):
|
|
if not source.strip():
|
|
return "삭제할 파일 경로를 입력하세요.", await list_docs()
|
|
try:
|
|
await container.document_service().delete_document(source.strip())
|
|
return f"삭제 완료: {os.path.basename(source.strip())}", await list_docs()
|
|
except Exception as e:
|
|
return f"오류: {e}", await list_docs()
|
|
|
|
|
|
# ── UI 구성 ──────────────────────────────────────────────────────
|
|
|
|
_BOX_STYLE = (
|
|
"background:#f9f9f9;border-left:3px solid #bbb;border-radius:6px;"
|
|
"padding:8px 14px;margin-bottom:6px;"
|
|
)
|
|
_CONTENT_STYLE = (
|
|
"margin-top:6px;white-space:pre-wrap;font-size:0.85em;"
|
|
"color:#555;max-height:160px;overflow-y:auto;"
|
|
)
|
|
|
|
|
|
def _last_line(text: str) -> str:
|
|
"""현재 진행 중인 마지막 비어있지 않은 줄 반환."""
|
|
lines = [l for l in text.split("\n") if l.strip()]
|
|
return lines[-1] if lines else text.strip()
|
|
|
|
|
|
def _live_html(text: str) -> str:
|
|
"""스트리밍 중 현재 줄만 보여주는 단순 div (details 미사용 → 닫힘 현상 없음)."""
|
|
return (
|
|
f'<div style="{_BOX_STYLE}">'
|
|
f'<strong class="streaming-indicator">⏳ 분석 중...</strong>'
|
|
f'<div style="{_CONTENT_STYLE}">{_html.escape(text)} ▌</div>'
|
|
f'</div>'
|
|
)
|
|
|
|
|
|
def _thinking_html(text: str) -> str:
|
|
"""완료 후 전체 내용을 접기/펼치기로 표시."""
|
|
return (
|
|
f'<details style="{_BOX_STYLE}">'
|
|
f'<summary style="cursor:pointer;font-weight:bold;">💭 분석 완료</summary>'
|
|
f'<div style="{_CONTENT_STYLE}">{_html.escape(text)}</div>'
|
|
f'</details>'
|
|
)
|
|
|
|
|
|
def _status_html(status: str) -> str:
|
|
"""내용 없이 상태만 표시하는 단순 헤더."""
|
|
return (
|
|
f'<div style="{_BOX_STYLE}">'
|
|
f'<strong>🤔 {_html.escape(status)}</strong>'
|
|
f'</div>'
|
|
)
|
|
|
|
|
|
def _sources_html(sources: list) -> str:
|
|
"""RAG 출처 목록을 접기/펼치기로 표시."""
|
|
items = "".join(
|
|
f"<li>{_html.escape(s['filename'])}"
|
|
+ (f" — {s['page']}페이지" if "page" in s else "")
|
|
+ "</li>"
|
|
for s in sources
|
|
)
|
|
return (
|
|
f'<details style="{_BOX_STYLE}">'
|
|
f'<summary style="cursor:pointer;font-weight:bold;">📄 출처 ({len(sources)}개)</summary>'
|
|
f'<ul style="margin:6px 0;padding-left:18px;font-size:0.85em;color:#555;">{items}</ul>'
|
|
f'</details>'
|
|
)
|
|
|
|
|
|
_CUSTOM_CSS = """
|
|
footer { display: none !important; }
|
|
|
|
/* 입력 영역 */
|
|
.send-btn { min-height: 80px !important; align-self: stretch !important; }
|
|
|
|
/* 헤더 (D2-11) */
|
|
.app-header {
|
|
align-items: center !important;
|
|
padding-bottom: 12px !important;
|
|
border-bottom: 1px solid var(--border-color-primary);
|
|
margin-bottom: 4px !important;
|
|
}
|
|
.app-header > .wrap, .app-header > div > .wrap {
|
|
padding: 0 !important;
|
|
background: transparent !important;
|
|
border: none !important;
|
|
box-shadow: none !important;
|
|
}
|
|
|
|
/* 채팅 버블 스타일 (D2-14) */
|
|
.message-wrap .user {
|
|
background: #dbeafe !important;
|
|
border-color: #93c5fd !important;
|
|
border-bottom-right-radius: 4px !important;
|
|
}
|
|
.message-wrap .bot {
|
|
background: #f8fafc !important;
|
|
border-color: #e2e8f0 !important;
|
|
border-bottom-left-radius: 4px !important;
|
|
}
|
|
|
|
/* 응답 스트리밍 애니메이션 (D2-15) */
|
|
@keyframes blink { 0%, 100% { opacity: 1; } 50% { opacity: 0.35; } }
|
|
.streaming-indicator { animation: blink 1.2s ease-in-out infinite; display: inline-block; }
|
|
|
|
/* 반응형 (D2-16) */
|
|
@media (max-width: 768px) {
|
|
.send-btn { min-height: 56px !important; }
|
|
.app-header { flex-wrap: wrap; gap: 8px; }
|
|
.message-wrap .user, .message-wrap .bot { max-width: 92% !important; }
|
|
}
|
|
"""
|
|
|
|
_THEME = gr.themes.Soft(
|
|
primary_hue="blue",
|
|
secondary_hue="indigo",
|
|
neutral_hue="slate",
|
|
)
|
|
|
|
with gr.Blocks(title="율봇", css=_CUSTOM_CSS, theme=_THEME) as demo:
|
|
with gr.Row(elem_classes=["app-header"]):
|
|
gr.HTML("""
|
|
<div style="display:flex;align-items:center;gap:14px;padding:4px 0;">
|
|
<span style="font-size:2.2rem;line-height:1;">🤖</span>
|
|
<div>
|
|
<div style="font-size:1.6rem;font-weight:700;color:#1e293b;line-height:1.15;">율봇</div>
|
|
<div style="font-size:0.85rem;color:#64748b;margin-top:3px;">육아·금융 전문 AI 상담 도우미</div>
|
|
</div>
|
|
</div>
|
|
""")
|
|
user_selector = gr.Dropdown(
|
|
choices=USER_LABELS,
|
|
value=DEFAULT_USER,
|
|
label="사용자",
|
|
scale=0,
|
|
min_width=160,
|
|
)
|
|
|
|
user_state = gr.State(DEFAULT_USER)
|
|
run_ids_state = gr.State([])
|
|
|
|
with gr.Tab("대화"):
|
|
thinking_box = gr.HTML(value="")
|
|
chatbot = gr.Chatbot(label="율봇", height=500, elem_id="main-chatbot")
|
|
source_box = gr.HTML(value="")
|
|
with gr.Row():
|
|
msg_box = gr.Textbox(
|
|
placeholder="질문을 입력하세요... (Enter로 전송)",
|
|
show_label=False,
|
|
lines=2,
|
|
scale=5,
|
|
autofocus=True,
|
|
)
|
|
send_btn = gr.Button("전송", variant="primary", scale=1, elem_classes=["send-btn"])
|
|
gr.Examples(
|
|
examples=[
|
|
["육아휴직 급여 신청 방법을 알려주세요"],
|
|
["어린이집 입소 대기 기간은 얼마나 걸리나요?"],
|
|
["아이 의료비 세금 공제는 어떻게 하나요?"],
|
|
],
|
|
inputs=[msg_box],
|
|
label="💡 예시 질문",
|
|
)
|
|
|
|
with gr.Accordion("📷 이미지 첨부 (선택)", open=False):
|
|
image_input = gr.Image(
|
|
type="filepath",
|
|
show_label=False,
|
|
sources=["upload", "clipboard"],
|
|
height=160,
|
|
)
|
|
|
|
with gr.Accordion("🎤 음성으로 질문하기", open=False):
|
|
with gr.Row():
|
|
audio_input = gr.Audio(
|
|
sources=["microphone"],
|
|
type="filepath",
|
|
show_label=False,
|
|
scale=4,
|
|
)
|
|
transcribe_btn = gr.Button("음성 → 텍스트 변환", scale=1)
|
|
|
|
with gr.Row():
|
|
with gr.Column(scale=3):
|
|
with gr.Row():
|
|
show_thinking = gr.Checkbox(label="사고 과정 표시", value=True)
|
|
use_tts = gr.Checkbox(label="음성으로 답변 읽기 (TTS)", value=False)
|
|
with gr.Column(scale=1, min_width=120):
|
|
reset_btn = gr.Button("대화 초기화", size="sm")
|
|
|
|
tts_output = gr.Audio(label="음성 답변", autoplay=True, visible=False)
|
|
use_tts.change(lambda v: gr.Audio(visible=v), inputs=[use_tts], outputs=[tts_output])
|
|
|
|
user_selector.change(
|
|
switch_user,
|
|
inputs=[user_selector],
|
|
outputs=[chatbot, run_ids_state],
|
|
).then(
|
|
lambda u: u, inputs=[user_selector], outputs=[user_state]
|
|
)
|
|
|
|
transcribe_btn.click(transcribe_audio, inputs=[audio_input], outputs=[msg_box])
|
|
|
|
_respond_inputs = [msg_box, chatbot, show_thinking, user_state, use_tts, run_ids_state, image_input]
|
|
_respond_outputs = [chatbot, msg_box, tts_output, run_ids_state, thinking_box, source_box, image_input]
|
|
|
|
send_btn.click(respond, inputs=_respond_inputs, outputs=_respond_outputs)
|
|
msg_box.submit(respond, inputs=_respond_inputs, outputs=_respond_outputs)
|
|
reset_btn.click(reset_chat, inputs=[user_state], outputs=[chatbot, run_ids_state])
|
|
|
|
chatbot.like(
|
|
handle_feedback,
|
|
inputs=[chatbot, run_ids_state, user_state],
|
|
outputs=[],
|
|
)
|
|
|
|
with gr.Tab("문서 등록"):
|
|
gr.Markdown("PDF 또는 TXT 파일을 업로드하면 율봇이 내용을 참고해 답변합니다.")
|
|
file_input = gr.File(
|
|
file_types=[".pdf", ".txt"],
|
|
file_count="multiple",
|
|
label="파일 선택",
|
|
)
|
|
with gr.Row():
|
|
ingest_btn = gr.Button("문서 수집", variant="primary", scale=0, min_width=200)
|
|
ingest_status = gr.Textbox(label="결과", interactive=False, visible=False)
|
|
ingest_btn.click(ingest_files, inputs=[file_input], outputs=[ingest_status])
|
|
|
|
with gr.Tab("문서 관리"):
|
|
gr.Markdown("Qdrant에 등록된 문서 목록입니다. 불필요한 문서를 삭제할 수 있습니다.")
|
|
doc_table = gr.Dataframe(
|
|
headers=["파일명", "전체 경로"],
|
|
label="등록된 문서",
|
|
interactive=False,
|
|
)
|
|
refresh_btn = gr.Button("새로고침")
|
|
gr.Markdown("---")
|
|
with gr.Row():
|
|
delete_source = gr.Textbox(
|
|
label="삭제할 파일 경로",
|
|
placeholder="위 표에서 전체 경로를 복사해 붙여넣으세요",
|
|
scale=4,
|
|
)
|
|
delete_btn = gr.Button("삭제", variant="stop", scale=1)
|
|
delete_status = gr.Textbox(label="결과", interactive=False)
|
|
|
|
refresh_btn.click(list_docs, outputs=[doc_table])
|
|
delete_btn.click(delete_doc, inputs=[delete_source], outputs=[delete_status, doc_table])
|
|
doc_table.select(select_doc_row, inputs=[doc_table], outputs=[delete_source])
|
|
demo.load(list_docs, outputs=[doc_table])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
demo.launch(
|
|
server_name=container.config().server_host,
|
|
server_port=container.config().server_port,
|
|
)
|