자율적으로 판단하고 대화하는 AI를 만드는 단계입니다.
요약: Agentic RAG란 무엇인가요?
핵심 목표
- 단순히 정보를 가져오기(fetch)만 하는 것이 아니라,
- 사용자의 요청을 이해하고
- 적절한 판단을 하며
- 정보를 검색하고
- 자연스럽게 대화로 응답할 수 있는 자율적인 AI 에이전트를 만드는 것입니다.
AI 에이전트의 핵심 개념
1. 자율성 (Autonomy)
- 사람의 개입 없이 스스로 작동
- 사용자의 입력에 따라 적절한 정보를 검색하고 응답
2. 인식 (Perception)
- 텍스트, 음성, 이미지 등의 입력 데이터를 받아들이고 분석
- "이 식당에서 아이들에게 적합한 메뉴가 뭐야?" → 메뉴 정보 검색
3. 목표 지향적 행동 (Goal-Oriented Behavior)
- 특정 목표(예: 사용자 질문에 대한 최적의 응답 제공)를 달성하기 위해 작동
- "건강한 식단을 찾고 있어." → 저칼로리 요리 추천
4. 적응성 (Adaptability)
- 사용자 질문이나 상황에 맞춰 반응을 조정
- "메뉴 말고 분위기에 대해 알고 싶어." → 식당 내부 사진 및 분위기 설명 추가
5. 지속적인 상호작용 (Interaction)
- 단순히 정보를 전달하는 것이 아니라 대화를 지속하면서 피드백을 반영
- "이 요리는 너무 매워 보이는데 다른 추천 있어?" → 대체 추천 제공
AI 에이전트의 작동 원리: Sense - Think - Act
1. Sense (인식)
- 사용자의 질문을 이해
- 예: "비건 메뉴가 있나요?"
2. Think (분석 및 의사결정)
- 질문의 의도를 분석
- 데이터베이스에서 관련 정보를 검색
3. Act (행동 수행)
- 적절한 응답을 생성
- 예: "네, 채식주의자를 위한 두부 샐러드와 비건 파스타가 있습니다."
RAG 시스템에서 AI 에이전트의 역할
RAG 시스템에서는 AI 에이전트가 단순히 정보를 검색하는 것이 아니라, 사용자의 맥락과 필요에 맞춰 가장 적절한 데이터를 찾아서 응답을 생성합니다.
예를 들어, 고객이 "매운 요리를 추천해 줘"라고 했을 때, AI 에이전트는 다음을 수행합니다.
1. 고객이 "매운 요리"를 언급했음을 감지 (Sense)
2. 메뉴에서 매운 요리를 검색하고, 고객이 기존에 좋아한 메뉴 데이터를 분석 (Think)
3. "당신이 선호할 만한 매운 요리는 매운 소고기 볶음과 칠리 새우입니다."라고 응답 (Act)
AI 에이전트는 단순한 검색 시스템이 아니라 사용자와 상호작용하며 최적의 정보를 제공하는 시스템입니다.
Agentic RAG 구축
대화의 문맥을 기억하고, 추가 정보를 요청하며, 더 자연스럽게 고객과 상호작용하는 AI 시스템을 구축해보겠습니다.
워크플로우 작성 후 보면서 함수와 워크플로우를 한 개씩 작성하는 것이 좋습니다.
1. 환경 설정 및 라이브러리 설치
먼저 필요한 라이브러리를 설치하고 import 합니다.
# 필수 라이브러리 설치
!pip install -q langchain-community langchain-openai unstructured faiss-cpu langgraph
# 드라이브 마운트
from google.colab import drive
drive.mount('/content/drive')
%cd /content/drive/MyDrive/GenAI/RAG/Agentic RAG
# API 키 연결
from google.colab import userdata
api_key = userdata.get('openai_key')
# 라이브러리 Import
from langchain_community.document_loaders import UnstructuredExcelLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.vectorstores.faiss import FAISS
from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.schema.output_parser import StrOutputParser
from typing import List # 상태 관리에 사용
from typing_extensions import TypedDict # 상태 관리에 사용
from IPython.display import display, Image
from langgraph.graph import StateGraph, END2. Google Drive 연결 및 데이터 로딩
메뉴 데이터셋 (엑셀 파일) 로딩
우리는 dim_sum_montage.xlsx 파일을 사용합니다.
# 파일 경로 설정
file = 'dim sum montijo.xlsx'
# 엑셀 파일 로더 사용(LangChain)
loader = UnstructuredExcelLoader(file, mode="elements")
data = loader.load()
print(f"데이터 로드 완료 : {len(data)} 개의 문서")
print(data[:5])3. OpenAI Embeddings 및 벡터 데이터베이스 설정
메뉴 데이터를 벡터 임베딩(embedding)으로 변환한 후, FAISS 벡터 데이터베이스에 저장합니다.
import os
# API Key 설정
os.environ["OPENAI_API_KEY"] = api_key
# Open API Embeddings 생성
embeddings = OpenAIEmbeddings()
# FAISS 벡터 데이터베이스에 저장
db = FAISS.from_documents(data, embeddings)
상태 (State)란?
현재 대화의 진행 상황을 추적하는 역할을 합니다.
예를 들어, 사용자가 "오늘의 스페셜 메뉴는?"이라고 물었을 때,
AI가 같은 질문에 반복해서 같은 답을 하지 않도록 대화의 단계(Conversation Step) 를 저장합니다.
메모리 (Memory)란?
사용자의 이전 질문, 선호도, 알레르기 정보 등을 장기적으로 저장하는 역할을 합니다.
예를 들어, 사용자가 채식 메뉴를 찾고 있다면, 이후 추천을 할 때 자동으로 채식 위주 메뉴를 추천할 수 있습니다.
4. 상태 및 메모리 관리에 필요한 데이터 구조 정의
우리는 Python의 TypedDict를 활용하여 상태(State) 및 메모리를 정의합니다.
# 상태(state) 관리에 필요한 데이터 구조
class ConversationState(TypedDict):
start: bool # 대화 시작 여부
conversation: int # 대화 횟수
question: str # 사용자의 마지막 질문
answer: str # AI의 마지막 응답
topic: bool # 질문이 관련된 주제인지 여부
documents: list # 검색된 문서 목록
recursion_limit: int # 무한 루프 방지 (최대 질문 횟수)
memory: list # 장기 메모리 (사용자 선호도 저장)주요 필드 설명
| start | 대화가 시작되었는지 여부 |
| conversation | 현재 대화 횟수 (Step) |
| question | 사용자의 마지막 질문 |
| answer | AI의 마지막 응답 |
| topic | 질문이 관련된 주제인지 여부 |
| documents | 검색된 문서 목록 |
| question_limit | 최대 질문 횟수 (무한 루프 방지) |
| memory | 사용자 관련 정보 (예: 알레르기, 선호 음식) |
5. 상태 및 메모리 초기화 함수 만들기
AI가 처음 실행될 때 기본적인 상태 및 메모리를 초기화하는 함수를 만듭니다.
def initialize_state() -> ConversationState:
""" 대화 상태 초기화 """
return ConversationState(
start=True,
conversation=0,
question="",
answer="",
topic=True,
documents=[],
question_limit=5,
memory=[]
)
# 상태 초기화 실행
state = initialize_state()
print("초기 상태:", state)6. AI가 대화 흐름을 유지하도록 상태 업데이트 함수 만들기
AI가 대화할 때마다 질문 및 응답을 업데이트하는 함수입니다.
def update_state(state: ConversationState, user_question: str, ai_response: str, retrieved_docs: List[str]):
""" 대화 상태 업데이트 """
state["conversation"] += 1 # 대화 횟수 증가
state["question"] = user_question # 새로운 질문 저장
state["answer"] = ai_response # 새로운 응답 저장
state["documents"] = retrieved_docs # 검색된 문서 저장
# 사용자 질문이 "관련된 주제"인지 확인 (예: 메뉴 질문이 아닌 날씨 질문)
if "menu" in user_question.lower() or "dish" in user_question.lower():
state["topic"] = True
else:
state["topic"] = False
# 메모리 업데이트 (예: 알레르기 정보 저장)
if "allergic" in user_question.lower():
state["memory"].append(user_question)
return state
# 예제 실행
state = update_state(state, "Do you have gluten-free options?", "Yes! We have a gluten-free menu.", ["Gluten-Free Menu"])
print("업데이트된 상태:", state)7. AI의 첫 응답을 설정하는 "환영 메시지 함수" 만들기
사용자가 처음 접속했을 때 환영 메시지를 보내고 대화를 시작하도록 합니다.
def greet_user(state: ConversationState):
""" 대화 시작 및 환영 메시지 """
if state["start"]:
state["start"] = False
return "Hello! Welcome to our restaurant. How can I assist you today?"
return None
# 환영 메시지 실행
print("AI의 첫 응답:", greet_user(state))8. AI가 무한 루프를 방지하도록 "대화 종료 함수" 만들기
사용자가 너무 많은 질문을 하면 자동으로 대화를 종료하는 기능을 추가합니다.
def check_conversation_limit(state: ConversationState) -> bool:
""" 대화 횟수가 너무 많으면 종료 """
if state["conversation"] >= state["question_limit"]:
return True
return False
# 예제 실행
print("대화 종료 여부:", check_conversation_limit(state))
Agent State란?
Agent State는 AI 에이전트가 대화 중 현재 상태를 저장하고 관리하는 역할을 합니다.
이를 통해 사용자의 질문을 기억하고, 반복 질문을 방지하며, 문맥을 유지할 수 있습니다.
9. Agent State를 저장할 데이터 구조 생성 (TypedDict 사용)
Agent State 클래스 정의
우리는 Python의 TypedDict를 사용하여 Agent State를 정의합니다.
from typing_extensions import TypedDict
from typing import List
# 1. Agent State 데이터 구조 정의
class AgentState(TypedDict):
start: bool # 대화 시작 여부 (True = 시작)
conversation: int # 현재 대화 횟수 (몇 번째 질문인지)
question: str # 사용자의 가장 최근 질문
answer: str # AI의 가장 최근 응답
topic: bool # 질문이 관련된 주제인지 여부 (True = 관련 있음)
documents: List[str] # 검색된 문서 목록
recursion_limit: int # 무한 루프 방지 (최대 질문 횟수)
memory: List[str] # 장기 메모리 (사용자 선호도 저장)주요 필드 설명
| start | 대화가 시작되었는지 여부 |
| conversation | 현재 대화 횟수 (Step) |
| question | 사용자의 가장 최근 질문 |
| answer | AI의 가장 최근 응답 |
| topic | 질문이 관련된 주제인지 여부 |
| documents | 검색된 문서 목록 |
| recursion_limit | 무한 루프 방지 (최대 질문 횟수) |
| memory | 사용자 관련 정보 (예: 알레르기, 선호 음식) |
10. 초기 Agent State 설정
이제 AI가 처음 실행될 때 기본적인 상태를 초기화하는 함수를 만듭니다.
def initialize_agent_state() -> AgentState:
""" 초기 Agent State 설정 """
return AgentState(
start=True,
conversation=0,
question="",
answer="",
topic=True,
documents=[],
recursion_limit=5,
memory=[]
)
# 상태 초기화 실행
agent_state = initialize_agent_state()
print("초기 Agent State:", agent_state)이제 agent_state 변수가 초기화되었으며, AI는 첫 대화를 시작할 준비가 되었습니다.
11. State Graph 생성
Agent State를 활용한 State Graph를 초기화합니다.
State Graph란?
대화의 흐름을 추적하고 관리하는 구조로, AI가 질문을 받고 응답하는 프로세스를 조율하는 역할을 합니다.
from langgraph.graph import StateGraph
# 1. State Graph 초기화
workflow = StateGraph(AgentState)
print("State Graph 초기화 완료")이제 AI가 대화를 진행할 수 있도록 State Graph가 준비되었습니다.
이제 AI가 대화의 흐름을 추적하면서, 자연스럽게 사용자와 상호작용할 수 있습니다.
12. 인사(Greetings) 기능 구현
AI가 처음 대화를 시작할 때 손님을 맞이하고 질문을 받는 역할을 합니다.
greetings() 함수 구현
# 1. 인사(Greetings) 함수 정의
def greetings(state: AgentState) -> AgentState:
""" 디지털 웨이터가 손님을 맞이하고 첫 질문을 받는 함수 """
# 웨이터의 첫 인사
print("Hello! Welcome to the restaurant. I will be your waiter. How can I help you?")
# 사용자 입력 받기
user_input = input("You: ")
# 상태 업데이트 (대화 시작, 질문 저장)
state["question"] = user_input
state["conversation"] = 1 # 첫 번째 대화
state["memory"] = [] # 초기 메모리 설정
return state # 업데이트된 상태 반환설명
- AI가 첫 인사 메시지를 출력
- input() 함수를 사용해 사용자 질문을 받음
- state 변수에 대화 상태(질문, 대화 단계, 메모리) 업데이트
13. Workflow에 인사 기능 추가
이제 Agentic RAG 워크플로우에 greetings() 기능을 추가합니다.
워크플로우에 노드 추가
# 2. Workflow에 인사(Greetings) 노드 추가
workflow.add_node("greetings", greetings) # 'greetings' 노드 추가
workflow.set_entry_point("greetings") # 시작점을 'greetings'로 설정설명
- "greetings" 노드를 워크플로우의 첫 번째 노드로 추가
- set_entry_point("greetings")를 사용해 AI가 첫 질문을 받도록 설정
14. Workflow를 컴파일하고 시각화
이제 워크플로우를 컴파일하고, 그래프로 확인합니다.
워크플로우 컴파일 및 실행
from IPython.display import display
# 3. Workflow 컴파일
app = workflow.compile()
# 4. Workflow 시각화 (Mermaid 다이어그램)
display(Image(app.get_graph(xray=True).draw_mermaid_png()))출력 결과
AI의 인사 기능이 포함된 워크플로우를 시각적으로 확인 가능합니다.
15. 질문 평가 로직 개요
AI가 손님의 질문을 분석해 두 가지 경우를 판단해야 합니다.
1. 질문이 식당과 관련된 경우 → 정상적으로 정보를 검색하여 응답
2. 질문이 무관한 경우 → 예의 있게 대화를 종료하거나 주제를 바꿈
예제:
사용자의 질문 AI 판단 (Topic)
| "오늘의 스페셜 메뉴는 뭐야?" | True (관련 질문) |
| "여기 WiFi 비밀번호 뭐야?" | True (관련 질문) |
| "오늘 날씨 어때?" | False (무관한 질문) |
| "축구 경기 결과 알려줘." | False (무관한 질문) |
16. check_question() 함수 구현
AI 모델이 질문을 평가하는 함수를 만들어 Boolean 값 (True/False)을 반환하도록 합니다.
check_question() 함수 코드
# 1. 질문 평가 함수 정의
def check_question(state: AgentState) -> AgentState:
""" 사용자의 질문이 식당과 관련된지 평가하는 함수 """
# 사용자의 질문 가져오기
question = state["question"]
# 시스템 프롬프트 정의 (질문 적절성 평가)
system_prompt = """
You are an evaluator for a restaurant's digital assistant.
Your job is to assess whether the customer's question is relevant to the restaurant.
Respond with 'True' if the question is suitable and related to the restaurant,
otherwise respond with 'False'. Only return 'True' or 'False' in your response.
"""
# ChatPromptTemplate을 사용해 프롬프트 생성
template = ChatPromptTemplate.from_messages([
("system", system_prompt),
("human", "{question}")
])
prompt = template.format(question=question)
# GPT-4o 모델을 사용해 판단 수행
model = ChatOpenAI(model="gpt-4o-mini", api_key=api_key)
response = model.invoke(prompt)
# Topic 값 업데이트 (True or False)
state["topic"] = response.content.strip() == "True"
return state설명
- state["question"] → 사용자의 질문을 가져옴
- system_prompt → AI가 질문을 평가하도록 지침 제공
- ChatPromptTemplate → 사용자 질문을 입력으로 받음
- ChatOpenAI(model="gpt-4o-mini") → OpenAI GPT-4o-mini를 사용해 판단
- state["topic"] → AI 응답이 "True"이면 적절한 질문, "False"이면 무관한 질문
17. 워크플로우에 check_question() 추가
이제 check_question()을 Agentic RAG 워크플로우에 추가합니다.
워크플로우 노드 추가
# 2. Workflow에 Check Question 노드 추가
workflow.add_node("check_question", check_question)18. 시각화: 현재 Agentic RAG 구조
이제 Agentic RAG의 현재 상태를 그래프로 시각화합니다.
# 3. Workflow 컴파일
app = workflow.compile()
# 4. Workflow 시각화
display(Image(app.get_graph(xray=True).draw_mermaid_png()))출력 결과:
AI의 질문 라우팅 흐름
이전 단계에서 check_question() 함수를 통해 AI가 질문의 적절성을 판단하는 로직을 만들었습니다.
이제 AI가 판단한 결과를 바탕으로, 적절한 경우와 무관한 경우를 분기 처리합니다.
19. topic_router() 함수 구현
AI가 check_question()의 결과 (state["topic"])를 보고 다음 단계로 라우팅하도록 합니다.
topic_router() 코드
def topic_router(state: AgentState) -> str: #질문이 적절한지 (On-Topic) 여부에 따라 다음 단계를 결정하는 함수
return "on_topic" if state["topic"] else "off_topic"20. 분기처리 함수 구현
AI가 식당과 무관한 질문에 예의 있게 응답하고 대화를 종료하는 기능을 추가합니다.
off_topic_response() 코드
# 분기처리
def retrieve_documents(state: AgentState) -> AgentState:
state["answer"] = "관련 문서를 검색하고 응답합니다."
print(state["answer"])
return state
def off_topic_response(state: AgentState) -> AgentState:
if state["conversation"] <= 1:
state["answer"] = "죄송합니다, 저는 레스토랑과 메뉴에 대한 질문만 받을 수 있습니다."
else:
state["answer"] = "도움이 되어 기쁩니다! 언제든 질문해주세요."
print(state["answer"])
return state설명
- 첫 번째 질문이면 "죄송합니다, 메뉴 관련 질문만 가능합니다."
- 대화가 여러 번 오간 후라면 "도움이 되어 기쁩니다!"
21. topic_router()를 워크플로우에 추가
이제 topic_router()를 Agentic RAG 워크플로우에 추가해야 합니다.
workflow에 분기 처리 추가
# 1. 조건 분기 함수 정의 (Runnable 형태로 사용 가능)
def topic_router(state):
if state['topic'] == 'finance':
return 'finance'
elif state['topic'] == 'menu':
return 'menu'
else:
return 'unknown' # 안전한 처리
# 2. 분기 처리 (add_conditional_edges)
workflow.add_conditional_edges(
"check_question",
topic_router, # 문자열이 아닌, 직접 정의한 함수
{
"finance": "finance_step",
"menu": "menu_step",
"unknown" : "off_topic_response" # 예외처리 가능
}
)설명
- check_question() → topic_router() 이동
- topic_router() → 질문이 적절하면 정보 검색으로, 무관하면 대화 종료
- off_topic_response() → 최종 종료 (end)
22. retrieve_docs() 함수 구현
이제 사용자의 질문과 기존 대화 내용을 바탕으로 관련 정보를 검색하는 기능을 추가합니다.
retrieve_docs() 코드
def retrieve_docs(state: AgentState) -> AgentState:
""" 질문 및 메모리를 기반으로 관련 문서를 검색하는 함수 """
# 1. 현재까지의 대화 내용을 메모리에 저장
memory = " ".join(state["memory"]) # 대화 이력을 문자열로 변환
# 2. 질문과 대화 맥락을 결합하여 검색 쿼리 생성
search_query = f"{memory} {state['question']}".strip()
# 3. 벡터 데이터베이스에서 관련 문서 검색 (유사도 검색)
docs = database.similarity_search(str(search_query), k=5) # 가장 관련 있는 5개 문서 검색
# 4. 검색된 문서의 내용 저장
state["documents"] = [doc.page_content for doc in docs]
# 5. 메모리에 질문 추가 (대화 맥락 유지)
state["memory"].append(state["question"])
return state설명
- state["memory"] → AI가 기억하는 이전 대화 내용을 가져옴
- search_query → 이전 대화 + 새로운 질문을 하나의 검색 쿼리로 만듦
- database.similarity_search() → 유사도 검색으로 관련 문서 5개를 찾음
- state["documents"] → 검색된 문서를 저장
- state["memory"].append(state["question"]) → 현재 질문을 메모리에 추가
23. retrieve_docs()를 워크플로우에 추가
이제 retrieve_docs()를 Agentic RAG의 워크플로우에 연결해야 합니다.
retrieve_docs()를 워크플로우에 추가
# 1. 정보 검색 기능을 워크플로우에 추가
workflow.add_node("retrieve_docs", retrieve_docs)
# 2. 질문의 적절성을 평가한 후, 검색 또는 종료 결정
workflow.add_conditional_edges(
"check_question", # 질문 분석 후
"topic_router", # 적절한 질문인지 라우팅
{
"on_topic": "retrieve_docs", # 질문이 적절하면 정보 검색
"off_topic": "off_topic_response" # 질문이 무관하면 대화 종료
}
)
# 3. 무관한 질문을 받은 경우, 대화 종료로 연결
workflow.add_edge("off_topic_response", "end")설명
- retrieve_docs() → 적절한 질문이면 실행됨
- topic_router() → "on_topic"이면 retrieve_docs() 실행, "off_topic"이면 대화 종료
- off_topic_response() → 무관한 질문을 받은 경우 대화 종료
이번 단계에서는 답변 생성 기능 (generate_answer()) 을 구현하고, 워크플로우에 연결합니다.
답변 생성 시스템 개요
이전 단계에서 retrieve_docs()를 통해 AI가 적절한 질문이면 관련 정보를 검색하도록 했습니다.
이제 AI가 검색된 정보를 바탕으로 답변을 생성하는 기능을 추가합니다.
AI의 답변 생성 흐름
1. AI가 검색된 정보(documents)와 대화 맥락(memory)을 이용해 답변 생성
2. 답변을 "웨이터처럼 자연스럽게" 생성 (필요하면 추가 질문도 유도 가능)
3. 사용자가 새로운 질문을 하면 이전 대화 맥락을 유지하면서 응답
24. generate_answer() 함수 구현
AI가 검색된 문서(documents)와 대화 맥락(memory)을 바탕으로 답변을 생성하도록 합니다.
generate_answer() 코드
def generate_answer(state: AgentState) -> AgentState:
""" AI가 검색된 정보를 바탕으로 답변을 생성하는 함수 """
# 1. 대화 상태에서 질문 및 검색된 문서 가져오기
question = state["question"]
documents = state["documents"]
memory = state["memory"]
# 2. GPT 모델 초기화
model = ChatOpenAI(
model="gpt-4o-mini",
api_key=API_KEY
)
# 3. 시스템 프롬프트 설정 (AI 웨이터 역할)
system_prompt = """You are a waiter at a restaurant.
Your job is to assist customers with menu-related questions.
Answer in a professional yet friendly tone.
Do not refer to yourself as a waiter.
Keep responses concise but informative."""
# 4. AI가 사용할 프롬프트 템플릿 설정
prompt_template = ChatPromptTemplate.from_messages([
("system", system_prompt),
("human", "Context: {docs}\\nConversation history: {memory}\\nCustomer Question: {question}")
])
# 5. 프롬프트 생성
prompt = prompt_template.format(
docs=documents,
memory=memory,
question=question
)
# 6. AI 모델을 사용해 답변 생성
response_text = model.invoke(prompt)
# 7. 생성된 답변을 상태에 저장
state["answer"] = response_text.content
return state설명
- state["question"] → 사용자의 질문을 가져옴
- state["documents"] → 검색된 문서를 가져옴
- state["memory"] → 기존 대화 맥락을 가져옴
- AI 웨이터 프롬프트 설정
- "웨이터처럼 자연스럽게 답변하라"
- "자신을 웨이터라고 하지 마라"
- "너무 짧거나 길지 않게 답변하라"
- ChatOpenAI 를 사용해 AI가 답변을 생성
- 생성된 답변을 state["answer"] 에 저장
예제
User: "추천 메뉴가 있나요?"
AI: "전통적인 샤오롱바오가 인기 메뉴입니다! 돼지고기, 청경채, 표고버섯을 넣어 만듭니다."25. generate_answer()를 워크플로우에 추가
이제 generate_answer()를 Agentic RAG의 워크플로우에 연결해야 합니다.
generate_answer()를 워크플로우에 추가
# 1. AI 답변 생성 기능을 워크플로우에 추가
workflow.add_node("generate_answer", generate_answer)
# 2. 정보 검색 후, AI가 답변을 생성하도록 연결
workflow.add_edge("retrieve_docs", "generate_answer")설명
- retrieve_docs() → AI가 검색한 정보를 기반으로
- generate_answer() → AI가 답변을 생성
26. 답변 개선이 필요한 이유
generate_answer()는 사용자의 질문과 검색된 문서를 바탕으로 AI가 답변을 생성합니다.
하지만, AI가 항상 완벽한 문장을 생성하는 것은 아니므로, "자연스럽게 다듬는 과정"이 필요합니다.
AI 답변 개선 흐름
1. generate_answer() → 답변을 생성
2. improve_answer() → 답변을 다듬고, 불필요한 부분을 제거
3. 필요하면 후속 질문을 추가하여 대화 지속 가능
27. improve_answer() 함수 구현
이제 AI가 생성한 답변을 자연스럽게 다듬고, 대화를 이어가기 위한 후속 질문을 추가하는 기능을 구현합니다.
improve_answer() 코드
def improve_answer(state: AgentState) -> AgentState:
""" AI가 생성한 답변을 다듬고, 후속 질문을 추가하는 함수 """
# 1. 현재 상태에서 질문, 답변, 메모리 가져오기
question = state["question"]
answer = state["answer"]
memory = state["memory"]
# 2. GPT 모델 설정
model = ChatOpenAI(
model="gpt-4o-mini",
api_key=API_KEY
)
# 3. 시스템 프롬프트 설정
system_prompt = """As a waiter, review and refine the response to a customer's question.
Your task is to:
1. Ensure the answer is friendly and informative.
2. Edit or remove unnecessary parts without adding new information.
3. Maintain a polite and professional tone.
4. Provide only the improved answer without introductory phrases.
5. Conclude with an open-ended question to invite further inquiries.
6. Consider conversation history for a coherent response.
7. Avoid long sentences by breaking them into clear, concise statements.
Deliver a refined response that enhances the customer's experience."""
# 4. 프롬프트 템플릿 설정
prompt_template = ChatPromptTemplate.from_messages([
("system", system_prompt),
("human", "Conversation history: {memory}\\nCustomer Question: {question}\\nWaiter Answer: {answer}")
])
# 5. 프롬프트 생성
prompt = prompt_template.format(
memory=memory,
question=question,
answer=answer
)
# 6. AI 모델을 사용해 답변 개선
response_text = model.invoke(prompt)
# 7. 개선된 답변을 상태에 저장
state["answer"] = response_text.content
# 8. 메모리에 저장 (대화 이력 유지)
state["memory"].append(response_text.content)
return state설명
- state["question"], state["answer"], state["memory"] → 현재 질문과 AI 답변을 가져옴
- GPT-4o를 사용해 AI 웨이터 역할의 시스템 프롬프트 설정
- 프롬프트 템플릿에서 기존 AI 답변을 다듬도록 설정
- ChatOpenAI 를 사용해 AI가 자연스럽게 답변을 수정
- 최종 개선된 답변을 state["answer"]에 저장
- 새로운 답변을 메모리에 추가하여 대화 지속 가능
예제
User: "추천 메뉴가 있나요?"
AI (1차 생성): "샤오롱바오는 인기 메뉴입니다. 돼지고기와 청경채가 들어 있습니다."
AI (2차 개선): "샤오롱바오는 육즙 가득한 만두로, 돼지고기와 청경채를 넣어 만듭니다. 향긋한 국물과 함께 즐기기 좋습니다. 혹시 다른 추천이 필요하신가요?"28. improve_answer()를 워크플로우에 추가
이제 improve_answer()를 Agentic RAG의 워크플로우에 연결해야 합니다.
improve_answer()를 워크플로우에 추가
# 1. AI 답변 개선 기능을 워크플로우에 추가
workflow.add_node("improve_answer", improve_answer)
# 2. AI가 답변을 생성한 후, 개선하도록 연결
workflow.add_edge("generate_answer", "improve_answer")설명
- generate_answer() → AI가 초기 답변을 생성
- improve_answer() → AI가 답변을 다듬고 후속 질문을 추가
이제 사용자가 추가 질문을 할 수 있도록 지속적인 대화를 유지하는 기능을 추가하겠습니다
즉, AI가 생성한 답변이 끝난 후, 사용자가 후속 질문을 하면 이를 저장하고, 계속 대화를 이어가도록 설정합니다.
29. further_question() 함수 구현
AI의 최종 답변에는 항상 후속 질문이 포함되도록 설정했기 때문에, 이제 사용자의 추가 질문을 받아야 합니다.
further_question() 코드
def further_question(state: AgentState) -> AgentState:
""" 사용자의 후속 질문을 받아 지속적인 대화를 유지하는 함수 """
# 1. 공백 한 줄 추가하여 가독성 확보
print("\\n")
# 2. 사용자의 후속 질문 입력받기
user_input = input("User: ")
# 3. 새로운 질문을 상태(State)에 업데이트
state["question"] = user_input
# 4. 대화 단계 (conversation count) 업데이트
state["conversation"] += 1
# 5. 메모리에 사용자 질문 추가
state["memory"].append(user_input)
return state설명
- 공백 한 줄을 추가하여 출력 가독성 향상
- input()을 사용하여 사용자의 추가 질문을 입력받음
- 새로운 질문을 state["question"]에 저장
- 대화 단계를 +1 증가하여 conversation flow 유지
- 새로운 질문을 메모리에 추가하여 문맥을 유지
예제
User: "추천 메뉴가 있나요?"
AI: "샤오롱바오는 육즙 가득한 만두로, 돼지고기와 청경채를 넣어 만듭니다. 향긋한 국물과 함께 즐기기 좋습니다. 혹시 다른 추천이 필요하신가요?"
User: "음료 추천도 가능할까요?"
AI: "따뜻한 우롱차는 샤오롱바오와 잘 어울립니다. 혹시 다른 특정 음료를 원하시나요?"30. further_question()을 워크플로우에 추가
이제 대화 흐름을 유지하도록 워크플로우에 추가해야 합니다.
further_question()을 워크플로우에 추가
# 1. 후속 질문을 처리하는 노드 추가
workflow.add_node("further_question", further_question)
# 2. AI가 답변을 개선한 후, 사용자의 추가 질문을 받을 수 있도록 연결
workflow.add_edge("improve_answer", "further_question")
# 3. 새로운 질문을 받은 후, 다시 질문 확인 단계로 연결
workflow.add_edge("further_question", "check_question")설명
- improve_answer() → AI가 최종 답변을 개선한 후, 후속 질문을 받을 수 있도록 설정
- further_question() → 사용자의 질문을 저장한 후, 다시 check_question()으로 이동
- 끊임없는 대화가 가능하도록 설정
31. 최종 테스트
이제 Agentic RAG 시스템을 테스트해봅시다!
# AI Agentic RAG 실행
result = app.invoke({
"start": True,
"recursion_limit": 50
})테스트 입력 & 출력
User: "추천 메뉴가 있나요?"
AI: "샤오롱바오는 육즙 가득한 만두로, 돼지고기와 청경채를 넣어 만듭니다. 향긋한 국물과 함께 즐기기 좋습니다. 혹시 다른 추천이 필요하신가요?"
User: "음료 추천도 가능할까요?"
AI: "따뜻한 우롱차는 샤오롱바오와 잘 어울립니다. 혹시 다른 특정 음료를 원하시나요?"
User: "나는 해산물 알레르기가 있어요."
AI: "알겠습니다. 해산물이 없는 요리를 추천드리겠습니다. 혹시 알레르기 외에 선호하는 재료가 있나요?"이제 AI가 사용자의 질문을 받아 문맥을 유지하며 지속적인 대화를 진행할 수 있습니다
32. 최종 코드 정리
# 질문
def topic_router(state: AgentState) -> str: # 질문이 적절한지 (On-Topic) 여부에 따라 다음 단계를 결정하는 함수
return "on_topic" if state["topic"] else "off_topic"
# 1. 분기처리
def off_topic_response(state: AgentState) -> AgentState:
if state["conversation"] <= 1:
state["answer"] = "죄송합니다, 저는 레스토랑과 메뉴에 대한 질문만 받을 수 있습니다."
else:
state["answer"] = "도움이 되어 기쁩니다! 언제든 질문해주세요."
print(state["answer"])
return state
# 2. 사용자의 질문과 기존 대화 내용을 바탕으로 관련 정보 검색 하는 기능
def retrieve_documents(state: AgentState) -> AgentState:
""" 질문 및 메모리를 기반으로 관련 문서를 검색하는 함수 """
# 현재까지의 대화 내용을 메모리에 저장
memory = " ".join(state["memory"]) # 대화 이력을 문자열로 변환
# 질문과 대화 맥락을 결합하여 검색 쿼리 생성
search_query = f"{memory} {state['question']}".strip()
# 벡터 데이터베이스에서 관련 문서 검색 (유사도 검색)
docs = database.similarity_search(str(search_query), k=5) # 가장 관련 있는 5개 문서 검색
# 검색된 문서의 내용 저장
state["documents"] = [doc.page_content for doc in docs]
# 메모리에 질문 추가 (대화 맥락 유지)
state["memory"].append(state["question"])
return state
# 4. 검색된 문서와 맥락을 바탕으로 답변 생성
def generate_answer(state: AgentState) -> AgentState:
""" AI가 검색된 정보를 바탕으로 답변을 생성하는 함수 """
# 대화 상태에서 질문 및 검색된 문서 가져오기
question = state["question"]
documents = state["documents"]
memory = state["memory"]
# GPT 모델 초기화
model = ChatOpenAI(
model="gpt-4o-mini",
api_key=api_key
)
# 시스템 프롬프트 설정 (AI 웨이터 역할)
system_prompt = """You are a waiter at a restaurant.
Your job is to assist customers with menu-related questions.
Answer in a professional yet friendly tone.
Do not refer to yourself as a waiter.
Keep responses concise but informative."""
# AI가 사용할 프롬프트 템플릿 설정
prompt_template = ChatPromptTemplate.from_messages([
("system", system_prompt),
("human", "Context: {docs}\nConversation history: {memory}\nCustomer Question: {question}")
])
# 프롬프트 생성
prompt = prompt_template.format(
docs=documents,
memory=memory,
question=question
)
# AI 모델을 사용해 답변 생성
response_text = model.invoke(prompt)
# 생성된 답변을 상태에 저장
state["answer"] = response_text.content
return state
# 5. 답변 개선
def improve_answer(state: AgentState) -> AgentState:
""" AI가 생성한 답변을 다듬고, 후속 질문을 추가하는 함수 """
# 현재 상태에서 질문, 답변, 메모리 가져오기
question = state["question"]
answer = state["answer"]
memory = state["memory"]
# GPT 모델 설정
model = ChatOpenAI(
model="gpt-4o-mini",
api_key=api_key
)
# 시스템 프롬프트 설정
system_prompt = """
웨이터로서 고객의 질문에 대한 답변을 검토하고 다듬습니다.
당신의 임무는 다음과 같습니다:
1. 답변이 친절하고 유익한지 확인하세요.
2. 새로운 정보를 추가하지 않고 불필요한 부분을 편집하거나 제거하세요.
3. 예의 바르고 전문적인 어조를 유지하세요.
4. 소개 문구 없이 개선된 답변만 제공하세요.
5. 개방형 질문으로 마무리하여 추가 문의를 유도합니다.
6. 일관된 답변을 위해 대화 기록을 고려하세요.
7. 긴 문장을 명확하고 간결한 문장으로 구분하여 피하세요.
고객의 경험을 향상시키는 세련된 응답을 제공합니다.
"""
# 프롬프트 템플릿 설정
prompt_template = ChatPromptTemplate.from_messages([
("system", system_prompt),
("human", "Conversation history: {memory}\nCustomer Question: {question}\nWaiter Answer: {answer}")
])
# 프롬프트 생성
prompt = prompt_template.format(
memory=memory,
question=question,
answer=answer
)
# AI 모델을 사용해 답변 개선
response_text = model.invoke(prompt)
# 개선된 답변을 상태에 저장
state["answer"] = response_text.content
# 메모리에 저장 (대화 이력 유지)
state["memory"].append(response_text.content)
return state
def further_question(state: AgentState) -> AgentState: # 사용자의 후속 질문을 받아 지속적인 대화를 유지하는 함수
# 공백 한 줄 추가하여 가독성 확보
print("\n")
# 사용자의 후속 질문 입력받기
user_input = input("User: ")
# 새로운 질문을 상태(State)에 업데이트
state["question"] = user_input
# 대화 단계 (conversation count) 업데이트
state["conversation"] += 1
# 메모리에 사용자 질문 추가
state["memory"].append(user_input)
return state
## ------------- 워크 플로우---------------------
from IPython.display import display
# 1. 상태그래프 시작
# workflow = StateGraph(AgentState)
# 2. 노드 추가
workflow.add_node("greetings", greetings) # 'greetings' 노드 추가
workflow.set_entry_point("greetings") # 시작점을 'greetings'로 설정
workflow.add_node("check_question", check_question) # check_question
workflow.add_conditional_edges( # 질문 적절성 판단 후, 다음 단계 연결
"check_question", # 질문 분석 후
topic_router, # 라우팅 단계로 이동
{
"on_topic": "retrieve_documents", # 정보 검색으로 이동
"off_topic": "off_topic_response" # 대화 종료
}
)
workflow.add_node("off_topic_response", off_topic_response) # 무관한 질문 응답
workflow.add_node("retrieve_documents", retrieve_documents)
workflow.add_node("generate_answer", generate_answer)
workflow.add_node("improve_answer", improve_answer) # AI 답변 개선 기능을 워크플로우에 추가
workflow.add_node("further_question", further_question) # 후속 질문을 처리하는 노드 추가
#엔트리 설정
workflow.set_entry_point("greetings")
# Edge 노드
workflow.add_edge("greetings", "check_question")# Workflow에 Check Question 노드 추가
workflow.add_edge("off_topic_response", END) # 대화 종료 노드와 최종 종료 연결
workflow.add_edge("retrieve_documents", "generate_answer")# 정보 검색 후, AI가 답변을 생성하도록 연결
workflow.add_edge("generate_answer", "improve_answer") # AI가 답변을 생성한 후, 개선하도록 연결
workflow.add_edge("improve_answer", "further_question") # AI가 답변을 개선한 후, 사용자의 추가 질문을 받을 수 있도록 연결
workflow.add_edge("further_question", "check_question") # 새로운 질문을 받은 후, 다시 질문 확인 단계로 연결
최종 테스트
이런식으로 입력 칸이 나오고 입력을 하면 답변에 따라 대답이 달라지게 됩니다.
현재 DB에 연결이 안되어 있기 때문에 정확하게 답변하지 못하는 모습입니다.
반응형
'RAG' 카테고리의 다른 글
| Crew AI 실습 (0) | 2025.05.02 |
|---|---|
| AI Agents with CrewAI (0) | 2025.05.02 |
| Multimodal Data Project (1) | 2025.05.02 |
| Multimodal RAG (1) | 2025.05.02 |
| OpenAI RAG (0) | 2025.05.02 |