Je veux apporter des information en plus dans le c...

Question

Je veux apporter des information en plus dans le chatbot avec mon utilisateur j'ai setup un server whoogle et il marche bien j'aimerais que tu me créé une fonctionnaité de recherche internet qui quand elle est activé se deroule comme ceci :

prompt utilisateur
si recherche activé :
requete = analyser_le_prompt pour créé une requete sur internet(whoogle)
v = envois de la requete
siteweb= analyse des liens retourner sur google par v(donc les lieins trouvé dans la requeête whoogle en json ou autre)
for lien in site web:
a = scrapper_lecontenue(lien)
si informationquipeuventêtreinetressante dans a alors
variableinfointeressant += informaiton qui peuvent extraite

ajout de variable qui peuvent être interessant aux contexte de la requete et demarage d'un chat normal :
"""
from gevent import monkey
monkey.patch_all()
import os
import re
import json
import logging
from typing import Any, Dict, List, Optional, Tuple
import time
import requests

from flask import Flask, request, Response, jsonify, stream_with_context
from flask_cors import CORS
from bs4 import BeautifulSoup

from groq import Groq
from e2b_code_interpreter import Sandbox

Configuration du logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(name)

app = Flask(name)
CORS(app)

Clés API (variables d'environnement ou valeurs par défaut sécurisées)

GROQ_API_KEY = os.environ.get("GROQ_API_KEY")
if not GROQ_API_KEY:
logger.warning("GROQ_API_KEY non trouvé dans les variables d'environnement. Le service Groq ne fonctionnera pas.")
GROQ_API_KEY = "clé_groq_par_defaut_non_securisee" # A REMPLACER ou configurer via env var
E2B_API_KEY = os.environ.get("E2B_API_KEY")
if not E2B_API_KEY:
logger.warning("E2B_API_KEY non trouvé dans les variables d'environnement. Le sandbox e2b ne fonctionnera pas.")
E2B_API_KEY = "clé_e2b_par_defaut_non_securisee" # A REMPLACER ou configurer via env var

PISTON_API_URL = "https://emkc.org/api/v2/piston/execute" # URL de l'API Piston
WHOOGLE_INSTANCE_URL = os.environ.get("WHOOGLE_INSTANCE_URL", "http://localhost:5001/search") # URL Whoogle, localhost par défaut

groq_client = Groq(api_key=GROQ_API_KEY)

Patched line: Increased timeout to 600 seconds (10 minutes) to address potential Sandbox Timeout errors.

sandbox = Sandbox(api_key=E2B_API_KEY, timeout=600) # Timeout pour l'exécution du code en secondes

VALID_MODELS = [
"distil-whisper-large-v3-en", "gemma2-9b-it", "llama-3.3-70b-versatile", "llama-3.1-8b-instant",
"llama-guard-3-8b", "llama3-70b-8192", "llama3-8b-8192", "mixtral-8x7b-32768", "whisper-large-v3",
"whisper-large-v3-turbo", "deepseek-r1-distill-llama-70b-specdec", "deepseek-r1-distill-llama-70b",
"llama-3.3-specdec", "llama-3.2-1b-preview", "llama-3.2-3b-preview", "llama-3.2-11b-vision-preview",
"llama-3.2-90b-vision-preview", "llama3-70b-8192"
]

CODE_PATTERN = re.compile(r"python\s*([\s\S]*?)")
CODE_REQUIRED_MARKER = "[CODE_REQUIRED]"
CANVAS_START_MARKER = "[CANVAS_START]"
CANVAS_END_MARKER = "[CANVAS_END]"
WEB_SEARCH_MARKER = "[WEB_SEARCH]" # Marqueur pour déclencher explicitement la recherche web

Limites pour le contexte web (Whoogle)

MAX_CONTEXT_CHARS_PER_PAGE = 2000
MAX_TOTAL_CONTEXT_CHARS = 7000
WHOOGLE_CACHE: Dict[str, str] = {} # Cache pour les résultats de Whoogle

http_session = requests.Session() # Session HTTP globale pour réutilisation des connexions
http_session.headers.update({"User-Agent": "PlinkoChatBot/1.0"}) # User-Agent personnalisé

def whoogle_web_search(query: str) -> str:
"""
Effectue une recherche sur Whoogle (instance locale) et renvoie le texte extrait.
Utilise un cache pour limiter les appels répétés.
"""
if not WHOOGLE_INSTANCE_URL or "localhost" in WHOOGLE_INSTANCE_URL:
logger.warning("L'URL de Whoogle n'est pas configurée ou pointe vers localhost. Assurez-vous que Whoogle est accessible.")

text
if query in WHOOGLE_CACHE:
    logger.info("Utilisation du cache pour la requête Whoogle: %s", query)
    return WHOOGLE_CACHE[query]

context = ""
total_length = 0
try:
    params = {"q": query}
    response = http_session.get(WHOOGLE_INSTANCE_URL, params=params, timeout=30) # Timeout augmenté pour la recherche
    response.raise_for_status()  # Lève une exception pour les codes d'erreur HTTP
    soup = BeautifulSoup(response.text, 'html.parser')
    links = [a['href'] for a in soup.find_all('a', href=True) if a['href'].startswith('http')] # Liens absolus seulement
    for link in links[:3]: # Limite à 3 liens pour éviter un contexte trop long
        if total_length >= MAX_TOTAL_CONTEXT_CHARS:
            logger.info("Limite maximale de contexte web atteinte.")
            break
        try:
            page_resp = http_session.get(link, timeout=10) # Timeout par page plus court
            page_resp.raise_for_status()
            page_soup = BeautifulSoup(page_resp.text, 'html.parser')
            text = page_soup.get_text(separator='\n', strip=True)
            truncated_text = text[:MAX_CONTEXT_CHARS_PER_PAGE] # Troncature par page
            context += truncated_text + "\n\n"
            total_length += len(truncated_text)
            logger.info("Contenu récupéré depuis %s (longueur %d)", link, len(truncated_text))
        except requests.RequestException as page_e:
            logger.warning("Erreur lors de la récupération de la page %s : %s", link, page_e)
            continue # Continue avec le lien suivant en cas d'erreur de page
except requests.RequestException as e:
    logger.error("Recherche Whoogle échouée pour '%s' : %s", query, e)
    return "Erreur lors de la recherche web avec Whoogle." # Message d'erreur clair

WHOOGLE_CACHE[query] = context # Mise en cache du contexte récupéré
return context

def perform_web_search(query: str) -> str:
"""Wrapper pour la recherche web avec gestion du marqueur."""
if WEB_SEARCH_MARKER in query:
search_term = query.replace(WEB_SEARCH_MARKER, "").strip()
if search_term:
logger.info("Recherche web déclenchée par marqueur pour : %s", search_term)
return whoogle_web_search(search_term)
else:
logger.warning("Marqueur de recherche web détecté mais requête vide.")
return "" # Requête vide, pas de recherche
return "" # Pas de marqueur, pas de recherche

def extract_code_blocks(text: str) -> Tuple[Optional[str], Optional[str]]:
"""Extrait les blocs de code Python et gère le marqueur [CODE_REQUIRED]."""
if CODE_REQUIRED_MARKER in text:
text_no_marker = text.replace(CODE_REQUIRED_MARKER, "", 1).strip()
code_match = CODE_PATTERN.search(text_no_marker)
if code_match:

text
        return code_match.group(1).strip(), None
else:
    code_match = CODE_PATTERN.search(text)
    if code_match:
        return None, code_match.group(1).strip() # Pas de code à exécuter, code à afficher seulement
return None, None # Ni code à exécuter, ni code à afficher

def run_code_in_sandbox(code: str) -> str:
"""Exécute le code Python dans le sandbox e2b et retourne le résultat."""
start_time = time.time()
try:
logger.info("Début exécution code dans sandbox e2b.")
execution = None
method_found = False
for method in ["run_code_sync", "execute_sync", "execute_code", "execute", "run_code", "run"]: # Méthodes possibles
if hasattr(sandbox, method):
logger.info(f"Utilisation de la méthode sandbox.{method}")
execution = getattr(sandbox, method)(code)
method_found = True
break # Sortir dès qu'une méthode fonctionnelle est trouvée

text
    if not method_found:
        error_msg = "Aucune méthode d'exécution de code disponible dans le sandbox."
        logger.error(error_msg)
        return error_msg # Retourner message d'erreur clair

    logs = execution.logs
    stdout_logs = logs.stdout if hasattr(logs, 'stdout') else [] # Gestion si stdout n'est pas dispo
    stderr_logs = logs.stderr if hasattr(logs, 'stderr') else [] # Gestion si stderr n'est pas dispo

    stdout = "\n".join(stdout_logs)
    stderr = "\n".join(stderr_logs)

    if stderr: # Priorité aux erreurs
        return f"Stdout:\n{stdout}\n\nStderr:\n{stderr}"
    return stdout # Retourner stdout si pas d'erreur

except Exception as e:
    duration = time.time() - start_time
    error_msg = f"Erreur lors de l'exécution du code dans e2b après {duration:.2f} secondes: {str(e)}"
    logger.exception(error_msg)
    return f"Erreur exécution code: {str(e)}" # Message d'erreur plus informatif

def extract_numeric_result(exec_result: str) -> Optional[str]:
"""Extrait la dernière ligne numérique du résultat d'exécution de code."""
if "Stdout:" in exec_result: # Gestion si résultat contient Stdout/Stderr
stdout_section = exec_result.split("Stdout:\n", 1)[1].split("\n\nStderr:")[0].strip()
else:
stdout_section = exec_result.strip()
lines = [l.strip() for l in stdout_section.splitlines() if l.strip()] # Lignes non vides
if not lines:
return None # Retourne None si pas de lignes
last_line = lines[-1]
if re.search(r'\d', last_line): # Recherche d'un nombre dans la dernière ligne
return last_line # Retourne la dernière ligne si elle contient un nombre
return None # Retourne None si pas de nombre trouvé dans la dernière ligne

@app.route('/models', methods=['GET'])
def get_models() -> Response:
"""Retourne la liste des modèles valides."""
return jsonify({"models": VALID_MODELS})

@app.route('/chat', methods=['POST'])
def chat_endpoint() -> Response:
"""
Point d'entrée principal pour le chat. Gère le streaming SSE, l'exécution de code et les canvas.
Inclut la recherche web si activée et pertinente.
"""
data: Dict[str, Any] = request.get_json() or {}
messages: List[Dict[str, str]] = data.get('messages', [])
model: str = data.get('model', 'llama3-70b-8192').strip()
execute_code: bool = data.get('execute_code', True)
enable_web_search: bool = data.get('enable_web_search', False)

text
# Validation de la requête
if not messages or not isinstance(messages, list) or not model or model not in VALID_MODELS:
    logger.warning("Requête chat invalide: messages=%s, model=%s", messages, model)
    return jsonify({"error": "Requête chat invalide : messages ou modèle incorrects."}), 400

last_user_message = messages[-1]['content'] if messages else "" # Récupérer le dernier message utilisateur
web_search_context = ""
if enable_web_search:
    web_search_context = perform_web_search(last_user_message) # Recherche web si activée

# Construction du prompt système enrichi
system_prompt_content = (
    "🎯 **Rôle : Expert en calcul numérique précis et assistant polyvalent avec canvas interactifs (texte et code).**\n"
    "**Mission :** Répondre aux questions, effectuer des calculs numériques exacts, et assister l'utilisateur avec des canvas éditables et interactifs pour le **texte** (avec formatage Markdown) et le **code**. Adopte un style expert et méthodique. **Maintiens le contexte de conversation, en particulier autour du code, des canvas et des résultats précédents.**\n\n" # Ajout instruction contexte général
    "**Si la question DEMANDE un CALCUL NUMÉRIQUE :**\n"
    "1. Réponds d'abord avec une courte phrase d'introduction expliquant que tu vas utiliser un code Python pour résoudre le problème. **Sois clair sur le contexte du calcul, en te référant si nécessaire aux étapes précédentes de la conversation et aux données ou résultats déjà obtenus.**\n" # Ajout instruction contexte calcul
    "2. Sur la ligne suivante, ajoute TOUJOURS le marqueur `[CODE_REQUIRED]` pour indiquer qu'un code va suivre.\n"
    "3. Ensuite, fournis le **BLOC DE CODE PYTHON** minimal, lisible et correct pour effectuer UNIQUEMENT LE CALCUL demandé et afficher UNIQUEMENT LE RÉSULTAT NUMÉRIQUE via `print(resultat_numerique)`.\n"
    "**IMPORTANT :** Ne génère RIEN D'AUTRE que ces 3 éléments, et dans cet ordre précis. Arrête-toi **ICI**. Attends l'exécution du code.\n\n" # Instruction arrêt génération et attente execution
    "**[INSTRUCTION SUITE - APRES EXECUTION DU CODE]**\n" # Nouveau marqueur pour la suite des instructions
    "Après avoir fourni le résultat du code, CONTINUE la conversation pour expliquer le résultat EN DETAILS, en explicitant clairement et précisément **LE Resultat du code python** que tu viens d'utiliser** pour le calcul. IL faut que l'explication du resultat qui fasse avancé la conversation sois adapté a la situation. Concentre-toi UNIQUEMENT sur quelque chose qui repond a la reponse de l'utiilsateur .  Fais référence au contexte de la question initiale de l'utilisateur pour bien relier l'explication au besoin, et **mentionne explicitement  le résultat du code répond à la question ou fait avancer la conversation.**\n\n" # Instruction ajoutée et renforcée pour expliquer *seulement* le code utilisé et le relier au contexte, + lien résultat/question
    "**Si la question DEMANDE la CRÉATION ou MODIFICATION de CODE dans un canvas :**\n"
    "1. Indique clairement que tu vas créer un **canvas de code** interactif pour cela. Précise que l'utilisateur pourra te demander des modifications dans un 'chat de canvas' dédié qui va s'ouvrir.\n"
    "2. Encadre le contenu initial du canvas avec les marqueurs `[CANVAS_START]` et `[CANVAS_END]`.\n"
    "   Exemple : 'Très bien, un canvas de code interactif est lancé pour un quick sort en C. Vous pouvez maintenant me demander des modifications directement dans le chat de canvas qui vient de s'ouvrir :\n[CANVAS_START]\n```c\n// Code Quick Sort en C\n...\n```\n[CANVAS_END]'\n"
    "3. **IMPORTANT :** Ne fournis PAS de code markdown ```python ... ``` après le canvas, sauf si cela est explicitement demandé dans la question initiale et pertinent en dehors du canvas.\n\n"
    "**Si la question DEMANDE la CRÉATION ou MODIFICATION de TEXTE RICHE (avec markdown) dans un canvas :**\n"
    "1. Indique clairement que tu vas créer un **canvas de texte** interactif pour cela.  Précise que l'utilisateur pourra te demander des modifications dans un 'chat de canvas' dédié qui va s'ouvrir. Le texte dans ce canvas **supporte le formatage markdown**.\n"
    "2. Encadre le contenu initial du canvas de texte avec les marqueurs `[CANVAS_START]` et `[CANVAS_END]`.\n"
    "   Exemple : 'Très bien, un canvas de texte interactif est lancé. Le texte supporte le **gras**, l'*italique*, les listes, etc. Vous pouvez maintenant me demander des modifications directement dans le chat de canvas qui vient de s'ouvrir :\n[CANVAS_START]\n# Titre principal en gras\n\n* Item 1\n* Item 2\n\n[CANVAS_END]'\n"
    "3. **IMPORTANT :** Ne fournis PAS de code markdown ```python ... ``` après le canvas de texte, sauf si cela est explicitement demandé dans la question initiale et pertinent en dehors du canvas.\n\n"
    "**Si l'utilisateur pose une question dans le 'chat de canvas' :**\n"
    "  - Interprète sa demande comme une instruction pour MODIFIER le contenu du canvas actuel.\n"
    "  - Fournis le NOUVEAU contenu du canvas ENTIÈREMENT MIS À JOUR encadré par `[CANVAS_START]` et `[CANVAS_END]`.\n"
    "  - Ajoute une courte phrase de confirmation indiquant que le canvas a été mis à jour. **Maintiens le contexte des modifications précédentes du canvas et des instructions données dans le chat principal.**\n\n" # Ajout instruction contexte canvas chat
    "**Si la question NE DEMANDE PAS de CALCUL NUMÉRIQUE NI DE CANVAS :**\n"
    "Réponds directement et clairement à la question. Fournis une réponse informative et concise. **Prends en compte le contexte de la conversation pour formuler ta réponse.** " # Ajout instruction contexte réponse générale
    "Si tu dois donner un exemple de code, utilise un bloc markdown ```python ... ``` MAIS SANS JAMAIS ajouter `[CODE_REQUIRED]` ni de canvas."
)
if enable_web_search and web_search_context: # Ajouter contexte de recherche web au prompt si dispo
    system_prompt_content += f"\n\n**Contexte de recherche web (Whoogle):**\n{web_search_context}\n"

system_prompt = {"role": "system", "content": system_prompt_content}
messages_with_system_prompt = [system_prompt] + messages # Ajouter prompt système en tête

@stream_with_context
def stream_response():
    full_reply_text = ""
    code_to_execute = None
    code_to_display = None
    is_code_used = False
    canvas_content = None
    code_executed = False  # Flag pour indiquer si le code a été exécuté
    prompt_tokens = 0
    completion_tokens = 0
    total_tokens = 0
    initial_response_completed = False # Flag pour indiquer que la première partie de la réponse (avant execution) est terminée


    yield format_sse_event({"type": "analysis_status", "content": "start"}) # Début d'analyse

    try:
        response_stream = groq_client.chat.completions.create( # Appel à l'API Groq (streaming)
            model=model,
            messages=messages_with_system_prompt,
            temperature=0.0,
            stream=True,
        )

        for chunk in response_stream: # Itération sur les chunks de réponse
            if initial_response_completed: # Stop initial stream after code block
                break

            delta_content = chunk.choices[0].delta.content or ""
            # Supprimer le marqueur [CODE_REQUIRED] avant d'envoyer au frontend
            delta_content_no_marker = delta_content.replace(CODE_REQUIRED_MARKER, "")
            full_reply_text += delta_content_no_marker


            if CANVAS_START_MARKER in full_reply_text and CANVAS_END_MARKER in full_reply_text: # Détection canvas
                start_index = full_reply_text.find(CANVAS_START_MARKER) + len(CANVAS_START_MARKER)
                end_index = full_reply_text.find(CANVAS_END_MARKER)
                canvas_content_extracted = full_reply_text[start_index:end_index].strip()
                canvas_content = canvas_content_extracted
                delta_content_for_chat = delta_content_no_marker.replace(CANVAS_START_MARKER, "").replace(CANVAS_END_MARKER, "") # Nettoyage pour le chat
                yield format_sse_event({"type": "text", "content": delta_content_for_chat}) # Envoi texte (hors canvas markers)
                yield format_sse_event({"type": "canvas_start", "content": canvas_content}) # Envoi canvas
                continue # Passer au chunk suivant
            else:
                yield format_sse_event({"type": "text", "content": delta_content_no_marker}) # Envoi texte normal

            code_to_execute, code_to_display = extract_code_blocks(full_reply_text) # Extraction bloc code
            if code_to_execute:
                initial_response_completed = True # Marquer fin de première partie de réponse


        if execute_code and code_to_execute: # Exécution code si demandé et extrait
            is_code_used = True
            logger.info("Envoi événement CODE_BLOCK")
            yield format_sse_event({"type": "code_block", "content": code_to_execute}) # Envoi bloc de code

            yield format_sse_event({"type": "code_execution_start"}) # Indiquer début execution code (frontend)
            execution_result = run_code_in_sandbox(code_to_execute) # Execution dans sandbox
            numeric_result = extract_numeric_result(execution_result) # Extraction résultat numérique

            if numeric_result: # Si résultat numérique
                code_executed = True # Marquer comme exécuté
                logger.info("Envoi événement EXECUTED_CODE_RESULT")
                yield format_sse_event({"type": "executed_code_result", "content": numeric_result}) # Envoi résultat


                # Nouvelle requête à l'IA pour continuer la génération après le résultat du code, en demandant d'expliquer LE CODE UTILISE et uniquement cela.
                conclusion_prompt_content = f"**[INSTRUCTION SUITE - APRES EXECUTION DU CODE]** Tu as exécuté le code Python suivant:\n```python\n{code_to_execute}\n```\nLe résultat du calcul est: **{numeric_result}**. Maintenant, Continue pour une conversation logique et repondre a la question de l'utilisateur correcement Pour une a la question de base posée par l'utilisateur. il faut que le tout reste coherent car c'est apres l'execution du code que l'on voit dans le chat"
                conclusion_prompt = [{"role": "user", "content": conclusion_prompt_content}]
                conclusion_stream = groq_client.chat.completions.create(
                    model=model,
                    messages=conclusion_prompt,
                    temperature=0.0,
                    stream=True,
                )
                for chunk in conclusion_stream: # Stream de la conclusion
                    delta_content = chunk.choices[0].delta.content or ""
                    yield format_sse_event({"type": "text", "content": delta_content}) # Envoi de la conclusion en SSE text
                yield format_sse_event({"type": "final"}) # Fin de la réponse complète

            else: # Si erreur ou pas de résultat numérique
                code_executed = True # Marquer comme exécuté
                logger.info("Envoi événement EXECUTED_CODE_ERROR")
                yield format_sse_event({"type": "executed_code_error", "content": execution_result}) # Envoi erreur
                yield format_sse_event({"type": "final", "result": "Fin de la réponse (erreur ou pas de résultat numérique)."}) # Fin réponse

        # Récupération des tokens utilisés (Groq API ne retourne pas directement les tokens dans le stream,
        # il faudrait faire un appel séparé non-stream pour avoir ces infos, ou les estimer localement.
        # Pour simplifier, on met des valeurs à zéro ici -  à améliorer si Groq fournit ces infos)
        prompt_tokens = 0 # Remplacer par la vraie valeur si disponible
        completion_tokens = 0 # Remplacer par la vraie valeur si disponible
        total_tokens = prompt_tokens + completion_tokens


        if not code_executed: # Si pas de code exécuté (ou déjà géré ci-dessus)
            logger.info("Envoi événement FINAL (sans code exécuté)")
            yield format_sse_event({"type": "final", "prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens, "total_tokens": total_tokens}) # Fin de la réponse (standard)
        else:
             yield format_sse_event({"type": "final", "prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens, "total_tokens": total_tokens}) # Fin de la réponse (standard)


    except Exception as e: # Gestion des erreurs globales
        error_message = f"Erreur serveur IA: {str(e)}"
        logger.exception(error_message)
        yield format_sse_event({"type": "error", "content": error_message}) # Envoi événement erreur

    finally: # S'assurer que l'état d'analyse est toujours terminé
        yield format_sse_event({"type": "analysis_status", "content": "end"}) # Fin d'analyse
        yield format_sse_event({"type": "done", "is_code_execution": is_code_used}) # Marquer comme terminé (frontend)

return Response(stream_with_context(stream_response()), mimetype='text/event-stream')

@app.route('/canvas_modify', methods=['POST'])
def canvas_modify_endpoint() -> Response:
data = request.get_json()
if not data:
logger.warning("Requête canvas_modify sans données.")
return jsonify({"error": "Requête canvas_modify invalide : corps de requête manquant."}), 400

text
user_message = data.get('message')
canvas_content = data.get('canvas_content')
model = data.get('model', 'llama3-70b-8192').strip()

if not user_message or canvas_content is None:
    logger.warning("Requête canvas_modify incomplète: message=%s, canvas_content=%s", user_message, canvas_content)
    return jsonify({"error": "Requête canvas_modify invalide : message ou contenu canvas manquants."}), 400

canvas_chat_messages = [ # Messages pour le chat de canvas (prompting spécifique)
    {"role": "system", "content": "Tu es dans un chat de canvas. L'utilisateur va te demander de modifier le contenu du canvas suivant. Fournis UNIQUEMENT le NOUVEAU contenu du canvas ENTIÈREMENT MIS À JOUR, encadré par [CANVAS_START] et [CANVAS_END].  Ne réponds pas avec du texte en dehors de ces marqueurs, sauf une courte phrase de confirmation après le canvas. **Maintiens le contexte des modifications précédentes du canvas, en te basant sur l'historique de ce chat canvas.**"}, # Ajout instruction contexte canvas + historique canvas chat
    {"role": "assistant", "content": f"[CANVAS_START]\n{canvas_content}\n[CANVAS_END] Voici le canvas actuel. Dites-moi comment le modifier."},
    {"role": "user", "content": user_message}
]

try:
    response_stream = groq_client.chat.completions.create( # Appel API Groq (non streaming)
        model=model,
        messages=canvas_chat_messages,
        temperature=0.0,
        stream=False,
    )
    ai_response_text = response_stream.choices[0].message.content or "" # Récupérer texte réponse IA

    if CANVAS_START_MARKER in ai_response_text and CANVAS_END_MARKER in ai_response_text: # Vérifier marqueurs canvas
        start_index = ai_response_text.find(CANVAS_START_MARKER) + len(CANVAS_START_MARKER)
        end_index = ai_response_text.find(CANVAS_END_MARKER)
        updated_canvas_content = ai_response_text[start_index:end_index].strip() # Extraction contenu canvas
        response_message = "Canvas mis à jour." # Message de confirmation
        logger.info("Canvas modifié avec succès.")
        return jsonify({"updated_canvas_content": updated_canvas_content, "response_message": response_message}) # Retourne canvas mis à jour
    else:
        logger.error("Réponse IA canvas_modify invalide: marqueurs CANVAS_START/END manquants.")
        return jsonify({"error": "Réponse IA invalide pour la modification du canvas", "full_response": ai_response_text}), 500 # Erreur si marqueurs manquants

except Exception as e: # Gestion erreurs
    error_message = f"Erreur serveur (canvas_modify): {str(e)}"
    logger.exception(error_message)
    return jsonify({"error": error_message}), 500 # Retourne erreur serveur

@app.route('/run', methods=['POST'])
def run_code_endpoint() -> Response:
"""
Point d'accès pour exécuter du code via l'API Piston (pour canvas code).
Langages supportés par Piston.
"""
data = request.get_json()
if not data:
logger.warning("Requête run sans données.")
return jsonify({"error": "Requête run invalide : corps de requête manquant."}), 400

text
language = data.get("language")
code = data.get("code")
stdin = data.get("stdin", "")

if not language or not code:
    logger.warning("Requête run incomplète: language=%s, code=%s", language, code)
    return jsonify({"error": "Requête run invalide : langage et code requis."}), 400

payload = { # Payload pour l'API Piston
    "language": language,
    "version": "*",  # Dernière version du langage
    "files": [{"content": code}],
    "stdin": stdin
}

try:
    piston_response = http_session.post(PISTON_API_URL, json=payload, timeout=30) # Appel API Piston (timeout augmenté)
    piston_response.raise_for_status() # Gestion erreurs HTTP Piston

    logger.info("Code exécuté avec Piston avec succès.")
    return jsonify(piston_response.json()) # Retourne la réponse brute de Piston

except requests.RequestException as e: # Gestion erreurs requête Piston
    error_message = f"Erreur lors de l'appel à l'API Piston: {str(e)}"
    logger.exception(error_message)
    piston_text = piston_response.text if 'piston_response' in locals() and piston_response is not None else "" # Récupérer texte réponse Piston si dispo
    return jsonify({"error": "Erreur d'exécution du code via l'API Piston.", "piston_response": piston_text, "details": error_message}), 500 # Retourne erreur + détails

except Exception as e: # Gestion autres erreurs
    error_message = f"Erreur inattendue lors de l'exécution du code: {str(e)}"
    logger.exception(error_message)
    return jsonify({"error": "Erreur inattendue lors de l'exécution du code.", "details": error_message}), 500

@app.route('/run', methods=['OPTIONS'])
def run_code_options() -> Response:
"""Gère les requêtes OPTIONS pour l'endpoint /run (CORS preflight)."""
return jsonify({'message': 'OPTIONS requests allowed'}), 200 # Réponse CORS preflight

def format_sse_event(data: Dict[str, Any]) -> str:
"""Formate les données en event SSE (Server-Sent Events)."""
return f"data: {json.dumps(data)}\n\n" # Format SSE standard

if name == 'main':
if os.environ.get('FLASK_ENV') == 'development':
app.run(debug=True, port=5000)
"""

Answer 1

Voici un exemple de solution en Python qui ajoute une fonctionnalité de recherche internet « intelligente » à votre chatbot. L’idée est d’analyser la requête utilisateur pour en extraire une requête de recherche, d’utiliser Whoogle pour obtenir les résultats (les liens), puis de scrapper chaque page pour en extraire les passages potentiellement intéressants (par exemple, les phrases contenant des mots-clés issus de la requête). Ces informations seront ensuite ajoutées au contexte de la discussion.

Vous pouvez, par exemple, ajouter les fonctions suivantes à votre code existant (en complément de la fonction déjà présente qui interroge Whoogle) :

python
def extract_interesting_info(content: str, query: str) -> str:
    """
    Extrait des phrases intéressantes du contenu qui contiennent au moins un mot-clé issu de la requête.
    Cette fonction découpe le texte en phrases et ne garde que celles qui contiennent un des mots-clés.
    """
    # On considère comme mots-clés tous les termes de la requête
    keywords = query.split()
    # Découpage simple en phrases (vous pouvez utiliser une approche plus fine si besoin)
    sentences = content.split('.')
    interesting_sentences = []
    for sentence in sentences:
        # Vérifier si un des mots-clés est présent dans la phrase
        for kw in keywords:
            if kw.lower() in sentence.lower():
                interesting_sentences.append(sentence.strip())
                break  # On arrête dès qu'un mot-clé est trouvé dans la phrase
    return ". ".join(interesting_sentences)

def perform_internet_search(query: str) -> str:
    """
    Analyse la requête utilisateur afin de créer une requête de recherche,
    interroge l'instance Whoogle, récupère les liens obtenus et scrappe chacune des pages pour en extraire
    les informations intéressantes en fonction de la requête.
    
    Retourne une chaîne contenant les informations pertinentes à ajouter au contexte.
    """
    search_query = query.strip()
    if not search_query:
        return ""
    
    interesting_info = ""
    try:
        # Envoyer la requête à Whoogle
        params = {"q": search_query}
        response = http_session.get(WHOOGLE_INSTANCE_URL, params=params, timeout=30)
        response.raise_for_status()
        soup = BeautifulSoup(response.text, 'html.parser')
        # Extraction des liens absolus (ceux qui commencent par "http")
        links = [a['href'] for a in soup.find_all('a', href=True) if a['href'].startswith('http')]
        logger.info("Recherche Whoogle pour '%s' trouvée %d liens", search_query, len(links))
        # Limiter à 3 liens afin d'éviter d'encombrer le contexte
        for link in links[:3]:
            try:
                page_resp = http_session.get(link, timeout=10)
                page_resp.raise_for_status()
                page_soup = BeautifulSoup(page_resp.text, 'html.parser')
                page_text = page_soup.get_text(separator='\n', strip=True)
                # Extraire les phrases qui semblent intéressantes en fonction des mots-clés
                info = extract_interesting_info(page_text, search_query)
                # On ajoute l'info si elle est suffisamment longue (par exemple, plus de 50 caractères)
                if info and len(info) > 50:
                    interesting_info += f"Source: {link}\n{

**Warning:**
The response is not complete, would not consume credits.

Je veux apporter des information en plus dans le c...

Question

Configuration du logging

Clés API (variables d'environnement ou valeurs par défaut sécurisées)

Patched line: Increased timeout to 600 seconds (10 minutes) to address potential Sandbox Timeout errors.

Limites pour le contexte web (Whoogle)

Réflexion

Réponse

Partager cette Q&R