File size: 2,535 Bytes
b7fa003
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
# Fonctions utilitaires
from typing import Any, Dict, List


def filter_tools_for_context(tools: List[Any], conversation_messages: List[Dict[str, Any]], keywords: List[str]) -> List[Any]:
    """Filtrage simple des outils selon contexte récent et mots-clés."""
    recent_text = " ".join(
        msg["content"] if isinstance(msg.get("content"), str) else " ".join(
            item.get("text", "") for item in msg.get("content", []) if isinstance(item, dict)
        )
        for msg in conversation_messages[-5:]
    ).lower()

    selected = []
    for tool in tools:
        name = (tool.get("name") if isinstance(tool, dict) else getattr(tool, "name", "")).lower()
        desc = (tool.get("description") if isinstance(tool, dict) else getattr(tool, "description", "")).lower()
        if not keywords or any(kw.lower() in (name + desc + recent_text) for kw in keywords):
            selected.append(tool)
    return selected


def summarize_latest_results(conversation_messages: List[Dict[str, Any]]) -> str:
    """Résumé des derniers résultats outils."""
    summaries = []
    for msg in conversation_messages:
        if msg.get("role") == "user" and isinstance(msg.get("content"), list):
            for item in msg["content"]:
                if isinstance(item, dict) and item.get("type") == "tool_result":
                    summaries.append(item.get("content", ""))
    return "\n".join(summaries[-5:]).strip()


def count_tokens(conversation_messages: List[Dict[str, Any]]) -> int:
    """Comptage naïf de tokens (optimisation mémoire)."""
    total = 0
    for msg in conversation_messages:
        content = msg.get("content")
        if isinstance(content, str):
            total += len(content.split())
        elif isinstance(content, list):
            for item in content:
                if isinstance(item, dict) and "text" in item:
                    total += len(item["text"].split())
                elif isinstance(item, str):
                    total += len(item.split())
    return total


def trim_conversation(conversation_messages: List[Dict[str, Any]], keep_last_n: int = 5) -> List[Dict[str, Any]]:
    """Réduction du contexte conversationnel."""
    if len(conversation_messages) <= keep_last_n:
        return conversation_messages
    trimmed = conversation_messages[-keep_last_n:]
    trimmed.insert(0, {"role": "system", "content": "Résumé des messages précédents supprimés pour raison de contexte."})
    return trimmed