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()