feat: delete support for older LLM models

- removed support for FuctionMessage

Author: [email protected]

src/agent/graph.py

@@ -1,16 +1,14 @@
-from copy import copy
 from langgraph.graph import StateGraph, MessagesState, START, END
-from langchain_core.messages import FunctionMessage, SystemMessage, ToolMessage
+from langchain_core.messages import SystemMessage, ToolMessage, BaseMessage, AnyMessage
 from langchain_core.runnables import RunnableConfig
-# from langchain_ollama import ChatOllama
 from langchain_google_genai import ChatGoogleGenerativeAI
 
 from agent.tools import tools
 from agent.utils import (
     has_too_many_consecutive_tool_calls,
     message_has_tool_calls,
-    process_new_format_tool_calls,
-    process_old_format_function_call
+    process_tool_calls,
+    filter_empty_content_messages
 )
 
 
@@ -23,11 +21,6 @@ class Configuration:
 
 )
 
-# llm_name = "llama3.2:3b-instruct-fp16"
-# llm = ChatOllama(
-#     model=llm_name,
-#     temperature=0.1,
-# )
 llm_name = "gemini-2.5-flash"
 llm = ChatGoogleGenerativeAI(model=llm_name)
 llm_with_tools = llm.bind_tools(tools)
@@ -45,26 +38,8 @@ def agent_node(state: AgentState, config: RunnableConfig):
     if not messages or not isinstance(messages[0], SystemMessage):
         messages = [SYSTEM_MESSAGE] + messages
 
-    # Ensure no message has completely empty content (which causes Gemini errors)
-    processed_messages = []
-    for message in messages:
-        content = getattr(message, "content", "")
-
-        # If content is empty or just whitespace, provide minimal content
-        if not content or (isinstance(content, str) and not content.strip()):
-            # Create a copy with minimal non-empty content
-            new_message = copy(message)
-            if isinstance(message, (ToolMessage, FunctionMessage)):
-                new_message.content = "Completed"
-            elif isinstance(message, SystemMessage):
-                # Keep system message as is (it should have content)
-                processed_messages.append(message)
-                continue
-            else:
-                new_message.content = "..."
-            processed_messages.append(new_message)
-        else:
-            processed_messages.append(message)
+    # Filter messages to ensure no empty content (prevents Gemini errors)
+    processed_messages = filter_empty_content_messages(messages)
 
     response = llm_with_tools.invoke(processed_messages)
     return {"messages": [response]}
@@ -73,7 +48,7 @@ def agent_node(state: AgentState, config: RunnableConfig):
 def should_continue(state: AgentState) -> str:
     """Determine if we should continue or end."""
     messages = state["messages"]
-    last_message = messages[-1]
+    last_message: AnyMessage = messages[-1]
 
     # Prevent infinite loops by checking recent tool usage
     if has_too_many_consecutive_tool_calls(messages):
@@ -87,24 +62,20 @@ def should_continue(state: AgentState) -> str:
 
 
 def tools_node(state: AgentState, config: RunnableConfig):
-    """Handle tool execution for both new and old tool calling formats."""
-    last_message = state["messages"][-1]
+    """Handle tool execution using the modern tool calling format."""
+    last_message: AnyMessage = state["messages"][-1]
 
-    # Handle new tool_calls format (preferred)
+    # Handle tool_calls format
     tool_calls = getattr(last_message, "tool_calls", None)
     if tool_calls:
-        messages_to_add = process_new_format_tool_calls(tool_calls, tools)
+        messages_to_add = process_tool_calls(tool_calls, tools)
         return {"messages": messages_to_add}
 
-    # Handle old function_call format (fallback)
-    additional_kwargs = getattr(last_message, "additional_kwargs", {})
-    function_call = additional_kwargs.get("function_call")
-    if function_call:
-        message = process_old_format_function_call(function_call, tools)
-        return {"messages": [message]}
+    # log a warning for observation
+    print("Warning: No tool calls found in the last message. This may indicate an issue.")
 
-    # No tool calls found
-    return {"messages": [FunctionMessage(content="No tool calls found", name="error")]}
+    # No tool calls found - this shouldn't happen with modern LLMs
+    return {"messages": [ToolMessage(content="No tool calls found", tool_call_id="error")]}
 
 
 workflow = StateGraph(AgentState)

src/agent/utils.py

@@ -1,11 +1,15 @@
 """Utility functions for the agent graph."""
 
-import json
-from typing import List, Dict, Any, Optional
-from langchain_core.messages import ToolMessage, FunctionMessage
+from copy import copy
+from typing import List, Dict, Any, Sequence
+from langchain_core.messages import BaseMessage, ToolMessage, AnyMessage, SystemMessage
 
 
-def has_too_many_consecutive_tool_calls(messages: list, max_calls: int = 3, look_back: int = 10) -> bool:
+def has_too_many_consecutive_tool_calls(
+    messages: Sequence[AnyMessage],
+    max_calls: int = 3,
+    look_back: int = 10
+) -> bool:
     """Check if there are too many consecutive tool calls to prevent infinite loops."""
     recent_messages = messages[-look_back:] if len(
         messages) > look_back else messages
@@ -22,23 +26,16 @@ def has_too_many_consecutive_tool_calls(messages: list, max_calls: int = 3, look
     return consecutive_tool_calls >= max_calls
 
 
-def message_has_tool_calls(message) -> bool:
-    """Check if a message contains tool calls in either format."""
-    # Check new tool_calls format
-    if hasattr(message, "tool_calls") and getattr(message, "tool_calls", None):
-        return True
+def message_has_tool_calls(message: AnyMessage) -> bool:
+    """Check if a message contains tool calls."""
+    # Check for tool_calls format (modern standard)
+    tool_calls = getattr(message, "tool_calls", None)
+    return bool(tool_calls)
 
-    # Check old function_call format
-    if hasattr(message, "additional_kwargs"):
-        additional_kwargs = getattr(message, "additional_kwargs", {})
-        return "function_call" in additional_kwargs
 
-    return False
-
-
-def is_tool_response(message) -> bool:
-    """Check if a message is a tool or function response."""
-    return hasattr(message, "__class__") and message.__class__.__name__ in ["ToolMessage", "FunctionMessage"]
+def is_tool_response(message: AnyMessage) -> bool:
+    """Check if a message is a tool response."""
+    return isinstance(message, ToolMessage)
 
 
 def execute_tool(tool, tool_name: str, tool_input: Dict[str, Any]) -> str:
@@ -61,8 +58,8 @@ def find_tool_by_name(tools: List, tool_name: str):
     return next((t for t in tools if t.name == tool_name), None)
 
 
-def process_new_format_tool_calls(tool_calls: List[Dict], tools: List) -> List[ToolMessage]:
-    """Process tool calls in the new format and return ToolMessage list."""
+def process_tool_calls(tool_calls: List[Dict], tools: List) -> List[ToolMessage]:
+    """Process tool calls and return ToolMessage list."""
     messages_to_add = []
 
     for tool_call in tool_calls:
@@ -82,15 +79,26 @@ def process_new_format_tool_calls(tool_calls: List[Dict], tools: List) -> List[T
     return messages_to_add
 
 
-def process_old_format_function_call(function_call: Dict, tools: List) -> FunctionMessage:
-    """Process function call in the old format and return FunctionMessage."""
-    tool_name = function_call["name"]
-    tool_input = json.loads(function_call["arguments"])
-
-    tool = find_tool_by_name(tools, tool_name)
-    if tool:
-        result_content = execute_tool(tool, tool_name, tool_input)
-    else:
-        result_content = f"Tool '{tool_name}' not found"
+def filter_empty_content_messages(messages: List[AnyMessage]) -> List[AnyMessage]:
+    """Filter messages to ensure no message has empty content (prevents Gemini errors)."""
+    processed_messages = []
+    for message in messages:
+        content = getattr(message, "content", "")
+
+        # If content is empty or just whitespace, provide minimal content
+        if not content or (isinstance(content, str) and not content.strip()):
+            # Create a copy with minimal non-empty content
+            new_message = copy(message)
+            if isinstance(message, ToolMessage):
+                new_message.content = "Completed"
+            elif isinstance(message, SystemMessage):
+                # Keep system message as is (it should have content)
+                processed_messages.append(message)
+                continue
+            else:
+                new_message.content = "..."
+            processed_messages.append(new_message)
+        else:
+            processed_messages.append(message)
 
-    return FunctionMessage(content=result_content, name=tool_name)
+    return processed_messages