Projet final : assembler l'agent complet

Felicitations, tu as toutes les briques. Il est temps de tout assembler. Dans cette derniere etape, tu vas creer une classe AgentWeb qui integre tout ce que tu as construit : les outils, la boucle agent, la memoire, le routing, la gestion d'erreurs, et la citation de sources.

L'objectif est d'avoir une classe unique, propre, configurable, que tu pourrais utiliser dans un vrai projet. Le genre de classe que tu mettrais dans un package Python avec des tests, une doc, et un fichier de configuration.

Ecris une classe AgentWeb qui prend un dictionnaire de configuration a la construction :

config = {
"max_iterations": 5,
"max_chars_page": 3000,
"max_resultats_recherche": 5,
"outils_actifs": ["recherche", "lecture_page", "calculatrice"]
}

La classe doit avoir :

__init__(self, config, llm_fn=None, search_fn=None, fetch_fn=None) :
Initialise l'agent avec sa config et les fonctions injectables.
Cree la memoire (MemoireAgent ou un simple dict cache).
Prepare la boite a outils selon outils_actifs.

run(self, question) :
Classifie la question (calcul direct ou recherche necessaire).
Utilise le cache pour les recherches deja faites.
Execute la boucle agent (LLM + outils) ou repond directement.
Renvoie un dictionnaire avec :
"reponse": la reponse finale
"sources": liste des sources utilisees (peut etre vide)
"iterations": nombre d'iterations
"outils_utilises": liste des noms d'outils appeles
"depuis_cache": True/False selon si les resultats venaient du cache

stats(self) :
Renvoie les statistiques globales : nombre de questions traitees, cache hits, outils les plus utilises.

Pour les tests, tout est injectable : llm_fn simule le LLM, search_fn simule la recherche, fetch_fn simule le telechargement de page.

Exemple :

config = {"max_iterations": 5, "max_resultats_recherche": 3, "max_chars_page": 1000, "outils_actifs": ["recherche", "calculatrice"]}
agent = AgentWeb(config, llm_fn=fake_llm, search_fn=fake_search)
resultat = agent.run("Qui a cree Python ?")
resultat["reponse"] contient la reponse
resultat["sources"] contient les sources utilisees

Tests (4/5)

Agent avec reponse directe

def fake_llm(messages):
    return {"type": "message", "content": "Reponse directe."}

config = {"max_iterations": 5, "outils_actifs": ["recherche"]}
agent = AgentWeb(config, llm_fn=fake_llm)
r = agent.run("Bonjour")
assert r["reponse"] == "Reponse directe."
assert r["iterations"] == 1
assert r["outils_utilises"] == []
assert r["sources"] == []

Agent avec recherche

import json
appels = [0]

def fake_llm(messages):
    appels[0] += 1
    if appels[0] == 1:
        return {"type": "tool_call", "tool_calls": [
            {"function": {"name": "rechercher", "arguments": json.dumps({"query": "Python"})}}
        ]}
    return {"type": "message", "content": "Python est genial [1]."}

def fake_search(query, max_results):
    return [{"title": "Python.org", "url": "https://python.org", "content": "Le site officiel."}]

config = {"max_iterations": 5, "max_resultats_recherche": 3, "outils_actifs": ["recherche"]}
agent = AgentWeb(config, llm_fn=fake_llm, search_fn=fake_search)
r = agent.run("Parle-moi de Python")
assert "Python" in r["reponse"]
assert r["iterations"] == 2
assert "rechercher" in r["outils_utilises"]
assert len(r["sources"]) >= 1

Cache fonctionne entre deux appels

import json
compteur_search = [0]

def fake_search(query, max_results):
    compteur_search[0] += 1
    return [{"title": "Resultat", "url": "https://x.com", "content": "Contenu"}]

appels = [0]
def fake_llm(messages):
    appels[0] += 1
    if appels[0] % 2 == 1:
        return {"type": "tool_call", "tool_calls": [
            {"function": {"name": "rechercher", "arguments": json.dumps({"query": "test"})}}
        ]}
    return {"type": "message", "content": "OK"}

config = {"max_iterations": 5, "outils_actifs": ["recherche"]}
agent = AgentWeb(config, llm_fn=fake_llm, search_fn=fake_search)
agent.run("test 1")
appels[0] = 0
agent.run("test 2")  # meme recherche "test" deja en cache
assert compteur_search[0] == 1, f"La recherche ne doit etre appelee qu'une fois grace au cache, pas {compteur_search[0]}"
assert agent.run("bilan")["depuis_cache"] == False or True  # le cache est verifie

Statistiques globales

def fake_llm(messages):
    return {"type": "message", "content": "OK"}

config = {"max_iterations": 5, "outils_actifs": []}
agent = AgentWeb(config, llm_fn=fake_llm)
agent.run("Question 1")
agent.run("Question 2")
agent.run("Question 3")
s = agent.stats()
assert s["questions_traitees"] == 3, f"3 questions traitees, pas {s['questions_traitees']}"
assert "cache_hits" in s
assert "outils_utilises" in s

+ 0 tests cachés

Indices (3 disponibles)

Solution officielle

import json
import re

class AgentWeb:
    def __init__(self, config, llm_fn=None, search_fn=None, fetch_fn=None):
        self.config = config
        self.llm_fn = llm_fn
        self.search_fn = search_fn
        self.fetch_fn = fetch_fn
        self.cache = {}
        self.historique_questions = []
        self.compteur_outils = {}
        self.total_cache_hits = 0

    def _rechercher(self, query):
        query_lower = query.lower()
        if query_lower in self.cache:
            self.total_cache_hits += 1
            return self.cache[query_lower], True
        max_r = self.config.get("max_resultats_recherche", 5)
        if self.search_fn:
            resultats = self.search_fn(query, max_r)
        else:
            resultats = []
        self.cache[query_lower] = resultats
        return resultats, False

    def _calculer(self, expression):
        import ast, operator
        ops = {ast.Add: operator.add, ast.Sub: operator.sub,
               ast.Mult: operator.mul, ast.Div: operator.truediv,
               ast.Pow: operator.pow}
        def evaluer(n):
            if isinstance(n, ast.Constant) and isinstance(n.value, (int, float)):
                return n.value
            if isinstance(n, ast.BinOp) and type(n.op) in ops:
                return ops[type(n.op)](evaluer(n.left), evaluer(n.right))
            if isinstance(n, ast.UnaryOp) and isinstance(n.op, ast.USub):
                return -evaluer(n.operand)
            raise ValueError()
        try:
            return str(evaluer(ast.parse(expression, mode="eval").body))
        except Exception:
            return "Erreur de calcul"

    def run(self, question):
        outils_utilises = []
        sources = []
        depuis_cache = False
        iterations = 0

        if not self.llm_fn:
            return {"reponse": "Pas de LLM configure.", "sources": [],
                    "iterations": 0, "outils_utilises": [], "depuis_cache": False}

        messages = [{"role": "user", "content": question}]
        max_iter = self.config.get("max_iterations", 5)

        for i in range(max_iter):
            iterations = i + 1
            reponse_llm = self.llm_fn(messages)

            if reponse_llm.get("type") == "message":
                reponse_text = reponse_llm["content"]
                numeros = set(int(m.group(1)) for m in re.finditer(r'\[(\d+)\]', reponse_text))
                for num in sorted(numeros):
                    if num - 1 < len(sources):
                        pass
                break

            if reponse_llm.get("type") == "tool_call":
                for tc in reponse_llm.get("tool_calls", []):
                    nom = tc["function"]["name"]
                    args = json.loads(tc["function"]["arguments"])
                    outils_utilises.append(nom)
                    self.compteur_outils[nom] = self.compteur_outils.get(nom, 0) + 1

                    if nom == "rechercher" and "recherche" in self.config.get("outils_actifs", []):
                        resultats, from_cache = self._rechercher(args.get("query", question))
                        depuis_cache = from_cache
                        sources = [{"numero": j+1, "titre": r.get("title", ""), "url": r.get("url", "")} for j, r in enumerate(resultats)]
                        messages.append({"role": "tool", "content": json.dumps(resultats), "name": nom})
                    elif nom == "calculer" and "calculatrice" in self.config.get("outils_actifs", []):
                        res = self._calculer(args.get("expression", ""))
                        messages.append({"role": "tool", "content": res, "name": nom})
                    else:
                        messages.append({"role": "tool", "content": f"Outil {nom} non disponible", "name": nom})
        else:
            reponse_text = "Nombre maximum d'iterations atteint."

        self.historique_questions.append(question)

        return {
            "reponse": reponse_text,
            "sources": sources,
            "iterations": iterations,
            "outils_utilises": outils_utilises,
            "depuis_cache": depuis_cache
        }

    def stats(self):
        return {
            "questions_traitees": len(self.historique_questions),
            "cache_hits": self.total_cache_hits,
            "requetes_en_cache": len(self.cache),
            "outils_utilises": dict(self.compteur_outils)
        }

← Interface CLI avec Rich CLI basique avec Click →

solution.py

Bravo!

Projet final : assembler l'agent complet

Tests (4/5)

Indices (3 disponibles)

Solution officielle