Streamlit과 LangGraph를 사용한 Front-end 구현
소개
이 글에서는 Streamlit과 LangGraph를 활용하여 대화형 프론트엔드를 구현하는 방법을 설명합니다. 이 프로젝트는 사용자의 음성을 텍스트로 변환하고, 이를 분석하여 적절한 응답을 제공하는 기능을 포함합니다.
주요 기술
- Streamlit: 대화형 웹 애플리케이션 개발
- LangGraph: LLM 기반 대화 흐름 관리
- Whisper: 음성-텍스트 변환
- PyTorch: 딥러닝 모델 실행
- Transformers: 자연어 처리 모델 활용
- ChatOllama: 대화형 AI 모델
- OllamaEmbeddings: 임베딩 생성
Streamlit 을 이용한 Front-end 구현
1. 환경 설정 및 모델 로드
import os
import torch
import json
import streamlit as st
from transformers import AutoTokenizer, AutoModelForSequenceClassification
from faster_whisper import WhisperModel
from myassistant_chatbot import MyAssistantChatbot
def load_whisper_model():
device = 'cuda' if torch.cuda.is_available() else 'cpu'
model = WhisperModel("large-v3", device=device, compute_type="float16")
return model
def load_intent_model(device):
tokenizer = AutoTokenizer.from_pretrained("./intent_model")
model = AutoModelForSequenceClassification.from_pretrained("./intent_model").to(device).eval()
return tokenizer, model
def init_chatbot():
return MyAssistantChatbot("data/vision_life_log.csv", "예전에 내가 했거나 과거에 한 일에 대한 질문이라면")
이 코드에서는 Whisper 모델과 자연어 처리 모델을 로드하며, 챗봇을 초기화합니다.
2. Streamlit UI 구성
st.set_page_config(page_title="My Best Assistant")
st.title("My Best Assistant")
st.markdown("##### 무엇이든 말씀하세요!")
if "chatbot" not in st.session_state:
st.session_state.chatbot = init_chatbot()
st.session_state.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
st.session_state.whisper_model = load_whisper_model()
st.session_state.tokenizer, st.session_state.model = load_intent_model(st.session_state.device)
if "messages" not in st.session_state:
st.session_state.messages = []
Streamlit을 활용하여 간단한 웹 UI를 구성합니다. 또한 세션 상태를 활용하여 챗봇과 모델을 유지합니다.
3. 사용자 입력 처리
prompt = st.chat_input("무엇이든 말씀하시면 친절히 답변드리겠습니다.")
if prompt:
with st.chat_message("user"):
st.markdown(prompt)
st.session_state.messages.append({"role": "user", "content": {"text": prompt}, "image": False})
사용자가 텍스트 입력을 하면 이를 세션에 저장하고 UI에 표시합니다.
4. 음성 입력 및 인식
voice = st.audio_input("")
recognized_text = ""
if voice:
raw_bytes = voice.read()
whisper_model = st.session_state.whisper_model
segments, _ = whisper_model.transcribe(raw_bytes, beam_size=5)
recognized_text = "".join(segment.text for segment in segments)
st.write(f"음성 인식 결과: {recognized_text}")
사용자가 음성을 입력하면 Whisper 모델을 활용해 텍스트로 변환하고 화면에 표시합니다.
5. 모델을 통한 의도 분석
def predict_intent(text, tokenizer, model, device):
inputs = tokenizer(text, return_tensors="pt", truncation=True).to(device)
with torch.no_grad():
logits = model(**inputs).logits
probs = torch.nn.functional.softmax(logits, dim=-1)[0]
max_prob, pred_id = torch.max(probs, dim=0)
return model.config.id2label[pred_id.item()] if max_prob.item() > 0.99 else "UNKNOWN"
if recognized_text:
label = predict_intent(recognized_text, st.session_state.tokenizer, st.session_state.model, st.session_state.device)
st.write(f"예측된 의도: {label}")
사용자의 음성을 텍스트로 변환한 후, Transformer 기반 의도 분류 모델을 활용하여 의미를 분석합니다.
6. 챗봇 응답 생성 및 UI 반영
if prompt or recognized_text:
input_text = prompt if prompt else recognized_text
response = st.session_state.chatbot.invoke(input_text)
generation = response["generation"]
with st.chat_message("assistant"):
st.markdown(generation)
st.session_state.messages.append({"role": "assistant", "content": {"text": generation}, "image": False})
사용자의 입력을 기반으로 챗봇이 응답을 생성하고 UI에 출력합니다.
LangGraph를 사용한 Front-end 챗봇 구현
1. 환경 설정 및 모델 로드
import os
import torch
import json
import streamlit as st
from transformers import AutoTokenizer, AutoModelForSequenceClassification
from faster_whisper import WhisperModel
from langchain_community.chat_models import ChatOllama
from langchain_community.embeddings import OllamaEmbeddings
from langgraph.graph import StateGraph, END
from myassistant_chatbot import MyAssistantChatbot
def load_whisper_model():
device = 'cuda' if torch.cuda.is_available() else 'cpu'
model = WhisperModel("large-v3", device=device, compute_type="float16")
return model
def load_intent_model(device):
tokenizer = AutoTokenizer.from_pretrained("./intent_model")
model = AutoModelForSequenceClassification.from_pretrained("./intent_model").to(device).eval()
return tokenizer, model
def init_chatbot():
return MyAssistantChatbot("data/vision_life_log.csv", "예전에 내가 했거나 과거에 한 일에 대한 질문이라면")
이 코드에서는 Whisper 모델과 자연어 처리 모델을 로드하며, 챗봇을 초기화합니다.
2. Streamlit UI 구성
st.set_page_config(page_title="My Best Assistant")
st.title("My Best Assistant")
st.markdown("##### 무엇이든 말씀하세요!")
if "chatbot" not in st.session_state:
st.session_state.chatbot = init_chatbot()
st.session_state.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
st.session_state.whisper_model = load_whisper_model()
st.session_state.tokenizer, st.session_state.model = load_intent_model(st.session_state.device)
if "messages" not in st.session_state:
st.session_state.messages = []
Streamlit을 활용하여 간단한 웹 UI를 구성합니다. 또한 세션 상태를 활용하여 챗봇과 모델을 유지합니다.
3. 사용자 입력 처리
prompt = st.chat_input("무엇이든 말씀하시면 친절히 답변드리겠습니다.")
if prompt:
with st.chat_message("user"):
st.markdown(prompt)
st.session_state.messages.append({"role": "user", "content": {"text": prompt}, "image": False})
사용자가 텍스트 입력을 하면 이를 세션에 저장하고 UI에 표시합니다.
4. 음성 입력 및 인식
voice = st.audio_input("")
recognized_text = ""
if voice:
raw_bytes = voice.read()
whisper_model = st.session_state.whisper_model
segments, _ = whisper_model.transcribe(raw_bytes, beam_size=5)
recognized_text = "".join(segment.text for segment in segments)
st.write(f"음성 인식 결과: {recognized_text}")
사용자가 음성을 입력하면 Whisper 모델을 활용해 텍스트로 변환하고 화면에 표시합니다.
5. 모델을 통한 의도 분석
def predict_intent(text, tokenizer, model, device):
inputs = tokenizer(text, return_tensors="pt", truncation=True).to(device)
with torch.no_grad():
logits = model(**inputs).logits
probs = torch.nn.functional.softmax(logits, dim=-1)[0]
max_prob, pred_id = torch.max(probs, dim=0)
return model.config.id2label[pred_id.item()] if max_prob.item() > 0.99 else "UNKNOWN"
if recognized_text:
label = predict_intent(recognized_text, st.session_state.tokenizer, st.session_state.model, st.session_state.device)
st.write(f"예측된 의도: {label}")
사용자의 음성을 텍스트로 변환한 후, Transformer 기반 의도 분류 모델을 활용하여 의미를 분석합니다.
6. LangGraph를 활용한 대화 흐름 관리
class State(TypedDict):
question: str
generation: str
data: List[str]
code: List[str]
graph = StateGraph(State)
graph.add_node("init_answer", chatbot.route_question)
graph.add_node("plain_answer", chatbot.plain_answer)
graph.add_node("lifelogging_answer", chatbot.lifelogging_answer)
graph.add_edge("plain_answer", END)
graph.add_edge("lifelogging_answer", END)
LangGraph를 활용하여 대화 흐름을 관리하며, 질문에 따라 적절한 응답을 제공합니다.
7. 챗봇 응답 생성 및 UI 반영
if prompt or recognized_text:
input_text = prompt if prompt else recognized_text
response = st.session_state.chatbot.invoke(input_text)
generation = response["generation"]
with st.chat_message("assistant"):
st.markdown(generation)
st.session_state.messages.append({"role": "assistant", "content": {"text": generation}, "image": False})
사용자의 입력을 기반으로 챗봇이 응답을 생성하고 UI에 출력합니다.
결론
위 코드는 Streamlit과 LangGraph를 활용하여 사용자 친화적인 챗봇을 구현하는 방법을 보여줍니다.
import os
import io
import torch
import tempfile
import numpy as np
from PIL import Image
import streamlit as st
from streamlit_extras.stylable_container import stylable_container
from myassistant_chatbot import MyAssistantChatbot
from pydub import AudioSegment
from faster_whisper import WhisperModel
from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch.nn.functional as F
import torch.nn as nn
import json
import cv2
st.set_page_config(page_title="My Best Assistant")
st.title("My Best Assistant")
st.markdown(
"""
##### 무엇이든 말씀하세요!
"""
)
lifelogging_path = "data/vision_life_log.csv"
lifelogging_description = "예전에 내가 했거나 과거에 한 일에 대한 질문이라면"
FASTER_WHISPER_MODEL_PATH = "large-v3"
INTENT_MODEL_DIR = "./intent_model"
JSON_PATH = "./intent_model/intent_map.json"
THRESHOLD = 0.99
EXP_TEMPERATURE = 0.0162 # 학습 후 최종 exp(temperature)= 1.0163 = ln(1.0163) ≈ 0.0162
class ModelWithTemperature(nn.Module):
# 사전 학습된 모델의 logits에 temperature를 곱해 확률을 보정하는 Wrapper 클래스
def __init__(self, base_model, init_temp=1.0):
super().__init__()
self.base_model = base_model
# 내부적으로는 'log scale' 파라미터 사용
# exp(self.temperature) 형태로 실제 temperature 계산
self.temperature = nn.Parameter(torch.ones(1) * init_temp)
def forward(self, input_ids=None, attention_mask=None, labels=None):
outputs = self.base_model(
input_ids=input_ids,
attention_mask=attention_mask,
labels=labels,
output_hidden_states=False,
return_dict=True
)
logits = outputs.logits
# 실제 temperature = exp(self.temperature)
temp = torch.exp(self.temperature)
scaled_logits = logits / temp
return scaled_logits
@st.cache_resource
def load_intent_map(json_file_path):
# JSON 로드
try:
with open(json_file_path, "r", encoding="utf-8") as f:
intent_json_map = json.load(f)
return intent_json_map
except FileNotFoundError:
print(f"__debug JSON file not found at: {json_file_path}")
return {}
except json.JSONDecodeError as e:
print(f"__debug Error decoding JSON: {e}")
return {}
intent_json_map = load_intent_map(JSON_PATH)
@st.cache_resource
def load_whisper_model():
# Whisper 모델 로드
device = "cuda" if torch.cuda.is_available() else "cpu"
model = WhisperModel(FASTER_WHISPER_MODEL_PATH, device=device, compute_type="float16")
return model
@st.cache_resource
def load_intent_model_with_temp(device):
# Intent 모델 + Temperature Scaling 초기화
# Base classifier 로드
tokenizer = AutoTokenizer.from_pretrained(INTENT_MODEL_DIR)
base_classifier = AutoModelForSequenceClassification.from_pretrained(INTENT_MODEL_DIR)
base_classifier.eval()
base_classifier.to(device)
# ModelWithTemperature
model_temp = ModelWithTemperature(base_model=base_classifier)
model_temp.temperature.data = torch.tensor([EXP_TEMPERATURE]).to(device)
model_temp.eval()
model_temp.to(device)
return tokenizer, model_temp
def predict_with_temperature_single(
text,
tokenizer,
model_temp,
device,
threshold=THRESHOLD
):
if not text.strip():
return None
# 토크나이징
inputs = tokenizer(text, return_tensors="pt", truncation=True).to(device)
print(f"__debug Input Text: {text}")
with torch.no_grad():
# scaled logits 계산
scaled_logits = model_temp(
input_ids=inputs["input_ids"],
attention_mask=inputs["attention_mask"]
)
# softmax 확률 계산
probabilities = F.softmax(scaled_logits, dim=-1)[0] # shape: [num_labels]
probs_list = probabilities.tolist()
# 최댓값 및 인덱스 파악
max_prob = max(probs_list)
pred_id = probs_list.index(max_prob)
print("__debug Softmax Prob:", [round(p, 4) for p in probs_list])
print(f"__debug Max Probability: {max_prob:.4f}, Index: {pred_id}, Threshold: {threshold}")
# Threshold 처리(UNKNOWN)
if max_prob < threshold:
print("__debug Prediction under threshold\n")
return None
else:
label = model_temp.base_model.config.id2label[pred_id]
print(f"__debug Predicted Label: {label}\n")
return label
@st.cache_resource
def init_chatbot():
chatbot = MyAssistantChatbot(
lifelogging_path,
lifelogging_description,
)
return chatbot
# 챗봇 초기화
if "chatbot" not in st.session_state:
with st.spinner():
chatbot = init_chatbot()
st.session_state.chatbot = chatbot
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
st.session_state.device = device
whisper_model = load_whisper_model()
st.session_state.whisper_model = whisper_model
tokenizer, model_with_temp = load_intent_model_with_temp(device)
st.session_state.tokenizer = tokenizer
st.session_state.model_with_temp = model_with_temp
# 메세지 초기화
if "messages" not in st.session_state:
st.session_state.messages = []
# 이미지 상태를 저장할 변수
# if 'image' not in st.session_state:
# st.session_state.image = None
# 메세지 초기화
# if "messages" not in st.session_state:
# st.session_state.youtube = []
for conversation in st.session_state.messages:
# print("conversation[\"role\"]")
# print(conversation["role"])
with st.chat_message(conversation["role"]):
print(f"demo.py/for conversation in st.session_state.messages {conversation}")
st.markdown(conversation["content"]["text"])
# st.markdown(conversation["content"]) # steven.kim
if "image" in conversation.keys() and conversation["image"]:
st.image(conversation["content"]["image"])
# 사용자 입력 처리
if prompt := st.chat_input("무엇이든 말씀하시면 친절히 답변드리겠습니다."):
with st.chat_message("user"):
st.markdown(prompt)
print(f"demo.py/if prompt := st.chat_input(): with st.chat_message(user): {prompt}")
st.session_state.messages.append({"role": "user", "content": {"text": prompt}, "image": False})
# st.session_state.messages.append({"role": "user", "content": prompt}) # steven.kim
st.markdown("")
st.markdown("")
st.markdown("")
st.markdown("")
recognized_text = ""
label = ""
with stylable_container(
key="bottom_content",
css_styles="""
{
position: fixed;
bottom: 110px;
width: 100%;
display: flex;
justify-content: center; /* 중앙 정렬 */
z-index: 10; /* 우선 순위 설정 */
}
""",
):
print("with stylable_container():")
voice = st.audio_input("")
if voice is not None and prompt is None:
print("with stylable_container():if voice is not None:")
# 16kHz 모노 변환
raw_bytes = voice.read()
original_audio = AudioSegment.from_file(io.BytesIO(raw_bytes), format="wav")
converted_audio = original_audio.set_frame_rate(16000).set_channels(1)
# 임시 wav 파일로 저장
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_wav:
converted_audio.export(temp_wav.name, format="wav")
temp_wav.flush()
temp_path = temp_wav.name
# 음성 인식
whisper_model = st.session_state.whisper_model
segments, info = whisper_model.transcribe(temp_path, beam_size=5)
recognized_text = "".join(segment.text for segment in segments)
# 결과 표시
st.write(f"음성 인식 결과: {recognized_text}")
print(f"with stylable_container():if voice is not None: 음성 인식 결과: {recognized_text}")
os.remove(temp_path)
if recognized_text.strip():
label = predict_with_temperature_single(
text=recognized_text,
tokenizer=st.session_state.tokenizer,
model_temp=st.session_state.model_with_temp,
device=st.session_state.device,
threshold=THRESHOLD
)
if not label:
label_none = "UNKNOWN"
st.write(f"Predicted Label: {label_none}")
print(f"demo.py/if user_text.strip():if not label: {recognized_text}")
prompt = recognized_text
else:
st.write(f"Predicted Label: {label}")
print(f"demo.py/if user_text.strip():else: {label}")
if label == "ACTIVATE_YOUTUBE":
print(f"demo.py/if user_text.strip():if label == ACTIVATE_YOUTUBE: {label}")
else:
print(f"demo.py/if user_text.strip():if label is not ACTIVATE_YOUTUBE: {label}")
### 여기부터 1
if voice is not None and recognized_text is not None:
with st.chat_message("user"):
st.markdown(recognized_text)
print(f"demo.py/if voice is not None: with st.chat_message(user):{recognized_text}")
st.session_state.messages.append({"role": "user", "content": {"text": recognized_text}, "image": False})
print(f"voice and label: {voice}, {label}")
if voice is not None and label == "ACTIVATE_YOUTUBE":
print(f"demo.py/if voice is not None and label == ACTIVATE_YOUTUBE: {label}")
st.video("https://youtu.be/mCyY8pQDpJM?si=sd324t9fTg9EDZBM")
else voice is not None and label == "ACTIVATE_MUSIC":
print(f"demo.py/if voice is not None and label == ACTIVATE_MUSIC: {label}")
st.audio("./answer_me.mp3", format='audio/mp3')
label = ""
voice = None
### 여기까지 1
# 챗봇 응답 처리
if prompt is not None:
# st.session_state.image = None #steven.kim
# st.session_state.youtube = None
response = st.session_state.chatbot.invoke(prompt)
generation = response["generation"]
# prompt = None
with st.chat_message("assistant"):
if "data" in response.keys() and "lifelogging" in response["data"]:
print("demo.py/response: lifelogging in response.keys()")
print(response["data"][0])
st.markdown(generation)
image = np.array(Image.open("lifelog.png")) # 이미지 경로 받아서 넣거나, 이미지 이름에 맞게 생성 요청
st.image(image)
st.session_state.messages.append({"role": "assistant", "content": {"text": generation, "image": image}, "image": True})
os.remove("lifelog.png")
elif "data" in response.keys() and len(response["data"]) > 0: # 20250225_myassistant_pics
print("demo.py/response: data in response.keys()")
print(response["data"][0])
if "png" in response["data"][0] or "jpg" in response["data"][0] or "webp" in response["data"][0]:
print("demo.py/response: png jpg webp in data")
st.markdown(generation)
st.session_state.image = response["data"][0]
if st.session_state.image is not None:
st.image(st.session_state.image, caption=response["data"][0], use_container_width =True)
st.session_state.image = None
response["data"][0] = None
elif "youtu.be" in response["data"][0]:
print("demo.py/response: youtube in data")
print(response["data"][0])
st.markdown(generation)
st.session_state.youtube = response["data"][0]
if st.session_state.youtube is not None:
st.video(st.session_state.youtube)
st.session_state.messages.append({"role": "assistant", "content": {"text": generation}, "image": False})
st.session_state.youtube = None
else:
print(f"demo.py/response: else: {generation}")
st.markdown(generation)
print(f"demo.py/response: else: st.markdown{generation}")
st.session_state.messages.append({"role": "assistant", "content": {"text": generation}, "image": False})
# CSS 스타일을 추가하여 대화 메시지 영역의 높이를 조정
st.markdown("""
<style>
.stMainBlockContainer {
margin-bottom: 50px; /* 오디오 입력 필드와 겹치지 않도록 하단 여백 추가 */
}
</style>
""", unsafe_allow_html=True)
from typing import List, Optional, TypedDict
from langchain_community.chat_models import ChatOllama
from langchain_community.embeddings import OllamaEmbeddings
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langgraph.graph import END, StateGraph
from utils import *
from myassistant_lifelogging import MyAssistantLifeLogging
from myassistant_pics import MyAssistantPics # 20250225_myassistant_pics
from myassistant_youtube import MyAssistantYoutube # 20250227_myassistant_youtube
# 그래프 상태 속성 정의
class State(TypedDict):
question: str
generation: str
data: List[str]
code: List[str]
class MyAssistantChatbot:
def __init__(
self,
lifelogging_path: Optional[str] = None,
lifelogging_description: Optional[str] = None,
) -> None:
self.llm = ChatOllama(model="exaone3.5:32b") # exaone3.5:2.4b 1.6GB; exaone3.5:7.8b 4.8GB; exaone3.5:32b 19GB
self.route_llm = ChatOllama(model="exaone3.5:32b", format="json", temperature=0)
self.embeddings = OllamaEmbeddings(model="exaone3.5:32b")
self.lifelogging_path = lifelogging_path
self.lifelogging_description = lifelogging_description
# 그래프 구성
self.graph = StateGraph(State)
self.graph.add_node("init_answer", self.route_question)
self.graph.add_node("youtube_answer", self.youtube_answer) # 20250227_myassistant_youtube
self.graph.add_node("plain_answer", self.plain_answer)
self.graph.add_node("lifelogging_answer", self.lifelogging_answer)
self.graph.add_node("pics_answer", self.pics_answer) # 20250225_myassistant_pics
self.graph.set_entry_point("init_answer")
# 간선을 정의합니다.
# self.graph.add_edge("rag", "answer_with_retrieval")
# END는 종결 지점을 의미합니다.
self.graph.add_edge(
"plain_answer", END
) # self.graph.set_finish_point("answer")와 동일합니다.
self.graph.add_edge( # 20250225_myassistant_pics
"pics_answer", END
)
self.graph.add_edge( # 20250227_myassistant_youtube
"youtube_answer", END
)
# self.graph.add_edge("answer_with_data", END)
self.graph.add_conditional_edges(
"init_answer",
self.extract_route,
{
"plain_answer": "plain_answer",
"lifelogging_answer": "lifelogging_answer",
"pics_answer": "pics_answer", # 20250225_myassistant_pics
"youtube_answer": "youtube_answer", # 20250227_myassistant_youtube
},
)
self.graph = self.graph.compile()
def invoke(self, question) -> str:
self.hallucination_count = 0
answer = self.graph.invoke({"question": question})
print("===생성 종료===")
return answer
# 라우팅된 질문 추출
def extract_route(self, state: State) -> str:
return state["generation"]
# 질문 라우팅
def route_question(self, state: State):
print("route_question() ---질문 라우팅---")
route_system_message = "You are the best AI Assistant.\n"
usable_tools_list = ["`plain_answer`"]
if self.lifelogging_path is not None:
route_system_message += f"{self.lifelogging_description} lifelogging_answer를 활용하세요.\n"
usable_tools_list.append("`lifelogging_answer`")
route_system_message += (
f"질문에 http로 시작되는 Image URL이 포함되어 있으면 pics_answer를 사용하세요. \n"
)
usable_tools_list.extend(["`pics_answer`"])
route_system_message += (
f"질문에 http로 시작되는 Youtube URL이 포함되어 있으면 youtube_answer 사용하세요. \n"
)
usable_tools_list.extend(["`youtube_answer`"])
route_system_message += "그 외의 질문이라면 plain_answer로 충분합니다.\n"
usable_tools_text = ", ".join(usable_tools_list)
route_system_message += f"주어진 질문에 맞춰 {usable_tools_text} 중 하나를 선택하세요.\n"
route_system_message += "답변은 `route`, `reason` 두개의 key 값을 가지는 JSON으로 답변하고, " + \
"route에는 선택한 종류, reason에는 판단한 이유를 적되, 반드시 reason 키를 먼저 작성하세요."
route_user_message = "{question}"
route_prompt = ChatPromptTemplate.from_messages(
[("system", route_system_message), ("human", route_user_message)]
)
print(f"route_question()/route_prompt: {route_prompt}")
router_chain = route_prompt | self.route_llm | JsonOutputParser()
print(f"route_question()/state: {state}")
# 현재 날짜와 시간
dt_info = DateTimeInfo()
current_datetime_text = dt_info.get_current_datetime_text()
output = router_chain.invoke({"question": state["question"]})
print(f"route_question()/output: {output}")
route = output["route"]
return {
"question": state["question"],
"generation": route.lower().strip(),
"code": [],
"data": [],
}
def plain_answer(self, state: State):
"""
데이터를 쿼리하지 않고 답변을 바로 생성합니다.
Args:
state (dict): 현재 그래프 상태
Returns:
state (dict): LLM의 답변을 포함한 새로운 State
"""
print("---답변 생성---") # 현재 상태를 확인하기 위한 Print문
question = state["question"]
dt_info = DateTimeInfo() # 현재 날짜와 시간
current_datetime_text = dt_info.get_current_datetime_text()
state["question"] += f"\n 오늘 날짜와 시간은 {current_datetime_text} 입니다. \n" # 20250225_myassistant_pics
return {
"question": question,
"generation": self.llm.invoke(question).content,
"data": [],
"code": [],
}
def lifelogging_answer(self, state: State):
print("lifelogging_answer() ---답변 생성---")
question = state["question"]
lifelogginganswer = MyAssistantLifeLogging(
self.lifelogging_path,
self.lifelogging_description,
)
return {
"question": question,
"generation": lifelogginganswer.lifelogging_answer(state),
"data": ["lifelogging"],
"code": [],
}
# 20250225_myassistant_pics
def pics_answer(self, state: State):
"""
# llava-hf/llava-onevision-qwen2-7b-ov-chat-hf
# https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/transformers/tasks/ai2d-demo.jpg
Args:
state (dict): 현재 그래프 상태
Returns:
state (dict): LLM의 답변을 포함한 새로운 State
"""
print("---그림 URL 답변 생성---")
question = state["question"]
picsanswer = MyAssistantPics(
self.lifelogging_path,
self.lifelogging_description,
)
return {
"question": question,
"generation": picsanswer.pics_answer(state)["generation"],
"data": state["data"],
"code": [],
}
# 20250227_myassistant_youtube
def youtube_answer(self, state: State):
"""
Args:
state (dict): 현재 그래프 상태
Returns:
state (dict): LLM의 답변을 포함한 새로운 State
"""
print("---Youtube URL 요약 생성---")
question = state["question"]
youtubeanswer = MyAssistantYoutube(
self.lifelogging_path,
self.lifelogging_description,
)
return {
"question": question,
"generation": youtubeanswer.youtube_answer(state)["generation"],
"data": state["data"],
"code": [],
}
class DateTimeInfo:
def __init__(self):
pass
@staticmethod
def get_current_datetime_text():
# 현재 날짜와 시간을 가져옵니다.
current_datetime = datetime.now()
# 현재 날짜와 시간을 텍스트로 리턴합니다.
return current_datetime.strftime("%Y년 %m월 %d일 %H시 %M분")
# # 사용 예시
# if __name__ == "__main__":
# dt_info = DateTimeInfo()
# current_datetime_text = dt_info.get_current_datetime_text()
# print(current_datetime_text)
0 댓글