고객 문의를 분류하고, 감정을 분석하며, 필요에 따라 적절한 응답을 제공하거나 문제를 에스컬레이션(담당자에게 전달)하도록 설계.
주요 구성 요소
- 상태 관리 (State Management): TypedDict를 사용하여 각 고객 문의의 상태를 정의하고 관리합니다.
- 문의 분류 (Query Categorization): 고객 문의를 기술(Technical), 요금(Billing), 일반(General) 카테고리로 분류합니다.
- 감정 분석 (Sentiment Analysis): 고객 문의의 **감정적 톤(긍정, 중립, 부정)**을 판단합니다.
- 응답 생성 (Response Generation): 문의 유형과 감정에 따라 적절한 응답을 생성합니다.
- 에스컬레이션 메커니즘 (Escalation Mechanism): 부정적인 감정이 포함된 문의를 자동으로 인간 상담원에게 전달합니다.
LangGraph 개념
from typing import Dict, TypedDict
from langgraph.graph import StateGraph, END
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from IPython.display import display, Image
from langchain_core.runnables.graph import MermaidDrawMethod
from dotenv import load_dotenv
import os
# Load environment variables and set OpenAI API key
load_dotenv()
os.environ["OPENAI_API_KEY"] = os.getenv('OPENAI_API_KEY')
Global State 구조체 정의
class State(TypedDict):
query: str
category: str
sentiment: str
response: str
노드에 연결할 함수 정의
def categorize(state: State) -> State:
"""고객 문의를 기술(Technical), 청구(Billing), 또는 일반(General) 범주로 분류."""
prompt = ChatPromptTemplate.from_template(
"다음 고객 문의를 다음 범주 중 하나로 분류하십시오:: "
"Technical, Billing, General. Query: {query}"
)
chain = prompt | ChatOpenAI(temperature=0)
category = chain.invoke({"query": state["query"]}).content
return {"category": category}
def analyze_sentiment(state: State) -> State:
"""고객 문의의 감정을 긍정(Positive), 중립(Neutral), 또는 부정(Negative) 으로 분석"""
prompt = ChatPromptTemplate.from_template(
"다음 고객 문의의 감정을 분석하시오."
"Respond with either 'Positive', 'Neutral', or 'Negative'. Query: {query}"
)
chain = prompt | ChatOpenAI(temperature=0)
sentiment = chain.invoke({"query": state["query"]}).content
return {"sentiment": sentiment}
def handle_technical(state: State) -> State:
"""해당 문의에 대한 기술 지원 답변을 제공"""
prompt = ChatPromptTemplate.from_template(
"다음 문의에 대한 기술 지원 답변을 제공하시오: {query}"
)
chain = prompt | ChatOpenAI(temperature=0)
response = chain.invoke({"query": state["query"]}).content
return {"response": response}
def handle_billing(state: State) -> State:
"""해당 문의에 대한 결제 지원 답변을 제공"""
prompt = ChatPromptTemplate.from_template(
"다음 문의에 대한 결제 지원 답변을 제공하시오.: {query}"
)
chain = prompt | ChatOpenAI(temperature=0)
response = chain.invoke({"query": state["query"]}).content
return {"response": response}
def handle_general(state: State) -> State:
"""해당 문의에 대한 일반 지원 답변을 제공"""
prompt = ChatPromptTemplate.from_template(
"다음 문의에 대한 일반 지원 답변을 제공하시오.: {query}"
)
chain = prompt | ChatOpenAI(temperature=0)
response = chain.invoke({"query": state["query"]}).content
return {"response": response}
def escalate(state: State) -> State:
"""부정적인 감정이 감지되었으므로, 해당 문의를 인간 상담원에게 이관"""
return {"response": "해당 문의는 부정적인 감정으로 인해 인간 상담원에게 이관되었습니다."}
def route_query(state: State) -> str:
"""문의의 감정 및 카테고리에 따라 해당 문의를 적절히 배정"""
if state["sentiment"] == "Negative":
return "escalate"
elif state["category"] == "Technical":
return "handle_technical"
elif state["category"] == "Billing":
return "handle_billing"
else:
return "handle_general"
Graph 생성후 node 추가, edge 추가
# Create the graph
workflow = StateGraph(State)
# Add nodes
workflow.add_node("categorize", categorize)
workflow.add_node("analyze_sentiment", analyze_sentiment)
workflow.add_node("handle_technical", handle_technical)
workflow.add_node("handle_billing", handle_billing)
workflow.add_node("handle_general", handle_general)
workflow.add_node("escalate", escalate)
# Add edges
workflow.add_edge("categorize", "analyze_sentiment")
workflow.add_conditional_edges(
"analyze_sentiment",
route_query,
{
"handle_technical": "handle_technical",
"handle_billing": "handle_billing",
"handle_general": "handle_general",
"escalate": "escalate"
}
)
workflow.add_edge("handle_technical", END)
workflow.add_edge("handle_billing", END)
workflow.add_edge("handle_general", END)
workflow.add_edge("escalate", END)
# Set entry point
workflow.set_entry_point("categorize")
# Compile the graph
app = workflow.compile()
실행함수 정의
def run_customer_support(query: str) -> Dict[str, str]:
"""Process a customer query through the LangGraph workflow.
Args:
query (str): The customer's query
Returns:
Dict[str, str]: A dictionary containing the query's category, sentiment, and response
"""
results = app.invoke({"query": query})
return {
"category": results["category"],
"sentiment": results["sentiment"],
"response": results["response"]
}
상황별 입력
# escalate
query = "My internet connection keeps dropping. Can you help?"
result = run_customer_support(query)
print(f"Query: {query}")
print(f"Category: {result['category']}")
print(f"Sentiment: {result['sentiment']}")
print(f"Response: {result['response']}")
print("\n")
# handle_technical
query = "I need help talking to chatGPT"
result = run_customer_support(query)
print(f"Query: {query}")
print(f"Category: {result['category']}")
print(f"Sentiment: {result['sentiment']}")
print(f"Response: {result['response']}")
print("\n")
# handle_billing
query = "where can i find my receipt?"
result = run_customer_support(query)
print(f"Query: {query}")
print(f"Category: {result['category']}")
print(f"Sentiment: {result['sentiment']}")
print(f"Response: {result['response']}")
print("\n")
# handle_general
query = "What are your business hours?"
result = run_customer_support(query)
print(f"Query: {query}")
print(f"Category: {result['category']}")
print(f"Sentiment: {result['sentiment']}")
print(f"Response: {result['response']}")
'data science > AI Agents' 카테고리의 다른 글
| CrewAi Flows - 개요, 설치방법 (0) | 2025.10.24 |
|---|---|
| CWD(Coordinator Worker Delegator) Model (0) | 2025.10.20 |
| CrewAI & LangChain - 영업 통화 분석 (1) | 2025.10.16 |
| agno - Memory 관리 (0) | 2025.10.16 |
| agno - 웹사이트 정보 취득후 블로그 컨텐츠 제작 Agent (0) | 2025.10.16 |