이제 이전에 했던 ai 식사 추천 챗봇에 langgraph와 rag 기능을 추가하여 다시 리팩토링하며 langgraph에 대해 복습한다.
server.py
from fastapi import FastAPI
from pydantic import BaseModel
from dotenv import load_dotenv
from langchain_core.messages import HumanMessage
from agent_with_graph import 랭그래프객체
app = FastAPI(title='식사 메뉴 추천 에이전트')
class UserRequest(BaseModel):
question:str
@app.post('/chat')
async def llm_endpoint(req:UserRequest):
try:
prompt = {'messages':[HumanMessage(content=req.question)]}
# 재시도 컨셉 =>
config = {"recursion_limit":5} # 최초 요청에 실패하면 최대 5회까지 다시 시도함
final_state = 랭그래프객체.invoke(prompt,config=config)
res = final_state['messages'][-1].content
return {"response":res}
except Exception as e:
return{'response':f'오류 발생{str(e)}'}최소한으로 필요한 fastapi 엔드포인트 기능만 남겨둔 후 기존에 있던거 싹 날린다, 다 노드화 하고 툴로 바꿔서 langgraph로 묶을거다,
agent_with_graph.py
전체코드
'''
langgraph 기반 LLM 사용에 대한 모듈
'''
from dotenv import load_dotenv
import os
load_dotenv()
from typing import TypedDict, List
from langgraph.graph import StateGraph, END, MessagesState, START
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate
from langchain_core.messages import HumanMessage, BaseMessage
from langchain_aws import ChatBedrockConverse, ChatBedrock
from langgraph.prebuilt import ToolNode, tools_condition
from tools import rag_search
# 1. LLM 모델 구성, client 생략
llm = ChatBedrockConverse(model = os.getenv('BEDROCK_MODEL_ID'),
region_name = os.getenv('AWS_REGION'),
temperature = 0.5,
max_tokens = 1000)
# 2. 외부 도구 가져오기 및 LLM에 등록
tools = [rag_search]
llm_with_tools = llm.bind_tools(tools)
# 3. fewshot 프롬프트 -> 참고용
examples = [
{"input":"비 오는 날 국물이 땡겨" ,"output":"국룰이죠, 칼국수와 잔치국수가 좋습니다."},
{"input":"다이어트를 위해 오늘 칼로리 낮은 것으로 ","output":"관리하시는군요, 닭가슴, 샐러드 드세요"},
]
example_format = ChatPromptTemplate.from_messages([
('human',"{input}"),
('ai',"{output}")
])
# 3. FewShot 템플릿 구성
few_shot_prompt = FewShotChatMessagePromptTemplate(
examples = examples,
example_prompt = example_format
)
# 4. 시스템 프롬프트
final_prompt = ChatPromptTemplate.from_messages([
# 페르소나 지정
# 도구 사용 => 현재는 rag => 임시편성으로 비어있음 => rag에는 LLM이 모르는 식당 정보가 준비되어있음.
('system','당신의 센스있는 식사 메뉴 추천 전문가입니다, 사용자의 상황에 맞춰서 메뉴를 추천하고, 필요하면 도구를 사용하여 식당을 찾으세요'),
# 퓨샷
few_shot_prompt,
# 사용자 입력
('human','{messages}')
])
# 5. 랭그래프 상태
class AgentState(TypedDict):
messages:List[BaseMessage]
# 6. 노드 정의
# 6-1. 사용자의 질의(말)을 듣고 생각하는 단계 구성 (메뉴 추천 + 도구 사용 결정)
def thinking_node(state:AgentState):
# 6-1-1. 현재 상태의 프롬프트 실제 내용 획득 (페르소나 + 퓨샷 + 사용자 질의)
messages = state['messages']
# 6-1-2. 체인 구성 (prompt + LLM) -> 랭그래프의 특정 노드에 랭체인 결합되어 있는 구조
chain = final_prompt | llm_with_tools
# 6-1-3. LLM에게 질문에 대한 요청
res = chain.invoke({'messages':messages})
return {'messages':[ res ]}
# 6-2. LLM이 도구 사용을 결정했다면 -> 실제로 도구 사용 - 간단한 MCP 개념 - RAG
def tool_node(state:AgentState):
# 툴 사용 -> rag 이용한 검색증강
last_msg = state['messages'][-1]
# 툴 사용 체크
print('last_msg.tool_calls',last_msg.tool_calls)
if last_msg.tool_calls: # 없으면 []
tool = last_msg.tool_calls[0] # 등록된 도구가 1개 -> 인덱스 번호 0번
# 사내 데이터 검색, 판례 검색
tool_output = rag_search.invoke( tool['args'] ) # 검색 증강 (벡터 디비 검색=> 유사도 1개 획득(가게데이터) -> 응답)
# 해결 결과를 프롬프트에 추가
return {'messages':[
HumanMessage(content=f'[사내 데이터 검색 결과]: {tool_output}\n 제공된 정보를 기반으로 최종 답변을 해주세요 ')
]}
# 6-3. 검색의 결과를 바탕으로 최종 답변 (추론) 생성
def final_answer_node(state:AgentState):
# 최종 프롬프트 획득 (기존 주고받은 내용 + 검색 증강 내용)
final_msg = state['messages']
print('final_msg', final_msg )
# LLM 질의 -> tool 필요 없음
res = llm.invoke(final_msg)
return {'messages':[ res ]}
# 7. 랭그래프 연결
workflow = StateGraph(AgentState) # 에이전트 상태 그래프 연동
workflow.add_node("thinking",thinking_node)
workflow.add_node("tools",tool_node)
workflow.add_node("final_answer",final_answer_node)
workflow.set_entry_point('thinking') # 사용자 질의 후 최초 invoke 가 진입할 노드
# 핵심 : 에이전트가 상태값을 유지하면서 LLM에서 호출로 마무리 할지? 아니면 도구를 써서 마무리 할지 판단(LLM한테)
# 에이전트의 역할 : 점심/저녁 등 식사 메뉴를 추천하는 기능
def check_tool_node(state:AgentState): # 도구 사용 여부 체크(초대형 LLM들은 도구 사용 필요 거의 없음)
# LLM의 마지막 응답 결과 추출 -> 대화 내용중 마지막 내용은 LLM이 대답한것임
last_msg = state['messages'][-1]
print("1차 노드 수행 후 LLM의 응답값 : ", last_msg)
print("1차 노드 수행 후 도구 사용 여부 : ", last_msg.tool_calls)
# 대답의 내용 구조 중 툴에 대한 언급, 표현 등이 있는지 체크
# llm.bind_tools(tools) 사용으로 인해서 생기는 표현 -> 스스로 납득하지 못할 값이 나오면 쓰라는 그 말임 -> 특정 표식을 하게됨
if last_msg.tool_calls: # 문자열로 툴에 대한 값이 세팅됨
return 'tools' # 커스텀 지정값(노드의 이름 지정) -> 툴 노드로 가라는
return END # LLM의 답변으로 충분하다. 도구 사용 X, 대화를 마무리한다.
pass
workflow.add_conditional_edges("thinking", check_tool_node) # 조건부 엣지
workflow.add_edge("tools", 'final_answer')
workflow.add_edge('final_answer', END)
# 8. 랭그래프 컴파일 -> 워크플로우 객체
# 랭그래프객체 => 전역변수
랭그래프객체 = workflow.compile()
먼저 llm 모델을 구성한다.
그 이후 외부 도구를 가져와서 llm에 '너 이 툴 써도 돼' 라고 선언해준다.
이 툴은 tools.py라는 파일에 있다. 그래서 모듈에서 마치 라이브러리 가져오는것처럼
from tools import rag_search 코드가 있는거다.
tools.py
'''
각종 툴을 모은 모듈
'''
from langchain_core.tools import tool
from rag_store import search_stores
@tool
def rag_search(cate : str) -> str:
'''
특정 메뉴 카테고리 입력받아서 -> RAG 이용 -> 유사도 검색 -> 실제 식당정보등 반환
'''
# RAG 검색
res = search_stores(cate) # 기본값은 2개 설정
return res if res else "관련 식당 정보를 찾을 수 없습니다."
tools.py는 rag_store.py를 참조 한다.
rag_store.py
'''
- RAG 기반 검색기능 제공
- FAISS에 등록된 식당/메뉴/가격 정보를 벡터형태로 제공
- 유사도 검색 기능 기본 제공
- LLM이 사용자질의를 보고 직접 답변이 불가할 경우 -> 도구 사용 요청 -> RAG를 통해서 검색 -> 내용 제공
'''
from langchain_community.vectorstores import FAISS
from langchain_aws import BedrockEmbeddings
import boto3
import os
from dotenv import load_dotenv
import os
load_dotenv()
# 1. BedrockEmbeddings 객체 생성
tokenizer = BedrockEmbeddings(
client = boto3.client(service_name='bedrock-runtime', region_name=os.getenv('AWS_REGION')),
model_id = 'amazon.titan-embed-text-v1'
)
# 2. 더미 음식점 데이터(차후, 실 데이터로 교체)
data = [
"가게명: 스파이시 웍, 메뉴: 마라탕, 꿔바로우, 특징: 아주 매움, 스트레스 풀림, 가격: 15000원",
"가게명: 헬시 샐러드, 메뉴: 닭가슴살 샐러드, 샌드위치, 특징: 다이어트, 가벼움, 신선함, 가격: 9000원",
"가게명: 엄마손 백반, 메뉴: 김치찌개, 제육볶음, 특징: 집밥 스타일, 가성비, 든든함, 가격: 8000원",
"가게명: 골든 스시, 메뉴: 초밥 세트, 우동, 특징: 고급스러움, 깔끔함, 월급날 추천, 가격: 25000원",
"가게명: 해장국 천국, 메뉴: 뼈해장국, 순대국, 특징: 국물 진함, 비 오는 날 추천, 가격: 10000원"
]
# 3. 데이터 -> 벡터화 -> FAISS에 저장
vector_db = FAISS.from_texts(data, embedding=tokenizer)
# 4. 함수로 제공(다른데서도 써야함) : 질의 -> 검색 -> 유사도 순으로 후보 k(1개의 문장으로 구성)개 반환
def search_stores(query:str, k:int = 2 ):
docs = vector_db.similarity_search(query, k)
return "\n".join( [ doc.page_content for doc in docs ] )
이 파일이 사실상 툴의 기능을 하는것이다.
이 파일에서는 llm을 불러오지 않고 "오로지"rag의 기능만 사용한다.
일단 데이터가 없기 때문에 임베딩 모델을 생성하고, 더미로 음식점을 추천하는 데이터만 넣어둔다.
그리고 데이터를 백터화해서 faiss에 저장하고, 함수로 제공한다, 다른데에서도 가져와서 써야하니까
다시 tools.py를 보면 rag_store 에서 search_store를 가져와서 rag_search라는 함수를 정의했다.
그리고 여기서 카테고리를 입력받아서 rag를 이용해서 검색하고, 실제로 식당 정보를 반환한다.
다시 agent_woth_graph로 돌아오면
'''
langgraph 기반 LLM 사용에 대한 모듈
'''
from dotenv import load_dotenv
import os
load_dotenv()
from typing import TypedDict, List
from langgraph.graph import StateGraph, END, MessagesState, START
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate
from langchain_core.messages import HumanMessage, BaseMessage
from langchain_aws import ChatBedrockConverse, ChatBedrock
from langgraph.prebuilt import ToolNode, tools_condition
from tools import rag_search
# 1. LLM 모델 구성, client 생략
llm = ChatBedrockConverse(model = os.getenv('BEDROCK_MODEL_ID'),
region_name = os.getenv('AWS_REGION'),
temperature = 0.5,
max_tokens = 1000)
# 2. 외부 도구 가져오기 및 LLM에 등록
tools = [rag_search]
llm_with_tools = llm.bind_tools(tools)
# 3. fewshot 프롬프트 -> 참고용
examples = [
{"input":"비 오는 날 국물이 땡겨" ,"output":"국룰이죠, 칼국수와 잔치국수가 좋습니다."},
{"input":"다이어트를 위해 오늘 칼로리 낮은 것으로 ","output":"관리하시는군요, 닭가슴, 샐러드 드세요"},
]
example_format = ChatPromptTemplate.from_messages([
('human',"{input}"),
('ai',"{output}")
])
# 3. FewShot 템플릿 구성
few_shot_prompt = FewShotChatMessagePromptTemplate(
examples = examples,
example_prompt = example_format
)
# 4. 시스템 프롬프트
final_prompt = ChatPromptTemplate.from_messages([
# 페르소나 지정
# 도구 사용 => 현재는 rag => 임시편성으로 비어있음 => rag에는 LLM이 모르는 식당 정보가 준비되어있음.
('system','당신의 센스있는 식사 메뉴 추천 전문가입니다, 사용자의 상황에 맞춰서 메뉴를 추천하고, 필요하면 도구를 사용하여 식당을 찾으세요'),
# 퓨샷
few_shot_prompt,
# 사용자 입력
('human','{messages}')
])
# 5. 랭그래프 상태
class AgentState(TypedDict):
messages:List[BaseMessage]
# 6. 노드 정의
# 6-1. 사용자의 질의(말)을 듣고 생각하는 단계 구성 (메뉴 추천 + 도구 사용 결정)
def thinking_node(state:AgentState):
# 6-1-1. 현재 상태의 프롬프트 실제 내용 획득 (페르소나 + 퓨샷 + 사용자 질의)
messages = state['messages']
# 6-1-2. 체인 구성 (prompt + LLM) -> 랭그래프의 특정 노드에 랭체인 결합되어 있는 구조
chain = final_prompt | llm_with_tools
# 6-1-3. LLM에게 질문에 대한 요청
res = chain.invoke({'messages':messages})
return {'messages':[ res ]}
# 6-2. LLM이 도구 사용을 결정했다면 -> 실제로 도구 사용 - 간단한 MCP 개념 - RAG
def tool_node(state:AgentState):
# 툴 사용 -> rag 이용한 검색증강
last_msg = state['messages'][-1]
# 툴 사용 체크
print('last_msg.tool_calls',last_msg.tool_calls)
if last_msg.tool_calls: # 없으면 []
tool = last_msg.tool_calls[0] # 등록된 도구가 1개 -> 인덱스 번호 0번
# 사내 데이터 검색, 판례 검색
tool_output = rag_search.invoke( tool['args'] ) # 검색 증강 (벡터 디비 검색=> 유사도 1개 획득(가게데이터) -> 응답)
# 해결 결과를 프롬프트에 추가
return {'messages':[
HumanMessage(content=f'[사내 데이터 검색 결과]: {tool_output}\n 제공된 정보를 기반으로 최종 답변을 해주세요 ')
]}
# 6-3. 검색의 결과를 바탕으로 최종 답변 (추론) 생성
def final_answer_node(state:AgentState):
# 최종 프롬프트 획득 (기존 주고받은 내용 + 검색 증강 내용)
final_msg = state['messages']
print('final_msg', final_msg )
# LLM 질의 -> tool 필요 없음
res = llm.invoke(final_msg)
return {'messages':[ res ]}
# 7. 랭그래프 연결
workflow = StateGraph(AgentState) # 에이전트 상태 그래프 연동
workflow.add_node("thinking",thinking_node)
workflow.add_node("tools",tool_node)
workflow.add_node("final_answer",final_answer_node)
workflow.set_entry_point('thinking') # 사용자 질의 후 최초 invoke 가 진입할 노드
# 핵심 : 에이전트가 상태값을 유지하면서 LLM에서 호출로 마무리 할지? 아니면 도구를 써서 마무리 할지 판단(LLM한테)
# 에이전트의 역할 : 점심/저녁 등 식사 메뉴를 추천하는 기능
def check_tool_node(state:AgentState): # 도구 사용 여부 체크(초대형 LLM들은 도구 사용 필요 거의 없음)
# LLM의 마지막 응답 결과 추출 -> 대화 내용중 마지막 내용은 LLM이 대답한것임
last_msg = state['messages'][-1]
print("1차 노드 수행 후 LLM의 응답값 : ", last_msg)
print("1차 노드 수행 후 도구 사용 여부 : ", last_msg.tool_calls)
# 대답의 내용 구조 중 툴에 대한 언급, 표현 등이 있는지 체크
# llm.bind_tools(tools) 사용으로 인해서 생기는 표현 -> 스스로 납득하지 못할 값이 나오면 쓰라는 그 말임 -> 특정 표식을 하게됨
if last_msg.tool_calls: # 문자열로 툴에 대한 값이 세팅됨
return 'tools' # 커스텀 지정값(노드의 이름 지정) -> 툴 노드로 가라는
return END # LLM의 답변으로 충분하다. 도구 사용 X, 대화를 마무리한다.
pass
workflow.add_conditional_edges("thinking", check_tool_node) # 조건부 엣지
workflow.add_edge("tools", 'final_answer')
workflow.add_edge('final_answer', END)
# 8. 랭그래프 컴파일 -> 워크플로우 객체
# 랭그래프객체 => 전역변수
랭그래프객체 = workflow.compile()
2번 외부 도구 가져오기 이후부터 진행이다.
그 다음부터는 fewshot 프롬프트 적용, 템플리 ㅅ구성, 시스템 프롬프트 까지는 이전과 같다.
랭그래프가 추가되었는데
# 5. 에 있는 랭그래프 상태는 랭그래프가 주고받는 어떤 데이터도 이와 같은 형식을 유지하고 간다고 선언하는 내용이다.
그래서 모든 데이터에 다 messages라는 키를 가지고 있는 dict형태라고 말하는거라고 생각하면 된다.
그 이후 랭그래프에서 실제로 돌아갈 노드를 구성한다.
세 가지 노드를 정의한다.
첫번째 노드에서 메뉴 추천 그리고 도구를 사용할 지 말지를 정한다.
프롬프트의 내용을 획득하고 llm과 프롬프트를 결합한다.
그리고 질문을 요청한다.
두번째 노드에서는 도구를 사용한다고 결정했을 시에, rag를 이용한 검색을 한다.
이 떄 rag_search를 사용해서 검색을 시도한다.
세번째 노드에서 최종 답변을 생성한다.
마지막 결과물로 주고 받은 내용과 검색 증강 내용을 결합하여 최적의 답을 도출해낸다. 이때는 tool이 필요가 없겠지?
마지막으로 노드들을 연결해준다.
last_msg에 -1이 붙어있는 이유가 모든 대화중에서 마지막 내용만 출력하라는 뜻이기 때문이다.
마지막이 제일 복잡함시롱
그리고 마지막 함수에서 더해준 노드들을 엣지를 더해서 방향을 정해준다.
이 때 조건부 엣지에 있는 check_tool_node가 툴을 사용할지 말지 정해주는거다.
마지막 메세지에 tool_calls가 있다면 (툴이 필요하다고 요청한다면)
tools 노드로 가서 툴을 사용하라는 것이다.
없다면, end final_answer로 가는거다.
그리고 이 코드는 프론트, 백엔드까지 가기 때문에 랭그래프객체라는 객체를 타고 다시 백엔드로 간다.
server.py
from fastapi import FastAPI
from pydantic import BaseModel
from dotenv import load_dotenv
from langchain_core.messages import HumanMessage
from agent_with_graph import 랭그래프객체
app = FastAPI(title='식사 메뉴 추천 에이전트')
class UserRequest(BaseModel):
question:str
@app.post('/chat')
async def llm_endpoint(req:UserRequest):
try:
prompt = {'messages':[HumanMessage(content=req.question)]}
# 재시도 컨셉 =>
config = {"recursion_limit":5} # 최초 요청에 실패하면 최대 5회까지 다시 시도함
final_state = 랭그래프객체.invoke(prompt,config=config)
res = final_state['messages'][-1].content
return {"response":res}
except Exception as e:
return{'response':f'오류 발생{str(e)}'}
백엔드 코드다.
여기서 res = 랭그래프객체가 invoke한 결과물을 받아서 res를 post방식으로 ㅍ프론트엔드로 보낸다.
app.py
import streamlit as st
import requests as req
# a = 1
# print('사용자 입력후 엔터치면 계속 전체가 구동되는지 점검')#, a)
# 전역설정
API_URL = 'http://localhost:8000/chat' # fastapi 주소
st.set_page_config(page_title='식사 메뉴 해결사', page_icon='🍔')
st.title('AI 식사 메뉴 해결사 - 킹')
st.caption('예상, 점심/저녁 등 시점, 날씨, 기분, 단체여부 등 알려주시면 메뉴를 추천해드립니다.')
# session state 초기화 -> 현재 코드가 몇 번이고 재실행되더라도 데이터를 유지, 전역
if 'messages' not in st.session_state: # 최초에는 아무것도 없음
st.session_state.messages = [
# 페르소나는 백엔드에서 구성
{
'role' : 'assistant',
'content' : '안녕하세요! 오늘 식사는 어떤 것이 땡기나요?(예산, 점심/저녁 등 시점, 날씨, 기분, 단체여부 등 알려주시면 메뉴를 추천해드립니다.)'
}
]
# 이전 대화내용 화면 출력
for msg in st.session_state.messages:
with st.chat_message(msg['role']): # assistant or user
st.markdown(msg['content'])
# ui
# prompt = st.chat_input('현재 상황을 자세히 입력하세요')
# print(prompt)
# a += 1
# if prompt:
# prompt 입력값을 받아서 -> 존재하면 -> 작업 진행
# 대입 표현식(혹은 왈러스 연산자) -> 나오지 않은 문법임
if prompt:= st.chat_input('현재 상황을 자세히 입력하세요'):
# 사용자 질의 처리 진행
# 1. 사용자의 입력 내용을 전역 상태 관리 변수에 추가
st.session_state.messages.append({
'role':'user',
'content' : prompt
})
# 2. 사용자 입력 후 -> 마크다운 표기
# 화면에 방금 추가된 내용을 바로 반영하여 출력해라
with st.chat_message('user'): # user로 고정했음
st.markdown(prompt) # 화면에 텍스트 내용 출력
pass
# 3. LLM에게 문의 -> 서버 요청 -> bedrock 요청 -> bedrock 응답
# 서버 응답 -> assistant의 응답
with st.chat_message('assistant'):
# msg_holder = st.empty()
# msg_holder.markdown('고민중....ㅡ.ㅡ')
with st.spinner('고민중....ㅡ.ㅡ'): # 로딩창
# 3-1. 서버측으로 사용자의 질의 전송
result = None
try:
res = req.post(API_URL,json={'question':prompt})
if res.status_code == 200: # 응답 성공 뜻
result = res.json().get('response','응답 없음')
pass
else:
result = f'서버측 오류 {res.status_code}'
# 추후, 백엔드 구성 후 교체
# import time
# time.sleep(3)
# res = '더미 응답 : 치킨으로 가보세요!'
except Exception as e:
# 더미 예외처리 구성
print('에러',e)
result='LLM 사용자가 너무 많습니다, 10초후에 다시 시도해주세요'
# 3-2. 화면 처리
st.markdown(result)
# 3-3. 전역 상태 관리 변수에 추가
st.session_state.messages.append({
'role':'assistant',
'content':'res'
})
pass
pass마지막 프론트엔드.py이다.
수정된건 이 부분이다.
서버측으로 사용자의 질의를 전송하는 내용인 3-1에서 post방식으로 사용자 질의를 전송하고, 그걸 res로 다시 받아낸다.
그리고 응답이 성공하면 result에 그 답이 담겨서, 바로 화면처리 해버리는것이다.
결과물
수업시간에 한거랑 뭔가 좀 다르고 불안정하긴 한데 일단 작동은 한다.
데이터 또한 더미데이터에 입력한 rag기반으로 잘 도출되는 모습을 볼 수 있다.
오늘의 복습은 여기까지