Files
llm_gateway_chatbot/main.py
2025-07-18 17:32:56 +09:00

198 lines
7.1 KiB
Python

import json
import os
import google.generativeai as genai
import pandas as pd
import requests
import streamlit as st
# --- 설정 ---
SPEC_FILE = "static/api_spec.json"
LOGO_FILE = "static/logo.png"
API_SPEC_URL = "http://172.16.10.176:8888/openapi.json"
# --- 도우미 함수 ---
def load_api_spec():
"""로컬 파일에서 API 명세를 불러옵니다."""
try:
with open(SPEC_FILE, "r", encoding="utf-8") as f:
return json.load(f)
except (FileNotFoundError, json.JSONDecodeError) as e:
st.error(f"API 명세 파일을 불러오는 중 오류 발생: {e}")
return None
def fetch_and_save_spec():
"""URL에서 최신 API 명세를 가져와 로컬에 저장합니다."""
try:
response = requests.get(API_SPEC_URL)
response.raise_for_status() # 잘못된 상태 코드에 대해 예외를 발생시킵니다
spec_data = response.json()
with open(SPEC_FILE, "w", encoding="utf-8") as f:
json.dump(spec_data, f, indent=2, ensure_ascii=False)
st.success(f"성공적으로 API 명세를 가져와 업데이트했습니다: {API_SPEC_URL}")
return spec_data
except requests.exceptions.RequestException as e:
st.error(f"API 명세를 가져오는 데 실패했습니다: {e}")
return None
except json.JSONDecodeError:
st.error("응답에서 JSON을 파싱하는 데 실패했습니다.")
return None
# --- Gemini AI 설정 ---
try:
api_key = os.getenv("GOOGLE_API_KEY")
if not api_key:
st.error(
"GOOGLE_API_KEY 환경 변수가 설정되지 않았습니다. .env.chatbot 파일에서 설정해주세요."
)
st.stop()
genai.configure(api_key=api_key)
model = genai.GenerativeModel("gemini-1.5-flash")
except Exception as e:
st.error(f"Gemini AI 설정 실패: {e}")
st.stop()
# --- Streamlit UI 설정 ---
st.set_page_config(
page_title="API Guide Chatbot",
page_icon="🤖",
layout="wide",
initial_sidebar_state="expanded",
)
# --- 커스텀 CSS ---
st.markdown(
"""
<style>
/* 다크 모드 기본 테마와 유사하게 맞춤 */
.stApp {
background-color: #0E117;
}
/* 채팅 메시지 스타일 */
[data-testid="chat-message-container"] {
border-radius: 0.5rem;
padding: 1rem;
margin-bottom: 1rem;
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
}
</style>
""",
unsafe_allow_html=True,
)
# --- 메인 앱 로직 ---
# 시작 시 API 명세 불러오기
api_spec = load_api_spec()
# --- 사이드바 ---
with st.sidebar:
if os.path.exists(LOGO_FILE):
st.image(LOGO_FILE, use_container_width=True)
st.header("⚙️ Controls")
if st.button("🔄 API 명세 새로고침"):
with st.spinner("최신 API 명세를 가져오는 중..."):
api_spec = fetch_and_save_spec()
st.session_state.messages = [] # 새로고침 시 대화 기록 삭제
st.rerun()
if api_spec:
with st.expander("📘 API 상세정보 보기", expanded=True):
st.info(f"**Title:** {api_spec.get('info', {}).get('title', 'N/A')}")
st.text(f"Version: {api_spec.get('info', {}).get('version', 'N/A')}")
paths = api_spec.get("paths", {})
if paths:
endpoint_data = []
for path, methods in paths.items():
for method, details in methods.items():
endpoint_data.append(
{
"Method": method.upper(),
"Endpoint": path,
"summary": details.get("summary", "요약 없음"),
}
)
df = pd.DataFrame(endpoint_data)
st.dataframe(df, use_container_width=True)
else:
st.warning("API 명세를 불러올 수 없습니다.")
# --- 메인 채팅 인터페이스 ---
st.markdown(
"<h1 style='text-align: center;'>💬 LLM-Gateway Guide Chatbot</h1>",
unsafe_allow_html=True,
)
st.caption(f"현재 사용 중인 API 명세: {API_SPEC_URL}")
# 대화 기록 초기화
if "messages" not in st.session_state:
st.session_state.messages = []
# 시작 메시지
if not st.session_state.messages:
st.info("안녕하세요! LLM-Gateway 에 대해 궁금한 점을 물어보세요.")
# 앱 재실행 시 기록에서 대화 메시지 표시
for message in st.session_state.messages:
avatar = "🧑‍💻" if message["role"] == "user" else "👻"
with st.chat_message(message["role"], avatar=avatar):
st.markdown(message["content"])
# 사용자 입력 수락
if prompt := st.chat_input("LLM-Gateway 에 대해 질문하세요..."):
if not api_spec:
st.error("질문을 처리할 수 없습니다: API 명세가 로드되지 않았습니다.")
else:
# 대화 기록에 사용자 메시지 추가
st.session_state.messages.append({"role": "user", "content": prompt})
# 채팅 메시지 컨테이너에 사용자 메시지 표시
with st.chat_message("user", avatar="🧑‍💻"):
st.markdown(prompt)
# 채팅 메시지 컨테이너에 어시스턴트 응답 표시
with st.chat_message("assistant", avatar="👻"):
message_placeholder = st.empty()
with st.spinner("답변을 생성하는 중..."):
try:
# Gemini를 위한 프롬프트 준비
full_prompt = f"""
당신은 다음 LLM-Gateway API 명세에 대한 전문가 어시스턴트입니다.
당신의 임무는 제공된 JSON 데이터를 기반으로만 질문에 답변하는 것입니다.
정보를 지어내지 마세요. 만약 명세에 답변이 없다면 없다고 말하세요.
모든 답변은 반드시 한국어로 제공해주세요.
**LLM-Gateway API 명세 (JSON):**
```json
{json.dumps(api_spec, indent=2, ensure_ascii=False)}
```
**사용자 질문:**
{prompt}
"""
response = model.generate_content(full_prompt)
if response.parts:
response_text = response.text
else:
# 응답이 차단될 수 있는 경우를 처리합니다
response_text = (
"죄송합니다, 해당 질문에 대한 답변을 드릴 수 없습니다."
)
message_placeholder.markdown(response_text)
st.session_state.messages.append(
{"role": "assistant", "content": response_text}
)
except Exception as e:
error_text = f"오류가 발생했습니다: {e}"
message_placeholder.markdown(error_text)
st.session_state.messages.append(
{"role": "assistant", "content": error_text}
)