eric/pmomusic
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

update the LLM Docker description

Browse Source
This commit is contained in:
Eric Coissac
2025-09-06 17:47:24 +02:00
parent 2a0d69e6b8
commit a0260b8520
6 changed files with 325 additions and 533 deletions
Download Patch File Download Diff File Expand all files Collapse all files

487
LLM/Docker/app.py
View File

@@ -7,9 +7,12 @@ import textwrap
import requests
import httpx
import socket
import json
import random
import traceback
from fastapi.responses import StreamingResponse
from fastapi import FastAPI, Query, HTTPException, Request, Body
from langchain_community.vectorstores import Chroma
from langchain_chroma import Chroma
from langchain_ollama import OllamaEmbeddings
from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
@@ -17,55 +20,42 @@ from ddgs import DDGS
from unstructured.cleaners.core import clean_extra_whitespace, clean_non_ascii_chars, replace_unicode_quotes
from datetime import datetime, timezone, timedelta
from pydantic import BaseModel, Field, PrivateAttr, ValidationError
from typing import Optional, List, Dict, Any
from typing import Optional, List, Dict, Any, Union, Literal
from collections import Counter
# --- Configuration via variables d'environnement ---
PERSIST_DIR = os.environ.get("CHROMA_PERSIST_DIR", "/chroma_db")
CACHE_DIR = os.environ.get("RESPONSE_CACHE_DIR", "/response_cache")
MODEL_NAME = os.environ.get("OLLAMA_MODEL", "llama3:13b")
EMBED_MODEL = os.environ.get("EMBED_MODEL", "mxbai-embed-large:latest")
GENERATE_MODEL = os.environ.get("GENERATE_MODEL", "deepseek-coder:6.7b-instruct")
CHAT_MODEL = os.environ.get("CHAT_MODEL", GENERATE_MODEL)
EMBED_MODEL = os.environ.get("EMBED_MODEL", "nomic-embed-text:latest")
SRC_PATH=os.environ.get("SRC_PATH", ".")
OLLAMA_HOST = os.environ.get("OLLAMA_HOST", "http://host.docker.internal:11434")
PROG_LANG = os.environ.get("PROG_LANG", "go")
try:
CHUNCK_SIZE = int(os.environ.get("CHUNCK_SIZE", "300"))
except (ValueError, TypeError):
CHUNCK_SIZE = 300
try:
CHUNCK_OVERLAP = int(os.environ.get("CHUNCK_OVERLAP", "50"))
except (ValueError, TypeError):
CHUNCK_OVERLAP = 50
try:
QUERY_TIMEOUT = int(os.environ.get("QUERY_TIMEOUT", "120"))
except (ValueError, TypeError):
QUERY_TIMEOUT = 120
# Configuration
OLLAMA_BASE_URL = "http://127.0.0.1:11434"
VECTORSTORE = None # Initialisé ailleurs
DDGS_SEARCH_ENABLED = True
os.makedirs(PERSIST_DIR, exist_ok=True)
os.makedirs(CACHE_DIR, exist_ok=True)
# Configuration de l'URL de base d'Ollama
def get_ollama_base_url():
"""Détermine dynamiquement l'URL d'Ollama"""
# 1. Vérifier la variable d'environnement
if "OLLAMA_BASE_URL" in os.environ:
return os.environ["OLLAMA_BASE_URL"]
# 2. Tester la connectivité locale
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.settimeout(1)
s.connect(("localhost", 11434))
return "http://localhost:11434"
except (socket.timeout, ConnectionRefusedError):
pass
# 3. Essayer l'adresse spéciale Docker
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.settimeout(1)
s.connect(("host.docker.internal", 11434))
return "http://host.docker.internal:11434"
except (socket.timeout, ConnectionRefusedError):
pass
# 4. Fallback pour les environnements cloud
return "http://127.0.0.1:11434"
OLLAMA_BASE_URL = get_ollama_base_url()
# --- Nettoyage du code ---
def clean_code_content(content: str) -> str:
@@ -131,35 +121,39 @@ app = FastAPI()
# --- Traitement des chemins ---
paths = SRC_PATH.split(":")
if not paths:
paths = ["."]
src_paths_directories = SRC_PATH.split(":")
if not src_paths_directories:
src_paths_directories = ["."]
# --- Initialisation ---
vectorstore = None
code_hash = ""
def build_vectorstore():
global vectorstore, code_hash, paths
print("🔹 Construction du vectorstore...", file=sys.stderr)
global vectorstore, code_hash, src_paths_directories
print("🔹 Construction du vectorstore ...", file=sys.stderr)
# Hash du code pour hot-reload
new_hash = hash_code_dir(paths)
new_hash = hash_code_dir(src_paths_directories)
if vectorstore and new_hash == code_hash:
print("🔹 Pas de changement dans /code, utilisation du vectorstore existant", file=sys.stderr)
return
code_hash = new_hash
print(f" 🔹 Programmation language: {PROG_LANG}", file=sys.stderr)
print(f" 🔹 Chunck size: {CHUNCK_SIZE}", file=sys.stderr)
print(f" 🔹 Chunck overlap: {CHUNCK_OVERLAP}", file=sys.stderr)
# Text splitter optimisé Go
go_splitter = RecursiveCharacterTextSplitter.from_language(
language="go",
chunk_size=800,
chunk_overlap=150 #,
#separators=["\n\n", "\nfunc ", "}\n\n", "\n//", "\n/*", "\t"]
code_splitter = RecursiveCharacterTextSplitter.from_language(
language=PROG_LANG,
chunk_size=CHUNCK_SIZE,
chunk_overlap=CHUNCK_OVERLAP,
keep_separator=True
)
all_docs = []
for path in paths:
for path in src_paths_directories:
abs_path = os.path.join("/code", path) if path != "." else "/code"
print(f" 🔹 Chargement du code Go depuis: {abs_path}", file=sys.stderr)
loader = DirectoryLoader(
@@ -176,32 +170,69 @@ def build_vectorstore():
all_docs.extend(loaded_docs)
print(f"🔹 {len(all_docs)} documents après chargement", file=sys.stderr)
splits = go_splitter.split_documents(all_docs)
print(f"🔹 {len(splits)} chunks créés", file=sys.stderr)
embedding = NomicEmbeddingsWrapper(model=EMBED_MODEL, base_url=OLLAMA_BASE_URL)
embedding = NomicEmbeddingsWrapper(model=EMBED_MODEL, base_url=OLLAMA_HOST)
chat_collection = Chroma.from_documents(
documents=all_docs,
embedding=embedding,
persist_directory=PERSIST_DIR,
collection_metadata={"hnsw:space": "cosine"},
collection_name="chat_context"
)
splits = code_splitter.split_documents(all_docs)
print(f"🔹 {len(splits)} chunks créés", file=sys.stderr)
splits = [doc for doc in splits if doc.page_content.strip()]
# Ajout des statistiques de longueur du split
bins=50
counter = Counter(int(len(split.page_content) / bins) * bins for split in splits)
print("🔹 Histogramme de la longueur des splits :", file=sys.stderr)
max_value = max(counter.values())
for length, count in sorted(counter.items()):
normalized_count = int((count / max_value) * 50)
print(f" {length+1:6}-{length+bins:-6}: { '#' * normalized_count}", file=sys.stderr)
print(f"🔹 {len(splits)} fragments non vides à intégrer", file=sys.stderr)
# Créer ou recharger Chroma
vectorstore = Chroma.from_documents(
documents=splits,
# 2. Vectorstore pour la Génération (splits courts)
gen_collection = Chroma.from_documents(
documents=splits, # Morceaux de 200-400 tokens
embedding=embedding,
collection_name="code_completion",
persist_directory=PERSIST_DIR,
collection_metadata={"hnsw:space": "cosine"}
)
vectorstore = {
"chat": chat_collection,
"generate": gen_collection
}
print("🔹 Vectorstore créé", file=sys.stderr)
# --- Formatage du contexte ---
def format_context(docs: list) -> str:
context = []
extraits = {}
print("🔹 Les fichiers suivants ont été selectionnés:", file=sys.stderr)
for i, doc in enumerate(docs):
source = doc.metadata.get('source', 'unknown')
filename = os.path.basename(source)
context.append(f"### Fichier: {filename} (Extrait {i+1}) ###")
context.append(textwrap.indent(doc.page_content, ' '))
print(f" 🔹 {filename} -- extrait {i+1} --", file=sys.stderr)
if doc.page_content not in extraits:
extraits[doc.page_content] = True
context.append(f"### Fichier: {filename} (Extrait {i+1}) ###")
context.append(textwrap.indent(doc.page_content, ' '))
print(f" 🔹 {filename} -- fin extrait {i+1} --", file=sys.stderr)
else:
print(f" 🔸 {filename} -- extrait {i+1} duppliqué et éliminé --", file=sys.stderr)
return "\n\n".join(context)
def format_iso_time_with_ns():
@@ -250,8 +281,8 @@ class ChatRequest(BaseModel):
messages: List[ChatMessage]
format: Optional[str] = None
options: Optional[Dict[str, Any]] = None
stream: bool = False
keep_alive: Optional[str] = None
stream: bool = True
keep_alive: Optional[Union[str, int]] = None # Modification ici
class EmbeddingRequest(BaseModel):
model: str
@@ -262,11 +293,11 @@ class EmbeddingResponse(BaseModel):
embedding: List[float]
# Fonctions utilitaires
async def perform_rag_search(prompt: str, k: int = 4) -> str:
async def perform_rag_search(mode: Literal["generate", "chat"], prompt: str, k: int = 4) -> str:
"""Effectue une recherche RAG et retourne le contexte"""
build_vectorstore()
rag_docs = vectorstore.similarity_search(prompt, k=k)
rag_docs = vectorstore[mode].similarity_search(prompt, k=k)
return format_context(rag_docs) if rag_docs else "Aucun contexte trouvé."
async def perform_web_search(prompt: str, k: int = 2) -> str:
@@ -275,171 +306,265 @@ async def perform_web_search(prompt: str, k: int = 2) -> str:
return "Recherche web désactivée"
try:
from duckduckgo_search import DDGS
from ddgs import DDGS
with DDGS() as ddgs:
results = list(ddgs.text(prompt, max_results=k))
web_info = "\n".join(f"- [{r['title']}]({r['href']}): {r['body'][:150]}..." for r in results) if results else "Aucun résultat web trouvé."
print(f"🔹 {len(results)} résultats trouvés sur le web", file=sys.stderr)
for i, r in enumerate(results):
print(f" - {i+1}. {r['title']} : {r['href']}", file=sys.stderr)
web_info = "\n".join(
f"- [{r['title']}]({r['href']}): {r['body'][:150]}..."
for r in results
) if results else "Aucun résultat web trouvé."
except Exception as e:
return f"Erreur recherche web: {str(e)}"
def build_enhanced_prompt(original_prompt: str, rag_context: str, web_context: str) -> str:
"""Construit un prompt enrichi avec les contextes"""
return f"""
### CONTEXTE RAG (Code) ###
{rag_context or "Aucun contexte code disponible"}
def build_enhanced_prompt(
mode: Literal["generate", "chat"],
question: str,
rag_context: str,
web_context: str):
if mode == "chat":
prompt = f"""
# Consigne
### CONTEXTE WEB ###
{web_context or "Aucune information web disponible"}
Vous êtes un expert en programmation {PROG_LANG}. Répondez à la question en utilisant le contexte fourni (extraits de code) et les informations web si disponibles.
Autant que possible tu indiqueras tes sources, url, nom du fichier source...
### QUESTION UTILISATEUR ###
{original_prompt}
# Contexte de la question:
## **Contexte Code (extraits pertinents):**
{rag_context}
## **Informations Web:**
{web_context}
## **Instructions:**
- Répondez de manière concise et précise à la question
- Si la réponse se trouve dans le contexte code, citez le fichier et l'extrait correspondant
- Si vous utilisez les informations web, citez la source
- Si la question est en anglais, répondez en anglais. Sinon, en français
- Pour les extraits de code, conservez le formatage et l'indentation
# **Question** à laquelle tu dois répondre
{question}
"""
else:
prompt=f"""
# Consigne
Vous êtes un expert en programmation {PROG_LANG}. Essayer de concevoir un petit bout de code permetant de résoudre la question
# Contexte de la question:
## **Contexte Code (extraits pertinents):**
{rag_context}
## **Instructions:**
- Rédigez les commentaires de code dans la même langue que le code qui vous est fourni. À défaut en anglais.
- Nommez les variables dans la même langue que le code qui vous est fourni. À défaut en anglais.
- Si la question est en anglais, répondez en anglais. Sinon, en français
- Pour les extraits de code, conservez le formatage et l'indentation
# **Question** à laquelle tu dois répondre
{question}
"""
return prompt
# Endpoints compatibles Ollama
# --- Fonctions utilitaires factorisées ---
async def build_augmented_prompt(
mode: Literal["generate", "chat"],
question: str
) -> str:
"""Construit un prompt enrichi avec contextes RAG et web"""
rag_context = await perform_rag_search(mode,question,k= 2 if mode=='chat' else 8)
if mode == "chat":
web_context = await perform_web_search(question)
else:
web_context = ""
return build_enhanced_prompt(
mode=mode,
question=question,
rag_context=rag_context,
web_context=web_context
)
async def _stream_ollama_response(response: httpx.Response, model_name: str):
"""Générateur pour le streaming de la réponse de chat au format Ollama"""
start_time = datetime.now(timezone.utc).isoformat()
async for line in response.aiter_lines():
if line.strip():
try:
chunk = json.loads(line)
# Construction du message conforme à l'API Ollama
message_chunk = {
"model": model_name,
"created_at": start_time,
"message": {
"role": "assistant",
"content": chunk.get("message", {}).get("content", "") if "message" in chunk else chunk.get("content", "")
},
"done": chunk.get("done", False)
}
# Ajout des champs optionnels
for field in ["total_duration", "load_duration", "prompt_eval_count", "eval_count"]:
if field in chunk:
message_chunk[field] = chunk[field]
yield f"data: {json.dumps(message_chunk)}\n\n"
except json.JSONDecodeError:
yield "event: error\ndata: Invalid JSON chunk\n\n"
yield "event: end\ndata: Stream completed\n\n"
# --- Endpoints API ---
@app.post("/api/generate")
async def generate_endpoint(request_data: GenerateRequest = Body(...)):
"""Endpoint pour la génération avec gestion du streaming"""
try:
# Récupération des contextes RAG et web
rag_context = await perform_rag_search(request_data.prompt)
web_context = await perform_web_search(request_data.prompt)
# Construction du prompt enrichi
enhanced_prompt = build_enhanced_prompt(
original_prompt=request_data.prompt,
rag_context=rag_context,
web_context=web_context
)
enhanced_prompt = await build_augmented_prompt("generate",request_data.prompt)
# Préparation du payload pour Ollama
# Appel à Ollama
ollama_payload = {
"model": request_data.model,
"model": GENERATE_MODEL,
"prompt": enhanced_prompt,
"stream": request_data.stream,
"options": request_data.options or {}
}
# Appel à Ollama
async with httpx.AsyncClient() as client:
response = await client.post(
f"{OLLAMA_BASE_URL}/api/generate",
f"{OLLAMA_HOST}/api/generate",
json=ollama_payload,
timeout=120.0
timeout=QUERY_TIMEOUT
)
response.raise_for_status()
# Gestion des réponses NON-STREAMING
if not request_data.stream:
result = response.json()
return {
"model": result["model"],
"response": result["response"],
"done": result["done"],
"context": result.get("context"),
"total_duration": result.get("total_duration")
}
# Gestion des réponses STREAMING
return response.json()
else:
async def generate():
# --- CORRECTION DU STREAMING ---
async def generate_stream():
"""Générateur pour le streaming des résultats"""
full_response = ""
async for line in response.aiter_lines():
if line.strip():
try:
chunk = json.loads(line)
# 1. Format SSE valide avec double newline
yield f"data: {json.dumps(chunk)}\n\n"
# 2. Accumuler la réponse complète pour les logs
full_response += chunk.get("response", "")
# 3. Envoyer périodiquement un keep-alive
if random.random() < 0.1: # 10% des chunks
yield ": keep-alive\n\n"
# 4. Fin du stream
if chunk.get("done", False):
break
except json.JSONDecodeError:
print(f"⚠️ Ligne JSON invalide: {line}")
yield f"event: error\ndata: Invalid JSON line\n\n"
# 5. Envoyer un message de fin explicite
yield "event: end\ndata: Stream completed\n\n"
# 6. Log de la réponse complète
print(f"🔹 Réponse complète ({len(full_response)} caractères): {full_response[:200]}...", file=sys.stderr)
# Log de la réponse complète (optionnel)
print(f"🔹 Réponse complète: {full_response}")
async for chunk in response.aiter_text():
# Transférer directement les chunks
yield chunk
# Fermeture propre du stream
# yield "data: [DONE]\n\n"
# 7. Configuration de la réponse avec des headers spécifiques
# Utilisez text/plain au lieu de text/event-stream
return StreamingResponse(
generate(),
media_type="text/event-stream",
generate_stream(),
media_type="text/plain",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no" # Important pour Nginx
"X-Accel-Buffering": "no"
}
)
)
except httpx.RequestError as e:
raise HTTPException(500, f"Erreur de connexion à Ollama: {str(e)}")
except json.JSONDecodeError as e:
raise HTTPException(500, f"Erreur de décodage JSON: {str(e)}")
except Exception as e:
raise HTTPException(500, f"Erreur interne: {str(e)}")
@app.post("/api/chat")
async def chat(request: ChatRequest):
"""Endpoint /api/chat avec enrichissement du dernier message"""
# Copie profonde des messages
processed_messages = [msg.dict() for msg in request.messages]
# Enrichissement uniquement du dernier message utilisateur
if processed_messages and processed_messages[-1]["role"] == "user":
last_msg = processed_messages[-1]["content"]
async def chat_endpoint(request_data: ChatRequest):
try:
messages = [msg.dict() for msg in request_data.messages]
rag_context = await perform_rag_search(last_msg)
web_context = await perform_web_search(last_msg)
if messages and messages[-1]["role"] == "user":
original_question = messages[-1]["content"]
try:
# Limiter la taille du contexte
augmented_prompt = await build_augmented_prompt("chat",original_question)
messages[-1]["content"] = augmented_prompt[-8000:] # Truncate to the last 8000 chars
print(f"🔹 Prompt enrichi ({len(augmented_prompt)} caractères)", file=sys.stderr)
except Exception as e:
print(f"⚠️ Erreur d'enrichissement: {str(e)}", file=sys.stderr)
messages[-1]["content"] = original_question # Fallback to original
enhanced_content = build_enhanced_prompt(
original_prompt=last_msg,
rag_context=rag_context,
web_context=web_context
)
# Préparation du payload pour Ollama
ollama_payload = {
"model": CHAT_MODEL,
"messages": messages,
"stream": request_data.stream,
"options": request_data.options or {}
}
if request_data.keep_alive is not None:
if isinstance(request_data.keep_alive, int):
ollama_payload["keep_alive"] = f"{request_data.keep_alive}s"
else:
ollama_payload["keep_alive"] = request_data.keep_alive
processed_messages[-1]["content"] = enhanced_content
# Appel au vrai Ollama
async with httpx.AsyncClient() as client:
try:
response = await client.post(
f"{OLLAMA_BASE_URL}/api/chat",
json={
"model": request.model,
"messages": processed_messages,
"format": request.format,
"options": request.options,
"stream": request.stream,
"keep_alive": request.keep_alive
},
timeout=120.0
)
response.raise_for_status()
if request.stream:
return response.iter_lines()
return response.json()
except httpx.RequestError as e:
raise HTTPException(500, f"Erreur de connexion à Ollama: {str(e)}")
sopload = json.dumps(ollama_payload, indent=2)
print(f" 🔹 Taille du Payload vers Ollama : {len(sopload)} octets...", file=sys.stderr)
print(f" 🔹 Début du payload : {sopload}...", file=sys.stderr)
# Appel à Ollama
async with httpx.AsyncClient() as client:
try:
response = await client.post(
f"{OLLAMA_HOST}/api/chat",
json=ollama_payload,
timeout=QUERY_TIMEOUT
)
response.raise_for_status()
# Gestion des réponses NON-STREAMING
if not request_data.stream:
return response.json()
# --- CORRECTION DU STREAMING ---
async def generate_stream():
"""Générateur pour le streaming des résultats"""
async for chunk in response.aiter_text():
# Transférer directement les chunks
yield chunk
# Fermeture propre du stream
# yield "data: [DONE]\n\n"
# Utilisez text/plain au lieu de text/event-stream
return StreamingResponse(
generate_stream(),
media_type="text/plain",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no"
}
)
# ... [gestion des erreurs existante] ...
except httpx.HTTPStatusError as e:
error_detail = e.response.text if e.response else str(e)
print(f"🚨 Erreur HTTP Ollama ({e.response.status_code}): {error_detail}", file=sys.stderr)
raise HTTPException(502, f"Erreur Ollama: {error_detail}")
except httpx.RequestError as e:
print(f"🚨 Erreur réseau Ollama: {str(e)}", file=sys.stderr)
raise HTTPException(503, f"Ollama non disponible: {str(e)}")
except Exception as e:
print(f"🚨 Erreur interne: {traceback.format_exc()}", file=sys.stderr)
raise HTTPException(500, f"Erreur interne: {str(e)}")
@app.post("/api/embeddings")
async def embeddings(request: EmbeddingRequest):
@@ -447,7 +572,7 @@ async def embeddings(request: EmbeddingRequest):
async with httpx.AsyncClient() as client:
try:
response = await client.post(
f"{OLLAMA_BASE_URL}/api/embeddings",
f"{OLLAMA_HOST}/api/embeddings",
json=request.dict()
)
response.raise_for_status()
@@ -461,7 +586,7 @@ async def list_models():
"""Proxy pour lister les modèles disponibles"""
async with httpx.AsyncClient() as client:
try:
response = await client.get(f"{OLLAMA_BASE_URL}/api/tags")
response = await client.get(f"{OLLAMA_HOST}/api/tags")
response.raise_for_status()
return response.json()