From 8b39e0fe34f410e2197e60de3cc9018ac21acba7 Mon Sep 17 00:00:00 2001 From: Nicolas Fernandez Date: Wed, 25 Mar 2026 15:47:29 -0300 Subject: [PATCH] Add LanguageAndMessageNotInfo tool and integrate with CogsolFrameworkAgent --- agents/cogsolframeworkagent/agent.py | 3 +- agents/migrations/0004_auto_20260325_1545.py | 12 +++++ agents/tools.py | 48 ++++++++++++++++++++ 3 files changed, 62 insertions(+), 1 deletion(-) create mode 100644 agents/migrations/0004_auto_20260325_1545.py diff --git a/agents/cogsolframeworkagent/agent.py b/agents/cogsolframeworkagent/agent.py index 6daa30a..b8b5adf 100644 --- a/agents/cogsolframeworkagent/agent.py +++ b/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 diff --git a/agents/migrations/0004_auto_20260325_1545.py b/agents/migrations/0004_auto_20260325_1545.py new file mode 100644 index 0000000..d450dc7 --- /dev/null +++ b/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//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//faqs.py`\n2. The agent will automatically load FAQs from the faqs.py file""",\n\n "fixed_response": """1. Add this class to `agents//fixed.py`\n2. The agent will automatically load fixed responses""",\n\n "lesson": """1. Add this class to `agents//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//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 --ingestion-config {snake_name}`""",\n\n "retrieval": """1. Add this class to `data/retrievals.py`\n2. Ensure the topic exists in `data//`\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'), + ] diff --git a/agents/tools.py b/agents/tools.py index 89215d7..9dd7d0a 100644 --- a/agents/tools.py +++ b/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 {}