85 lines
2.7 KiB
Python
85 lines
2.7 KiB
Python
import json
|
|
|
|
from mistralai.client import Mistral
|
|
from . import config
|
|
|
|
_client = Mistral(api_key=config.MISTRAL_API_KEY)
|
|
|
|
_history: list[dict] = []
|
|
|
|
|
|
def reset_history() -> None:
|
|
"""Efface l'historique de conversation."""
|
|
_history.clear()
|
|
|
|
|
|
def chat(user_message: str) -> str:
|
|
"""Envoie un message au LLM, gère les appels d'outils MCP et retourne la réponse."""
|
|
from . import mcp_client
|
|
|
|
_history.append({"role": "user", "content": user_message})
|
|
|
|
manager = mcp_client.get_manager()
|
|
tools = manager.get_mistral_tools() if manager.has_tools else None
|
|
|
|
while True:
|
|
messages = [{"role": "system", "content": config.SYSTEM_PROMPT}] + _history
|
|
|
|
kwargs: dict = {"model": config.LLM_MODEL, "messages": messages}
|
|
if tools:
|
|
kwargs["tools"] = tools
|
|
|
|
response = _client.chat.complete(**kwargs)
|
|
choice = response.choices[0]
|
|
msg = choice.message
|
|
|
|
if msg.tool_calls:
|
|
# 1. Ajouter le message assistant (avec les appels d'outils) à l'historique
|
|
_history.append({
|
|
"role": "assistant",
|
|
"content": msg.content or "",
|
|
"tool_calls": [
|
|
{
|
|
"id": tc.id,
|
|
"type": "function",
|
|
"function": {
|
|
"name": tc.function.name,
|
|
"arguments": tc.function.arguments,
|
|
},
|
|
}
|
|
for tc in msg.tool_calls
|
|
],
|
|
})
|
|
|
|
# 2. Exécuter chaque outil et ajouter les résultats
|
|
for tc in msg.tool_calls:
|
|
tool_name = tc.function.name
|
|
try:
|
|
args = (
|
|
json.loads(tc.function.arguments)
|
|
if isinstance(tc.function.arguments, str)
|
|
else tc.function.arguments
|
|
)
|
|
print(f" 🔧 [MCP] {tool_name}({_short_args(args)})")
|
|
result = manager.call_tool(tool_name, args)
|
|
except Exception as e:
|
|
result = f"Erreur lors de l'appel à {tool_name} : {e}"
|
|
|
|
_history.append({
|
|
"role": "tool",
|
|
"content": result,
|
|
"tool_call_id": tc.id,
|
|
})
|
|
|
|
# 3. Reboucler pour obtenir la réponse finale
|
|
|
|
else:
|
|
reply = msg.content or ""
|
|
_history.append({"role": "assistant", "content": reply})
|
|
return reply
|
|
|
|
|
|
def _short_args(args: dict) -> str:
|
|
text = json.dumps(args, ensure_ascii=False)
|
|
return text[:80] + "…" if len(text) > 80 else text
|