''' import gradio as gr from huggingface_hub import InferenceClient """ 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 = "" 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 """ For information on how to customize the ChatInterface, peruse the gradio docs: https://www.gradio.app/docs/chatinterface """ demo = gr.ChatInterface( respond, additional_inputs=[ gr.Textbox(value="You are a friendly Chatbot.", label="System message"), gr.Slider(minimum=1, maximum=2048, value=512, step=1, label="Max new tokens"), gr.Slider(minimum=0.1, maximum=4.0, value=0.7, step=0.1, label="Temperature"), gr.Slider( minimum=0.1, maximum=1.0, value=0.95, step=0.05, label="Top-p (nucleus sampling)", ), ], ) if __name__ == "__main__": demo.launch() import gradio as gr from langchain.chains import LLMChain from langchain.prompts import PromptTemplate from langchain_huggingface import HuggingFaceEndpoint from langgraph.graph import StateGraph,END,START from typing import TypedDict class InputState(TypedDict): string_var :str numeric_var :int def changeState(input: InputState): print(f"Current value: {input}") return input # Define the LLM models llm1 = HuggingFaceEndpoint(model='t5-small') llm2 = HuggingFaceEndpoint(model='t5-large') # Define the agent functions def agent1(response): return f"Agent 1: {response}" def agent2(response): return f"Agent 2: {response}" # Define the prompts and LLM chains chain1 = LLMChain(llm=llm1, prompt=PromptTemplate( input_variables=["query"], template="You are in state s1. {{query}}" )) chain2 = LLMChain(llm=llm2, prompt=PromptTemplate( input_variables=["query"], template="You are in state s2. {{query}}" )) # Create a state graph with required schemas for inputs and outputs graph = StateGraph(InputState) # Add states to the graph graph.add_node("s1",changeState) graph.add_node("s2",changeState) # Define transitions graph.add_edge(START, "s1") # Transition from s1 to s2 graph.add_edge("s1", "s2") # Transition from s2 to s1 graph.add_edge("s2", END) # Initialize the current state current_state = "s1" def handle_input(query): global current_state output = '' # Process user input based on current state if current_state == "s1": output = chain1.invoke(input=query) # Invoke chain1 with user input response = agent1(output) # Process output through Agent 1 current_state = "s2" # Transition to state s2 elif current_state == "s2": output = chain2.invoke(input=query) # Invoke chain2 with user input response = agent2(output) # Process output through Agent 2 current_state = "s1" # Transition back to state s1 return response # Create the Gradio interface with gr.Blocks() as demo: gr.Markdown("# Chatbot Interface") chatbot_interface = gr.Chatbot() user_input = gr.Textbox(label="Your Message", placeholder="Type something here...") submit_btn = gr.Button("Send") # Define the behavior of the submit button submit_btn.click( fn=lambda input_text: handle_input(input_text), # Handle user input inputs=[user_input], outputs=chatbot_interface ) # Launch the Gradio application demo.launch() ''' ''' from typing import Annotated, Sequence, TypedDict import operator import functools from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage from langchain_community.tools.tavily_search import TavilySearchResults from langchain_experimental.tools import PythonREPLTool from langchain.agents import create_openai_tools_agent from langchain_huggingface import HuggingFacePipeline from langgraph.graph import StateGraph, END from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline # SETUP: HuggingFace Model and Pipeline #name = "meta-llama/Llama-3.2-1B" #name="deepseek-ai/DeepSeek-R1-Distill-Qwen-32B" #name="deepseek-ai/deepseek-llm-7b-chat" #name="openai-community/gpt2" #name="deepseek-ai/DeepSeek-R1-Distill-Qwen-1.5B" #name="microsoft/Phi-3.5-mini-instruct" name="Qwen/Qwen2.5-7B-Instruct-1M" tokenizer = AutoTokenizer.from_pretrained(name,truncation=True) tokenizer.pad_token = tokenizer.eos_token model = AutoModelForCausalLM.from_pretrained(name) pipe = pipeline( "text-generation", model=model, tokenizer=tokenizer, device_map="auto", max_new_tokens=500, # text to generate for outputs ) print ("pipeline is created") # Wrap in LangChain's HuggingFacePipeline llm = HuggingFacePipeline(pipeline=pipe) # Members and Final Options members = ["Researcher", "Coder"] options = ["FINISH"] + members # Supervisor prompt system_prompt = ( "You are a supervisor tasked with managing a conversation between the following workers: {members}." " Given the following user request, respond with the workers to act next. Each worker will perform a task" " and respond with their results and status. When all workers are finished, respond with FINISH." ) # Prompt template required for the workflow prompt = ChatPromptTemplate.from_messages( [ ("system", system_prompt), MessagesPlaceholder(variable_name="messages"), ("system", "Given the conversation above, who should act next? Or Should we FINISH? Select one of: {options}"), ] ).partial(options=str(options), members=", ".join(members)) print ("Prompt Template created") # Supervisor routing logic def route_tool_response(llm_response): """ Parse the LLM response to determine the next step based on routing logic. """ if "FINISH" in llm_response: return "FINISH" for member in members: if member in llm_response: return member return "Unknown" def supervisor_chain(state): """ Supervisor logic to interact with HuggingFacePipeline and decide the next worker. """ messages = state.get("messages", []) print(f"[TRACE] Supervisor received messages: {messages}") # Trace input messages user_prompt = prompt.format(messages=messages) try: llm_response = pipe(user_prompt, max_new_tokens=500)[0]["generated_text"] print(f"[TRACE] LLM Response: {llm_response}") # Trace LLM interaction except Exception as e: raise RuntimeError(f"LLM processing error: {e}") next_action = route_tool_response(llm_response) print(f"[TRACE] Supervisor deciding next action: {next_action}") # Trace state changes return {"next": next_action} # AgentState definition class AgentState(TypedDict): messages: Annotated[Sequence[BaseMessage], operator.add] next: str # Create tools tavily_tool = TavilySearchResults(max_results=5) python_repl_tool = PythonREPLTool() # Create agents with their respective prompts research_agent = create_openai_tools_agent( llm=llm, tools=[tavily_tool], prompt=ChatPromptTemplate.from_messages( [ SystemMessage(content="You are a web researcher."), MessagesPlaceholder(variable_name="messages"), MessagesPlaceholder(variable_name="agent_scratchpad"), # Add required placeholder ] ), ) print ("Created agents with their respective prompts") code_agent = create_openai_tools_agent( llm=llm, tools=[python_repl_tool], prompt=ChatPromptTemplate.from_messages( [ SystemMessage(content="You may generate safe Python code for analysis."), MessagesPlaceholder(variable_name="messages"), MessagesPlaceholder(variable_name="agent_scratchpad"), # Add required placeholder ] ), ) print ("create_openai_tools_agent") # Create the workflow workflow = StateGraph(AgentState) # Nodes workflow.add_node("Researcher", research_agent) # Pass the agent directly (no .run required) workflow.add_node("Coder", code_agent) # Pass the agent directly workflow.add_node("supervisor", supervisor_chain) # Add edges for workflow transitions for member in members: workflow.add_edge(member, "supervisor") workflow.add_conditional_edges( "supervisor", lambda x: x["next"], {k: k for k in members} | {"FINISH": END} # Dynamically map workers to their actions ) print("[DEBUG] Workflow edges added: supervisor -> members/FINISH based on 'next'") # Define entry point workflow.set_entry_point("supervisor") print(workflow) # Compile the workflow graph = workflow.compile() #from IPython.display import display, Image #display(Image(graph.get_graph().draw_mermaid_png())) # Properly formatted initial state initial_state = { "messages": [ #HumanMessage(content="Code hello world and print it to the terminal.") # Correct format for user input HumanMessage(content="Write Code for printing \"hello world\" in Python. Keep it precise.") # Correct format for user input ] } # Execute the workflow try: print(f"[TRACE] Initial workflow state: {initial_state}") result = graph.invoke(initial_state) print(f"[TRACE] Workflow Result: {result}") # Final workflow result except Exception as e: print(f"[ERROR] Workflow execution failed: {e}") ''' from typing import Dict from pydantic import BaseModel from langgraph.graph import StateGraph, END from langchain_core.prompts import PromptTemplate from langchain_core.messages import HumanMessage from langchain_huggingface import HuggingFacePipeline from transformers import pipeline, AutoTokenizer, AutoModelForCausalLM import gradio as gr import torch print(torch.cuda.is_available()) # Check if a GPU is available a = torch.Tensor([1]).cuda() # Should succeed without errors print(a) # Define HuggingFace LLM def create_llm(): model_name = "Qwen/Qwen2.5-7B-Instruct-1M" tokenizer = AutoTokenizer.from_pretrained(model_name) model = AutoModelForCausalLM.from_pretrained(model_name) llm_pipeline = pipeline( task="text-generation", model=model, tokenizer=tokenizer, device=0, max_new_tokens=500, ) return HuggingFacePipeline(pipeline=llm_pipeline) # Mock Tools for Registration and Scheduling def registration_agent(visitor_details: Dict[str, str]) -> Dict[str, bool]: """Check if the visitor is registered.""" visitor_name = visitor_details.get("visitor_name") visitor_mobile = visitor_details.get("visitor_mobile") # Mock registration lookup registered_visitors = [{"visitor_name": "John Doe", "visitor_mobile": "1234567890"}] is_registered = any( v["visitor_name"] == visitor_name and v["visitor_mobile"] == visitor_mobile for v in registered_visitors ) return {"registered": is_registered} def scheduling_agent(scheduling_details: Dict[str, str]) -> Dict[str, str]: """Schedule an appointment.""" doctor_name = scheduling_details.get("doctor_name") department_name = scheduling_details.get("department_name") # Mock scheduling logic appointment_status = "Scheduled successfully" return {"status": appointment_status, "doctor_name": doctor_name, "department_name": department_name} # Define the VisitorState schema class VisitorState(BaseModel): visitor_name: str = "" visitor_mobile: str = "" doctor_name: str = "" department_name: str = "" messages: list = [] def input_state(state: VisitorState) -> dict: """Initial state to collect visitor details.""" return {"messages": ["Please provide your name and mobile number."], "next": "RegistrationState"} def registration_state(state: VisitorState) -> dict: """State to check visitor registration.""" result = registration_agent( {"visitor_name": state.visitor_name, "visitor_mobile": state.visitor_mobile} ) if result["registered"]: return {"messages": ["Visitor is registered."], "next": "SchedulingState"} else: return {"messages": ["Visitor not found in records. Please register first."], "next": END} def scheduling_state(state: VisitorState) -> dict: """State to schedule an appointment.""" result = scheduling_agent( {"doctor_name": state.doctor_name, "department_name": state.department_name} ) return { "messages": [ f"Appointment {result['status']} with Dr.{result['doctor_name']} in {result['department_name']} department." ], "next": END, } # Define Workflow workflow = StateGraph(VisitorState) workflow.add_node("InputState", input_state) workflow.add_node("RegistrationState", registration_state) workflow.add_node("SchedulingState", scheduling_state) # Define edges workflow.add_edge("InputState", "RegistrationState") workflow.add_conditional_edges( "RegistrationState", lambda state: state.get("next"), { "SchedulingState": "SchedulingState", END: END, } ) workflow.set_entry_point("InputState") compiled_graph = workflow.compile() # Gradio Frontend def gradio_interface(visitor_name, visitor_mobile, doctor_name, department_name): """Gradio interface function.""" state = VisitorState( visitor_name=visitor_name, visitor_mobile=visitor_mobile, doctor_name=doctor_name, department_name=department_name, ) # Execute workflow result = compiled_graph.invoke(state.dict()) return result["messages"][0] iface = gr.Interface( fn=gradio_interface, inputs=[ gr.Textbox(label="Visitor Name"), gr.Textbox(label="Visitor Mobile Number"), gr.Textbox(label="Doctor Name"), gr.Textbox(label="Department Name"), ], outputs="textbox", ) if __name__ == "__main__": iface.launch(share=True)