Implement language detection in the agent

agents/cogsolframeworkagent/agent.py

@@ -1,13 +1,14 @@
 from cogsol.agents import BaseAgent, genconfigs
 from cogsol.prompts import Prompts
 from ..searches import CogsolFrameworkDocsSearch, CogsolAPIsDocsSearch
-from ..tools import CogSolScaffoldGenerator
+from ..tools import CogSolScaffoldGenerator, LanguageAndMessageNotInfo
 
 
 class CogsolFrameworkAgent(BaseAgent):
     system_prompt = Prompts.load("cogsolframeworkagent.md")
     generation_config = genconfigs.QA()
     tools = [CogsolFrameworkDocsSearch(), CogsolAPIsDocsSearch(), CogSolScaffoldGenerator()]
+    pretools = [LanguageAndMessageNotInfo()]
     max_responses = 20
     max_msg_length = 2048
     max_consecutive_tool_calls = 3

agents/migrations/0004_auto_20260325_1545.py

@@ -0,0 +1,12 @@
+# Generated by CogSol 0.2.1 on 2026-03-25 15:45
+from cogsol.db import migrations
+
+
+class Migration(migrations.Migration):
+    initial = False
+    dependencies = [('agents', '0003_auto_20260115_1059')]
+    operations = [
+        migrations.AlterField(model_name='CogSolScaffoldGenerator', name='__code__', value='def _to_class_name(self, name: str) -> str:\n        """Convert name to PascalCase class name."""\n        # Remove non-alphanumeric chars and split\n        import re\n        words = re.split(r\'[\\s_\\-]+\', name)\n        return \'\'.join(word.capitalize() for word in words if word)\n\ndef _to_snake_case(self, name: str) -> str:\n        """Convert name to snake_case."""\n        import re\n        # Insert underscore before uppercase letters and convert to lowercase\n        s1 = re.sub(r\'[\\s\\-]+\', \'_\', name)\n        s2 = re.sub(r\'([a-z])([A-Z])\', r\'\\1_\\2\', s1)\n        return s2.lower()\n\ndef _generate_agent(self, class_name: str, snake_name: str, desc: str, options: dict) -> str:\n        tools_import = ""\n        tools_list = "[]"\n        if options.get("tools"):\n            tool_names = options["tools"]\n            tools_import = f"\\nfrom .tools import {\', \'.join(tool_names)}"\n            tools_list = f"[{\', \'.join(f\'{t}()\' for t in tool_names)}]"\n\n        temperature = options.get("temperature", 0.3)\n\n        return f\'\'\'from cogsol.agents import BaseAgent, genconfigs\nfrom cogsol.prompts import Prompts{tools_import}\n\n\nclass {class_name}Agent(BaseAgent):\n    """\n    {desc}\n    """\n    # Core configuration\n    system_prompt = Prompts.load("{snake_name}.md")\n    generation_config = genconfigs.QA()\n    temperature = {temperature}\n\n    # Tools\n    tools = {tools_list}\n    pretools = []\n\n    # Limits\n    max_interactions = 20\n    user_message_length = 2048\n    consecutive_tool_calls_limit = 5\n\n    # Behaviors\n    initial_message = "Hello! How can I help you today?"\n    no_information_message = "I don\'t have information on that topic."\n\n    # Features\n    streaming = False\n    realtime = False\n\n    class Meta:\n        name = "{class_name}Agent"\n        chat_name = "{class_name.replace(\'_\', \' \')}"\n        # logo_url = "https://example.com/logo.png"\n        # primary_color = "#007bff"\'\'\'\n\ndef _generate_tool(self, class_name: str, snake_name: str, desc: str, options: dict) -> str:\n        # Build parameters from options or use default example\n        params = options.get("parameters", [\n            {"name": "query", "description": "Input query", "type": "string", "required": True}\n        ])\n\n        param_decorators = []\n        param_args = []\n        param_docs = []\n\n        for p in params:\n            p_name = p.get("name", "param") if isinstance(p, dict) else p\n            p_desc = p.get("description", "Parameter description") if isinstance(p, dict) else "Parameter description"\n            p_type = p.get("type", "string") if isinstance(p, dict) else "string"\n            p_required = p.get("required", False) if isinstance(p, dict) else False\n\n            param_decorators.append(\n                f\'        {p_name}={{"description": "{p_desc}", "type": "{p_type}", "required": {p_required}}}\'\n            )\n\n            py_type = {"string": "str", "integer": "int", "boolean": "bool", "number": "float"}.get(p_type, "str")\n            default = {"str": \'""\', "int": "0", "bool": "False", "float": "0.0"}.get(py_type, \'""\')\n            param_args.append(f"{p_name}: {py_type} = {default}")\n            param_docs.append(f"        {p_name}: {p_desc}")\n\n        params_str = ",\\n".join(param_decorators)\n        args_str = ", ".join(param_args)\n        docs_str = "\\n".join(param_docs)\n\n        return f\'\'\'from cogsol.tools import BaseTool, tool_params\n\n\nclass {class_name}Tool(BaseTool):\n    """\n    {desc}\n    """\n    name = "{snake_name}"\n    description = "{desc}"\n\n    @tool_params(\n{params_str}\n    )\n    def run(self, chat=None, data=None, secrets=None, log=None, {args_str}):\n        """\n{docs_str}\n        """\n        if log:\n            log(f"Running {class_name}Tool...")\n\n        # TODO: Implement your tool logic here\n        result = "Tool executed successfully"\n\n        return result\'\'\'\n\ndef _generate_retrieval_tool(self, class_name: str, snake_name: str, desc: str, options: dict) -> str:\n        retrieval_class = options.get("retrieval_class", f"{class_name}Retrieval")\n\n        return f\'\'\'from cogsol.tools import BaseRetrievalTool\nfrom data.retrievals import {retrieval_class}\n\n\nclass {class_name}Search(BaseRetrievalTool):\n    """\n    {desc}\n    """\n    name = "{snake_name}_search"\n    description = "{desc}"\n    retrieval = {retrieval_class}()\n    # Optional: customize parameters (default includes \'question\')\n    # parameters = [\n    #     {{"name": "question", "description": "Search query", "type": "string", "required": True}}\n    # ]\'\'\'\n\ndef _generate_faq(self, class_name: str, snake_name: str, desc: str, options: dict) -> str:\n        question = options.get("question", "What is the answer to this common question?")\n        answer = options.get("answer", desc)\n\n        return f\'\'\'from cogsol.tools import BaseFAQ\n\n\nclass {class_name}FAQ(BaseFAQ):\n    """\n    {desc}\n    """\n    question = "{question}"\n    answer = """{answer}"""\'\'\'\n\ndef _generate_fixed_response(self, class_name: str, snake_name: str, desc: str, options: dict) -> str:\n        key = options.get("key", snake_name)\n        response = options.get("response", desc)\n\n        return f\'\'\'from cogsol.tools import BaseFixedResponse\n\n\nclass {class_name}Fixed(BaseFixedResponse):\n    """\n    {desc}\n    """\n    key = "{key}"\n    response = """{response}"""\'\'\'\n\ndef _generate_lesson(self, class_name: str, snake_name: str, desc: str, options: dict) -> str:\n        content = options.get("content", desc)\n        context = options.get("context_of_application", "general")\n\n        return f\'\'\'from cogsol.tools import BaseLesson\n\n\nclass {class_name}Lesson(BaseLesson):\n    """\n    {desc}\n    """\n    name = "{class_name.replace(\'_\', \' \')}"\n    content = """{content}"""\n    context_of_application = "{context}"\'\'\'\n\ndef _generate_topic(self, class_name: str, snake_name: str, desc: str, options: dict) -> str:\n        return f\'\'\'from cogsol.content import BaseTopic\n\n\nclass {class_name}Topic(BaseTopic):\n    """\n    {desc}\n    """\n    name = "{snake_name}"\n\n    class Meta:\n        description = "{desc}"\'\'\'\n\ndef _generate_metadata_config(self, class_name: str, snake_name: str, desc: str, options: dict) -> str:\n        meta_type = options.get("type", "STRING")\n        values = options.get("possible_values", [])\n        values_str = f"\\n    possible_values = {values}" if values else ""\n\n        return f\'\'\'from cogsol.content import BaseMetadataConfig, MetadataType\n\n\nclass {class_name}Metadata(BaseMetadataConfig):\n    """\n    {desc}\n    """\n    name = "{snake_name}"\n    type = MetadataType.{meta_type.upper()}{values_str}\n    filtrable = True\n    required = False\'\'\'\n\ndef _generate_ingestion_config(self, class_name: str, snake_name: str, desc: str, options: dict) -> str:\n        pdf_mode = options.get("pdf_parsing_mode", "OCR")\n        chunking = options.get("chunking_mode", "AGENTIC_SPLITTER")\n        max_size = options.get("max_size_block", 2000)\n\n        return f\'\'\'from cogsol.content import BaseIngestionConfig, PDFParsingMode, ChunkingMode\n\n\nclass {class_name}Config(BaseIngestionConfig):\n    """\n    {desc}\n    """\n    name = "{snake_name}"\n    pdf_parsing_mode = PDFParsingMode.{pdf_mode}\n    chunking_mode = ChunkingMode.{chunking}\n    max_size_block = {max_size}\n    chunk_overlap = 100\'\'\'\n\ndef _generate_retrieval(self, class_name: str, snake_name: str, desc: str, options: dict) -> str:\n        topic = options.get("topic", snake_name)\n        num_refs = options.get("num_refs", 10)\n\n        return f\'\'\'from cogsol.content import BaseRetrieval, ReorderingStrategy\n# from data.formatters import DetailedFormatter  # Uncomment if using custom formatters\n\n\nclass {class_name}Retrieval(BaseRetrieval):\n    """\n    {desc}\n    """\n    name = "{snake_name}_search"\n    topic = "{topic}"\n    num_refs = {num_refs}\n    reordering = False\n    strategy_reordering = ReorderingStrategy.NONE\n    # formatters = {{"Text Document": DetailedFormatter}}  # Uncomment for custom formatting\n    filters = []\'\'\'\n\ndef _get_next_steps(self, component_type: str, class_name: str, snake_name: str) -> str:\n        steps = {\n            "agent": f"""1. Create the prompt file at `agents/{snake_name}agent/prompts/{snake_name}.md`\n2. Add the agent file at `agents/{snake_name}agent/agent.py`\n3. Create `__init__.py` that exports your agent\n4. Run `python manage.py makemigrations agents` and `python manage.py migrate agents`""",\n\n            "tool": """1. Add this class to `agents/tools.py` or `agents/<your_agent>/tools.py`\n2. Import and add it to your agent\'s `tools` list\n3. Implement the tool logic in the `run` method""",\n\n            "retrieval_tool": """1. Ensure the referenced Retrieval exists in `data/retrievals.py`\n2. Add this class to `agents/searches.py`\n3. Import and add it to your agent\'s `tools` list""",\n\n            "faq": """1. Add this class to `agents/<your_agent>/faqs.py`\n2. The agent will automatically load FAQs from the faqs.py file""",\n\n            "fixed_response": """1. Add this class to `agents/<your_agent>/fixed.py`\n2. The agent will automatically load fixed responses""",\n\n            "lesson": """1. Add this class to `agents/<your_agent>/lessons.py`\n2. The agent will automatically load lessons""",\n\n            "topic": f"""1. Create directory `data/{snake_name}/`\n2. Add this code to `data/{snake_name}/__init__.py`\n3. Create `data/{snake_name}/metadata.py` for metadata configs\n4. Run migrations: `python manage.py makemigrations data` and `python manage.py migrate data`""",\n\n            "metadata_config": """1. Add this class to `data/<topic>/metadata.py`\n2. Run `python manage.py makemigrations data`\n3. Run `python manage.py migrate data`""",\n\n            "ingestion_config": f"""1. Add this class to `data/ingestion.py`\n2. Use with: `python manage.py ingest <topic> <path> --ingestion-config {snake_name}`""",\n\n            "retrieval": """1. Add this class to `data/retrievals.py`\n2. Ensure the topic exists in `data/<topic>/`\n3. Run `python manage.py makemigrations data` and `python manage.py migrate data`\n4. Create a retrieval tool in `agents/searches.py` to use it"""\n        }\n\n        return steps.get(component_type, "Check the CogSol documentation for next steps.")\n\n@tool_params(\n    component_type={\n        "description": "Type of component to generate: \'agent\', \'tool\', \'retrieval_tool\', \'faq\', \'fixed_response\', \'lesson\', \'topic\', \'metadata_config\', \'ingestion_config\', or \'retrieval\'",\n        "type": "string",\n        "required": True\n    },\n    name={\n        "description": "Name for the component (e.g., \'CustomerSupport\', \'ProductSearch\'). Will be used as class name.",\n        "type": "string",\n        "required": True\n    },\n    description={\n        "description": "Brief description of what this component does. Used in docstrings and description fields.",\n        "type": "string",\n        "required": False\n    },\n    extra_options={\n        "description": "Optional JSON string with extra options. For tools: parameters list. For agents: tool names, temperature. For retrievals: topic name, num_refs.",\n        "type": "string",\n        "required": False\n    }\n)\ndef run(self, chat=None, data=None, secrets=None, log=None, component_type: str = "", name: str = "", description: str = "", extra_options: str = ""):\n    """\n    component_type: The type of CogSol component to generate.\n    name: The name for the new component class.\n    description: Description of what the component does.\n    extra_options: Additional configuration as JSON string.\n    """\n    import json\n\n    if not component_type or not name:\n        return "Error: Both \'component_type\' and \'name\' are required."\n\n    # Parse extra options if provided\n    options = {}\n    if extra_options:\n        try:\n            options = json.loads(extra_options)\n        except json.JSONDecodeError:\n            pass  # Use empty options if parsing fails\n\n    # Clean the name to be a valid Python class name\n    class_name = self._to_class_name(name)\n    snake_name = self._to_snake_case(name)\n    desc = description or f"A {component_type} for {name}"\n\n    generators = {\n        "agent": self._generate_agent,\n        "tool": self._generate_tool,\n        "retrieval_tool": self._generate_retrieval_tool,\n        "faq": self._generate_faq,\n        "fixed_response": self._generate_fixed_response,\n        "lesson": self._generate_lesson,\n        "topic": self._generate_topic,\n        "metadata_config": self._generate_metadata_config,\n        "ingestion_config": self._generate_ingestion_config,\n        "retrieval": self._generate_retrieval,\n    }\n\n    generator = generators.get(component_type.lower())\n    if not generator:\n        valid_types = ", ".join(generators.keys())\n        return f"Error: Unknown component type \'{component_type}\'. Valid types are: {valid_types}"\n\n    code = generator(class_name, snake_name, desc, options)\n\n    return f"## Generated {component_type.replace(\'_\', \' \').title()} Code\\n\\n```python\\n{code}\\n```\\n\\n**Next steps:**\\n{self._get_next_steps(component_type, class_name, snake_name)}"', entity='tools', scope='fields'),
+        migrations.CreateTool(name='LanguageAndMessageNotInfo', fields={'name': 'language_and_message_not_info', 'description': 'Tool to detect language and set context for assistant', 'parameters': {}, '__code__': 'def run(self, chat=None, data=None, secrets=None, log=None):\n    import pycld2 as cld2\n    from translate import Translator\n\n    if chat is None or not hasattr(chat, "messages"):\n        return {}\n\n    if data is None:\n        data = {}\n    prompt_params = data.setdefault("prompt_params", {})\n    context = prompt_params.setdefault("context", {})\n\n    assistant = getattr(chat, "assistant", None)\n    no_info_message = getattr(assistant, "not_info_message", "")\n\n    last_user = chat.messages.filter(role="user").order_by("msg_num").last()\n    last_user_msg = getattr(last_user, "content", "")\n\n    is_reliable, _text_bytes_found, details = cld2.detect(last_user_msg)\n    is_reliable_not_info, _text_bytes_not_info, details_not_info = cld2.detect(\n        no_info_message\n    )\n\n    if is_reliable:\n        context["The language you should answer to the user is"] = details[0][0]\n        if is_reliable_not_info:\n            context["Message of not having information"] = Translator(\n                to_lang=details[0][1], from_lang=details_not_info[0][1]\n            ).translate(no_info_message)\n        else:\n            context["Message of not having information"] = Translator(\n                to_lang=details[0][1], from_lang="es"\n            ).translate(no_info_message)\n    else:\n        context["The language you should answer to the user is"] = (\n            f"same language of the last message, that was: \'{last_user_msg}\'"\n        )\n        context["Message of not having information"] = no_info_message\n\n    return {}'}),
+        migrations.AlterField(model_name='CogsolFrameworkAgent', name='pretools', value=['language_and_message_not_info'], entity='agents', scope='fields'),
+    ]

agents/tools.py

@@ -366,3 +366,51 @@ def _get_next_steps(self, component_type: str, class_name: str, snake_name: str)
         }
 
         return steps.get(component_type, "Check the CogSol documentation for next steps.")
+
+
+class LanguageAndMessageNotInfo(BaseTool):
+    """Imported language pretool from tenant."""
+
+    name = "language_and_message_not_info"
+    description = "Tool to detect language and set context for assistant"
+
+    def run(self, chat=None, data=None, secrets=None, log=None):
+        import pycld2 as cld2
+        from translate import Translator
+
+        if chat is None or not hasattr(chat, "messages"):
+            return {}
+
+        if data is None:
+            data = {}
+        prompt_params = data.setdefault("prompt_params", {})
+        context = prompt_params.setdefault("context", {})
+
+        assistant = getattr(chat, "assistant", None)
+        no_info_message = getattr(assistant, "not_info_message", "")
+
+        last_user = chat.messages.filter(role="user").order_by("msg_num").last()
+        last_user_msg = getattr(last_user, "content", "")
+
+        is_reliable, _text_bytes_found, details = cld2.detect(last_user_msg)
+        is_reliable_not_info, _text_bytes_not_info, details_not_info = cld2.detect(
+            no_info_message
+        )
+
+        if is_reliable:
+            context["The language you should answer to the user is"] = details[0][0]
+            if is_reliable_not_info:
+                context["Message of not having information"] = Translator(
+                    to_lang=details[0][1], from_lang=details_not_info[0][1]
+                ).translate(no_info_message)
+            else:
+                context["Message of not having information"] = Translator(
+                    to_lang=details[0][1], from_lang="es"
+                ).translate(no_info_message)
+        else:
+            context["The language you should answer to the user is"] = (
+                f"same language of the last message, that was: '{last_user_msg}'"
+            )
+            context["Message of not having information"] = no_info_message
+
+        return {}