app.py

'''

import gradio as gr
import os
if os.environ.get("SPACES_ZERO_GPU") is not None:
    import spaces
else:
    class spaces:
        @staticmethod
        def GPU(func):
            def wrapper(*args, **kwargs):
                return func(*args, **kwargs)
            return wrapper

@spaces.GPU
def fake_gpu():
  pass

# Define a function to respond to user input
def respond(message, history):
    # Create a response based on the user's message
    response = "You said: " + message
    
    # Append the message and response to history
    history.append({"role": "user", "content": message})
    history.append({"role": "assistant", "content": response})
    
    return history

# Create the Gradio interface
with gr.Blocks() as demo:
    gr.Markdown("# Chatbot Interface")
    
    # Initialize chatbot with the new message type
    chatbot_interface = gr.Chatbot(type='messages')  # Specify type='messages'
    user_input = gr.Textbox(label="Your Message", placeholder="Type something...")
    submit_btn = gr.Button("Send")
    
    # Define the behavior of the submit button
    submit_btn.click(fn=respond, inputs=[user_input, chatbot_interface], outputs=chatbot_interface)

# Launch the Gradio application
demo.launch()





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 transformers import pipeline, AutoTokenizer, AutoModelForCausalLM
from langchain_huggingface import HuggingFacePipeline
from langchain.tools import Tool
from langchain.agents import create_react_agent
from langgraph.graph import StateGraph, END
from pydantic import BaseModel
import gradio as gr
import os
if os.environ.get("SPACES_ZERO_GPU") is not None:
    import spaces
else:
    class spaces:
        @staticmethod
        def GPU(func):
            def wrapper(*args, **kwargs):
                return func(*args, **kwargs)
            return wrapper

@spaces.GPU
def fake_gpu():
  pass

# ---------------------------------------
# Step 1: Define Hugging Face LLM (Qwen/Qwen2.5-7B-Instruct-1M)
# ---------------------------------------
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=-1,  # CPU mode, set to 0 for GPU
        max_new_tokens=200
    )
    return HuggingFacePipeline(pipeline=llm_pipeline)

# ---------------------------------------
# Step 2: Create Agents
# ---------------------------------------
llm = create_llm()

# Registration Agent
registration_agent = Tool(
    name="registration_check",
    description="Check if a patient is registered.",
    func=lambda details: registration_tool(details.get("visitor_name"), details.get("visitor_mobile"))
)

# Scheduling Agent
scheduling_agent = Tool(
    name="schedule_appointment",
    description="Fetch available time slots for a doctor.",
    func=lambda details: doctor_slots_tool(details.get("doctor_name"))
)

# Payment Agent
payment_agent = Tool(
    name="process_payment",
    description="Generate a payment link and confirm the payment.",
    func=lambda details: confirm_payment_tool(details.get("transaction_id"))
)

# Email Agent
email_agent = Tool(
    name="send_email",
    description="Send appointment confirmation email to the visitor.",
    func=lambda details: email_tool(
        details.get("visitor_email"),
        details.get("appointment_details"),
        details.get("hospital_location")
    )
)

# ---------------------------------------
# Step 3: Tools and Mock Functions
# ---------------------------------------
def registration_tool(visitor_name: str, visitor_mobile: str) -> bool:
    registered_visitors = [{"visitor_name": "John Doe", "visitor_mobile": "1234567890"}]
    return any(
        v["visitor_name"] == visitor_name and v["visitor_mobile"] == visitor_mobile
        for v in registered_visitors
    )

def register_visitor(visitor_name: str, visitor_mobile: str) -> bool:
    """Register a new user if not already registered."""
    return True  # Simulate successful registration

def doctor_slots_tool(doctor_name: str):
    available_slots = {
        "Dr. Smith": ["10:00 AM", "2:00 PM"],
        "Dr. Brown": ["12:00 PM"]
    }
    return available_slots.get(doctor_name, [])

def payment_tool(amount: float):
    """Generate a payment link."""
    return f"http://mock-payment-link.com/pay?amount={amount}"

def confirm_payment_tool(transaction_id: str) -> dict:
    """Confirm the payment."""
    if transaction_id == "TIMEOUT":
        return {"status": "FAILED", "reason_code": "timeout"}
    elif transaction_id == "SUCCESS":
        return {"status": "SUCCESS", "reason_code": None}
    else:
        return {"status": "FAILED", "reason_code": "other_error"}

def email_tool(visitor_email: str, appointment_details: str, hospital_location: str) -> bool:
    """Simulate sending an email to the visitor with appointment details."""
    print(f"Sending email to {visitor_email}...")
    print(f"Appointment Details: {appointment_details}")
    print(f"Hospital Location: {hospital_location}")
    # Simulate success
    return True

# ---------------------------------------
# Step 4: Define Workflow States
# ---------------------------------------
class VisitorState(BaseModel):
    visitor_name: str = ""
    visitor_mobile: str = ""
    visitor_email: str = ""
    doctor_name: str = ""
    department_name: str = ""
    selected_slot: str = ""
    messages: list = []
    payment_confirmed: bool = False
    email_sent: bool = False

def input_state(state: VisitorState):
    """InputState: Collect visitor details."""
    return {"messages": ["Please provide your name, mobile number, and email."], "next": "RegistrationState"}

def registration_state(state: VisitorState):
    """Registration State: Check and register visitor."""
    is_registered = registration_tool(state.visitor_name, state.visitor_mobile)
    if is_registered:
        return {"messages": ["Visitor is registered."], "next": "SchedulingState"}
    else:
        successfully_registered = register_visitor(state.visitor_name, state.visitor_mobile)
        if successfully_registered:
            return {"messages": ["Visitor has been successfully registered."], "next": "SchedulingState"}
        else:
            return {"messages": ["Registration failed. Please try again later."], "next": END}

def scheduling_state(state: VisitorState):
    """SchedulingState: Fetch available slots for a doctor."""
    available_slots = doctor_slots_tool(state.doctor_name)
    if available_slots:
        state.selected_slot = available_slots[0]
        return {"messages": [f"Slot selected for {state.doctor_name}: {state.selected_slot}"], "next": "PaymentState"}
    else:
        return {"messages": [f"No available slots for {state.doctor_name}."], "next": END}

def payment_state(state: VisitorState):
    """PaymentState: Generate payment link and confirm."""
    payment_link = payment_tool(500)
    state.messages.append(f"Please proceed to pay at: {payment_link}")

    # Simulate payment confirmation
    payment_response = confirm_payment_tool("SUCCESS")
    if payment_response["status"] == "SUCCESS":
        state.payment_confirmed = True
        return {"messages": ["Payment successful. Appointment is being finalized."], "next": "FinalState"}
    elif payment_response["reason_code"] == "timeout":
        return {"messages": ["Payment timed out. Retrying payment..."], "next": "PaymentState"}
    else:
        return {"messages": ["Payment failed due to an error. Please try again later."], "next": END}

def final_state(state: VisitorState):
    """FinalState: Send email confirmation and finalize the appointment."""
    if state.payment_confirmed:
        appointment_details = f"Doctor: {state.doctor_name}\nTime: {state.selected_slot}"
        hospital_location = "123 Main St, Springfield, USA"
        email_success = email_tool(state.visitor_email, appointment_details, hospital_location)

        if email_success:
            state.email_sent = True
            return {"messages": [f"Appointment confirmed. Details sent to your email: {state.visitor_email}"], "next": END}
        else:
            return {"messages": ["Appointment confirmed, but failed to send email. Please contact support."], "next": END}
    else:
        return {"messages": ["Payment confirmation failed. Appointment could not be finalized."], "next": END}

# ---------------------------------------
# Step 5: Build Langgraph Workflow
# ---------------------------------------
workflow = StateGraph(VisitorState)

# Add nodes
workflow.add_node("InputState", input_state)
workflow.add_node("RegistrationState", registration_state)
workflow.add_node("SchedulingState", scheduling_state)
workflow.add_node("PaymentState", payment_state)
workflow.add_node("FinalState", final_state)

# Define edges
workflow.add_edge("InputState", "RegistrationState")
workflow.add_edge("RegistrationState", "SchedulingState")
workflow.add_edge("SchedulingState", "PaymentState")
workflow.add_edge("PaymentState", "FinalState")

# Entry Point
workflow.set_entry_point("InputState")
compiled_graph = workflow.compile()

# ---------------------------------------
# Step 6: Gradio Interface
# ---------------------------------------
def gradio_interface(visitor_name, visitor_mobile, visitor_email, doctor_name, department_name):
    """Interface for Gradio application."""
    state = VisitorState(
        visitor_name=visitor_name,
        visitor_mobile=visitor_mobile,
        visitor_email=visitor_email,
        doctor_name=doctor_name,
        department_name=department_name,
    )
    # Execute workflow
    result = compiled_graph.invoke(state.model_dump())
    return "\n".join(result["messages"])

iface = gr.Interface(
    fn=gradio_interface,
    inputs=[
        gr.Textbox(label="Visitor Name"),
        gr.Textbox(label="Visitor Mobile Number"),
        gr.Textbox(label="Visitor Email"),
        gr.Textbox(label="Doctor Name"),
        gr.Textbox(label="Department Name"),
    ],
    outputs="textbox",
)

# Execute the Gradio interface
if __name__ == "__main__":
    iface.launch()