관련 시리즈
더보기
관련 시리즈 목록
- 2024.11.11 - [A.I.(인공지능) & M.L.(머신러닝)/LLM] - [LangGrpah] 1. LangGrpah(랭그래프)를 이용한 AI Workflow 관리하기
- 2024.11.13 - [A.I.(인공지능) & M.L.(머신러닝)/LLM] - [LangGrpah] 2. LangGrpah(랭그래프)의 핵심 요소의 개념적 이해
- 2024.11.14 - [A.I.(인공지능) & M.L.(머신러닝)/LLM] - [LangGrpah] 3. Chain과 Agent를 이용한 Workflow 구현 - 쿼리 추출 모델 (현재글)
- 2024.11.20 - [A.I.(인공지능) & M.L.(머신러닝)/LLM] - [LangGrpah] 4. LangGraph를 이용한 RAG 및 검색 Agent 개발
이번 포스팅에서는 실제 예시와 함께 Workflow를 구현해보고자 한다.
최종적인 목표는 온라인 쇼핑몰에서 동작하는 LLM Agent를 설계하는 것이다.
이를 달성하기 위한 하위 Task들을 구현해보고자 하겠다.
실제 LangGraph에서 제공하는 몇 가지 기술들에 대한 설명은
해당 기술을 사용하면서 간략하게 소개하도록 하겠다.
1. 전체적인 Workflow 설계
LangGraph를 이용한 Workflow를 구현하기 위해서는
미리 어떤 단계를 통해 작업이 진행될지 설계하여야 한다.
본 포스팅에서 구현한 프로젝트는 유저의 Persona를 입력받아
적당한 검색 쿼리를 생성하는 것이 최종 목표이다.
추후 이러한 쿼리를 토대로 실제 쇼핑몰에 입력하고, 결과를 받아 Agent가 처리하도록 만들기 위함이다.
이를 달성하기 위해 필자는 단계를 다음과 같이 나누어 구성해보았다.
- 입력된 Persona 단서를 활용하여, 임의의 페르소나를 가진 캐릭터를 생성하는 단계
- 생성된 페르소나를 활용하여 적당한 검색 문장을 생성하는 단계
- 검색 문장 리스트가 적합한지 검증하는 단계
3-1. 만약 해당 리스트가 적합하지 않다면, 해당 쿼리를 수정하는 단계
3-2. 만약 해당 리스트가 적합하다면, 해당 리스트를 반환
2. 그래프의 단순 Node 구현
Graph의 START의 경우에도 일종의 Node이기 때문에, 특정 State에 대한 Input을 요구한다.
따라서 User Input이 모든 단계의 첫 번째가 되는 것이 기본이다.
다만 필자는 System의 설명이 먼저 나오고, 그 다음 User의 Input을 받도록 설계하고 싶었다.
이러한 요구 때문에 START 다음으로 Input Stage를 구성하여 이를 달성하도록 만들었다.
코드를 통해 살펴보자.
# 먼저 필요한 라이브러리를 모두 임포트하자.
import os
from pydantic import BaseModel
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import tools_condition
from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
# API키를 가져와 입력해주자. 필자의 경우 txt로 저장된 파일로부터 가져와 입력하였다.
# 모델의 경우 OpenAI의 모델이 아닌 다른 모델을 사용하여도 무관하다.
# 그러한 경우 적절한 API키를 입력해두자.
with open('API_KEY.txt', 'r') as api:
os.environ["OPENAI_API_KEY"] = api.read()
먼저 START -> USER INPUT 단계를 만들어보자.
해당 단계는 None을 입력받아 User INPUT으로 전달하는 단계이다.
이후 USER INPUT은 System 메세지를 출력하고, 유저로부터 Input을 받을 것이다.
# State 정의
# START로부터 None의 입력을 받게 된다.
# 따라서 실제 필요한 인자는 아무것도 없어도 되지만
# Pass나 ... 을 입력해 class를 정의하면 에러가 발생한다.
class InputState(TypedDict):
start_input: str
# user_input을 받는 Node를 다음과 같이 작성하였다.
# start_input을 입력받아 (실제로는 아무것도 입력되지 않음)
# 아래의 기능을 수행하고, "user_input"이 담긴 Dict를 반환한다.
# 다음 Node의 경우 이러한 변수를 가진 State를 가지고 있어야 한다.
def user_input_node(state: InputState):
print("================================= Make Persona =================================")
print("페르소나를 결정합니다. 성별, 나이, 거주지, 취미 등 정보를 알려주세요.")
# time.sleep(1)
user_input = input("User: ")
return {"user_input": user_input}
이제 사용자의 입력값을 받았으니, 이를 토대로 정형화된 페르소나를 구축하고자 한다.
페르소나에 대한 정보를 문장 형태로 사용하여도 무관하겠지만
여기서는 정형화된 형태로 값들을 저장하고 싶다.
이에 따라 LLM을 활용하여, 주어진 문장으로부터 정형화된 페르소나 Dict를 만드는 Node를 설계하였다.
# 이전 Node에서 user_input을 전달하므로 이러한 변수가 반드시 포함되어야 한다.
# 이후 데이터를 만들어 chracter_persona_dict에 저장하려고 한다.
class PersonaState(TypedDict):
user_input: str
character_persona_dict: dict
# 노드 생성 전, LLM을 설계해보자.
# 해당 노드에서 작동하는 LLM은 다음과 같은 정보를 반환하도록 만들었다.
# Sturctured_output을 만들기 위해 아래와 같이 작성하였다.
class Persona_Output(TypedDict):
"""
Sturctured_output을 생성하기위한 클래스
"""
character_age: Annotated[str, ..., "An age of the Persona"]
character_sex: Annotated[str, ..., "A sex of the Persona"]
character_location: Annotated[str, ..., "A place where the persona might live"]
character_interest: Annotated[str, ..., "Interests that the persona might have"]
character_hobby: Annotated[str, ..., "Hobbies that the persona might have"]
persona_model = ChatOpenAI(model="gpt-4o-mini")
persona_model = persona_model.with_structured_output(Persona_Output)
# 이를 활용하는 노드를 다음과 같이 설계하였다.
# System Prompt를 통해 작업을 지시하였다. 만약 정확한 정보가 제시되지 않으면 임의의 값을 채워넣도록 요청했다.
def persona_setup_node(state: PersonaState):
messages = [
("system", """
You are the expert in determining your character's persona.
Extract the character's 'age', 'sex, 'location', 'interest', and 'hobbies' from the values entered by the user.
If no information is available, it will return a randomised set of appropriate information that must be entered.
Answers must be in Korean.
"""),
("human", state['user_input'])
]
response = persona_model.invoke(messages)
print("================================= Persona Setup =================================")
print(f"입력된 정보:{state['user_input']}")
print(f"성별: {response['character_sex']}")
print(f"나이: {response['character_age']}")
print(f"거주지: {response['character_location']}")
print(f"흥미: {response['character_interest']}")
print(f"취미: {response['character_hobby']}")
return {"character_persona_dict": response}
해당 노드는 LangChain의 with_structured_output를 활용해 dict형태의 Output을 Parsing한다.
Prompt를 통해, 유저 Input으로부터 주어진 5개의 정보를 채워넣는 역할을 지시하였다.
여기까지 진행했으면 실제로 어떻게 작동되는지 한 번 살펴보자.
이를 위해 필요한 요소들을 아래에 추가적으로 구현하였다.
# 그래프 전체적으로 사용할 State를 정의
class OverallState(TypedDict):
user_input: str
messages: Annotated[list, add_messages]
character_persona_dict: dict
# 그래프를 만드는 builder를 정의. input을 지정해주지 않으면
# OverallState를 START에서 Input으로 요구하게 됨.
graph_builder = StateGraph(OverallState, input=InputState)
# 그래프의 Node를 추가함. 노드의 이름과 노드 함수를 인자로 받음.
graph_builder.add_node("User Input", user_input_node)
graph_builder.add_node("Persona Setup", persona_setup_node)
# 그래프의 Edge를 추가함. 시작과 끝은 항상 START에서 END로 가야함
graph_builder.add_edge(START, "User Input")
graph_builder.add_edge("User Input", "Persona Setup")
graph_builder.add_edge("Persona Setup", END)
# 해당 그래프를 컴파일
graph = graph_builder.compile()
# 해당 그래프의 도식을 그려서 저장
with open("graph_output.png", "wb") as f:
f.write(graph.get_graph().draw_mermaid_png())
config = {"configurable": {"thread_id": "1"}}
# 그래프 호출. 아까 이야기 했듯 start_input에는 아무것도 입력하지 않음.
# 여기서도 마찬가지로 dict형태의 입력이 요구됨.
graph.invoke({"start_input": ""})
여기까지 구현한 그래프의 도식은 위와 같다.
아주 간단하게 Graph가 시작되고, User Input Node에서 유저의 입력을 받은 뒤
Persona Setup Node로 보내고 여기서 정형화된 값을 반환하도록 한다.
실제로 실행해보자.
해당 그래프를 invoke하면 바로 User Input Node로 진입하여 위와 같이 Input을 요구한다.
유저의 입력은 바로 Persona Setup Node로 전달되고, 여기서는 LLM이 명시된 작업을 수행한다.
이후 그래프는 종료되게 된다.
두 번째 노드까지는 매우 정상적으로 잘 작동하는 듯 보인다.
그러면 이어서 이러한 정보를 받아, 적당한 Sentence를 생성하는 과정까지 가보도록 하자.
# 주어진 정보를 통해 문장을 생성하는 역할을 수행할 것이다.
# 문장을 생성 후, 검증, 재작성 등의 작업을 수행하게 될 것이기 때문에
# Agent의 기능을 위한 messages와, 다른 변수들도 추가로 정의하였다.
class SearchQueryState(TypedDict):
messages: Annotated[list, add_messages]
character_persona_dict: dict
query_list: list
is_revise: bool
# 노드를 정의하기에 앞서 마찬가지로 견고한 출력을 위해 Parser를 작성하였다.
class Search_Output(TypedDict):
"""
Sturctured_output을 생성하기위한 클래스
"""
query_list: Annotated[list, ..., "List of queries that customers have entered in your shop"]
search_model = ChatOpenAI(model="gpt-4o")
search_model = search_model.with_structured_output(Search_Output)
# 출력의 성능을 보장하기 위해 LangChain에서 제공하는 FewshotPrompt를 작성하였다.
# 1-Shot 정보가 제공되지 않는 경우 성능이 상당히 떨어지는 모습이 관찰된다.
examples = [
{"input":
"""
User Sex: 여자,
User Age: 20대,
User Location: 서울 강남,
User Interest: 최신 화장법,
User Hobby: 공원 산책
""",
"output":
['피부진정용 필링패드', '수분에센스', '스틱형 파운데이션', '강아지 간식', '강아지용 배변패드', '강아지 장난감']
},
]
example_prompt = ChatPromptTemplate.from_messages(
[
("human", "{input}"),
("ai", "{output}"),
]
)
few_shot_prompt = FewShotChatMessagePromptTemplate(
example_prompt=example_prompt,
examples=examples,
)
# 위에서 정의된 기능들을 추가하여 문장을 생성하는 Node를 설계하였다.
def search_setence_node(state: SearchQueryState):
prompt = ChatPromptTemplate.from_messages([
("system","""
You're a great marketing manager, and you're working on inferring customer search queries.
Given the customer information, generate appropriate search quries that customers might enter to find products in your shopping mall.
Make sure to clearly present the actual product names that a user with that persona would search for in your retail mall.
"""),
few_shot_prompt,
("human", """
User Sex: {sex},
User Age: {age},
User Location: {location},
User Interest: {interest},
User Hobby: {hobby}
"""),
])
chain = prompt | search_model
response = chain.invoke(
{
"sex": state['character_persona_dict']['character_sex'],
"age": state['character_persona_dict']['character_age'],
"location": state['character_persona_dict']['character_location'],
"interest": state['character_persona_dict']['character_interest'],
"hobby": state['character_persona_dict']['character_hobby'],
}
)
print("=============================== Search Queries ===============================")
print(response['query_list'])
return {"query_list": response}
핵심은 여전히 Input의 State와 Output의 State만 잘 맞추어 주면 된다는 것이다.
이렇게 작성된 Node를 추가하여 다시 실행해보자.
# 노드가 추가되었으니, 노드와 엣지를 조금 수정하자
graph_builder.add_node("User Input", user_input_node)
graph_builder.add_node("Persona Setup", persona_setup_node)
graph_builder.add_node("Search Sentence", search_setence_node) # 새로 만든 노드를 추가하자.
graph_builder.add_edge(START, "User Input")
graph_builder.add_edge("User Input", "Persona Setup")
graph_builder.add_edge("Persona Setup", "Search Sentence") # 노드에 순서에 맞추어 엣지를 조금 변경하자.
graph_builder.add_edge("Search Sentence", END) # 이제 문장을 생성한 뒤 마무리 되어야 한다.
잘 작동하는 것을 볼 수 있다.
여기까지는 단방향으로 설계된 Node와 Graph를 작성하였다.
Input과 Output을 제대로 작성하고 연결한다면 손쉽게 Workflow를 작성할 수 있다.
이러한 방식으로 LLM을 사용하여 기능을 구현할 수도 있지만
LLM에게 작업을 결정하도록 구현하는 것도 가능하다.
3. Agent형태의 Node 구현
LangGraph에서 강력하게 지원하는 기능 중 하나는 바로 Agent 형태의 Node를 견고하게 설계할 수 있다는 것이다.
LLM이 스스로 작업을 결정하고, 이를 수행하도록 만드는 것이 핵심이다.
이를 구현하기 위해서는 State의 변수들을 추가하여 상태를 추적할 수 있도록 한 뒤
LLM의 tools_call 여부를 검사함으로써, 상태들을 업데이트할지 결정하고
add_conditional_edges 를 통해 현재 상태에 따른 경로를 결정하는 것이다.
언뜻 복잡해보이는 과정이지만 이를 구현하는 것은 크게 어렵지 않다. 천천히 따라가보자.
먼저 핵심적인 기능인 tools을 구현해보자.
# BaseModel을 이용한 커스텀 툴의 구현.
class QueryReviseAssistance(BaseModel):
"""Escalate the conversation.
Use only if the given search query is a strong mismatch with the customer's information.
Use this tool even if given search query is seriously inappropriate to enter into the search bar of an online retailer like Amazon.
Never call the tool if the same input is still being given as before.
To use this function, return 'query_list'.
"""
query_list: list
query_check_model = ChatOpenAI(model="gpt-4o-mini")
query_check_model = query_check_model.bind_tools([QueryReviseAssistance])
LangChain에서는 bind_tools이라는 method를 제공하여, 특정한 tool 들을 사용할 수 있도록 하고 있다.
커스텀 툴을 생성하며, 정확한 사용 방식을 문장의 형태로 제공하는 것이 전부이다.
이렇게 되면, LLM은 주어진 문장으로부터 특정한 작업이 필요하다고 생각하면 해당 tool을 호출하게 된다.
def query_check_node(state: SearchQueryState):
print("=============================== Query Check ===============================")
prompt = ChatPromptTemplate.from_messages([
("system","""
You are a search manager.
Based on a given customer persona, if you think that customer would search for the given queries, return the given queries as a list.
Never call the tool if the same input is still being given as before.
"""),
("human", """
User Sex: {sex},
User Age: {age},
User Location: {location},
User Interest: {interest},
User Hobby: {hobby}
Queries: {queries}
"""),
])
chain = prompt | query_check_model
response = chain.invoke(
{
"sex": state['character_persona_dict']['character_sex'],
"age": state['character_persona_dict']['character_age'],
"location": state['character_persona_dict']['character_location'],
"interest": state['character_persona_dict']['character_interest'],
"hobby": state['character_persona_dict']['character_hobby'],
"queries": state['query_list']['query_list'],
}
)
is_revise = False
if (
response.tool_calls
and response.tool_calls[0]["name"] == QueryReviseAssistance.__name__
):
print("Revise Requires")
is_revise = True
return {"messages": [response], "is_revise": is_revise}
노드의 구현은 거의 똑같으나 여기서는 output parser를 이용하지 않고 llm의 메세지를 출력하도록 한다.
이렇게 함으로써 LLM이 스스로 판단하여 tool_calls을 반환할 것인지, 아니면 AI 메세지를 반환할지 결정하게 된다.
여기서 tool_calls이 반환되고 호출된 tool이 앞서 정의한 것과 같다면 State를 업데이트 하도록 설계하였다.
- 즉 ... 문장과 작업이 주어지면
- LLM이 tools을 호출할지 그냥 답변을 생성할지 결정하고
- 이러한 결정이 메세지의 형태로 반환되며
- 이렇게 반환된 값을 보고 State를 업데이트할지 결정
이러한 과정을 통해 어떠한 Action을 취할지 결정하게 되는 것이다.
# 마찬가지로 리스트 형태의 반환을 위한 Output Parser
# 이전에 사용한 것과 거의 똑같지만, 구분을 위해 새로 하나를 더 정의함
class QueryCheck_Output(TypedDict):
"""
Sturctured_output을 생성하기위한 클래스
"""
query_list: Annotated[list, ..., "List of queries that customers might have entered in search-bar of your online retail shop"]
query_revise_model = ChatOpenAI(model="gpt-4o")
query_revise_model = query_revise_model.with_structured_output(QueryCheck_Output)
# 해당 기능을 수행하는 node 설정
def query_revise_node(state: SearchQueryState):
print("=============================== Query Revise ===============================")
prompt = ChatPromptTemplate.from_messages([
("system",
"""
You are a validator who fixes errors in a given query.
From the list of queries given, remove or modify the queries that do not match the user's information appropriately.
Be sure to delete highly irrelevant data.
Be sure to remove search terms that you wouldn't use on a shopping site like Amazon.
Return the modified queries as a list.
"""
),
("human",
"""
User Sex: {sex},
User Age: {age},
User Location: {location},
User Interest: {interest},
User Hobby: {hobby}
Queries: {queries}
"""
)])
chain = prompt | query_revise_model
response = chain.invoke(
{
"sex": state['character_persona_dict']['character_sex'],
"age": state['character_persona_dict']['character_age'],
"location": state['character_persona_dict']['character_location'],
"interest": state['character_persona_dict']['character_interest'],
"hobby": state['character_persona_dict']['character_hobby'],
"queries": state['query_list'],
}
)
print(response['query_list'])
return {"query_list": response, "is_revise": False}
이후 해당 tool이 호출되었을 때 기능을 수행하는 노드를 만들어준다.
마지막으로 조건부 엣지를 만들기 위해 커스텀 라우팅 함수를 만들어준다.
# 커스텀 라우팅 함수 추가.
def select_next_node(state: SearchQueryState):
if state["is_revise"]:
return "is_revise"
return tools_condition(state)
여기서 tools_condition 함수는 LangGraph에서 기본적으로 제공하는 함수이다.
해당 함수는 tool_calls 및 END를 라우팅하는 함수이지만
위와 같이 state를 추가적으로 넣어서 반환되는 값을 다르게 만들 수 있다.
해당 함수는 앞서 설명한 것과 같이 state를 통해 조건을 검증한 뒤
Literal한 문자를 반환하는 함수이다.
이렇게 정의된 함수는 조건에 따라
'is_revise', 'tools', '__end__' 셋 중 하나를 반환하게 된다(tools와 __end__는 기본으로 존재함)
앞서 이야기 하였듯이 Edge는 일종의 라우팅이라는 기능을 수행하는 Node이기 때문에
만약 다른 방식을 구현하고 싶다면 다른 방식으로도 구현이 가능하다.
def select_next_node(state: SearchQueryState):
if state["is_revise"]:
return "is_revise"
return "__end__"
예를 들어 위와 같이 단순하게 수정인지 아닌지만 검사해야 한다면 이처럼 구현해도 무관하다.
# (출발노드, 라우팅을 담당하는 함수, 라우팅 함수가 반환한 값에 따라 어떤 노드로 연결할지를 지시해주는 DICT)
graph_builder.add_conditional_edges(
"Query Check",
select_next_node,
{"is_revise": "Query Revise Tool", END: END}
)
이후 마지막으로 다음과 같이 조건부 엣지를 생성해주면 끝이다.
select_next_node 함수에 의해 is_revise가 반환되면 Query Revise Tool로 연결되며
__end__가 반환되면 END 노드로 연결되게 된다.
이런 방식을 통해 최종적으로 이런 모습이 구현된다.
Query Check 노드에서는 LLM이 스스로 Query Revise Tool이라는 도구를 호출할지 결정하고
이를 토대로 Query의 완성도 향상시키게 된다.
실제 출력결과를 보면 위와 같다.
모델은 자의적으로 Queries의 수정이 필요하다고 생각하였고 Revise Tool을 호출하였다.
Revise Tool은 검증을 수행하였지만, 같은 값을 반환하게 되었고, 이를 몇 번 반복하다가 종료되는 모습을 보이고 있다.
이처럼 Agent는 특정 분기에서 자신의 Action을 선택할 수 있으며
이러한 Agent를 다중으로 배치하거나, 여러 분기에 배치함으로써 더욱 정밀한 LLM 기반 Product를 개발할 수 있다.
다음에는 RAG 시스템 및 검색 시스템에 대해서도 알아보자.
전체 코드
더보기
# 라이브러리 임포트.
# 대부분의 상용 LLM 모델을 Langchain 자체 라이브러리에서 로드해야함. 안 그러면 쓰는게 너무 어려움
import os
import time
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_core.messages import BaseMessage, FunctionMessage, HumanMessage
from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate
from langgraph.checkpoint.memory import MemorySaver
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_community.agent_toolkits import PlayWrightBrowserToolkit
from langchain_community.tools.playwright.utils import create_async_playwright_browser
from langchain_openai import ChatOpenAI
from typing_extensions import TypedDict
from pydantic import BaseModel
# API 키 미리 로드해서 환경변수에 입력.
with open('API_KEY.txt', 'r') as api:
os.environ["OPENAI_API_KEY"] = api.read()
with open('TAVILY_API.txt', 'r') as api:
os.environ["TAVILY_API_KEY"] = api.read()
# 메모리 기능. 이전 대화 기록.
memory = MemorySaver()
# State는 노드가 다루는 변수이기도 하면서, 전역적인 상태를 다루게 됨.
# 모든 노드는 TypedDict를 다루기 때문에 모든 입력에서 Dict를 요구함(다른걸 쓰면 다른거 요구)
# 여기서 개별 딕트의 변수가 필요하고, 모든 변수가 입력될 필요는 없음.
class OverallState(TypedDict):
user_input: str
messages: Annotated[list, add_messages]
character_persona_dict: dict
class InputState(TypedDict):
start_input: str
class PersonaState(TypedDict):
user_input: str
character_persona_dict: dict
class SearchQueryState(TypedDict):
messages: Annotated[list, add_messages]
character_persona_dict: dict
query_list: list
previous_query: list
is_revise: bool
class EndState(TypedDict):
messages: Annotated[list, add_messages]
query_list: list
# Node는 단순하게 생각하면 그냥 함수임. 근데 State로 진행을 결정하는.
# 시작노드 - 페르소나에 대한 정보를 요구하는 노드임
def user_input_node(state: InputState):
print("================================= Make Persona =================================")
print("페르소나를 결정합니다. 성별, 나이, 거주지, 취미 등 정보를 알려주세요.")
# time.sleep(1)
user_input = input("User: ")
return {"user_input": user_input}
# 노드2 - 입력된 문장으로부터 페르소나에 관한 정보를 추출하고, 정보가 없는 경우 이를 채워넣는 노드.
class Persona_Output(TypedDict):
"""
Sturctured_output을 생성하기위한 클래스
"""
character_age: Annotated[str, ..., "An age of the Persona"]
character_sex: Annotated[str, ..., "A sex of the Persona"]
character_location: Annotated[str, ..., "A place where the persona might live"]
character_interest: Annotated[str, ..., "Interests that the persona might have"]
character_hobby: Annotated[str, ..., "Hobbies that the persona might have"]
persona_model = ChatOpenAI(model="gpt-4o-mini")
persona_model = persona_model.with_structured_output(Persona_Output)
# 페르소나를 반환하는 매우 경직된 LLM.
# 정보가 없는 경우 임의의 값을 채워넣도록 되어있음.
def persona_setup_node(state: PersonaState):
messages = [
("system", """
You are the expert in determining your character's persona.
Extract the character's 'age', 'sex, 'location', 'interest', and 'hobbies' from the values entered by the user.
If no information is available, it will return a randomised set of appropriate information that must be entered.
Answers must be in Korean.
"""),
("human", state['user_input'])
]
response = persona_model.invoke(messages)
print("================================= Persona Setup =================================")
print(f"입력된 정보:{state['user_input']}")
print(f"성별: {response['character_sex']}")
print(f"나이: {response['character_age']}")
print(f"거주지: {response['character_location']}")
print(f"흥미: {response['character_interest']}")
print(f"취미: {response['character_hobby']}")
return {"character_persona_dict": response}
# 노드 3 - 페르소나를 토대로 적절한 검색 키워드를 생성하는 놈. 매우 랜덤하게 했으면 좋겠는데 페르소나가 한정적인게 문제.
# 랜덤한 상황적 설명을 주는 것이 좋다고 생각하는데 어떻게 할지 결정해야 할듯. (이걸 툴로 만들면 참 좋을듯.)
class Search_Output(TypedDict):
"""
Sturctured_output을 생성하기위한 클래스
"""
query_list: Annotated[list, ..., "List of queries that customers have entered in your shop"]
search_model = ChatOpenAI(model="gpt-4o")
search_model = search_model.with_structured_output(Search_Output)
examples = [
{"input":
"""
User Sex: 여자,
User Age: 20대,
User Location: 서울 강남,
User Interest: 최신 화장법,
User Hobby: 공원 산책
""",
"output":
['피부진정용 필링패드', '수분에센스', '스틱형 파운데이션', '강아지 간식', '강아지용 배변패드', '강아지 장난감']
},
]
example_prompt = ChatPromptTemplate.from_messages(
[
("human", "{input}"),
("ai", "{output}"),
]
)
few_shot_prompt = FewShotChatMessagePromptTemplate(
example_prompt=example_prompt,
examples=examples,
)
def search_setence_node(state: SearchQueryState):
prompt = ChatPromptTemplate.from_messages([
("system","""
You're a great marketing manager, and you're working on inferring customer search queries.
Given the customer information, generate appropriate search quries that customers might enter to find products in your shopping mall.
Make sure to clearly present the actual product names that a user with that persona would search for in your retail mall.
"""),
few_shot_prompt,
("human", """
User Sex: {sex},
User Age: {age},
User Location: {location},
User Interest: {interest},
User Hobby: {hobby}
"""),
])
chain = prompt | search_model
response = chain.invoke(
{
"sex": state['character_persona_dict']['character_sex'],
"age": state['character_persona_dict']['character_age'],
"location": state['character_persona_dict']['character_location'],
"interest": state['character_persona_dict']['character_interest'],
"hobby": state['character_persona_dict']['character_hobby'],
}
)
print("=============================== Search Queries ===============================")
print(response['query_list'])
return {"query_list": response}
# 노드 4, revise_tool - 반환된 서치쿼리가 적당한지 검증하는 노드임.
class QueryReviseAssistance(BaseModel):
"""Escalate the conversation.
Use only if the given search query is a strong mismatch with the customer's information.
Use this tool even if given search query is seriously inappropriate to enter into the search bar of an online retailer like Amazon.
Never call the tool if the same input is still being given as before.
To use this function, return 'query_list'.
"""
query_list: list
query_check_model = ChatOpenAI(model="gpt-4o-mini", temperature=1.3)
query_check_model = query_check_model.bind_tools([QueryReviseAssistance])
def query_check_node(state: SearchQueryState):
print("=============================== Query Check ===============================")
prompt = ChatPromptTemplate.from_messages([
("system","""
You are a search manager.
Based on a given customer persona, if you think that customer would search for the given queries, return the given queries as a list.
"""),
("human", """
User Sex: {sex},
User Age: {age},
User Location: {location},
User Interest: {interest},
User Hobby: {hobby}
Queries: {queries}
"""),
])
chain = prompt | query_check_model
response = chain.invoke(
{
"sex": state['character_persona_dict']['character_sex'],
"age": state['character_persona_dict']['character_age'],
"location": state['character_persona_dict']['character_location'],
"interest": state['character_persona_dict']['character_interest'],
"hobby": state['character_persona_dict']['character_hobby'],
"queries": state['query_list']['query_list'],
}
)
is_revise = False
if (
response.tool_calls
and response.tool_calls[0]["name"] == QueryReviseAssistance.__name__
):
print("Revise Requires")
is_revise = True
return {"messages": [response], "is_revise": is_revise}
class QueryCheck_Output(TypedDict):
"""
Sturctured_output을 생성하기위한 클래스
"""
query_list: Annotated[list, ..., "List of queries that customers might have entered in search-bar of your online retail shop"]
query_revise_model = ChatOpenAI(model="gpt-4o")
query_revise_model = query_revise_model.with_structured_output(QueryCheck_Output)
def query_revise_node(state: SearchQueryState):
print("=============================== Query Revise ===============================")
prompt = ChatPromptTemplate.from_messages([
("system",
"""
You are a validator who fixes errors in a given query.
From the list of queries given, remove or modify the queries that do not match the user's information appropriately.
Be sure to delete highly irrelevant data.
Be sure to remove search terms that you wouldn't use on a shopping site like Amazon.
Return the modified queries as a list.
"""
),
("human",
"""
User Sex: {sex},
User Age: {age},
User Location: {location},
User Interest: {interest},
User Hobby: {hobby}
Queries: {queries}
"""
)])
chain = prompt | query_revise_model
response = chain.invoke(
{
"sex": state['character_persona_dict']['character_sex'],
"age": state['character_persona_dict']['character_age'],
"location": state['character_persona_dict']['character_location'],
"interest": state['character_persona_dict']['character_interest'],
"hobby": state['character_persona_dict']['character_hobby'],
"queries": state['query_list'],
}
)
print(response['query_list'])
return {"query_list": response, "is_revise": False}
# 그래프 빌드 및 노드 / 엣지 추가
# 커스텀 노드 라우팅 함수 추가.
def select_next_node(state: SearchQueryState):
if state["is_revise"]:
return "is_revise"
return '__end__'
graph_builder = StateGraph(OverallState, input=InputState, output=EndState)
graph_builder.add_node("User Input", user_input_node)
graph_builder.add_node("Persona Setup", persona_setup_node)
graph_builder.add_node("Search Sentence", search_setence_node)
graph_builder.add_node("Query Check", query_check_node)
graph_builder.add_node("Query Revise Tool", query_revise_node)
graph_builder.add_edge(START, "User Input")
graph_builder.add_edge("User Input", "Persona Setup")
graph_builder.add_edge("Persona Setup", "Search Sentence")
graph_builder.add_edge("Search Sentence", "Query Check")
graph_builder.add_edge("Query Revise Tool", "Query Check")
graph_builder.add_conditional_edges(
"Query Check",
select_next_node,
{"is_revise": "Query Revise Tool", END: END}
)
graph = graph_builder.compile(checkpointer=memory)
# graph = graph_builder.compile()
with open("graph_output.png", "wb") as f:
f.write(graph.get_graph().draw_mermaid_png())
config = {"configurable": {"thread_id": "1"}}
graph.invoke({"start_input": ""}, config)'A.I.(인공지능) & M.L.(머신러닝) > LLM' 카테고리의 다른 글
| [Anthropic Computer-Use] LLM Cluade를 이용한 자동화 에이전트(11.28 내용 추가) (0) | 2024.11.22 |
|---|---|
| [LangGrpah] 4. LangGraph를 이용한 RAG 및 검색 Agent 개발 (24) | 2024.11.20 |
| [LangGrpah] 2. LangGrpah(랭그래프)의 핵심 요소의 개념적 이해 (0) | 2024.11.13 |
| [LangGrpah] 1. LangGrpah(랭그래프)를 이용한 AI Workflow 관리하기 (0) | 2024.11.11 |
| Adapting While Learning (2) | 2024.11.10 |
