목차
- 작성 개요
- 상태 정의
- 상담사, 고객 역할 정의
- 에이전트 시뮬레이션 정의하기
- 그래프 정의 하기
- 실행
1. 작성 개요
그럼 LangGraph와 Agent의 사용 양상을 더욱 다각도로 확인해볼 필요가 있다. 첫번째로는 시뮬레이션이다.
에이전트를 활용할 수 있는 다양한 방법 중 하나로 해당 게시물에서는 간단한 대화 시뮬레이션을 정리하려고 한다.
대화 시뮬레이션은 내가 챗봇을 만들었을 때 그 성능을 점검해 볼 때, 하는 것이 적절한다. 사람이 만든 챗봇을 하나하나 테스트 하는 것이 어렵기 때문에, 이와 같이 유저 역할을 하는 체인을 하나 더 만들어 시뮬레이션을 하면, 그 품이 줄어들 수 있다.
해당 게시물은 상담사와 고객의 대화를 시뮬레이션 할 것이다. 이와 같은 관계를 학습하면서 다양한 상황, 그리고 2명 이상의 사람들의 대화를 시뮬레이션 해볼 수 있을 것 같다.
해당 게시물은 테디 노트님의 유료 강의를 듣고 복습차 정리한 게시물이다. 게시물이 부족하다고 여겨지면 유료 강의를 구매할 것을 추천한다.
2. 상태 정의
from langgraph.graph.message import add_messages
from typing import Annotated
from typing_extensions import TypedDict
# State 정의
class State(TypedDict):
messages: Annotated[list, add_messages] # 사용자 - 상담사 간의 대화 메시지기본적인 형태의 상태이고, 해당 class를 보면 사용자와 상담사 간의 대화 메시지를 모두 한 리스트에 모아 보관할 것임을 확인할 수 있다.
3. 상담사, 고객 역할 정의
1) 상담사 역할을 하는 챗봇
from typing import List
from langchain_teddynote.models import LLMs, get_model_name
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage
from langchain_core.output_parsers import StrOutputParser
# 모델 이름 설정
MODEL_NAME = get_model_name(LLMs.GPT4)
def call_chatbot(messages: List[BaseMessage]) -> dict:
# LangChain ChatOpenAI 모델을 Agent 로 변경할 수 있습니다.
prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"You are a customer support agent for an airline. Answer in Korean.",
),
MessagesPlaceholder(variable_name="messages"),
]
)
model = ChatOpenAI(model=MODEL_NAME, temperature=0.6)
chain = prompt | model | StrOutputParser()
return chain.invoke({"messages": messages})2) 고객 역할을 하는 챗봇
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
def create_scenario(name: str, instructions: str):
# 시스템 프롬프트를 정의: 필요에 따라 변경
system_prompt_template = """You are a customer of an airline company. \
You are interacting with a user who is a customer support person. \
Your name is {name}.
# Instructions:
{instructions}
[IMPORTANT]
- When you are finished with the conversation, respond with a single word 'FINISHED'
- You must speak in Korean."""
# 대화 메시지와 시스템 프롬프트를 결합하여 채팅 프롬프트 템플릿을 생성합니다.
prompt = ChatPromptTemplate.from_messages(
[
("system", system_prompt_template),
MessagesPlaceholder(variable_name="messages"),
]
)
# 특정 사용자 이름과 지시사항을 사용하여 프롬프트를 부분적으로 채웁니다.
prompt = prompt.partial(name=name, instructions=instructions)
return prompt
# 사용자 지시사항을 정의합니다.
instructions = """You are trying to get a refund for the trip you took to Paris. \
You want them to give you ALL the money back. This trip happened last year."""
# 사용자 이름을 정의합니다.
name = "Jiwoo"
create_scenario(name, instructions).pretty_print()
# OpenAI 챗봇 모델을 초기화합니다.
model = ChatOpenAI(model=MODEL_NAME, temperature=0.6)
# 시뮬레이션된 사용자 대화를 생성합니다.
simulated_user = create_scenario(name, instructions) | model | StrOutputParser()
from langchain_core.messages import HumanMessage
# 시뮬레이션된 사용자에게 메시지를 전달(상담사 -> 고객)
messages = [HumanMessage(content="안녕하세요? 어떻게 도와 드릴까요?")]
simulated_user.invoke({"messages": messages})
=======결과======
안녕하세요. 저는 작년에 파리로 여행을 갔었는데, 그 여행에 대한 환불을 요청하고 싶습니다. 모든 돈을 되돌려받고 싶습니다. 도와주실 수 있나요?그리고 테스트하는 사람 역할의 챗봇도 아무 설정 없이 대화를 할 수 없기 때문에 시나리오를 만들어야 하고, 시나리오는 프롬프트 형태로 만들 수 있다.
4. 에이전트 시뮬레이션 정의하기
1) 노드 만들기
from langchain_core.messages import AIMessage
# 상담사 역할(AI Assistant) 노드 정의
def ai_assistant_node(state: State):
# 상담사 응답 호출
ai_response = call_chatbot(state["messages"])
# AI 상담사의 응답을 반환
return {"messages": [("assistant", ai_response)]}
# 시뮬레이션된 사용자(Simulated User) 노드 정의
def simulated_user_node(state: State):
# 메시지 타입을 교환: AI -> Human, Human -> AI
new_messages = _swap_roles(state["messages"])
# 시뮬레이션된 사용자를 호출
response = simulated_user.invoke({"messages": new_messages})
return {"messages": [("user", response)]}
대화는 두명이 하기 때문에 노드는 2개, 아까 만들어둔 시뮬레이션 된 사용자와 상담사를 노드로 만들어준다.
2) swap roles
def _swap_roles(messages):
# 메시지의 역할을 교환: 시뮬레이션 사용자 단계에서 메시지 타입을 AI -> Human, Human -> AI 로 교환합니다.
new_messages = []
for m in messages:
if isinstance(m, AIMessage):
# AIMessage 인 경우, HumanMessage 로 변환합니다.
new_messages.append(HumanMessage(content=m.content))
else:
# HumanMessage 인 경우, AIMessage 로 변환합니다.
new_messages.append(AIMessage(content=m.content))
return new_messages해당 구조에서 가장 중요한 테크닉이라고 보인다. 우선 양쪽의 대화를 같은 리스트 안에 모아 두고 있기 때문에, 정리가 필요하기도 하고, 두 노드가 대화를 하고 있기 때문에 생기는 본질적인 문제를 해결하기 위함이다.
우선 어떤 채팅형 모델이든 결과를 반환할 때는 AI메시지로 반환한다. 하지만 입력을 받을 때는 Human메시지로 받아야 한다. 하지만 두 노드 모두 AI메시지로 반환하기 때문에 적절한 대화가 이루어지기 위해서는 AI메시지를 Human 메시지로 바꿔서 전달해야 한다. 이와 같은 부분을 개선하기 위한 태크닉이다.
5. 그래프 정의 하기
1) 종료 시점 로직
def should_continue(state: State):
# 메시지 리스트의 길이가 6보다 크면 'end'를 반환합니다.
if len(state["messages"]) > 6:
return "end"
# 마지막 메시지의 내용이 'FINISHED'라면 'end'를 반환합니다.
elif state["messages"][-1].content == "FINISHED":
return "end"
# 위의 조건에 해당하지 않으면 'continue'를 반환합니다.
else:
return "continue"위에 상담사 노드 프롬프트에 대화 종료시점에 [FINISHED]라는 결과를 반환하라고 언급했고, 해당 내용이 나오면 대화가 종료될 수 있도록 해야 한다.
2) 그래프 정의
from langgraph.graph import END, StateGraph
# StateGraph 인스턴스 생성
graph_builder = StateGraph(State)
# 노드 정의
graph_builder.add_node("simulated_user", simulated_user_node)
graph_builder.add_node("ai_assistant", ai_assistant_node)
# 엣지 정의 (챗봇 -> 시뮬레이션된 사용자)
graph_builder.add_edge("ai_assistant", "simulated_user")
# 조건부 엣지 정의
graph_builder.add_conditional_edges(
"simulated_user",
should_continue,
{
"end": END, # 종료 조건이 충족되면 시뮬레이션을 중단
"continue": "ai_assistant", # 종료 조건이 충족되지 않으면 상담사 역할 노드로 메시지를 전달
},
)
# 시작점 설정
graph_builder.set_entry_point("ai_assistant")
# 그래프 컴파일
simulation = graph_builder.compile()구조는 간단하다. 둘이 대화하다, 목표가 달성되어 FINISHED 받으면 대화를 종료하면 된다.
6. 실행
from langchain_core.runnables import RunnableConfig
from langchain_teddynote.messages import stream_graph, random_uuid
# config 설정(재귀 최대 횟수, thread_id)
config = RunnableConfig(recursion_limit=10, configurable={"thread_id": random_uuid()})
# 입력 메시지 설정
inputs = {
"messages": [HumanMessage(content="안녕하세요? 저 지금 좀 화가 많이 났습니다^^")]
}
# 그래프 스트리밍
stream_graph(simulation, inputs, config, node_names=["simulated_user", "ai_assistant"])
프로그래밍 기록 + 공부 기록