Utilisation des API LLM — OpenAI & Anthropic

pip install openai anthropic

import os
from openai import OpenAI
import anthropic
 
openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
anthropic_client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

response = openai_client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "Tu es un assistant technique."},
        {"role": "user", "content": "Explique le principe de l'injection de dépendances en Python."},
    ],
    temperature=0.7,
    max_tokens=1024,
)
 
print(response.choices[0].message.content)
# Accès au token usage :
print(response.usage.prompt_tokens, response.usage.completion_tokens)

response = anthropic_client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Explique le principe de l'injection de dépendances en Python."},
    ],
)
 
# Claude renvoie une liste de blocs (text, tool_use, ...)
print(response.content[0].text)
print(response.usage.input_tokens, response.usage.output_tokens)

import asyncio
from openai import AsyncOpenAI
 
async def main():
    client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Bonjour, donne-moi une recette de crêpes."}],
    )
    print(response.choices[0].message.content)
 
asyncio.run(main())

import asyncio
import anthropic
 
async def main():
    client = anthropic.AsyncAnthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
    async with client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": "Bonjour, donne-moi une recette de crêpes."}],
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)
    print()  # fin de ligne
 
asyncio.run(main())

stream = openai_client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Écris un poème sur les microservices."}],
    stream=True,
)
 
for chunk in stream:
    delta = chunk.choices[0].delta
    if delta.content:
        print(delta.content, end="", flush=True)
print()

with anthropic_client.messages.stream(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Écris un poème sur les microservices."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
print()

# Squelette OpenAI
response = openai_client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=tools,
    tool_choice="auto",
)
 
message = response.choices[0].message
if message.tool_calls:
    for tc in message.tool_calls:
        args = json.loads(tc.function.arguments)
        result = dispatch(tc.function.name, args)
    messages.append(message)
    messages.append({"role": "tool", "tool_call_id": tc.id, "content": json.dumps(result)})
    # rappeler le modèle avec le résultat en contexte

response = openai_client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Extrais : nom, âge, ville de Marie, 30 ans, Lyon."}],
    response_format={"type": "json_schema", "json_schema": {
        "name": "profil",
        "schema": {"type": "object", "properties": {
            "nom": {"type": "string"}, "age": {"type": "integer"}, "ville": {"type": "string"}
        }}
    }},
)
data = json.loads(response.choices[0].message.content)

import time
import openai
from openai import APIError, RateLimitError, APITimeoutError
 
def call_llm_with_retry(client, model, messages, max_retries=5, backoff=2.0, **kwargs):
    for attempt in range(max_retries):
        try:
            return client.chat.completions.create(
                model=model, messages=messages, **kwargs
            )
        except RateLimitError:
            wait = backoff * (2 ** attempt) + 0.5  # jitter
            print(f"Rate limit — attente de {wait:.1f}s (essai {attempt+1}/{max_retries})")
            time.sleep(wait)
        except APITimeoutError:
            wait = backoff * (attempt + 1)
            print(f"Timeout — attente de {wait:.1f}s")
            time.sleep(wait)
        except APIError as e:
            # 5xx = probable erreur serveur passagère
            if 500 <= e.status_code < 600:
                wait = backoff * (2 ** attempt)
                print(f"Erreur serveur ({e.status_code}) — attente de {wait:.1f}s")
                time.sleep(wait)
            else:
                # Erreur côté client (paramètres invalides) : pas de retry
                raise
    raise RuntimeError(f"Échec après {max_retries} tentatives pour {model}")

TOKEN_COST = {
    "gpt-4o":      {"input": 0.0025/1000,  "output": 0.01/1000},
    "gpt-4o-mini": {"input": 0.00015/1000,  "output": 0.0006/1000},
    "claude-sonnet-4-20250514": {"input": 3.0/1_000_000,  "output": 15.0/1_000_000},
}
 
def estimate_cost(response, model: str) -> float:
    usage = response.usage
    cost_table = TOKEN_COST[model]
    return usage.prompt_tokens * cost_table["input"] + usage.completion_tokens * cost_table["output"]
 
# Usage cumulé sur plusieurs appels
total_cost = 0.0
for resp in responses:
    total_cost += estimate_cost(resp, resp.model)
print(f"Coût total estimé : {total_cost:.4f} $")