Upload 2 files

agent.py

@@ -1,9 +1,9 @@
 import re
-from typing import Tuple
+from typing import Dict, List
 
 from memory import get_relevant_context, save_interaction
 from model import get_model_manager
-from tools import calculator_tool, datetime_tool, web_search_tool
+from tools import calculator_tool, datetime_tool, text_stats_tool, web_search_tool
 
 
 class AgentRouter:
@@ -11,7 +11,7 @@ class AgentRouter:
         self.model = get_model_manager()
 
     @staticmethod
-    def _classify(message: str) -> str:
+    def _detect_tool_intents(message: str) -> List[str]:
         lower = message.lower()
 
         web_keywords = [
@@ -50,6 +50,14 @@ class AgentRouter:
             "plus",
             "minus",
         ]
+        text_stats_keywords = [
+            "word count",
+            "count words",
+            "character count",
+            "text stats",
+            "text statistics",
+            "count characters",
+        ]
 
         tokens = set(re.findall(r"\b\w+\b", lower))
 
@@ -58,24 +66,25 @@ class AgentRouter:
                 return keyword in lower
             return keyword in tokens
 
-        has_web_intent = any(has_phrase_or_token(k) for k in web_keywords)
-        has_datetime_intent = any(has_phrase_or_token(k) for k in datetime_keywords)
+        intents: List[str] = []
 
-        # Give web queries priority when both are present (e.g., "latest AI news today").
-        if has_web_intent:
-            return "web"
+        if any(has_phrase_or_token(k) for k in web_keywords):
+            intents.append("web_search")
 
-        if has_datetime_intent:
-            return "datetime"
+        if any(has_phrase_or_token(k) for k in datetime_keywords):
+            intents.append("datetime")
 
         if any(has_phrase_or_token(k) for k in calc_keywords):
-            return "calculator"
+            intents.append("calculator")
 
         # Fallback detection for math-like expressions.
-        if re.search(r"[0-9][0-9\s\+\-\*/\(\)\.\^%]+", lower):
-            return "calculator"
+        if "calculator" not in intents and re.search(r"[0-9][0-9\s\+\-\*/\(\)\.\^%]+", lower):
+            intents.append("calculator")
+
+        if any(has_phrase_or_token(k) for k in text_stats_keywords):
+            intents.append("text_stats")
 
-        return "llm"
+        return intents if intents else ["llm"]
 
     @staticmethod
     def _extract_expression(message: str) -> str:
@@ -88,36 +97,58 @@ class AgentRouter:
             if cleaned.lower().startswith(prefix):
                 cleaned = cleaned[len(prefix):].strip()
 
-        return cleaned.strip() or message.strip()
+        matches = re.findall(r"[0-9\s\+\-\*/\(\)\.\^%]+", cleaned)
+        ranked = [m.strip() for m in matches if re.search(r"\d", m) and re.search(r"[\+\-\*/\^%]", m)]
+        if ranked:
+            return max(ranked, key=len).replace("^", "**")
 
-    def _tool_context(self, route: str, message: str) -> Tuple[str, str]:
-        if route == "datetime":
-            tool_result = datetime_tool()
-            return "datetime", tool_result
+        return cleaned.strip() or message.strip()
 
-        if route == "web":
-            tool_result = web_search_tool(message, max_results=5)
-            return "web_search", tool_result
+    def _run_tools(self, intents: List[str], message: str) -> Dict[str, str]:
+        outputs: Dict[str, str] = {}
 
-        if route == "calculator":
-            expression = self._extract_expression(message)
-            tool_result = calculator_tool(expression)
-            return "calculator", f"Expression: {expression}\nResult: {tool_result}"
+        for intent in intents:
+            if intent == "datetime":
+                outputs["datetime"] = datetime_tool()
+            elif intent == "web_search":
+                outputs["web_search"] = web_search_tool(message, max_results=5)
+            elif intent == "calculator":
+                expression = self._extract_expression(message)
+                result = calculator_tool(expression)
+                outputs["calculator"] = f"Expression: {expression}\nResult: {result}"
+            elif intent == "text_stats":
+                outputs["text_stats"] = text_stats_tool(message)
 
-        return "llm", ""
+        return outputs
 
     @staticmethod
-    def _direct_tool_response(tool_name: str, tool_output: str) -> str:
-        if tool_name == "datetime":
-            return tool_output
-
-        if tool_name == "calculator":
-            lines = tool_output.splitlines()
-            result_line = next((line for line in lines if line.startswith("Result:")), "Result: N/A")
+    def _friendly_direct_response(tool_outputs: Dict[str, str]) -> str:
+        lines: List[str] = ["Sure, here you go:"]
+
+        if "datetime" in tool_outputs:
+            date_line = ""
+            time_line = ""
+            for line in tool_outputs["datetime"].splitlines():
+                if line.startswith("Current date:"):
+                    date_line = line.replace("Current date:", "").strip()
+                if line.startswith("Current time:"):
+                    time_line = line.replace("Current time:", "").strip()
+            if date_line or time_line:
+                lines.append(f"- Date and time: {date_line} {time_line}".strip())
+
+        if "calculator" in tool_outputs:
+            result_line = next(
+                (line for line in tool_outputs["calculator"].splitlines() if line.startswith("Result:")),
+                "Result: N/A",
+            )
             result = result_line.replace("Result:", "").strip()
-            return f"The result is {result}."
+            lines.append(f"- Calculation result: {result}")
+
+        if "text_stats" in tool_outputs:
+            stats = tool_outputs["text_stats"].replace("\n", " | ")
+            lines.append(f"- Text stats: {stats}")
 
-        return tool_output
+        return "\n".join(lines)
 
     @staticmethod
     def _is_unhelpful_web_response(text: str) -> bool:
@@ -158,27 +189,52 @@ class AgentRouter:
 
         return "Here are the latest web results:\n" + "\n".join(bullets)
 
+    @staticmethod
+    def _extra_tools_summary(tool_outputs: Dict[str, str]) -> str:
+        extra: List[str] = []
+        if "datetime" in tool_outputs:
+            extra.append(tool_outputs["datetime"])
+        if "calculator" in tool_outputs:
+            extra.append(tool_outputs["calculator"])
+        if "text_stats" in tool_outputs:
+            extra.append(tool_outputs["text_stats"])
+
+        if not extra:
+            return ""
+
+        return "\n\nAdditional tool outputs:\n" + "\n\n".join(extra)
+
     def respond(self, user_id: str, message: str) -> str:
         memory_context = get_relevant_context(user_id, message)
-        route = self._classify(message)
-        tool_name, tool_output = self._tool_context(route, message)
+        intents = self._detect_tool_intents(message)
 
-        # For deterministic utility queries, respond directly from tools.
-        if tool_name in {"datetime", "calculator"}:
-            response = self._direct_tool_response(tool_name, tool_output)
+        if intents == ["llm"]:
+            response = self.model.generate(
+                message=message,
+                memory_context=memory_context,
+                tool_context="",
+            )
             save_interaction(user_id, message, response)
             return response
 
-        if tool_name == "llm":
-            tool_context = ""
-        else:
-            tool_context = f"Tool used: {tool_name}\n{tool_output}"
+        tool_outputs = self._run_tools(intents, message)
 
-        if tool_name == "web_search":
+        deterministic_only = set(tool_outputs.keys()).issubset({"datetime", "calculator", "text_stats"})
+        if deterministic_only:
+            response = self._friendly_direct_response(tool_outputs)
+            save_interaction(user_id, message, response)
+            return response
+
+        tool_context_parts = []
+        for tool_name, tool_output in tool_outputs.items():
+            tool_context_parts.append(f"Tool used: {tool_name}\n{tool_output}")
+        tool_context = "\n\n".join(tool_context_parts)
+
+        if "web_search" in tool_outputs:
             web_instruction = (
                 "Answer using only the provided web results. "
                 "Do not say you lack real-time access. "
-                "Provide a concise summary with sources."
+                "Provide a concise, friendly summary with sources."
             )
             response = self.model.generate(
                 message=f"{web_instruction}\n\nUser request: {message}",
@@ -187,7 +243,11 @@ class AgentRouter:
             )
 
             if self._is_unhelpful_web_response(response):
-                response = self._summarize_web_tool_output(tool_output, message)
+                response = self._summarize_web_tool_output(tool_outputs["web_search"], message)
+
+            extra = self._extra_tools_summary(tool_outputs)
+            if extra:
+                response = f"{response}{extra}".strip()
 
             save_interaction(user_id, message, response)
             return response

tools.py

@@ -1,6 +1,7 @@
 import ast
 import datetime as dt
 import math
+import re
 from functools import lru_cache
 from typing import Dict, List
 
@@ -127,3 +128,14 @@ def datetime_tool() -> str:
         f"Current date: {now.strftime('%Y-%m-%d')}\n"
         f"Current time: {now.strftime('%H:%M:%S %Z')}"
     )
+
+
+def text_stats_tool(text: str) -> str:
+    cleaned = text.strip()
+    if not cleaned:
+        return "Words: 0\nCharacters: 0\nSentences: 0"
+
+    words = len(re.findall(r"\b\w+\b", cleaned))
+    chars = len(cleaned)
+    sentences = len([s for s in re.split(r"[.!?]+", cleaned) if s.strip()])
+    return f"Words: {words}\nCharacters: {chars}\nSentences: {sentences}"