참고 사이트 : https://wikidocs.net/233801
챗봇: 어떤 도움이 필요하신가요? 비상사태 대처나 대피소 정보가 필요하시면 말씀해 주세요.
사용자: 내 장소는 동대구역이야
챗봇: 동대구역 근처에서 대피소를 찾고 계신가요? 그렇다면 가장 가까운 대피소를 찾아드리겠습니다. 확인을 원하시면 말씀해 주세요.
사용자: 근처 대피소를 찾아줘
챗봇: 대피소를 찾고자 하는 주소를 알려주시면 가장 가까운 대피소를 찾아드리겠습니다.
사용자: 내 장소의 대피소를 찾아줘
챗봇: 대피소를 찾고자 하는 주소를 알려주시면 가장 가까운 대피소를 찾아드리겠습니다.
사용자: 아까 말했던 장소
챗봇: 죄송하지만, 이전 대화 내용을 기억할 수 없습니다. 대피소를 찾거나 다른 도움이 필요하시면 현재 위치나 주소를 알려주세요.
이런 식으로 이전 대화 내용을 기억할 수 없다고 나와서 제대로된 답변을 받을 수 없었음.
# 비상사태 대처 매뉴얼 전문 챗봇 프롬프트 템플릿 정의
chat_template = ChatPromptTemplate.from_messages(
[
("system", (
"당신은 비상사태 대처 매뉴얼 전문 챗봇입니다. "
"재난 상황(지진, 화재, 홍수, 전쟁 등)이 발생했을 때 사용자가 안전하게 대피할 수 있도록 최적의 정보를 제공하는 것이 목표입니다.\n\n"
"제공된 컨텍스트만 사용해서, 질문에 답변하세요."
"아래의 지침에 따라 응답하세요:\n"
"1. 역할 정의: 사용자에게 신뢰할 수 있는 정보를 제공하고, 필요한 경우 함수 호출을 통해 가장 가까운 대피소를 추천하세요.\n"
"2. 대화 스타일: 간결하고 명확하며 사용자 친화적인 언어를 사용하고, 긴급 상황에 맞는 전문적인 톤을 유지하세요.\n"
"3. 긴급 연락처: 추가적인 도움이 필요할 경우 즉시 긴급 연락처(예: 119, 112)를 안내하세요.\n"
"4. 정보의 정확성과 최신성: 최신 데이터를 결합해 응답하세요. 데이터 부족 시 안전한 방향으로 안내하고 추가 도움을 요청하도록 권장하세요.\n"
"5. 함수 호출 지침: 대피소 검색이나 위치 관련 질문에 적절한 함수를 호출하여 데이터를 검색하세요.\n"
"6. 다양한 사용자 고려: 복잡한 용어 대신 쉬운 표현을 사용하세요.\n"
"7. 추가 지침: 필요한 경우 질문을 되묻고, 제공 정보가 명확한지 점검하세요."
)),
("human", "안녕하세요!"),
("ai", "안녕하세요! 저는 비상사태에서 안전한 대처를 도와드리는 전문 AI 챗봇입니다. 무엇을 도와드릴까요?"),
("human", "{user_input}"),
]
)
def get_user_input():
return input("사용자: ")
def handle_function_call(response):
function_call = response.additional_kwargs.get("function_call")
if function_call:
function_name = function_call["name"]
function_args = json.loads(function_call["arguments"])
if function_name == "find_nearest_shelters":
function_result = find_nearest_shelters(**function_args)
print(f"챗봇: 가장 가까운 대피소는 {function_result}입니다.")
else:
print(f"알 수 없는 함수 호출: {function_name}")
else:
print(f"챗봇: {response.content}")
def process_response(response):
if response.additional_kwargs.get("function_call"):
handle_function_call(response)
else:
print(f"챗봇: {response.content}")
class ModelInvocation(Runnable):
def invoke(self, input, config=None):
messages = chat_template.format_messages(user_input=input)
return model.invoke(messages)
class ResponseProcessor(Runnable):
def invoke(self, input, config=None):
process_response(input)
return input
chat_chain = (
ModelInvocation()
| ResponseProcessor()
)
def main():
# 프로그램 시작 시 초기 AI 메시지를 출력합니다.
initial_messages = chat_template.format_messages(user_input="")
response = model.invoke(initial_messages)
process_response(response)
while True:
user_input = get_user_input()
if user_input.lower() in ["종료", "exit", "quit"]:
print("대화를 종료합니다.")
break
response = chat_chain.invoke(user_input)
if __name__ == "__main__":
main()
# 대화 기록을 저장할 메모리 초기화
memory = ConversationBufferMemory(return_messages=True)
class ModelInvocation(Runnable):
def invoke(self, input, config=None):
# 이전 대화 기록 불러오기
past_messages = memory.load_memory_variables({})["history"]
# 현재 사용자 입력을 메시지로 변환
user_message = chat_template.format_messages(user_input=input)
# 이전 대화 기록과 현재 입력을 결합하여 모델 호출
messages = past_messages + user_message
# 모델 응답 생성
response = model.invoke(messages)
# 현재 대화를 메모리에 저장
memory.save_context({"input": input}, {"output": response.content})
return response
챗봇: 어떤 도움이 필요하신가요? 비상사태 대처나 대피소 정보가 필요하시면 말씀해 주세요.
사용자: 내 장소는 동대구역이야
챗봇: 가장 가까운 대피소는
가장 가까운 대피소 정보 도출됨
~~ 입니다.
사용자: 내 장소가 어디라고?
챗봇: 당신이 이전에 언급한 장소는 "동대구역"입니다. 이 위치와 관련하여 도움이 필요하신가요? 예를 들어, 가장 가까운 대피소를 찾고 싶으신가요?
이렇게 langchain
에서 제공하는 ConversationBufferMemory
을 사용해서 대화를 메모리에 저장하여 기억 할 수 있도록 코드를 수정해서 해결했다.
함수호출을 사용했을때 반환한 결과를 그대로 응답하고 나머지 질문에는 대답을 못하는 문제가 발생했다.
예를들어, 서울시청역이야 비상상황 대처 매뉴얼을 알려줘 -> 서울시청역 근처 대피소 정보만 반환하고 비상상황 대처 매뉴얼에는 응답하지 않고 대답이 종료됨.
def handle_function_call(response):
function_call = response.additional_kwargs.get("function_call")
if function_call:
function_name = function_call["name"]
function_args = json.loads(function_call["arguments"])
if function_name == "find_nearest_shelters":
function_result = find_nearest_shelters(**function_args)
print(f"챗봇: 가장 가까운 대피소는 {function_result}입니다.")
else:
print(f"알 수 없는 함수 호출: {function_name}")
else:
print(f"챗봇: {response.content}")
def handle_function_call(response):
function_call = response.additional_kwargs.get("function_call")
if function_call:
function_name = function_call.get("name")
function_args = function_call.get("arguments")
# arguments가 문자열인지 확인하고, 그렇지 않으면 그대로 사용
if isinstance(function_args, str):
function_args = json.loads(function_args)
if function_name == "find_nearest_shelters":
function_result = find_nearest_shelters(**function_args)
function_message = FunctionMessage(
name=function_name,
content=function_result
)
# 대화 기록에 함수 메시지 추가
memory.save_context({"input": function_message.content}, {"output": function_result})
# 모델에 함수 결과 전달하여 최종 응답 생성
final_response = model.invoke(memory.load_memory_variables({})["history"])
print(f"챗봇: {final_response.content}")
# 대화 기록에 AI 메시지 추가
memory.save_context({"input": function_result}, {"output": final_response.content})
else:
print(f"알 수 없는 함수 호출: {function_name}")
else:
print(f"챗봇: {response.content}")
문제 해결
arguments
가 문자열(JSON) 또는 딕셔너리일 수 있어 처리 방식이 달라야 함.FunctionMessage
를 생성하고 이를 대화 기록에 추가.