customer_agent / app.py
omvishesh's picture
Update app.py
94eeb73 verified
import gradio as gr
from huggingface_hub import InferenceClient
from typing import TypedDict, Dict
from langgraph.graph import StateGraph, END
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables.graph import MermaidDrawMethod
from IPython.display import display, Image
class State(TypedDict):
query: str
category: str
sentiment: str
response: str
from langchain_groq import ChatGroq
llm = ChatGroq(
temperature=0,
groq_api_key="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",
model_name="llama-3.3-70b-versatile"
)
def categorize(state: State) -> State:
"""Categorize the query."""
prompt = ChatPromptTemplate.from_template(
"Categorize the following customer query into one of these categories: "
"Technical, Billing, General. Query: {query}"
)
chain = prompt | llm
category = chain.invoke({"query": state["query"]}).content
return {"category": category}
def analyze_sentiment(state: State) -> State:
"""Analyze sentiment of the query."""
prompt = ChatPromptTemplate.from_template(
"Analyze the sentiment of the following customer query "
"Response with either 'Positive', 'Neutral', or 'Negative'. Query: {query}"
)
chain = prompt | llm
sentiment = chain.invoke({"query": state["query"]}).content
return {"sentiment": sentiment}
def handle_technical(state: State) -> State:
"""Handle technical queries."""
prompt = ChatPromptTemplate.from_template(
"Provide a technical support response to the following query: {query}"
)
chain = prompt | llm
response = chain.invoke({"query": state["query"]}).content
return {"response": response}
def handle_billing(state: State) -> State:
"""Handle billing queries."""
prompt = ChatPromptTemplate.from_template(
"Provide a billing support response to the following query: {query}"
)
chain = prompt | llm
response = chain.invoke({"query": state["query"]}).content
return {"response": response}
def handle_general(state: State) -> State:
"""Handle general queries."""
prompt = ChatPromptTemplate.from_template(
"Provide a general support response to the following query: {query}"
)
chain = prompt | llm
response = chain.invoke({"query": state["query"]}).content
return {"response": response}
def escalate(state: State) -> State:
"""Escalate negative sentiment queries."""
return {"response": "This query has been escalated to a human agent due to its negative sentiment."}
def route_query(state: State) -> State:
"""Route query based on category and sentiment."""
if state["sentiment"] == "Negative":
return "escalate"
elif state["category"] == "Technical":
return "handle_technical"
elif state["category"] == "Billing":
return "handle_billing"
else:
return "handle_general"
workflow = StateGraph(State)
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)
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)
workflow.set_entry_point("categorize")
app = workflow.compile()
# Define the function that integrates the workflow.
def run_customer_support(query: str) -> Dict[str, str]:
results = app.invoke({"query": query})
return {
"Category": results['category'],
"Sentiment": results['sentiment'],
"Response": results['response']
}
"""
For more information on `huggingface_hub` Inference API support, please check the docs: https://huggingface.co/docs/huggingface_hub/v0.22.2/en/guides/inference
"""
client = InferenceClient("HuggingFaceH4/zephyr-7b-beta")
def respond(
message,
history: list[tuple[str, str]],
system_message,
max_tokens,
temperature,
top_p
):
messages = [{"role": "system", "content": system_message}]
for val in history:
if val[0]:
messages.append({"role": "user", "content": val[0]})
if val[1]:
messages.append({"role": "assistant", "content": val[1]})
messages.append({"role": "user", "content": message})
response = ""
# Simulate streaming from the client
for message in client.chat_completion(
messages,
max_tokens=max_tokens,
stream=True,
temperature=temperature,
top_p=top_p
):
token = message.choices[0].delta.content
response += token
yield response
# Define a custom Gradio Chat Interface with hidden sliders
with gr.Blocks() as demo:
gr.Markdown("### AI-Powered Customer Support Assistant")
chatbot = gr.ChatInterface(
respond,
additional_inputs=[
gr.Textbox(
value="You are a friendly chatbot.",
label="System Message",
info="Customize how the assistant behaves in conversations."
),
gr.Slider(
minimum=1,
maximum=2048,
value=512,
step=1,
label="Max New Tokens",
visible=False
),
gr.Slider(
minimum=0.1,
maximum=4.0,
value=0.7,
step=0.1,
label="Temperature",
visible=False
),
gr.Slider(
minimum=0.1,
maximum=1.0,
value=0.95,
step=0.05,
label="Top-p (Nucleus Sampling)",
visible=False
),
]
)
gr.Markdown("### Instructions")
gr.Textbox(
value="Enter your query, select response settings, and start the conversation.",
interactive=False,
)
if __name__ == "__main__":
demo.launch()