[LangGraph] 로컬여행가이드 에이전트 코드 살펴보기
이전 게시물에서 여행 유형을 기반으로 사용자의 여행을 더 정밀하게 분석할 수 있는 추가 기능에 대해 소개했다.
에이전트가 사용자의 여행 유형을 어떻게 대화에 반영하고, 사용자의 여행 선호도를 어떻게 분석하는지 코드를 통해 살펴보자.
(직접 코드로 보는게 이해가 더 빠를수도 있으나 짧게 설명을 진행합니다)
LangGraph 에이전트 작동 흐름
이전 게시물에서 이미 작동 흐름에 대해 설명하였다.
코드의 이해를 위해 LangGraph에 대해 조금 설명하자면, LangGraph는 그래프 기반의 워크 플로우로 노드(Node)와 엣지(Edge), 상태(State)를 활용하여 위와 같은 흐름으로 구조화 할 수 있다.
보라색 네모칸은 LangGraph에서 노드를 의미하며 각 노드는 하나의 작업을 수행하는 함수로 상태(State)를 매개변수로 받아 업데이트된 상태(State)를 반환한다. 에이전트의 상태(State)는 여러 노드가 작업을 수행하며 값을 공유하고 업데이트하는 저장소이다. 이렇게 구현된 노드를 엣지(Edge)를 통해 연결하는데 여러 조건 분기도 가능하다.
자세한 LangGraph의 설명은 공식 문서를 살펴보자🙂
LangGraph 공식 문서
1. 메인 코드
# 사용자별 에이전트 생성 함수
def create_user_agent(userId: Optional[str], model: LanguageModelLike, tools: Union[ToolExecutor, Sequence[BaseTool], ToolNode]) -> CompiledGraph:
# 사용자별 mcheckpointer 가져오기
user_checkpointer = get_user_checkpointer(userId)
# 에이전트 생성
return create_my_agent(
model=model,
tools=tools,
checkpointer=user_checkpointer
)
class QuestionRequest(BaseModel):
userMessage: str
userId: Optional[str] = None
tbtiType: Optional[str] = None
class AiResponse(BaseModel):
typeNum: Optional[int] = None
answer: str
place: Optional[List[Dict]] = None
app = FastAPI()
# 호출할 함수 리스트 가져오기
tools = tools_of_travel["list_of_func"] + tools_of_type["list_of_func"]
# 이전 state 값 저장
previous_state = {
"messages" : None,
"previous_result" : None,
"final_response" : None,
"tbti_of_user" : None,
"filtering" : {}
}
db = database
@app.post("/ask-ai/", response_model=AiResponse)
async def ask_ai(request: QuestionRequest):
global previous_state
userId = request.userId or "false"
userMessage = request.userMessage
tbtiType = request.tbtiType
try:
db.reconnect()
# 사용자 ID를 기반으로 에이전트 생성
user_agent = create_user_agent(userId, llm, tools)
# TBTI 유형 저장
previous_state["tbti_of_user"] = tbtiType
# 사용자 ID를 포함한 설정 구성
config = {"configurable": {"thread_id": userId, "user_id": userId}}
system_prompt = """
- You are a tour guide called 'TBTI'. Ask the user a short and clear question.
- Only up to five locations will be notified.
- Don't ask a question what type of trip the user wants.
- Don't ask specifically what kind of trip users want.
"""
messages_list = [("system", f"{system_prompt}")]
messages_list.append(("human", f"{userMessage}"))
previous_state["messages"] = messages_list
# 에이전트 실행
response = user_agent.invoke(previous_state, config)
previous_state = response
# JSON 직렬화 시 SecretStr 값 처리
return response['final_response']
except Exception as e:
print("에러 발생: ", e)
raise HTTPException(status_code=500, detail="AI 처리 중 오류 발생")
finally:
db.unconnect()
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8001)
FastAPI를 통해 사용자의 질문을 받고 프론트단으로 답변을 전달한다.
- 전달받는값: 사용자 질문, 사용자 ID, 사용자의 여행 유형(TBTI)
- 응답값: typeNum, AI 챗봇 응답, 여행지 추천 장소
주요한 부분은 create_user_agent에서 사용자만의 에이전트를 생성하고, 사용자의 질문 및 시스템 프롬프트를 에이전트 상태(State)에 저장하여 에이전트를 실행시킨다.
previous_state 딕셔너리에 에이전트의 이전 상태값을 저장하는 이유는 사용자의 여행 유형이나 이전 대화 기록을 통해 사용자와의 대화를 자연스럽게 연결하기 위함이다.
이제 에이전트의 구현 코드를 살펴보자
2. 에이전트 구현 코드
def create_my_agent(
model: LanguageModelLike,
tools: Union[ToolExecutor, Sequence[BaseTool], ToolNode],
checkpointer: Optional[BaseCheckpointSaver] = None
) -> CompiledGraph:
# LLM 시스템 프롬프트 리스트 로드 - 작동되는 함수에 따라 시스템 프롬프트 내용 다름
configuration_for_answers = system_informations_of_functions
# 도구 호출 가능한 모델 로드
model_with_tools = model.bind_tools(tools)
def save_user_info(state: AgentState):
return {'tbti_of_user': state["tbti_of_user"]}
...... 더보기create_my_agent 는 에이전트 생성 함수로 LLM, LLM 외부 도구(Tools), 채팅 메모리를 매개변수로 받아 CompiledGraph를 리턴한다. 이 함수에는 노드(Node)로 작동되는 작업들이 구현되어 있으며 노드와 엣지로 작업 흐름 또한 구현한다.
에이전트의 상태 및 작업 함수들을 하나씩 살펴보자.
2-1. 에이전트 상태(State)
# 노드에 전달되는 state
class AgentState(TypedDict):
messages : Annotated[list, add_messages]
previous_result : Optional[str]
final_response : Optional[dict]
tbti_of_user : Optional[str]
filtering : Optional[dict] LangGraph에서 에이전트의 상태(State)를 TypedDict을 사용하여 다음과 같이 정의할 수 있고, 모든 노드에서 이 값들을 공유하고 수정할 수 있다. 여행 가이드의 경우로 다섯가지의 요소만 저장을 했다.
- messages : 대화 기록 저장
- previous_result : 이전 단계에서의 생성된 AI 답변
- final_response : 사용자에게 전달될 최종 답변
- tbti_of_user : 사용자의 여행 유형 TBTI
- filtering : 사용자의 여행 특성 (데이터베이스 검색 필터 조건)
이 값은 사용자의 여행 유형에 따라 추가되는 필터로 사용자에게 맞는 여행지를 검색할 때 사용되는 정보이다.
2-2. 노드(Node) 작업 구현
노드의 흐름대로 노드에 어떤 작업이 이뤄지는지 설명하고자 한다.
1. 사용자의 여행 유형 저장
# 첫 노드
def save_user_info(state: AgentState):
return {'tbti_of_user': state["tbti_of_user"]}
# 검색 필터를 생성해야 하는지 파악
def should_create_filter(state: AgentState):
filtering = state['filtering']
tbti = state['tbti_of_user']
if len(filtering) > 0:
return "start talking"
elif len(filtering) == 0 and tbti != None:
return "create filter"
else:
return "start talking"
사용자의 여행 유형을 서버에서 받게 되면 유형을 에이전트 상태(State)에 저장하고, 여행 유형을 에이전트가 처음 받게 되면 검색 필터가 존재하지 않는다.
따라서 should_create_filter를 통해 검색 필터가 없는지 파악하고, 없다면 create filter 문자열을 리턴한다. 리턴된 값을 통해 검색 필터를 생성하는 작업 함수로 이동할 수 있다. 이 함수는 조건부 엣지(Conditional Edge)를 만들기 위해 사용되는데 에이전트 작동 흐름을 다시 보면 첫 노드 start-node에서 두 갈래로 나누어진 조건부 엣지를 확인할 수 있다.
2. 검색 필터 생성 및 시스템 프롬프트 추가
# 새로운 검색 필터 생성 노드
def generate_new_filter(state: AgentState):
messages = state["messages"]
filtering = {}
system_prompt = ['Ask a question in Korean one by one.']
# 사용자 여행 유형 가져오기
tbti = state['tbti_of_user']
try:
if tbti not in ['AIEU', 'AIEP', 'AIFU', 'AIFP', 'ASEU', 'ASEP', 'ASFU', 'ASFP', 'CIEU', 'CIEP', 'CIFU', 'CIFP', 'CSEU', 'CSEP', 'CSFU', 'CSFP'] and tbti == None:
return {"filtering": filtering, "messages": messages}
else:
# 여행 유형에 따른 검색 필터 및 추가 질문 준비
tbti = list(tbti)
for one_type in tbti:
match one_type:
case 'A':
filtering["mood"] = "(mood == 0)"
case 'C':
filtering["mood"] = "(mood == 1)"
case 'I':
system_prompt.append(tools_of_type['check_companion_animal']["added_system_message"])
case 'P':
pass
case 'S':
system_prompt.append(tools_of_type['check_child']["added_system_message"])
system_prompt.append(tools_of_type['check_companion_animal']["added_system_message"])
case 'E':
system_prompt.append(tools_of_type['check_distance']["added_system_message"])
case 'F':
filtering["parking"] = "(parking == true)"
case 'U':
filtering["reservation"] = "(reservation == true)"
except Exception as e:
print(e, " 올바른 TBTI 유형을 전달하세요.")
added_system_msg = ' '.join(system_prompt)
messages = [
("system", f"{added_system_msg}")
]
return {"filtering": filtering, "messages": messages}
위의 작업은 사용자의 여행 유형에 따른 검색 필터를 에이전트 상태(State)에 저장한다. 특정 유형에 경우, 추가적인 여행 특성을 구체적으로 파악하기 위해 시스템 프롬프트를 추가하여 LLM이 사용자의 여행에 대해 추가적인 질문을 하도록 유도한다.
만약 사용자의 여행 유형이 APSU 인 경우 👆
- 검색 필터 filtering: mood == 0, reservation == True 저장
S유형에 속하기 때문에 LLM의 시스템 프롬프트에는
"Ask first if users travel with their children."
"Ask first if users are going to travel with their pets."위의 두 문장이 추가되어 사용자 여행에 대해 구체적으로 알기 위한 LLM의 추가 질문이 이어질 수 있다.
3. AI 모델 로드 및 답변 생성
# 사용할 AI 모델 로드 및 AI 답변 처리
def talk_to_model(state: AgentState):
response = model_with_tools.invoke(state['messages'])
last_response = response.content.replace('\"', '\'')
last_response = f'{{\"answer\": \"{last_response}\", \"place\": null}}'
# AI 답변을 json 형식의 문자열로 만들어 previous_result에 저장 / 답변 history에 저장
return {"previous_result" : last_response , "messages" : [response]}
# 도구를 작동 시킬 지 파악
def should_continue(state: AgentState):
messages = state["messages"]
last_message = messages[-1]
# 함수 호출이 없으면 바로 사용자에게 리턴
if not last_message.tool_calls:
return "pass"
# 있으면 워크플로우 지속
else:
return "work"
상태(State)에 사용자의 여행 유형과 검색 필터가 모두 저장되고 나면 사용자의 질문에 대한 LLM 답변을 생성하는 작업으로 진행된다.
LLM 답변을 생성하여 JSON 형식의 문자열로 만들고 리턴하며 상태 속 previous_result 요소에 답변을 저장한다.
사용자에 질문을 LLM이 받게 되면 단순한 답변을 바로 생성해도 되는지, 추가 작업을 위해 외부 도구 호출이 필요한지 파악한다. 외부 도구 호출이 필요할 경우, should_continue 함수에서 work 문자열을 리턴하여 도구(Tools) 노드로 이동시킨다. 호출 가능한 외부 도구는 아래 경로로 확인할 수 있다.
4. 필터링 추가 기능
# 검색 필터를 추가할 지, 검색 결과를 통한 답변을 생성할 지 파악
def should_make_answer(state: AgentState):
name_of_tools = tools_of_type.keys()
tool_messages = state["messages"][-1]
if tool_messages.name in name_of_tools:
return "add type"
else:
return "make answer"
# 여행 취향을 파악하기 위한 추가 질문 답변 처리
def process_type_result(state: AgentState):
filtering = state["filtering"]
messages = state["messages"]
ai_message = filter_messages(messages, include_types=[AIMessage])[-1]
tool_call_ids = [item['id'] for item in ai_message.tool_calls]
# 이전 실행된 tool 결과 가져오기
tool_messages = filter_messages(messages, include_types=[ToolMessage], include_ids=tool_call_ids)
for tool in tool_messages:
content = tool.content
split_result = content.split(',')
if split_result[1] != 'null':
filtering[split_result[0]] = split_result[1]
else:
if split_result[0] in filtering:
del filtering[split_result[0]]
print('filteirng: ', filtering)
return {"filtering": filtering}여행 가이드 챗봇이 가진 외부 도구에는 여행지 추천, 여행 계획 수립 도구가 있지만, 사용자의 여행 특성을 추가하는 도구도 존재한다.
만약 👆
- 사용자 질문: 나 강아지랑 여행가고 싶어
# 반려동물과 함께 여행하는지 아닌지에 대한 답변에 작동되어 답변 저장
@tool
def check_companion_animal(response: bool) -> str:
"""check whether the users are going to travel with their pets or not and save their answers to the parameters.
Args:
response: if the user is going to travel with a pet, put 'true'. if not, put 'false'.
"""
if response == True:
filter_string = 'animal,(animal == true)'
else:
filter_string = 'animal,null'
return filter_string위의 도구가 실행되고, 위의 도구는 animal,(animal == true) 문자열을 리턴한다. 위 값은 process_type_result 함수에서 후처리 되어 에이전트 상태(State)의 filtering에 (animal == true) 값이 추가되도록 한다.
결과적으로, 사용자가 어떠한 여행을 원하는지 구체적인 정보를 이러한 방식으로 알아낼 수 있고, 사용자의 여행 특성이 변화한다면 즉각적으로 수정하여 LLM이 반영할 수 있도록 한다.
❓필터링(filtering) 값은 언제 쓰이는가?
@tool
def recommand_travel_destination(question : str, location : str, area : str, filtering: Annotated[dict, InjectedState('filtering')]) -> str:
"""
recommand the various places that user wants to know or to travel
It only works when user wants to know the various places.
It doesn't work when user told to plan the trip and when user told to reserve the place.
Args:
question: input the user's questions.
location: input the area of Korea to travel, e.g. 서울 or 부산 or 대구 or 강원도
area: Enter only the following words to indicate where the place in the user's question belongs to the following Korean administrative districts. e.g. 강원특별자치도
- 한국 행정 구역 : 서울특별시, 부산광역시, 인천광역시, 대구광역시, 대전광역시, 광주광역시, 울산광역시, 세종특별자치시, 경기도, 충청북도, 충청남도, 전라남도, 경상북도, 경상남도, 강원특별자치도, 전북특별자치도, 제주특별자치도
"""
milvus = database
# 사용자 질문 벡터화
vector = embedding(question)
# 필터링 생성 후 테이블 검색 진행
milvus_filter = None
area_filter = f"area_name == '{area}'"
if bool(filtering):
area_filter = area_filter + ' && '
another = ' && '.join(filtering.values())
milvus_filter = area_filter + another
else:
milvus_filter = area_filter
results_localCreator, results_nowLocal = milvus.search_all_tables(embedding=vector, filtering=milvus_filter, top_k=5)
# 쿼리 결과 합치기
total_results = milvus.get_formatted_results(results_localCreator, results_nowLocal)
return f"user question: {question} \n\nreference: \n{total_results}"여행지 추천 함수가 작동 시 위의 함수의 파라미터로 가져와, milvus 데이터 베이스 필터 조건으로 쓰여 사용자에게 맞는 장소만 가져오는 역할을 하게됨
5. 도구 호출 후, AI 답변 생성 및 전달
# 도구 작동 후 함수 결과 LLM에게 최종 전달 후 답변 생성
def respond_after_calling_tools(state: AgentState):
# 작동된 마지막 도구 메시지 가져오기
messages = state["messages"]
last_tool_message = messages[-1]
# 작동된 함수 이름 가져오기
name_of_functions_called = last_tool_message.name
# 함수 리턴값 가져오기 = 검색 결과 가져오기
reference = last_tool_message.content
# 작동된 도구에 맞는 시스템 프롬프트 가져오기
system_prompt = configuration_for_answers[name_of_functions_called]['system_prompt']
response_format = configuration_for_answers[name_of_functions_called]['response_format']
# 결과 참고해서 LLM 답변 생성
messages = [
{"role":"system", "content": f"{system_prompt}"},
{"role": "user", "content":f"{reference}"}
]
llm_response = chat_completion_request(
messages=messages,
response_format=response_format
).choices[0].message.content
return {"previous_result": llm_response}
def post_processing_of_answer(state: AgentState):
ai_answer = state["previous_result"]
escaped_response = escape_json_strings(ai_answer)
return {"final_response" : escaped_response}일반적인 여행지 추천과 같은 외부 도구가 실행된 후,
실행 결과를 LLM이 참고하여 답변을 생성할 수 있도록 한다. 여러가지 외부 도구들이 존재할 것이고, 작동된 도구에 맞는 시스템 프롬프트, LLM의 답변 형식을 불러와 LLM을 실행한다. LLM의 답변은 에이전트 상태의 previous_result에 저장한다.
post_processing_of_answer 노드에서는 json 문자열을 딕셔너리 자료형으로 변환하는 작업을 수행하고, 최종 답변으로 에이전트 상태(State)에 저장한다.
6. 에이전트 노드 생성 및 흐름 구현
# 새로운 그래프 정의
workflow = StateGraph(AgentState)
# 각 노드 생성
workflow.add_node("start-node", save_user_info)
workflow.add_node("generate-filter", generate_new_filter)
workflow.add_node("talk-to-human", talk_to_model)
workflow.add_node("tools", ToolNode(tools))
workflow.add_node("add-filter", process_type_result)
workflow.add_node("respond", respond_after_calling_tools)
workflow.add_node("json-processing", post_processing_of_answer)
# 그래프 진입 포인트 설정
workflow.set_entry_point("start-node")
workflow.add_conditional_edges(
"start-node",
should_create_filter,
{
"start talking": "talk-to-human",
"create filter": "generate-filter",
},
)
workflow.add_edge("generate-filter", "talk-to-human")
workflow.add_conditional_edges(
"talk-to-human",
should_continue,
{
"work": "tools",
"pass": "json-processing"
}
)
workflow.add_conditional_edges(
"tools",
should_make_answer,
{
"add type": "add-filter",
"make answer": "respond"
}
)
workflow.add_edge("add-filter", "talk-to-human")
workflow.add_edge("respond", "json-processing")
workflow.add_edge("json-processing", END)
graph = workflow.compile(
checkpointer=checkpointer
)
위의 모든 작업 함수들로 노드를 생성하고, 그래프의 진입 포인트, 엣지를 통하여 그래프의 작업 흐름을 생성한다.
compile 메서드로 생성된 graph가 바로 에이전트가 된다. Main 코드에서 생성된 에이전트는 사용자의 상호작용하며 기록된 대화 내용을 통해 사용자의 여행 유형을 답변에 반영하며, 사용자의 여행 계획이나 취향이 변화였을 때 즉각적으로 수정할 수 있었다.
7. 느낀 점 😶
조금 지난 코드인데 이 프로그램을 개발할 당시, 어떤 구조로 어떻게 개발해야 할지 까마득했던 기억이 난다. 현재, AI 에이전트의 라이브러리나 다양한 도구가 많이 개발된 것을 보아 새로운 문서와 기능들을 공부하여 더욱 보완된 에이전트를 개발하고 싶다.
