Initial commit

.gitignore

@@ -0,0 +1,5 @@
+.env
+uv.lock
+.idea
+.python-version
+__pycache__

README.md

@@ -0,0 +1 @@
+A template for chatbot streaming from langgraph with gradio

app.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python
+
+from uuid import uuid4
+import logging
+import logging.config
+import json
+
+import gradio as gr
+from dotenv import load_dotenv
+from langgraph.types import RunnableConfig
+
+from graph import GraphProcessingState, graph
+
+logger = logging.getLogger(__name__)
+
+def setup_logging():
+    with open("logging-config.json") as fh:
+        config = json.load(fh)
+    logging.config.dictConfig(config)
+
+async def chat_fn(message, history, input_graph_state, uuid):
+    try:
+        input_graph_state.user_input = message
+        input_graph_state.history = history
+        config = RunnableConfig()
+        config["configurable"] = {}
+        config["configurable"]["thread_id"] = uuid
+
+        output = ""
+        async for msg, metadata in graph.astream(
+                    {"user_input": input_graph_state.user_input, "history": input_graph_state.history},
+                    config=config,
+                    stream_mode="messages",
+                ):
+            # assistant_node is the name we defined in the langraph graph
+            if metadata['langgraph_node'] == "assistant_node" and msg.content:
+                output += msg.content
+                yield output
+    except Exception:
+        logger.exception("Exception occurred")
+        user_error_message = "There was an error processing your request. Please try again."
+        yield user_error_message  # , input_graph_state
+
+def clear():
+    return GraphProcessingState(), uuid4()
+
+if __name__ == "__main__":
+    load_dotenv()
+    setup_logging()
+    logger.info("Starting the interface")
+    with gr.Blocks(title="Langgraph Template", fill_height=True, css="footer {visibility: hidden}") as app:
+        gradio_graph_state = gr.State(
+            value=GraphProcessingState
+        )
+        uuid_state = gr.State(
+            uuid4
+        )
+        chatbot = gr.Chatbot(
+            # avatar_images=(None, "assets/ai-avatar.png"),
+            type="messages",
+            # placeholder=WELCOME_MESSAGE,/
+            scale=1,
+        )
+        chatbot.clear(fn=clear, outputs=[gradio_graph_state, uuid_state])
+        chat_interface = gr.ChatInterface(
+            chatbot=chatbot,
+            fn=chat_fn,
+            additional_inputs=[
+                gradio_graph_state,
+                uuid_state,
+            ],
+            additional_outputs=[
+                # gradio_graph_state
+            ],
+            type="messages",
+            multimodal=False,
+        )
+
+    app.launch(
+        server_name="127.0.0.1",
+        server_port=7860,
+        # favicon_path="assets/favicon.ico"
+    )

graph.py

@@ -0,0 +1,98 @@
+from typing import Annotated
+
+import aiohttp
+from langchain_core.messages import AnyMessage
+from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
+from langchain_core.tools import tool
+from langchain_openai import ChatOpenAI
+from langgraph.graph.state import CompiledStateGraph
+from langgraph.prebuilt import ToolNode
+from langgraph.types import RunnableConfig
+from langgraph.graph import Graph, StateGraph, END, MessagesState, add_messages
+from pydantic import BaseModel, Field, ValidationError
+from trafilatura import extract
+
+@tool
+async def download_website_text(url: str, config: RunnableConfig) -> str:
+    """Downloads the text from a website
+
+    args:
+        url: The URL of the website
+    """
+    async with aiohttp.ClientSession() as session:
+        async with session.get(url) as response:
+            downloaded = await response.text()
+    result = extract(downloaded, include_formatting=True, include_links=True, output_format='json', with_metadata=True)
+    if result:
+        return result
+    return "No text found on the website"
+
+tools = [download_website_text]
+
+ASSISTANT_SYSTEM_PROMPT = "You are a helpful assistant."
+assistant_model = ChatOpenAI(model="gpt-4o-mini", tags=["assistant"]).bind_tools(tools)
+
+class GraphProcessingState(BaseModel):
+    user_input: str = Field(default_factory=str, description="The original user input")
+    history: list[dict] = Field(default_factory=list, description="Chat history")  # type: ignore
+    messages: Annotated[list[AnyMessage], add_messages] = Field(default_factory=list)
+
+
+async def assistant_node(state: GraphProcessingState, config=None):
+    prompt = ChatPromptTemplate.from_messages(
+        [
+            ("system", ASSISTANT_SYSTEM_PROMPT),
+            MessagesPlaceholder(variable_name="chat_history"),
+            ("user", "{user_input}"),
+            *state.messages,
+        ]
+    )
+    chain = prompt | assistant_model
+    response = await chain.ainvoke({"user_input": state.user_input, "chat_history": state.history}, config)
+    return {"messages": response}
+
+def assistant_cond_edge(state: GraphProcessingState, config=None):
+    if not state.messages[-1].content:
+        return "tools"
+    return END
+
+def define_workflow() -> CompiledStateGraph:
+    """Defines the workflow graph"""
+    # Initialize the graph
+    workflow = StateGraph(GraphProcessingState)
+
+    # Add nodes
+    workflow.add_node("assistant_node", assistant_node)
+    workflow.add_node("tools", ToolNode(tools))
+
+    workflow.add_edge("tools", "assistant_node")
+
+    # Conditional routing
+    workflow.add_conditional_edges(
+        "assistant_node",
+        # If the latest message (result) from assistant is a tool call -> tools_condition routes to tools
+        # If the latest message (result) from assistant is a not a tool call -> tools_condition routes to END
+        assistant_cond_edge,
+    )
+    # Set end nodes
+    workflow.set_entry_point("assistant_node")
+    # workflow.set_finish_point("assistant_node")
+
+    return workflow.compile()
+
+graph = define_workflow()
+#
+# async def process_user_input_graph(input_state: GraphProcessingState, thread_id=None) -> GraphProcessingState:
+#     config: RunnableConfig = RunnableConfig()
+#     if "configurable" not in config:
+#         config["configurable"] = {}
+#     if thread_id:
+#         config["configurable"]["thread_id"] = thread_id
+#     final_state_dict = await graph.ainvoke(
+#         input_state,
+#         config=config,
+#     )
+#     final_state = GraphProcessingState(**final_state_dict)
+#     final_state.user_input = ""
+#     final_state.history = []
+#     return final_state

logging-config.json

@@ -0,0 +1,38 @@
+{
+  "version": 1,
+  "disable_existing_loggers": true,
+  "formatters": {
+    "simple": {
+      "format": "%(levelname)s: %(message)s"
+    },
+    "detailed": {
+      "format": "[%(levelname)s|%(module)s|L%(lineno)d] %(asctime)s: %(message)s",
+      "datefmt": "%Y-%m-%dT%H:%M:%S%z"
+    }
+  },
+  "handlers": {
+    "stdout": {
+      "class": "logging.StreamHandler",
+      "level": "INFO",
+      "formatter": "detailed",
+      "stream": "ext://sys.stdout"
+    },
+    "file": {
+      "class": "logging.handlers.RotatingFileHandler",
+      "level": "DEBUG",
+      "formatter": "detailed",
+      "filename": "logs/my_app.log",
+      "maxBytes": 10000,
+      "backupCount": 3
+    }
+  },
+  "loggers": {
+    "root": {
+      "level": "DEBUG",
+      "handlers": [
+        "stdout",
+        "file"
+      ]
+    }
+  }
+}

pyproject.toml

@@ -0,0 +1,16 @@
+[project]
+name = "gradio-langgraph-template"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+    "aiohttp>=3.11.12",
+    "gradio==5.16.2",
+    "langchain-core==0.3.37",
+    "langchain-openai==0.3.6",
+    "langgraph==0.2.74",
+    "pydantic==2.10.6",
+    "python-dotenv==1.0.1",
+    "trafilatura==2.0.0",
+]

setup.cfg

@@ -0,0 +1,2 @@
+[flake8]
+extend-ignore = E302,E501,E305,E402