import os import json import time from typing import Any, Dict, List, Optional, Tuple import requests import streamlit as st from openai import OpenAI # ---------------------------- # Helpers # ---------------------------- def get_headers(api_key: str) -> Dict[str, str]: return {"Content-Type": "application/json", "Authorization": f"Bearer {api_key}"} def pretty(obj: Any) -> str: return json.dumps(obj, ensure_ascii=False, indent=2) def safe_get_env(name: str, default: str = "") -> str: v = os.getenv(name) return v if v is not None else default def make_client(base_url: str, api_key: str) -> OpenAI: return OpenAI(base_url=base_url, api_key=api_key) def http_get_json(url: str, headers: Dict[str, str], timeout: int = 30) -> Dict[str, Any]: r = requests.get(url, headers=headers, timeout=timeout) r.raise_for_status() return r.json() def http_post_json(url: str, headers: Dict[str, str], payload: Dict[str, Any], timeout: int = 60) -> Dict[str, Any]: r = requests.post(url, headers=headers, json=payload, timeout=timeout) r.raise_for_status() return r.json() def provider_defaults() -> Dict[str, Dict[str, str]]: return { "OpenAI": { "base_url": "https://api.openai.com/v1", "env_key": "OPENAI_API_KEY", "notes": "Supporte Responses API (web_search, image) + Chat/Embeddings/Models.", }, "Groq": { "base_url": "https://api.groq.com/openai/v1", "env_key": "GROQ_API_KEY", "notes": "OpenAI-compatible pour Chat/Models/Embeddings (selon offre). Pas de web_search OpenAI.", }, "Ollama": { "base_url": "https://ollama.com/v1", "env_key": "OLLAMA_API_KEY", "notes": "OpenAI-compatible local. Models/Chat ok selon config. Embeddings selon modèles dispo.", }, "Albert (Etalab)": { "base_url": "https://albert.api.etalab.gouv.fr/v1", "env_key": "ALBERT_API_KEY", "notes": "OpenAI-compatible (selon endpoints activés). Pas de web_search OpenAI.", }, } def is_openai_provider(name: str, base_url: str) -> bool: return name.lower().startswith("openai") or "api.openai.com" in base_url def extract_model_ids(models_payload: Any) -> List[str]: """ Normalise la sortie /models (ou SDK models.list) en liste d'IDs. """ if models_payload is None: return [] if isinstance(models_payload, dict) and isinstance(models_payload.get("data"), list): ids = [] for m in models_payload["data"]: if isinstance(m, dict) and "id" in m: ids.append(m["id"]) return sorted(list(dict.fromkeys(ids))) # parfois payload déjà sous forme list if isinstance(models_payload, list): ids = [] for m in models_payload: if isinstance(m, dict) and "id" in m: ids.append(m["id"]) return sorted(list(dict.fromkeys(ids))) return [] def pick_default(ids: List[str], preferred: List[str]) -> int: """ Renvoie l'index d'un modèle préféré s'il existe, sinon 0. """ lower = {m.lower(): i for i, m in enumerate(ids)} for p in preferred: if p.lower() in lower: return lower[p.lower()] return 0 # ---------------------------- # Streamlit UI # ---------------------------- st.set_page_config(page_title="LLM API Playground (pédagogique)", layout="wide") st.title("🧪 Mini-app de requêtes API sur LLMs") st.caption("Choisir un provider → on charge `/models` → chaque onglet propose un menu déroulant des modèles.") with st.sidebar: st.header("⚙️ Configuration") defaults = provider_defaults() provider_name = st.selectbox("Fournisseur", list(defaults.keys()), index=0) base_url = st.text_input("Base URL", value=defaults[provider_name]["base_url"]) env_key = defaults[provider_name]["env_key"] st.write(f"Variable d’environnement attendue : `{env_key}`") api_key = st.text_input( "API key (optionnel si déjà dans l'env)", value="", type="password", help=f"Laisse vide si tu as déjà exporté {env_key} dans ton environnement.", ) if not api_key: api_key = safe_get_env(env_key, "") st.markdown("---") st.info(defaults[provider_name]["notes"]) st.markdown("---") show_raw = st.toggle("Afficher requête/réponse brutes", value=True) timeout_s = st.slider("Timeout HTTP (s)", 10, 120, 45, 5) if not api_key and provider_name != "Ollama": st.warning(f"Pas de clé détectée. Renseigne l’API key dans la sidebar ou exporte `{env_key}`.") client = make_client(base_url=base_url, api_key=api_key if api_key else "NO_KEY") # ---------------------------- # Load models once per provider/base_url/api_key (cache) # ---------------------------- @st.cache_data(show_spinner=False, ttl=300) def load_models_cached(base_url: str, api_key: str, timeout_s: int) -> Tuple[List[str], Optional[Dict[str, Any]], Optional[str]]: """ Retourne (ids, raw_payload, error_msg). On tente d'abord via HTTP GET /models (le plus universel). """ try: models_url = f"{base_url}/models" raw = http_get_json(models_url, headers=get_headers(api_key), timeout=timeout_s) ids = extract_model_ids(raw) if not ids: return [], raw, "Réponse /models reçue, mais aucun `id` exploitable n'a été trouvé." return ids, raw, None except Exception as e: return [], None, str(e) with st.spinner("Chargement des modèles du provider…"): model_ids, models_raw, models_err = load_models_cached(base_url, api_key, timeout_s) if models_err: st.sidebar.warning(f"Impossible de charger /models : {models_err}") elif model_ids: st.sidebar.success(f"Modèles chargés : {len(model_ids)}") else: st.sidebar.warning("Aucun modèle détecté (structure inattendue).") if show_raw and models_raw: with st.expander("Debug: réponse brute /models"): st.code(pretty(models_raw), language="json") def model_selector( label: str, ids: List[str], preferred: List[str], key: str, fallback_value: str, ) -> str: """ Retourne un model_id : - si ids dispo: selectbox - sinon: text_input fallback """ if ids: idx = pick_default(ids, preferred) return st.selectbox(label, ids, index=idx, key=key) return st.text_input(label, value=fallback_value, key=key + "_fallback") tabs = st.tabs( [ "1) Lister les modèles", "2) Embeddings", "3) Chat completion", "4) Extraction JSON", "5) OpenAI web_search", "6) OpenAI image → description", ] ) # ---------------------------- # 1) Models # ---------------------------- with tabs[0]: st.subheader("1) Lister les modèles disponibles (`GET /models`)") colA, colB = st.columns([1, 1], vertical_alignment="top") with colA: if st.button("🔄 Recharger /models", type="primary"): load_models_cached.clear() st.rerun() st.write("IDs détectés :") if model_ids: st.dataframe({"model_id": model_ids}) else: st.info("Pas de liste exploitable. (Voir la réponse brute dans la sidebar si activée.)") with colB: st.write("À retenir") st.markdown( "- Le dropdown des autres onglets dépend de cette liste.\n" "- Si `/models` est bloqué par un provider, l’app retombe sur un champ texte." ) # ---------------------------- # 2) Embeddings # ---------------------------- with tabs[1]: st.subheader("2) Embeddings (`POST /embeddings`)") colA, colB = st.columns([1, 1], vertical_alignment="top") with colA: emb_model = model_selector( "Modèle d'embeddings", model_ids, preferred=["text-embedding-3-small", "text-embedding-ada-002"], key="emb_model", fallback_value="text-embedding-3-small", ) text = st.text_area("Texte à embedder", value="This is a test", height=120) if st.button("🧬 Calculer embeddings", type="primary"): try: emb_url = f"{base_url}/embeddings" payload = {"model": emb_model, "input": text} t0 = time.time() resp = http_post_json(emb_url, headers=get_headers(api_key), payload=payload, timeout=timeout_s) dt = time.time() - t0 st.success(f"OK — {dt:.2f}s") try: emb = resp["data"][0]["embedding"] st.write(f"Dimension : **{len(emb)}**") except Exception: st.info("Impossible d'extraire `data[0].embedding` (structure différente).") usage = resp.get("usage", {}) if usage: st.write(f"Usage : `{pretty(usage)}`") if show_raw: st.code(pretty(resp), language="json") except Exception as e: st.error(f"Erreur: {e}") # ---------------------------- # 3) Chat completion # ---------------------------- with tabs[2]: st.subheader("3) Chat completion (`POST /chat/completions`)") colA, colB = st.columns([1, 1], vertical_alignment="top") with colA: chat_model = model_selector( "Modèle (chat)", model_ids, preferred=["gpt-4o-mini", "gpt-4.1-mini", "gpt-4", "llama", "mixtral"], key="chat_model", fallback_value="gpt-4o-mini", ) prompt = st.text_area( "Prompt utilisateur", value="Explique la différence entre modèles de langage encodeur et modèle de langage décodeur.", height=140, ) max_tokens = st.slider("max_completion_tokens", 32, 512, 200, 16) temperature = st.slider("temperature", 0.0, 1.5, 0.3, 0.1) if st.button("💬 Générer", type="primary"): try: completion_url = f"{base_url}/chat/completions" payload = { "model": chat_model, "messages": [{"role": "user", "content": prompt}], "max_completion_tokens": max_tokens, "temperature": temperature, "stream": False, } t0 = time.time() resp = http_post_json(completion_url, headers=get_headers(api_key), payload=payload, timeout=timeout_s) dt = time.time() - t0 st.success(f"OK — {dt:.2f}s") try: content = resp["choices"][0]["message"]["content"] st.markdown("### Réponse") st.write(content) except Exception: st.info("Structure de réponse inattendue. Regarde la réponse brute.") if show_raw: st.code(pretty(resp), language="json") except Exception as e: st.error(f"Erreur: {e}") # ---------------------------- # 4) JSON extraction # ---------------------------- with tabs[3]: st.subheader("4) Extraction structurée en JSON (`response_format`)") colA, colB = st.columns([1, 1], vertical_alignment="top") with colA: json_model = model_selector( "Modèle (extraction)", model_ids, preferred=["gpt-4o-mini", "gpt-4.1-mini"], key="json_model", fallback_value="gpt-4o-mini", ) system = st.text_input("System prompt", value="You are a data extractor") text = st.text_area( "Texte à analyser", value="Le 12 janvier 2023, Marie Curie a rencontré Albert Einstein à Paris.", height=120, ) temp = st.slider("temperature (extraction)", 0.0, 1.0, 0.1, 0.05) if st.button("🧾 Extraire en JSON", type="primary"): try: resp = client.chat.completions.create( model=json_model, messages=[ {"role": "system", "content": system}, { "role": "user", "content": f"Extract the places, persons and dates from the following text and respond in JSON format: {text}", }, ], temperature=temp, response_format={"type": "json_object"}, stream=False, ) content = resp.choices[0].message.content st.markdown("### JSON retourné") try: st.json(json.loads(content)) except Exception: st.code(content, language="json") if show_raw: raw = resp.model_dump() if hasattr(resp, "model_dump") else resp st.code(pretty(raw), language="json") except Exception as e: st.error(f"Erreur: {e}") st.info("Certains providers/modèles ne supportent pas `response_format`.") # ---------------------------- # 5) OpenAI web_search (Responses API) # ---------------------------- with tabs[4]: st.subheader("5) OpenAI — Tool `web_search` via Responses API") if not is_openai_provider(provider_name, base_url): st.warning("Cet onglet est conçu pour OpenAI (Responses API + tool web_search).") else: colA, colB = st.columns([1, 1], vertical_alignment="top") with colA: resp_model = model_selector( "Modèle (Responses)", model_ids, preferred=["gpt-5", "gpt-4.1", "gpt-4.1-mini"], key="resp_model", fallback_value="gpt-5", ) reasoning = st.selectbox("reasoning.effort", ["minimal", "low", "medium", "high"], index=3) verbosity = st.selectbox("text.verbosity", ["low", "medium", "high"], index=1) prompt = st.text_area( "Prompt", value=( "Trouve le ppn de l'autorité personne Albert Camus dans le référentiel français Idref." ), height=180, ) if st.button("Lancer web_search", type="primary"): try: resp = client.responses.create( model=resp_model, input=prompt, tools=[{"type": "web_search"}], reasoning={"effort": reasoning}, text={"verbosity": verbosity}, stream=False, ) st.markdown("### Réponse") st.write(resp.output_text if getattr(resp, "output_text", None) else "(pas de output_text)") if show_raw: raw = resp.model_dump() if hasattr(resp, "model_dump") else resp st.code(pretty(raw), language="json") except Exception as e: st.error(f"Erreur: {e}") with colB: st.write("À retenir") st.markdown("- Ici le modèle **appelle un outil**.\n- Ce tab reste OpenAI-only dans cette démo.") # ---------------------------- # 6) OpenAI image description # ---------------------------- with tabs[5]: st.subheader("6) OpenAI — Décrire une image (Responses API, multimodal)") if not is_openai_provider(provider_name, base_url): st.warning("Cet onglet est conçu pour OpenAI (Responses API + input_image).") else: colA, colB = st.columns([1, 1], vertical_alignment="top") with colA: img_model = model_selector( "Modèle (multimodal)", model_ids, preferred=["gpt-4.1-mini", "gpt-4o-mini"], key="img_model", fallback_value="gpt-4.1-mini", ) image_url = st.text_input( "Image URL", value="https://github.com/gegedenice/divers-files/raw/ca7c12ae2955a804b8a050c0f9ce77e2c0ef3aad/aude_edouard.jpg", ) instruction = st.text_area("Instruction", value="Décris précisément cette image. Réponds en français.", height=120) st.image(image_url, caption="Aperçu (si accessible)", use_container_width=True) if st.button("Décrire", type="primary"): try: resp = client.responses.create( model=img_model, input=[ { "role": "user", "content": [ {"type": "input_text", "text": instruction}, {"type": "input_image", "image_url": image_url}, ], } ], ) st.markdown("### Description") st.write(resp.output_text) if show_raw: raw = resp.model_dump() if hasattr(resp, "model_dump") else resp st.code(pretty(raw), language="json") except Exception as e: st.error(f"Erreur: {e}") st.markdown("---") st.caption("Astuce : si `/models` ne marche pas chez un provider, la fallback text_input permet quand même de tester.")