Spaces:
Sleeping
Sleeping
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() | |