validator for messages

graph.py

@@ -1,6 +1,7 @@
 import logging
 import os
 from typing import Annotated
+from typing_extensions import TypedDict
 
 import aiohttp
 from langchain_core.messages import AnyMessage
@@ -8,13 +9,15 @@ from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
 from langchain_core.tools import tool
 from langgraph.graph.state import CompiledStateGraph
 from langgraph.prebuilt import ToolNode, tools_condition
-from langgraph.graph import StateGraph, END, add_messages
+from langgraph.graph import StateGraph, START, END, add_messages
 from langchain_community.tools import TavilySearchResults
 from pydantic import BaseModel, Field
 from trafilatura import extract
 from langchain_anthropic import ChatAnthropic
 from langgraph.prebuilt import ToolNode
 from langgraph.checkpoint.memory import MemorySaver
+from langgraph.types import Command, interrupt
+
 
 
 
@@ -43,6 +46,11 @@ def evaluate_idea_completion(response) -> bool:
 
     return all(keyword in response_text for keyword in required_keywords)
 
+@tool
+def human_assistance(query: str) -> str:
+    """Request assistance from a human."""
+    human_response = interrupt({"query": query})
+    return human_response["data"]
 
 @tool
 async def download_website_text(url: str) -> str:
@@ -58,7 +66,7 @@ async def download_website_text(url: str) -> str:
         logger.error(f"Failed to download {url}: {str(e)}")
         return f"Error retrieving website content: {str(e)}"
 
-tools = [download_website_text]
+tools = [download_website_text, human_assistance]
 memory = MemorySaver()
 
 
@@ -92,42 +100,19 @@ class GraphProcessingState(BaseModel):
     search_enabled: bool = Field(default=True, description="Whether to enable search tools")
     idea_complete: bool = Field(default=False)
 
-
-async def planning_node(state: GraphProcessingState, config=None):
-    # Define the system prompt for planning
-    planning_prompt = "Based on the user's idea, create a detailed step-by-step plan to build the DIY product."
-
-    # Combine the planning prompt with any existing prompts
-    if state.prompt:
-        final_prompt = "\n".join([planning_prompt, state.prompt, ASSISTANT_SYSTEM_PROMPT_BASE])
-    else:
-        final_prompt = "\n".join([planning_prompt, ASSISTANT_SYSTEM_PROMPT_BASE])
-
-    # Create the prompt template
-    prompt = ChatPromptTemplate.from_messages(
-        [
-            ("system", final_prompt),
-            MessagesPlaceholder(variable_name="messages"),
+    @root_validator
+    def remove_empty_messages(cls, values):
+        messages = values.get("messages", [])
+        values["messages"] = [
+            msg for msg in messages if getattr(msg, "content", "").strip()
         ]
-    )
+        return values
 
-    # Bind tools if necessary
-    assistant_tools = []
-    if state.tools_enabled.get("download_website_text", True):
-        assistant_tools.append(download_website_text)
-    if search_enabled and state.tools_enabled.get("tavily_search_results_json", True):
-        assistant_tools.append(tavily_search_tool)
-    assistant_model = model.bind_tools(assistant_tools)
 
-    # Create the chain and invoke it
-    chain = prompt | assistant_model
-    response = await chain.ainvoke({"messages": state.messages}, config=config)
 
-    return {
-        "messages": response
-    }
 
-async def assistant_node(state: GraphProcessingState, config=None):
+
+async def chatbot(state: GraphProcessingState, config=None):
     assistant_tools = []
     if state.tools_enabled.get("download_website_text", True):
         assistant_tools.append(download_website_text)
@@ -146,8 +131,16 @@ async def assistant_node(state: GraphProcessingState, config=None):
         ]
     )
     chain = prompt | assistant_model
-    response = await chain.ainvoke({"messages": state.messages}, config=config)
 
+    valid_messages = [msg for msg in state.messages if getattr(msg, "content", "").strip()]
+
+    response = await chain.ainvoke({"messages": valid_messages}, config=config)
+
+    # message = llm_with_tools.invoke(state["messages"])
+    # Because we will be interrupting during tool execution,
+    # we disable parallel tool calling to avoid repeating any
+    # tool invocations when we resume.
+    assert len(response.tool_calls) <= 1
     idea_complete = evaluate_idea_completion(response)
 
     return {
@@ -155,23 +148,6 @@ async def assistant_node(state: GraphProcessingState, config=None):
         "idea_complete": idea_complete
     }
 
-# def assistant_cond_edge(state: GraphProcessingState):
-#     last_message = state.messages[-1]
-#     if hasattr(last_message, "tool_calls") and last_message.tool_calls:
-#         logger.info(f"Tool call detected: {last_message.tool_calls}")
-#         return "tools"
-#     return END
-def assistant_routing(state: GraphProcessingState) -> str:
-    last_message = state.messages[-1]
-    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
-        logger.info("Tool call detected. Routing to 'tools' node.")
-        return "tools"
-    elif state.idea_complete:
-        logger.info("Idea is complete. Routing to 'planning_node'.")
-        return "planning_node"
-    else:
-        logger.info("Idea is incomplete. Routing back to 'assistant_node'.")
-        return "assistant_node"
 
 
 
@@ -181,20 +157,19 @@ def define_workflow() -> CompiledStateGraph:
     workflow = StateGraph(GraphProcessingState)
 
     # Add nodes
-    workflow.add_node("assistant_node", assistant_node)
     workflow.add_node("tools", ToolNode(tools))
-    workflow.add_node("planning_node", planning_node)
+    workflow.add_node("chatbot", chatbot)
 
-    # Edges
-    workflow.add_edge("tools", "assistant_node")
-    workflow.add_edge("planning_node", "assistant_node")
+  
+
+    workflow.add_conditional_edges(
+        "chatbot",
+        tools_condition,
+    )
 
-    # Conditional routing   
-    workflow.add_conditional_edges("assistant_node", assistant_routing)
+    workflow.add_edge("tools", "chatbot")
+    workflow.add_edge(START, "chatbot")
 
-    # Set end nodes
-    workflow.set_entry_point("assistant_node")
-    # workflow.set_finish_point("assistant_node")
     compiled_graph = workflow.compile(checkpointer=memory)
     try:
         img_bytes = compiled_graph.get_graph().draw_mermaid_png()
@@ -208,4 +183,7 @@ def define_workflow() -> CompiledStateGraph:
 
     return compiled_graph
 
-graph = define_workflow()
+graph = define_workflow()
+
+
+

oldgraph.py

@@ -0,0 +1,264 @@
+import logging
+import os
+from typing import Annotated
+from typing_extensions import TypedDict
+
+import aiohttp
+from langchain_core.messages import AnyMessage
+from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
+from langchain_core.tools import tool
+from langgraph.graph.state import CompiledStateGraph
+from langgraph.prebuilt import ToolNode, tools_condition
+from langgraph.graph import StateGraph, START, END, add_messages
+from langchain_community.tools import TavilySearchResults
+from pydantic import BaseModel, Field
+from trafilatura import extract
+from langchain_anthropic import ChatAnthropic
+from langgraph.prebuilt import ToolNode
+from langgraph.checkpoint.memory import MemorySaver
+from langgraph.types import Command, interrupt
+
+class State(TypedDict):
+    messages: Annotated[list, add_messages]
+
+
+logger = logging.getLogger(__name__)
+ASSISTANT_SYSTEM_PROMPT_BASE = """"""
+search_enabled = bool(os.environ.get("TAVILY_API_KEY"))
+
+def evaluate_idea_completion(response) -> bool:
+    """
+    Evaluates whether the assistant's response indicates a complete DIY project idea.
+    You can customize the logic based on your specific criteria.
+    """
+    # Example logic: Check if the response contains certain keywords
+    required_keywords = ["materials", "dimensions", "tools", "steps"]
+
+    # Determine the type of response and extract text accordingly
+    if isinstance(response, dict):
+        # If response is a dictionary, extract values and join them into a single string
+        response_text = ' '.join(str(value).lower() for value in response.values())
+    elif isinstance(response, str):
+        # If response is a string, convert it to lowercase
+        response_text = response.lower()
+    else:
+        # If response is of an unexpected type, convert it to string and lowercase
+        response_text = str(response).lower()
+
+    return all(keyword in response_text for keyword in required_keywords)
+
+@tool
+def human_assistance(query: str) -> str:
+    """Request assistance from a human."""
+    human_response = interrupt({"query": query})
+    return human_response["data"]
+
+@tool
+async def download_website_text(url: str) -> str:
+    """Download the text from a website"""
+    try:
+        async with aiohttp.ClientSession() as session:
+            async with session.get(url) as response:
+                response.raise_for_status()
+                downloaded = await response.text()
+        result = extract(downloaded, include_formatting=True, include_links=True, output_format='json', with_metadata=True)
+        return result or "No text found on the website"
+    except Exception as e:
+        logger.error(f"Failed to download {url}: {str(e)}")
+        return f"Error retrieving website content: {str(e)}"
+
+tools = [download_website_text, human_assistance]
+memory = MemorySaver()
+
+
+if search_enabled:
+    tavily_search_tool = TavilySearchResults(
+        max_results=5,
+        search_depth="advanced",
+        include_answer=True,
+        include_raw_content=True,
+    )
+    tools.append(tavily_search_tool)
+else:
+    print("TAVILY_API_KEY environment variable not found. Websearch disabled")
+
+weak_model = ChatAnthropic(
+            model="claude-3-5-sonnet-20240620",
+            temperature=0,
+            max_tokens=1024,
+            timeout=None,
+            max_retries=2,
+            # other params...
+        )
+model = weak_model
+assistant_model = weak_model
+
+class GraphProcessingState(BaseModel):
+    # user_input: str = Field(default_factory=str, description="The original user input")
+    messages: Annotated[list[AnyMessage], add_messages] = Field(default_factory=list)
+    prompt: str = Field(default_factory=str, description="The prompt to be used for the model")
+    tools_enabled: dict = Field(default_factory=dict, description="The tools enabled for the assistant")
+    search_enabled: bool = Field(default=True, description="Whether to enable search tools")
+    idea_complete: bool = Field(default=False)
+
+
+async def planning_node(state: GraphProcessingState, config=None):
+    # Define the system prompt for planning
+    planning_prompt = "Based on the user's idea, create a detailed step-by-step plan to build the DIY product."
+
+    # Combine the planning prompt with any existing prompts
+    if state.prompt:
+        final_prompt = "\n".join([planning_prompt, state.prompt, ASSISTANT_SYSTEM_PROMPT_BASE])
+    else:
+        final_prompt = "\n".join([planning_prompt, ASSISTANT_SYSTEM_PROMPT_BASE])
+
+    # Create the prompt template
+    prompt = ChatPromptTemplate.from_messages(
+        [
+            ("system", final_prompt),
+            MessagesPlaceholder(variable_name="messages"),
+        ]
+    )
+
+    # Bind tools if necessary
+    assistant_tools = []
+    if state.tools_enabled.get("download_website_text", True):
+        assistant_tools.append(download_website_text)
+    if search_enabled and state.tools_enabled.get("tavily_search_results_json", True):
+        assistant_tools.append(tavily_search_tool)
+    assistant_model = model.bind_tools(assistant_tools)
+
+    # Create the chain and invoke it
+    chain = prompt | assistant_model
+    response = await chain.ainvoke({"messages": state.messages}, config=config)
+
+    return {
+        "messages": response
+    }
+
+async def assistant_node(state: GraphProcessingState, config=None):
+    assistant_tools = []
+    if state.tools_enabled.get("download_website_text", True):
+        assistant_tools.append(download_website_text)
+    if search_enabled and state.tools_enabled.get("tavily_search_results_json", True):
+        assistant_tools.append(tavily_search_tool)
+    assistant_model = model.bind_tools(assistant_tools)
+    if state.prompt:
+        final_prompt = "\n".join([state.prompt, ASSISTANT_SYSTEM_PROMPT_BASE])
+    else:
+        final_prompt = ASSISTANT_SYSTEM_PROMPT_BASE
+
+    prompt = ChatPromptTemplate.from_messages(
+        [
+            ("system", final_prompt),
+            MessagesPlaceholder(variable_name="messages"),
+        ]
+    )
+    chain = prompt | assistant_model
+    response = await chain.ainvoke({"messages": state.messages}, config=config)
+
+    idea_complete = evaluate_idea_completion(response)
+
+    return {
+        "messages": response,
+        "idea_complete": idea_complete
+    }
+
+async def chatbot(state: GraphProcessingState, config=None):
+    assistant_tools = []
+    if state.tools_enabled.get("download_website_text", True):
+        assistant_tools.append(download_website_text)
+    if search_enabled and state.tools_enabled.get("tavily_search_results_json", True):
+        assistant_tools.append(tavily_search_tool)
+    assistant_model = model.bind_tools(assistant_tools)
+    if state.prompt:
+        final_prompt = "\n".join([state.prompt, ASSISTANT_SYSTEM_PROMPT_BASE])
+    else:
+        final_prompt = ASSISTANT_SYSTEM_PROMPT_BASE
+
+    prompt = ChatPromptTemplate.from_messages(
+        [
+            ("system", final_prompt),
+            MessagesPlaceholder(variable_name="messages"),
+        ]
+    )
+    chain = prompt | assistant_model
+
+    response = await chain.ainvoke({"messages": state.messages}, config=config)
+
+    # message = llm_with_tools.invoke(state["messages"])
+    # Because we will be interrupting during tool execution,
+    # we disable parallel tool calling to avoid repeating any
+    # tool invocations when we resume.
+    assert len(response.tool_calls) <= 1
+    idea_complete = evaluate_idea_completion(response)
+
+    return {
+        "messages": response,
+        "idea_complete": idea_complete
+    }
+
+# def assistant_cond_edge(state: GraphProcessingState):
+#     last_message = state.messages[-1]
+#     if hasattr(last_message, "tool_calls") and last_message.tool_calls:
+#         logger.info(f"Tool call detected: {last_message.tool_calls}")
+#         return "tools"
+#     return END
+def assistant_routing(state: GraphProcessingState) -> str:
+    last_message = state.messages[-1]
+    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
+        logger.info("Tool call detected. Routing to 'tools' node.")
+        return "tools"
+    elif state.idea_complete:
+        logger.info("Idea is complete. Routing to 'planning_node'.")
+        return "planning_node"
+    else:
+        logger.info("Idea is incomplete. Routing back to 'assistant_node'.")
+        return "assistant_node"
+
+
+
+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_node("planning_node", planning_node)
+    workflow.add_node("chatbot", chatbot)
+
+    # Edges
+    # workflow.add_edge("tools", "assistant_node")
+    # workflow.add_edge("planning_node", "assistant_node")
+
+
+    workflow.add_conditional_edges(
+        "chatbot",
+        tools_condition,
+    )
+
+    workflow.add_edge("tools", "chatbot")
+    workflow.add_edge(START, "chatbot")
+
+    # Conditional routing   
+    # workflow.add_conditional_edges("assistant_node", assistant_routing)
+
+    # # Set end nodes
+    # workflow.set_entry_point("assistant_node")
+    # workflow.set_finish_point("assistant_node")
+    compiled_graph = workflow.compile(checkpointer=memory)
+    try:
+        img_bytes = compiled_graph.get_graph().draw_mermaid_png()
+        with open("graph.png", "wb") as f:
+            f.write(img_bytes)
+        print("Graph image saved as graph.png")
+    except Exception as e:
+        print("Can't print the graph:")
+        print(e)
+
+
+    return compiled_graph
+
+graph = define_workflow()