13.08.2026 20:25 перед внедрением коррекции поведения ИИ при использовании инструментов (создание флагов мусорных сообщений для очистки)

2026-08-13 20:25:27 +03:00
parent fc1a608a5f
commit f481fffdb7
18 changed files with 1562 additions and 1254 deletions
@@ -0,0 +1,53 @@
+"""
+Вспомогательные утилиты динамического календаря и парсинга дат.
+"""
+import re
+from datetime import datetime, timedelta
+
+DAYS_RU = [
+    "понедельник", "вторник", "среда", "четверг", 
+    "пятница", "суббота", "воскресенье"
+]
+
+def parse_relative_date_ru(text: str) -> str:
+    now = datetime.now()
+    text_lower = text.lower() if text else ""
+    
+    match = re.search(r'(\d{2}\.\d{2}\.\d{4})', text)
+    if match:
+        return match.group(1)
+
+    if "вчера" in text_lower:
+        return (now - timedelta(days=1)).strftime("%d.%m.%Y")
+    elif "позавчера" in text_lower:
+        return (now - timedelta(days=2)).strftime("%d.%m.%Y")
+    elif "сегодня" in text_lower:
+        return now.strftime("%d.%m.%Y")
+        
+    return (now - timedelta(days=1)).strftime("%d.%m.%Y")
+
+def get_dynamic_calendar_context() -> str:
+    now = datetime.now()
+    current_wd = now.weekday()
+    
+    lines = [
+        f"СЕГОДНЯ: {DAYS_RU[current_wd].upper()}, {now.strftime('%d.%m.%Y')} (время сервера: {now.strftime('%H:%M:%S')}).",
+        "\nСПРАВОЧНИК ДАТ ДЛЯ ОТВЕТОВ (БЕРИ ДАТЫ СТРОГО ОТСЮДА):",
+        f"• Сегодня: {now.strftime('%d.%m.%Y')} ({DAYS_RU[current_wd]})",
+        f"• Вчера: {(now - timedelta(days=1)).strftime('%d.%m.%Y')} ({DAYS_RU[(current_wd - 1) % 7]})",
+        f"• Позавчера: {(now - timedelta(days=2)).strftime('%d.%m.%Y')} ({DAYS_RU[(current_wd - 2) % 7]})",
+        "\nПрошедшие дни недели:"
+    ]
+    
+    for days_back in range(1, 8):
+        dt = now - timedelta(days=days_back)
+        day_name = DAYS_RU[dt.weekday()]
+        
+        if days_back == 7:
+            label = f"Прошлый {day_name}" if dt.weekday() in [0, 1, 3, 6] else f"Прошлая {day_name}"
+            lines.append(f"• {label} (ровно неделю назад): {dt.strftime('%d.%m.%Y')}")
+        else:
+            label = f"Ближайший прошедший {day_name}" if dt.weekday() in [0, 1, 3, 6] else f"Ближайшая прошедшая {day_name}"
+            lines.append(f"• {label} / {day_name}: {dt.strftime('%d.%m.%Y')}")
+            
+    return "\n".join(lines)
@@ -0,0 +1,71 @@
+"""
+Модуль чистки сырых тегов Ollama и перехвата Tool-вызовов (Tool Injector).
+"""
+import re
+import json
+import logging
+from typing import List, Dict, Any
+
+logger = logging.getLogger("TOOL_INJECTOR")
+
+def clean_raw_tool_tags(text: str) -> str:
+    if not text:
+        return ""
+    text = re.sub(r'\{"name":\s*"db_[^}]+\}\s*(</tool_call>)?', '', text)
+    text = re.sub(r'<tool_call>.*?</tool_call>', '', text, flags=re.DOTALL)
+    text = re.sub(r'</tool_call>\w*\[\]\(\)', '', text)
+    text = re.sub(r'</tool_call>', '', text)
+    return text.strip()
+
+def clean_raw_tool_tags(text: str) -> str:
+    if not text:
+        return ""
+    text = re.sub(r'\{"name":\s*"db_[^}]+\}\s*(</tool_call>)?', '', text)
+    text = re.sub(r'<tool_call>.*?</tool_call>', '', text, flags=re.DOTALL)
+    text = re.sub(r'</tool_call>\w*\[\]\(\)', '', text)
+    text = re.sub(r'</tool_call>', '', text)
+    return text.strip()
+
+def clean_output(text: str) -> str:
+    if not text:
+        return text
+    artifacts = ["почемучто", "почто", "почему что"]
+    lower_text = text.lower()
+    for art in artifacts:
+        if lower_text.startswith(art):
+            text = text[len(art):].lstrip(",.!?:; -")
+    return text.strip()
+
+def inject_tools_if_needed(user_message: str, raw_text_content: str, tool_calls: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    if tool_calls:
+        return tool_calls
+
+    # Если Ollama отдала JSON-структуру вызова функции как простой текст — помогаем распарсить
+    if '{"name":' in raw_text_content or '<tool_call>' in raw_text_content:
+        try:
+            match = re.search(r'\{"name":\s*"([^"]+)",\s*"(?:params|arguments|properties)":\s*(\{.*?\})\}', raw_text_content)
+            if match:
+                fn_name = match.group(1)
+                fn_args = json.loads(match.group(2))
+                return [{"function": {"name": fn_name, "arguments": fn_args}}]
+        except Exception:
+            pass
+
+    return tool_calls
+
+def inject_tools_if_needed(user_message: str, raw_text_content: str, tool_calls: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    if tool_calls:
+        return tool_calls
+
+    # Единственная задача Python — поймать JSON, если модель написала его текстом
+    if '{"name":' in raw_text_content or '<tool_call>' in raw_text_content:
+        try:
+            match = re.search(r'\{"name":\s*"([^"]+)",\s*"(?:params|arguments|properties)":\s*(\{.*?\})\}', raw_text_content)
+            if match:
+                fn_name = match.group(1)
+                fn_args = json.loads(match.group(2))
+                return [{"function": {"name": fn_name, "arguments": fn_args}}]
+        except Exception:
+            pass
+
+    return tool_calls