Normanthiia – Formation des formateurs NSI & SNT
Bruno Mermet – Licence CC BY-NC-SA – 09/26
Voir partie “Utiliser un LLM poids ouverts”
pip install langchain
pip install langchain--ollama
pip install langgraph
pip install langchain-mcp-adapters
pip install llama-index
pip install llama-index-readers-file
pip install llama-index-llms-ollama
pip install llama-index-embeddings-ollama
Installer également dans ollama ces 2 LLM :
nomic-embed-textqwen2.5:1.5bfrom langchain_ollama import ChatOllama
model = ChatOllama(
model="llama3.1",
temperature=0
)
messages = [
("system", "Tu es un professeur d'histoire-géographie."),
("human", "Donne la liste des présidents de la 5è république")
]
réponse = model.invoke(messages)
texte = réponse.content
temps = int(réponse.response_metadata['total_duration'] / 1E7)/100
tokens_en_entrée = réponse.usage_metadata['input_tokens']
tokens_en_sortie = réponse.usage_metadata['output_tokens']
print("réponse :", texte)
print(f"durée : {temps}s")
print("nombre de tokens en entrée :", tokens_en_entrée)
print("nombre de tokens en sortie :", tokens_en_sortie)from langchain_ollama import ChatOllama
from langchain.agents import create_agent
model = ChatOllama(
model="llama3.1",
temperature=0
)
agent = create_agent(
model=model,
system_prompt=("Tu es un professeur d'histoire géographie", "Tu aimes donner beaucoup de précisions")
)
res = agent.invoke({
"messages": [{
"role": "user",
"content": "Donne la liste des présidents de la 5è république"
}]
})
réponse = res["messages"][-1]
print(réponse)
texte = réponse.content
temps = int(réponse.response_metadata['total_duration'] / 1E7)/100
tokens_en_entrée = réponse.usage_metadata['input_tokens']
tokens_en_sortie = réponse.usage_metadata['output_tokens']
print("réponse :", texte)
print(f"durée : {temps}s")
print("nombre de tokens en entrée :", tokens_en_entrée)
print("nombre de tokens en sortie :", tokens_en_sortie)
from langchain.agents import create_agent
from langchain.tools import tool
from langchain_ollama import ChatOllama
# ─────────────────────────────────────────────
# Les outils disponibles pour l'agent
# ─────────────────────────────────────────────
@tool
def get_animal_préféré(nom_personne: str) -> str:
"""
Donne l'animal préféré d'une personne précisée par son nom.
@param nom_personne: un nom de personne
@return: l'animal préféré de la personne dont le nom a été passé en paramètre
"""
print("utilisation de get_animal_préféré pour ", nom_personne)
if "mermet" in nom_personne.lower():
retour = "âne"
else:
retour = "cheval"
print("valeur renvoyée :", retour)
return retour
@tool
def get_nom_formateur(formation: str) -> str:
"""
Donne le nom du responsable d'une formation.
@param formation: le nom de la formation
@return: le nom de la personne responsable de la formation
"""
# Simulation d'un moteur de recherche
print("utilisation de get_formateur avec", formation)
if "formateur" in formation.lower():
retour = "Bruno Mermet"
else:
retour = "Armelle Klein"
print("valeur renvoyée :", retour)
return retour
# ─────────────────────────────────────────────
# Le modèle
# ─────────────────────────────────────────────
model = ChatOllama(
model="qwen3:4b",
temperature=0
)
# ─────────────────────────────────────────────
# L'agent
# ─────────────────────────────────────────────
agent = create_agent(
model=model,
tools=[get_animal_préféré, get_nom_formateur],
# system_prompt="""
# Tu es un assistant.
# Tu peux utiliser les outils disponibles lorsque cela est nécessaire.
# Décide toi-même quel outil utiliser.
# """
system_prompt="""
Tu es un assistant.
Utilise prioritairement les outils disponibles en leur faisant confiance.
Tu peux décomposer le problème en sous-problèmes pour mieux le résoudre.
"""
)
# ─────────────────────────────────────────────
# Une question
# ─────────────────────────────────────────────
result = agent.invoke({
"messages": [
{
"role": "user",
"content": "Qui est le responsable de la formation 'formateur' ? Quel est son animal préféré ? Donne-également l'animal préféré du responsable de la formation 'SNT'."
}
]
})
print(result["messages"][-1].content)
Le prompt est passé à l’agent d’entrée La sortie est récupérée depuis l’agent de sortie
from typing import TypedDict
from langchain_ollama import ChatOllama
from langgraph.graph import StateGraph, START, END
# ─────────────────────────────────────────────
# Le modèle
# ─────────────────────────────────────────────
model = ChatOllama(
# model="llama3.1",
model="qwen3:4b",
temperature=0
)
# ─────────────────────────────────────────────
# L'état partagé par les agents
# ─────────────────────────────────────────────
class State(TypedDict):
question: str
informations: str
reponse: str
# ─────────────────────────────────────────────
# Agent chercheur
# ─────────────────────────────────────────────
def chercheur(state: State):
print("appel de chercheur,state=",state)
prompt = f"""
Tu es un agent chercheur.
Question :
{state["question"]}
Donne les informations importantes permettant
à un autre agent de répondre à cette question.
"""
result = model.invoke(prompt)
return {
"informations": result.content
}
# ─────────────────────────────────────────────
# Agent rédacteur
# ─────────────────────────────────────────────
def redacteur(state: State):
print("appel de redacteur, state=", state)
prompt = f"""
Tu es un agent rédacteur.
Question :
{state["question"]}
Informations fournies par le chercheur :
{state["informations"]}
Rédige une réponse claire et concise.
"""
result = model.invoke(prompt)
return {
"reponse": result.content
}
# ─────────────────────────────────────────────
# Construction du graphe
# ─────────────────────────────────────────────
graph = StateGraph(State)
graph.add_node("chercheur", chercheur)
graph.add_node("redacteur", redacteur)
graph.add_edge(START, "chercheur")
graph.add_edge("chercheur", "redacteur")
graph.add_edge("redacteur", END)
agent = graph.compile()
# ─────────────────────────────────────────────
# Exécution
# ─────────────────────────────────────────────
result = agent.invoke({
"question":
"Pourquoi Python est-il populaire en intelligence artificielle ?"
})
print("Modèle utilisé :", model)
print(result["reponse"])
Prompt ⟶ Chercheur ⟶ Rédacteur
🢑 │
│ │
│ 🢓
└───────Vérificateur
│
│
🢓
Réponse
from typing import TypedDict
from langchain_ollama import ChatOllama
from langgraph.graph import StateGraph, START, END
# ─────────────────────────────────────────────
# Le modèle
# ─────────────────────────────────────────────
model = ChatOllama(
# model="llama3.1",
model="qwen3:4b",
temperature=0
)
# ─────────────────────────────────────────────
# État partagé
# ─────────────────────────────────────────────
class State(TypedDict):
question: str
informations: str
reponse: str
verification: str
# ─────────────────────────────────────────────
# Agent chercheur
# ─────────────────────────────────────────────
def chercheur(state: State):
print("\n[CHERCHEUR]")
prompt = f"""
Tu es un agent chercheur.
Question :
{state["question"]}
Donne les informations importantes permettant
de répondre à cette question.
"""
if state.get("reponse", "") != "":
prompt += "Tu avais déjà proposé la réponse " + state["reponse"] + "."
prompt += "Mais celle-ci présentait un problème : " + state["informations"] + "."
prompt += "Ta réponse doit corriger ce problème."
print("PROMPT : ", prompt)
result = model.invoke(prompt)
print(result.content)
return {
"informations": result.content
}
# ─────────────────────────────────────────────
# Agent rédacteur
# ─────────────────────────────────────────────
def redacteur(state: State):
print("\n[REDACTEUR]")
prompt = f"""
Tu es un agent rédacteur.
Question :
{state["question"]}
Voici éventuellement une réponse précédente :
{state.get("reponse", "")}
Informations disponibles :
{state["informations"]}
Rédige une réponse correcte et concise.
Corrige les éventuelles erreurs de la réponse précédente.
"""
result = model.invoke(prompt)
print(result.content)
return {
"reponse": result.content
}
# ─────────────────────────────────────────────
# Agent vérificateur 2
# ─────────────────────────────────────────────
def verificateur2(state: State):
print("\n[VERIFICATEUR 2]")
if "java" not in state["reponse"].lower():
print("pb java")
return {
"verification": "CORRIGER : il n'y a pas de comparaison avec java"
}
if "efficacité" not in state["reponse"].lower():
print("pb efficacité")
return {
"verification": "CORRIGER : l'efficacité n'est pas abordée"
}
if "langchain" not in state["reponse"].lower():
print("pb langchain")
return {
"verification": "CORRIGER : l'existence de la bibliothèque langchain n'est pas abordée"
}
print("OK")
return {
"verification": "OK"
}
# ─────────────────────────────────────────────
# Décision : faut-il recommencer ?
# ─────────────────────────────────────────────
def decision(state: State):
if state["verification"].strip().startswith("OK"):
return "fin"
return "corriger"
# ─────────────────────────────────────────────
# Construction du graphe
# ─────────────────────────────────────────────
graph = StateGraph(State)
graph.add_node("chercheur", chercheur)
graph.add_node("redacteur", redacteur)
graph.add_node("verificateur", verificateur2)
graph.add_edge(START, "chercheur")
graph.add_edge("chercheur", "redacteur")
graph.add_edge("redacteur", "verificateur")
graph.add_conditional_edges(
"verificateur",
decision,
{
"fin": END,
"corriger": "chercheur"
}
)
agent = graph.compile()
# ─────────────────────────────────────────────
# Exécution
# ─────────────────────────────────────────────
result = agent.invoke({
"question":
"Pourquoi Python est-il populaire en intelligence artificielle ? Réponds en moins de 800 mots"
})
print("Modèle utilisé :", model)
print("\n==============================")
print("RÉPONSE FINALE")
print("==============================")
print(result["reponse"])
Dans un SMA à base d’agents IA, on peut souvent trouver différents types d’agents :
On utilise le MCP OpenAQ, permettant d’avoir des informations sur la qualité de l’air
import asyncio
from langchain.agents import create_agent
from langchain_ollama import ChatOllama
from langchain_mcp_adapters.client import MultiServerMCPClient
async def main():
# 1. Connexion au serveur MCP
client = MultiServerMCPClient({
"weather": {
"transport": "http",
"url": "https://openaq.caseyjhand.com/mcp",
}
})
# 2. Récupération des outils MCP
tools = await client.get_tools()
print("Outils disponibles :")
for tool in tools:
print("OUTIL :", tool.name)
print("DESCRIPTION :", tool.description)
print("SHEMA :", tool.args_schema)
print()
asyncio.run(main())Ici, c’est l’outil openaq_list_countries qui va être utilisé.
import asyncio
from langchain.agents import create_agent
from langchain_ollama import ChatOllama
from langchain_mcp_adapters.client import MultiServerMCPClient
async def main():
# 1. Connexion au serveur MCP
client = MultiServerMCPClient({
"weather": {
"transport": "http",
"url": "https://openaq.caseyjhand.com/mcp",
}
})
# 2. Récupération des outils MCP
tools = await client.get_tools()
# 3. Modèle
model = ChatOllama(
model="qwen3:4b",
#model="llama3.1:latest",
temperature=0
)
# 4. Agent LangChain
agent = create_agent(
model,
tools
)
# 5. Question
result = await agent.ainvoke({
"messages": [
{
"role": "user",
"content": "Donne la liste de stations donnant la qualité de l'air en France."
}
]
})
# 6. Réponse finale
print(result["messages"][-1].content)
asyncio.run(main())Ici, les outils openaq_list_countries et openaq_get_readings qui vont être utilisés.
python
import asyncio
from langchain.agents import create_agent
from langchain_ollama import ChatOllama
from langchain_mcp_adapters.client import MultiServerMCPClient
async def main():
# 1. Connexion au serveur MCP
client = MultiServerMCPClient({
"weather": {
"transport": "http",
"url": "https://openaq.caseyjhand.com/mcp",
}
})
# 2. Récupération des outils MCP
tools = await client.get_tools()
# 3. Modèle
model = ChatOllama(
model="qwen3:4b",
#model="llama3.1:latest",
temperature=0
)
# 4. Agent LangChain
agent = create_agent(
model,
tools
)
# 5. Question
result = await agent.ainvoke({
"messages": [
{
"role": "user",
"content": "Après avoir récupéré l'identifiant de la station située à Rouen en France, donne les dernières mesures des différents capteurs de cette station."
}
]
})
# 6. Réponse finale
print(result["messages"][-1].content)
asyncio.run(main())
Google collab est un notebook permettant d’avoir accès gratuitement à un GPU (si disponible).
Pour utiliser un GPU si disponible :
Exécution -> Modifier le type d’exécution -> GPU T4
!pip install colab-xterm
%load_ext colabxterm
%xterm
curl -fsSL https://ollama.com/install.sh | sh
ollama serve &
ollama pull llama3.2
N.B : il faudra certainement relancer régulièrement ollama
!ollama list
!pip install langgraph langchain-ollama
Exemple de programme permettant de :
Ce programme utilise :
llama-indexnomic-embed-text pour l’indexationqwen2.5:1.5b pour la rechercheLes 2 LLM tournent en local via ollama.
Le code est ici. La structure du code est très discutable (généré essentiellement par Open Code, et non amélioré à dessein).