clovax-tax-chatbot / gradio_app.py
bissal's picture
Complete RAGโ†’LLMโ†’Web pipeline integration ๐Ÿค– Generated with Claude Code
4aa05c6
Raw History Blame Contribute Delete
8.27 kB
# gradio_app.py - ํ…Œ์ŠคํŠธ ๊ธฐ๋Šฅ์ด ํฌํ•จ๋œ Gradio ์•ฑ
import gradio as gr
import os
import sys
from datetime import datetime
# ์‹œ์Šคํ…œ ๋ชจ๋“ˆ import
try:
from tax_consultant import tax_consultant
from rag_system import search_tax_law, is_rag_available, get_rag_system
from space_test import gradio_test
RAG_AVAILABLE = True
except ImportError as e:
print(f"โš ๏ธ ๋ชจ๋“ˆ import ์‹คํŒจ: {e}")
RAG_AVAILABLE = False
def chat_with_tax_consultant(message, history):
"""์„ธ๋ฌด ์ƒ๋‹ด ์ฑ—๋ด‡"""
try:
if RAG_AVAILABLE:
# RAG ์‹œ์Šคํ…œ ์‚ฌ์šฉ
response = search_tax_law(message)
else:
# ํด๋ฐฑ ์‘๋‹ต
response = f"ํ˜„์žฌ ์‹œ์Šคํ…œ ์ ๊ฒ€ ์ค‘์ž…๋‹ˆ๋‹ค. ์งˆ๋ฌธ: {message}"
return response
except Exception as e:
return f"์˜ค๋ฅ˜๊ฐ€ ๋ฐœ์ƒํ–ˆ์Šต๋‹ˆ๋‹ค: {str(e)}"
def quick_test():
"""๋น ๋ฅธ ์‹œ์Šคํ…œ ํ…Œ์ŠคํŠธ"""
try:
test_results = []
# ํ™˜๊ฒฝ ์ •๋ณด
space_id = os.getenv('SPACE_ID')
test_results.append(f"๐ŸŒŸ ํ™˜๊ฒฝ: {'ํ—ˆ๊น…ํŽ˜์ด์Šค ์ŠคํŽ˜์ด์Šค' if space_id else '๋กœ์ปฌ'}")
test_results.append(f"๐Ÿ“… ์‹œ๊ฐ„: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
# CUDA ํ™•์ธ
try:
import torch
cuda_available = torch.cuda.is_available()
test_results.append(f"๐Ÿ”ฅ CUDA: {'์‚ฌ์šฉ๊ฐ€๋Šฅ' if cuda_available else '์‚ฌ์šฉ๋ถˆ๊ฐ€'}")
if cuda_available:
test_results.append(f"๐ŸŽฎ GPU: {torch.cuda.get_device_name()}")
except:
test_results.append("โŒ PyTorch ๋กœ๋“œ ์‹คํŒจ")
# RAG ์‹œ์Šคํ…œ ํ™•์ธ
if RAG_AVAILABLE:
try:
rag = get_rag_system()
test_results.append(f"โœ… RAG ์‹œ์Šคํ…œ: ์ •์ƒ")
test_results.append(f"๐Ÿ–ฅ๏ธ ๋””๋ฐ”์ด์Šค: {rag.device}")
if rag.vector_db:
test_results.append(f"๐Ÿ“š ๋ฒกํ„ฐ DB: {rag.vector_db.ntotal}๊ฐœ ๋ฒกํ„ฐ")
else:
test_results.append("โš ๏ธ ๋ฒกํ„ฐ DB: ๋ฏธ๊ตฌ์ถ•")
except Exception as e:
test_results.append(f"โŒ RAG ์˜ค๋ฅ˜: {str(e)}")
else:
test_results.append("โŒ RAG ์‹œ์Šคํ…œ: ์‚ฌ์šฉ๋ถˆ๊ฐ€")
# ๊ฐ„๋‹จํ•œ ๊ฒ€์ƒ‰ ํ…Œ์ŠคํŠธ
if RAG_AVAILABLE:
try:
import time
start_time = time.time()
result = search_tax_law("์ทจ๋“์„ธ ์„ธ์œจ")
response_time = time.time() - start_time
test_results.append(f"๐Ÿ” ๊ฒ€์ƒ‰ ํ…Œ์ŠคํŠธ: {response_time:.2f}์ดˆ")
test_results.append(f"๐Ÿ“ ์‘๋‹ต ๊ธธ์ด: {len(result)}์ž")
except Exception as e:
test_results.append(f"โŒ ๊ฒ€์ƒ‰ ํ…Œ์ŠคํŠธ ์‹คํŒจ: {str(e)}")
return "\n".join(test_results)
except Exception as e:
return f"ํ…Œ์ŠคํŠธ ์‹คํŒจ: {str(e)}"
def full_system_test():
"""์ „์ฒด ์‹œ์Šคํ…œ ํ…Œ์ŠคํŠธ"""
try:
output, results = gradio_test()
return output
except Exception as e:
return f"์ „์ฒด ํ…Œ์ŠคํŠธ ์‹คํŒจ: {str(e)}"
def sample_questions():
"""์ƒ˜ํ”Œ ์งˆ๋ฌธ ์ œ๊ณต"""
samples = [
"์ทจ๋“์„ธ์œจ์ด ์–ผ๋งˆ์ธ๊ฐ€์š”?",
"1์„ธ๋Œ€ 1์ฃผํƒ์ž ๊ฐ๋ฉด ํ˜œํƒ์€?",
"์‹ ํ˜ผ๋ถ€๋ถ€ ์ทจ๋“์„ธ ํŠน๋ก€๋Š”?",
"๋†์ง€ ์ทจ๋“์„ธ ๊ณ„์‚ฐ๋ฐฉ๋ฒ•์€?",
"๋‹ค์ฃผํƒ์ž ์ค‘๊ณผ์„ธ์œจ์€?",
"์ทจ๋“์„ธ ์‹ ๊ณ  ๊ธฐํ•œ์€ ์–ธ์ œ์ธ๊ฐ€์š”?",
"์ฆ์—ฌ๋ฐ›์€ ๋ถ€๋™์‚ฐ์˜ ์ทจ๋“์„ธ๋Š”?",
"์ƒ์† ์ฃผํƒ ์ทจ๋“์„ธ ์‹ ๊ณ ๊ธฐํ•œ์€?"
]
return "\n".join([f"{i+1}. {q}" for i, q in enumerate(samples)])
# Gradio ์ธํ„ฐํŽ˜์ด์Šค ๊ตฌ์„ฑ
def create_interface():
"""Gradio ์ธํ„ฐํŽ˜์ด์Šค ์ƒ์„ฑ"""
with gr.Blocks(title="์ทจ๋“์„ธ AI ์ƒ๋‹ด๋ด‡", theme=gr.themes.Soft()) as demo:
gr.Markdown("""
# ๐Ÿ›๏ธ ์ทจ๋“์„ธ AI ์ƒ๋‹ด๋ด‡
## ํ—ˆ๊น…ํŽ˜์ด์Šค ์ŠคํŽ˜์ด์Šค์—์„œ ์‹คํ–‰๋˜๋Š” RAG ๊ธฐ๋ฐ˜ ์„ธ๋ฌด ์ƒ๋‹ด ์‹œ์Šคํ…œ
""")
with gr.Tab("๐Ÿ’ฌ ์„ธ๋ฌด ์ƒ๋‹ด"):
with gr.Row():
with gr.Column(scale=3):
chatbot = gr.Chatbot(
value=[],
height=400,
label="์ทจ๋“์„ธ ์ƒ๋‹ด๋ด‡"
)
msg = gr.Textbox(
placeholder="์ทจ๋“์„ธ ๊ด€๋ จ ์งˆ๋ฌธ์„ ์ž…๋ ฅํ•˜์„ธ์š”...",
label="์งˆ๋ฌธ",
lines=2
)
with gr.Row():
submit_btn = gr.Button("์งˆ๋ฌธํ•˜๊ธฐ", variant="primary")
clear_btn = gr.Button("๋Œ€ํ™” ์ดˆ๊ธฐํ™”")
with gr.Column(scale=1):
gr.Markdown("### ๐Ÿ’ก ์ƒ˜ํ”Œ ์งˆ๋ฌธ")
samples_text = gr.Textbox(
value=sample_questions(),
lines=10,
label="ํด๋ฆญํ•˜์—ฌ ๋ณต์‚ฌ",
interactive=False
)
with gr.Tab("๐Ÿงช ์‹œ์Šคํ…œ ํ…Œ์ŠคํŠธ"):
with gr.Row():
with gr.Column():
gr.Markdown("### ๋น ๋ฅธ ํ…Œ์ŠคํŠธ")
quick_test_btn = gr.Button("๋น ๋ฅธ ์‹œ์Šคํ…œ ์ฒดํฌ", variant="secondary")
quick_result = gr.Textbox(
lines=10,
label="ํ…Œ์ŠคํŠธ ๊ฒฐ๊ณผ",
interactive=False
)
with gr.Column():
gr.Markdown("### ์ „์ฒด ํ…Œ์ŠคํŠธ")
full_test_btn = gr.Button("์ „์ฒด ์‹œ์Šคํ…œ ํ…Œ์ŠคํŠธ", variant="primary")
full_result = gr.Textbox(
lines=15,
label="์ƒ์„ธ ํ…Œ์ŠคํŠธ ๊ฒฐ๊ณผ",
interactive=False
)
with gr.Tab("๐Ÿ“Š ์‹œ์Šคํ…œ ์ •๋ณด"):
gr.Markdown(f"""
### ์‹œ์Šคํ…œ ์ •๋ณด
- **์‹คํ–‰ ํ™˜๊ฒฝ**: {'ํ—ˆ๊น…ํŽ˜์ด์Šค ์ŠคํŽ˜์ด์Šค' if os.getenv('SPACE_ID') else '๋กœ์ปฌ ํ™˜๊ฒฝ'}
- **Python ๋ฒ„์ „**: {sys.version}
- **RAG ์‹œ์Šคํ…œ**: {'์‚ฌ์šฉ๊ฐ€๋Šฅ' if RAG_AVAILABLE else '์‚ฌ์šฉ๋ถˆ๊ฐ€'}
- **์‹คํ–‰ ์‹œ๊ฐ„**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
### ์‚ฌ์šฉ๋ฒ•
1. **์„ธ๋ฌด ์ƒ๋‹ด ํƒญ**: ์ทจ๋“์„ธ ๊ด€๋ จ ์งˆ๋ฌธ์„ ์ž…๋ ฅํ•˜์„ธ์š”
2. **์‹œ์Šคํ…œ ํ…Œ์ŠคํŠธ ํƒญ**: ์‹œ์Šคํ…œ ์ƒํƒœ๋ฅผ ํ™•์ธํ•˜์„ธ์š”
3. ์ƒ˜ํ”Œ ์งˆ๋ฌธ์„ ์ฐธ๊ณ ํ•˜์—ฌ ๋‹ค์–‘ํ•œ ์ทจ๋“์„ธ ์ƒํ™ฉ์„ ๋ฌธ์˜ํ•˜์„ธ์š”
### ์ฃผ์˜์‚ฌํ•ญ
- ๋ณธ ์‹œ์Šคํ…œ์€ ์ฐธ๊ณ ์šฉ์œผ๋กœ๋งŒ ์‚ฌ์šฉํ•˜์„ธ์š”
- ์ •ํ™•ํ•œ ์„ธ๋ฌด ์ƒ๋‹ด์€ ์ „๋ฌธ๊ฐ€์™€ ์ƒ์˜ํ•˜์‹œ๊ธฐ ๋ฐ”๋ž๋‹ˆ๋‹ค
""")
# ์ด๋ฒคํŠธ ํ•ธ๋“ค๋Ÿฌ
def respond(message, chat_history):
bot_message = chat_with_tax_consultant(message, chat_history)
chat_history.append((message, bot_message))
return "", chat_history
# ๋ฒ„ํŠผ ์ด๋ฒคํŠธ ์—ฐ๊ฒฐ
msg.submit(respond, [msg, chatbot], [msg, chatbot])
submit_btn.click(respond, [msg, chatbot], [msg, chatbot])
clear_btn.click(lambda: [], None, chatbot)
quick_test_btn.click(quick_test, None, quick_result)
full_test_btn.click(full_system_test, None, full_result)
return demo
if __name__ == "__main__":
# ํ™˜๊ฒฝ์— ๋”ฐ๋ฅธ ์„ค์ •
space_id = os.getenv('SPACE_ID')
if space_id:
print(f"๐Ÿš€ ํ—ˆ๊น…ํŽ˜์ด์Šค ์ŠคํŽ˜์ด์Šค์—์„œ ์‹คํ–‰: {space_id}")
# ํ—ˆ๊น…ํŽ˜์ด์Šค ์ŠคํŽ˜์ด์Šค ํ™˜๊ฒฝ
demo = create_interface()
demo.launch(
server_name="0.0.0.0",
server_port=7860,
share=False
)
else:
print("๐Ÿ’ป ๋กœ์ปฌ ํ™˜๊ฒฝ์—์„œ ์‹คํ–‰")
# ๋กœ์ปฌ ๊ฐœ๋ฐœ ํ™˜๊ฒฝ
demo = create_interface()
demo.launch(
server_name="127.0.0.1",
server_port=7860,
share=True
)