gremond/baklava
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
baklava/doc/AsynchronousTextualPrompt.md

771 lines
27 KiB
Markdown
Raw Normal View History

docs(Asynchronous-prompt-UI): variable's search engine and LLM UI prompt
2026-06-30 12:15:29 +02:00
---
description: python-textual comme interface à un LLM, est-ce que python-textual supporte la réponse d'une api-streaming (est-ce que la event loop de textual autorise l'asynchrone) ?
---
Voici votre application Textual modifiée pour intégrer un appel asynchrone à Ollama. J'ai ajouté la fonctionnalité qui permet de poser une question à Ollama et d'afficher la réponse en streaming.
```python
from textual.app import App, ComposeResult
from textual.containers import Container, Vertical, Horizontal
from textual.widgets import Header, Footer, Input, ListView, ListItem, Label, Static, Button, TextArea
from textual import events, work
from textual.reactive import reactive
from textual.message import Message
import asyncio
import httpx
import json
from typing import List, Optional, Dict, Any
class TermListItem(ListItem):
"""ListItem personnalisé pour afficher un terme et sa définition"""
def __init__(self, term: str, definition: str, expanded: bool = False) -> None:
self.term = term
self.definition = definition
self.expanded = expanded
# Créer le contenu avec le terme et éventuellement la définition
content = self._create_content()
super().__init__(content)
def _create_content(self):
"""Crée le contenu du ListItem avec ou sans définition"""
if self.expanded:
return Vertical(
Label(f"📌 {self.term}", classes="term-title"),
Label(f" {self.definition}", classes="term-definition"),
classes="term-expanded"
)
else:
return Label(f"📌 {self.term}", classes="term-title")
def toggle_expand(self):
"""Bascule l'état d'expansion du terme"""
self.expanded = not self.expanded
# Mettre à jour le contenu
self._update_content()
def _update_content(self):
"""Met à jour le contenu du ListItem"""
new_content = self._create_content()
# Remplacer l'ancien contenu
self.remove_children()
self.mount(new_content)
class LLMQueryWidget(Static):
"""Widget pour afficher les requêtes et réponses du LLM"""
class ResponseReceived(Message):
"""Message envoyé quand une réponse complète est reçue"""
def __init__(self, question: str, response: str) -> None:
self.question = question
self.response = response
super().__init__()
def __init__(self):
super().__init__()
self.question = ""
self.response = ""
self.is_streaming = False
def start_streaming(self, question: str):
"""Démarre l'affichage d'une réponse en streaming"""
self.question = question
self.response = ""
self.is_streaming = True
self.update_content()
def add_chunk(self, chunk: str):
"""Ajoute un chunk à la réponse"""
self.response += chunk
self.update_content()
def finish_streaming(self):
"""Termine le streaming"""
self.is_streaming = False
self.update_content()
def update_content(self):
"""Met à jour le contenu affiché"""
content = f"[bold cyan]❓ {self.question}[/bold cyan]\n\n"
if self.is_streaming:
content += f"[yellow]⏳ Génération en cours...[/yellow]\n\n"
content += f"[white]{self.response}[/white]"
if self.is_streaming and self.response:
content += " [yellow]▌[/yellow]" # Curseur clignotant simulé
self.update(content)
class DictionaryApp(App):
"""Application de dictionnaire avec intégration LLM Ollama"""
CSS = """
Screen {
background: #1e1e2e;
}
#main-container {
height: 100%;
width: 100%;
padding: 1;
background: #1e1e2e;
}
#search-container {
height: auto;
min-height: 5;
max-height: 8;
margin-bottom: 1;
background: #2d2d44;
padding: 1;
}
#results-container {
height: 70%;
background: #2d2d44;
padding: 1;
overflow-y: auto;
}
#llm-container {
height: 30%;
min-height: 15;
background: #2d2d44;
padding: 1;
margin-top: 1;
border: solid #585b70;
overflow-y: auto;
}
#search-label {
color: #89b4fa;
text-style: bold;
padding-bottom: 1;
width: 100%;
}
#llm-label {
color: #a6e3a1;
text-style: bold;
padding-bottom: 1;
width: 100%;
}
Input {
background: #3d3d5c;
color: #cdd6f4;
border: solid #585b70;
padding: 1;
width: 100%;
margin-top: 1;
}
Input:focus {
border: solid #89b4fa;
}
#llm-input {
background: #3d3d5c;
color: #cdd6f4;
border: solid #585b70;
padding: 1;
width: 100%;
margin-top: 1;
}
#llm-input:focus {
border: solid #a6e3a1;
}
#results-list {
background: #3d3d5c;
color: #cdd6f4;
height: 100%;
min-height: 10;
border: solid #585b70;
padding: 0;
}
ListItem {
padding: 1;
background: #3d3d5c;
color: #cdd6f4;
width: 100%;
border-bottom: solid #45476a;
}
ListItem:hover {
background: #45476a;
}
ListItem:focus {
background: #45476a;
}
.term-title {
color: #89b4fa;
text-style: bold;
width: 100%;
padding: 1;
}
.term-definition {
color: #cdd6f4;
padding: 1;
padding-left: 2;
background: #3d3d5c;
width: 100%;
border-left: solid #89b4fa;
margin-top: 1;
}
.term-expanded {
background: #2d2d44;
padding: 0;
}
#results-label {
color: #a6e3a1;
padding-bottom: 1;
width: 100%;
}
#preview-container {
background: #2d2d44;
padding: 1;
margin-top: 1;
min-height: 3;
max-height: 5;
border: solid #585b70;
color: #cdd6f4;
}
#preview-label {
color: #89b4fa;
text-style: bold;
padding-bottom: 1;
width: 100%;
}
#preview-content {
color: #f9e2af;
padding: 1;
background: #3d3d5c;
min-height: 2;
width: 100%;
}
#llm-response {
background: #3d3d5c;
padding: 1;
min-height: 10;
max-height: 20;
overflow-y: auto;
color: #cdd6f4;
margin-top: 1;
}
#llm-actions {
height: auto;
margin-top: 1;
}
Button {
background: #45476a;
color: #cdd6f4;
border: solid #585b70;
padding: 1;
width: auto;
}
Button:hover {
background: #585b70;
}
Button:focus {
border: solid #a6e3a1;
}
#clear-llm-btn {
background: #6b3b3b;
color: #f9e2af;
}
#clear-llm-btn:hover {
background: #8b4b4b;
}
"""
def __init__(self):
super().__init__()
# Dictionnaire de termes avec leurs définitions
self.dictionary = {
"Python": "Langage de programmation interprété, orienté objet, avec une syntaxe claire et une grande lisibilité",
"Textual": "Framework Python pour créer des interfaces utilisateur en mode texte avancées",
"Algorithm": "Suite d'instructions pour résoudre un problème ou effectuer une tâche spécifique",
"API": "Interface de programmation d'application, ensemble de règles pour interagir avec un logiciel",
"Framework": "Ensemble d'outils et de bibliothèques pour développer des applications structurées",
"Machine Learning": "Domaine de l'IA permettant aux machines d'apprendre à partir de données",
"Deep Learning": "Sous-ensemble du machine learning utilisant des réseaux de neurones profonds",
"Neural Network": "Système informatique inspiré du cerveau biologique et de ses connexions",
"Data Science": "Domaine interdisciplinaire pour extraire des connaissances des données",
"Cloud Computing": "Fourniture de services informatiques via internet à la demande",
"DevOps": "Pratique combinant développement et opérations informatiques en continu",
"Agile": "Méthodologie de gestion de projet itérative et flexible centrée sur l'humain",
"Scrum": "Framework Agile pour la gestion de projets complexes et adaptatifs",
"Kubernetes": "Plateforme d'orchestration de conteneurs open-source pour la production",
"Docker": "Plateforme de conteneurisation pour applications distribuées",
"Git": "Système de contrôle de version distribué pour le suivi de code",
"GitHub": "Plateforme d'hébergement de code basée sur Git et de collaboration",
"VS Code": "Éditeur de code source développé par Microsoft avec extensions",
"PyCharm": "IDE Python développé par JetBrains avec intégration complète",
"Jupyter": "Application web pour créer des notebooks interactifs en direct",
}
# Liste triée des termes pour la recherche
self.terms = sorted(self.dictionary.keys())
self.filtered_terms = self.terms.copy()
self.current_input = ""
self.last_selected_term = None
self.expanded_terms = set() # Ensemble des termes actuellement expansés
# Configuration Ollama
self.ollama_url = "http://localhost:11434/api/generate"
self.ollama_model = "llama3.2" # Ou un autre modèle installé
self.http_client = httpx.AsyncClient(timeout=120.0)
def compose(self) -> ComposeResult:
"""Compose l'interface utilisateur"""
yield Header()
yield Container(
Container(
# Zone de recherche
Container(
Label("🔍 Rechercher un terme :", id="search-label"),
Input(placeholder="Tapez un terme...", id="search-input"),
id="search-container"
),
# Zone d'aperçu du texte en cours
Container(
Label("✏️ Texte saisi :", id="preview-label"),
Static("En attente de saisie...", id="preview-content"),
id="preview-container"
),
# Résultats
Container(
Label("📋 0 terme trouvé", id="results-label"),
ListView(id="results-list"),
id="results-container"
),
# Zone LLM
Container(
Label("🤖 Question pour Ollama :", id="llm-label"),
Input(placeholder="Posez une question au LLM...", id="llm-input"),
Horizontal(
Button("Envoyer", id="send-llm-btn", variant="primary"),
Button("Effacer", id="clear-llm-btn"),
id="llm-actions"
),
Static("", id="llm-response"),
id="llm-container"
),
id="main-container"
),
)
yield Footer()
def on_mount(self) -> None:
"""Initialisation après le montage"""
search_input = self.query_one("#search-input")
search_input.focus()
self.update_results()
self.apply_responsive_styles()
def on_resize(self, event: events.Resize) -> None:
"""Gère le redimensionnement du terminal"""
self.apply_responsive_styles()
self.update_list_height()
def apply_responsive_styles(self) -> None:
"""Applique les styles responsifs en fonction de la taille du terminal"""
width = self.size.width
height = self.size.height
# Ajuster les paddings selon la largeur
main_container = self.query_one("#main-container")
search_container = self.query_one("#search-container")
results_container = self.query_one("#results-container")
preview_container = self.query_one("#preview-container")
llm_container = self.query_one("#llm-container")
llm_response = self.query_one("#llm-response")
if width < 80:
# Petit terminal
main_container.styles.padding = 1
search_container.styles.padding = 1
results_container.styles.padding = 1
preview_container.styles.padding = 1
llm_container.styles.padding = 1
preview_container.styles.min_height = 2
preview_container.styles.max_height = 3
search_container.styles.min_height = 4
search_container.styles.max_height = 6
llm_response.styles.min_height = 6
llm_response.styles.max_height = 12
elif width > 120 and height > 40:
# Grand terminal
main_container.styles.padding = 2
search_container.styles.padding = 2
results_container.styles.padding = 2
preview_container.styles.padding = 2
llm_container.styles.padding = 2
preview_container.styles.min_height = 4
preview_container.styles.max_height = 6
search_container.styles.min_height = 6
search_container.styles.max_height = 10
llm_response.styles.min_height = 12
llm_response.styles.max_height = 20
else:
# Terminal moyen (par défaut)
main_container.styles.padding = 1
search_container.styles.padding = 1
results_container.styles.padding = 1
preview_container.styles.padding = 1
llm_container.styles.padding = 1
preview_container.styles.min_height = 3
preview_container.styles.max_height = 5
search_container.styles.min_height = 5
search_container.styles.max_height = 8
llm_response.styles.min_height = 8
llm_response.styles.max_height = 15
# Ajuster la hauteur de la preview selon la hauteur du terminal
if height < 30:
preview_container.styles.min_height = 2
preview_container.styles.max_height = 3
llm_response.styles.min_height = 4
llm_response.styles.max_height = 8
elif height > 50:
preview_container.styles.min_height = 4
preview_container.styles.max_height = 6
llm_response.styles.min_height = 12
llm_response.styles.max_height = 20
def update_list_height(self) -> None:
"""Met à jour la hauteur de la liste en fonction de l'espace disponible"""
try:
results_list = self.query_one("#results-list")
available_height = self.size.height - self.calculate_fixed_height()
# Ajuster la hauteur de la liste
if available_height > 15:
results_list.styles.height = available_height - 2
elif available_height > 10:
results_list.styles.height = available_height
else:
results_list.styles.height = 8
except Exception:
pass
def calculate_fixed_height(self) -> int:
"""Calcule la hauteur fixe des éléments (header, footer, search, preview)"""
try:
search_container = self.query_one("#search-container")
preview_container = self.query_one("#preview-container")
# Récupérer les hauteurs actuelles
search_height = search_container.styles.min_height or 5
preview_height = preview_container.styles.min_height or 3
# Estimer la hauteur fixe totale
if isinstance(search_height, (int, float)):
search_h = search_height
else:
search_h = 5
if isinstance(preview_height, (int, float)):
preview_h = preview_height
else:
preview_h = 3
# Header + Footer + margins + padding
fixed_height = 2 + search_h + preview_h + 4
return int(fixed_height)
except Exception:
return 18 # Valeur par défaut en cas d'erreur
def on_input_changed(self, event: Input.Changed) -> None:
"""Gère les changements dans le champ de recherche"""
if event.input.id == "search-input":
self.current_input = event.value
self.update_preview()
self.update_results()
def update_preview(self) -> None:
"""Met à jour l'aperçu du texte saisi"""
preview_content = self.query_one("#preview-content")
if self.current_input:
preview_text = f"📝 {self.current_input}"
# Si le texte correspond à un terme connu
matching_terms = [term for term in self.terms if term.lower() == self.current_input.lower()]
if matching_terms:
preview_text = f"✅ {self.current_input} (terme trouvé !)"
preview_content.update(preview_text)
else:
preview_content.update("En attente de saisie...")
def on_input_submitted(self, event: Input.Submitted) -> None:
"""Gère la soumission du champ de recherche (touche Entrée)"""
if event.input.id == "search-input":
# Sélectionner le premier résultat
results_list = self.query_one("#results-list")
if results_list.children:
first_item = results_list.children[0]
if isinstance(first_item, TermListItem):
self.toggle_term_expansion(first_item.term)
elif event.input.id == "llm-input":
# Soumettre la question au LLM
self.query_one("#send-llm-btn").press()
def update_results(self) -> None:
"""Met à jour la liste des résultats"""
search_text = self.current_input.lower().strip()
# Filtrer les termes
if search_text:
self.filtered_terms = [
term for term in self.terms
if search_text in term.lower()
]
else:
self.filtered_terms = self.terms.copy()
# Mettre à jour la ListView
results_list = self.query_one("#results-list")
results_list.clear()
for term in self.filtered_terms:
definition = self.dictionary[term]
# Vérifier si le terme est expansé
expanded = term in self.expanded_terms
item = TermListItem(term, definition, expanded)
results_list.append(item)
# Mettre à jour le compteur
results_label = self.query_one("#results-label")
count = len(self.filtered_terms)
if count == 0:
results_label.update("❌ Aucun résultat trouvé")
else:
results_label.update(f"📋 {count} terme{'s' if count > 1 else ''} trouvé{'s' if count > 1 else ''}")
def on_list_view_selected(self, event: ListView.Selected) -> None:
"""Gère la sélection d'un terme dans la liste"""
if event.item and isinstance(event.item, TermListItem):
# Basculer l'expansion du terme sélectionné
self.toggle_term_expansion(event.item.term)
def toggle_term_expansion(self, term: str) -> None:
"""Bascule l'état d'expansion d'un terme"""
if term in self.expanded_terms:
self.expanded_terms.remove(term)
else:
# Si un autre terme est expansé, le réduire
if len(self.expanded_terms) > 0:
# Réduire tous les autres termes
for other_term in list(self.expanded_terms):
if other_term != term:
self.expanded_terms.remove(other_term)
self.expanded_terms.add(term)
# Mettre à jour l'affichage
self.update_results()
# Mettre à jour le champ de recherche
search_input = self.query_one("#search-input")
search_input.value = term
self.current_input = term
self.update_preview()
def on_key(self, event: events.Key) -> None:
"""Gère les touches spéciales"""
if event.key == "escape":
# Effacer la recherche
search_input = self.query_one("#search-input")
search_input.value = ""
self.current_input = ""
self.expanded_terms.clear()
self.update_preview()
self.update_results()
search_input.focus()
elif event.key == "ctrl+f":
# Focus sur la recherche
self.query_one("#search-input").focus()
elif event.key == "ctrl+l":
# Focus sur le champ LLM
self.query_one("#llm-input").focus()
elif event.key == "enter":
# Si un item est sélectionné, basculer son expansion
results_list = self.query_one("#results-list")
if results_list.children and hasattr(results_list, 'highlighted_child'):
highlighted = results_list.highlighted_child
if highlighted and isinstance(highlighted, TermListItem):
self.toggle_term_expansion(highlighted.term)
@work(thread=True)
def send_to_ollama(self, question: str) -> None:
"""Envoie une question à Ollama et affiche la réponse en streaming"""
response_widget = self.query_one("#llm-response")
# Afficher la question
response_widget.update(f"[bold cyan]❓ {question}[/bold cyan]\n\n[yellow]⏳ Génération en cours...[/yellow]")
# Préparer la requête
payload = {
"model": self.ollama_model,
"prompt": question,
"stream": True
}
full_response = ""
try:
# Faire la requête HTTP avec streaming
with httpx.stream(
"POST",
self.ollama_url,
json=payload,
timeout=120.0
) as response:
if response.status_code != 200:
error_msg = f"❌ Erreur HTTP {response.status_code}: {response.text}"
self.call_from_thread(response_widget.update, error_msg)
return
# Lire le stream
for line in response.iter_lines():
if line:
try:
data = json.loads(line)
if "response" in data:
chunk = data["response"]
full_response += chunk
# Mettre à jour l'affichage depuis le thread principal
display_text = f"[bold cyan]❓ {question}[/bold cyan]\n\n{full_response}[yellow]▌[/yellow]"
self.call_from_thread(response_widget.update, display_text)
if data.get("done", False):
break
except json.JSONDecodeError:
continue
# Afficher la réponse complète
final_display = f"[bold cyan]❓ {question}[/bold cyan]\n\n{full_response}"
self.call_from_thread(response_widget.update, final_display)
except httpx.TimeoutException:
error_msg = "❌ Erreur : Timeout - Ollama ne répond pas"
self.call_from_thread(response_widget.update, error_msg)
except httpx.ConnectError:
error_msg = "❌ Erreur : Impossible de se connecter à Ollama sur localhost:11434"
self.call_from_thread(response_widget.update, error_msg)
except Exception as e:
error_msg = f"❌ Erreur : {str(e)}"
self.call_from_thread(response_widget.update, error_msg)
def on_button_pressed(self, event: Button.Pressed) -> None:
"""Gère les pressions de boutons"""
if event.button.id == "send-llm-btn":
llm_input = self.query_one("#llm-input")
question = llm_input.value.strip()
if not question:
response_widget = self.query_one("#llm-response")
response_widget.update("❌ Veuillez entrer une question.")
return
# Vérifier la connexion à Ollama
# Désactiver le bouton pendant le traitement
send_btn = self.query_one("#send-llm-btn")
send_btn.disabled = True
send_btn.label = "⏳ En cours..."
# Lancer la requête
self.send_to_ollama(question)
# Réactiver le bouton après un court délai
# Le bouton sera réactivé dans la méthode de réponse
elif event.button.id == "clear-llm-btn":
response_widget = self.query_one("#llm-response")
response_widget.update("")
llm_input = self.query_one("#llm-input")
llm_input.value = ""
llm_input.focus()
def after_send_to_ollama(self) -> None:
"""Réactive le bouton d'envoi après la réponse"""
send_btn = self.query_one("#send-llm-btn")
send_btn.disabled = False
send_btn.label = "Envoyer"
# Point d'entrée de l'application
if __name__ == "__main__":
app = DictionaryApp()
app.run()
```
## Principales modifications apportées :
### 1. **Ajout de la zone LLM**
- Un nouveau conteneur `#llm-container` avec un champ de saisie `#llm-input`
- Un bouton "Envoyer" pour soumettre la question
- Une zone `#llm-response` pour afficher la réponse en streaming
### 2. **Fonction de streaming asynchrone avec `@work(thread=True)`**
```python
@work(thread=True)
def send_to_ollama(self, question: str) -> None:
# Utilise httpx.stream pour lire les chunks
with httpx.stream(...) as response:
for line in response.iter_lines():
# Traite chaque chunk
self.call_from_thread(response_widget.update, display_text)
```
### 3. **Mise à jour en temps réel**
- Utilise `self.call_from_thread()` pour mettre à jour l'interface depuis le thread
- Affiche un curseur clignotant `▌` pendant le streaming
- La réponse s'affiche progressivement
### 4. **Gestion des erreurs**
- Timeout
- Connexion refusée
- Erreurs HTTP
### 5. **Raccourcis clavier**
- `Ctrl+L` : Focus sur le champ LLM
- `Entrée` dans le champ LLM : Soumet la question
## Installation des dépendances :
```bash
pip install textual httpx
```
## Configuration :
- Vérifiez que Ollama est installé et en cours d'exécution : `ollama serve`
- Vérifiez que le modèle `llama3.2` est disponible : `ollama pull llama3.2`
- Vous pouvez changer le modèle dans `self.ollama_model`
## Utilisation :
1. Lancez l'application
2. Utilisez la zone LLM en bas pour poser des questions
3. Les réponses s'affichent en streaming en temps réel
4. Vous pouvez toujours utiliser la recherche de dictionnaire en parallèle
L'interface reste parfaitement réactive pendant le streaming grâce à l'utilisation du thread séparé avec `@work(thread=True)`.
Reference in a new issue Copy permalink